Is Your Traffic Actually Proxied? Building an Egress Canary for Python Scrapers

Every proxy integration I have debugged at 2 a.m. had the same shape: the code believed it was proxied, the logs said "proxy configured," and the traffic was leaving the building through the front door, naked. Proxy plumbing is not a boolean you flip at startup. It is a live property of every request you send, and a dozen perfectly reasonable code paths can quietly un-set it while everything downstream keeps reporting success.

The thesis of this article is simple: never assume your scraper is proxied. Verify egress continuously with a canary — a periodic, real check of "which IP did that request actually leave from, and was it the one I asked for?" — the same way you'd health-check a database you refuse to trust.

The silent-bypass catalog

These are the failure modes, with mechanisms. None of them are exotic; most are the default behavior of well-loved libraries.

• trust_env surprises. Both requests and httpx default to trust_env=True, which means stray HTTP_PROXY, HTTPS_PROXY, and NO_PROXY values in the environment silently reshape your routing. The nasty one is NO_PROXY: its matching is case-insensitive in practice but case-sensitive-looking in config files, and its suffix semantics vary across versions — an entry of example.com that was meant to exempt an internal API can accidentally exempt api.example.com, your actual scrape target, too. One forgotten export NO_PROXY=.example.com in a shell profile, and every request to that domain goes direct while your session still carries proxy settings that are never consulted.
• Session reuse and redirects dropping proxy-like state. Libraries that accept a proxies argument don't all carry it across the same boundaries. Some clients bind proxy configuration at connection-pool creation time, so mid-life changes to the session's proxy map do nothing on reused keep-alive connections. Redirect handling is a classic gap: a 302 to a different host can be followed with the new host matched against NO_PROXY, so request #1 goes through your exit and request #2 does not — same job, same session, two different egresses.
• The refactor special: proxy=None. A signature change adds proxy: str | None = None, a call site forgets the argument, and None doesn't mean "use the default" — it means "no proxy," explicitly. An empty dict {} for proxies does the same thing more quietly. Type checkers love it. ipinfo.io loves it more.
• Missing SOCKS support, with two different personalities. SOCKS proxies need PySocks (requests[socks]). Without it, requests raises loudly — good. Other stacks are worse: some async HTTP clients and HTTPX-with-fallback configurations will downgrade or ignore proxy-like settings they can't service, so a socks5:// URL becomes decorative. Same typo, different blast radius depending on which library got upgraded last.
• Startup ordering. Gunicorn or uvicorn workers spawned before your config-management system drops the proxy environment variables into the container. The main process has them, the workers inherited an older environ at fork() time, and no amount of re-reading config in request handlers fixes an env that was captured at process birth. Half your fleet is proxied; the other half is your datacenter's IP getting rate-limited.
• The geo lie. The traffic is proxied. It is just proxied out of the wrong country, because someone updated a country-code parameter in one config file and not the one your worker actually reads. Note what happened here: a naive assert ip != my_datacenter_ip — the "is it proxied?" check most teams write — passes. The check was a boolean; the intent was a country.

Why startup-only checks don't cut it

The standard mitigation — ping an echo endpoint once during boot and fail fast — covers exactly one moment in time. Config drifts after boot. Dev and prod have different environments (that's the entire premise of trust_env biting you). Per-worker variance means your boot check passing in worker 0 says nothing about worker 7, which was forked into a different environ. Verification has to be a pulse, not a snapshot.

The verifier

The core idea: send one request to an echo endpoint (ipinfo.io, ifconfig.me, icanhazip.com — anything that reports your observed egress) through the same session and proxy configuration that production traffic uses, and compare the answer against what you believe your configuration means.

One design decision matters: match on ASN and country, not raw IP. Rotating residential exits legitimately return a different IP every call; asserting IP equality will flood you with false alarms. The raw IP is still worth recording — it becomes important for the sticky-session check later.

"""egress_canary/verify.py"""
from dataclasses import dataclass, field
import ipaddress

import requests

ECHO_URL = "https://ipinfo.io/json"  # echo endpoint: reports observed egress


class EgressError(RuntimeError):
    """Raised when observed egress does not match intent."""
    DIRECT_TRAFFIC = "DIRECT_TRAFFIC"
    WRONG_GEO = "WRONG_GEO"
    UNKNOWN_ASN = "UNKNOWN_ASN"
    ECHO_UNREACHABLE = "ECHO_UNREACHABLE"

    def __init__(self, kind: str, detail: str, observed: dict | None = None):
        super().__init__(f"[{kind}] {detail}")
        self.kind = kind
        self.observed = observed or {}


@dataclass
class ExpectedEgress:
    country: str | None = None            # e.g. "DE"
    asn_prefixes: tuple[str, ...] = ()    # e.g. ("AS12345", "AS67890")
    direct_cidrs: tuple[str, ...] = ()    # your own ranges = unproxied
    timeout: float = 10.0


@dataclass
class VerifyResult:
    ip: str
    country: str
    org: str
    history: list = field(default_factory=list)  # internal use


def verify_egress(session: requests.Session, expected: ExpectedEgress) -> VerifyResult:
    """Send one echo request THROUGH `session` (same proxies, env,
    trust_env behavior as production traffic) and assert intent."""
    try:
        resp = session.get(ECHO_URL, timeout=expected.timeout)
        resp.raise_for_status()
        data = resp.json()
    except requests.RequestException as exc:
        raise EgressError(EgressError.ECHO_UNREACHABLE,
                          f"echo endpoint failed: {exc}") from exc

    ip = data.get("ip", "")
    country = data.get("country", "")
    org = data.get("org", "")  # e.g. "AS12345 Example ISP"
    result = VerifyResult(ip=ip, country=country, org=org)

    # 1. Direct-traffic detection: our own infrastructure ranges.
    try:
        addr = ipaddress.ip_address(ip)
        for cidr in expected.direct_cidrs:
            if addr in ipaddress.ip_network(cidr):
                raise EgressError(
                    EgressError.DIRECT_TRAFFIC,
                    f"egress {ip} is inside our own range {cidr} — "
                    f"request left UNPROXIED",
                    {"ip": ip},
                )
    except ValueError:
        pass  # not parseable; ASN check below still applies

    # 2. Geo intent: proxied, but to the wrong place.
    if expected.country and country.upper() != expected.country.upper():
        raise EgressError(
            EgressError.WRONG_GEO,
            f"expected country {expected.country}, observed {country!r} "
            f"(ip={ip}, org={org})",
            {"ip": ip, "country": country},
        )

    # 3. ASN intent — match ASN, not IP, so rotation is allowed.
    if expected.asn_prefixes and not any(
        asn in org for asn in expected.asn_prefixes
    ):
        raise EgressError(
            EgressError.UNKNOWN_ASN,
            f"observed org {org!r} matches none of "
            f"{expected.asn_prefixes} (ip={ip})",
            {"ip": ip, "org": org},
        )

    return result

A typical scrape job wires it in like this:

"""scrape_job.py"""
import requests

from egress_canary.verify import ExpectedEgress, EgressError, verify_egress

PROXIES = {
    "http":  "http://user:[e-mail]:8443",
    "https": "http://user:[e-mail]:8443",
}

EXPECTED = ExpectedEgress(
    country="DE",
    asn_prefixes=("AS24940", "AS9009"),  # illustrative: provider ASN pool
    direct_cidrs=("203.0.113.0/24",),   # our datacenter, per RFC 5737
)


def build_session() -> requests.Session:
    s = requests.Session()
    s.proxies = PROXIES
    # Belt and suspenders: if your config IS the proxies dict,
    # do not let stray env vars veto it. Decide this deliberately;
    # if you DO rely on env proxies, leave trust_env=True and let
    # the canary catch what it does.
    s.trust_env = False
    return s


def run(urls: list[str]) -> None:
    session = build_session()
    try:
        verify_egress(session, EXPECTED)
    except EgressError as exc:
        # Refuse to scrape unverified. Log, page, exit non-zero.
        print(f"egress verification failed: {exc}")
        raise
    for url in urls:
        page = session.get(url, timeout=15)
        # ... parse ...

Illustrative output when the bypass is real:

egress verification failed: [DIRECT_TRAFFIC] egress 203.0.113.42 is
inside our own range 203.0.113.0/24 — request left UNPROXIED

Making it continuous

A canary is one cheap echo request amortized over many real ones. The two knobs are frequency (every_n_requests, every_seconds — whichever fires first) and what you do on failure. My preference: fail loud but structured — a counter you alert on, plus an exception in jobs where one bad exit contaminates the whole batch.

"""egress_canary/canary.py"""
import threading
import time
from collections import Counter

from egress_canary.verify import EgressError, ExpectedEgress, verify_egress


class EgressCanary:
    """Wrap a live session; spend one echo request every N calls
    or every T seconds, whichever comes first."""

    def __init__(self, session, expected: ExpectedEgress,
                 every_n_requests: int = 200, every_seconds: float = 300.0,
                 raise_on_failure: bool = False):
        self._session = session
        self._expected = expected
        self._n = every_n_requests
        self._t = every_seconds
        self._calls = 0
        self._last_check = 0.0
        self._lock = threading.Lock()
        self.counts: Counter[str] = Counter()
        self.last_error: EgressError | None = None

    @property
    def proxy_verified_ratio(self) -> float:
        checked = self.counts["verified"] + self.counts["failed"]
        return self.counts["verified"] / checked if checked else 1.0

    @property
    def unexpected_direct_count(self) -> int:
        return self.counts["DIRECT_TRAFFIC"]

    def request(self, url: str, **kwargs):
        with self._lock:
            self._calls += 1
            due = (self._calls >= self._n
                   or (time.monotonic() - self._last_check) >= self._t)
        if due:
            self._check()
        return self._session.get(url, **kwargs)

    def _check(self) -> None:
        self._calls = 0
        self._last_check = time.monotonic()
        try:
            verify_egress(self._session, self._expected)
            self.counts["verified"] += 1
            self.last_error = None
        except EgressError as exc:
            self.counts["failed"] += 1
            self.counts[exc.kind] += 1
            self.last_error = exc
            if self.raise_on_failure:
                raise
        # In real life, push self.counts and the two ratio gauges to
        # Prometheus/Datadog here instead of keeping them in-process.

Scrape jobs report proxy_verified_ratio (alert below, say, 0.99) and unexpected_direct_count (alert above 0 — a single direct egress is one config bug too many). The echo call itself goes through self._session, so it inherits exactly the proxy config, redirect behavior, and trust_env semantics of real traffic. A canary on a freshly built session proves nothing about the session doing the work.

And the CI gate — this is the one that catches the proxy=None refactor before it ships:

"""tests/test_egress_smoke.py"""
import pytest

from egress_canary.verify import EgressError, verify_egress
from scrape_job import EXPECTED, build_session


def test_egress_is_proxied():
    session = build_session()
    try:
        result = verify_egress(session, EXPECTED)
    except EgressError as exc:
        pytest.fail(f"CI egress smoke failed: {exc}")
    assert result.country == EXPECTED.country

Illustrative CI failure:

FAILED tests/test_egress_smoke.py::test_egress_is_proxied -
  CI egress smoke failed: [WRONG_GEO] expected country DE,
  observed 'NL' (ip=185.x.y.z, org=AS9009 ...)

The intent check — and the sticky/rotating symmetry test

Verifying the country is the "intent check" the naive boolean can't do: proxy_verified == True is satisfied by a Netherlands exit when the config says Germany, and only comparing observed country against config-declared country catches it. But you can go one level deeper, because egress patterns carry meaning.

If your session uses a sticky (persistent) endpoint — same exit IP for the life of the session, the way you keep a shopping cart or pagination stable — then consecutive canary checks on the same session should return the same IP. If your configuration is rotating, consecutive checks should return different IPs. Now watch what that symmetry gives you: it catches the config typo where a sticky endpoint got swapped for a rotating one (or vice versa) — a bug that passes every boolean check, every ASN check, every country check, and only reveals itself as forty blocked pagination runs. So verify_egress results should be recorded per session: same-IP across checks when sticky is declared, differing IPs when rotation is declared, and flip the canary to raise_on_failure=True if the observed pattern contradicts the declared one. The canary stops being just "is a proxy in front of me?" and becomes "is my architecture still what I think it is?"

Closing

Egress verification is cheap: one echo request per few hundred, a dataclass of intent, and a typed exception. Silent bypasses are not cheap — they cost you ban rates measured in weeks, and an incident postmortem that starts with "but the code sets proxies." Stop asking whether the code says proxy. Ask the internet what IP it saw, on a schedule, through the same session doing the work. The answer is the only source of truth you have.

Disclosure: I use Thordata's residential proxies for this project. New users get 500MB free — code thor020 (10% off): https://www.thordata.com/?ls=uXcSHJzx&lk=02-tele