商业许可证客户端：验证、检查状态、测试¶

// Verifies vssaas License Certificates offline.
import { createHash, createPublicKey, verify, type KeyObject } from 'node:crypto';
import { readFileSync } from 'node:fs';
import { fileURLToPath } from 'node:url';

export type Environment = 'development' | 'staging' | 'production';
export type ErrorCode =
  | 'malformed' | 'unknown_key' | 'bad_signature' | 'wrong_audience'
  | 'wrong_environment' | 'not_yet_valid' | 'expired' | 'invalid_claims';
export class LicenseCertificateError extends Error {
  readonly code: ErrorCode;
  constructor(code: ErrorCode) {
    super(code);
    this.code = code;
  }
}
const fail = (code: ErrorCode): never => {
  throw new LicenseCertificateError(code);
};

type Shape = Record<string, 'str' | 'int'>;
const CLAIMS: Shape = {
  v: 'int', kid: 'str', iss: 'str', aud: 'str', env: 'str', serial: 'str', license_id: 'str',
  seat_id: 'str', organization_id: 'str', inst: 'str', edition: 'str', agreement_version: 'str',
  term_ends_at: 'int', updates_until: 'int', iat: 'int', nbf: 'int', exp: 'int', hard_exp: 'int',
  refresh_after: 'int',
};
const ENDORSEMENT: Shape = {
  v: 'int', kid: 'str', public_key: 'str', env: 'str', nbf: 'int', exp: 'int', root_kid: 'str',
};
const ENVIRONMENTS = ['development', 'staging', 'production'];
const TOKEN = /^[A-Za-z0-9._-]{1,64}$/;
const KID = /^[0-9a-f]{16}$/;

// Strict, unpadded, canonical base64url (Buffer.from alone is lenient).
export function b64url(text: string): Buffer {
  if (!/^[A-Za-z0-9_-]*$/.test(text) || text.length % 4 === 1) fail('malformed');
  const bytes = Buffer.from(text, 'base64url');
  if (bytes.toString('base64url') !== text) fail('malformed');
  return bytes;
}
const is32 = (text: string) => {
  try {
    return b64url(text).length === 32;
  } catch {
    return false;
  }
};
export const keyIdFor = (raw: Uint8Array) => createHash('sha256').update(raw).digest('hex').slice(0, 16);
export const ed25519Key = (raw: Uint8Array): KeyObject =>
  createPublicKey({ key: { kty: 'OKP', crv: 'Ed25519', x: Buffer.from(raw).toString('base64url') }, format: 'jwk' });

// One flat JSON object: exactly the keys in `shape`, each once, never null,
// with the declared type, and nothing but whitespace after it. JSON.parse
// would silently keep the last duplicate key, so scan the tokens.
function exactObject(payload: Buffer, shape: Shape): Record<string, any> {
  const text = new TextDecoder('utf-8', { ignoreBOM: true }).decode(payload);
  const out = new Map<string, string | number>();
  let at = 0;
  const take = (re: RegExp) => {
    re.lastIndex = at;
    const m = re.exec(text);
    if (m) at = re.lastIndex;
    return m ? m[0] : null;
  };
  const ws = () => take(/[ \t\n\r]*/y);
  const str = () => take(/"(?:[^"\\\u0000-\u001f]|\\(?:["\\/bfnrt]|u[0-9A-Fa-f]{4}))*"/y);
  const num = () => take(/-?(?:0|[1-9][0-9]*)(?:\.[0-9]+)?(?:[eE][+-]?[0-9]+)?/y);
  ws();
  if (text[at++] !== '{') fail('malformed');
  ws();
  if (text[at] === '}') at++;
  else
    for (;;) {
      ws();
      const rawKey = str() ?? fail('malformed');
      const key = JSON.parse(rawKey) as string;
      if (!Object.hasOwn(shape, key) || out.has(key)) fail('malformed');
      ws();
      if (text[at++] !== ':') fail('malformed');
      ws();
      if (text[at] === '"') {
        const value = str();
        if (value === null || shape[key] !== 'str') fail('malformed');
        out.set(key, JSON.parse(value!));
      } else {
        const value = num(); // null, true, false, {} and [] all fail here
        if (value === null || shape[key] !== 'int' || !/^-?(?:0|[1-9][0-9]*)$/.test(value)) fail('malformed');
        if (!Number.isSafeInteger(Number(value))) fail('malformed');
        out.set(key, Number(value));
      }
      ws();
      const next = text[at++];
      if (next === '}') break;
      if (next !== ',') fail('malformed');
    }
  ws();
  if (at !== text.length || out.size !== Object.keys(shape).length) fail('malformed');
  return Object.fromEntries(out);
}

function open(envelope: string, prefix: string, domain: string, shape: Shape) {
  if (Buffer.byteLength(envelope, 'utf8') > 8192 || !envelope.startsWith(prefix)) fail('malformed');
  const parts = envelope.slice(prefix.length).split('.');
  if (parts.length !== 2) fail('malformed');
  const payload = b64url(parts[0]);
  const signature = b64url(parts[1]);
  if (payload.length === 0 || signature.length !== 64) fail('malformed');
  const fields = exactObject(payload, shape);
  return { fields, message: Buffer.concat([Buffer.from(domain, 'utf8'), payload]), signature };
}

// roots: root_kid -> root public key. Returns the signing key it vouches for.
export function verifyEndorsement(value: string, roots: Map<string, KeyObject>, env: Environment, now: number) {
  const { fields: e, message, signature } = open(value, 'vsle1.', 'vssaas-license-signing-key-v1\x00', ENDORSEMENT);
  const root = roots.get(e.root_kid) ?? fail('unknown_key');
  if (!verify(null, message, root, signature)) fail('bad_signature');
  if (e.env !== env) fail('wrong_environment');
  if (e.v !== 1 || !ENVIRONMENTS.includes(e.env) || !is32(e.public_key)) fail('invalid_claims');
  const raw = b64url(e.public_key);
  if (e.kid !== keyIdFor(raw) || !KID.test(e.root_kid) || e.nbf <= 0 || e.exp <= e.nbf) fail('invalid_claims');
  if (now < e.nbf) fail('not_yet_valid');
  if (now >= e.exp) fail('expired');
  return { kid: e.kid as string, key: ed25519Key(raw) };
}

// keys: kid -> signing key whose endorsement verified at the same `now`.
export function verifyCertificate(envelope: string, keys: Map<string, KeyObject>, env: Environment, now: number) {
  const { fields: c, message, signature } = open(envelope, 'vslc1.', 'vssaas-license-certificate-v1\x00', CLAIMS);
  const key = keys.get(c.kid) ?? fail('unknown_key');
  if (!verify(null, message, key, signature)) fail('bad_signature');
  if (c.aud !== 'voicestudio-desktop') fail('wrong_audience');
  if (c.env !== env) fail('wrong_environment');
  const valid =
    c.v === 1 && c.iss === 'vssaas' && ENVIRONMENTS.includes(c.env) && KID.test(c.kid) &&
    [c.serial, c.license_id, c.seat_id, c.organization_id, c.edition, c.agreement_version].every((t) => TOKEN.test(t)) &&
    is32(c.inst) && c.iat > 0 && c.updates_until > 0 && c.nbf > 0 && c.nbf <= c.iat && c.exp > c.iat &&
    c.hard_exp >= c.exp && c.term_ends_at >= c.hard_exp && c.refresh_after > c.iat;
  if (!valid) fail('invalid_claims');
  if (now < c.nbf) fail('not_yet_valid');
  if (now >= c.hard_exp) fail('expired');
  return { claims: c, inGrace: now >= c.exp };
}

// Builds the key set from GET /v1/license-signing-keys, trusting only `roots`.
export function trustedKeys(published: { endorsement: string }[], rootKeys: string[], env: Environment, now: number) {
  const roots = new Map(rootKeys.map((r) => [keyIdFor(b64url(r)), ed25519Key(b64url(r))] as const));
  const keys = new Map<string, KeyObject>();
  for (const { endorsement } of published) {
    try {
      const k = verifyEndorsement(endorsement, roots, env, now);
      keys.set(k.kid, k.key);
    } catch {} // an endorsement that does not verify contributes no key
  }
  return keys;
}

// CLI: <certificate or response.json> <signing-keys.json> <root,...> <env> [now]
if (process.argv[1] === fileURLToPath(import.meta.url)) {
  const [certFile, keysFile, roots, env, nowArg] = process.argv.slice(2);
  const raw = readFileSync(certFile, 'utf8').trim();
  const envelope = raw.startsWith('vslc1.') ? raw : JSON.parse(raw).certificate;
  const now = nowArg ? Number(nowArg) : Math.floor(Date.now() / 1000);
  const keys = trustedKeys(JSON.parse(readFileSync(keysFile, 'utf8')).keys, roots.split(','), env as Environment, now);
  try {
    const { claims, inGrace } = verifyCertificate(envelope, keys, env as Environment, now);
    console.log(JSON.stringify({ valid: true, in_grace: inGrace, claims }, null, 2));
  } catch (e) {
    console.log(JSON.stringify({ valid: false, error: (e as LicenseCertificateError).code ?? String(e) }));
    process.exitCode = 1;
  }
}

"""Verifies vssaas License Certificates offline."""
from __future__ import annotations

import base64
import hashlib
import json
import re
import sys
import time

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

ENVIRONMENTS = ("development", "staging", "production")
CLAIMS = {"v": int, "kid": str, "iss": str, "aud": str, "env": str, "serial": str, "license_id": str,
          "seat_id": str, "organization_id": str, "inst": str, "edition": str, "agreement_version": str,
          "term_ends_at": int, "updates_until": int, "iat": int, "nbf": int, "exp": int, "hard_exp": int,
          "refresh_after": int}
ENDORSEMENT = {"v": int, "kid": str, "public_key": str, "env": str, "nbf": int, "exp": int, "root_kid": str}
TOKEN = re.compile(r"[A-Za-z0-9._-]{1,64}")  # always fullmatch: "$" would accept a trailing newline
KID = re.compile(r"[0-9a-f]{16}")


class LicenseCertificateError(Exception):
    """code: malformed, unknown_key, bad_signature, wrong_audience,
    wrong_environment, not_yet_valid, expired or invalid_claims."""

    def __init__(self, code: str) -> None:
        super().__init__(code)
        self.code = code


def b64url(text: str) -> bytes:
    """Strict, unpadded, canonical base64url."""
    if not re.fullmatch(r"[A-Za-z0-9_-]*", text) or len(text) % 4 == 1:
        raise LicenseCertificateError("malformed")
    raw = base64.urlsafe_b64decode(text + "=" * (-len(text) % 4))
    if base64.urlsafe_b64encode(raw).rstrip(b"=").decode() != text:
        raise LicenseCertificateError("malformed")
    return raw


def _is32(text: str) -> bool:
    try:
        return len(b64url(text)) == 32
    except LicenseCertificateError:
        return False


def key_id_for(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()[:16]


class _Pairs(list):
    pass


def _reject(_constant):
    raise ValueError("NaN and Infinity are not JSON")


def _exact_object(payload: bytes, shape: dict) -> dict:
    try:  # json.loads rejects a leading BOM and trailing data
        pairs = json.loads(payload.decode("utf-8", "replace"), object_pairs_hook=_Pairs, parse_constant=_reject)
    except (ValueError, RecursionError):
        raise LicenseCertificateError("malformed") from None
    if not isinstance(pairs, _Pairs):
        raise LicenseCertificateError("malformed")
    out = {}
    for key, value in pairs:
        want = shape.get(key)
        # type() excludes bool and null; the int range is Go's int64.
        if want is None or key in out or type(value) is not want or (want is int and not -2**63 <= value < 2**63):
            raise LicenseCertificateError("malformed")
        out[key] = value
    if len(out) != len(shape):
        raise LicenseCertificateError("malformed")
    return out


def _open(envelope: str, prefix: str, domain: bytes, shape: dict):
    if len(envelope.encode("utf-8", "surrogatepass")) > 8192 or not envelope.startswith(prefix):
        raise LicenseCertificateError("malformed")
    parts = envelope[len(prefix):].split(".")
    if len(parts) != 2:
        raise LicenseCertificateError("malformed")
    payload, signature = b64url(parts[0]), b64url(parts[1])
    if not payload or len(signature) != 64:
        raise LicenseCertificateError("malformed")
    return _exact_object(payload, shape), domain + payload, signature


def _signed(key: Ed25519PublicKey, signature: bytes, message: bytes) -> bool:
    try:
        key.verify(signature, message)
        return True
    except InvalidSignature:
        return False


def verify_endorsement(value: str, roots: dict, env: str, now: int) -> tuple[str, Ed25519PublicKey]:
    """roots: root_kid -> root public key. Returns (kid, signing key)."""
    e, message, signature = _open(value, "vsle1.", b"vssaas-license-signing-key-v1\x00", ENDORSEMENT)
    root = roots.get(e["root_kid"])
    if root is None:
        raise LicenseCertificateError("unknown_key")
    if not _signed(root, signature, message):
        raise LicenseCertificateError("bad_signature")
    if e["env"] != env:
        raise LicenseCertificateError("wrong_environment")
    if (e["v"] != 1 or e["env"] not in ENVIRONMENTS or not _is32(e["public_key"])
            or e["kid"] != key_id_for(b64url(e["public_key"])) or not KID.fullmatch(e["root_kid"])
            or e["nbf"] <= 0 or e["exp"] <= e["nbf"]):
        raise LicenseCertificateError("invalid_claims")
    if now < e["nbf"]:
        raise LicenseCertificateError("not_yet_valid")
    if now >= e["exp"]:
        raise LicenseCertificateError("expired")
    return e["kid"], Ed25519PublicKey.from_public_bytes(b64url(e["public_key"]))


def verify_certificate(envelope: str, keys: dict, env: str, now: int) -> tuple[dict, bool]:
    """keys: kid -> signing key whose endorsement verified at the same `now`.
    Returns (claims, in_grace)."""
    c, message, signature = _open(envelope, "vslc1.", b"vssaas-license-certificate-v1\x00", CLAIMS)
    key = keys.get(c["kid"])
    if key is None:
        raise LicenseCertificateError("unknown_key")
    if not _signed(key, signature, message):
        raise LicenseCertificateError("bad_signature")
    if c["aud"] != "voicestudio-desktop":
        raise LicenseCertificateError("wrong_audience")
    if c["env"] != env:
        raise LicenseCertificateError("wrong_environment")
    tokens = ("serial", "license_id", "seat_id", "organization_id", "edition", "agreement_version")
    if not (c["v"] == 1 and c["iss"] == "vssaas" and c["env"] in ENVIRONMENTS and KID.fullmatch(c["kid"])
            and all(TOKEN.fullmatch(c[t]) for t in tokens) and _is32(c["inst"])
            and c["iat"] > 0 and c["updates_until"] > 0 and 0 < c["nbf"] <= c["iat"] and c["exp"] > c["iat"]
            and c["hard_exp"] >= c["exp"] and c["term_ends_at"] >= c["hard_exp"] and c["refresh_after"] > c["iat"]):
        raise LicenseCertificateError("invalid_claims")
    if now < c["nbf"]:
        raise LicenseCertificateError("not_yet_valid")
    if now >= c["hard_exp"]:
        raise LicenseCertificateError("expired")
    return c, now >= c["exp"]


def trusted_keys(published: list, root_keys: list[str], env: str, now: int) -> dict:
    """Key set from GET /v1/license-signing-keys, trusting only root_keys."""
    roots = {key_id_for(b64url(r)): Ed25519PublicKey.from_public_bytes(b64url(r)) for r in root_keys}
    keys = {}
    for item in published:
        try:
            kid, key = verify_endorsement(item["endorsement"], roots, env, now)
            keys[kid] = key
        except LicenseCertificateError:
            pass  # an endorsement that does not verify contributes no key
    return keys


if __name__ == "__main__":  # <certificate or response.json> <signing-keys.json> <root,...> <env> [now]
    cert_file, keys_file, roots_csv, env_name = sys.argv[1:5]
    now = int(sys.argv[5]) if len(sys.argv) > 5 else int(time.time())
    raw = open(cert_file).read().strip()
    envelope = raw if raw.startswith("vslc1.") else json.loads(raw)["certificate"]
    key_set = trusted_keys(json.load(open(keys_file))["keys"], roots_csv.split(","), env_name, now)
    try:
        claims, in_grace = verify_certificate(envelope, key_set, env_name, now)
        print(json.dumps({"valid": True, "in_grace": in_grace, "claims": claims}, indent=2))
    except LicenseCertificateError as exc:
        print(json.dumps({"valid": False, "error": exc.code}))
        sys.exit(1)

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { createHash } from 'node:crypto';
import { readFileSync } from 'node:fs';
import { b64url, keyIdFor, trustedKeys, verifyCertificate, type Environment } from './license-certificate.ts';

const golden = JSON.parse(readFileSync(new URL('./golden.json', import.meta.url), 'utf8'));
const env = golden.environment as Environment;
const keys = trustedKeys([{ endorsement: golden.endorsement.value }], [golden.root.public_key], env, golden.endorsement.now);
const codes: Record<string, string> = {
  ErrMalformed: 'malformed', ErrUnknownKey: 'unknown_key', ErrBadSignature: 'bad_signature',
  ErrWrongAudience: 'wrong_audience', ErrWrongEnvironment: 'wrong_environment',
  ErrNotYetValid: 'not_yet_valid', ErrExpired: 'expired', ErrInvalidClaims: 'invalid_claims',
};

test('file shape and key IDs', () => {
  assert.equal(createHash('sha256').update(readFileSync(new URL('./golden.json', import.meta.url))).digest('hex'),
    'f9008f69434581a3941700680955cac7669c8963a7d88d10d87dba08e23edbec');
  assert.equal(keyIdFor(b64url(golden.root.public_key)), golden.root.kid);
  assert.equal(keyIdFor(b64url(golden.signing_key.public_key)), golden.signing_key.kid);
  assert.deepEqual([...keys.keys()], [golden.signing_key.kid]); // the root itself is never a certificate key
  assert.equal(golden.valid.length, 3);
  assert.equal(golden.invalid.length, 8);
});
for (const v of golden.valid) {
  test(`accepts ${v.name}`, () => {
    const { claims, inGrace } = verifyCertificate(v.envelope, keys, env, v.now);
    assert.deepEqual(claims, v.claims);
    assert.equal(inGrace, v.in_grace);
    const payload = b64url(v.envelope.split('.')[1]);
    assert.equal(payload.toString('utf8'), v.payload_json);
    assert.equal(createHash('sha256').update(payload).digest('hex'), v.payload_sha256);
  });
}
for (const v of golden.invalid) {
  test(`rejects ${v.name}`, () => {
    assert.throws(() => verifyCertificate(v.envelope, keys, env, v.now), { code: codes[v.error] });
  });
}

import hashlib
import json
from pathlib import Path

import pytest

from license_certificate import LicenseCertificateError, b64url, key_id_for, trusted_keys, verify_certificate

GOLDEN_PATH = Path(__file__).with_name("golden.json")
GOLDEN = json.loads(GOLDEN_PATH.read_text())
ENV = GOLDEN["environment"]
KEYS = trusted_keys([{"endorsement": GOLDEN["endorsement"]["value"]}], [GOLDEN["root"]["public_key"]], ENV,
                    GOLDEN["endorsement"]["now"])
CODES = {"ErrMalformed": "malformed", "ErrUnknownKey": "unknown_key", "ErrBadSignature": "bad_signature",
         "ErrWrongAudience": "wrong_audience", "ErrWrongEnvironment": "wrong_environment",
         "ErrNotYetValid": "not_yet_valid", "ErrExpired": "expired", "ErrInvalidClaims": "invalid_claims"}


def test_file_shape_and_key_ids():
    assert hashlib.sha256(GOLDEN_PATH.read_bytes()).hexdigest() == \
        "f9008f69434581a3941700680955cac7669c8963a7d88d10d87dba08e23edbec"
    assert key_id_for(b64url(GOLDEN["root"]["public_key"])) == GOLDEN["root"]["kid"]
    assert key_id_for(b64url(GOLDEN["signing_key"]["public_key"])) == GOLDEN["signing_key"]["kid"]
    assert list(KEYS) == [GOLDEN["signing_key"]["kid"]]  # the root itself is never a certificate key
    assert (len(GOLDEN["valid"]), len(GOLDEN["invalid"])) == (3, 8)


@pytest.mark.parametrize("vector", GOLDEN["valid"], ids=lambda v: v["name"])
def test_accepts(vector):
    claims, in_grace = verify_certificate(vector["envelope"], KEYS, ENV, vector["now"])
    assert claims == vector["claims"] and in_grace == vector["in_grace"]
    payload = b64url(vector["envelope"].split(".")[1])
    assert payload.decode() == vector["payload_json"]
    assert hashlib.sha256(payload).hexdigest() == vector["payload_sha256"]


@pytest.mark.parametrize("vector", GOLDEN["invalid"], ids=lambda v: v["name"])
def test_rejects(vector):
    with pytest.raises(LicenseCertificateError) as caught:
        verify_certificate(vector["envelope"], KEYS, ENV, vector["now"])
    assert caught.value.code == CODES[vector["error"]]

POST /v1/license-seats/{seat_id}/certificates
Authorization: Bearer vsic_<43 base64url characters>
Content-Type: application/json

{ "last_certificate_serial": "<serial of the certificate held now>" }

GET /v1/license-signing-keys

{ "keys": [ { "kid": "0310b053afb12114", "public_key": "AxCwU6-…", "state": "active", "endorsement": "vsle1.…" } ] }