marcopan
fc5fbe6b65
Thoth (tht) è il prodotto, PSD è il cliente. Nessun riferimento al contesto
clinico nel codice.
Rinomine:
- comando+package nsp→tht (dir nsp/→tht/, 46 import, pyproject entry point)
- gate nsp-gate.js→tht-gate.js (+ rewrite token, relayIfNspFails→relayIfThtFails)
- workspace chirone.{example,test}.yaml→tht.{example,test}.yaml (generici)
- env THOTH_→THT_ (19 var) + NSP_ stragglers (NSP_HARNESS_ROOT, NSP_SESSION)
- commenti/docstring chirone/psdwp3/policlinico neutralizzati ('the reference
implementation', 'the DWH')
Aggiunto [tool.setuptools.packages.find] include=['tht*'] (necessario: l'auto-
discovery rompeva con tht/ + workspaces/ come top-level multipli).
.env operatore aggiornato in-place (prefissi THT_, valori preservati, gitignored).
Verifica: pytest 109 passed, npm test 14 pass, tht phase meta --json OK, zero
residui nsp/THOTH_/NSP_/chirone nel package.
115 lines
4.3 KiB
Python
115 lines
4.3 KiB
Python
"""Recupero della catena di certificati presentata da un endpoint HTTPS.
|
|
|
|
Serve al setup di una postazione *workstation* dietro una CA interna: scarica la
|
|
catena TLS del server REST e la salva in un bundle PEM da puntare con `THT_SSL_CA`
|
|
(consumato da `requests` via `verify=`). NON installa nulla nel trust store dell'OS.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import _ssl
|
|
import socket
|
|
import ssl
|
|
import tempfile
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
# Encoding atteso da Certificate.public_bytes() per la catena TLS non verificata.
|
|
_PEM_ENCODING = getattr(_ssl, "ENCODING_PEM", 1)
|
|
|
|
|
|
class CaFetchError(Exception):
|
|
"""Errore azionabile durante il recupero della catena CA."""
|
|
|
|
|
|
def parse_host_port(base_url: str) -> tuple[str, int]:
|
|
"""Estrae (host, port) da un URL REST https. Porta di default 443."""
|
|
parts = urlsplit(base_url)
|
|
if parts.scheme != "https":
|
|
raise CaFetchError(
|
|
f"URL non https: {base_url!r}. Il recupero CA ha senso solo su HTTPS."
|
|
)
|
|
if not parts.hostname:
|
|
raise CaFetchError(f"Host mancante nell'URL: {base_url!r}.")
|
|
return parts.hostname, parts.port or 443
|
|
|
|
|
|
def fetch_chain_pem(host: str, port: int = 443, timeout: int = 30) -> list[str]:
|
|
"""Restituisce la catena di certificati presentata da host:port come lista di PEM.
|
|
|
|
L'handshake è volutamente *non verificato* (CERT_NONE): stiamo recuperando la catena
|
|
per poter poi *stabilire* la fiducia, non per fidarci adesso. La verifica vera avviene
|
|
in seguito quando `THT_SSL_CA` punta al bundle salvato (es. `tht db ping`).
|
|
"""
|
|
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
try:
|
|
with socket.create_connection((host, port), timeout=timeout) as sock:
|
|
with ctx.wrap_socket(sock, server_hostname=host) as tls:
|
|
certs = _unverified_chain(tls)
|
|
except (OSError, ssl.SSLError) as e:
|
|
raise CaFetchError(
|
|
f"Impossibile connettersi a {host}:{port} per recuperare i certificati: {e}"
|
|
) from e
|
|
|
|
if not certs:
|
|
raise CaFetchError(
|
|
f"Nessun certificato presentato da {host}:{port}. "
|
|
f"In alternativa, estrai la catena a mano con: "
|
|
f"openssl s_client -showcerts -connect {host}:{port} -servername {host}"
|
|
)
|
|
return [_to_pem(c) for c in certs]
|
|
|
|
|
|
def _unverified_chain(tls: ssl.SSLSocket) -> list:
|
|
"""Catena presentata dal server. Metodo pubblico su Python >= 3.13, API interna su 3.12."""
|
|
public = getattr(tls, "get_unverified_chain", None)
|
|
if public is not None:
|
|
return list(public() or [])
|
|
sslobj = getattr(tls, "_sslobj", None)
|
|
getter = getattr(sslobj, "get_unverified_chain", None) if sslobj is not None else None
|
|
if getter is None:
|
|
raise CaFetchError(
|
|
"Questa versione di Python non espone la catena TLS. "
|
|
"Estrai la catena a mano con `openssl s_client -showcerts`."
|
|
)
|
|
return list(getter() or [])
|
|
|
|
|
|
def describe_pem(pem: str) -> str:
|
|
"""Riassunto leggibile (subject / issuer) di un certificato PEM, best-effort.
|
|
|
|
Serve a far riconoscere all'utente la CA interna attesa (verifica out-of-band).
|
|
Restituisce "" se il certificato non è decodificabile.
|
|
"""
|
|
try:
|
|
with tempfile.NamedTemporaryFile("w", suffix=".pem", delete=False) as fh:
|
|
fh.write(pem)
|
|
tmp = fh.name
|
|
try:
|
|
info = _ssl._test_decode_cert(tmp)
|
|
finally:
|
|
Path(tmp).unlink(missing_ok=True)
|
|
except (OSError, ssl.SSLError, ValueError):
|
|
return ""
|
|
subject = _name(info.get("subject"))
|
|
issuer = _name(info.get("issuer"))
|
|
return f"subject={subject} issuer={issuer}"
|
|
|
|
|
|
def _name(rdns) -> str:
|
|
"""Estrae il CN (o l'intero RDN) da una struttura subject/issuer di _test_decode_cert."""
|
|
if not rdns:
|
|
return "?"
|
|
parts = {k: v for rdn in rdns for (k, v) in rdn}
|
|
return parts.get("commonName") or ", ".join(f"{k}={v}" for k, v in parts.items())
|
|
|
|
|
|
def _to_pem(cert) -> str:
|
|
"""Converte un certificato (_ssl.Certificate o DER bytes) in PEM."""
|
|
if isinstance(cert, (bytes, bytearray)):
|
|
return ssl.DER_cert_to_PEM_cert(bytes(cert))
|
|
pem = cert.public_bytes(_PEM_ENCODING)
|
|
return pem if isinstance(pem, str) else pem.decode("ascii")
|