Compare commits

...
13 Commits
9 changed files with 659 additions and 10 deletions
+34 -5
View File
@@ -117,6 +117,14 @@ def build(build: bool = True, copy: bool = True, test: bool = False, dest: str |
name_used = f"{base_name}_{ts}" name_used = f"{base_name}_{ts}"
print(f"Previous artifact appears locked; building as {name_used} to avoid PermissionError") print(f"Previous artifact appears locked; building as {name_used} to avoid PermissionError")
# Ensure cert folder is included for test builds as well (if present)
cert_dir = project_root / "cert"
if cert_dir.exists() and cert_dir.is_dir():
arg = f"{str(cert_dir)};cert"
# Avoid adding duplicate entries
if arg not in add_data_args:
add_data_args += ["--add-data", arg]
PyInstaller.__main__.run([ PyInstaller.__main__.run([
str(project_root / "nice.py"), str(project_root / "nice.py"),
"--onefile", "--onefile",
@@ -149,13 +157,34 @@ def build(build: bool = True, copy: bool = True, test: bool = False, dest: str |
if not artifact_dir.exists(): if not artifact_dir.exists():
artifact_dir = project_root / "dist" artifact_dir = project_root / "dist"
# Look for a matching artifact # Look for a matching artifact. Prefer exe files named Uploader*.exe,
candidates = list(artifact_dir.glob("Uploader*")) + list(artifact_dir.glob("*.exe")) # then any .exe, then directories named Uploader* (onedir builds).
if not candidates: # Exclude obvious non-artifacts like log files.
print(f"No build artifact found in {artifact_dir}, skipping copy") candidates = []
# Uploader*.exe (preferred)
candidates.extend(sorted(artifact_dir.glob("Uploader*.exe")))
# any .exe
candidates.extend(sorted(artifact_dir.glob("*.exe")))
# onedir: directories starting with Uploader
candidates.extend([p for p in sorted(artifact_dir.glob("Uploader*")) if p.is_dir()])
# Filter out files that look like logs or non-executables
def is_valid_artifact(p: Path) -> bool:
if not p.exists():
return False
if p.is_file():
# require .exe for files (Windows builds)
return p.suffix.lower() == ".exe"
if p.is_dir():
return True
return False
valid = [p for p in candidates if is_valid_artifact(p)]
if not valid:
print(f"No build artifact (.exe or Uploader directory) found in {artifact_dir}, skipping copy")
return return
artifact = candidates[0] artifact = valid[0]
dest_path = Path(dest) if dest else default_dest dest_path = Path(dest) if dest else default_dest
dest_path.mkdir(parents=True, exist_ok=True) dest_path.mkdir(parents=True, exist_ok=True)
+21
View File
@@ -0,0 +1,21 @@
-----BEGIN CERTIFICATE-----
MIIDhzCCAm+gAwIBAgIQHLnU4uDoKaJAaqa5Yi/NoTANBgkqhkiG9w0BAQUFADBW
MQswCQYDVQQGEwJHQjERMA8GA1UECBMIQ29ybndhbGwxFTATBgNVBAoTDENvcm53
YWxsIE5IUzEdMBsGA1UEAxMUQ29ybndhbGwgSUNUIFJvb3QgQ0EwHhcNMTMwNjI2
MDkwNjExWhcNMzMwNjI2MDkxNjA1WjBWMQswCQYDVQQGEwJHQjERMA8GA1UECBMI
Q29ybndhbGwxFTATBgNVBAoTDENvcm53YWxsIE5IUzEdMBsGA1UEAxMUQ29ybndh
bGwgSUNUIFJvb3QgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCm
m8jF64E3wkxhf2BLfD3lT3v9RebGhJ684jtv0A6qL2BTe/Co4SJ/obYiDw82K4Vw
1O+jwIg77hVrH/YaCsjPiefxxwL/T0bSURlXq5Q8pCQaHjd1iH1JNY1WfW0Ywkni
a7xbm6/2+zwsyTHfhuSU+zmSQxejOX+Ffz6MWZSb+JjppOH6OBKTHGw69JIXcAUu
wT3GSAdYyYKgP6i6jfsEeJkq+s/hd6cVxFWPrWJJaGtfIvLZT38vZeitYDYT9Ad0
gIQPDyLH35tuNszFWkU5DA/U6xXvWkAV7Mq/EaGSVHckGeQiGAQNmV74KugqpcC/
9reOFKEbc6qvDdGaASjZAgMBAAGjUTBPMAsGA1UdDwQEAwIBhjAPBgNVHRMBAf8E
BTADAQH/MB0GA1UdDgQWBBShR/WOiN4mvXi2tYLGtbeAaCXE7DAQBgkrBgEEAYI3
FQEEAwIBADANBgkqhkiG9w0BAQUFAAOCAQEAEMca1BLnKJIXRZhJUTfLrSWtQDgV
T1jGnHFKngPxWUB1u0fVwpnqeuHSWQ2qkNvEZoBQFbJhEny83r8thLamJB1UHEyj
yhWrtjif7dne3LwzLtmiMZIKPjeyCa+zzl5YW+lUenQErl48PC7sJc5gq4lZajOo
fsEtw5zZLWFqXE/KJNt4vrGhQ0SplphbOzau7VrFIFciiKHHHw3rnV5byjqYDy/Z
IJytye488Pi75lBKq8pWW6YbUFmKly5bqF6Q8oicd9y0/9GGmTZGrzp1WbspYF9Y
lc2p6bkhfQ1f7Q+8/sF0D/sGfjfo099OlRhc881Mwt5p8mEZtJTxk3SZlQ==
-----END CERTIFICATE-----
+19
View File
@@ -0,0 +1,19 @@
-----BEGIN CERTIFICATE-----
MIIDJjCCAg6gAwIBAgIIUW6l3kYeVMEwDQYJKoZIhvcNAQELBQAwMTEOMAwGA1UE
ChMFQ2lzY28xHzAdBgNVBAMTFkNpc2NvIFVtYnJlbGxhIFJvb3QgQ0EwHhcNMTYw
NjI4MTUzNzUzWhcNMzYwNjI4MTUzNzUzWjAxMQ4wDAYDVQQKEwVDaXNjbzEfMB0G
A1UEAxMWQ2lzY28gVW1icmVsbGEgUm9vdCBDQTCCASIwDQYJKoZIhvcNAQEBBQAD
ggEPADCCAQoCggEBAO7ZjfBSCaz5EMYSiWYoXjHPP/w7xFT4bXa82lOZ9CJJXDQw
bZpBdmuqX9UWo769LIAaSUvkYEeZqcTsjrx/7juPKoOErhJY0cPK12LU9PbHXqEd
XESIqBjdOC5oiIFHhTAKuuKRlL7rhPYkYhZtgdll4h0FLIG+xNsMVfzJb7z69X8Y
vF9r1drLkd7oR2xHuRkXgzeblFVpF+DRF7WXNhLy0By38ZxtClxYUSitdz53W0ic
maelG7EyCVNVxARxn5waaphRvki1hkuqqrm3JdlV165zAOdSz3JKzRISQinCTQuT
+RK/w0qLsDTyOVO/mEIVWLXu/Z1NtuXgj/jhegcCAwEAAaNCMEAwDgYDVR0PAQH/
BAQDAgEGMA8GA1UdEwEB/wQFMAMBAf8wHQYDVR0OBBYEFENzAN4kukAaQFQsfXzV
AEiJDHCkMA0GCSqGSIb3DQEBCwUAA4IBAQBIEoceSPZLmo5sLmgDfQA+Fq5BKztL
qg8aAvZdrbdMEKEBr1RDB0OAhuPcaaVxZi6Hjyql1N999Zmp8qIw/lLTt3VSTmEa
29uPgjdMGLl9KyfZjARiA/PPvPdHTwg7TMJOet+w7P5nWabLNW55+Wc/JzCSFE30
+0Kdz/jojxlA/8t0xYLCdS2UK7zC4kuAbojHLJDbIQO3HeEWwVmg4FO89AHVvC4R
Y+V0t7SaEradv6tPG9DHX7PLwjQ/Xs95NGDIJTeFwCRqYUlBu9iZjIvKba0e0tST
Vuyw2+P2HuWazjBPawGrbfyw+uO3KO4WnNGjMutJJ920o8B5M8gW1+Ye
-----END CERTIFICATE-----
+275 -4
View File
@@ -14,6 +14,7 @@ from datetime import datetime
from loguru import logger from loguru import logger
import sys import sys
import os
from anonymiser import Anonymizer from anonymiser import Anonymizer
import pydicom import pydicom
@@ -29,6 +30,8 @@ from local_file_picker import local_file_picker
from local_folder_picker import local_folder_picker from local_folder_picker import local_folder_picker
from typing import Optional from typing import Optional
import tempfile
import glob
from fastapi import Request from fastapi import Request
from fastapi.responses import RedirectResponse from fastapi.responses import RedirectResponse
@@ -36,12 +39,19 @@ from starlette.middleware.base import BaseHTTPMiddleware
import typer import typer
import logging import logging
import subprocess import subprocess
import shutil
from urllib.parse import urlparse
import socket
import platform import platform
import importlib.util import importlib.util
# traceback was previously imported for debugging but is unused # traceback was previously imported for debugging but is unused
# TODO fix ssl # SSL verification setting. Can be a boolean (True/False) or a path to
# a CA bundle (PEM file) used by requests' 'verify' parameter.
# Default is True. It can be overridden via CLI options (--ca-bundle,
# --insecure) or environment variables: NICE_UPLOADER_CA_BUNDLE,
# REQUESTS_CA_BUNDLE or SSL_CERT_FILE.
VERIFY_SSL = True VERIFY_SSL = True
SOCKET_PATH = "tcp://localhost:9976" SOCKET_PATH = "tcp://localhost:9976"
@@ -79,8 +89,14 @@ logging.getLogger('niceGUI').setLevel(logging.INFO)
# application is frozen into a single executable (PyInstaller --onefile). # application is frozen into a single executable (PyInstaller --onefile).
try: try:
if getattr(sys, "frozen", False): if getattr(sys, "frozen", False):
# When frozen, place logs next to the executable # When frozen by PyInstaller, data files are extracted to a temporary
base_path = Path(sys.executable).resolve().parent # folder available as sys._MEIPASS. Fall back to the executable
# directory when _MEIPASS is not present.
meipass = getattr(sys, "_MEIPASS", None)
if meipass:
base_path = Path(meipass)
else:
base_path = Path(sys.executable).resolve().parent
else: else:
base_path = Path(__file__).resolve().parent base_path = Path(__file__).resolve().parent
except Exception: except Exception:
@@ -380,6 +396,135 @@ def clear_anonymized_files():
return return
def ssl_check_cmd(url: str) -> str:
"""Run openssl s_client -showcerts against the given URL host and return output.
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
port = parsed.port or (443 if parsed.scheme in ("https", "") else 80)
except Exception:
host = url
port = 443
logger.debug(f"ssl_check_cmd: host={host} port={port}")
openssl_path = shutil.which("openssl")
if openssl_path:
cmd = [openssl_path, "s_client", "-showcerts", "-connect", f"{host}:{port}", "-servername", host]
logger.debug(f"Running openssl: {' '.join(cmd)}")
try:
# Provide a small stdin so s_client can complete instead of
# waiting for interactive input. Capture both stdout and stderr.
proc = subprocess.run(cmd, input='\n', capture_output=True, text=True, timeout=60)
out = proc.stdout + "\n" + proc.stderr
logger.debug(f"OpenSSL finished, {len(out)} bytes returned, rc={proc.returncode}")
# Some older OpenSSL builds (or default s_client options) may
# negotiate an incompatible protocol version with modern servers
# and return a short error like 'tlsv1 alert protocol version'. If
# that appears, try again forcing TLS1.2 which most servers still
# support. This helps when the openssl on PATH is old (eg bundled
# with other software) and defaults to legacy negotiation.
if "tlsv1 alert protocol version" in out or "SSL23_GET_SERVER_HELLO" in out:
logger.debug("OpenSSL reported protocol version issue; retrying with -tls1_2")
try:
cmd2 = cmd + ["-tls1_2"]
proc2 = subprocess.run(cmd2, input='\n', capture_output=True, text=True, timeout=30)
out2 = proc2.stdout + "\n" + proc2.stderr
logger.debug(f"OpenSSL retry finished, {len(out2)} bytes returned, rc={proc2.returncode}")
return out + "\n--- retry with -tls1_2 ---\n" + out2
except Exception:
logger.exception("OpenSSL tls1_2 retry failed")
return out + "\nNote: OpenSSL appears to be too old to negotiate modern TLS.\nConsider installing a newer OpenSSL (eg Git for Windows or a Win64 OpenSSL build).\n"
return out
except subprocess.TimeoutExpired:
logger.debug("OpenSSL timed out")
return (
"OpenSSL check timed out after 60s\n"
"This can happen if the connection is blocked by a proxy or firewall. "
"Try running the openssl command manually or check your network/proxy settings.\n"
)
except Exception as e:
logger.exception("OpenSSL check failed")
return f"OpenSSL check failed: {e}\n"
else:
# 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
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:
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"
def upload_files(q, rqst): def upload_files(q, rqst):
# global rqst#, uploaded_files # global rqst#, uploaded_files
@@ -668,6 +813,32 @@ def main_page():
ui.button("Clear anon path", on_click=clear_anonymized_files).tooltip( ui.button("Clear anon path", on_click=clear_anonymized_files).tooltip(
"Delete all files in ANON_PATH" "Delete all files in ANON_PATH"
) )
async def ssl_check_ui() -> None:
# Show a running dialog immediately so user sees progress
with ui.dialog() as running_dlg:
with ui.card().classes("p-4"):
ui.label("SSL check running...").classes("text-h6")
running_dlg.open()
try:
logger.debug("Starting SSL check via UI")
output = await run.cpu_bound(ssl_check_cmd, BASE_SITE_URL)
except Exception as e:
logger.exception("SSL check failed")
running_dlg.close()
ui.notify(f"SSL check failed: {e}", color="negative")
return
# close the running dialog and show the results in a new dialog
running_dlg.close()
with ui.dialog() as d:
with ui.card().classes("p-4"):
ui.label("SSL Check output").classes("text-h6")
ui.code(output).props("white-space: pre-wrap")
ui.button("Close", on_click=d.close)
d.open()
ui.button("SSL check", on_click=ssl_check_ui).tooltip("Run openssl s_client -showcerts against the configured site and show output")
def shutdown_app(): def shutdown_app():
app.shutdown() app.shutdown()
@@ -707,7 +878,11 @@ def launch_app(
work_dir: Path = typer.Option(WORK_DIR, help="Working directory"), work_dir: Path = typer.Option(WORK_DIR, help="Working directory"),
nng: bool = typer.Option(True, help="Use nng"), nng: bool = typer.Option(True, help="Use nng"),
native_mode: bool = typer.Option(False, help="Use native mode"), native_mode: bool = typer.Option(False, help="Use native mode"),
debug: bool = typer.Option(False, help="Debug mode") debug: bool = typer.Option(False, help="Debug mode"),
ca_bundle: Optional[Path] = typer.Option(
None, help="Path to a CA bundle (PEM file) to verify HTTPS connections"
),
insecure: bool = typer.Option(False, help="Disable SSL verification (insecure)"),
): ):
global WORK_DIR, EXPORT_PATH, PROCESS_PATH, ANON_PATH, DEBUG global WORK_DIR, EXPORT_PATH, PROCESS_PATH, ANON_PATH, DEBUG
@@ -719,6 +894,102 @@ def launch_app(
logger.debug(f"Work dir: {WORK_DIR}") logger.debug(f"Work dir: {WORK_DIR}")
# Configure SSL verification behaviour. VERIFY_SSL can be:
# - True (default): use system certs
# - False: disable verification (insecure)
# - path (str): path to CA bundle PEM file used by requests
global VERIFY_SSL
env_ca = (
os.environ.get("NICE_UPLOADER_CA_BUNDLE")
or os.environ.get("REQUESTS_CA_BUNDLE")
or os.environ.get("SSL_CERT_FILE")
)
if insecure:
VERIFY_SSL = False
logger.warning(
"SSL verification disabled (insecure). Only use for troubleshooting in trusted networks."
)
elif ca_bundle is not None:
VERIFY_SSL = str(ca_bundle)
logger.info(f"Using CA bundle from CLI: {VERIFY_SSL}")
elif env_ca:
VERIFY_SSL = env_ca
logger.info(f"Using CA bundle from environment: {VERIFY_SSL}")
else:
VERIFY_SSL = True
# If verification is disabled, suppress urllib3 insecure warnings to keep logs clean
if VERIFY_SSL is False:
try:
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
except Exception:
pass
# If a bundled cert.pem was included in the frozen app, prefer that
# when the user hasn't explicitly provided a CA bundle or disabled
# verification. This lets builders ship a corporate CA with the app.
try:
# Prefer an entire bundled cert/ folder (so users can include many
# certificate files). If present, concatenate .pem/.cer/.crt files
# into a temporary combined bundle and use that. Otherwise fall back
# to a single cert.pem file if present.
bundled_dir = base_path / "cert"
if bundled_dir.exists() and bundled_dir.is_dir() and VERIFY_SSL is True:
# Collect candidate cert files
patterns = ["*.pem", "*.cer", "*.crt"]
files = []
for p in patterns:
files.extend(sorted(bundled_dir.glob(p)))
if files:
# Create a temporary combined CA bundle file (persisted)
tmp = tempfile.NamedTemporaryFile(delete=False, prefix="nice_uploader_ca_", suffix=".pem")
included = []
try:
import base64
for f in files:
try:
with open(f, "rb") as fh:
data = fh.read()
# If the file already contains a PEM block, write as-is
if b"-----BEGIN CERTIFICATE-----" in data:
tmp.write(data)
tmp.write(b"\n")
included.append(str(f))
else:
# Assume DER/binary and convert to PEM by base64-encoding
b64 = base64.encodebytes(data)
tmp.write(b"-----BEGIN CERTIFICATE-----\n")
tmp.write(b64)
tmp.write(b"-----END CERTIFICATE-----\n")
included.append(str(f) + " (converted from DER)")
except Exception:
# skip unreadable files
logger.debug(f"Skipping unreadable cert file: {f}")
tmp.flush()
tmp.close()
VERIFY_SSL = str(Path(tmp.name))
logger.info(f"Using bundled CA bundle dir combined to {VERIFY_SSL}")
logger.debug(f"Bundled cert files: {included}")
except Exception as e:
try:
tmp.close()
except Exception:
pass
logger.debug(f"Failed to build combined CA bundle: {e}")
else:
bundled = base_path / "cert.pem"
if bundled.exists() and VERIFY_SSL is True:
VERIFY_SSL = str(bundled)
logger.info(f"Using bundled CA bundle at {VERIFY_SSL}")
except Exception:
pass
# When running as a frozen executable, the native/webview backend # When running as a frozen executable, the native/webview backend
# (which depends on pythonnet) can fail to load inside the bundled # (which depends on pythonnet) can fail to load inside the bundled
# runtime. Disable native mode automatically for frozen builds so the # runtime. Disable native mode automatically for frozen builds so the
+6 -1
View File
@@ -8,7 +8,12 @@ here = Path(__file__).resolve().parent
project_root = here project_root = here
# Prepare datas and binaries lists; include icon resources # Prepare datas and binaries lists; include icon resources
datas = [(str(project_root / 'icon'), 'icon'), (str(project_root / 'icon' / 'icon1.png'), 'icon')] datas = [
(str(project_root / 'icon'), 'icon'),
(str(project_root / 'icon' / 'icon1.png'), 'icon'),
# include user-supplied CA bundle folder (all files) so frozen app can use it
(str(project_root / 'cert'), 'cert')
]
binaries = [] binaries = []
# Try to include pythonnet runtime DLLs so Python.Runtime.dll is available # Try to include pythonnet runtime DLLs so Python.Runtime.dll is available
+1
View File
@@ -11,3 +11,4 @@ dicom2jpg
pyinstaller pyinstaller
nicegui nicegui
pywebview pywebview
psutil
+259
View File
@@ -0,0 +1,259 @@
#!/usr/bin/env python3
"""Standalone SSL check utility to be frozen for testing.
Usage:
python ssl_check.py --url https://www.penracourses.org.uk --cert-dir ./cert
What it does:
- Runs `openssl s_client -showcerts` against the host (if openssl on PATH).
- Extracts PEM cert blocks from the output and inspects each with `openssl x509`.
- Loads cert files from --cert-dir (PEM or DER) and converts DER to PEM as needed.
- Compares server chain Issuers with Subjects of bundled certs and reports which CA certs are missing.
This script is intentionally small and self-contained so it can be frozen with PyInstaller
for quick testing on Windows where the system openssl may be too old.
"""
from __future__ import annotations
import argparse
import base64
import re
import shutil
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import List, Tuple
def find_openssl() -> str | None:
return shutil.which("openssl")
def run_openssl_s_client(host: str, port: int = 443, tls_flag: str | None = None, timeout: int = 60) -> str:
openssl = find_openssl()
if not openssl:
raise RuntimeError("openssl not found on PATH")
cmd = [openssl, "s_client", "-showcerts", "-connect", f"{host}:{port}", "-servername", host]
if tls_flag:
cmd.append(tls_flag)
# give a newline so s_client won't wait for interactive input
proc = subprocess.run(cmd, input="\n", capture_output=True, text=True, timeout=timeout)
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)
def write_pems_to_temp(pems: List[str]) -> List[Path]:
out = []
for i, pem in enumerate(pems):
f = Path(tempfile.gettempdir()) / f"ssl_check_cert_{i}.pem"
f.write_text(pem)
out.append(f)
return out
def inspect_cert(path: Path) -> Tuple[str, str, str]:
"""Return (subject, issuer, fingerprint) by calling openssl x509."""
openssl = find_openssl()
if not openssl:
raise RuntimeError("openssl not found on PATH")
proc = subprocess.run([openssl, "x509", "-in", str(path), "-noout", "-subject", "-issuer", "-fingerprint"], capture_output=True, text=True)
out = proc.stdout + proc.stderr
subj = re.search(r"subject=\s*(.*)", out)
issu = re.search(r"issuer=\s*(.*)", out)
fp = re.search(r"Fingerprint=([A-F0-9:]+)", out)
return (subj.group(1).strip() if subj else "", issu.group(1).strip() if issu else "", fp.group(1).strip() if fp else "")
def load_cert_files_from_dir(cert_dir: Path) -> List[Path]:
files = []
if not cert_dir.exists():
return files
for pattern in ("*.pem", "*.crt", "*.cer"):
files.extend(sorted(cert_dir.glob(pattern)))
out = []
for f in files:
data = f.read_bytes()
if b"-----BEGIN CERTIFICATE-----" in data:
# already PEM
out.append(f)
continue
# assume DER -> convert to PEM
b64 = base64.encodebytes(data).decode("ascii")
pem = "-----BEGIN CERTIFICATE-----\n" + b64 + "-----END CERTIFICATE-----\n"
tmp = Path(tempfile.gettempdir()) / f"ssl_check_bundle_{f.name}.pem"
tmp.write_text(pem)
out.append(tmp)
return out
def main(argv: List[str]) -> int:
p = argparse.ArgumentParser()
p.add_argument("--url", required=True, help="URL or host to check (eg https://www.penracourses.org.uk)")
p.add_argument("--cert-dir", default="cert", help="Directory with bundled certs")
p.add_argument("--force-tls12", action="store_true", help="Force -tls1_2 on openssl if initial attempt failed")
args = p.parse_args(argv)
# parse host/port
from urllib.parse import urlparse
parsed = urlparse(args.url if "//" in args.url else "//" + args.url)
host = parsed.hostname or args.url
port = parsed.port or 443
openssl = find_openssl()
if not openssl:
print("openssl not found on PATH; please install Git for Windows or OpenSSL and re-run.")
# we still attempt Python fallback but chain won't be available
else:
print(f"Using openssl: {openssl}")
out = ""
if openssl:
try:
out = run_openssl_s_client(host, port)
if "tlsv1 alert protocol version" in out or "SSL23_GET_SERVER_HELLO" in out:
print("Server rejected legacy TLS; retrying with -tls1_2")
out2 = run_openssl_s_client(host, port, tls_flag="-tls1_2")
out = out + "\n--- retry result ---\n" + out2
except subprocess.TimeoutExpired:
print("OpenSSL check timed out")
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:
print(f"Found {len(pems)} PEM blocks in openssl output")
server_paths = write_pems_to_temp(pems)
server_info = [inspect_cert(p) for p in server_paths]
print("Server chain:")
for i, (s, iss, fp) in enumerate(server_info):
print(f"[{i}] SUBJECT: {s}")
print(f" ISSUER: {iss}")
print(f" FP: {fp}")
else:
print("No certificate chain found from openssl (server may have rejected the handshake or openssl is incompatible)")
server_info = []
# load bundled certs
cert_dir = Path(args.cert_dir)
bundled = load_cert_files_from_dir(cert_dir)
print(f"Found {len(bundled)} bundled cert files in {cert_dir}")
bundled_info = [inspect_cert(p) for p in bundled]
subjects = {info[0]: p for info, p in zip(bundled_info, bundled)}
# Compare: ensure each server issuer is present in bundled subjects
missing = set()
for s, iss, fp in server_info:
if iss and iss not in subjects:
missing.add(iss)
if not server_info:
print("No server certs to compare; check network/openssl compatibility")
return 2
if not missing:
print("All server issuers are present in the bundled certs. OK")
return 0
else:
print("Missing issuer certificates in bundled certs:")
for m in missing:
print(" - ", m)
return 3
if __name__ == "__main__":
try:
rc = main(sys.argv[1:])
except Exception as e:
print(f"ssl_check failed: {e}")
rc = 1
sys.exit(rc)
+38
View File
@@ -0,0 +1,38 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['ssl_check.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='ssl_check',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
+6
View File
@@ -0,0 +1,6 @@
PS T:\rad-tools\uploader\Uploader_test> .\ssl_check.exe --url http://www.penracourses.org.uk --cert-dir ./cert
Using openssl: C:\Sybase\OCS-15_0\bin\openssl.EXE
Server rejected legacy TLS; retrying with -tls1_2
No certificate chain found from openssl (server may have rejected the handshake or openssl is incompatible)
Found 2 bundled cert files in cert
No server certs to compare; check network/openssl compatibility