#!/usr/bin/env python3 """Check one HTTPS request stage by stage: how far did it get, and was the request sent? Standard library only. Sends one HEAD, never a body. Never prints the proxy's username or password. export HTTPS_PROXY=http://USERNAME:PASSWORD@proxy.example.net:8080 python3 ssl_eof_probe.py https://example.com/ Use --cafile if the target's certificate needs a private CA. Through an https:// proxy URL it checks the proxy's handshake and CONNECT, then stops, because it does not run TLS inside TLS. It speaks HTTP CONNECT only, so it does not test SOCKS proxies. With no proxy set it stops with an error, so it can't report a direct connection as a pass: pass --no-proxy to test from your own address. Exit status 0 means a response came back. """ import argparse import base64 import json import os import socket import ssl import sys from urllib.parse import unquote, urlsplit def read_reply(sock): reply = b"" while b"\r\n\r\n" not in reply: chunk = sock.recv(4096) if not chunk: return reply reply += chunk return reply def probe(url, proxy, cafile, proxy_cafile, timeout): target = urlsplit(url) host, port = target.hostname, target.port or 443 proxy_parts = urlsplit(proxy) if proxy else None secrets = [] report = {"target": f"{host}:{port}", "proxy": None, "stage": "start", "connect_status": None, "failed_at": None, "error": None, "request_sent": False, "http_status": None} proxy_port = None if proxy_parts: try: proxy_port = proxy_parts.port except ValueError: # its text can hold part of a password with an unencoded slash, so it is not printed proxy_port = None if not proxy_parts.hostname or not proxy_port: report.update(failed_at="proxy_url", error="the proxy URL needs a host and a port") return report if proxy_parts.scheme not in ("http", "https"): report.update(failed_at="proxy_url", error=f"this probe speaks HTTP CONNECT only, so it can't test a {proxy_parts.scheme} proxy") return report report["proxy"] = f"{proxy_parts.scheme}://{proxy_parts.hostname}:{proxy_port}" secrets = [value for value in (proxy_parts.username, proxy_parts.password) if value] secrets += [unquote(value) for value in secrets] def failed(where, exc): text = f"{type(exc).__name__}: {exc}" for secret in secrets: text = text.replace(secret, "***") report.update(failed_at=where, error=text) return report try: address = (proxy_parts.hostname, proxy_port) if proxy_parts else (host, port) sock = socket.create_connection(address, timeout=timeout) except OSError as exc: return failed("proxy_tcp" if proxy_parts else "target_tcp", exc) report["stage"] = "tcp" if proxy_parts: if proxy_parts.scheme == "https": try: context = ssl.create_default_context(cafile=proxy_cafile or cafile) sock = context.wrap_socket(sock, server_hostname=proxy_parts.hostname) except OSError as exc: return failed("proxy_tls", exc) report["stage"] = "proxy_tls" login = "" if proxy_parts.username: token = base64.b64encode(f"{unquote(proxy_parts.username)}:{unquote(proxy_parts.password or '')}".encode()).decode() login = f"Proxy-Authorization: Basic {token}\r\n" try: sock.sendall(f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n{login}\r\n".encode()) reply = read_reply(sock) except OSError as exc: return failed("connect_reply", exc) if not reply: report.update(failed_at="connect_reply", error="the proxy closed the connection without answering CONNECT") return report try: report["connect_status"] = int(reply.split(b" ", 2)[1]) except (IndexError, ValueError): report.update(failed_at="connect_reply", error="the proxy's reply is not HTTP") return report if report["connect_status"] != 200: report.update(failed_at="connect_refused", error=f"CONNECT answered {report['connect_status']}") return report report["stage"] = "tunnel" if proxy_parts.scheme == "https": report["note"] = "the target's TLS is not checked through an https:// proxy" return report try: context = ssl.create_default_context(cafile=cafile) tls = context.wrap_socket(sock, server_hostname=host) except OSError as exc: return failed("target_tls", exc) report["stage"] = "target_tls" report["tls_version"] = tls.version() path = target.path or "/" try: tls.sendall(f"HEAD {path} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: ssl-eof-probe\r\nConnection: close\r\n\r\n".encode()) report["request_sent"] = True report["stage"] = "request_written" first = tls.recv(4096) except OSError as exc: return failed("response", exc) if not first: report.update(failed_at="response", error="the connection ended before any response") return report report["stage"] = "response" try: report["http_status"] = int(first.split(b" ", 2)[1]) except (IndexError, ValueError): report.update(failed_at="response", error="the reply is not HTTP") return report def reading(report): if report["failed_at"] is None: if report["http_status"] is not None: return f"response received: HTTP {report['http_status']}" return "the tunnel opened; the request was not sent" if report["failed_at"] == "proxy_url": return f"not run: {report['error']}" if report["request_sent"]: return "unknown: the request was written, then the connection failed; it may have reached the target" return f"not sent: this attempt failed at {report['failed_at']}, before any HTTP bytes were written" def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("url") parser.add_argument("--proxy", help="proxy URL; defaults to HTTPS_PROXY or https_proxy from the environment") parser.add_argument("--no-proxy", action="store_true", help="ignore proxy settings and connect directly") parser.add_argument("--cafile") parser.add_argument("--proxy-cafile") parser.add_argument("--timeout", type=float, default=10) parser.add_argument("--json", action="store_true") args = parser.parse_args() proxy = None if args.no_proxy else args.proxy or os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") if not proxy and not args.no_proxy: print("no proxy set: pass --proxy or set HTTPS_PROXY, or pass --no-proxy to connect directly from your own address", file=sys.stderr) return 2 report = probe(args.url, proxy, args.cafile, args.proxy_cafile, args.timeout) report["reading"] = reading(report) if args.json: print(json.dumps(report)) else: for key in ("target", "proxy", "stage", "connect_status", "failed_at", "error", "request_sent", "http_status", "reading"): print(f"{key}: {report[key]}") return 0 if report["http_status"] is not None else 1 if __name__ == "__main__": sys.exit(main())