bonaminandClaude Opus 5.5
13a451a756
- services/license.py: verifies the cloud's Ed25519-signed license token
(public key built in) and decides purely: valid → licensed until expiry,
then a 5-day grace, then blocked - never mid-service (deferred while a
workday is open, applied at close). Works offline for as long as the
license lasts: the "unlicensed after 72h without heartbeat" rule is gone.
- Tamper resistance: an edited token fails the signature ("unverified");
a clock earlier than the latest provable time (token issued_at, newest
order in the DB, stored high-water mark; 1 day tolerance) → "clock".
- apply_license() re-evaluates from the stored token at startup, after
every heartbeat attempt and when a workday closes. Cloud lock/unlock from
the token keeps the workday-deferred behaviour. Transition: a cloud
without tokens is trusted 72h per successful heartbeat.
- FIX: the promised 5-day grace after expiry never happened - the cloud's
licensed=false was applied immediately (402 on everything).
- FIX: license_state.json lived inside the container and was lost on every
re-creation; it now lives in the data volume (old path read once).
- /api/system/status: offline_days, license_verified, license_problem,
grace_over; lock_reason "clock"/"unverified"; grace days from the license
module (rounded up).
Tests: 18 unit checks (signature, tamper, other site/key, 364 days
offline, grace ±workday, inactive, clock rollback, transition) + 17 E2E
checks with a real cloud + site process (400 days offline, tampered file →
402, clock behind newest order, expiry deferred until workday close,
renewal, remote lock/unlock).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
107 lines
4.8 KiB
Python
107 lines
4.8 KiB
Python
"""
|
|
Offline-capable licensing (KI-006).
|
|
|
|
The cloud is needed to RENEW a license, not to RUN one. Every heartbeat
|
|
brings a license token signed by the cloud (Ed25519). The site stores it and
|
|
enforces it by itself, so a venue that paid for a year keeps working for that
|
|
year even if it never goes online again. The old rule ("unlicensed after 72h
|
|
without a heartbeat") is gone.
|
|
|
|
Tamper resistance:
|
|
- Editing the stored token breaks the signature → treated as no license.
|
|
- Turning the clock back: "now" may not be earlier than the latest time this
|
|
system has provably seen — the token's signed cloud time, the newest order
|
|
in the database, and a stored high-water mark (1 day tolerance).
|
|
Limits (documented): someone with root on the server could still patch the
|
|
code itself, and a remote lock only reaches a site when it next goes online.
|
|
|
|
Rules (evaluate()):
|
|
valid token, clock OK:
|
|
active and not expired → licensed
|
|
expired: 5-day grace → licensed, warnings in the manager
|
|
grace over / site deactivated → unlicensed, but never mid-service:
|
|
deferred while a workday is open
|
|
no valid token:
|
|
old cloud without tokens, heartbeat said licensed < 72h ago → licensed (transition)
|
|
otherwise → unlicensed ("unverified")
|
|
Locks from the cloud (token.locked) keep the existing workday-deferred behaviour.
|
|
"""
|
|
import base64
|
|
import json
|
|
import math
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from cryptography.exceptions import InvalidSignature
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
|
|
|
|
# Matches LICENSE_SIGNING_KEY in the cloud's .env (generated 2026-09-28).
|
|
# Built into the code on purpose: a key taken from configuration could simply
|
|
# be replaced together with a self-made token.
|
|
LICENSE_PUBLIC_KEY = "2oeFHV6hgAlJsx/ZBvG6fqmWYn5tjSW5hURrrPhLoOw="
|
|
|
|
EXPIRY_GRACE = timedelta(days=5)
|
|
CLOCK_TOLERANCE = timedelta(days=1)
|
|
LEGACY_UNSIGNED_OK = timedelta(hours=72) # only while the cloud sends no tokens
|
|
|
|
|
|
def _b64url_decode(s: str) -> bytes:
|
|
return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
|
|
|
|
|
|
def parse_dt(value) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
dt = datetime.fromisoformat(value) if isinstance(value, str) else value
|
|
except ValueError:
|
|
return None
|
|
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def verify_token(token: str | None, site_id: str, public_key_b64: str = LICENSE_PUBLIC_KEY) -> dict | None:
|
|
"""Payload of a genuine token for this site, else None."""
|
|
if not token or "." not in token:
|
|
return None
|
|
body, _, sig = token.partition(".")
|
|
try:
|
|
Ed25519PublicKey.from_public_bytes(base64.b64decode(public_key_b64)).verify(_b64url_decode(sig), body.encode())
|
|
payload = json.loads(_b64url_decode(body))
|
|
except (InvalidSignature, ValueError, TypeError):
|
|
return None
|
|
if payload.get("v") != 1 or payload.get("site_id") != site_id:
|
|
return None
|
|
if not parse_dt(payload.get("expires_at")) or not parse_dt(payload.get("issued_at")):
|
|
return None
|
|
return payload
|
|
|
|
|
|
def evaluate(payload: dict | None, now: datetime, floor: datetime | None, workday_open: bool,
|
|
legacy_ok_at: datetime | None = None) -> dict:
|
|
"""Pure license decision. `floor` = latest time this system has provably seen."""
|
|
base = {"license_verified": payload is not None, "license_problem": None, "grace_over": False,
|
|
"days_until_expiry": None, "grace_expires_at": None, "grace_days_remaining": None}
|
|
|
|
if payload is None:
|
|
legacy = legacy_ok_at is not None and now - legacy_ok_at <= LEGACY_UNSIGNED_OK
|
|
return {**base, "licensed": legacy, "license_problem": None if legacy else "unverified"}
|
|
|
|
if floor is not None and now < floor - CLOCK_TOLERANCE:
|
|
return {**base, "licensed": False, "license_problem": "clock"}
|
|
|
|
expires = parse_dt(payload["expires_at"])
|
|
grace_end = expires + EXPIRY_GRACE
|
|
days_until = (expires - now).days # negative once expired
|
|
grace_over = now > grace_end
|
|
fields = {
|
|
**base,
|
|
"expires_at": expires.isoformat(),
|
|
"days_until_expiry": days_until,
|
|
"grace_expires_at": grace_end.isoformat() if days_until < 0 else None,
|
|
# Rounded up: 2 days 23 hours left reads as "3 days", as people count it
|
|
"grace_days_remaining": math.ceil((grace_end - now) / timedelta(days=1)) if days_until < 0 and not grace_over else None,
|
|
"grace_over": grace_over,
|
|
}
|
|
problem = "inactive" if not payload.get("active", True) else "expired" if grace_over else None
|
|
# Never cut a restaurant off mid-service: an open workday finishes first
|
|
return {**fields, "licensed": problem is None or workday_open, "license_problem": problem}
|