diff --git a/nice.py b/nice.py index fe468b7..77fb292 100644 --- a/nice.py +++ b/nice.py @@ -401,6 +401,8 @@ def ssl_check_cmd(url: str) -> str: Falls back to a minimal Python-based peer certificate fetch if openssl is not available. """ + from urllib.parse import urlparse + try: parsed = urlparse(url) host = parsed.hostname or url @@ -450,20 +452,74 @@ def ssl_check_cmd(url: str) -> str: logger.exception("OpenSSL check failed") return f"OpenSSL check failed: {e}\n" else: - # Fallback: use Python ssl to fetch the peer certificate (leaf only) + # Fallback: use Python ssl to fetch the peer certificate (leaf only). + # If a proxy is configured, perform an HTTP CONNECT to the proxy first + # and then do TLS over the established tunnel so the result matches the + # path used by the running Python process. import ssl + from urllib.parse import urlparse + import base64 - logger.debug("OpenSSL not found; using Python ssl fallback") + def fetch_via_proxy(proxy_url: str) -> str: + parsed = urlparse(proxy_url) + proxy_host = parsed.hostname + proxy_port = parsed.port or (443 if parsed.scheme == "https" else 80) + + auth_header = None + if parsed.username: + user = parsed.username + pwd = parsed.password or "" + auth_header = base64.b64encode(f"{user}:{pwd}".encode("utf-8")).decode("ascii") + + s = socket.create_connection((proxy_host, proxy_port), timeout=15) + try: + req = f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n" + if auth_header: + req += f"Proxy-Authorization: Basic {auth_header}\r\n" + req += "\r\n" + s.sendall(req.encode("ascii")) + + # read headers + resp = b"" + while b"\r\n\r\n" not in resp: + chunk = s.recv(4096) + if not chunk: + break + resp += chunk + if len(resp) > 65536: + break + + first_line = resp.split(b"\r\n", 1)[0].decode("ascii", errors="ignore") + if not first_line.startswith("HTTP/") or ("200" not in first_line): + raise RuntimeError(f"Proxy CONNECT failed: {first_line}") + + ctx = ssl.create_default_context() + with ctx.wrap_socket(s, server_hostname=host) as ss: + ss.settimeout(15) + der = ss.getpeercert(True) + return ssl.DER_cert_to_PEM_cert(der) + except Exception: + s.close() + raise + + logger.debug("OpenSSL not found; using Python ssl fallback (proxy-aware)") + # Prefer HTTPS_PROXY then HTTP_PROXY + proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") or os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy") try: - ctx = ssl.create_default_context() - s = socket.socket(socket.AF_INET) - with ctx.wrap_socket(s, server_hostname=host) as ss: - ss.settimeout(15) - ss.connect((host, port)) - der = ss.getpeercert(True) - pem = ssl.DER_cert_to_PEM_cert(der) - logger.debug("Python ssl fallback succeeded") - return pem + if proxy: + logger.debug(f"Using proxy for ssl check: {proxy}") + pem = fetch_via_proxy(proxy) + else: + ctx = ssl.create_default_context() + s = socket.socket(socket.AF_INET) + with ctx.wrap_socket(s, server_hostname=host) as ss: + ss.settimeout(15) + ss.connect((host, port)) + der = ss.getpeercert(True) + pem = ssl.DER_cert_to_PEM_cert(der) + + logger.debug("Python ssl fallback succeeded") + return pem except Exception as e: logger.exception("Python SSL fallback failed") return f"Python SSL fallback failed: {e}\n" diff --git a/ssl_check.py b/ssl_check.py index ab84ee4..e8934a0 100644 --- a/ssl_check.py +++ b/ssl_check.py @@ -20,6 +20,7 @@ import argparse import base64 import re import shutil +import os import subprocess import sys import tempfile @@ -43,6 +44,59 @@ def run_openssl_s_client(host: str, port: int = 443, tls_flag: str | None = None return proc.stdout + "\n" + proc.stderr +def fetch_cert_via_proxy(host: str, port: int, proxy: str, timeout: int = 20) -> List[str]: + """Connect to proxy, issue CONNECT, perform TLS handshake and return PEM of the leaf cert.""" + from urllib.parse import urlparse + import socket + import ssl + import base64 + + parsed = urlparse(proxy) + proxy_host = parsed.hostname + proxy_port = parsed.port or (443 if parsed.scheme == 'https' else 80) + + auth_header = None + if parsed.username: + user = parsed.username + pwd = parsed.password or "" + cred = f"{user}:{pwd}".encode("utf-8") + auth_header = base64.b64encode(cred).decode("ascii") + + s = socket.create_connection((proxy_host, proxy_port), timeout=timeout) + try: + connect_req = f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n" + if auth_header: + connect_req += f"Proxy-Authorization: Basic {auth_header}\r\n" + connect_req += "\r\n" + s.sendall(connect_req.encode("ascii")) + + # read response headers + resp = b"" + while b"\r\n\r\n" not in resp: + chunk = s.recv(4096) + if not chunk: + break + resp += chunk + if len(resp) > 65536: + break + + header = resp.split(b"\r\n\r\n", 1)[0].decode("ascii", errors="ignore") + # Simple status check + first_line = header.splitlines()[0] if header.splitlines() else "" + if not first_line.startswith("HTTP/") or ("200" not in first_line): + raise RuntimeError(f"Proxy CONNECT failed: {first_line}") + + # Wrap the socket with SSL and fetch peer cert (leaf) + ctx = ssl.create_default_context() + with ctx.wrap_socket(s, server_hostname=host) as ss: + der = ss.getpeercert(True) + pem = ssl.DER_cert_to_PEM_cert(der) + return [pem] + except Exception: + s.close() + raise + + def extract_pem_blocks(s: str) -> List[str]: pattern = re.compile(r"-----BEGIN CERTIFICATE-----(?:.|\n)*?-----END CERTIFICATE-----", re.M) return pattern.findall(s) @@ -127,6 +181,33 @@ def main(argv: List[str]) -> int: except Exception as e: print(f"OpenSSL check failed: {e}") + # If openssl couldn't fetch a chain (or wasn't available), attempt a proxy-aware + # fetch if HTTPS_PROXY or HTTP_PROXY is set. This helps in corporate networks + # where direct TLS is blocked and a proxy must be used. + if not extract_pem_blocks(out): + proxy = None + for key in ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy"): + proxy = os.environ.get(key) + if proxy: + break + if proxy: + try: + print(f"Attempting proxy CONNECT via {proxy}") + pems = fetch_cert_via_proxy(host, port, proxy) + if pems: + print(f"Fetched {len(pems)} PEM(s) via proxy") + server_paths = write_pems_to_temp(pems) + server_info = [inspect_cert(p) for p in server_paths] + print("Server chain (proxy fetched):") + for i, (s, iss, fp) in enumerate(server_info): + print(f"[{i}] SUBJECT: {s}") + print(f" ISSUER: {iss}") + print(f" FP: {fp}") + else: + print("Proxy CONNECT succeeded but no certs found") + except Exception as e: + print(f"Proxy CONNECT fetch failed: {e}") + # extract certs from openssl output if any pems = extract_pem_blocks(out) if pems: