marcopan
eb3bde90e2
Ports the leaf data-layer modules and validates them: - mschema/ (models, eligibility, merge, render), db/ (connection, sampling, introspect, fetch_ca), rest/client.py -- renamed psdwp3->nsp, verbatim. - L0 (testcontainers, real Postgres): db connection read-only enforcement (psd_ro cannot CREATE/INSERT), introspect against a known schema (tables, columns, types, comments, FKs, enum, composite PK), sampling most-frequent values + truncation reporting. 15 tests, ~4s. - L1 (fake data): rest/client RPC contract (mocked transport -- X-API-Key header, payloads, base_url slash handling, HTTP/network error surfacing), mschema/render 3 formats (markdown, mschema-text, schema-dict) + eligibility rules (wide_text excluded, short_text/numeric/enum/temporal/ boolean eligible, annotation override wins). 25 tests. pyproject registers l0/l2 markers + addopts '-m not l2' (L2 opt-in). Deferred to their dependency-porting tasks: test_rrf.py (search needs vectorstore, B3) and the 11 CLI contract tests (need _guards/session, wired when each command lands). 'Not assumed reliable' now has real teeth for the data layer; CLI/search contracts follow.
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 `PSD_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 `PSD_SSL_CA` punta al bundle salvato (es. `nsp 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")
|