#!/usr/bin/env python3
"""Reproduce proxy error and bypass behavior without an external proxy.

Run: python3 proxy-diagnostics-lab.py
Save results: python3 proxy-diagnostics-lab.py --output proxy-lab-results.json

Requires Python 3.9+ and curl. Requests is optional: its two timeout tests
are skipped when the package is absent. No package installation is needed
for the curl tests. Every listener and destination uses 127.0.0.1. The
proxy fixture returns controlled responses and never forwards a request.
No firewall, browser, environment or system proxy settings are changed.

These are client-behavior fixtures, not mobile-proxy service benchmarks.
Do not add production credentials or replace the loopback destinations.
"""

import argparse
import base64
import contextlib
import datetime
import http.server
import importlib.metadata
import json
import os
from pathlib import Path
import platform
import shutil
import socket
import subprocess
import tempfile
import threading
import time
import unittest
import urllib.parse
import warnings


LOOPBACK = "127.0.0.1"
OBSERVATIONS = []
CURL = shutil.which("curl")


class FixtureHandler(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, *_args):
        # Request logs are unnecessary and could contain credentials if edited.
        pass

    def do_CONNECT(self):
        expected = "Basic " + base64.b64encode(b"fixture-user:fixture-password").decode("ascii")
        authenticated = self.headers.get("Proxy-Authorization") == expected
        self.server.observations.append({
            "method": "CONNECT",
            "has_proxy_authorization": bool(self.headers.get("Proxy-Authorization")),
            "fixture_authentication_accepted": authenticated,
        })
        status = 407 if self.server.auth_required and not authenticated else self.server.connect_status
        self.send_response(status)
        if status == 407:
            self.send_header("Proxy-Authenticate", 'Basic realm="local-fixture"')
        self.send_header("Content-Length", "0")
        self.send_header("Connection", "close")
        self.end_headers()
        self.close_connection = True

    def do_GET(self):
        self.server.observations.append({"method": "GET"})
        body = self.server.kind.encode("ascii")
        if self.path == "/delayed-body":
            body = b"late"
        elif self.path == "/drip":
            body = b"123456"
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        try:
            if self.path == "/delayed-body":
                time.sleep(0.8)
            if self.path == "/drip":
                # Each gap is below the read timeout, while total duration is above it.
                for byte in body:
                    time.sleep(0.15)
                    self.wfile.write(bytes([byte]))
                    self.wfile.flush()
            else:
                self.wfile.write(body)
                self.wfile.flush()
        except (BrokenPipeError, ConnectionResetError):
            # Expected when the client reaches its timeout before the fixture sends.
            pass


@contextlib.contextmanager
def fixture(kind="origin", connect_status=502, auth_required=False):
    server = http.server.ThreadingHTTPServer((LOOPBACK, 0), FixtureHandler)
    server.daemon_threads = False
    server.kind = kind
    server.connect_status = connect_status
    server.auth_required = auth_required
    server.observations = []
    thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.02})
    thread.start()
    try:
        assert server.server_address[0] == LOOPBACK
        yield server
    finally:
        server.shutdown()
        server.server_close()  # Wait for the short-lived request threads as well.
        thread.join(timeout=3)
        if thread.is_alive():
            raise RuntimeError("Fixture server failed to stop")


def clean_proxy_environment():
    return {
        key: value for key, value in os.environ.items()
        if key.lower() not in ("http_proxy", "https_proxy", "all_proxy", "no_proxy")
    }


def curl_request(url, proxy_port=None, no_proxy="", overall_timeout=2, auth=False):
    """Run curl with a private config, controlled environment and a loopback URL."""
    parsed = urllib.parse.urlsplit(url)
    if (parsed.scheme not in ("http", "https") or parsed.hostname != LOOPBACK
            or parsed.username or parsed.password or not parsed.port):
        raise ValueError("This fixture accepts only numeric loopback URLs")
    env = clean_proxy_environment()
    env["NO_PROXY"] = no_proxy
    with tempfile.TemporaryDirectory(prefix="proxy-client-lab-") as directory:
        config = Path(directory) / "fixture.curl"
        body_file = Path(directory) / "body.txt"
        options = [
            "silent", "show-error", "connect-timeout = 1",
            "max-time = " + str(overall_timeout),
        ]
        if proxy_port is not None:
            options.append('proxy = "http://127.0.0.1:' + str(proxy_port) + '"')
        if auth:
            # These fixture values are disposable and have no access to any service.
            options.append('proxy-user = "fixture-user:fixture-password"')
        with os.fdopen(os.open(config, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as file:
            file.write("\n".join(options) + "\n")
        run = subprocess.run([
            CURL, "--disable", "--config", str(config),
            "--proto", "=http,https", "--output", str(body_file),
            "--write-out", "http=%{http_code}\nconnect=%{http_connect}\ntotal=%{time_total}\nremote=%{remote_ip}\n",
            url,
        ], env=env, capture_output=True, text=True, timeout=5, check=False)
        fields = dict(line.split("=", 1) for line in run.stdout.splitlines() if "=" in line)
        result = {
            "curl_exit": run.returncode,
            "http_status": int(fields.get("http", "0")),
            "connect_status": int(fields.get("connect", "0")),
            "elapsed_seconds": float(fields.get("total", "0")),
            "remote_ip": fields.get("remote", ""),
        }
        # The fixture body identifies the local route. Never retain curl stderr:
        # edits to an example could make error messages include sensitive values.
        body = body_file.read_text() if body_file.exists() else ""
        if body in ("origin", "proxy", "late", "123456"):
            result["fixture_body"] = body
        return result


def record(case, result, explanation):
    OBSERVATIONS.append({"case": case, "observed": result, "interpretation": explanation})


@unittest.skipUnless(CURL, "curl is required for these tests")
class CurlDiagnostics(unittest.TestCase):
    def test_connect_407(self):
        with fixture("proxy", 502, auth_required=True) as proxy:
            result = curl_request("https://127.0.0.1:1/", proxy.server_port)
            self.assertEqual(result["connect_status"], 407)
            self.assertNotEqual(result["curl_exit"], 0)
            self.assertEqual(proxy.observations[0]["method"], "CONNECT")
            self.assertFalse(proxy.observations[0]["has_proxy_authorization"])
            record("http_connect_407", result,
                   "The authentication fixture rejected CONNECT without credentials, before a destination TLS connection.")

    def test_connect_authentication_positive_control(self):
        with fixture("proxy", 502, auth_required=True) as proxy:
            result = curl_request("https://127.0.0.1:1/", proxy.server_port, auth=True)
            self.assertEqual(result["connect_status"], 502)
            self.assertNotEqual(result["curl_exit"], 0)
            self.assertTrue(proxy.observations[0]["fixture_authentication_accepted"])
            record("http_connect_authentication_positive_control", {
                **result, "fixture_authentication_accepted": True,
            }, "Valid fixture credentials passed the authentication gate. The fixture then deliberately returned 502 without forwarding; this is not a successful tunnel.")

    def test_connect_502(self):
        with fixture("proxy", 502) as proxy:
            result = curl_request("https://127.0.0.1:1/", proxy.server_port)
            self.assertEqual(result["connect_status"], 502)
            self.assertNotEqual(result["curl_exit"], 0)
            self.assertEqual(proxy.observations[0]["method"], "CONNECT")
            record("http_connect_502", result,
                   "The proxy fixture was reachable and rejected CONNECT with 502. No origin was contacted.")

    def test_connection_refused(self):
        # Select a fresh ephemeral port, then close our own socket. A socket that
        # remains bound without listening can time out rather than refuse on macOS.
        # A rare port reuse by another process makes this check fail; nothing else
        # is stopped to force the expected result.
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as unused:
            unused.bind((LOOPBACK, 0))
            port = unused.getsockname()[1]
        result = curl_request("https://127.0.0.1:1/", port)
        record("proxy_connection_refused", result,
               "curl attempted a recently closed loopback port. Exit 7 with CONNECT 0 identifies a refused connection in this run.")
        self.assertEqual(result["curl_exit"], 7)
        self.assertEqual(result["connect_status"], 0)

    def test_total_timeout_during_body(self):
        with fixture() as origin:
            result = curl_request("http://127.0.0.1:" + str(origin.server_port) + "/delayed-body",
                                  overall_timeout=0.25)
            self.assertEqual(result["curl_exit"], 28)
            self.assertEqual(result["http_status"], 200)
            self.assertEqual(result["connect_status"], 0)
            record("curl_total_timeout_after_headers", result,
                   "HTTP 200 headers arrived, but the delayed body exceeded curl's total deadline.")

    def test_no_proxy_bypasses_explicit_proxy(self):
        with fixture() as origin, fixture("proxy") as proxy:
            url = "http://127.0.0.1:" + str(origin.server_port) + "/route"
            proxied = curl_request(url, proxy.server_port)
            bypassed = curl_request(url, proxy.server_port, no_proxy=LOOPBACK)
            self.assertEqual(proxied["curl_exit"], 0)
            self.assertEqual(proxied.get("fixture_body"), "proxy")
            self.assertEqual(bypassed["curl_exit"], 0)
            self.assertEqual(bypassed.get("fixture_body"), "origin")
            self.assertEqual(len(proxy.observations), 1)
            self.assertEqual(len(origin.observations), 1)
            record("curl_no_proxy_with_explicit_proxy", {
                "empty_NO_PROXY": proxied, "NO_PROXY_127_0_0_1": bypassed,
                "proxy_requests": len(proxy.observations), "origin_requests": len(origin.observations),
            }, "For this curl version and exact host match, NO_PROXY bypassed an explicitly configured HTTP proxy.")


class RequestsTimeouts(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        try:
            # Some macOS Python builds warn about their TLS library. These tests
            # use plain HTTP on loopback and make no TLS compatibility claim.
            with warnings.catch_warnings():
                warnings.simplefilter("ignore")
                import requests
        except ImportError:
            raise unittest.SkipTest("Optional requests package is not installed")
        cls.requests = requests

    def test_read_timeout_waiting_for_body(self):
        with fixture() as origin, self.requests.Session() as client:
            client.trust_env = False
            url = "http://127.0.0.1:" + str(origin.server_port) + "/delayed-body"
            start = time.monotonic()
            caught = None
            try:
                client.get(url, timeout=(1, 0.25))
            except self.requests.exceptions.RequestException as error:
                caught = type(error).__name__
                # Requests can wrap a body-read timeout in ConnectionError.
                self.assertIn("Read timed out", str(error))
            self.assertIsNotNone(caught)
            record("requests_read_timeout", {
                "exception": caught, "read_timeout_seconds": 0.25,
                "elapsed_seconds": round(time.monotonic() - start, 6),
            }, "The fixture sent headers, then paused the body longer than the configured read timeout.")

    def test_read_timeout_is_not_total_deadline(self):
        with fixture() as origin, self.requests.Session() as client:
            client.trust_env = False
            url = "http://127.0.0.1:" + str(origin.server_port) + "/drip"
            start = time.monotonic()
            with client.get(url, timeout=(1, 0.4)) as response:
                self.assertEqual(response.content, b"123456")
            elapsed = time.monotonic() - start
            self.assertGreater(elapsed, 0.4)
            record("requests_read_timeout_not_total_deadline", {
                "http_status": response.status_code, "received_bytes": 6,
                "read_timeout_seconds": 0.4, "fixture_byte_interval_seconds": 0.15,
                "elapsed_seconds": round(elapsed, 6),
            }, "The body completed after more than 0.4 seconds because each gap between bytes stayed below the read timeout.")


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--output", type=Path, help="Also save the JSON result to this new file")
    args = parser.parse_args()
    if args.output and args.output.exists():
        parser.error("Output already exists; choose a new path to preserve previous results")
    suite = unittest.TestSuite([
        unittest.defaultTestLoader.loadTestsFromTestCase(CurlDiagnostics),
        unittest.defaultTestLoader.loadTestsFromTestCase(RequestsTimeouts),
    ])
    outcome = unittest.TextTestRunner(verbosity=2).run(suite)
    try:
        requests_version = importlib.metadata.version("requests")
    except importlib.metadata.PackageNotFoundError:
        requests_version = None
    curl_version = subprocess.run([CURL, "--version"], capture_output=True, text=True, timeout=5).stdout.splitlines()[0] if CURL else None
    result = {
        "checked_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "scope": "Controlled local client behavior, not a mobile proxy service test",
        "versions": {"python": platform.python_version(), "operating_system": platform.system(),
                     "curl": curl_version, "requests": requests_version},
        "network": {"bind_host": LOOPBACK, "destinations": "127.0.0.1 only", "forwarding": False,
                    "proxy_environment": "Cleared only in curl child processes; Requests trust_env=False",
                    "credentials": "Disposable fixture values only", "cleanup": "All fixture sockets closed"},
        "tests": {"run": outcome.testsRun, "failures": len(outcome.failures), "errors": len(outcome.errors),
                  "skipped": len(outcome.skipped), "successful": outcome.wasSuccessful(),
                  "failure_details": [{"test": item.id(), "reason": traceback.strip().splitlines()[-1]}
                                      for item, traceback in outcome.failures],
                  "error_details": [{"test": item.id(), "reason": traceback.strip().splitlines()[-1]}
                                    for item, traceback in outcome.errors],
                  "skip_details": [{"test": item.id(), "reason": reason} for item, reason in outcome.skipped]},
        "observations": OBSERVATIONS,
        "limits": [
            "curl error numbers and exception wrappers are reported for the versions used here.",
            "CONNECT fixtures reject the tunnel; successful proxy tunneling and destination TLS are not tested.",
            "Timing observations illustrate timeout behavior and are not latency benchmarks.",
            "NO_PROXY coverage is one exact IPv4 host match, not every suffix, wildcard or CIDR rule.",
            "No Chrome, SOCKS5, carrier network or production rental is tested by this script.",
            "Requests timing tests may fail on a heavily stalled machine; inspect the actual result before drawing conclusions.",
        ],
    }
    serialized = json.dumps(result, indent=2) + "\n"
    print(serialized, end="")
    if args.output:
        with args.output.open("x", encoding="utf-8") as file:
            file.write(serialized)
    return 0 if outcome.wasSuccessful() and CURL else 1


if __name__ == "__main__":
    raise SystemExit(main())