video_monitor/monitor_runtime/licensing.py

"""Offline signed licenses. Private signing material is never used here."""
import base64
import hashlib
import json
import os
import subprocess
import threading
import time
from datetime import datetime, timezone
from functools import lru_cache
from pathlib import Path
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from monitor_runtime.paths import DESKTOP, data, public_key_path, atomic_json

PRODUCT = 'monitor'
MAX_LICENSE_BYTES = 32768
_GUARD = threading.RLock()

class LicenseError(ValueError):
    pass

def canonical(payload):
    return json.dumps(payload, sort_keys=True, separators=(',', ':'), ensure_ascii=False).encode('utf-8')

def timestamp(value):
    dt = datetime.fromisoformat(value.replace('Z', '+00:00'))
    if dt.tzinfo is None:
        raise LicenseError('授权时间必须包含时区')
    return dt.timestamp()

@lru_cache(maxsize=1)
def machine_identity():
    if os.name != 'nt':
        raise LicenseError('机器授权仅支持 Windows')
    import winreg
    with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r'SOFTWARE\Microsoft\Cryptography',
                        0, winreg.KEY_READ | winreg.KEY_WOW64_64KEY) as key:
        guid = str(winreg.QueryValueEx(key, 'MachineGuid')[0]).strip().lower()
    result = subprocess.run(
        ['powershell.exe', '-NoProfile', '-NonInteractive', '-Command',
         '(Get-CimInstance Win32_ComputerSystemProduct).UUID'],
        capture_output=True, timeout=20, creationflags=subprocess.CREATE_NO_WINDOW, check=True)
    uuid = result.stdout.decode('utf-8', errors='replace').strip().lower()
    import uuid as uuid_module
    try:
        uuid = str(uuid_module.UUID(uuid))
    except ValueError as exc:
        raise LicenseError('无法读取系统 UUID；请检查 Windows CIM 服务') from exc
    weak = uuid in ('00000000-0000-0000-0000-000000000000', 'ffffffff-ffff-ffff-ffff-ffffffffffff')
    if not guid:
        raise LicenseError('无法读取 Windows MachineGuid')
    digest = hashlib.sha256(('monitor-machine-v1|' + uuid + '|' + guid).encode()).hexdigest()
    return {'fingerprint': 'v1:' + digest, 'weak_uuid': weak}

def request_document():
    return {'schema_version': 1, 'product': PRODUCT, **machine_identity()}

def verify_document(document, public_pem, fingerprint, now=None):
    now = time.time() if now is None else now
    try:
        if set(document) != {'payload', 'signature'}:
            raise LicenseError('授权文件结构无效')
        payload = document['payload']
        required = {'schema_version', 'product', 'license_id', 'customer', 'machine_fingerprint',
                    'issued_at', 'expires_at'}
        if not isinstance(payload, dict) or set(payload) != required:
            raise LicenseError('授权字段无效')
        key = serialization.load_pem_public_key(public_pem)
        if not isinstance(key, Ed25519PublicKey):
            raise LicenseError('发行公钥类型无效')
        key.verify(base64.b64decode(document['signature'], validate=True), canonical(payload))
        if payload['schema_version'] != 1 or payload['product'] != PRODUCT:
            raise LicenseError('授权版本或产品不匹配')
        if payload['machine_fingerprint'] != fingerprint:
            raise LicenseError('授权与本机不匹配')
        if not all(isinstance(payload[k], str) and 0 < len(payload[k]) <= 200 for k in ('customer', 'license_id')):
            raise LicenseError('客户或授权编号无效')
        if timestamp(payload['issued_at']) > now + 300:
            raise LicenseError('授权尚未生效，或系统时间不正确')
        if payload['expires_at'] is not None:
            expiry = timestamp(payload['expires_at'])
            if expiry <= timestamp(payload['issued_at']) or now >= expiry:
                raise LicenseError('授权已过期')
        return payload
    except LicenseError:
        raise
    except Exception as exc:
        raise LicenseError('授权签名或格式无效') from exc

def _clock(now):
    path = data('license-clock.json')
    previous = 0
    if path.exists():
        try:
            from monitor_runtime.windows import protect
            encoded = json.loads(path.read_text(encoding='utf-8'))['protected']
            previous = float(protect(base64.b64decode(encoded), decrypt=True).decode('ascii'))
        except Exception as exc:
            raise LicenseError('本地授权时间记录损坏，请恢复备份并检查系统时间') from exc
    if now + 300 < previous:
        raise LicenseError('检测到系统时间回拨，请校准系统时间')
    if now > previous + 60:
        from monitor_runtime.windows import protect
        encoded = base64.b64encode(protect(str(now).encode('ascii'), machine=True)).decode('ascii')
        atomic_json(path, {'protected': encoded})

def check_license():
    if not DESKTOP:
        return {'valid': True, 'reason': 'source development'}
    with _GUARD:
        try:
            path = data('license.json')
            if not path.exists():
                raise LicenseError('尚未导入授权')
            if path.stat().st_size > MAX_LICENSE_BYTES:
                raise LicenseError('授权文件过大')
            pem = public_key_path().read_bytes()
            payload = verify_document(json.loads(path.read_text(encoding='utf-8')),
                                      pem, machine_identity()['fingerprint'])
            _clock(time.time())
            return {'valid': True, 'reason': '授权有效', 'license': payload}
        except Exception as exc:
            return {'valid': False, 'reason': str(exc) if isinstance(exc, LicenseError) else '无法读取发行公钥或授权文件'}

def require_license():
    result = check_license()
    if not result['valid']:
        raise LicenseError(result['reason'])

def import_license(raw):
    if len(raw) > MAX_LICENSE_BYTES:
        raise LicenseError('授权文件过大')
    with _GUARD:
        try:
            document = json.loads(raw)
            payload = verify_document(document, public_key_path().read_bytes(), machine_identity()['fingerprint'])
            _clock(time.time())
        except LicenseError:
            raise
        except Exception as exc:
            raise LicenseError('无法读取授权文件或发行公钥') from exc
        atomic_json(data('license.json'), document)
        return payload