Add SSL certificate check functionality and UI integration
This commit is contained in:
@@ -39,6 +39,9 @@ from starlette.middleware.base import BaseHTTPMiddleware
|
|||||||
import typer
|
import typer
|
||||||
import logging
|
import logging
|
||||||
import subprocess
|
import subprocess
|
||||||
|
import shutil
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
import socket
|
||||||
import platform
|
import platform
|
||||||
import importlib.util
|
import importlib.util
|
||||||
|
|
||||||
@@ -393,6 +396,46 @@ 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.
|
||||||
|
"""
|
||||||
|
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
|
||||||
|
|
||||||
|
openssl_path = shutil.which("openssl")
|
||||||
|
if openssl_path:
|
||||||
|
cmd = [openssl_path, "s_client", "-showcerts", "-connect", f"{host}:{port}", "-servername", host]
|
||||||
|
try:
|
||||||
|
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
|
||||||
|
out = proc.stdout + "\n" + proc.stderr
|
||||||
|
return out
|
||||||
|
except Exception as e:
|
||||||
|
return f"OpenSSL check failed: {e}\n"
|
||||||
|
else:
|
||||||
|
# Fallback: use Python ssl to fetch the peer certificate (leaf only)
|
||||||
|
import ssl
|
||||||
|
|
||||||
|
try:
|
||||||
|
ctx = ssl.create_default_context()
|
||||||
|
with ctx.wrap_socket(
|
||||||
|
socket.socket(socket.AF_INET), server_hostname=host
|
||||||
|
) as s:
|
||||||
|
s.settimeout(10)
|
||||||
|
s.connect((host, port))
|
||||||
|
der = s.getpeercert(True)
|
||||||
|
pem = ssl.DER_cert_to_PEM_cert(der)
|
||||||
|
return pem
|
||||||
|
except Exception as e:
|
||||||
|
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
|
||||||
|
|
||||||
@@ -681,6 +724,16 @@ 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:
|
||||||
|
# Run the openssl/Python SSL check in a thread and show output in a dialog
|
||||||
|
output = await run.cpu_bound(ssl_check_cmd, BASE_SITE_URL)
|
||||||
|
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)
|
||||||
|
|
||||||
|
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()
|
||||||
@@ -790,13 +843,26 @@ def launch_app(
|
|||||||
if files:
|
if files:
|
||||||
# Create a temporary combined CA bundle file (persisted)
|
# Create a temporary combined CA bundle file (persisted)
|
||||||
tmp = tempfile.NamedTemporaryFile(delete=False, prefix="nice_uploader_ca_", suffix=".pem")
|
tmp = tempfile.NamedTemporaryFile(delete=False, prefix="nice_uploader_ca_", suffix=".pem")
|
||||||
|
included = []
|
||||||
try:
|
try:
|
||||||
|
import base64
|
||||||
|
|
||||||
for f in files:
|
for f in files:
|
||||||
try:
|
try:
|
||||||
with open(f, "rb") as fh:
|
with open(f, "rb") as fh:
|
||||||
tmp.write(fh.read())
|
data = fh.read()
|
||||||
# Ensure certs are separated
|
# If the file already contains a PEM block, write as-is
|
||||||
tmp.write(b"\n")
|
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:
|
except Exception:
|
||||||
# skip unreadable files
|
# skip unreadable files
|
||||||
logger.debug(f"Skipping unreadable cert file: {f}")
|
logger.debug(f"Skipping unreadable cert file: {f}")
|
||||||
@@ -804,6 +870,7 @@ def launch_app(
|
|||||||
tmp.close()
|
tmp.close()
|
||||||
VERIFY_SSL = str(Path(tmp.name))
|
VERIFY_SSL = str(Path(tmp.name))
|
||||||
logger.info(f"Using bundled CA bundle dir combined to {VERIFY_SSL}")
|
logger.info(f"Using bundled CA bundle dir combined to {VERIFY_SSL}")
|
||||||
|
logger.debug(f"Bundled cert files: {included}")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
try:
|
try:
|
||||||
tmp.close()
|
tmp.close()
|
||||||
|
|||||||
Reference in New Issue
Block a user