Function attempt_gate

examples/policy-advisor/ctf.py:217–259

Try to pass through a gate. Returns ``("pass", flag)`` on success, ``("blocked", reason)`` when the proxy denied the connection (retryable), or ``("error", detail)`` for a real upstream failure (not retryable).

(gate: dict)

def attempt_gate(gate: dict) -> tuple[str, str]:
    """Try to pass through a gate.

    Returns ``("pass", flag)`` on success, ``("blocked", reason)`` when the
    proxy denied the connection (retryable), or ``("error", detail)`` for a
    real upstream failure (not retryable).
    """
    if gate.get("use_curl"):
        return attempt_gate_curl(gate)
    try:
        req = urllib.request.Request(
            gate["url"],
            headers=gate.get("headers") or {},
            method=gate["method"],
        )
        if gate.get("body"):
            req.data = gate["body"].encode("utf-8")

        with urllib.request.urlopen(req, timeout=15) as resp:
            data = resp.read().decode("utf-8")
            flag = gate["extract"](data)
            return "pass", flag

    except urllib.error.HTTPError as exc:
        if exc.code == 403:
            return "blocked", "blocked by sandbox proxy (403)"
        return "error", f"HTTP {exc.code} from {gate['host']}"

    except urllib.error.URLError as exc:
        if _is_proxy_block(exc):
            return "blocked", "blocked by sandbox proxy"
        reason = str(exc.reason)
        if "timed out" in reason:
            return "blocked", "connection timed out"
        return "blocked", f"connection failed ({reason})"

    except (ConnectionError, OSError, socket.timeout) as exc:
        if _is_proxy_block(exc):
            return "blocked", "connection refused by proxy"
        return "blocked", f"network error ({exc})"

    except Exception as exc:  # noqa: BLE001
        return "error", f"unexpected error ({exc})"


# -- Banner / victory ----------------------------------------------------------