Compare commits

...
32 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
Ross f0f50381ec disable native mode for frozen builds to avoid pythonnet loader issues 2025-11-24 12:40:01 +00:00
Ross 5f2bcb81a4 include pythonnet package and clr_loader in build process for runtime compatibility 2025-11-24 12:24:10 +00:00
Ross f42c1a1839 include pythonnet runtime DLLs in build process to prevent missing dependencies 2025-11-24 12:18:15 +00:00
Ross 094d66b137 rename test binary from 'anon_gui_test' to 'Uploader_test' for consistency 2025-11-24 12:15:00 +00:00
Ross 6366557d20 Merge branch 'main' of git.xkjq.uk:ross/uploader 2025-11-24 12:11:35 +00:00
Ross fa469a79fc fix build process: use absolute paths for icon folder in PyInstaller configuration to prevent runtime errors 2025-11-24 12:11:12 +00:00
Ross Kruger e93e7a8861 . 2025-11-24 11:57:58 +00:00
Ross 6db2a8fcf3 enhance build process: include dicognito package data in PyInstaller build to prevent runtime errors 2025-11-24 11:57:00 +00:00
Ross 21328b53dd refactor logging setup: improve log file path handling for development and frozen application 2025-11-24 11:48:13 +00:00
Ross eb061992a3 enhance build script and logging: add destination option to build command, log output to uploader.log, and improve artifact handling 2025-11-24 11:34:43 +00:00
Ross Kruger a8c48c12f9 v 2025-11-24 11:23:22 +00:00
Ross c735b45a25 refactor login handling: replace inline logic with perform_login function for improved clarity and maintainability 2025-11-24 10:54:42 +00:00
Ross 05c80a8fd7 remove wx and clean up deps 2025-11-24 10:51:20 +00:00
15 changed files with 1153 additions and 2072 deletions
+2
View File
@@ -14,3 +14,5 @@ test/*
*.build
C:/* # Windows paths on linux
uploader.log
-1596
View File
File diff suppressed because it is too large Load Diff
+178 -20
View File
@@ -1,46 +1,204 @@
import shutil
import importlib.util
from pathlib import Path
import PyInstaller.__main__
import typer
from rich import print
from rich.progress import Progress
import time
from datetime import datetime
app = typer.Typer()
@app.command()
def build(build: bool = True, copy: bool = True, test: bool = False):
def build(build: bool = True, copy: bool = True, test: bool = False, dest: str | None = None):
"""Windows-focused build script.
- `--test` creates a one-file test binary in `dist\test`.
- `--copy` (default) copies the artifact to the shared network path used previously.
"""
project_root = Path(__file__).resolve().parent
if build:
if test:
print("Building Windows test one-file bundle with PyInstaller")
# On Windows use ';' as add-data separator (src;dest)
# Use absolute path for the icon folder so PyInstaller can find it
icon_src = str(project_root / "icon")
add_data_args = ["--add-data", f"{icon_src};icon"]
# Try to include dicognito package data (release_notes.md) which
# PyInstaller may miss. This prevents runtime FileNotFoundError for
# dicognito/release_notes.md when the package expects the file at runtime.
try:
spec = importlib.util.find_spec("dicognito")
if spec and spec.submodule_search_locations:
dicognito_path = Path(spec.submodule_search_locations[0])
release_notes = dicognito_path / "release_notes.md"
if release_notes.exists():
# PyInstaller on Windows expects 'src;dest'
add_data_args += ["--add-data", f"{str(release_notes)};dicognito"]
else:
# include whole package folder if the single file wasn't found
add_data_args += ["--add-data", f"{str(dicognito_path)};dicognito"]
except Exception as e:
print(f"Warning: couldn't locate dicognito package: {e}")
# Try to include pythonnet runtime DLLs to avoid missing
# Python.Runtime.dll at runtime (used by pythonnet / webview).
add_binary_args = []
try:
spec_py = importlib.util.find_spec("pythonnet")
if spec_py and spec_py.submodule_search_locations:
pn_path = Path(spec_py.submodule_search_locations[0])
runtime_dir = pn_path / "runtime"
if runtime_dir.exists():
for f in runtime_dir.rglob("*.dll"):
# PyInstaller on Windows expects 'src;dest'
add_binary_args += ["--add-binary", f"{str(f)};pythonnet\\runtime"]
# also include the full pythonnet package folder as data so
# package-relative lookups succeed at runtime
add_data_args += ["--add-data", f"{str(pn_path)};pythonnet"]
# try to include clr_loader package as well
try:
spec_clr = importlib.util.find_spec("clr_loader")
if spec_clr and spec_clr.submodule_search_locations:
clr_path = Path(spec_clr.submodule_search_locations[0])
add_data_args += ["--add-data", f"{str(clr_path)};clr_loader"]
except Exception:
pass
except Exception as 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]
print("Building with PyInstaller")
PyInstaller.__main__.run([
"anon_gui.spec"
str(project_root / "nice.py"),
"--onefile",
"--name",
name_used,
"--distpath",
str(dist_dir),
"--workpath",
str(work_dir),
"--specpath",
str(spec_dir),
] + add_data_args + add_binary_args + [
# add some common hidden imports that PyInstaller sometimes misses
"--hidden-import",
"bs4",
"--hidden-import",
"requests",
"--hidden-import",
"nicegui",
])
else:
print("Building using spec: nice_uploader.spec")
PyInstaller.__main__.run([str(project_root / "nice_uploader.spec")])
if copy:
print("Copying file")
source = Path(r"C:\Users\krugerro\Desktop\uploader\dist\anon_gui.exe" )
dest =Path(r"T:\rad-tools\uploader" )
# Windows-only default destination used previously in the repo
default_dest = Path(r"\\ict\Go\RCH\Shared\Dragon-Xray\rad-tools\uploader\Uploader")
artifact_dir = project_root / "dist" / ("test" if test else "")
if not artifact_dir.exists():
artifact_dir = project_root / "dist"
# Look for a matching artifact. Prefer exe files named Uploader*.exe,
# then any .exe, then directories named Uploader* (onedir builds).
# Exclude obvious non-artifacts like log files.
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()])
n = 0
while True:
if n > 5:
break
# 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
s = n if n > 0 else ""
filename = Path(f"cris_tools{s}.exe")
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
print(f"Destination: {filename}")
try:
shutil.copy(source, dest / filename)
break
artifact = valid[0]
except PermissionError:
n = n + 1
dest_path = Path(dest) if dest else default_dest
dest_path.mkdir(parents=True, exist_ok=True)
if artifact.is_file():
target = dest_path / artifact.name
print(f"Copying {artifact} -> {target}")
shutil.copy2(artifact, target)
else:
target = dest_path / artifact.name
print(f"Copying directory {artifact} -> {target}")
if target.exists():
shutil.rmtree(target)
shutil.copytree(artifact, target)
if __name__ == "__main__":
+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);
}
}
}
+491 -57
View File
@@ -14,6 +14,7 @@ from datetime import datetime
from loguru import logger
import sys
import os
from anonymiser import Anonymizer
import pydicom
@@ -29,17 +30,29 @@ from local_file_picker import local_file_picker
from local_folder_picker import local_folder_picker
from typing import Optional
import tempfile
import glob
from fastapi import Request
from fastapi.responses import RedirectResponse
from starlette.middleware.base import BaseHTTPMiddleware
import typer
import logging
import subprocess
import shutil
from urllib.parse import urlparse
import socket
import platform
import importlib.util
import traceback
# traceback was previously imported for debugging but is unused
# TODO fix ssl
VERIFY_SSL = False
# 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
SOCKET_PATH = "tcp://localhost:9976"
@@ -72,6 +85,26 @@ DEBUG = False
logging.getLogger('niceGUI').setLevel(logging.INFO)
# Choose a log file path that works both during development and when the
# application is frozen into a single executable (PyInstaller --onefile).
try:
if getattr(sys, "frozen", False):
# 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
else:
base_path = Path(__file__).resolve().parent
except Exception:
base_path = Path.cwd()
log_path = base_path / "uploader.log"
logger.add(str(log_path), rotation="10 MB")
if DEBUG:
BASE_SITE_URL = "http://localhost:8000"
EXPORT_PATH = Path("./test/export")
@@ -119,6 +152,7 @@ async def read_nng_messages():
# This shouldn't be necessary
global SHUTDOWN
SHUTDOWN = True
logger.debug("Triggering shutdown due to socket in use")
return
# asyncio.create_task(read_nng_messages())
@@ -362,6 +396,135 @@ def clear_anonymized_files():
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):
# global rqst#, uploaded_files
@@ -449,39 +612,26 @@ def upload_files(q, rqst):
)
def perform_login(user: str, pw: str) -> dict:
"""Module-level blocking login function suitable for run.cpu_bound.
@ui.page("/login")
def login() -> Optional[RedirectResponse]:
dark = ui.dark_mode()
dark.enable()
def try_login() -> (
None
): # local function to avoid passing username and password as arguments
# Because I haven't gotten around to implementing a proper login system yet
# we just log in via requests and store teh session in the global rqst variable
user = username.value
pw = password.value
global rqst, LOGIN_SUCCESS, LOGGED_IN_USER
rqst = requests.session()
Returns a dict with keys:
- error: str on error
- session: requests.Session on success
- token: csrf token if found
- success: bool
"""
s = requests.session()
try:
rsp = rqst.get(LOGIN_URL, verify=VERIFY_SSL)
except requests.exceptions.ConnectionError as e:
ui.notify("Connection error!", color="negative")
r = s.get(LOGIN_URL, verify=VERIFY_SSL, timeout=10)
except Exception as e:
logger.debug(e)
print(traceback.format_exc())
return
return {"error": str(e)}
token = (
BeautifulSoup(rsp.content, "html.parser").find("input").attrs["value"]
) # , attr={"name": "csrfmiddlewaretoken"}).attrs("value")
# token = rsp.cookies["csrftoken"]
# header = {"X-CSRFToken": token}
# cookies = {"csrftoken": token}
try:
token = BeautifulSoup(r.content, "html.parser").find("input").attrs.get("value")
except Exception:
token = None
data = {
"username": user,
@@ -490,35 +640,53 @@ def login() -> Optional[RedirectResponse]:
"next": "/",
}
rqst.headers["Referer"] = LOGIN_URL
s.headers["Referer"] = LOGIN_URL
rsp = rqst.post(
LOGIN_URL,
data=data,
# headers=header,
# cookies=cookies
verify=VERIFY_SSL
)
print(rsp.content)
soup = BeautifulSoup(rsp.content, "html.parser")
try:
r2 = s.post(LOGIN_URL, data=data, verify=VERIFY_SSL, timeout=10)
except Exception as e:
logger.debug(e)
return {"error": str(e)}
soup = BeautifulSoup(r2.content, "html.parser")
if soup.find("button") and soup.find("button").find(string="Login"):
print("login fail")
pass
return {"session": s, "success": False}
else:
print("login success")
return {"session": s, "success": True, "token": token}
@ui.page("/login")
def login() -> Optional[RedirectResponse]:
dark = ui.dark_mode()
dark.enable()
async def try_login() -> None:
# Run blocking network calls off the UI thread to keep the page responsive.
user = username.value
pw = password.value
global rqst, LOGIN_SUCCESS, LOGGED_IN_USER
# Use the module-level perform_login function (picklable) to avoid
# multiprocessing pickling errors when using run.cpu_bound.
result = await run.cpu_bound(perform_login, user, pw)
if result.get("error"):
ui.notify("Connection error!", color="negative")
logger.debug(result.get("error"))
return
if result.get("success"):
rqst = result.get("session")
token = result.get("token")
if token:
rqst.headers["X-CSRFToken"] = token
LOGIN_SUCCESS = True
LOGGED_IN_USER = user
user_info_ui.refresh()
# user_label.clear()
# user_label.text = f"User: {user}"
# print(rqst.headers)
rqst.headers["X-CSRFToken"] = token
if LOGIN_SUCCESS:
ui.notify("Logged in!", color="positive")
ui.navigate.to("/")
else:
@@ -537,6 +705,7 @@ def login() -> Optional[RedirectResponse]:
return None
@logger.catch
@ui.page("/")
def main_page():
async def watch_for_shutdown():
@@ -561,7 +730,7 @@ def main_page():
globals(), "LOGIN_SUCCESS", backward=lambda v: not v
)
updates = ui.list()
# placeholder for update list (previously assigned but unused)
loaded_series_ui()
@@ -644,6 +813,32 @@ def main_page():
ui.button("Clear anon path", on_click=clear_anonymized_files).tooltip(
"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():
app.shutdown()
@@ -678,11 +873,16 @@ def main_page():
ui.button("About", on_click=about_dialog.open).props("outline color=white")
@logger.catch
def launch_app(
work_dir: Path = typer.Option(WORK_DIR, help="Working directory"),
nng: bool = typer.Option(True, help="Use nng"),
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
@@ -694,6 +894,111 @@ def launch_app(
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
# (which depends on pythonnet) can fail to load inside the bundled
# runtime. Disable native mode automatically for frozen builds so the
# app falls back to launching in the browser and remains usable.
if getattr(sys, "frozen", False):
if native_mode:
logger.debug("Running frozen executable: disabling native mode to avoid pythonnet loader issues")
native_mode = False
if nng:
socket_error = False
try:
@@ -701,6 +1006,102 @@ def launch_app(
pass
except pynng.exceptions.AddressInUse:
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:
with pynng.Pair0(dial=SOCKET_PATH, send_timeout=4000, recv_timeout=4000) as sub:
sub.send(b"loaded")
@@ -722,8 +1123,41 @@ def launch_app(
port = native.find_open_port()
#port = 8001
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)
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:
logger.error(e)
pass
+67
View File
@@ -0,0 +1,67 @@
# -*- mode: python ; coding: utf-8 -*-
import os
import importlib.util
from pathlib import Path
here = Path(__file__).resolve().parent
project_root = here
# Prepare datas and binaries lists; include icon resources
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 = []
# Try to include pythonnet runtime DLLs so Python.Runtime.dll is available
try:
spec_py = importlib.util.find_spec("pythonnet")
if spec_py and spec_py.submodule_search_locations:
pn_path = Path(spec_py.submodule_search_locations[0])
runtime_dir = pn_path / "runtime"
if runtime_dir.exists():
for f in runtime_dir.rglob("*.dll"):
# destination folder inside the frozen app
binaries.append((str(f), "pythonnet/runtime"))
except Exception:
# best-effort; if we can't locate pythonnet at build time, user can add it manually
pass
a = Analysis(
['nice.py'],
pathex=[str(project_root)],
binaries=binaries,
datas=datas,
hiddenimports=['bs4', 'requests', 'nicegui', 'pynng'],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='nice_uploader',
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,
)
+3
View File
@@ -0,0 +1,3 @@
import webview
webview.create_window('Hello world', 'https://pywebview.flowrl.com/hello')
webview.start()
+4 -2
View File
@@ -1,12 +1,14 @@
typer
wxpython
pydicom
dicognito
python-datauri
bs4
requests
zmq
pynng
blake3
loguru
dicom2jpg
pyinstaller
nicegui
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)
+5 -8
View File
@@ -2,19 +2,17 @@
a = Analysis(
['anon_gui.py'],
['ssl_check.py'],
pathex=[],
binaries=[],
datas=[
('icon\\icon1.ico', 'icon'),
('icon\\icon1.png', 'icon'),
],
datas=[],
hiddenimports=[],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
noarchive=False,
optimize=0,
)
pyz = PYZ(a.pure)
@@ -24,18 +22,17 @@ exe = EXE(
a.binaries,
a.datas,
[],
name='anon_gui',
name='ssl_check',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=False,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon='icon\\icon1.ico'
)
+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
-384
View File
@@ -1,384 +0,0 @@
# ===================================================================
# imViewer-Simple.py
#
# An example program that opens uncompressed DICOM images and
# converts them via numPy and PIL to be viewed in wxWidgets GUI
# apps. The conversion is currently:
#
# pydicom->NumPy->PIL->wxPython.Image->wxPython.Bitmap
#
# Gruesome but it mostly works. Surely there is at least one
# of these steps that could be eliminated (probably PIL) but
# haven't tried that yet and I may want some of the PIL manipulation
# functions.
#
# This won't handle RLE, embedded JPEG-Lossy, JPEG-lossless,
# JPEG2000, old ACR/NEMA files, or anything wierd. Also doesn't
# handle some RGB images that I tried.
#
# Have added Adit Panchal's LUT code. It helps a lot, but needs
# to be further generalized. Added test for window and/or level
# as 'list' type - crude, but it worked for a bunch of old MR and
# CT slices I have.
#
# Testing: moderate
# Tested on Windows 7 and MacOS 10.13 using numpy 1.15.2,
# Pillow 5.2.0, Python 2.7.13 / 3.6.5, and wxPython 4.0.3
#
# Originally written by Dave Witten: Nov. 11, 2009
# Updated by Aditya Panchal: Oct. 9, 2018
# ===================================================================
from io import BytesIO
import pathlib
from PIL import Image
import dicom2jpg
import pydicom
import wx
import cv2
have_PIL = True
try:
import PIL.Image
except ImportError:
have_PIL = False
have_numpy = True
try:
import numpy as np
except ImportError:
have_numpy = False
# ----------------------------------------------------------------
# Initialize image capabilities.
# ----------------------------------------------------------------
wx.InitAllImageHandlers()
def MsgDlg(window, string, caption='OFAImage', style=wx.YES_NO | wx.CANCEL):
"""Common MessageDialog."""
dlg = wx.MessageDialog(window, string, caption, style)
result = dlg.ShowModal()
dlg.Destroy()
return result
class ViewerFrame(wx.Frame):
"""Class for main window."""
def __init__(self, parent, title, initial_file=None):
"""Create the pydicom image example's main frame window."""
style = wx.DEFAULT_FRAME_STYLE | wx.SUNKEN_BORDER | wx.CLIP_CHILDREN
wx.Frame.__init__(
self,
parent,
id=-1,
title="",
pos=wx.DefaultPosition,
size=wx.Size(1024, 768),
style=style)
# --------------------------------------------------------
# Set up the menubar.
# --------------------------------------------------------
self.mainmenu = wx.MenuBar()
# Make the 'File' menu.
menu = wx.Menu()
item = menu.Append(wx.ID_ANY, '&Open...\tCtrl+O', 'Open file for editing')
self.Bind(wx.EVT_MENU, self.OnFileOpen, item)
item = menu.Append(wx.ID_ANY, 'E&xit', 'Exit Program')
self.Bind(wx.EVT_MENU, self.OnFileExit, item)
self.mainmenu.Append(menu, '&File')
# Attach the menu bar to the window.
self.SetMenuBar(self.mainmenu)
# --------------------------------------------------------
# Set up the main splitter window.
# --------------------------------------------------------
self.mainSplitter = wx.SplitterWindow(self)
self.mainSplitter.SetMinimumPaneSize(1)
# -------------------------------------------------------------
# Create the folderTreeView on the left.
# -------------------------------------------------------------
dsstyle = wx.TR_LINES_AT_ROOT | wx.TR_HAS_BUTTONS
self.dsTreeView = wx.TreeCtrl(self.mainSplitter, style=dsstyle)
# --------------------------------------------------------
# Create the ImageView on the right pane.
# --------------------------------------------------------
imstyle = wx.VSCROLL | wx.HSCROLL | wx.CLIP_CHILDREN
self.imView = wx.Panel(self.mainSplitter, style=imstyle)
self.imView.Bind(wx.EVT_PAINT, self.OnPaint)
self.imView.Bind(wx.EVT_ERASE_BACKGROUND, self.OnEraseBackground)
self.imView.Bind(wx.EVT_SIZE, self.OnSize)
# --------------------------------------------------------
# Install the splitter panes.
# --------------------------------------------------------
self.mainSplitter.SplitVertically(self.dsTreeView, self.imView)
self.mainSplitter.SetSashPosition(300, True)
# --------------------------------------------------------
# Initialize some values
# --------------------------------------------------------
self.dcmdsRoot = False
self.foldersRoot = False
self.loadCentered = True
self.bitmap = None
self.Show(True)
print(initial_file)
if initial_file is not None:
path = pathlib.Path(initial_file)
self.show_file2(path.name, str(path))
def OnFileExit(self, event):
"""Exits the program."""
self.Destroy()
event.Skip()
def OnSize(self, event):
"""Window 'size' event."""
self.Refresh()
def OnEraseBackground(self, event):
"""Window 'erase background' event."""
pass
def populateTree(self, ds):
""" Populate the tree in the left window with the [desired]
dataset values"""
if not self.dcmdsRoot:
self.dcmdsRoot = self.dsTreeView.AddRoot(text="DICOM Objects")
else:
self.dsTreeView.DeleteChildren(self.dcmdsRoot)
self.recurse_tree(ds, self.dcmdsRoot)
self.dsTreeView.ExpandAll()
def recurse_tree(self, ds, parent, hide=False):
""" order the dicom tags """
for data_element in ds:
if isinstance(data_element.value, str):
text = str(data_element)
ip = self.dsTreeView.AppendItem(parent, text=text)
else:
ip = self.dsTreeView.AppendItem(parent, text=str(data_element))
if data_element.VR == "SQ":
for i, ds in enumerate(data_element.value):
item_describe = data_element.name.replace(" Sequence", "")
item_text = "%s %d" % (item_describe, i + 1)
rjust = item_text.rjust(128)
parentNodeID = self.dsTreeView.AppendItem(ip, text=rjust)
self.recurse_tree(ds, parentNodeID)
# --- Most of what is important happens below this line ---------------------
def OnFileOpen(self, event):
"""Opens a selected file."""
msg = 'Choose a file to add.'
dlg = wx.FileDialog(self, msg, '', '', '*.*', wx.FD_OPEN)
if dlg.ShowModal() == wx.ID_OK:
fullPath = dlg.GetPath()
imageFile = dlg.GetFilename()
# checkDICMHeader()
self.show_file2(imageFile, fullPath)
def OnPaint(self, event):
"""Window 'paint' event."""
dc = wx.PaintDC(self.imView)
dc = wx.BufferedDC(dc)
# paint a background just so it isn't *so* boring.
dc.SetBackground(wx.Brush("WHITE"))
dc.Clear()
dc.SetBrush(wx.Brush("GREY", wx.CROSSDIAG_HATCH))
windowsize = self.imView.GetSize()
dc.DrawRectangle(0, 0, windowsize[0], windowsize[1])
bmpX0 = 0
bmpY0 = 0
if self.bitmap is not None:
if self.loadCentered:
bmpX0 = (windowsize[0] - self.bitmap.Width) / 2
bmpY0 = (windowsize[1] - self.bitmap.Height) / 2
dc.DrawBitmap(self.bitmap, int(bmpX0), int(bmpY0), False)
# ------------------------------------------------------------
# ViewerFrame.ConvertWXToPIL()
# Expropriated from Andrea Gavana's
# ShapedButton.py in the wxPython dist
# ------------------------------------------------------------
def ConvertWXToPIL(self, bmp):
""" Convert wx.Image Into PIL Image. """
width = bmp.GetWidth()
height = bmp.GetHeight()
im = wx.EmptyImage(width, height)
im.fromarray("RGBA", (width, height), bmp.GetData())
return im
# ------------------------------------------------------------
# ViewerFrame.ConvertPILToWX()
# Expropriated from Andrea Gavana's
# ShapedButton.py in the wxPython dist
# ------------------------------------------------------------
def ConvertPILToWX(self, pil, alpha=True):
""" Convert PIL Image Into wx.Image. """
if alpha:
image = wx.Image(pil.size[0], pil.size[1], clear=True)
image.SetData(pil.convert("RGB").tobytes())
image.SetAlpha(pil.convert("RGBA").tobytes()[3::4])
else:
image = wx.Image(pil.size[0], pil.size[1], clear=True)
new_image = pil.convert('RGB')
data = new_image.tobytes()
image.SetData(data)
return image
def get_LUT_value(self, data, window, level):
"""Apply the RGB Look-Up Table for the given
data and window/level value."""
if not have_numpy:
raise ImportError("Numpy is not available. "
"See http://numpy.scipy.org/ "
"to download and install")
win = window
lvl = level
e = [
0, 255,
lambda data: ((data - (lvl - 0.5)) / (win - 1) + 0.5) * (255 - 0)
]
return np.piecewise(data, [
data <= (lvl - 0.5 - (win - 1) / 2),
data > (lvl - 0.5 + (win - 1) / 2)
], e)
# -----------------------------------------------------------
# ViewerFrame.loadPIL_LUT(dataset)
# Display an image using the Python Imaging Library (PIL)
# -----------------------------------------------------------
def loadPIL_LUT(self, dataset):
if not have_PIL:
raise ImportError("Python Imaging Library is not available."
" See http://www.pythonware.com/products/pil/"
" to download and install")
if 'PixelData' not in dataset:
raise TypeError("Cannot show image -- "
"DICOM dataset does not have pixel data")
# can only apply LUT if these values exist
if ('WindowWidth' not in dataset) or ('WindowCenter' not in dataset):
bits = dataset.BitsAllocated
samples = dataset.SamplesPerPixel
if bits == 8 and samples == 1:
mode = "L"
elif bits == 8 and samples == 3:
mode = "RGB"
# not sure about this -- PIL source says is
# 'experimental' and no documentation.
elif bits == 16:
# Also, should bytes swap depending
# on endian of file and system??
mode = "I;16"
else:
msg = "Don't know PIL mode for %d BitsAllocated" % (bits)
msg += " and %d SamplesPerPixel" % (samples)
raise TypeError(msg)
size = (dataset.Columns, dataset.Rows)
# Recommended to specify all details by
# http://www.pythonware.com/library/pil/handbook/image.htm
im = PIL.Image.frombuffer(mode, size, dataset.PixelData, "raw",
mode, 0, 1)
else:
ew = dataset['WindowWidth']
ec = dataset['WindowCenter']
ww = int(ew.value[0] if ew.VM > 1 else ew.value)
wc = int(ec.value[0] if ec.VM > 1 else ec.value)
image = self.get_LUT_value(dataset.pixel_array, ww, wc)
# Convert mode to L since LUT has only 256 values:
# http://www.pythonware.com/library/pil/handbook/image.htm
im = PIL.Image.fromarray(image).convert('L')
return im
def show_file(self, imageFile, fullPath):
""" Load the DICOM file, make sure it contains at least one
image, and set it up for display by OnPaint(). ** be
careful not to pass a unicode string to read_file or it will
give you 'fp object does not have a defer_size attribute,
or some such."""
ds = pydicom.dcmread(str(fullPath))
# change strings to unicode
ds.decode()
self.populateTree(ds)
if 'PixelData' in ds:
self.dImage = self.loadPIL_LUT(ds)
if self.dImage is not None:
tmpImage = self.ConvertPILToWX(self.dImage, False)
self.bitmap = wx.Bitmap(tmpImage)
self.Refresh()
def show_file2(self, imageFile, fullPath):
""" Load the DICOM file, make sure it contains at least one
image, and set it up for display by OnPaint(). ** be
careful not to pass a unicode string to read_file or it will
give you 'fp object does not have a defer_size attribute,
or some such."""
ds = pydicom.dcmread(str(fullPath))
# change strings to unicode
ds.decode()
self.populateTree(ds)
with open(fullPath, "rb") as f:
img_data = dicom2jpg.io2img(BytesIO(f.read()))
img = Image.fromarray(img_data)
#img.show()
tmpImage = self.ConvertPILToWX(img, False)
self.bitmap = wx.Bitmap(tmpImage)
self.Refresh()
# ------ This is just the initialization of the App ----
# =======================================================
# The main App Class.
# =======================================================
class BasicDicomViewer(wx.App):
"""Image Application."""
def __init__(self, initial_file=None, redirect=False, filename=None, useBestVisual=False, clearSigInt=True):
self.initial_file = initial_file
super().__init__(redirect=redirect, filename=filename, useBestVisual=useBestVisual, clearSigInt=clearSigInt)
def OnInit(self):
"""Create the Image Application."""
print(1)
ViewerFrame(None, 'Simple Dicom Viewer', initial_file=self.initial_file)
return True
# ------------------------------------------------------
# If this file is running as main or a
# standalone test, begin execution here.
# ------------------------------------------------------
if __name__ == '__main__':
app = BasicDicomViewer()
app.MainLoop()