From 86f5f1be68830aeb7ce6c9fbef6aa4433885cc4b Mon Sep 17 00:00:00 2001 From: cproudlock Date: Sat, 18 Jul 2026 20:21:57 -0400 Subject: [PATCH] ADR-013 Phase 1: signed plugin artifacts (pack/validate/keygen) Packaging + provenance for the plugin marketplace. No runtime behavior change yet - verification is available on demand; enforcing it at plugin load/migrate and pulling from a shelf are Phase 2. - signing.py: ed25519 key pairs + provenance. Provenance is a sorted per-file SHA-256 map plus metadata; the detached signature covers the exact serialized provenance bytes, so verifying is re-hash files, re-serialize, check signature. verify() accepts any of several trusted keys (rotation). Uses cryptography (already a dependency). - packaging.py: pack() builds a signed -.shopdbplugin (zip + PROVENANCE.json + PROVENANCE.sig). verify_artifact()/verify_dir() re-hash and check the signature, and flag a tampered file, an unexpected file, a wrong/absent key - all fail closed. - CLI: `flask plugin keygen` (publisher key pair), `flask plugin pack --key` (validates then signs), and `flask plugin validate` extended to a signed artifact by path (--pubkey, else PLUGIN_TRUSTED_KEYS). - config PLUGIN_TRUSTED_KEYS: os.pathsep-separated public-key PEM paths, delivered with the site config, never read from the shelf. .env.example documents it. - docs/PLUGIN-SIGNING.md: curator flow (keygen offline, review, pack, publish, pin keys, rotate). The signature proves an artifact is exactly what a curator signed, not that the code is safe - human review before signing is the control. 11 tests: sign/verify round trip, wrong key, provenance excludes noise, serialize determinism, pack + verify, tamper -> hash mismatch, extra file, no-key fail-closed, verify_dir. 1028 pass, naming green. --- .env.example | 6 + docs/PLUGIN-SIGNING.md | 75 ++++++++++++ shopdb/config.py | 10 ++ shopdb/plugins/cli.py | 203 ++++++++++++++++++++++++++++----- shopdb/plugins/packaging.py | 131 +++++++++++++++++++++ shopdb/plugins/signing.py | 136 ++++++++++++++++++++++ tests/test_plugin_packaging.py | 153 +++++++++++++++++++++++++ 7 files changed, 687 insertions(+), 27 deletions(-) create mode 100644 docs/PLUGIN-SIGNING.md create mode 100644 shopdb/plugins/packaging.py create mode 100644 shopdb/plugins/signing.py create mode 100644 tests/test_plugin_packaging.py diff --git a/.env.example b/.env.example index 6263201..cf7870c 100644 --- a/.env.example +++ b/.env.example @@ -61,6 +61,12 @@ ZABBIX_TOKEN= # COLLECTOR_API_KEY= # COLLECTOR_API_KEY_COMPUTERS= +# ---- Trusted plugin publisher keys (ADR-013, optional) ---- +# Public-key PEM paths (OS path separator: ':' on Linux, ';' on Windows) used +# to verify signed plugin artifacts. Delivered with this config, NEVER from the +# plugin shelf. Empty on a site that does not adopt marketplace plugins. +# PLUGIN_TRUSTED_KEYS=/etc/shopdb/keys/curator.pub + # ---- Employee directory database (optional, read-only) ---- # Separate HR/employee lookup DB consumed by the notifications plugin and the # public shopfloor kiosks. Leave unset if the feature is not used; there is no diff --git a/docs/PLUGIN-SIGNING.md b/docs/PLUGIN-SIGNING.md new file mode 100644 index 0000000..06730a7 --- /dev/null +++ b/docs/PLUGIN-SIGNING.md @@ -0,0 +1,75 @@ +# Plugin signing and packaging (curator guide) + +ADR-013 Phase 1. How a plugin becomes a signed, verifiable artifact and how a +site trusts it. The signature proves an artifact is EXACTLY what a curator +reviewed and signed - it does not prove the code is safe. Human review before +signing is the actual safety control; the signature makes that review's verdict +tamper-evident all the way to the point of execution. + +Requires the `cryptography` package (already a dependency). + +## One-time: create the publisher key pair + +``` +flask plugin keygen --out ./keys --name curator +``` + +Writes `keys/curator.key` (PRIVATE) and `keys/curator.pub` (public). + +- Keep the `.key` OFFLINE with the curator. It is the only thing that can sign a + trusted artifact. Never put it on the plugin shelf or in the repo. +- Distribute the `.pub` with each site's deployed config and pin it (below). +- Rotation: generate a new pair, pin BOTH public keys on sites for an overlap + window (`verify` accepts any trusted key), then retire the old one. + +## Per plugin: review, then pack + +1. Review the plugin's source. This is the security gate - read what it does. +2. Validate and package in one step: + + ``` + flask plugin pack printers --key ./keys/curator.key --publisher west-jefferson + ``` + + `pack` refuses to sign a directory that does not validate (manifest schema, + name/dir match, core_version, dependencies on disk). On success it writes + `printers-.shopdbplugin` - a zip of the plugin plus: + + - `PROVENANCE.json`: name, version, publisher, created, and a sorted + `{file: sha256}` map of every packaged file. + - `PROVENANCE.sig`: a detached ed25519 signature over the exact + `PROVENANCE.json` bytes. + +3. Publish the artifact to the shelf (a SharePoint-synced or copied folder). + Transport is untrusted; the signature is what makes it safe. + +## Verify an artifact + +``` +flask plugin validate dist/printers-1.0.0.shopdbplugin --pubkey ./keys/curator.pub +``` + +Checks, fail-closed: signature against the trusted key(s), every file's hash, +no unexpected files, manifest schema, and that the plugin's `core_version` +admits this framework's contract version. Any changed byte in any file fails +the hash check; a signature from an untrusted key fails the signature check. + +## Pin trusted keys on a site + +Set `PLUGIN_TRUSTED_KEYS` to one or more public-key PEM paths, separated by the +OS path separator (`:` on Linux, `;` on Windows), in the site's environment: + +``` +PLUGIN_TRUSTED_KEYS=/etc/shopdb/keys/curator.pub:/etc/shopdb/keys/curator-next.pub +``` + +Keys are read only from this deployed config, never from the shelf - a folder an +attacker could write must not also carry the keys that authenticate it. With no +keys set, `validate` on an artifact fails closed (unverifiable). + +## What is NOT in this phase + +Verification is available but not yet enforced automatically at plugin load or +migrate time, and there is no `adopt` command that pulls from a shelf yet. Those +land in Phase 2 (verify-at-load, verify-at-migrate, the signed shelf index). +For now, verify artifacts manually with `validate` before installing. diff --git a/shopdb/config.py b/shopdb/config.py index 4f8b22d..4da2a1a 100644 --- a/shopdb/config.py +++ b/shopdb/config.py @@ -60,6 +60,16 @@ class Config: # API key for the unattended PowerShell collector scripts COLLECTOR_API_KEY = os.environ.get('COLLECTOR_API_KEY', '') + # Trusted plugin publisher public keys (ADR-013). os.pathsep-separated PEM + # file paths, delivered out-of-band with the deployed config - NEVER read + # from the plugin shelf. Used to verify signed plugin artifacts. Empty on a + # site that does not adopt marketplace plugins. + PLUGIN_TRUSTED_KEYS = [ + path.strip() + for path in os.environ.get('PLUGIN_TRUSTED_KEYS', '').split(os.pathsep) + if path.strip() + ] + ZABBIX_ENABLED = os.environ.get('ZABBIX_ENABLED', 'false').lower() == 'true' ZABBIX_URL = os.environ.get('ZABBIX_URL', '') ZABBIX_TOKEN = os.environ.get('ZABBIX_TOKEN', '') diff --git a/shopdb/plugins/cli.py b/shopdb/plugins/cli.py index cc5adcc..90a5d28 100644 --- a/shopdb/plugins/cli.py +++ b/shopdb/plugins/cli.py @@ -280,38 +280,28 @@ def new_plugin(name: str, description: str, overwrite: bool): click.echo(f' 6. Run: pytest plugins/{name}/tests/') -@plugin_cli.command('validate') -@click.argument('name') -@with_appcontext -def validate_plugin(name: str): - """Validate a plugin directory against the manifest schema + contract. +def _dep_name(dep: str) -> str: + """Bare plugin name from a dependency spec, dropping any PEP440 range.""" + for sep in ('>', '<', '=', '!', '~', ' '): + dep = dep.split(sep)[0] + return dep.strip() - Pre-publish gate (directory mode). Checks: manifest loads and its name - matches the directory, required/typed fields per the manifest schema, the - core_version range admits this framework's contract version, and every - declared hard dependency exists on disk. Exits non-zero on any failure. - Usage: flask plugin validate printers - """ +def _validate_directory(pm, name: str) -> list: + """Run the directory-mode checks, echoing each. Returns a failures list.""" from shopdb import __contract_version__ from ..exceptions import PluginError - pm = current_app.extensions.get('plugin_manager') - if not pm: - click.echo(click.style("Plugin manager not initialized", fg='red')) - raise SystemExit(1) - failures = [] - # 1. Manifest loads (raises on missing/unparseable/name-mismatch). try: manifest = pm.loader.load_manifest(name) except PluginError as e: click.echo(click.style(f" manifest: {e}", fg='red')) - raise SystemExit(1) - click.echo(click.style(" manifest loads + name matches directory", fg='green')) + return [str(e)] + click.echo(click.style( + " manifest loads + name matches directory", fg='green')) - # 2. Schema. schema_errors = _check_against_schema(manifest, _load_manifest_schema()) if schema_errors: for err in schema_errors: @@ -320,7 +310,6 @@ def validate_plugin(name: str): else: click.echo(click.style(" schema OK", fg='green')) - # 3. Contract version range admits this framework. try: pm.loader.check_contract_version(name, __contract_version__) click.echo(click.style( @@ -329,24 +318,184 @@ def validate_plugin(name: str): failures.append(str(e)) click.echo(click.style(f" core_version: {e}", fg='red')) - # 4. Declared hard dependencies exist on disk (name-only; ranges stripped - # later when the loader gains range semantics). available = set(pm.loader.discover_plugins()) + dep_failures = [] for dep in manifest.get('dependencies', []): - depname = dep.split('>')[0].split('<')[0].split('=')[0].split('!')[0].split('~')[0].strip() + depname = _dep_name(dep) if depname not in available: + dep_failures.append(depname) failures.append(f"dependency '{depname}' not found on disk") click.echo(click.style( f" dependency '{depname}' not found on disk", fg='red')) - if not failures and manifest.get('dependencies'): + if manifest.get('dependencies') and not dep_failures: click.echo(click.style(" dependencies present on disk", fg='green')) + return failures + + +def _validate_artifact(pubkey_paths, artifact_path: str) -> list: + """Verify a signed .shopdbplugin: signature + hashes + manifest schema + + contract. Returns a failures list.""" + from shopdb import __contract_version__ + from packaging.specifiers import SpecifierSet + from packaging.version import Version + from .packaging import verify_artifact + from .signing import load_trusted_keys + + keys = load_trusted_keys(pubkey_paths) + if not keys: + click.echo(click.style( + " no trusted keys (pass --pubkey or set PLUGIN_TRUSTED_KEYS)", + fg='red')) + + manifest, errors = verify_artifact(artifact_path, keys) + failures = list(errors) + for err in errors: + click.echo(click.style(f" {err}", fg='red')) + if not errors: + click.echo(click.style( + " signature trusted + every file hash intact", fg='green')) + + if manifest is not None: + schema_errors = _check_against_schema( + manifest, _load_manifest_schema()) + for err in schema_errors: + failures.append(f"schema: {err}") + click.echo(click.style(f" schema: {err}", fg='red')) + if not schema_errors: + click.echo(click.style(" schema OK", fg='green')) + + spec = manifest.get('core_version', '') + if spec and Version(__contract_version__) not in SpecifierSet(spec): + failures.append( + f"core_version {spec} excludes contract {__contract_version__}") + click.echo(click.style( + f" core_version {spec} excludes contract " + f"{__contract_version__}", fg='red')) + elif spec: + click.echo(click.style( + f" core_version admits contract {__contract_version__}", + fg='green')) + + return failures + + +@plugin_cli.command('validate') +@click.argument('target') +@click.option('--pubkey', 'pubkeys', multiple=True, + type=click.Path(exists=True), + help='Trusted public key PEM (artifact mode). Repeatable.') +@with_appcontext +def validate_plugin(target: str, pubkeys): + """Validate a plugin - directory (by name) or a signed artifact (by path). + + Directory mode (`flask plugin validate printers`): manifest loads + name + match, manifest schema, core_version admits this contract, dependencies + exist on disk. + + Artifact mode (`flask plugin validate dist/printers-1.0.0.shopdbplugin`): + signature against trusted keys (--pubkey, else PLUGIN_TRUSTED_KEYS), every + file hash, no unexpected files, manifest schema + core_version. Exits + non-zero on any failure. + """ + pm = current_app.extensions.get('plugin_manager') + if not pm: + click.echo(click.style("Plugin manager not initialized", fg='red')) + raise SystemExit(1) + + is_artifact = target.endswith('.shopdbplugin') and Path(target).is_file() + if is_artifact: + keys = list(pubkeys) or current_app.config.get( + 'PLUGIN_TRUSTED_KEYS', []) + failures = _validate_artifact(keys, target) + label = Path(target).name + else: + failures = _validate_directory(pm, target) + label = target + click.echo("") if failures: click.echo(click.style( - f"{name}: INVALID ({len(failures)} problem(s))", fg='red')) + f"{label}: INVALID ({len(failures)} problem(s))", fg='red')) raise SystemExit(1) - click.echo(click.style(f"{name}: valid", fg='green')) + click.echo(click.style(f"{label}: valid", fg='green')) + + +@plugin_cli.command('keygen') +@click.option('--out', 'out_dir', default='.', type=click.Path(), + help='Directory to write the key pair into') +@click.option('--name', 'key_name', default='curator', + help='Base filename for the pair') +def keygen(out_dir: str, key_name: str): + """Generate an ed25519 publisher key pair for signing plugin artifacts. + + Writes .key (PRIVATE - keep offline with the curator) and .pub + (public - pin on each site via PLUGIN_TRUSTED_KEYS). Not app-bound. + """ + from .signing import generate_keypair + + out = Path(out_dir) + out.mkdir(parents=True, exist_ok=True) + private_pem, public_pem = generate_keypair() + + private_path = out / f'{key_name}.key' + public_path = out / f'{key_name}.pub' + private_path.write_bytes(private_pem) + public_path.write_bytes(public_pem) + try: + private_path.chmod(0o600) + except OSError: + pass + + click.echo(click.style(f"Private key: {private_path}", fg='green')) + click.echo(click.style(f"Public key: {public_path}", fg='green')) + click.echo("") + click.echo(click.style( + "Keep the .key OFFLINE. Never place it on the plugin shelf. Distribute " + "the .pub with each site's config and list it in PLUGIN_TRUSTED_KEYS.", + fg='yellow')) + + +@plugin_cli.command('pack') +@click.argument('name') +@click.option('--key', 'key_path', required=True, + type=click.Path(exists=True), help='Private signing key PEM') +@click.option('--publisher', default='', + help='Publisher id recorded in the provenance') +@click.option('--out', 'out_dir', default=None, type=click.Path(), + help='Output directory (default: the plugins/ directory)') +@with_appcontext +def pack_plugin(name: str, key_path: str, publisher: str, out_dir): + """Validate a plugin directory, then emit a signed artifact. + + Produces -.shopdbplugin. Fails without packing if the + directory does not validate. + + Usage: flask plugin pack printers --key curator.key --publisher west-jefferson + """ + from .signing import load_private_key + from .packaging import pack + + pm = current_app.extensions.get('plugin_manager') + if not pm: + click.echo(click.style("Plugin manager not initialized", fg='red')) + raise SystemExit(1) + + click.echo(f"Validating {name} ...") + failures = _validate_directory(pm, name) + if failures: + click.echo("") + click.echo(click.style( + f"Refusing to pack {name}: {len(failures)} validation problem(s)", + fg='red')) + raise SystemExit(1) + + plugin_dir = Path(pm.loader.plugins_dir) / name + private_key = load_private_key(Path(key_path).read_bytes()) + artifact = pack(plugin_dir, private_key, + publisher=publisher, out_dir=out_dir) + click.echo("") + click.echo(click.style(f"Packed + signed: {artifact}", fg='green')) @plugin_cli.command('apply-profile') diff --git a/shopdb/plugins/packaging.py b/shopdb/plugins/packaging.py new file mode 100644 index 0000000..2350d8e --- /dev/null +++ b/shopdb/plugins/packaging.py @@ -0,0 +1,131 @@ +"""Pack and verify signed plugin artifacts (.shopdbplugin) - ADR-013 Phase 1. + +A .shopdbplugin is a zip of a plugin directory plus PROVENANCE.json (a per-file +SHA-256 map + metadata) and PROVENANCE.sig (a detached ed25519 signature over +the exact provenance bytes). pack() builds and signs it; verify_artifact() and +verify_dir() re-hash the contents and check the signature against trusted keys. +""" + +import hashlib +import json +import zipfile +from pathlib import Path + +from . import signing + +ARTIFACT_SUFFIX = '.shopdbplugin' + + +def pack(plugin_dir, private_key, publisher: str = '', + out_dir=None, created: str = None) -> Path: + """Build a signed artifact from a plugin directory. Returns its path. + + Caller is expected to have validated the directory first (the CLI does). + """ + plugin_dir = Path(plugin_dir) + manifest = json.loads((plugin_dir / 'manifest.json').read_text()) + name = manifest['name'] + version = manifest['version'] + + provenance = signing.build_provenance( + plugin_dir, name, version, publisher, created) + provenance_bytes = signing.serialize_provenance(provenance) + signature = signing.sign(private_key, provenance_bytes) + + out_dir = Path(out_dir) if out_dir else plugin_dir.parent + out_dir.mkdir(parents=True, exist_ok=True) + artifact = out_dir / f'{name}-{version}{ARTIFACT_SUFFIX}' + + with zipfile.ZipFile(artifact, 'w', zipfile.ZIP_DEFLATED) as archive: + for relpath in provenance['files']: + archive.write(plugin_dir / relpath, relpath) + archive.writestr(signing.PROVENANCE_NAME, provenance_bytes) + archive.writestr(signing.PROVENANCE_SIG, signature) + + return artifact + + +def _verify_map(read_bytes, member_names, public_keys): + """Shared core: given a way to read a member's bytes and the set of member + names, verify signature + every file hash + no unexpected files. + + read_bytes(name) -> bytes. Returns (manifest_or_None, errors_list). + """ + errors = [] + if signing.PROVENANCE_NAME not in member_names \ + or signing.PROVENANCE_SIG not in member_names: + return None, ['missing PROVENANCE.json or PROVENANCE.sig'] + + provenance_bytes = read_bytes(signing.PROVENANCE_NAME) + signature = read_bytes(signing.PROVENANCE_SIG) + + if not public_keys: + errors.append('no trusted keys supplied - signature unverifiable') + elif not signing.verify(public_keys, provenance_bytes, signature): + errors.append('signature does not match any trusted key') + + try: + provenance = json.loads(provenance_bytes) + except json.JSONDecodeError as exc: + return None, errors + [f'PROVENANCE.json is not valid JSON: {exc}'] + + filemap = provenance.get('files', {}) + for relpath, expected in filemap.items(): + if relpath not in member_names: + errors.append(f'declared file missing: {relpath}') + continue + actual = hashlib.sha256(read_bytes(relpath)).hexdigest() + if actual != expected: + errors.append(f'hash mismatch: {relpath}') + + allowed = set(filemap) | {signing.PROVENANCE_NAME, signing.PROVENANCE_SIG} + for member in member_names: + if member not in allowed: + errors.append(f'unexpected file not in provenance: {member}') + + manifest = None + if 'manifest.json' in member_names: + try: + manifest = json.loads(read_bytes('manifest.json')) + except json.JSONDecodeError: + errors.append('manifest.json is not valid JSON') + + return manifest, errors + + +def verify_artifact(artifact_path, public_keys): + """Verify a .shopdbplugin file. Returns (manifest_or_None, errors_list). + + errors empty = signature trusted AND every file hash intact AND no extra + files. public_keys is a list of loaded ed25519 public keys (may be empty, + which fails closed with an 'unverifiable' error). + """ + with zipfile.ZipFile(artifact_path) as archive: + member_names = { + name for name in archive.namelist() if not name.endswith('/')} + return _verify_map(archive.read, member_names, public_keys) + + +def verify_dir(plugin_dir, public_keys): + """Verify an unpacked, adopted plugin directory that carries PROVENANCE + files (used by verify-at-load/migrate in Phase 2). Same guarantees as + verify_artifact. + """ + plugin_dir = Path(plugin_dir) + + member_names = set() + for path in plugin_dir.rglob('*'): + if path.is_file(): + member_names.add(path.relative_to(plugin_dir).as_posix()) + # Drop the noise the packer also excludes, so an on-disk __pycache__ does + # not read as an "unexpected file". + member_names = { + m for m in member_names + if not any(part in signing._EXCLUDE_DIRS for part in Path(m).parts) + and Path(m).suffix not in signing._EXCLUDE_SUFFIXES + } + + def read_bytes(relpath): + return (plugin_dir / relpath).read_bytes() + + return _verify_map(read_bytes, member_names, public_keys) diff --git a/shopdb/plugins/signing.py b/shopdb/plugins/signing.py new file mode 100644 index 0000000..be8627d --- /dev/null +++ b/shopdb/plugins/signing.py @@ -0,0 +1,136 @@ +"""Ed25519 signing + provenance for plugin artifacts (ADR-013 Phase 1). + +A plugin's provenance is a per-file SHA-256 map plus metadata (name, version, +publisher, created). The detached signature covers the EXACT serialized +provenance bytes, so verifying a plugin is: re-hash its files, re-serialize the +provenance the same way, and check the signature. Any changed byte in any file +changes a hash, which changes the serialized provenance, which fails the +signature. The signature proves the artifact is exactly what a curator reviewed +and signed - nothing about what the code does. + +Uses the cryptography package (already a dependency for MySQL 8 auth). +""" + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +PROVENANCE_NAME = 'PROVENANCE.json' +PROVENANCE_SIG = 'PROVENANCE.sig' + +# Never packed, hashed, or counted as an unexpected artifact member. +_EXCLUDE_DIRS = {'__pycache__', '.pytest_cache', '.mypy_cache'} +_EXCLUDE_SUFFIXES = {'.pyc', '.pyo'} +_CHUNK = 65536 + + +def generate_keypair(): + """Return (private_pem, public_pem) as PEM bytes for a new ed25519 key.""" + private_key = Ed25519PrivateKey.generate() + private_pem = private_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + public_pem = private_key.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + return private_pem, public_pem + + +def load_private_key(pem_bytes: bytes): + return serialization.load_pem_private_key(pem_bytes, password=None) + + +def load_public_key(pem_bytes: bytes): + return serialization.load_pem_public_key(pem_bytes) + + +def load_trusted_keys(pem_paths): + """Load public keys from a list of PEM file paths. Missing/unreadable paths + are skipped (they simply cannot vouch for a signature).""" + keys = [] + for path in pem_paths or []: + try: + keys.append(load_public_key(Path(path).read_bytes())) + except (OSError, ValueError): + continue + return keys + + +def _packable(plugin_dir: Path): + """Yield the plugin's real files (sorted, posix relpaths excluded from noise).""" + for path in sorted(plugin_dir.rglob('*')): + if not path.is_file(): + continue + rel = path.relative_to(plugin_dir) + if any(part in _EXCLUDE_DIRS for part in rel.parts): + continue + if path.suffix in _EXCLUDE_SUFFIXES: + continue + if path.name in (PROVENANCE_NAME, PROVENANCE_SIG): + continue + yield rel.as_posix(), path + + +def _sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with open(path, 'rb') as handle: + for chunk in iter(lambda: handle.read(_CHUNK), b''): + digest.update(chunk) + return digest.hexdigest() + + +def build_file_map(plugin_dir) -> dict: + """{posix relpath: sha256hex} for every packable file, sorted.""" + plugin_dir = Path(plugin_dir) + return {relpath: _sha256_file(path) for relpath, path in _packable(plugin_dir)} + + +def build_provenance(plugin_dir, name: str, version: str, + publisher: str = '', created: str = None) -> dict: + """Provenance dict. `created` defaults to now (UTC, second precision).""" + if created is None: + created = datetime.now(timezone.utc).replace( + microsecond=0, tzinfo=None).isoformat() + return { + 'name': name, + 'version': version, + 'publisher': publisher or '', + 'created': created, + 'files': build_file_map(plugin_dir), + } + + +def serialize_provenance(provenance: dict) -> bytes: + """Canonical bytes that get signed AND stored, so sign and verify agree. + + sort_keys + compact separators make this deterministic; the same dict always + serializes to the same bytes regardless of insertion order. + """ + return json.dumps( + provenance, sort_keys=True, separators=(',', ':')).encode('utf-8') + + +def sign(private_key, data: bytes) -> bytes: + return private_key.sign(data) + + +def verify(public_keys, data: bytes, signature: bytes) -> bool: + """True if the signature validates against ANY trusted public key. + + Multiple keys support overlap during key rotation. + """ + for key in public_keys: + try: + key.verify(signature, data) + return True + except InvalidSignature: + continue + return False diff --git a/tests/test_plugin_packaging.py b/tests/test_plugin_packaging.py new file mode 100644 index 0000000..17ad103 --- /dev/null +++ b/tests/test_plugin_packaging.py @@ -0,0 +1,153 @@ +"""Signed plugin artifact packaging tests (ADR-013 Phase 1). + +Covers the provenance + ed25519 signing that the marketplace trust model rests +on: a valid signature over an intact file set verifies, and every tampering +path (flipped file byte, wrong key, missing/extra file, no key) fails closed. +""" + +import json +import zipfile + +import pytest + +from shopdb.plugins import signing, packaging + + +def _write_plugin(plugin_dir, name='demo', version='1.0.0', deps=None): + plugin_dir.mkdir(parents=True, exist_ok=True) + (plugin_dir / 'manifest.json').write_text(json.dumps({ + 'name': name, 'version': version, 'description': 'demo', + 'dependencies': deps or [], 'core_version': '>=0.1.0,<1.0.0', + 'api_prefix': f'/api/{name}', + })) + (plugin_dir / 'plugin.py').write_text('# demo plugin\n') + (plugin_dir / 'models').mkdir() + (plugin_dir / 'models' / '__init__.py').write_text('') + # noise that must NOT be packed or hashed + (plugin_dir / '__pycache__').mkdir() + (plugin_dir / '__pycache__' / 'x.pyc').write_bytes(b'\x00\x01') + return plugin_dir + + +# --- signing primitives ----------------------------------------------------- + +def test_sign_verify_round_trip(): + private_pem, public_pem = signing.generate_keypair() + priv = signing.load_private_key(private_pem) + pub = signing.load_public_key(public_pem) + + data = b'provenance bytes' + sig = signing.sign(priv, data) + + assert signing.verify([pub], data, sig) is True + assert signing.verify([pub], b'other bytes', sig) is False + + +def test_wrong_key_does_not_verify(): + priv = signing.load_private_key(signing.generate_keypair()[0]) + other_pub = signing.load_public_key(signing.generate_keypair()[1]) + sig = signing.sign(priv, b'data') + assert signing.verify([other_pub], b'data', sig) is False + + +def test_provenance_excludes_noise(tmp_path): + plugin_dir = _write_plugin(tmp_path / 'demo') + filemap = signing.build_file_map(plugin_dir) + assert 'manifest.json' in filemap + assert 'models/__init__.py' in filemap + assert not any('__pycache__' in k or k.endswith('.pyc') for k in filemap) + assert signing.PROVENANCE_NAME not in filemap + + +def test_serialize_provenance_is_order_independent(): + a = {'name': 'x', 'version': '1', 'files': {'b': '2', 'a': '1'}} + b = {'files': {'a': '1', 'b': '2'}, 'version': '1', 'name': 'x'} + assert signing.serialize_provenance(a) == signing.serialize_provenance(b) + + +# --- pack + verify ---------------------------------------------------------- + +@pytest.fixture +def keypair(): + private_pem, public_pem = signing.generate_keypair() + return (signing.load_private_key(private_pem), + signing.load_public_key(public_pem)) + + +def test_pack_then_verify_ok(tmp_path, keypair): + priv, pub = keypair + plugin_dir = _write_plugin(tmp_path / 'demo') + + artifact = packaging.pack(plugin_dir, priv, publisher='site-a', + out_dir=tmp_path / 'dist') + assert artifact.name == 'demo-1.0.0.shopdbplugin' + + manifest, errors = packaging.verify_artifact(artifact, [pub]) + assert errors == [] + assert manifest['name'] == 'demo' + + +def test_verify_fails_with_no_keys(tmp_path, keypair): + priv, _ = keypair + artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv, + out_dir=tmp_path / 'dist') + _, errors = packaging.verify_artifact(artifact, []) + assert any('unverifiable' in e for e in errors) + + +def test_verify_fails_with_wrong_key(tmp_path, keypair): + priv, _ = keypair + wrong_pub = signing.load_public_key(signing.generate_keypair()[1]) + artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv, + out_dir=tmp_path / 'dist') + _, errors = packaging.verify_artifact(artifact, [wrong_pub]) + assert any('signature does not match' in e for e in errors) + + +def test_verify_detects_tampered_file(tmp_path, keypair): + priv, pub = keypair + artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv, + out_dir=tmp_path / 'dist') + + # Rewrite the zip with one member's bytes changed, keeping provenance + sig. + with zipfile.ZipFile(artifact) as zin: + members = {n: zin.read(n) for n in zin.namelist()} + members['manifest.json'] += b'tampered' + with zipfile.ZipFile(artifact, 'w', zipfile.ZIP_DEFLATED) as zout: + for name, data in members.items(): + zout.writestr(name, data) + + _, errors = packaging.verify_artifact(artifact, [pub]) + assert any('hash mismatch' in e for e in errors) + + +def test_verify_detects_extra_file(tmp_path, keypair): + priv, pub = keypair + artifact = packaging.pack(_write_plugin(tmp_path / 'demo'), priv, + out_dir=tmp_path / 'dist') + with zipfile.ZipFile(artifact, 'a', zipfile.ZIP_DEFLATED) as z: + z.writestr('sneaky.py', 'print("surprise")') + + _, errors = packaging.verify_artifact(artifact, [pub]) + assert any('unexpected file' in e for e in errors) + + +def test_verify_dir_round_trip(tmp_path, keypair): + """An unpacked directory carrying PROVENANCE files verifies (Phase 2 uses + this for verify-at-load).""" + priv, pub = keypair + plugin_dir = _write_plugin(tmp_path / 'demo') + + provenance = signing.build_provenance(plugin_dir, 'demo', '1.0.0', 'site-a') + prov_bytes = signing.serialize_provenance(provenance) + (plugin_dir / signing.PROVENANCE_NAME).write_bytes(prov_bytes) + (plugin_dir / signing.PROVENANCE_SIG).write_bytes( + signing.sign(priv, prov_bytes)) + + _, errors = packaging.verify_dir(plugin_dir, [pub]) + assert errors == [] + + # Tamper a file on disk -> hash mismatch. + (plugin_dir / 'plugin.py').write_text('# changed\n') + _, errors = packaging.verify_dir(plugin_dir, [pub]) + assert any('hash mismatch' in e for e in errors)