mycc-toylom

"""Offline license check for the $3 one-time Toy-LOM entitlement.

A license is an Ed25519-signed token: base64url(json_payload) + "." + hex(sig),
verified locally against My-CC's license public key. There is NO mandatory
phone-home: verification is pure crypto on-device. An absent license runs in
`eval` mode (functional, flagged), so the safety floor never depends on
licensing.

Per the product split: $3 buys this versioned, signed snapshot that works
offline forever. Ongoing classifier updates, live escalation, and the caregiver
dashboard sit behind the subscription, not this token.
"""

from __future__ import annotations

import base64
import json
import os
import time
from typing import Optional

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey

# Baked My-CC license public key (hex). Overridable via env for staging; the
# real GA key is injected at release. Empty default => eval mode unless an env
# key is supplied.
_BAKED_PUBKEY = os.environ.get("MYCC_LICENSE_PUBKEY", "")


def issue_license(payload: dict, private_key: Ed25519PrivateKey) -> str:
    body = base64.urlsafe_b64encode(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).decode()
    sig = private_key.sign(body.encode()).hex()
    return f"{body}.{sig}"


def verify_license(token: Optional[str], pubkey_hex: Optional[str] = None,
                   now: Optional[float] = None, device_id: Optional[str] = None) -> dict:
    """Return {"valid": bool, "mode": "licensed"|"eval"|"invalid"|"expired"|"wrong_device", "payload": dict|None}.

    `now` is injectable for deterministic expiry tests; defaults to wall clock.
    `device_id`: if the token is device-bound (payload has device_id) it must
    match this host's id, so a $3 token bought for one toy cannot be copied to a
    fleet (HIGH-24). A token with no device_id is not device-bound (portable)."""
    pubkey_hex = pubkey_hex if pubkey_hex is not None else _BAKED_PUBKEY
    if not token:
        return {"valid": False, "mode": "eval", "payload": None}
    if not pubkey_hex:
        # No verification key available on this build: cannot assert validity,
        # but never phone home. Treat as eval so the guard still runs.
        return {"valid": False, "mode": "eval", "payload": None}
    try:
        body, sig_hex = token.split(".", 1)
        pk = Ed25519PublicKey.from_public_bytes(bytes.fromhex(pubkey_hex))
        pk.verify(bytes.fromhex(sig_hex), body.encode())
        payload = json.loads(base64.urlsafe_b64decode(body.encode()))
        if payload.get("product") != "toylom":
            return {"valid": False, "mode": "invalid", "payload": payload}
        # HIGH-24: a signed token is not "offline forever" if it carries an
        # expiry. The signature is valid, but an expired / not-yet-valid token
        # must not assert entitlement. The safety FLOOR is unaffected either way
        # (it never depends on the license); only the licensed/eval flag moves.
        if not _within_validity(payload, now=now):
            return {"valid": False, "mode": "expired", "payload": payload}
        # HIGH-24 device binding: if the token names a device, it must match.
        bound = payload.get("device_id")
        if bound and device_id is not None and str(bound) != str(device_id):
            return {"valid": False, "mode": "wrong_device", "payload": payload}
        return {"valid": True, "mode": "licensed", "payload": payload}
    except (InvalidSignature, ValueError, json.JSONDecodeError):
        return {"valid": False, "mode": "invalid", "payload": None}


def _as_epoch(value) -> Optional[float]:
    """Accept an epoch number or an ISO-8601 string; return epoch seconds."""
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return float(value)
    try:
        from datetime import datetime
        s = str(value).replace("Z", "+00:00")
        return datetime.fromisoformat(s).timestamp()
    except (ValueError, TypeError):
        return None


def _within_validity(payload: dict, now: Optional[float] = None) -> bool:
    """True if the token is inside [not_before, not_after]. Missing bounds mean
    open-ended on that side (a token with no not_after is offline-forever)."""
    t = now if now is not None else time.time()
    not_after = _as_epoch(payload.get("not_after"))
    if not_after is not None and t > not_after:
        return False
    not_before = _as_epoch(payload.get("not_before"))
    if not_before is not None and t < not_before:
        return False
    return True


def license_from_env_or_file() -> Optional[str]:
    tok = os.environ.get("MYCC_TOYLOM_LICENSE")
    if tok:
        return tok
    path = os.environ.get("MYCC_TOYLOM_LICENSE_FILE", "")
    if path and os.path.exists(path):
        with open(path, "r", encoding="utf-8") as fh:
            return fh.read().strip()
    return None