refactor(harness): renaming prodotto tht (Onda -1)
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.
This commit is contained in:
@@ -0,0 +1,114 @@
|
||||
"""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")
|
||||
Reference in New Issue
Block a user