ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ADR-013 Phase 1: signed plugin artifacts (pack/validate/keygen)

Browse Source 
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 <name>-<version>.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 <name>
  --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.
This commit is contained in:
cproudlock
2026-07-18 20:21:57 -04:00
parent d178726687
commit 86f5f1be68
7 changed files with 687 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.env.example
View File

@@ -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

75
docs/PLUGIN-SIGNING.md Normal file
View File

@@ -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-<version>.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.

10
shopdb/config.py
View File

@@ -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', '')

203
shopdb/plugins/cli.py
View File

@@ -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 <name>.key (PRIVATE - keep offline with the curator) and <name>.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 <name>-<version>.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')

131
shopdb/plugins/packaging.py Normal file
View File

@@ -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)

136
shopdb/plugins/signing.py Normal file
View File

@@ -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

153
tests/test_plugin_packaging.py Normal file
View File

@@ -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)