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 <data dir>/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 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,126 @@
|
||||
"""
|
||||
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}
|
||||
Reference in New Issue
Block a user