BrokenPipeError 'Future exception was never retrieved' during subprocess cleanup

[liquidsec]

Summary

During scan shutdown (commonly observed in a module's finish() phase), asyncio emits warnings of the form:

Future exception was never retrieved
future: <Future finished exception=BrokenPipeError()>
Traceback (most recent call last):
  File "bbot/core/helpers/command.py", line 223, in _write_proc_line
    await proc.stdin.drain()
  File ".../asyncio/streams.py", line 371, in drain
    await self._protocol._drain_helper()
  File ".../asyncio/streams.py", line 173, in _drain_helper
    await waiter
BrokenPipeError

The scan itself completes successfully — this is cosmetic noise, not a functional failure. No events are lost.

Root cause

In bbot/core/helpers/command.py (run_live's finally block), when a subprocess is still alive at teardown we:

1. proc.terminate()
2. await proc.wait() (subprocess dies, kernel breaks its stdin pipe)
3. input_task.cancel() (the task feeding stdin)

Between steps 2 and 3, the input_task is typically awaiting proc.stdin.drain(), which is awaiting an internal _drain_waiter Future. The cancellation at step 3 injects CancelledError — which is a BaseException, so the except Exception in _write_proc_line does not catch it, and the task tears down before the drain waiter is resolved.

Shortly after, asyncio's call_soon-scheduled _call_connection_lost(BrokenPipeError(...)) runs and calls waiter.set_exception(BrokenPipeError()) on the now-abandoned drain waiter. With nobody left to retrieve it, the Future is GC'd and asyncio logs the warning.

Reproducer

Any module that pipes stdin to a subprocess via run_process_live(..., input=...) (e.g. gowitness, nuclei, fingerprintx) can trigger it during scan shutdown. Observed running the waf_bypass module with active scans, where the long finish() phase increases the chance of overlapping teardown.

Proposed fix

After cancelling input_task, await it so pending callbacks (including the drain waiter's deferred exception set) are flushed under a context where the exception is retrieved:

if input_task is not None:
    input_task.cancel()
    with contextlib.suppress(BaseException):
        await input_task

Alternative: close proc.stdin before proc.terminate() so the writer side shuts down cleanly and the cancel race is avoided.

Impact

• Severity: low (warning only, no functional impact)
• Visibility: moderate (looks alarming in logs)