From 8924a1674747baa657c92b16ce05ff781ad0940d Mon Sep 17 00:00:00 2001 From: bonamin Date: Mon, 28 Sep 2026 17:41:24 +0300 Subject: [PATCH] feat(backend): self-managed TLS identity for the native app's encrypted LAN link services/tls_identity.py creates an EC P-256 key + self-signed cert (10y, SAN localhost/127.0.0.1/HOST_IP) under /tls at startup, and re-issues the cert with the SAME key when < 2 years remain. A cert that doesn't belong to the key is replaced. Nothing to renew by hand; a backup of the data directory keeps the identity. Runs in lifespan before the app is healthy, so the proxy (which waits for healthy) always finds the files. /api/system/identity and /api/system/status now include tls: {port: TLS_PORT (default 8443), spki_sha256} - the base64 SHA-256 of the public key that phones pin. Adds cryptography==46.0.4. Tests: create / restart (no change) / renewal 8 years later keeps the key and pin / foreign cert replaced; pin equals openssl's SPKI sha256. Co-Authored-By: Claude Opus 5.5 --- local_backend/main.py | 3 + local_backend/requirements.txt | 1 + local_backend/routers/system.py | 4 + local_backend/services/tls_identity.py | 126 +++++++++++++++++++++++++ 4 files changed, 134 insertions(+) create mode 100644 local_backend/services/tls_identity.py diff --git a/local_backend/main.py b/local_backend/main.py index 9df8129..aff8594 100644 --- a/local_backend/main.py +++ b/local_backend/main.py @@ -771,6 +771,9 @@ async def lifespan(app: FastAPI): ws_init_loop(loop) Base.metadata.create_all(bind=engine) _run_migrations() + # Before the app reports healthy: the proxy's :8443 needs these files to start + from services.tls_identity import ensure_tls_identity + ensure_tls_identity() start_print_retry_thread() pruned = prune_old_events(hours=24) if pruned: diff --git a/local_backend/requirements.txt b/local_backend/requirements.txt index a016c5d..eccd31c 100644 --- a/local_backend/requirements.txt +++ b/local_backend/requirements.txt @@ -9,3 +9,4 @@ bcrypt==4.2.0 pyjwt==2.9.0 httpx==0.27.2 python-multipart==0.0.9 +cryptography==46.0.4 diff --git a/local_backend/routers/system.py b/local_backend/routers/system.py index a5cff53..779c88f 100644 --- a/local_backend/routers/system.py +++ b/local_backend/routers/system.py @@ -20,6 +20,7 @@ from models.table import Table, TableGroup from services import printer_service from services.cloud_sync import _sync_once, _push_menu_snapshot from services.lan_ip import OVERRIDE_KEY, resolve_lan_ip, validate_lan_ip +from services.tls_identity import tls_info from middleware.license_check import license_state from config import settings @@ -50,6 +51,8 @@ def identity(db: Session = Depends(get_db)): "venue_name": (venue.value if venue and venue.value else None), "version": settings.VERSION, "api_version": API_VERSION, + # Encrypted LAN endpoint for the native app: https://:, pin spki_sha256 + "tls": tls_info(), } @@ -110,6 +113,7 @@ def system_status(db: Session = Depends(get_db), user: User = Depends(get_curren "site_id": settings.SITE_ID or None, "lan_ip": lan.get("effective"), "lan_ip_info": lan, + "tls": tls_info(), "printers": printer_statuses, } diff --git a/local_backend/services/tls_identity.py b/local_backend/services/tls_identity.py new file mode 100644 index 0000000..d8c7a86 --- /dev/null +++ b/local_backend/services/tls_identity.py @@ -0,0 +1,126 @@ +""" +The server's own TLS identity for the native app's encrypted LAN connection +(https://: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 + (/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: + """/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}