scalesApp/backend/api/licensing.py

"""Offline node-locked licensing.

Customer sites are air-gapped, so there is no activation server. The flow is:

  1. This machine reports a fingerprint built from several hardware components
  2. The customer sends that string to the vendor
  3. The vendor SIGNS it with their Ed25519 private key
  4. The customer pastes the result back; this module VERIFIES it

The critical property is that this code can only ever verify, never generate.
The private key never leaves the vendor's machine, so unpacking the application
does not let anyone mint licences. A shared-secret scheme would be worthless.

See .claude/plans/licensing.md for the threat model: this stops casual copying
and gives a clear contractual line. It does not stop a determined attacker, and
is not intended to.
"""

import base64
import hashlib
import json
import platform
import subprocess
from datetime import timedelta

from django.utils import timezone

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


# Vendor's Ed25519 public key (base64 raw, 32 bytes). The matching PRIVATE key
# lives only on the vendor's machine and must never enter this repository.
#
# Replace this placeholder with the real public key before shipping. While it is
# empty, no licence can validate and the system runs in grace mode.
LICENSE_PUBLIC_KEY = 'HAbweykY+P9jdb81BpfiGVXwM2BIgkh9tBjSHq7RN+I='

# How long an unlicensed or newly-invalid install keeps working normally. A
# licence check must never stop a weighbridge without warning, so a hardware
# failure becomes a support call rather than an outage.
GRACE_DAYS = 14

# Station limit once grace has expired.
UNLICENSED_STATIONS = 1

# A slot not refreshed within this window is released. The frontend refreshes
# every 30s, so a station must miss four in a row - a brief hiccup is harmless.
SLOT_TIMEOUT_SECONDS = 120

# Hardware components must match at least this many of four, so replacing a disk
# or a network card does not invalidate an otherwise identical machine.
MIN_COMPONENT_MATCHES = 3


def _run(command):
    """Run a shell command and return stripped stdout, or None on any failure."""
    try:
        out = subprocess.run(
            command, shell=True, capture_output=True, text=True, timeout=10
        )
        if out.returncode != 0:
            return None
        value = out.stdout.strip()
        return value or None
    except Exception:
        return None


def _hash(value):
    if not value:
        return None
    return hashlib.sha256(str(value).strip().encode('utf-8')).hexdigest()[:16]


def _powershell(script):
    """Run a PowerShell snippet and return its stripped output, or None."""
    return _run(
        'powershell -NoProfile -NonInteractive -Command "' + script.replace('"', '\\"') + '"'
    )


def _windows_components():
    """Hardware identifiers via PowerShell CIM.

    Uses PowerShell rather than wmic: wmic is deprecated and has been REMOVED
    from recent Windows 11 builds, where it silently yields nothing. Each lookup
    falls back to wmic so older machines still work.
    """
    parts = {}

    board = _powershell('(Get-CimInstance Win32_BaseBoard).SerialNumber')
    if not board:
        out = _run('wmic baseboard get serialnumber')
        if out:
            lines = [l.strip() for l in out.splitlines() if l.strip()]
            board = lines[1] if len(lines) > 1 else None
    if board and board.lower() not in ('none', 'default string', 'to be filled by o.e.m.'):
        parts['board'] = board

    disk = _powershell(
        '(Get-CimInstance Win32_DiskDrive | Select-Object -First 1).SerialNumber'
    )
    if not disk:
        out = _run('wmic diskdrive get serialnumber')
        if out:
            lines = [l.strip() for l in out.splitlines() if l.strip()]
            disk = lines[1] if len(lines) > 1 else None
    if disk:
        parts['disk'] = disk

    mac = _run('getmac /fo csv /nh')
    if mac:
        first = mac.splitlines()[0].split(',')[0].strip().strip('"')
        if first and first != 'N/A':
            parts['mac'] = first

    guid = _powershell(
        r"(Get-ItemProperty 'HKLM:\SOFTWARE\Microsoft\Cryptography').MachineGuid"
    )
    if not guid:
        out = _run(r'reg query "HKLM\SOFTWARE\Microsoft\Cryptography" /v MachineGuid')
        if out and 'MachineGuid' in out:
            guid = out.split()[-1]
    if guid:
        parts['guid'] = guid

    return parts


def _linux_components():
    parts = {}

    board = _run('cat /sys/class/dmi/id/board_serial 2>/dev/null')
    if board:
        parts['board'] = board

    disk = _run("lsblk -dno SERIAL 2>/dev/null | head -1")
    if disk:
        parts['disk'] = disk

    mac = _run(
        "cat /sys/class/net/$(ls /sys/class/net | grep -v lo | head -1)/address 2>/dev/null"
    )
    if mac:
        parts['mac'] = mac

    guid = _run('cat /etc/machine-id 2>/dev/null')
    if guid:
        parts['guid'] = guid

    return parts


def machine_components():
    """Hashed hardware identifiers for this machine.

    Missing components come back as None rather than raising - a virtual machine
    or restricted environment may not expose all of them, and the 3-of-4 rule
    tolerates that as long as enough are present.
    """
    raw = _windows_components() if platform.system() == 'Windows' else _linux_components()
    return {key: _hash(raw.get(key)) for key in ('board', 'disk', 'mac', 'guid')}


def machine_fingerprint():
    """Short human-readable fingerprint, for the customer to send to the vendor.

    Read aloud over the phone or copied from the admin screen, so it is grouped
    and uppercase with no ambiguous characters beyond hex.
    """
    components = machine_components()
    present = [v for v in components.values() if v]
    if not present:
        return 'UNAVAILABLE'
    digest = hashlib.sha256('|'.join(sorted(present)).encode('utf-8')).hexdigest()
    body = digest[:16].upper()
    return '-'.join(body[i:i + 4] for i in range(0, 16, 4))


def count_matching_components(licensed):
    """How many of the licensed component hashes still match this machine."""
    current = machine_components()
    matches = 0
    for key in ('board', 'disk', 'mac', 'guid'):
        want = (licensed or {}).get(key)
        if want and current.get(key) == want:
            matches += 1
    return matches


def verify_license(blob):
    """Verify a signed licence blob.

    Returns (valid, payload, reason). `payload` is returned even when the
    signature is valid but the machine does not match, so callers can report
    which machine the licence was actually issued for.

    Blob format: base64("<payload json>.<signature>")
    """
    if not blob:
        return False, None, 'No licence installed'

    if not LICENSE_PUBLIC_KEY:
        return False, None, 'No vendor public key compiled in'

    try:
        decoded = base64.b64decode(blob.strip()).decode('utf-8')
        payload_json, signature_b64 = decoded.rsplit('.', 1)
        payload = json.loads(payload_json)
        signature = base64.b64decode(signature_b64)
    except Exception:
        return False, None, 'Licence is malformed'

    try:
        public_key = Ed25519PublicKey.from_public_bytes(
            base64.b64decode(LICENSE_PUBLIC_KEY)
        )
        public_key.verify(signature, payload_json.encode('utf-8'))
    except InvalidSignature:
        return False, None, 'Licence signature is not valid'
    except Exception:
        return False, None, 'Licence could not be verified'

    matches = count_matching_components(payload.get('components'))
    if matches < MIN_COMPONENT_MATCHES:
        return False, payload, (
            f'Licence was issued for a different machine '
            f'({matches} of 4 hardware components match, {MIN_COMPONENT_MATCHES} required)'
        )

    # Perpetual licences have no expiry; the field exists so subscriptions can be
    # added later without changing the signed format.
    expires = payload.get('expires')
    if expires:
        try:
            if timezone.now() > timezone.datetime.fromisoformat(expires):
                return False, payload, f'Licence expired on {expires}'
        except Exception:
            return False, payload, 'Licence expiry date is malformed'

    return True, payload, None


def license_status():
    """Current licence state, including how many stations may connect.

    Falls back to a grace period so an unlicensed or newly-invalid install keeps
    working while the customer obtains a key.
    """
    from .models import License

    record = License.objects.first()
    blob = record.key_blob if record else ''
    valid, payload, reason = verify_license(blob)

    if valid:
        return {
            'valid': True,
            'customer': payload.get('customer', ''),
            'stations': int(payload.get('stations', 1)),
            'reason': None,
            'in_grace': False,
            'grace_days_left': None,
            'fingerprint': machine_fingerprint(),
        }

    # Invalid or absent - start (or continue) the grace period.
    now = timezone.now()
    if record is None:
        grace_started = None
    else:
        if record.grace_started is None:
            record.grace_started = now
            record.save(update_fields=['grace_started'])
        grace_started = record.grace_started

    if grace_started is None:
        # Nothing installed at all: grace runs from first use, tracked by the
        # placeholder row the status endpoint creates.
        grace_started = now

    days_used = (now - grace_started).days
    days_left = max(0, GRACE_DAYS - days_used)
    in_grace = days_left > 0

    licensed_stations = int(payload.get('stations', 1)) if payload else 1

    return {
        'valid': False,
        'customer': payload.get('customer', '') if payload else '',
        'stations': licensed_stations if in_grace else UNLICENSED_STATIONS,
        'reason': reason,
        'in_grace': in_grace,
        'grace_days_left': days_left,
        'fingerprint': machine_fingerprint(),
    }


def effective_station_limit():
    return license_status()['stations']


def live_slot_cutoff():
    return timezone.now() - timedelta(seconds=SLOT_TIMEOUT_SECONDS)