xenia-pos-local/local_backend/services/tls_identity.py

"""
The server's own TLS identity for the native app's encrypted LAN connection
(https://<LAN IP>:8443, plan step 7). Zero maintenance by design:

- Created automatically at backend startup if missing: an EC P-256 key and a
  self-signed certificate valid for 10 years, stored next to the database
  (<data dir>/tls/), so a backup of the data directory keeps the same identity.
- Re-issued automatically at startup when less than 2 years of validity remain
  — always with the SAME key.
- Phones pin the SHA-256 of the public key (SPKI), not the certificate, so
  renewals, expiry and IP changes never require touching a phone. Only a lost
  key (new machine without a backup) needs the waiters to re-scan the QR.

The proxy serves :8443 with these files (nginx-proxy/nginx.conf) and starts only
after the backend is healthy, i.e. after this ran.
"""
import base64
import hashlib
import ipaddress
import logging
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path

from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.x509.oid import NameOID

logger = logging.getLogger(__name__)

VALIDITY = timedelta(days=3650)
RENEW_BEFORE = timedelta(days=730)
TLS_PORT = int(os.environ.get("TLS_PORT", "8443"))

_cached_spki: str | None = None


def tls_dir() -> Path:
    """<directory of the SQLite database>/tls, overridable with TLS_DIR."""
    if os.environ.get("TLS_DIR"):
        return Path(os.environ["TLS_DIR"])
    from config import settings
    db_url = settings.DATABASE_URL
    db_path = db_url.split("sqlite:///", 1)[1] if db_url.startswith("sqlite:///") else "./pos.db"
    return Path(db_path).resolve().parent / "tls"


def spki_sha256(public_key) -> str:
    """Base64 SHA-256 of the DER SubjectPublicKeyInfo — what phones pin."""
    der = public_key.public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
    return base64.b64encode(hashlib.sha256(der).digest()).decode()


def _issue_cert(key, host_ip: str | None, now: datetime) -> x509.Certificate:
    name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "xenia-pos")])
    sans: list[x509.GeneralName] = [x509.DNSName("localhost"), x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]
    if host_ip:
        try:
            sans.append(x509.IPAddress(ipaddress.ip_address(host_ip)))
        except ValueError:
            pass
    return (
        x509.CertificateBuilder()
        .subject_name(name)
        .issuer_name(name)
        .public_key(key.public_key())
        .serial_number(x509.random_serial_number())
        .not_valid_before(now - timedelta(days=1))
        .not_valid_after(now + VALIDITY)
        .add_extension(x509.SubjectAlternativeName(sans), critical=False)
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .sign(key, hashes.SHA256())
    )


def ensure_tls_identity(directory: Path | None = None, now: datetime | None = None) -> dict:
    """Create the key/cert if missing, renew the cert (same key) if it expires
    within RENEW_BEFORE. Returns {spki_sha256, not_after, created, renewed}."""
    global _cached_spki
    directory = directory or tls_dir()
    now = now or datetime.now(timezone.utc)
    directory.mkdir(parents=True, exist_ok=True)
    key_file, cert_file = directory / "key.pem", directory / "cert.pem"
    created = renewed = False

    if key_file.exists():
        key = serialization.load_pem_private_key(key_file.read_bytes(), password=None)
    else:
        key = ec.generate_private_key(ec.SECP256R1())
        tmp = key_file.with_suffix(".tmp")
        tmp.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8,
                                          serialization.NoEncryption()))
        os.chmod(tmp, 0o600)
        os.replace(tmp, key_file)
        created = True

    cert = None
    if cert_file.exists() and not created:
        cert = x509.load_pem_x509_certificate(cert_file.read_bytes())
        if cert.public_key().public_numbers() != key.public_key().public_numbers():
            cert = None  # cert from another key — never serve a mismatched pair
    if cert is None or cert.not_valid_after_utc - now < RENEW_BEFORE:
        renewed = cert is not None
        cert = _issue_cert(key, os.environ.get("HOST_IP", "").strip() or None, now)
        tmp = cert_file.with_suffix(".tmp")
        tmp.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
        os.replace(tmp, cert_file)

    _cached_spki = spki_sha256(key.public_key())
    if created or renewed:
        logger.info("TLS identity %s (key pin %s, valid until %s)",
                    "created" if created else "renewed", _cached_spki, cert.not_valid_after_utc.date())
    return {"spki_sha256": _cached_spki, "not_after": cert.not_valid_after_utc.isoformat(),
            "created": created, "renewed": renewed}


def tls_info() -> dict | None:
    """What /api/system/identity advertises, or None if TLS isn't set up."""
    if _cached_spki is None:
        try:
            ensure_tls_identity()
        except Exception:
            logger.exception("TLS identity unavailable")
            return None
    return {"port": TLS_PORT, "spki_sha256": _cached_spki}