fix: tie proxy readiness to actual socket bind; report real port; sync stop

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/auto_reverse/proxy.py

@@ -71,10 +71,14 @@ class ProxyServer:
         return self._port
 
     def start(self) -> None:
+        requested_port = self._port
         ready = threading.Event()
         self._thread = threading.Thread(target=self._run, args=(ready,), daemon=True)
         self._thread.start()
-        ready.wait(timeout=10)
+        if not ready.wait(timeout=10):
+            raise RuntimeError(
+                f"proxy failed to start (port {requested_port} bind failed?)"
+            )
 
     def _run(self, ready: threading.Event) -> None:
         from mitmproxy.options import Options
@@ -90,11 +94,22 @@ class ProxyServer:
             master: Any = DumpMaster(opts, with_termlog=False, with_dumper=False)
             master.addons.add(CaptureAddon(self._store, self._archive_path))
             self._master = master
+            # Schedule the master so we can poll for the actual listen socket
+            # rather than signalling readiness before the bind has happened.
+            serve_task = asyncio.ensure_future(master.run())
+            ps: Any = master.addons.get("proxyserver")
+            while not ps.listen_addrs() and not serve_task.done():
+                await asyncio.sleep(0.01)
+            if ps.listen_addrs():
+                # Record the real bound port (matters for port=0 / ephemeral).
+                self._port = ps.listen_addrs()[0][1]
                 ready.set()
-            await master.run()
+            await serve_task
 
         self._loop.run_until_complete(_serve())
 
     def stop(self) -> None:
         if self._master is not None and self._loop is not None:
             self._loop.call_soon_threadsafe(self._master.shutdown)
+        if self._thread is not None:
+            self._thread.join(timeout=5)

tests/test_proxy.py

@@ -1,6 +1,9 @@
 from types import SimpleNamespace
 
-from auto_reverse.proxy import flow_from_mitm
+import pytest
+
+from auto_reverse.proxy import ProxyServer, flow_from_mitm
+from auto_reverse.store import FlowStore, ScopeFilter
 
 
 def _fake_mitm_flow():
@@ -32,3 +35,16 @@ def test_flow_from_mitm_handles_missing_response():
     flow.response = None
     captured = flow_from_mitm(flow)
     assert captured is None
+
+
+def test_proxy_server_binds_and_reports_real_port(tmp_path):
+    store = FlowStore(ScopeFilter(target_hosts={"ex.com"}))
+    server = ProxyServer(store, archive_path=tmp_path / "archive.log", port=0)
+    try:
+        server.start()
+    except Exception as exc:
+        pytest.skip(f"proxy bind unavailable: {exc}")
+    try:
+        assert server.port != 0  # real OS-assigned port after binding
+    finally:
+        server.stop()