Harden live SOCKS CI with safe diagnostics and session isolation

.github/workflows/ci.yml

@@ -69,6 +69,29 @@ jobs:
         with:
           name: distribution-tests
           path: ci-results/
+  live-proxy:
+    needs: [distribution]
+    runs-on: macos-latest
+    timeout-minutes: 10
+    env:
+      METAADS_CI_PROXY: ${{ secrets.METAADS_CI_PROXY }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - run: python -m pip install packaging
+      - uses: actions/download-artifact@v4
+        with:
+          name: release-distributions
+          path: dist/
+      - name: Actual Meta SOCKS modes with matched family sessions and safe metadata
+        run: python scripts/check_distribution.py --live --proxy-diagnostics
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: live-proxy-tests
+          path: ci-results/
   live-connectivity:
     needs: [quality, test, distribution]
     runs-on: macos-latest

scripts/check_distribution.py

@@ -71,6 +71,7 @@ def main() -> None:
     parser.add_argument("--minimum", action="store_true", help="Pin curl-cffi to the declared minimum")
     parser.add_argument("--live", action="store_true", help="Run actual Meta integration checks")
     parser.add_argument("--preflight", action="store_true", help="Run only the bounded live connectivity check")
+    parser.add_argument("--proxy-diagnostics", action="store_true", help="Compare real Meta SOCKS DNS modes")
     parser.add_argument("--evidence-dir", type=Path, default=Path("ci-results"))
     args = parser.parse_args()
     private_values = private_proxy_values(os.environ.get("METAADS_CI_PROXY", ""))
@@ -79,6 +80,8 @@ def main() -> None:
             print("::add-mask::" + value.replace("%", "%25").replace("\r", "%0D").replace("\n", "%0A"), flush=True)
     if args.preflight and not args.live:
         parser.error("--preflight requires --live")
+    if args.proxy_diagnostics and (not args.live or args.preflight):
+        parser.error("--proxy-diagnostics requires --live and excludes --preflight")
     root = Path(__file__).resolve().parents[1]
     evidence = args.evidence_dir.resolve()
     evidence.mkdir(parents=True, exist_ok=True)
@@ -149,12 +152,18 @@ def main() -> None:
         env.pop("METAADS_AUDIT_MIN_PYTHON", None)
         if args.minimum:
             env["METAADS_AUDIT_MIN_PYTHON"] = str(python)
-        label = "preflight" if args.preflight else "live" if args.live else "minimum" if args.minimum else "wheel"
-        target = "tests/test_live_preflight.py" if args.preflight else "tests"
+        label = "proxy" if args.proxy_diagnostics else (
+            "preflight" if args.preflight else "live" if args.live else "minimum" if args.minimum else "wheel"
+        )
+        target = "tests/test_live_proxy_transport.py" if args.proxy_diagnostics else (
+            "tests/test_live_preflight.py" if args.preflight else "tests"
+        )
         command = [str(python), "-I", "-m", "pytest", target, "-v",
                    f"--junitxml={evidence / (label + '.xml')}"]
         if args.live:
             command += ["--run-integration", "-m", "integration"]
+            if args.proxy_diagnostics:
+                command += ["-k", "different_python_dns_view or authenticated_socks_proxy", "-o", "junit_family=legacy"]
         else:
             command += ["-m", "not integration"]
         try:

tests/README.md

@@ -1,5 +1,35 @@
 # CI regression cases
 
+## SOCKS DNS modes and private CI sessions
+
+`live-proxy` installs the built wheel outside the checkout and runs all six
+SOCKS regression cases before the slower full-suite result is available. Both
+`socks5://` (client DNS) and `socks5h://` (proxy DNS) remain real Meta tests.
+Related parameter cases share a derived residential session; each test family
+has its own session so earlier full-suite traffic does not determine its exit.
+Explicit user-selected proxy sessions and other providers are preserved.
+
+The JUnit `meta_transport` property records actual Meta destination addresses,
+HTTP status, numeric GraphQL error codes, and whether an LSD token is present.
+It excludes response bodies, error messages, token values, cookies, request
+payloads, URL query strings, and proxy credentials. Privacy unit contracts use
+an ephemeral canary; integration cases collect actual Meta ads.
+
+Run the six cases with your private proxy configured:
+
+```bash
+python -m pytest tests/test_live_proxy_transport.py --run-integration -m integration -k "different_python_dns_view or authenticated_socks_proxy" -o junit_family=legacy
+```
+
+Or verify the distribution built in `dist/`:
+
+```bash
+python scripts/check_distribution.py --live --proxy-diagnostics
+```
+
+All diagnostics failures block CI. The full live suite still runs the same
+cases, and MCP live tests still require the core live suite to pass.
+
 ## Meta media tunnel failures
 
 These cases reproduce the two failures in tag CI run

tests/ci_network.py

@@ -1,12 +1,36 @@
 """Private optional upstream proxy configuration for actual Meta CI traffic."""
 
+import hashlib
 import os
 import uuid
 from urllib.parse import quote, unquote, urlsplit, urlunsplit
 
 from meta_ads_collector.proxy_pool import parse_proxy
 
 
+def scoped_ci_proxy(proxy, scope):
+    """Isolate a live test family while keeping all of its DNS modes on one exit.
+
+    Only our generated Apify sessions are scoped. A user's explicit session or
+    another proxy provider remains unchanged. Credentials never enter the label.
+    """
+    if not proxy:
+        return proxy
+    parsed = urlsplit(parse_proxy(proxy))
+    if parsed.hostname != "proxy.apify.com" or not parsed.username:
+        return proxy
+    parts = unquote(parsed.username).split(",")
+    sessions = [part for part in parts if part.startswith("session-")]
+    if len(sessions) != 1 or not sessions[0].startswith("session-metaads"):
+        return proxy
+    suffix = hashlib.sha256(scope.encode("utf-8")).hexdigest()[:12]
+    session = sessions[0][:44] + suffix
+    parts[parts.index(sessions[0])] = session
+    authority = quote(",".join(parts), safe="") + ":" + (parsed.password or "")
+    authority += "@" + parsed.hostname + (f":{parsed.port}" if parsed.port else "")
+    return urlunsplit(parsed._replace(netloc=authority))
+
+
 def validate_ci_proxy(proxy, required=False):
     if required and not proxy:
         raise ValueError("METAADS_CI_PROXY is required for live GitHub CI checks")

tests/meta_transport_diagnostics.py

@@ -0,0 +1,64 @@
+"""Allowlisted metadata from real Meta responses; never cookies, tokens or bodies."""
+
+from __future__ import annotations
+
+import inspect
+import json
+from urllib.parse import urlsplit
+
+
+def response_metadata(client, method, url, response):
+    """Do not serialize server messages, URLs, request arguments or credentials."""
+    parsed = urlsplit(url)
+    if parsed.hostname not in {"www.facebook.com", "web.facebook.com"}:
+        return None
+    codes = []
+    if parsed.path == "/api/graphql/":
+        try:
+            text = response.text
+            payload = json.loads(text[9:] if text.startswith("for (;;);") else text)
+            if isinstance(payload, dict):
+                for error in payload.get("errors", []) or []:
+                    if isinstance(error, dict) and type(error.get("code")) is int:
+                        codes.append(error["code"])
+        except (ValueError, TypeError):
+            pass
+    return {
+        "method": "POST" if method.upper() == "POST" else "GET",
+        "endpoint": "graphql" if parsed.path == "/api/graphql/" else "bootstrap",
+        "host": parsed.hostname,
+        "status": int(response.status_code),
+        "error_codes": codes,
+        "lsd_present": bool(getattr(client, "_tokens", {}).get("lsd")),
+    }
+
+
+class MetaTransportDiagnostics:
+    """Observe the existing request method without changing any network behavior."""
+
+    def __init__(self):
+        self.responses = []
+
+    def instrument(self, patch, client):
+        request = client._make_request
+
+        def record(method, url, response):
+            entry = response_metadata(client, method, url, response)
+            if entry is not None and len(self.responses) < 100:
+                self.responses.append(entry)
+
+        if inspect.iscoroutinefunction(request):
+            async def async_request(method, url, **kwargs):
+                response = await request(method, url, **kwargs)
+                record(method, url, response)
+                return response
+            patch.setattr(client, "_make_request", async_request)
+        else:
+            def sync_request(method, url, **kwargs):
+                response = request(method, url, **kwargs)
+                record(method, url, response)
+                return response
+            patch.setattr(client, "_make_request", sync_request)
+
+    def snapshot(self, destinations):
+        return {"responses": list(self.responses), "meta_destinations": sorted(set(destinations))}

tests/test_audit_packaging.py

@@ -63,6 +63,7 @@ def test_published_sdist_includes_support_for_the_tests_it_ships(tmp_path):
     required = {
         "tests/__init__.py", "tests/utils.py", "tests/conftest.py", "tests/audit_meta_samples.py",
         "tests/meta_full_sample.py", "tests/meta_forward_proxy.py", "tests/ci_network.py",
+        "tests/meta_transport_diagnostics.py",
         "tests/test_all_meta_fields.py",
         "tests/test_live_proxy_transport.py", "scripts/check_distribution.py",
         "tests/README.md", ".github/workflows/ci.yml", ".github/workflows/publish.yml",

tests/test_ci_configuration.py

@@ -40,7 +40,7 @@ def test_published_releases_keep_full_validation_and_private_proxy_forwarding():
 
 def test_live_failures_remain_blocking_and_mcp_live_runs_after_core_live():
     jobs = workflow("ci")["jobs"]
-    for name in ("live-connectivity", "live-meta", "mcp-live"):
+    for name in ("live-connectivity", "live-proxy", "live-meta", "mcp-live"):
         job = jobs[name]
         assert job.get("continue-on-error", "false") == "false"
         assert job["env"]["METAADS_CI_PROXY"] == "${{ secrets.METAADS_CI_PROXY }}"
@@ -50,6 +50,15 @@ def test_live_failures_remain_blocking_and_mcp_live_runs_after_core_live():
     assert any("scripts/check_distribution.py --live" in step.get("run", "") for step in jobs["live-meta"]["steps"])
 
 
+def test_sock_dns_diagnostics_run_on_actual_distribution_and_remain_blocking():
+    job = workflow("ci")["jobs"]["live-proxy"]
+    assert job["needs"] == ["distribution"]
+    assert job["runs-on"] == "macos-latest"
+    assert any("--live --proxy-diagnostics" in step.get("run", "") for step in job["steps"])
+    upload = [step for step in job["steps"] if step.get("uses", "").startswith("actions/upload-artifact")]
+    assert len(upload) == 1 and upload[0]["if"] == "always()"
+
+
 def test_public_package_verification_includes_core_and_mcp_live_tests():
     verification = workflow("publish")["jobs"]["verify-published"]
     commands = "\n".join(step.get("run", "") for step in verification["steps"])

tests/test_ci_network.py

@@ -78,3 +78,34 @@ def test_distribution_checks_do_not_filter_out_integration_regressions():
     # Captured-data fault injection is labeled separately in the report, but
     # must still be exercised alongside every other permanent regression.
     assert '"not controlled"' not in inspect.getsource(main)
+
+
+def test_scoped_proxy_matches_dns_cases_but_separates_families(monkeypatch):
+    from urllib.parse import unquote, urlsplit
+
+    from tests.ci_network import scoped_ci_proxy
+
+    monkeypatch.setenv("GITHUB_RUN_ID", "123456789")
+    configure_ci_network(monkeypatch, "http://groups-RESIDENTIAL,country-US:ephemeral@proxy.apify.com:8000")
+    shared = os.environ["METAADS_CI_PROXY"]
+    first = scoped_ci_proxy(shared, "authenticated_socks")
+    assert first == scoped_ci_proxy(shared, "authenticated_socks")
+    assert first != scoped_ci_proxy(shared, "independent_dns")
+    assert first != shared
+    parsed = urlsplit(first)
+    parameters = unquote(parsed.username).split(",")
+    assert "groups-RESIDENTIAL" in parameters and "country-US" in parameters
+    sessions = [item.removeprefix("session-") for item in parameters if item.startswith("session-")]
+    assert len(sessions) == 1 and len(sessions[0]) <= 50
+    assert parsed.password == "ephemeral" and parsed.hostname == "proxy.apify.com" and parsed.port == 8000
+
+
+@pytest.mark.parametrize("proxy", [
+    None,
+    "http://audit:ephemeral@127.0.0.1:8123",
+    "http://groups-RESIDENTIAL,session-userchosen:ephemeral@proxy.apify.com:8000",
+])
+def test_scoped_proxy_preserves_user_selected_session_and_other_providers(proxy):
+    from tests.ci_network import scoped_ci_proxy
+
+    assert scoped_ci_proxy(proxy, "authenticated_socks") == proxy

tests/test_live_proxy_transport.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import ipaddress
+import json
+import os
 import secrets
 from urllib.parse import quote
 
@@ -12,6 +14,29 @@
 from meta_ads_collector.collector import MetaAdsCollector
 
 from .meta_forward_proxy import MetaForwardProxy, MetaSocksProxy, _is_meta_host
+from .meta_transport_diagnostics import MetaTransportDiagnostics
+
+
+@pytest.fixture(autouse=True)
+def isolated_proxy_family(request, ci_meta_network, monkeypatch):
+    """Keep related DNS cases matched, separate from preceding live-suite load."""
+    from .ci_network import scoped_ci_proxy
+
+    if request.node.get_closest_marker("integration"):
+        proxy = os.environ.get("METAADS_CI_PROXY")
+        if proxy:
+            monkeypatch.setenv("METAADS_CI_PROXY", scoped_ci_proxy(proxy, request.node.originalname))
+
+
+@pytest.fixture
+def transport_diagnostics(record_property):
+    diagnostics = MetaTransportDiagnostics()
+    gateways = []
+    yield diagnostics, gateways
+    destinations = [host for gateway in gateways for host in gateway.connections]
+    evidence = json.dumps(diagnostics.snapshot(destinations), sort_keys=True)
+    record_property("meta_transport", evidence)
+    print("Meta transport metadata: " + evidence)
 
 
 def _proxy_string(proxy, form):
@@ -53,7 +78,9 @@ def test_forward_gateway_allows_meta_subdomains_without_allowing_lookalikes(host
 
 @pytest.mark.integration
 @pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
-def test_live_socks5_meta_collection_survives_a_different_python_dns_view(monkeypatch, async_mode):
+def test_live_socks5_meta_collection_survives_a_different_python_dns_view(
+    monkeypatch, async_mode, transport_diagnostics,
+):
     import asyncio
     import socket
 
@@ -68,16 +95,20 @@ def independent_dns_view(host, *args, **kwargs):
 
     monkeypatch.setattr(socket, "getaddrinfo", independent_dns_view)
     with MetaSocksProxy("audit@meta", "local/" + secrets.token_hex(12)) as proxy:
+        diagnostics, gateways = transport_diagnostics
+        gateways.append(proxy)
         address = _proxy_string(proxy, "url").replace("http://", "socks5://", 1)
         if async_mode:
             async def collect():
                 async with AsyncMetaAdsCollector(proxy=address, timeout=30, max_retries=2,
                                                  rate_limit_delay=0.5, jitter=0) as collector:
+                    diagnostics.instrument(monkeypatch, collector.client)
                     return await collector.collect(query="nike", country="US", max_results=3, page_size=2)
             ads = asyncio.run(collect())
         else:
             with MetaAdsCollector(proxy=address, timeout=30, max_retries=2,
                                   rate_limit_delay=0.5, jitter=0) as collector:
+                diagnostics.instrument(monkeypatch, collector.client)
                 ads = list(collector.search(query="nike", country="US", max_results=3, page_size=2))
         assert len(ads) == 3
         assert all(ad.id and ad.page and ad.page.id for ad in ads)
@@ -169,22 +200,28 @@ async def collect():
 @pytest.mark.integration
 @pytest.mark.parametrize("scheme", ["socks5", "socks5h"])
 @pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
-def test_live_authenticated_socks_proxy_collects_actual_meta_ads(scheme, async_mode):
+def test_live_authenticated_socks_proxy_collects_actual_meta_ads(
+    scheme, async_mode, monkeypatch, transport_diagnostics,
+):
     import asyncio
 
     from meta_ads_collector.async_collector import AsyncMetaAdsCollector
 
     with MetaSocksProxy("audit@meta", "local/" + secrets.token_hex(12)) as proxy:
+        diagnostics, gateways = transport_diagnostics
+        gateways.append(proxy)
         address = _proxy_string(proxy, "url").replace("http://", scheme + "://", 1)
         if async_mode:
             async def collect():
                 async with AsyncMetaAdsCollector(proxy=address, timeout=30, max_retries=2,
                                                  rate_limit_delay=0.5, jitter=0) as collector:
+                    diagnostics.instrument(monkeypatch, collector.client)
                     return await collector.collect(query="nike", country="US", max_results=3, page_size=2)
             ads = asyncio.run(collect())
         else:
             with MetaAdsCollector(proxy=address, timeout=30, max_retries=2,
                                   rate_limit_delay=0.5, jitter=0) as collector:
+                diagnostics.instrument(monkeypatch, collector.client)
                 ads = list(collector.search(query="nike", country="US", max_results=3, page_size=2))
         assert len(ads) == 3
         assert all(ad.id and ad.page and ad.page.id for ad in ads)