Function run_gate

examples/policy-advisor/ctf.py:325–382

Run a single gate with retry loop. Returns True on success.

(gate: dict)

def run_gate(gate: dict) -> bool:
    """Run a single gate with retry loop. Returns True on success."""
    num = gate["num"]
    total = len(GATES)

    print(f"  {CYAN}{BOLD}{'=' * 60}{RESET}")
    print(f"  {CYAN}{BOLD}  GATE {num}/{total}:  {gate['name'].upper()}{RESET}")
    proto = "https" if gate["port"] == 443 else "http"
    print(
        f"  {DIM}  {gate['hint']}{RESET}\n"
        f"  {DIM}  target: {proto}://{gate['host']}:{gate['port']}  "
        f"({gate['method']}){RESET}"
    )
    print(f"  {CYAN}{BOLD}{'=' * 60}{RESET}")
    print()

    for attempt in range(1, MAX_RETRIES + 1):
        log(
            "GATE",
            f"Gate {num}  attempt #{attempt}",
            host=gate["host"],
            port=gate["port"],
        )

        status, result = attempt_gate(gate)

        if status == "pass":
            log("PASS", f"Gate {num} UNLOCKED")
            log("FLAG", result)
            print()
            return True

        if status == "error":
            log("WARN", f"Gate {num} skipped: {result}")
            log("INFO", "This is an upstream error, not a proxy block")
            print()
            return True

        log("FAIL", f"Gate {num}: {result}")

        if attempt == 1:
            log("WARN", "Approve the recommendation in the TUI to proceed")
            log(
                "INFO",
                "TUI: select sandbox -> [r] drafts -> [a] approve",
            )

        for remaining in range(RETRY_INTERVAL, 0, -1):
            print(
                f"\r  {DIM}        retrying in {remaining:>2}s ...{RESET}",
                end="",
                flush=True,
            )
            time.sleep(1)
        print("\r" + " " * 50 + "\r", end="", flush=True)

    log("FAIL", f"Gate {num} timed out after {MAX_RETRIES} attempts")
    return False