Add proxy support for SSL certificate fetching in ssl_check utility
This commit is contained in:
Ross
@@ -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"
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user