Compare commits

...
19 Commits
Author SHA1 Message Date
Ross 134bbf85b6 Add proxy support for SSL certificate fetching in ssl_check utility 2025-11-24 16:44:37 +00:00
Ross Kruger b7242d19de . 2025-11-24 16:35:05 +00:00
Ross 854fd315fa Add standalone SSL check utility with comprehensive certificate validation 2025-11-24 16:25:50 +00:00
Ross ffc4c79def Enhance SSL check command to handle protocol version issues with OpenSSL, adding retry logic for TLS1.2 support. 2025-11-24 16:23:30 +00:00
Ross Kruger 8780a409de . 2025-11-24 16:21:06 +00:00
Ross 2973e86d45 Improve artifact detection logic in build script to prioritize executable files and directories, enhancing robustness and user feedback. 2025-11-24 16:13:09 +00:00
Ross bb0d3d8af1 Enhance SSL check functionality with improved logging and UI feedback 2025-11-24 16:10:04 +00:00
Ross 9c7777323d Add SSL certificate check functionality and UI integration 2025-11-24 15:59:01 +00:00
Ross Kruger 796027a971 . 2025-11-24 15:52:31 +00:00
Ross 3680716d71 Update build process to include certificate directory for SSL verification in frozen builds 2025-11-24 15:44:43 +00:00
Ross Kruger 7528202978 . 2025-11-24 15:43:05 +00:00
Ross 377be258de include cert.pem for SSL verification in frozen builds and update build process to handle CA bundles 2025-11-24 15:30:48 +00:00
Ross 84d18f25e1 add psutil to requirements for enhanced system monitoring 2025-11-24 15:05:14 +00:00
Ross Kruger a0ee79e318 . 2025-11-24 15:04:50 +00:00
Ross 27f4aeec5e add process detection for TCP port usage and improve native mode backend checks in launch_app function 2025-11-24 14:09:06 +00:00
Ross Kruger f1b8b81c76 . 2025-11-24 14:01:02 +00:00
Ross 305e47ba8f disable native mode by default in launch_app function 2025-11-24 13:52:57 +00:00
Ross 6a49f47a5b add native mode checks and fallback for pywebview in UI launch process; create new pywebview module 2025-11-24 13:47:32 +00:00
Ross b40e802627 enhance build process to handle locked artifacts and ensure directory creation 2025-11-24 13:43:15 +00:00
12 changed files with 943 additions and 18 deletions
+85 -9
View File
@@ -4,6 +4,8 @@ from pathlib import Path
import PyInstaller.__main__ import PyInstaller.__main__
import typer import typer
from rich import print from rich import print
import time
from datetime import datetime
app = typer.Typer() app = typer.Typer()
@@ -70,17 +72,70 @@ def build(build: bool = True, copy: bool = True, test: bool = False, dest: str |
except Exception as e: except Exception as e:
print(f"Warning: couldn't locate pythonnet runtime: {e}") print(f"Warning: couldn't locate pythonnet runtime: {e}")
# Ensure dist/work/spec directories exist
dist_dir = project_root / "dist" / "test"
work_dir = project_root / "build" / "test"
spec_dir = project_root / "build" / "specs"
dist_dir.mkdir(parents=True, exist_ok=True)
work_dir.mkdir(parents=True, exist_ok=True)
spec_dir.mkdir(parents=True, exist_ok=True)
# Decide build name and try to remove any locked previous artifact.
base_name = "Uploader_test"
output_exe = dist_dir / f"{base_name}.exe"
def try_remove_with_retries(path: Path, retries: int = 5, delay: float = 0.5) -> bool:
"""Try to remove a file with retries and backoff. Returns True if removed or not present."""
if not path.exists():
return True
for attempt in range(retries):
try:
path.unlink()
print(f"Removed existing file: {path}")
return True
except PermissionError:
# Try to move/rename it out of the way
try:
backup = path.with_suffix(f".old.{int(time.time())}")
path.rename(backup)
print(f"Renamed locked file to: {backup}")
return True
except Exception:
pass
print(f"File locked, retrying in {delay} seconds ({attempt+1}/{retries})")
time.sleep(delay)
delay *= 1.5
except Exception as e:
print(f"Failed to remove existing file {path}: {e}")
return False
return False
name_used = base_name
if not try_remove_with_retries(output_exe):
# If the previous artifact is locked, fall back to a unique name to avoid build failure
ts = datetime.now().strftime("%Y%m%dT%H%M%S")
name_used = f"{base_name}_{ts}"
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",
"--name", "--name",
"Uploader_test", name_used,
"--distpath", "--distpath",
str(project_root / "dist" / "test"), str(dist_dir),
"--workpath", "--workpath",
str(project_root / "build" / "test"), str(work_dir),
"--specpath", "--specpath",
str(project_root / "build" / "specs"), str(spec_dir),
] + add_data_args + add_binary_args + [ ] + add_data_args + add_binary_args + [
# add some common hidden imports that PyInstaller sometimes misses # add some common hidden imports that PyInstaller sometimes misses
"--hidden-import", "--hidden-import",
@@ -102,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-----
+12
View File
@@ -0,0 +1,12 @@
[package]
name = "launcher"
version = "0.1.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
nng = "1.0.1"
#rfd = "0.14.1"
#toml = "0.8.13"
#serde = { version = "1.0", features = ["derive"] }
+81
View File
@@ -0,0 +1,81 @@
use nng::{Protocol, Socket, Error};
use std::process::Command;
//use rfd::MessageDialog;
//use serde::Deserialize;
//use std::fs;
//use std::path::PathBuf;
//#[derive(Debug, Deserialize)]
//struct Settings {
// cris_tools_path: PathBuf,
// port: toml::Value,
//}
//fn load_settings() -> Result<Settings, toml::de::Error> {
// // Read the entire contents of the file
// let contents = fs::read_to_string("uploader.toml")
// .expect("Failed to read settings file");
//
// // Parse the file contents and deserialize into the Settings struct
// toml::from_str(&contents)
//}
fn main() {
//let settings = match load_settings() {
// Ok(settings) => settings,
// Err(e) => {
// eprintln!("Failed to load settings: {:?}", e);
// std::process::exit(1);
// }
//};
// Create a socket of type REQ (request)
let socket = Socket::new(Protocol::Pair0).expect("Failed to create socket");
// Connect to the NNG server
//match socket.dial(format!("tcp://localhost:{}", settings.port).as_str()) {
match socket.dial(format!("tcp://localhost:9976").as_str()) {
Ok(_) => {
println!("Connected to server");
}
Err(e) => {
println!("Failed to connect to server: {:?}", e);
println!("Launching uploader tool");
let output = Command::new("Uploader.exe")
.spawn();
std::process::exit(1);
}
}
// Get the command line argument
//let arg = std::env::args().nth(1).expect("Missing argument");
//let arg = match std::env::args().nth(1) {
// Some(arg) => arg,
// None => {
// eprintln!("Missing argument");
// std::process::exit(1);
// }
//};
// Send the argument to the server
//let message = "run/".to_string() + &arg;
let message = "loaded".to_string();
println!("Send message: {}", message);
socket.send(message.as_bytes()).expect("Failed to send message");
// Receive the response from the server
match socket.recv() {
Ok(response) => {
let response = String::from_utf8(response.to_vec()).expect("Failed to receive response");
println!("Response: {}", response);
}
Err(Error::TimedOut) => {
println!("Failed to receive response: Timeout");
}
Err(e) => {
println!("Failed to receive response: {:?}", e);
}
}
}
+410 -6
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,17 +30,29 @@ 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
from starlette.middleware.base import BaseHTTPMiddleware from starlette.middleware.base import BaseHTTPMiddleware
import typer import typer
import logging import logging
import subprocess
import shutil
from urllib.parse import urlparse
import socket
import platform
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
VERIFY_SSL = False # 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
SOCKET_PATH = "tcp://localhost:9976" SOCKET_PATH = "tcp://localhost:9976"
@@ -76,7 +89,13 @@ 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
# 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 base_path = Path(sys.executable).resolve().parent
else: else:
base_path = Path(__file__).resolve().parent base_path = Path(__file__).resolve().parent
@@ -133,6 +152,7 @@ async def read_nng_messages():
# This shouldn't be necessary # This shouldn't be necessary
global SHUTDOWN global SHUTDOWN
SHUTDOWN = True SHUTDOWN = True
logger.debug("Triggering shutdown due to socket in use")
return return
# asyncio.create_task(read_nng_messages()) # asyncio.create_task(read_nng_messages())
@@ -376,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
@@ -664,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()
@@ -702,8 +877,12 @@ def main_page():
def launch_app( 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(True, 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
@@ -715,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
@@ -731,6 +1006,102 @@ def launch_app(
pass pass
except pynng.exceptions.AddressInUse: except pynng.exceptions.AddressInUse:
logger.debug("Address in use") logger.debug("Address in use")
# Try to detect which process is holding the TCP port (helpful on Windows)
try:
port_str = SOCKET_PATH.split(':')[-1]
port = int(port_str)
except Exception:
port = None
if port:
def find_process_using_port(port: int):
"""Return a list of (pid, name) tuples for processes listening on TCP port.
Uses psutil if available; otherwise falls back to parsing `netstat -ano` and
resolving PIDs with `tasklist` on Windows. Returns empty list if none found.
"""
results = []
# Prefer psutil (fast, cross-platform) when available
try:
import psutil
for conn in psutil.net_connections(kind='tcp'):
laddr = conn.laddr
if not laddr:
continue
# laddr can be an address tuple (ip, port)
try:
conn_port = laddr.port
except Exception:
# on some platforms laddr may be a string
conn_port = int(str(laddr).split(':')[-1])
if conn_port == port and conn.status in ('LISTEN', 'LISTENING'):
pid = conn.pid
name = None
try:
if pid:
p = psutil.Process(pid)
name = p.name()
except Exception:
name = None
results.append((pid, name))
return results
except Exception:
pass
# Fallback: use netstat (Windows) and tasklist to resolve PIDs
try:
if platform.system().lower().startswith('win'):
cmd = ['netstat', '-ano', '-p', 'tcp']
proc = subprocess.run(cmd, capture_output=True, text=True)
out = proc.stdout.splitlines()
pids = set()
for line in out:
parts = line.split()
# Expected format: Proto Local Address Foreign Address State PID
if len(parts) >= 5:
local = parts[1]
state = parts[3]
pid = parts[4]
# local can be like '0.0.0.0:9976' or '[::]:9976'
if ':' in local and state.upper() in ('LISTEN', 'LISTENING'):
try:
local_port = int(local.rsplit(':', 1)[1])
except Exception:
continue
if local_port == port:
try:
pid_i = int(pid)
pids.add(pid_i)
except Exception:
pass
for pid in pids:
# Resolve process name via tasklist
try:
tl = subprocess.run(['tasklist', '/FI', f'PID eq {pid}', '/FO', 'CSV', '/NH'], capture_output=True, text=True)
line = tl.stdout.strip()
if line and '"' in line:
# CSV like "process.exe","1234","..."
cols = [c.strip('"') for c in line.split(',')]
name = cols[0] if cols else None
else:
name = None
except Exception:
name = None
results.append((pid, name))
return results
except Exception:
pass
return results
holders = find_process_using_port(port)
if holders:
for pid, name in holders:
logger.error(f"Port {port} appears in use by PID={pid} name={name}")
else:
logger.debug(f"No process detected listening on port {port} via psutil/netstat")
try: try:
with pynng.Pair0(dial=SOCKET_PATH, send_timeout=4000, recv_timeout=4000) as sub: with pynng.Pair0(dial=SOCKET_PATH, send_timeout=4000, recv_timeout=4000) as sub:
sub.send(b"loaded") sub.send(b"loaded")
@@ -752,8 +1123,41 @@ def launch_app(
port = native.find_open_port() port = native.find_open_port()
#port = 8001 #port = 8001
logger.debug(f"Run on port {port}") logger.debug(f"Run on port {port}")
#ui.run(reload=False, show=True, port=port, native=native_mode, favicon="icon/icon1.ico")
# If native mode was requested, ensure pywebview has a usable GUI
# backend (Qt or GTK). Use importlib.util.find_spec to check for
# available backends without importing them (avoids side-effects).
if native_mode:
try:
if importlib.util.find_spec('webview') is None:
logger.debug('pywebview package not found; disabling native mode')
native_mode = False
else:
backend_available = False
if importlib.util.find_spec('PyQt5') or importlib.util.find_spec('PySide6'):
backend_available = True
elif importlib.util.find_spec('gi'):
backend_available = True
if not backend_available:
logger.debug('pywebview backend (Qt/GTK) not available; disabling native mode')
native_mode = False
except Exception as e:
logger.debug(f'pywebview import/check failed: {e}; disabling native mode')
native_mode = False
# Try to run with the requested mode. If native startup fails at
# runtime, catch and fall back to browser mode so the app stays usable.
try:
ui.run(reload=False, show=True, port=port, native=native_mode) ui.run(reload=False, show=True, port=port, native=native_mode)
except Exception as e:
logger.error(f"Failed to start UI in native mode: {e}")
if native_mode:
logger.debug("Retrying without native mode (browser fallback)")
try:
ui.run(reload=False, show=True, port=port, native=False)
except Exception as e2:
logger.error(f"Fallback to browser mode failed: {e2}")
except Exception as e: except Exception as e:
logger.error(e) logger.error(e)
pass pass
+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
+3
View File
@@ -0,0 +1,3 @@
import webview
webview.create_window('Hello world', 'https://pywebview.flowrl.com/hello')
webview.start()
+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