Rotating Proxy Python: Sessions, Scheduling, and Exit Checks

python -m venv .venv
# Activate .venv using the command appropriate to your shell.
python -m pip install "requests[socks]" selenium
python -m pip freeze > requirements-lock.txt

"""Bounded, authorized exit-IP checks through a configured proxy gateway.

The username must be copied/generated according to the current provider docs.
This script never changes routes in response to a target restriction.
"""

import argparse
import json
import os
from ipaddress import ip_address
from urllib.parse import quote, urlsplit

import requests


DEFAULT_CHECK_URL = "https://api64.ipify.org?format=json"
STOP_STATUSES = {401, 403, 429}
MAX_RESPONSE_BYTES = 4096


def proxy_url_from_env(username_override=None):
    scheme = os.getenv("ROLA_PROXY_SCHEME", "socks5h")
    if scheme not in {"socks5h", "http"}:
        raise ValueError("ROLA_PROXY_SCHEME must be socks5h or http")

    host = os.environ["ROLA_PROXY_HOST"].strip()
    port = int(os.environ["ROLA_PROXY_PORT"])
    username = username_override or os.environ["ROLA_PROXY_USERNAME"]
    password = os.environ["ROLA_PROXY_PASSWORD"]

    if not host or "://" in host or "/" in host or "@" in host:
        raise ValueError("ROLA_PROXY_HOST must be a hostname, not a URL")
    if not 1 <= port <= 65535:
        raise ValueError("ROLA_PROXY_PORT is out of range")
    if not username or not password:
        raise ValueError("Proxy credentials must not be empty")

    return (
        f"{scheme}://{quote(username, safe='')}:"
        f"{quote(password, safe='')}@{host}:{port}"
    )


def validate_check_url(url):
    parts = urlsplit(url)
    if (
        parts.scheme not in {"http", "https"}
        or not parts.hostname
        or parts.username
        or parts.password
        or parts.fragment
        or (parts.query and url != DEFAULT_CHECK_URL)
        or (
            parts.scheme == "http"
            and (
                parts.hostname != "fixture.invalid"
                or os.getenv("ROLA_PROXY_SCHEME") != "http"
                or os.getenv("ROLA_PROXY_HOST") not in {"127.0.0.1", "localhost"}
            )
        )
    ):
        raise ValueError("Use HTTPS without credentials; HTTP is limited to the local fixture")
    return url


def check_exit(session, check_url, proxy_url, timeout=(5, 15)):
    with session.get(
        check_url,
        proxies={"http": proxy_url, "https": proxy_url},
        timeout=timeout,
        allow_redirects=False,
        stream=True,
    ) as response:
        if response.status_code in STOP_STATUSES:
            raise RuntimeError(f"HTTP {response.status_code}: stop; do not change routes")
        if response.status_code == 407:
            raise RuntimeError("HTTP 407: proxy authentication failed")
        if response.status_code != 200:
            raise RuntimeError(f"HTTP {response.status_code}: inspect this response separately")
        if int(response.headers.get("Content-Length", "0")) > MAX_RESPONSE_BYTES:
            raise ValueError("Exit-check response is too large")
        body = bytearray()
        for chunk in response.iter_content(chunk_size=1024):
            body.extend(chunk)
            if len(body) > MAX_RESPONSE_BYTES:
                raise ValueError("Exit-check response is too large")

    try:
        data = json.loads(body)
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        raise ValueError("Exit-check response is not valid JSON") from exc
    if not isinstance(data, dict):
        raise ValueError("Exit-check response must be a JSON object")
    raw_ip = data.get("ip")
    if not isinstance(raw_ip, str):
        raise ValueError("Exit-check 'ip' must be a string")
    try:
        observed_ip = str(ip_address(raw_ip.strip()))
    except ValueError as exc:
        raise ValueError("Exit-check response contains an invalid IP") from exc
    return {"ip": observed_ip, "country": str(data.get("country", "unknown"))}


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", default=DEFAULT_CHECK_URL)
    parser.add_argument("--checks", type=int, default=2)
    args = parser.parse_args(argv)

    if not 1 <= args.checks <= 3:
        parser.error("--checks must be between 1 and 3")

    check_url = validate_check_url(args.url)
    proxy_url = proxy_url_from_env()
    observed = []

    with requests.Session() as session:
        session.trust_env = False
        session.verify = os.getenv("APPROVED_CA_BUNDLE", True)
        for index in range(args.checks):
            try:
                result = check_exit(session, check_url, proxy_url)
            except requests.exceptions.ProxyError as exc:
                raise SystemExit("Proxy gateway or authentication failed") from exc
            except requests.exceptions.SSLError as exc:
                raise SystemExit("TLS verification failed; do not disable it") from exc
            except requests.exceptions.Timeout as exc:
                raise SystemExit("Connect or read timeout; inspect the network path") from exc
            except requests.exceptions.RequestException as exc:
                raise SystemExit(f"Transport error: {type(exc).__name__}") from exc
            except (RuntimeError, ValueError) as exc:
                raise SystemExit(str(exc)) from exc

            observed.append(result["ip"])
            print(f"check={index + 1} exit_ip={result['ip']} country={result['country']}")

    if len(observed) > 1:
        print(f"distinct_exits={len(set(observed))}/{len(observed)}")
        print("A repeated exit does not, by itself, prove that rotation is broken.")
    return observed


if __name__ == "__main__":
    main()

python rotating_proxy.py --checks 2

"""Local-only fixture: an authenticated HTTP proxy echo with two synthetic exits."""

import base64
import json
import os
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread

import rotating_proxy


class EchoProxy(BaseHTTPRequestHandler):
    request_number = 0
    total_requests = 0
    target_429_requests = 0
    slow_second_seen = 0

    def do_GET(self):
        EchoProxy.total_requests += 1
        if "/sleep" in self.path:
            time.sleep(0.3)
            return
        if "/status/429" in self.path:
            EchoProxy.target_429_requests += 1
            self.send_response(429)
            self.end_headers()
            return
        if "/slow-second-check" in self.path:
            EchoProxy.slow_second_seen += 1
            if EchoProxy.slow_second_seen == 2:
                time.sleep(0.7)

        header = self.headers.get("Proxy-Authorization", "")
        if not header.startswith("Basic "):
            self.send_response(407)
            self.send_header("Proxy-Authenticate", 'Basic realm="fixture"')
            self.end_headers()
            return

        credentials = base64.b64decode(header[6:]).decode("utf-8")
        username, _, password = credentials.partition(":")
        if password != "fixture-password":
            self.send_response(407)
            self.send_header("Proxy-Authenticate", 'Basic realm="fixture"')
            self.end_headers()
            return

        if "/bad-json" in self.path:
            body = b"{broken"
        elif "/bad-ip" in self.path:
            body = json.dumps({"ip": "not-an-ip"}).encode("utf-8")
        elif "/oversize" in self.path:
            body = json.dumps({"ip": "192.0.2.10", "pad": "x" * 5000}).encode("utf-8")
        else:
            if username.endswith("-f-1"):
                EchoProxy.request_number += 1
                ip = ["192.0.2.10", "192.0.2.11"][EchoProxy.request_number % 2]
            else:
                ip = "198.51.100.20"
            body = json.dumps({"ip": ip, "country": "fixture"}).encode("utf-8")

        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass


def run_fixture():
    server = ThreadingHTTPServer(("127.0.0.1", 0), EchoProxy)
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        os.environ.update(
            ROLA_PROXY_SCHEME="http",
            ROLA_PROXY_HOST="127.0.0.1",
            ROLA_PROXY_PORT=str(server.server_port),
            ROLA_PROXY_PASSWORD="fixture-password",
        )
        url = "http://fixture.invalid/ip.json"

        os.environ["ROLA_PROXY_USERNAME"] = "fixture-country-us-f-1"
        print("rotating fixture:")
        before = EchoProxy.total_requests
        rotating = rotating_proxy.main(["--url", url, "--checks", "2"])
        assert rotating == ["192.0.2.11", "192.0.2.10"]
        assert EchoProxy.total_requests - before == 2
        print("rotation values and call count: PASS")

        os.environ["ROLA_PROXY_USERNAME"] = "fixture_1-country-us-sessiontime-10"
        print("fixed-session fixture:")
        before = EchoProxy.total_requests
        fixed = rotating_proxy.main(["--url", url, "--checks", "2"])
        assert fixed == ["198.51.100.20", "198.51.100.20"]
        assert EchoProxy.total_requests - before == 2
        print("fixed values and call count: PASS")

        print("target rate-limit fixture:")
        before = EchoProxy.total_requests
        try:
            rotating_proxy.main(
                ["--url", "http://fixture.invalid/status/429", "--checks", "2"]
            )
        except SystemExit as exc:
            assert "stop; do not change routes" in str(exc)
            print("HTTP 429 stop rule: PASS")
        else:
            raise AssertionError("The client did not stop on HTTP 429")
        assert EchoProxy.total_requests - before == 1

        for path, expected in [
            ("bad-json", "not valid JSON"),
            ("bad-ip", "invalid IP"),
            ("oversize", "too large"),
        ]:
            try:
                rotating_proxy.main(
                    ["--url", f"http://fixture.invalid/{path}", "--checks", "1"]
                )
            except SystemExit as exc:
                assert expected in str(exc)
                print(f"{path} rejection: PASS")
            else:
                raise AssertionError(f"{path} was accepted")

        os.environ["ROLA_PROXY_PASSWORD"] = "wrong-password"
        try:
            rotating_proxy.main(["--url", url, "--checks", "1"])
        except SystemExit as exc:
            assert "authentication" in str(exc).lower()
            print("proxy authentication rejection: PASS")
        else:
            raise AssertionError("Bad proxy credentials were accepted")
        os.environ["ROLA_PROXY_PASSWORD"] = "fixture-password"

        with rotating_proxy.requests.Session() as session:
            session.trust_env = False
            try:
                rotating_proxy.check_exit(
                    session,
                    "http://fixture.invalid/sleep",
                    rotating_proxy.proxy_url_from_env(),
                    timeout=(0.1, 0.1),
                )
            except rotating_proxy.requests.exceptions.Timeout:
                print("transport timeout classification: PASS")
            else:
                raise AssertionError("Timeout was not raised")
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


if __name__ == "__main__":
    run_fixture()

python verify_local_fixture.py

rotating fixture:
check=1 exit_ip=192.0.2.11 country=fixture
check=2 exit_ip=192.0.2.10 country=fixture
distinct_exits=2/2
A repeated exit does not, by itself, prove that rotation is broken.
rotation values and call count: PASS
fixed-session fixture:
check=1 exit_ip=198.51.100.20 country=fixture
check=2 exit_ip=198.51.100.20 country=fixture
distinct_exits=1/2
A repeated exit does not, by itself, prove that rotation is broken.
fixed values and call count: PASS
target rate-limit fixture:
HTTP 429 stop rule: PASS
bad-json rejection: PASS
bad-ip rejection: PASS
oversize rejection: PASS
proxy authentication rejection: PASS
transport timeout classification: PASS

"""Minimal task-level proxy-session scheduler for authorized read-only checks.

Each task receives one route before it starts. A route never changes after a
target rejection. Deadline checks are soft admission limits, not forced network
request cancellation.
"""

import argparse
import math
import os
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from threading import Event, Lock

import requests

from rotating_proxy import (
    DEFAULT_CHECK_URL,
    STOP_STATUSES,
    check_exit,
    proxy_url_from_env,
    validate_check_url,
)


@dataclass(frozen=True)
class Task:
    label: str
    username: str
    target_url: str
    session_minutes: int
    expected_seconds: float
    session_age_seconds: float = 0.0
    diagnostic_allowance_seconds: float = 20.0
    safety_margin_seconds: float = 10.0


class TargetPacer:
    def __init__(self, gap_seconds):
        self.gap_seconds = gap_seconds
        self.next_allowed = 0.0
        self.lock = Lock()

    def wait_turn(self, stop, deadline):
        with self.lock:
            turn = max(time.monotonic(), self.next_allowed)
            self.next_allowed = turn + self.gap_seconds
        delay = turn - time.monotonic()
        if turn >= deadline or stop.wait(max(0, delay)):
            return False
        return time.monotonic() < deadline and not stop.is_set()


def run_task(task, check_url, pacer, stop, deadline):
    if stop.is_set() or time.monotonic() >= deadline:
        return task.label, "not_started", None
    if not 1 <= task.session_minutes <= 120:
        return task.label, "session_config_error", None
    estimates = (
        task.expected_seconds,
        task.session_age_seconds,
        task.diagnostic_allowance_seconds,
        task.safety_margin_seconds,
    )
    if (not all(math.isfinite(value) for value in estimates)
            or task.expected_seconds <= 0
            or any(value < 0 for value in estimates[1:])):
        return task.label, "session_config_error", None
    if sum(estimates) >= task.session_minutes * 60:
        return task.label, "session_budget_too_short", None

    task_started = time.monotonic()
    proxy_url = proxy_url_from_env(task.username)
    with requests.Session() as session:
        session.trust_env = False
        session.verify = os.getenv("APPROVED_CA_BUNDLE", True)
        try:
            if time.monotonic() >= deadline:
                return task.label, "deadline", None
            before = check_exit(session, check_url, proxy_url)["ip"]
            # Diagnostic time consumed so far reduces the declared budget.
            elapsed = time.monotonic() - task_started
            if (task.session_age_seconds + elapsed + task.expected_seconds
                    + task.diagnostic_allowance_seconds
                    + task.safety_margin_seconds >= task.session_minutes * 60):
                return task.label, "session_budget_too_short", before
            if not pacer.wait_turn(stop, deadline):
                return task.label, "not_started", before
            if stop.is_set() or time.monotonic() >= deadline:
                return task.label, "not_started", before

            with session.get(
                task.target_url,
                proxies={"http": proxy_url, "https": proxy_url},
                timeout=(5, 15),
                allow_redirects=False,
                stream=True,
            ) as response:
                status = response.status_code
            if status in STOP_STATUSES:
                stop.set()
                return task.label, f"target_stop_{status}", before
            if status == 407:
                return task.label, "proxy_auth_failure", before
            if status != 200:
                return task.label, f"target_status_{status}", before
            if time.monotonic() >= deadline:
                stop.set()
                return task.label, "deadline_after_target", before

            after = check_exit(session, check_url, proxy_url)["ip"]
            if time.monotonic() >= deadline:
                stop.set()
                return task.label, "deadline_after_verification", after
            if after != before:
                stop.set()
                return task.label, "session_exit_changed", after
            return task.label, "status_and_exit_checks_passed", after
        except requests.exceptions.ProxyError:
            return task.label, "proxy_transport_failure", None
        except requests.exceptions.Timeout:
            return task.label, "transport_timeout", None
        except requests.exceptions.SSLError:
            stop.set()
            return task.label, "tls_verification_failure", None
        except (requests.exceptions.RequestException, RuntimeError, ValueError) as exc:
            # No automatic route change or direct-connection fallback.
            if "stop; do not change routes" in str(exc):
                stop.set()
                return task.label, "diagnostic_stop_signal", None
            if "proxy authentication" in str(exc).lower():
                return task.label, "proxy_auth_failure", None
            return task.label, "diagnostic_or_transport_failure", None


def run(tasks, check_url=DEFAULT_CHECK_URL, max_workers=2, gap_seconds=1.0,
        deadline_seconds=30.0):
    if (not 1 <= max_workers <= 3
            or not math.isfinite(gap_seconds)
            or not 0 <= gap_seconds <= 60
            or not math.isfinite(deadline_seconds)
            or deadline_seconds <= 0):
        raise ValueError("Worker count, target gap, or deadline is invalid")
    validate_check_url(check_url)
    if not tasks:
        raise ValueError("At least one task is required")
    for task in tasks:
        validate_check_url(task.target_url)

    stop = Event()
    pacer = TargetPacer(gap_seconds)
    deadline = time.monotonic() + deadline_seconds
    with ThreadPoolExecutor(max_workers=max_workers) as pool:
        futures = [pool.submit(run_task, task, check_url, pacer, stop, deadline)
                   for task in tasks]
        results = [future.result() for future in futures]
    return results


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--target", required=True, help="Owned/authorized read-only URL")
    parser.add_argument("--check-url", default=DEFAULT_CHECK_URL)
    parser.add_argument("--gap-seconds", type=float, default=1.0)
    parser.add_argument("--deadline-seconds", type=float, default=30.0)
    parser.add_argument("--estimated-task-seconds", type=float, required=True,
                        help="Declared duration of each authorized target task")
    parser.add_argument("--session-age-seconds", type=float, required=True,
                        help="Declared age since this session assignment first began")
    parser.add_argument("--diagnostic-allowance-seconds", type=float, default=20.0)
    parser.add_argument("--safety-margin-seconds", type=float, default=10.0)
    args = parser.parse_args()

    usernames = [item.strip() for item in os.environ["ROLA_TASK_USERNAMES"].split(",")]
    if (not 1 <= len(usernames) <= 3 or any(not item for item in usernames)
            or len(set(usernames)) != len(usernames)):
        parser.error("ROLA_TASK_USERNAMES needs one to three distinct generated usernames")
    minutes = int(os.getenv("ROLA_SESSION_MINUTES", "10"))
    estimates = (args.estimated_task_seconds, args.session_age_seconds,
                 args.diagnostic_allowance_seconds, args.safety_margin_seconds)
    if (not all(math.isfinite(value) for value in estimates)
            or args.estimated_task_seconds <= 0
            or any(value < 0 for value in estimates[1:])):
        parser.error("Task/session estimates must be finite and non-negative")
    tasks = [Task(f"route-{index+1}", username, args.target, minutes,
                  args.estimated_task_seconds, args.session_age_seconds,
                  args.diagnostic_allowance_seconds, args.safety_margin_seconds)
             for index, username in enumerate(usernames)]
    for label, state, exit_ip in run(
        tasks, args.check_url, min(2, len(tasks)), args.gap_seconds,
        args.deadline_seconds,
    ):
        print(f"task={label} state={state} exit_ip={exit_ip or 'none'}")


if __name__ == "__main__":
    main()

python schedule_tasks.py --target https://your-owned-site.example/health --estimated-task-seconds 20 --session-age-seconds 0 --diagnostic-allowance-seconds 20 --gap-seconds 2 --deadline-seconds 60

"""Local-only assertions for task binding, pacing, stop, and session budget."""

import os
from http.server import ThreadingHTTPServer
from threading import Thread

from schedule_tasks import Task, run
from verify_local_fixture import EchoProxy


def verify():
    server = ThreadingHTTPServer(("127.0.0.1", 0), EchoProxy)
    thread = Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        os.environ.update(
            ROLA_PROXY_SCHEME="http",
            ROLA_PROXY_HOST="127.0.0.1",
            ROLA_PROXY_PORT=str(server.server_port),
            ROLA_PROXY_PASSWORD="fixture-password",
        )
        check_url = "http://fixture.invalid/ip.json"
        target_url = "http://fixture.invalid/page"
        tasks = [
            Task("route-1", "fixture_1-country-us-sessiontime-10", target_url, 10, 10),
            Task("route-2", "fixture_2-country-us-sessiontime-10", target_url, 10, 10),
        ]
        before = EchoProxy.total_requests
        results = run(tasks, check_url, max_workers=2, gap_seconds=0.1,
                      deadline_seconds=5)
        assert [state for _, state, _ in results] == [
            "status_and_exit_checks_passed", "status_and_exit_checks_passed"
        ]
        assert EchoProxy.total_requests - before == 6
        print("two bound tasks, two-worker cap, six local calls: PASS")

        restricted = [Task(task.label, task.username,
                           "http://fixture.invalid/status/429", 10, 10)
                      for task in tasks]
        before_429 = EchoProxy.target_429_requests
        results = run(restricted, check_url, max_workers=2, gap_seconds=0.5,
                      deadline_seconds=5)
        assert EchoProxy.target_429_requests - before_429 == 1
        assert "target_stop_429" in [state for _, state, _ in results]
        assert "not_started" in [state for _, state, _ in results]
        print("shared target 429 stops a second target call: PASS")

        too_short = [Task("short", tasks[0].username, target_url, 1, 55)]
        results = run(too_short, check_url, deadline_seconds=5)
        assert results[0][1] == "session_budget_too_short"
        print("short session budget rejection: PASS")

        aged = [Task("aged", tasks[0].username, target_url, 1, 20,
                     session_age_seconds=50)]
        results = run(aged, check_url, deadline_seconds=5)
        assert results[0][1] == "session_budget_too_short"
        print("declared session age and diagnostic budget rejection: PASS")

        invalid = [Task("nan", tasks[0].username, target_url, 10, float("nan"))]
        results = run(invalid, check_url, deadline_seconds=5)
        assert results[0][1] == "session_config_error"
        for kwargs in ({"gap_seconds": float("nan")},
                       {"deadline_seconds": float("nan")}):
            try:
                run(tasks, check_url, **kwargs)
            except ValueError:
                pass
            else:
                raise AssertionError("Non-finite scheduler limit was accepted")
        print("non-finite budget and scheduler limits rejected: PASS")

        EchoProxy.slow_second_seen = 0
        delayed = [Task("delayed", tasks[0].username, target_url, 10, 10)]
        results = run(delayed, "http://fixture.invalid/slow-second-check",
                      max_workers=1, gap_seconds=0, deadline_seconds=0.5)
        assert results[0][1] == "deadline_after_verification"
        print("deadline after final exit check: PASS")
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


if __name__ == "__main__":
    verify()

two bound tasks, two-worker cap, six local calls: PASS
shared target 429 stops a second target call: PASS
short session budget rejection: PASS
declared session age and diagnostic budget rejection: PASS
non-finite budget and scheduler limits rejected: PASS
deadline after final exit check: PASS

import json
from ipaddress import ip_address

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.proxy import Proxy, ProxyType

approved_jobs = [
    ("region-a", "proxy-a.example:8080"),
    ("region-b", "proxy-b.example:8080"),
]

for label, endpoint in approved_jobs:
    options = webdriver.ChromeOptions()
    options.proxy = Proxy({
        "proxyType": ProxyType.MANUAL,
        "httpProxy": endpoint,
        "sslProxy": endpoint,
    })
    driver = webdriver.Chrome(options=options)
    try:
        driver.set_page_load_timeout(15)
        driver.get("https://api64.ipify.org?format=json")
        body = driver.find_element(By.TAG_NAME, "body").text
        data = json.loads(body)  # A login/error HTML page is not an IP result.
        if not isinstance(data, dict) or not isinstance(data.get("ip"), str):
            raise ValueError("Neutral page did not return an IP string")
        observed_ip = str(ip_address(data["ip"].strip()))
        print(label, observed_ip)
    finally:
        driver.quit()