feat(cloud): heartbeat returns a signed license token (Ed25519)

Sites now enforce their license offline (client-services KI-006): the
cloud is needed to renew a license, not to run it. Each heartbeat carries
license_token = base64url(payload).base64url(signature), payload
{v:1, site_id, active, locked, lock_reason, expires_at, issued_at}, signed
with LICENSE_SIGNING_KEY (base64 raw Ed25519 private key, cloud .env). The
matching public key is built into the site code, so a stored token can't be
edited and a clock can't be set before issued_at unnoticed.

Additive field only (old sites ignore it). Without the key the field is
null and a warning is logged. Pins cryptography==46.0.4 (was transitive).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-09-28 22:16:32 +03:00
co-authored by Claude Opus 5.5
parent 15331cdccf
commit 96a014ecbe
6 changed files with 72 additions and 0 deletions
+4
View File
@@ -19,6 +19,10 @@ class Settings(BaseSettings):
ADMIN_USERNAME: str = "sysadmin" ADMIN_USERNAME: str = "sysadmin"
ADMIN_PASSWORD: str = "changeme" ADMIN_PASSWORD: str = "changeme"
LATEST_VERSION: str = "0.0.0" LATEST_VERSION: str = "0.0.0"
# Ed25519 private key (base64, raw 32 bytes) that signs sites' license tokens.
# Generated once; the matching public key is built into client-services
# (local_backend/services/license.py). Keep secret, back up.
LICENSE_SIGNING_KEY: str = ""
# Manager JWT (separate secret from admin JWT) # Manager JWT (separate secret from admin JWT)
MANAGER_JWT_SECRET: str = "change-me-manager-secret" MANAGER_JWT_SECRET: str = "change-me-manager-secret"
MANAGER_JWT_EXPIRE_HOURS: int = 72 MANAGER_JWT_EXPIRE_HOURS: int = 72
+1
View File
@@ -8,3 +8,4 @@ passlib[bcrypt]==1.7.4
bcrypt==4.0.1 bcrypt==4.0.1
python-multipart==0.0.9 python-multipart==0.0.9
httpx==0.27.2 httpx==0.27.2
cryptography==46.0.4
+2
View File
@@ -7,6 +7,7 @@ from config import settings
from database import get_db from database import get_db
from models.site import Site from models.site import Site
from schemas.site import HeartbeatRequest, HeartbeatResponse from schemas.site import HeartbeatRequest, HeartbeatResponse
from services.license_token import issue_license_token
router = APIRouter() router = APIRouter()
_pwd = CryptContext(schemes=["bcrypt"], deprecated="auto") _pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
@@ -40,4 +41,5 @@ def heartbeat(
latest_version=settings.LATEST_VERSION, latest_version=settings.LATEST_VERSION,
waiter_domain=site.waiter_domain, waiter_domain=site.waiter_domain,
site_numeric_id=site.id, site_numeric_id=site.id,
license_token=issue_license_token(site, now),
) )
+2
View File
@@ -73,3 +73,5 @@ class HeartbeatResponse(BaseModel):
latest_version: str | None = None latest_version: str | None = None
waiter_domain: str | None = None waiter_domain: str | None = None
site_numeric_id: int | None = None # cloud DB pk — needed by Connect sync loops site_numeric_id: int | None = None # cloud DB pk — needed by Connect sync loops
# Signed license the site stores and enforces offline (services/license_token.py)
license_token: str | None = None
View File
+63
View File
@@ -0,0 +1,63 @@
"""
Signed license tokens for on-site servers (client-services KI-006).
Every heartbeat answer carries a token the site stores and verifies with the
public key built into its code. The site then enforces the license OFFLINE:
it keeps running until `expires_at` (+ its grace), however long it is without
internet, and nobody can extend it by editing the stored copy.
Format: <base64url(payload JSON)>.<base64url(Ed25519 signature of the first part)>
Payload: { v: 1, site_id, active, locked, lock_reason, expires_at, issued_at }
issued_at = the cloud's clock — the site uses it as a floor against clock rollback.
Key: LICENSE_SIGNING_KEY in the cloud .env (base64 raw Ed25519 private key,
generated once; see docs). Without it, heartbeats carry no token — old-style
behaviour — and a warning is logged.
"""
import base64
import json
import logging
from datetime import datetime, timezone
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from config import settings
logger = logging.getLogger(__name__)
_key: Ed25519PrivateKey | None = None
_warned = False
def _b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def _signing_key() -> Ed25519PrivateKey | None:
global _key, _warned
if _key is None and settings.LICENSE_SIGNING_KEY:
_key = Ed25519PrivateKey.from_private_bytes(base64.b64decode(settings.LICENSE_SIGNING_KEY))
if _key is None and not _warned:
logger.warning("LICENSE_SIGNING_KEY not set — heartbeats carry no signed license token")
_warned = True
return _key
def _iso(dt: datetime) -> str:
return (dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)).astimezone(timezone.utc).isoformat()
def issue_license_token(site, now: datetime) -> str | None:
key = _signing_key()
if key is None:
return None
payload = {
"v": 1,
"site_id": site.site_id,
"active": bool(site.is_active),
"locked": bool(site.is_locked),
"lock_reason": site.lock_reason,
"expires_at": _iso(site.license_expires_at),
"issued_at": _iso(now),
}
body = _b64url(json.dumps(payload, separators=(",", ":"), sort_keys=True).encode())
return f"{body}.{_b64url(key.sign(body.encode()))}"