From 23cd81c6bc3b8564bb326b0422832e3e11a633f4 Mon Sep 17 00:00:00 2001 From: olatechy0 <302475381+olatechy0@users.noreply.github.com> Date: Sat, 17 Oct 2026 16:57:30 +0100 Subject: [PATCH] feat(plugins): publisher trust, signed catalog installs, and lock entries Implements the Phase 9 trust/install/lock slice: - #386: strict PLUGIN_TRUSTED_KEYS loading (canonical ids, real Ed25519 keys), startup validation via create_app, and secret-free publisher listings (flask plugins publishers, GET /plugins/api/trusted-publishers). - #385: detached Ed25519 manifest signature verification over the canonical manifest body, shared with the embedded-signature path; forged/untrusted signatures always fail, unsigned entries fail under PLUGIN_TRUST_POLICY=required. - #389: validated operator catalog (PLUGIN_CATALOG_FILE) and owner-only POST .../plugins/install-from-catalog; registers metadata only, audits trust refusals, never downloads code or grants capabilities. - #394: PluginLockEntry model (owner/workspace scoped) serialized through the canonical deterministic lockfile builder. Docs and .env.example updated; tests cover keys, signatures, catalog install failure/authorization paths, and lock serialization. --- .env.example | 11 +- app/__init__.py | 9 + app/config.py | 6 + app/models/__init__.py | 3 +- app/models/plugin.py | 96 +++++ app/plugins/routes.py | 266 +++++++++++-- app/services/plugin_catalog.py | 350 +++++++++++++++++ app/services/plugin_signatures.py | 221 +++++++++++ app/services/plugin_trust.py | 215 ++++++++++ app/services/plugins.py | 76 +--- app/services/plugins_cli.py | 30 +- docs/plugins.md | 162 +++++++- docs/security.md | 18 + tests/test_plugin_catalog_install.py | 560 +++++++++++++++++++++++++++ tests/test_plugin_lock_entries.py | 145 +++++++ tests/test_plugin_signatures.py | 255 ++++++++++++ tests/test_plugin_trust.py | 225 +++++++++++ 17 files changed, 2545 insertions(+), 103 deletions(-) create mode 100644 app/services/plugin_catalog.py create mode 100644 app/services/plugin_signatures.py create mode 100644 app/services/plugin_trust.py create mode 100644 tests/test_plugin_catalog_install.py create mode 100644 tests/test_plugin_lock_entries.py create mode 100644 tests/test_plugin_signatures.py create mode 100644 tests/test_plugin_trust.py diff --git a/.env.example b/.env.example index c612985..d2eb537 100644 --- a/.env.example +++ b/.env.example @@ -39,10 +39,19 @@ LLM_CACHE_MAX_ENTRIES=256 # Plugin manifest trust (optional). Unsigned plugins remain allowed by default. # Set PLUGIN_TRUST_POLICY=required to reject unsigned or invalid signatures. -# JSON object mapping publisher ids to base64 Ed25519 public keys. +# JSON object mapping canonical publisher ids to Ed25519 public keys; each key +# may be base64-encoded raw 32 bytes or a PEM SubjectPublicKeyInfo. Invalid key +# material aborts startup instead of silently disabling verification. PLUGIN_TRUST_POLICY=if-present PLUGIN_TRUSTED_KEYS={} +# Operator-controlled catalog for POST .../plugins/install-from-catalog +# (Phase 9, issue #389). Empty disables remote installation. The file is a JSON +# catalog of validated entries: id, version, https source, sha256 integrity, +# optional detached Ed25519 signature, and a full manifest. See +# docs/plugins.md. +PLUGIN_CATALOG_FILE= + # Plugin execution budget (Phase 9, issue #408): seconds an event handler or # lifecycle hook may run before it is abandoned (daemon thread), recorded as a # bounded error report, and dispatch continues. 0 disables the bound. diff --git a/app/__init__.py b/app/__init__.py index f20e058..f51d466 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -32,6 +32,15 @@ def create_app(config_name: str | None = None) -> Flask: app = Flask(__name__) app.config.from_object(config_by_name[config_name]) + # Reject malformed plugin trust configuration at startup instead of + # silently treating every publisher as untrusted (#386). + from app.services.plugin_trust import PluginTrustConfigError, validate_trust_config + + try: + validate_trust_config(app.config) + except PluginTrustConfigError as exc: + raise RuntimeError(f"Invalid plugin trust configuration: {exc}") from exc + # Initialize extensions with the application. db.init_app(app) migrate.init_app(app, db) diff --git a/app/config.py b/app/config.py index ad867c8..c9a14fc 100644 --- a/app/config.py +++ b/app/config.py @@ -115,6 +115,12 @@ class Config: PLUGIN_TRUST_POLICY = os.getenv("PLUGIN_TRUST_POLICY", "if-present") PLUGIN_TRUSTED_KEYS = _trusted_plugin_keys() + # Operator-controlled catalog used by the remote-install endpoint (#389). + # Empty by default so remote installation is disabled until an operator + # points it at a vetted catalog file; the endpoint never accepts a + # client-supplied manifest or URL. See ``app/services/plugin_catalog.py``. + PLUGIN_CATALOG_FILE = os.getenv("PLUGIN_CATALOG_FILE", "") + # Directory scanned by the read-only plugin discovery endpoint (#172). # Discovery only reads ``manifest.json`` files from plugin subdirectories # and never installs, loads, or mutates anything. Defaults to the repo's diff --git a/app/models/__init__.py b/app/models/__init__.py index 6dbf994..7274855 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -17,7 +17,7 @@ from app.models.message_attachment import MessageAttachment from app.models.notification import Notification from app.models.notification_preference import NotificationPreference -from app.models.plugin import CapabilityGrant, Plugin, PluginInstallation +from app.models.plugin import CapabilityGrant, Plugin, PluginInstallation, PluginLockEntry from app.models.plugin_error_report import PluginErrorReport from app.models.project import Project from app.models.project_chat_session import ProjectChatSession @@ -53,6 +53,7 @@ "Plugin", "PluginErrorReport", "PluginInstallation", + "PluginLockEntry", "Project", "ProjectChatSession", "ProjectComment", diff --git a/app/models/plugin.py b/app/models/plugin.py index 8a295ba..40bdba7 100644 --- a/app/models/plugin.py +++ b/app/models/plugin.py @@ -176,3 +176,99 @@ def __repr__(self) -> str: f"" ) + + +class PluginLockEntry(db.Model): + """A resolved plugin pinned for reproducible installs (#394). + + A lock row records exactly what a workspace resolved: the plugin id, the + pinned version, where it came from, and the sha256 integrity hash of the + artifact. Rows are scoped to both the workspace and its owner, so one + workspace can never read or overwrite another's lockfile, and a lock entry + never grants a capability — grants stay explicit and live in + ``CapabilityGrant``. + + The row is intentionally free of volatile fields in its lockfile form: + ``resolved_at`` is for operators, while + :meth:`~app.services.plugin_lockfile.build_lockfile` serializes only the + stable fields (and accepts these rows directly), keeping lockfiles + byte-stable across resolutions. + """ + + __tablename__ = "plugin_lock_entries" + + id = db.Column(db.Integer, primary_key=True) + workspace_id = db.Column( + db.Integer, + db.ForeignKey("workspaces.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + owner_id = db.Column( + db.Integer, + db.ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + plugin_id = db.Column(db.String(64), nullable=False, index=True) + version = db.Column(db.String(32), nullable=False) + source = db.Column(db.String(512), nullable=False) + integrity = db.Column(db.String(128), nullable=False) + dependencies = db.Column(db.JSON, default=[], nullable=False) + + resolved_at = db.Column( + db.DateTime, + default=lambda: datetime.now(UTC), + nullable=False, + ) + + # Relationships + workspace = db.relationship("Workspace", foreign_keys=[workspace_id], lazy="select") + owner = db.relationship("User", foreign_keys=[owner_id], lazy="select") + + __table_args__ = ( + db.UniqueConstraint("workspace_id", "plugin_id", name="uq_plugin_lock_entry"), + ) + + def to_lockfile_entry(self): + """Return the canonical :class:`LockfileEntry` for this row. + + Imported lazily so the model module does not depend on the service + layer at import time. + """ + from app.services.plugin_lockfile import LockfileEntry, normalize_integrity + + return LockfileEntry( + plugin_id=self.plugin_id, + version=self.version, + source=self.source, + integrity=normalize_integrity(self.integrity), + dependencies=tuple(self.dependencies or ()), + ) + + def to_dict(self) -> dict: + """Convert to dictionary representation. + + The result includes audit fields (``id``, ``workspace_id``, + ``owner_id``, ``resolved_at``) for API reads. Use + :meth:`to_lockfile_entry` (or pass the row to ``build_lockfile``) when + producing the canonical, byte-stable lockfile. + """ + return { + "id": self.id, + "workspace_id": self.workspace_id, + "owner_id": self.owner_id, + "plugin_id": self.plugin_id, + "version": self.version, + "source": self.source, + "integrity": self.integrity, + "dependencies": self.dependencies or [], + "resolved_at": self.resolved_at.isoformat() if self.resolved_at else None, + } + + def __repr__(self) -> str: + """String representation.""" + return ( + f"" + ) diff --git a/app/plugins/routes.py b/app/plugins/routes.py index 8211d0a..285150f 100644 --- a/app/plugins/routes.py +++ b/app/plugins/routes.py @@ -24,6 +24,12 @@ from app.services.events import emit_event from app.services.permissions import require_workspace_capability, resolve_workspace, role_for from app.services.plugin_audit import record_plugin_audit +from app.services.plugin_catalog import ( + CatalogEntryNotFoundError, + CatalogError, + CatalogUnavailableError, + load_plugin_catalog, +) from app.services.plugin_compat import ( IncompatibleCompatibilityError, InvalidCompatibilityError, @@ -39,9 +45,22 @@ discover_plugins, update_plugin, ) +from app.services.plugin_signatures import ( + VERIFY_IF_PRESENT, + PublisherSignatureError, + enforce_publisher_trust, +) +from app.services.plugin_trust import ( + PluginTrustConfigError, + trusted_publisher_ids, +) from app.services.plugins import ManifestValidationError, PluginError, PluginManifest +class PluginAlreadyInstalledError(ValueError): + """Raised when a plugin is already installed in the workspace.""" + + def _accessible_workspaces(user): """Return the workspaces ``user`` owns or is an active member of.""" seen: dict[int, Workspace] = {} @@ -103,6 +122,78 @@ def _capability_name(value: str) -> str | None: return None +def _upsert_plugin(manifest: PluginManifest) -> Plugin: + """Create or reuse the global ``Plugin`` row for a validated manifest. + + The manifest id is the identity, and an existing plugin keeps its entry + point: a conflicting entry point raises + :class:`PluginEntryPointMismatchError` so an install/update can never + retarget registered code. + """ + plugin = Plugin.query.filter_by(id=manifest.id).first() + if plugin is None: + plugin = Plugin( + id=manifest.id, + name=manifest.name, + version=manifest.version, + description=manifest.description, + author=manifest.author, + entry_point=manifest.entry_point, + capabilities=manifest.capabilities, + permissions=manifest.permissions or [], + dependencies=manifest.dependencies or [], + compatibility=manifest.compatibility, + configuration=manifest.configuration or {}, + trust_state=manifest.trust_state, + trust_publisher=manifest.trust_publisher, + ) + db.session.add(plugin) + return plugin + + if plugin.entry_point != manifest.entry_point: + raise PluginEntryPointMismatchError( + "Manifest entry_point does not match the registered plugin." + ) + + # Trust metadata is derived from the verified signature, so refresh it even + # when the global row already exists (for example after a key rotation). + plugin.trust_state = manifest.trust_state + plugin.trust_publisher = manifest.trust_publisher + return plugin + + +def _install_in_workspace(manifest: PluginManifest, workspace_id: int) -> Plugin: + """Register ``manifest`` and install it in ``workspace_id``. + + Shared by the local-manifest and catalog install routes so both paths keep + the same identity binding and duplicate-install handling. Neither path + creates capability grants. Returns the persisted plugin. + + Raises: + PluginAlreadyInstalledError: If the plugin is already installed in the + workspace (or a concurrent request won the unique constraint). + """ + plugin = _upsert_plugin(manifest) + if _installation(manifest.id, workspace_id) is not None: + db.session.rollback() + raise PluginAlreadyInstalledError + + installation = PluginInstallation( + plugin_id=manifest.id, + workspace_id=workspace_id, + enabled=True, + config=manifest.configuration or {}, + installed_by_id=current_user.id, + ) + db.session.add(installation) + try: + db.session.commit() + except Exception: # uniqueness races map to a duplicate install + db.session.rollback() + raise PluginAlreadyInstalledError from None + return plugin + + # -------------------------------------------------------------------------- # Pages # -------------------------------------------------------------------------- @@ -164,6 +255,33 @@ def api_inspect_plugin(workspace_id: int, plugin_id: str): return jsonify(_serialize(plugin, workspace_id)) +# -------------------------------------------------------------------------- +# API: trusted publisher key management (read-only, #386) +# -------------------------------------------------------------------------- + + +@bp.route("/api/trusted-publishers", methods=["GET"]) +@login_required +def api_list_trusted_publishers(): + """List the configured trusted publisher ids (never key material). + + The configured keys live in ``PLUGIN_TRUSTED_KEYS`` (an operator secret + setting). This read-only surface lets an operator confirm *which* publisher + ids are trusted and under which policy, without exposing any key bytes or + fingerprints. Invalid key material is rejected when the application loads + (``app.services.plugin_trust.validate_trust_config``). + """ + publishers = trusted_publisher_ids(current_app.config.get("PLUGIN_TRUSTED_KEYS") or {}) + policy = current_app.config.get("PLUGIN_TRUST_POLICY", VERIFY_IF_PRESENT) + return jsonify( + { + "publishers": publishers, + "count": len(publishers), + "policy": policy, + } + ) + + # -------------------------------------------------------------------------- # API: plugin discovery (read-only, local-directory based) # -------------------------------------------------------------------------- @@ -220,46 +338,130 @@ def api_install_plugin(workspace_id: int): except (IncompatibleCompatibilityError, InvalidCompatibilityError) as exc: return jsonify({"error": str(exc)}), 400 - plugin = Plugin.query.filter_by(id=manifest.id).first() - if plugin is None: - plugin = Plugin( - id=manifest.id, - name=manifest.name, - version=manifest.version, - description=manifest.description, - author=manifest.author, - entry_point=manifest.entry_point, - capabilities=manifest.capabilities, - permissions=manifest.permissions or [], - dependencies=manifest.dependencies or [], - compatibility=manifest.compatibility, - configuration=manifest.configuration or {}, - trust_state=manifest.trust_state, - trust_publisher=manifest.trust_publisher, - ) - db.session.add(plugin) - elif plugin.entry_point != manifest.entry_point: - return jsonify({"error": "Manifest entry_point does not match the registered plugin."}), 409 - - if _installation(manifest.id, workspace_id) is not None: + try: + plugin = _install_in_workspace(manifest, workspace_id) + except PluginEntryPointMismatchError: db.session.rollback() + return jsonify({"error": "Manifest entry_point does not match the registered plugin."}), 409 + except PluginAlreadyInstalledError: return jsonify({"error": "Plugin is already installed in this workspace."}), 409 - installation = PluginInstallation( - plugin_id=manifest.id, - workspace_id=workspace_id, - enabled=True, - config=manifest.configuration or {}, - installed_by_id=current_user.id, - ) - db.session.add(installation) + return jsonify(_serialize(plugin, workspace_id)), 201 + + +@bp.route( + "/api/workspaces//plugins/install-from-catalog", + methods=["POST"], +) +@login_required +@require_workspace_capability("manage_plugins") +def api_install_plugin_from_catalog(workspace_id: int): + """Install a plugin pinned by the operator-controlled catalog (#389). + + Owner-only. The request names a catalog entry (``plugin_id`` plus an + optional ``version``); a client cannot supply a manifest, an artifact URL, + or a publisher key. The entry is validated against the catalog, the + manifest is re-validated against the manifest schema, and the active + ``PLUGIN_TRUST_POLICY`` is enforced *before* anything is written: + + - unsigned entries are refused when the policy is ``required``; + - a forged, malformed, or untrusted signature is always refused; + - the artifact sha256 from the catalog is recorded in the response. + + Success registers metadata only: no artifact is downloaded, no plugin code + is executed, and **no capability grant is created**. Grants remain explicit + per workspace. + """ + data = request.get_json(silent=True) or {} + plugin_id = data.get("plugin_id") + version = data.get("version") + + if not isinstance(plugin_id, str) or not plugin_id.strip(): + return jsonify({"error": "plugin_id is required."}), 400 + if version is not None and (not isinstance(version, str) or not version.strip()): + return jsonify({"error": "version must be a non-empty string when provided."}), 400 + + try: + catalog = load_plugin_catalog(current_app.config) + except CatalogUnavailableError as exc: + return jsonify({"error": str(exc)}), 404 + except CatalogError as exc: + return jsonify({"error": f"Invalid plugin catalog: {exc}"}), 400 + try: + entry = catalog.get( + plugin_id.strip(), + version.strip() if version is not None else None, + ) + except CatalogEntryNotFoundError as exc: + return jsonify({"error": str(exc)}), 404 + except CatalogError as exc: + return jsonify({"error": f"Invalid plugin version: {exc}"}), 400 + + trusted_publishers = current_app.config.get("PLUGIN_TRUSTED_KEYS") or {} + try: + # Schema validation is trust-policy agnostic here: the detached (or + # embedded) signature is enforced separately below so a catalog entry + # signed out-of-band is not rejected as "unsigned". + manifest = PluginManifest.from_dict( + entry.manifest, + trusted_publishers=trusted_publishers, + trust_policy=VERIFY_IF_PRESENT, + ) + except ManifestValidationError as exc: + return jsonify({"error": f"Invalid plugin manifest: {exc}"}), 400 + except PluginError as exc: + return jsonify({"error": str(exc)}), 400 + + policy = current_app.config.get("PLUGIN_TRUST_POLICY", VERIFY_IF_PRESENT) + # A catalog entry may carry the publisher signature detached (catalog + # field) or embedded in the manifest; both sign the same canonical body, so + # the effective signature is whichever was supplied. + signature = entry.signature if entry.signature is not None else entry.manifest.get("signature") + try: + trust = enforce_publisher_trust( + entry.manifest, + signature, + trusted_publishers=trusted_publishers, + policy=policy, + ) + except PluginTrustConfigError as exc: + return jsonify({"error": str(exc)}), 400 + except PublisherSignatureError as exc: + # Refusals are auditable but never store artifact bytes or key material. + record_plugin_audit( + workspace_id, + "denied", + actor=current_user, + plugin_id=entry.plugin_id, + outcome="denied", + reason="publisher signature refused", + ) db.session.commit() - except Exception: # uniqueness races map to a duplicate install + return jsonify({"error": str(exc)}), 403 + + try: + enforce_compatible(manifest.compatibility, subject=f"Plugin {manifest.id}") + except (IncompatibleCompatibilityError, InvalidCompatibilityError) as exc: + return jsonify({"error": str(exc)}), 400 + + # The verified trust fact is what gets persisted, not whatever the catalog + # (or manifest) claimed. + manifest.trust_state = trust.state + manifest.trust_publisher = trust.publisher + + try: + plugin = _install_in_workspace(manifest, workspace_id) + except PluginEntryPointMismatchError: db.session.rollback() + return jsonify({"error": "Manifest entry_point does not match the registered plugin."}), 409 + except PluginAlreadyInstalledError: return jsonify({"error": "Plugin is already installed in this workspace."}), 409 - return jsonify(_serialize(plugin, workspace_id)), 201 + payload = _serialize(plugin, workspace_id) + payload["source"] = entry.source + payload["integrity"] = entry.integrity + return jsonify(payload), 201 @bp.route("/api/workspaces//plugins//update", methods=["POST"]) diff --git a/app/services/plugin_catalog.py b/app/services/plugin_catalog.py new file mode 100644 index 0000000..f43cd1a --- /dev/null +++ b/app/services/plugin_catalog.py @@ -0,0 +1,350 @@ +"""Validated plugin catalog entries used by remote installation (#389). + +The remote-install endpoint never trusts a client-supplied manifest: it resolves +``(plugin_id, version)`` against the operator-controlled catalog and installs +only what that catalog entry describes. This module is the smallest catalog +surface that makes that contract testable while the full marketplace catalog +issues (#373-#381) land: + +``PLUGIN_CATALOG_FILE`` points at a JSON document:: + + { + "version": 1, + "plugins": [ + { + "id": "hello-world", + "version": "0.1.0", + "source": "https://plugins.example.com/hello-world-0.1.0.zip", + "integrity": "sha256:...", + "signature": {"publisher": "acme", "algorithm": "ed25519", "value": "..."}, + "manifest": { ...a full manifest.json object... } + } + ] + } + +Every entry is validated at load time: ids and versions are canonical, the +source is https (or an opaque local identifier), ``integrity`` is a sha256 +digest, the signature shape is well formed, and the embedded manifest describes +the same plugin id and version. Loading performs **no network access** and the +install path executes no plugin code. + +Nothing here is a secret: the catalog carries public metadata only. +""" + +from __future__ import annotations + +import json +import re +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from packaging.version import InvalidVersion, Version + +from app.services.plugin_lockfile import LockfileError, normalize_integrity +from app.services.plugin_signatures import validate_signature_metadata + +#: Catalog document version understood by this loader. +CATALOG_VERSION = 1 + +#: Refuse absurdly large catalog files instead of reading them into memory. +MAX_CATALOG_BYTES = 1024 * 1024 + +#: Refuse catalogs with more entries than the installer can reasonably serve. +MAX_CATALOG_ENTRIES = 500 + +#: Longest accepted ``source`` reference. +MAX_SOURCE_LENGTH = 512 + +#: Manifest/plugin id format, matching ``plugins/plugin.schema.json``. +_PLUGIN_ID_RE = re.compile(r"^[a-z][a-z0-9_-]{0,63}$") + + +class CatalogError(ValueError): + """The plugin catalog or one of its entries is invalid.""" + + +class CatalogUnavailableError(CatalogError): + """No catalog is configured, or the configured file cannot be read.""" + + +class CatalogEntryNotFoundError(CatalogError): + """The requested plugin id/version is not present in the catalog.""" + + +@dataclass(frozen=True) +class CatalogEntry: + """One validated catalog entry. + + ``manifest`` is a plain mapping so it can be validated with + :meth:`~app.services.plugins.PluginManifest.from_dict` exactly like a + manifest uploaded by a client; ``signature`` is the *detached* publisher + signature for that manifest (it may be ``None``). + """ + + plugin_id: str + version: str + source: str + integrity: str + manifest: Mapping[str, Any] + signature: Mapping[str, str] | None = None + + @property + def publisher(self) -> str | None: + """Return the detached-signature publisher, when present.""" + if self.signature is None: + return None + return self.signature.get("publisher") + + def to_dict(self) -> dict[str, Any]: + """Return the entry as JSON-serializable data.""" + data: dict[str, Any] = { + "id": self.plugin_id, + "version": self.version, + "source": self.source, + "integrity": self.integrity, + "manifest": dict(self.manifest), + } + if self.signature is not None: + data["signature"] = dict(self.signature) + return data + + +def _require_mapping(value: Any, label: str) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise CatalogError(f"{label} must be an object") + return value + + +def _parse_version(value: Any, label: str) -> Version: + if not isinstance(value, str) or not value.strip(): + raise CatalogError(f"{label} is required") + try: + return Version(value.strip()) + except InvalidVersion as exc: + raise CatalogError(f"{label} is not a valid version: {value!r}") from exc + + +def _parse_source(value: Any) -> str: + if not isinstance(value, str) or not value.strip(): + raise CatalogError("catalog entry source is required") + source = value.strip() + if len(source) > MAX_SOURCE_LENGTH: + raise CatalogError(f"catalog entry source must be at most {MAX_SOURCE_LENGTH} characters") + if "://" in source: + scheme = source.split("://", 1)[0].lower() + if scheme != "https": + raise CatalogError("catalog entry source URLs must use https") + return source + + +def _parse_manifest(entry_id: str, entry_version: Version, value: Any) -> Mapping[str, Any]: + manifest = _require_mapping(value, "catalog entry manifest") + manifest_id = manifest.get("id") + if manifest_id != entry_id: + raise CatalogError( + f"catalog entry id {entry_id!r} does not match the manifest id {manifest_id!r}" + ) + manifest_version = _parse_version(manifest.get("version"), "catalog entry manifest version") + if manifest_version != entry_version: + raise CatalogError( + f"catalog entry version {entry_version} does not match the manifest " + f"version {manifest_version}" + ) + return dict(manifest) + + +def _parse_signature(value: Any) -> Mapping[str, str] | None: + if value is None: + return None + error = validate_signature_metadata(value) + if error: + raise CatalogError(error) + return {key: value[key] for key in ("publisher", "algorithm", "value")} + + +def parse_catalog_entry(value: Any) -> CatalogEntry: + """Validate one catalog entry mapping and return it. + + Raises: + CatalogError: If any required field is missing or malformed. + """ + entry = _require_mapping(value, "catalog entry") + + entry_id = entry.get("id") + if not isinstance(entry_id, str) or not _PLUGIN_ID_RE.match(entry_id): + raise CatalogError( + "catalog entry id must start with a lowercase letter and contain only " + "lowercase letters, numbers, hyphens, and underscores" + ) + + version = _parse_version(entry.get("version"), "catalog entry version") + source = _parse_source(entry.get("source")) + + try: + integrity = normalize_integrity(entry.get("integrity")) + except LockfileError as exc: + raise CatalogError(f"catalog entry integrity is invalid: {exc}") from exc + + signature = _parse_signature(entry.get("signature")) + manifest = _parse_manifest(entry_id, version, entry.get("manifest")) + + return CatalogEntry( + plugin_id=entry_id, + version=version.public, + source=source, + integrity=integrity, + manifest=manifest, + signature=signature, + ) + + +class PluginCatalog: + """An immutable, indexed collection of validated :class:`CatalogEntry` rows.""" + + def __init__(self, entries: list[CatalogEntry]): + ordered = sorted(entries, key=lambda item: (item.plugin_id, Version(item.version))) + self._entries = tuple(ordered) + self._by_id: dict[str, list[CatalogEntry]] = {} + for entry in ordered: + self._by_id.setdefault(entry.plugin_id, []).append(entry) + + def __len__(self) -> int: + """Number of catalog entries.""" + return len(self._entries) + + def list_entries(self) -> tuple[CatalogEntry, ...]: + """Return every entry, sorted by plugin id and version.""" + return self._entries + + def get(self, plugin_id: str, version: str | None = None) -> CatalogEntry: + """Return the entry for ``plugin_id`` (and ``version`` when given). + + Without a version the newest compatible entry for the plugin is + returned, using PEP 440 ordering. + + Raises: + CatalogEntryNotFoundError: If the id or version is not catalogued. + """ + matches = self._by_id.get(plugin_id, []) + if version is None: + if not matches: + raise CatalogEntryNotFoundError( + f"Plugin {plugin_id!r} is not in the plugin catalog." + ) + return matches[-1] + + requested = _parse_version(version, "version") + for entry in matches: + if Version(entry.version) == requested: + return entry + raise CatalogEntryNotFoundError( + f"Plugin {plugin_id!r} version {requested} is not in the plugin catalog." + ) + + +def parse_catalog(document: Any) -> PluginCatalog: + """Validate a catalog document and return a :class:`PluginCatalog`. + + Raises: + CatalogError: For a malformed document or invalid/duplicate entries. + """ + data = _require_mapping(document, "plugin catalog") + document_version = data.get("version", CATALOG_VERSION) + if document_version != CATALOG_VERSION: + raise CatalogError( + f"unsupported plugin catalog version {document_version!r} " + f"(expected {CATALOG_VERSION})" + ) + + raw_entries = data.get("plugins") + if not isinstance(raw_entries, list): + raise CatalogError("plugin catalog 'plugins' must be a list") + if len(raw_entries) > MAX_CATALOG_ENTRIES: + raise CatalogError(f"plugin catalog must contain at most {MAX_CATALOG_ENTRIES} entries") + + entries: list[CatalogEntry] = [] + seen: set[tuple[str, Version]] = set() + for index, raw_entry in enumerate(raw_entries): + try: + entry = parse_catalog_entry(raw_entry) + except CatalogError as exc: + raise CatalogError(f"catalog entry {index}: {exc}") from exc + + key = (entry.plugin_id, Version(entry.version)) + if key in seen: + raise CatalogError( + f"duplicate catalog entry for plugin {entry.plugin_id!r} " + f"version {entry.version!r}" + ) + seen.add(key) + entries.append(entry) + + return PluginCatalog(entries) + + +def load_catalog_file(path: str | Path) -> PluginCatalog: + """Read and validate the catalog JSON at ``path``. + + Raises: + CatalogUnavailableError: If the file is missing, unreadable, or too big. + CatalogError: If the contents are not a valid catalog. + """ + catalog_path = Path(path) + try: + size = catalog_path.stat().st_size + except OSError as exc: + raise CatalogUnavailableError( + f"Plugin catalog file could not be read: {catalog_path}" + ) from exc + if size > MAX_CATALOG_BYTES: + raise CatalogUnavailableError( + f"Plugin catalog file exceeds {MAX_CATALOG_BYTES} bytes: {catalog_path}" + ) + + try: + with catalog_path.open(encoding="utf-8") as handle: + document = json.load(handle) + except json.JSONDecodeError as exc: + raise CatalogError(f"Plugin catalog is not valid JSON: {exc}") from exc + except OSError as exc: + raise CatalogUnavailableError( + f"Plugin catalog file could not be read: {catalog_path}" + ) from exc + + return parse_catalog(document) + + +def load_plugin_catalog(config: Mapping[str, Any]) -> PluginCatalog: + """Load the catalog configured for ``config``. + + ``PLUGIN_CATALOG_FILE`` is empty by default, which deliberately disables + remote installation rather than silently reading an unrelated file. + + Raises: + CatalogUnavailableError: If no catalog is configured or it cannot be read. + CatalogError: If the configured catalog is invalid. + """ + configured = config.get("PLUGIN_CATALOG_FILE") + if not configured: + raise CatalogUnavailableError( + "Remote plugin installation is not configured (set PLUGIN_CATALOG_FILE)." + ) + return load_catalog_file(configured) + + +__all__ = [ + "CATALOG_VERSION", + "MAX_CATALOG_BYTES", + "MAX_CATALOG_ENTRIES", + "CatalogEntry", + "CatalogEntryNotFoundError", + "CatalogError", + "CatalogUnavailableError", + "PluginCatalog", + "load_catalog_file", + "load_plugin_catalog", + "parse_catalog", + "parse_catalog_entry", +] diff --git a/app/services/plugin_signatures.py b/app/services/plugin_signatures.py new file mode 100644 index 0000000..44dffb0 --- /dev/null +++ b/app/services/plugin_signatures.py @@ -0,0 +1,221 @@ +"""Detached publisher signature verification (#385). + +A publisher signs the canonical JSON body of a plugin manifest with an Ed25519 +private key. The signature may travel inside the manifest (``signature``) or +detached from it (a ``.sig`` file or a catalog field) — both forms sign exactly +the same bytes, so a detached signature can never be used to authenticate +metadata the embedded form would reject. + +The signed payload is :func:`canonical_manifest_body`: the manifest as compact, +key-sorted JSON with ``signature`` and the runtime-managed ``trust_state`` / +``trust_publisher`` fields removed. Those fields are produced by the host, so +including them would make a signature impossible to verify after a round-trip. + +Verification returns a :class:`PublisherTrust` fact and never leaks key +material. Two entry points cover the two call sites: + +* :func:`evaluate_signature` — the historical manifest behaviour, which reports + ``Unverified``/``Invalid`` instead of raising; +* :func:`enforce_publisher_trust` — the strict path used for installations, + where a *forged* signature always fails and a *missing* signature fails when + ``PLUGIN_TRUST_POLICY=required`` (``if-present`` only tolerates absence). +""" + +from __future__ import annotations + +import base64 +import binascii +import json +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + +from app.services.plugin_trust import ( + VERIFY_IF_PRESENT, + VERIFY_REQUIRED, + PluginTrustConfigError, + decode_ed25519_public_key, + validate_trust_policy, +) + +__all__ = [ + "INVALID", + "TRUSTED", + "UNVERIFIED", + "VERIFY_IF_PRESENT", + "VERIFY_REQUIRED", + "InvalidPublisherSignatureError", + "MissingPublisherSignatureError", + "PublisherSignatureError", + "PublisherTrust", + "canonical_manifest_body", + "enforce_publisher_trust", + "evaluate_signature", + "validate_signature_metadata", +] + +TRUSTED = "Trusted" +UNVERIFIED = "Unverified" +INVALID = "Invalid" + +#: Fields the host owns and that are never part of the signed bytes. +_UNSIGNED_FIELDS = frozenset({"signature", "trust_state", "trust_publisher"}) + + +class PublisherSignatureError(ValueError): + """A publisher signature was required but could not be verified.""" + + +class MissingPublisherSignatureError(PublisherSignatureError): + """No signature was supplied while the trust policy requires one.""" + + +class InvalidPublisherSignatureError(PublisherSignatureError): + """A supplied signature is malformed, forged, or from an untrusted publisher.""" + + +@dataclass(frozen=True) +class PublisherTrust: + """The outcome of checking one publisher signature.""" + + state: str + publisher: str | None + signature_present: bool + + @property + def trusted(self) -> bool: + """Whether the signature verified against a configured publisher key.""" + return self.state == TRUSTED + + def to_dict(self) -> dict[str, Any]: + """Return the trust fields as stored on a plugin record.""" + return { + "trust_state": self.state, + "trust_publisher": self.publisher, + } + + +def validate_signature_metadata(signature: Any) -> str | None: + """Validate the shape of optional signature metadata. + + Returns a human-readable error string, or ``None`` when the metadata is a + well-formed Ed25519 signature object. + """ + if signature is None: + return None + if not isinstance(signature, Mapping): + return "Invalid signature: must be an object" + if set(signature) != {"publisher", "algorithm", "value"}: + return "Invalid signature: expected publisher, algorithm, and value" + if not all(isinstance(signature.get(key), str) and signature[key] for key in signature): + return "Invalid signature: publisher, algorithm, and value must be non-empty strings" + if signature["algorithm"] != "ed25519": + return "Invalid signature: algorithm must be ed25519" + return None + + +def canonical_manifest_body(data: Mapping[str, Any]) -> bytes: + """Return the stable JSON bytes a publisher signs for ``data``. + + ``signature`` and the runtime-managed trust fields are excluded so the same + payload verifies whether the signature is embedded or detached. + """ + if not isinstance(data, Mapping): + raise TypeError("Plugin manifest must be a mapping") + body = {key: value for key, value in data.items() if key not in _UNSIGNED_FIELDS} + return json.dumps( + body, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode() + + +def evaluate_signature( + manifest: Mapping[str, Any], + signature: Any, + trusted_publishers: Mapping[str, str | bytes] | None, +) -> PublisherTrust: + """Check ``signature`` and report trust without raising. + + Returns ``Trusted`` only when the signature verifies against the key + configured for its ``publisher``. A missing signature is ``Unverified``; + malformed metadata, an unknown publisher, or a forged signature is + ``Invalid``. The caller decides whether the resulting state is acceptable + (see :func:`enforce_publisher_trust`). + """ + if signature is None: + return PublisherTrust(UNVERIFIED, None, False) + + metadata_error = validate_signature_metadata(signature) + if metadata_error: + publisher = signature.get("publisher") if isinstance(signature, Mapping) else None + return PublisherTrust(INVALID, publisher if isinstance(publisher, str) else None, True) + + publisher = signature["publisher"] + key_value = (trusted_publishers or {}).get(publisher) + if key_value is None: + return PublisherTrust(INVALID, publisher, True) + + try: + public_key = Ed25519PublicKey.from_public_bytes(decode_ed25519_public_key(key_value)) + public_key.verify( + base64.b64decode(signature["value"], validate=True), + canonical_manifest_body(manifest), + ) + except (PluginTrustConfigError, ValueError, TypeError, binascii.Error, InvalidSignature): + return PublisherTrust(INVALID, publisher, True) + return PublisherTrust(TRUSTED, publisher, True) + + +def enforce_publisher_trust( + manifest: Mapping[str, Any], + signature: Any, + *, + trusted_publishers: Mapping[str, str | bytes] | None, + policy: str = VERIFY_IF_PRESENT, +) -> PublisherTrust: + """Enforce the trust policy for an installation and return the trust fact. + + * A valid signature from a configured publisher is always accepted. + * A **forged or untrusted** signature is always refused, even under + ``if-present``: tolerating absence must not mean tolerating forgeries. + * A **missing** signature is refused when ``policy`` is ``required`` and + accepted (as ``Unverified``) when it is ``if-present``. + + Args: + manifest: The manifest mapping as published by the catalog. + signature: Detached signature metadata, or ``None``. + trusted_publishers: Mapping of publisher id to configured key. + policy: ``if-present`` or ``required``. + + Raises: + PluginTrustConfigError: If ``policy`` is not a known policy. + MissingPublisherSignatureError: If the policy requires a signature and + none was supplied. + InvalidPublisherSignatureError: If the signature is malformed, forged, + or from a publisher without a configured key. + """ + policy = validate_trust_policy(policy) + trust = evaluate_signature(manifest, signature, trusted_publishers) + + if trust.trusted: + return trust + + if trust.signature_present: + publisher_label = f" from {trust.publisher!r}" if trust.publisher else "" + raise InvalidPublisherSignatureError( + f"Publisher signature verification failed: {trust.state} signature" + f"{publisher_label}. The manifest was modified or the publisher key is not trusted." + ) + + if policy == VERIFY_REQUIRED: + raise MissingPublisherSignatureError( + "A publisher signature is required by PLUGIN_TRUST_POLICY=required, " + "but the catalog entry is unsigned." + ) + + return trust diff --git a/app/services/plugin_trust.py b/app/services/plugin_trust.py new file mode 100644 index 0000000..21d8d84 --- /dev/null +++ b/app/services/plugin_trust.py @@ -0,0 +1,215 @@ +"""Trusted publisher key management (#386). + +Operators configure the publishers they trust through ``PLUGIN_TRUSTED_KEYS``, +a JSON object mapping a publisher id to an Ed25519 public key. Supporting +multiple key encodings makes the setting easy to get wrong, and a silently +ignored key is a security problem: an operator who pastes a malformed key would +believe signatures are verified while every artifact is simply marked invalid. + +This module therefore owns one strict loader: + +* publisher ids are canonical (lowercase, ``[a-z0-9._-]``, max 64 characters); +* every key must decode to a real 32-byte Ed25519 public key, supplied either + as base64 (raw key), as a PEM SubjectPublicKeyInfo, or as raw bytes; +* invalid configuration raises :class:`PluginTrustConfigError` naming the + publisher and the problem, and **never** echoing key material; +* the read-only surface (:func:`trusted_publisher_ids`) exposes publisher ids + only, so an operator can audit what is configured without any secret leaking. + +The loader is imported by the application factory so bad key material is +rejected at startup (see :func:`validate_trust_config`). +""" + +from __future__ import annotations + +import base64 +import binascii +import json +import re +from collections.abc import Mapping +from typing import Any + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + +#: Accepted plugin trust policies (mirrored by ``app.services.plugin_signatures``). +VERIFY_IF_PRESENT = "if-present" +VERIFY_REQUIRED = "required" +TRUST_POLICIES = (VERIFY_IF_PRESENT, VERIFY_REQUIRED) + +#: Canonical publisher id: lowercase, starts alphanumeric, ``[a-z0-9._-]``. +PUBLISHER_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$") + +#: Ed25519 raw public keys are exactly 32 bytes. +ED25519_KEY_BYTES = 32 + + +class PluginTrustConfigError(ValueError): + """Raised when ``PLUGIN_TRUSTED_KEYS`` / ``PLUGIN_TRUST_POLICY`` is invalid.""" + + +def validate_publisher_id(publisher: Any) -> str: + """Return ``publisher`` when it is a canonical id, else raise. + + Args: + publisher: Configured publisher identifier. + + Raises: + PluginTrustConfigError: If the id is not a lowercase ``[a-z0-9._-]`` + string of at most 64 characters. + """ + if not isinstance(publisher, str) or not PUBLISHER_ID_RE.match(publisher): + raise PluginTrustConfigError( + "publisher ids must be lowercase [a-z0-9._-], start with a letter or " + "digit, and be at most 64 characters" + ) + return publisher + + +def decode_ed25519_public_key(value: str | bytes) -> bytes: + """Decode a configured Ed25519 public key to its 32 raw bytes. + + Accepts base64-encoded raw keys, PEM ``SubjectPublicKeyInfo`` text, and raw + ``bytes``. Error messages describe the encoding problem only; they never + include the supplied value. + + Raises: + PluginTrustConfigError: If the value is not a valid Ed25519 public key. + """ + if isinstance(value, bytes): + raw = value + elif isinstance(value, str): + text = value.strip() + if not text: + raise PluginTrustConfigError("the key is empty") + if text.startswith("-----BEGIN"): + try: + loaded = serialization.load_pem_public_key(text.encode("utf-8")) + except (TypeError, ValueError) as exc: + raise PluginTrustConfigError("the PEM value is not a valid public key") from exc + if not isinstance(loaded, Ed25519PublicKey): + raise PluginTrustConfigError("the PEM value is not an Ed25519 public key") + raw = loaded.public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + else: + try: + raw = base64.b64decode(text, validate=True) + except (binascii.Error, ValueError) as exc: + raise PluginTrustConfigError( + "the key is not valid base64 or a PEM public key" + ) from exc + else: + raise PluginTrustConfigError("the key must be a base64/PEM string or raw bytes") + + if len(raw) != ED25519_KEY_BYTES: + raise PluginTrustConfigError(f"an Ed25519 public key must be {ED25519_KEY_BYTES} raw bytes") + try: + Ed25519PublicKey.from_public_bytes(raw) + except ValueError as exc: # pragma: no cover - length is checked above + raise PluginTrustConfigError("the value is not a valid Ed25519 public key") from exc + return raw + + +def load_trusted_publishers( + raw: Mapping[str, str | bytes] | str | None, +) -> dict[str, bytes]: + """Validate ``raw`` and return ``{publisher: raw_public_key_bytes}``. + + ``raw`` may be the parsed mapping held in ``PLUGIN_TRUSTED_KEYS`` or the + original JSON string (which is what an operator sets), so the loader is + usable from configuration, the CLI, and tests alike. + + Raises: + PluginTrustConfigError: For malformed JSON, a non-object value, an + invalid publisher id, or invalid key material. + """ + if raw is None or raw == "": + return {} + + if isinstance(raw, str): + try: + parsed = json.loads(raw) + except json.JSONDecodeError as exc: + raise PluginTrustConfigError( + "PLUGIN_TRUSTED_KEYS must be a JSON object of publisher keys" + ) from exc + raw = parsed + + if not isinstance(raw, Mapping): + raise PluginTrustConfigError("PLUGIN_TRUSTED_KEYS must be a JSON object of publisher keys") + + publishers: dict[str, bytes] = {} + for publisher, value in raw.items(): + try: + publisher_id = validate_publisher_id(publisher) + publishers[publisher_id] = decode_ed25519_public_key(value) + except PluginTrustConfigError as exc: + raise PluginTrustConfigError( + f"trusted publisher {publisher!r} is invalid: {exc}" + ) from exc + return publishers + + +def validate_trust_policy(policy: Any) -> str: + """Return ``policy`` when it is one of :data:`TRUST_POLICIES`. + + Raises: + PluginTrustConfigError: For any other value. + """ + if policy not in TRUST_POLICIES: + allowed = ", ".join(TRUST_POLICIES) + raise PluginTrustConfigError( + f"PLUGIN_TRUST_POLICY must be one of: {allowed} (got {policy!r})" + ) + return policy + + +def trusted_publisher_ids( + raw: Mapping[str, str | bytes] | str | None, +) -> list[str]: + """Return the configured publisher ids, sorted, without any key material. + + This is the only publisher information the API/CLI expose: the configured + keys themselves (and even their fingerprints) never leave the process. + """ + return sorted(load_trusted_publishers(raw)) + + +def validate_trust_config(config: Mapping[str, Any]) -> dict[str, bytes]: + """Validate the trust settings of an application config mapping. + + Called by the application factory so invalid key material fails at startup + with a clear message instead of silently weakening verification. + + Raises: + PluginTrustConfigError: If the keys or the policy are invalid. + """ + try: + publishers = load_trusted_publishers(config.get("PLUGIN_TRUSTED_KEYS")) + except PluginTrustConfigError as exc: + raise PluginTrustConfigError(f"invalid PLUGIN_TRUSTED_KEYS: {exc}") from exc + + try: + validate_trust_policy(config.get("PLUGIN_TRUST_POLICY", VERIFY_IF_PRESENT)) + except PluginTrustConfigError as exc: + raise PluginTrustConfigError(f"invalid PLUGIN_TRUST_POLICY: {exc}") from exc + + return publishers + + +__all__ = [ + "ED25519_KEY_BYTES", + "PUBLISHER_ID_RE", + "TRUST_POLICIES", + "VERIFY_IF_PRESENT", + "VERIFY_REQUIRED", + "PluginTrustConfigError", + "decode_ed25519_public_key", + "load_trusted_publishers", + "trusted_publisher_ids", + "validate_publisher_id", + "validate_trust_config", + "validate_trust_policy", +] diff --git a/app/services/plugins.py b/app/services/plugins.py index e59914b..6be176d 100644 --- a/app/services/plugins.py +++ b/app/services/plugins.py @@ -7,8 +7,6 @@ - Custom exceptions for plugin system """ -import base64 -import binascii import importlib import json import logging @@ -21,21 +19,21 @@ from types import ModuleType from typing import Any -from cryptography.exceptions import InvalidSignature -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey - from app.services.plugin_compat import is_valid_compatibility from app.services.plugin_deps import PluginRef from app.services.plugin_execution import run_with_timeout +from app.services.plugin_signatures import ( + TRUSTED, + UNVERIFIED, + VERIFY_IF_PRESENT, + VERIFY_REQUIRED, + PublisherTrust, + evaluate_signature, + validate_signature_metadata, +) logger = logging.getLogger(__name__) -TRUSTED = "Trusted" -UNVERIFIED = "Unverified" -INVALID = "Invalid" -VERIFY_IF_PRESENT = "if-present" -VERIFY_REQUIRED = "required" - #: The declared plugin manifest contract. Runtime validation below is driven by #: this schema (see :func:`load_manifest_schema`) so the two cannot silently #: drift apart. @@ -275,7 +273,7 @@ def from_dict( raise ManifestValidationError("; ".join(errors)) signature = data.get("signature") - signature_error = _validate_signature_metadata(signature) + signature_error = validate_signature_metadata(signature) if trust_policy not in (VERIFY_IF_PRESENT, VERIFY_REQUIRED): raise ManifestValidationError(f"Invalid plugin trust policy: {trust_policy}") @@ -364,62 +362,14 @@ def to_dict(self) -> dict[str, Any]: } -def _validate_signature_metadata(signature: Any) -> str | None: - """Validate the shape of optional signature metadata.""" - if signature is None: - return None - if not isinstance(signature, dict): - return "Invalid signature: must be an object" - if set(signature) != {"publisher", "algorithm", "value"}: - return "Invalid signature: expected publisher, algorithm, and value" - if not all(isinstance(signature.get(key), str) and signature[key] for key in signature): - return "Invalid signature: publisher, algorithm, and value must be non-empty strings" - if signature["algorithm"] != "ed25519": - return "Invalid signature: algorithm must be ed25519" - return None - - -def _canonical_manifest_body(data: dict[str, Any]) -> bytes: - """Return the stable JSON representation signed by plugin publishers.""" - body = {key: value for key, value in data.items() if key != "signature"} - return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() - - -def _decode_public_key(value: str | bytes) -> bytes: - if isinstance(value, bytes): - return value - if value.startswith("-----BEGIN"): - from cryptography.hazmat.primitives import serialization - - return serialization.load_pem_public_key(value.encode()).public_bytes( - serialization.Encoding.Raw, - serialization.PublicFormat.Raw, - ) - return base64.b64decode(value, validate=True) - - def _verify_signature( data: dict[str, Any], signature: Any, trusted_publishers: dict[str, str | bytes], ) -> tuple[str, str | None]: - if signature is None: - return UNVERIFIED, None - if _validate_signature_metadata(signature): - return INVALID, signature.get("publisher") if isinstance(signature, dict) else None - publisher = signature["publisher"] - key_value = trusted_publishers.get(publisher) - if key_value is None: - return INVALID, publisher - try: - public_key = Ed25519PublicKey.from_public_bytes(_decode_public_key(key_value)) - public_key.verify( - base64.b64decode(signature["value"], validate=True), - _canonical_manifest_body(data), - ) - except (ValueError, TypeError, binascii.Error, InvalidSignature): - return INVALID, publisher - return TRUSTED, publisher + """Return the ``(trust_state, publisher)`` facts for an embedded signature.""" + trust: PublisherTrust = evaluate_signature(data, signature, trusted_publishers) + return trust.state, trust.publisher class Plugin: diff --git a/app/services/plugins_cli.py b/app/services/plugins_cli.py index abe716f..6e63981 100644 --- a/app/services/plugins_cli.py +++ b/app/services/plugins_cli.py @@ -7,6 +7,8 @@ flask plugins enable flask plugins disable flask plugins install [--json] + flask plugins publishers [--json] + flask plugins errors [--json] The CLI acts as the **operator**, not as an end user: it reads/writes the persisted ``Plugin`` rows and never performs per-user workspace authorization. @@ -24,7 +26,7 @@ import json import click -from flask import Flask +from flask import Flask, current_app #: Exit codes (success / not found / validation-or-other error). EXIT_OK = 0 @@ -171,6 +173,32 @@ def install_plugin(path: str, as_json: bool) -> int: click.echo(f"Installed plugin: {plugin.id} v{plugin.version}") return EXIT_OK + @plugins_group.command("publishers") + @click.option("--json", "as_json", is_flag=True, help="Emit machine-readable JSON.") + def list_publishers(as_json: bool) -> int: + """List configured trusted publisher ids (never key material).""" + from app.services.plugin_signatures import VERIFY_IF_PRESENT + from app.services.plugin_trust import trusted_publisher_ids + + publishers = trusted_publisher_ids(current_app.config.get("PLUGIN_TRUSTED_KEYS") or {}) + policy = current_app.config.get("PLUGIN_TRUST_POLICY", VERIFY_IF_PRESENT) + if as_json: + _write_json( + { + "publishers": publishers, + "count": len(publishers), + "policy": policy, + } + ) + return EXIT_OK + if not publishers: + click.echo("No trusted publishers configured.") + return EXIT_OK + for publisher in publishers: + click.echo(publisher) + click.echo(f"{len(publishers)} trusted publisher(s) configured (policy: {policy}).") + return EXIT_OK + @plugins_group.command("errors") @click.option("--json", "as_json", is_flag=True, help="Emit machine-readable JSON.") def plugin_errors(as_json: bool) -> int: diff --git a/docs/plugins.md b/docs/plugins.md index 6eaa04e..3641ae4 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -4,11 +4,14 @@ The AI Code Assistant ships an extensible plugin architecture so that Stellar, Soroban, GitHub, AI-provider, and developer-tooling features can be added without modifying the core application. -> **Status:** Phase 8. Manifest validation, the registry, the capability model, -> the event system with fail-closed dispatch-time capability enforcement, and a -> workspace-scoped plugin management API + UI are implemented and tested. -> Dependency resolution, compatibility/versioning, signing/trust, and a -> marketplace are **not** implemented yet — see [Contributing](#contributing). +> **Status:** Phase 8 + Phase 9 security foundations. Manifest validation, the +> registry, the capability model, the event system with fail-closed +> dispatch-time capability enforcement, a workspace-scoped plugin management API +> + UI, publisher trust (trusted keys, embedded and detached Ed25519 +> signatures), catalog-pinned remote installation, and the plugin lock-entry +> model are implemented and tested. A browsable marketplace, artifact +> download/verification, and remote update/rollback are **not** implemented yet +> — see [Contributing](#contributing). ## Concepts @@ -220,6 +223,7 @@ flask plugins inspect # full manifest metadata + state flask plugins enable # enable (operator scope) flask plugins disable # disable (operator scope) flask plugins install # register from a local manifest dir +flask plugins publishers # list trusted publisher ids (never keys) ``` Every command supports `--json` for stable, parseable output. Exit codes @@ -246,7 +250,9 @@ management operations. | `GET` | `/plugins/api/workspaces` | Workspaces the user can access | any member | | `GET` | `/plugins/api/workspaces//plugins` | List plugins + installation state | any member | | `GET` | `/plugins/api/workspaces//plugins/` | Inspect one plugin | any member | +| `GET` | `/plugins/api/trusted-publishers` | List configured trusted publisher ids (never keys) and the active policy | authenticated | | `POST` | `/plugins/api/workspaces//plugins/install` | Install a trusted/local manifest | owner | +| `POST` | `/plugins/api/workspaces//plugins/install-from-catalog` | Install a catalog-pinned plugin (trust policy enforced) | owner | | `POST` | `/plugins/api/workspaces//plugins//update` | Update a registered plugin from a manifest (compatibility-gated) | owner | | `POST` | `/plugins/api/workspaces//plugins//enable` | Enable the workspace installation | owner | | `POST` | `/plugins/api/workspaces//plugins//disable` | Disable the workspace installation | owner | @@ -270,6 +276,134 @@ Capabilities granted through the API must be declared in the plugin's manifest (declared vs granted are kept distinct). The `manage_plugins` role gate is the only way to grant/revoke through the UI; grants are never implicit. +### Publisher trust: keys and detached signatures (#385, #386) + +Operators list trusted publishers in `PLUGIN_TRUSTED_KEYS`, a JSON object that +maps a publisher id (lowercase `[a-z0-9._-]`, at most 64 characters) to an +Ed25519 public key. Keys may be base64-encoded raw 32-byte keys or PEM +`SubjectPublicKeyInfo` text: + +```bash +export PLUGIN_TRUSTED_KEYS='{"acme-labs": ""}' +export PLUGIN_TRUST_POLICY=required # or if-present (default) +``` + +Key material is validated when the application starts +(`app.services.plugin_trust.validate_trust_config`): a malformed key, a +non-canonical publisher id, or an unknown policy aborts startup with a clear +message instead of silently treating every signature as invalid. Keys are +never returned by the API or the CLI — inspect the configured ids with: + +```bash +flask plugins publishers # human-readable ids + policy +flask plugins publishers --json # {"publishers": [...], "count": N, "policy": ...} +``` + +`GET /plugins/api/trusted-publishers` exposes the same read-only view (any +authenticated user): publisher ids, a count, and the active policy, never key +bytes or fingerprints. + +A publisher signs the **canonical manifest body**: compact, key-sorted JSON with +`signature`, `trust_state`, and `trust_publisher` removed. The signature is a +base64 Ed25519 signature over those bytes and may be: + +- **embedded** in the manifest (`"signature": {"publisher", "algorithm": + "ed25519", "value"}`), or +- **detached**, delivered next to the manifest (a `.sig` file or the catalog + entry's `signature` field). + +`app/services/plugin_signatures.py` is the single verification path for both +forms. Policy handling is deliberately strict: + +| Situation | `if-present` | `required` | +| --------- | ------------ | ---------- | +| Valid signature from a configured publisher | accepted (`Trusted`) | accepted (`Trusted`) | +| No signature | accepted (`Unverified`) | refused | +| Forged/malformed signature, or unknown publisher | **refused** | **refused** | + +Tolerating absence must never mean tolerating forgeries, so an invalid +signature fails even in `if-present` mode. Local-path installs keep working +under the default policy; set `required` to demand signatures everywhere. + +### Remote install from the validated catalog (#389) + +`POST /plugins/api/workspaces//plugins/install-from-catalog` installs a +plugin that an operator has already published in the catalog. The request only +names the entry — a client can never supply a manifest, an artifact URL, or a +key: + +```bash +curl -X POST https:///plugins/api/workspaces/7/plugins/install-from-catalog \ + -H "Content-Type: application/json" \ + -d '{"plugin_id": "hello-world", "version": "0.1.0"}' +``` + +`version` is optional; without it the newest catalogued version wins (PEP 440 +ordering). `PLUGIN_CATALOG_FILE` points at the catalog JSON (abridged below; +`manifest` is a complete, schema-valid manifest): + +```json +{ + "version": 1, + "plugins": [ + { + "id": "hello-world", + "version": "0.1.0", + "source": "https://plugins.example.com/hello-world-0.1.0.zip", + "integrity": "sha256:<64 lowercase hex characters>", + "signature": { + "publisher": "acme-labs", + "algorithm": "ed25519", + "value": "" + }, + "manifest": { "id": "hello-world", "...": "see plugins/plugin.schema.json" } + } + ] +} +``` + +The endpoint is owner-only (`manage_plugins`) and fails closed: + +| Failure | Status | +| ------- | ------ | +| Missing/invalid `plugin_id` or `version` | `400` | +| No catalog configured, or unknown id/version | `404` | +| Invalid manifest or incompatible `compatibility` range | `400` | +| Trust refusal (unsigned under `required`, forged, unknown publisher) | `403` | +| Already installed, or entry point conflicts with a registered plugin | `409` | +| Non-member of the workspace | `404` | +| Member who is not the owner | `403` | + +On success it registers metadata only: the global `Plugin` row and the +workspace `PluginInstallation`. The response also reports the catalog `source` +and `integrity` so the resolved pin is visible. It does **not** download the +artifact, execute plugin code, or create capability grants, and trust refusals +are appended to the workspace audit trail. + +### Plugin lock entries (#394) + +`PluginLockEntry` (`plugin_lock_entries`) persists the resolved plugin set for +reproducible installs: plugin id, pinned version, source, sha256 integrity +hash, declared dependencies, the workspace, and its owner. A row is unique per +`(workspace_id, plugin_id)`, and lock entries never imply capability grants. + +Rows serialize through the existing canonical lockfile builder, so +`dumps_lockfile([*rows])` produces the same byte-stable document as any other +resolved entry and omits volatile audit fields (`resolved_at`, row/workspace +ids): + +```python +from app.models import PluginLockEntry +from app.services.plugin_lockfile import dumps_lockfile + +rows = PluginLockEntry.query.filter_by(workspace_id=workspace_id).all() +print(dumps_lockfile(rows)) +``` + +The Alembic migration for the table is tracked by the Phase 9 lockfile +migration issue; the model and its serialization are covered by +`tests/test_plugin_lock_entries.py`. + ### Updating a plugin `POST .../plugins//update` accepts the same `{"manifest": {...}}` @@ -429,6 +563,15 @@ See [security.md](security.md) for the full security model. In short: - Plugin handler failures are isolated from the request lifecycle. - Manifests are validated (no arbitrary code identifiers, no unknown capabilities). +- Publisher signatures are Ed25519 over the canonical manifest body; forged or + untrusted signatures are always refused, and `required` also refuses + unsigned entries. +- Trusted publisher keys are validated at startup and never exposed by the API + or CLI (only publisher ids are listed). +- Catalog installs are owner-only, resolve server-side, register metadata only, + and never download code or grant capabilities. +- Event data carries workspace/user context so plugins can scope work; the + Stellar analysis path reuses the Phase 7 content-access gate and fails closed. - Event data carries workspace/user context so plugins can scope work; the Stellar analysis path reuses the Phase 7 content-access gate and fails closed. @@ -456,6 +599,15 @@ is reset between tests): - `tests/test_plugins_api.py` — the workspace-scoped plugin management API (install/enable/disable/grant) with authorization fail-closed and workspace isolation. +- `tests/test_plugin_trust.py` — trusted publisher key loading (base64/PEM), + startup rejection of bad configuration, and the secret-free listing surfaces. +- `tests/test_plugin_signatures.py` — canonical payloads, detached signatures, + forged/unsigned handling, and the embedded-signature regression path. +- `tests/test_plugin_catalog_install.py` — catalog parsing/validation and the + remote-install endpoint: authorization, trust policy, 4xx failure paths, and + no implicit grants or code execution. +- `tests/test_plugin_lock_entries.py` — the `PluginLockEntry` model, scoping, + unique constraint, and canonical lockfile serialization. - `tests/test_plugin_audit.py` — the capability/state audit trail. - `tests/test_event_authorization.py` — dispatch-time capability enforcement. - `tests/test_event_wiring.py` — real routes emit events to authorized diff --git a/docs/security.md b/docs/security.md index 6555e05..e793160 100644 --- a/docs/security.md +++ b/docs/security.md @@ -55,6 +55,10 @@ kept in sync as the plugin system grows. | Lifecycle hooks + uninstall | Optional hooks run isolated/non-fatal on registry enable/disable/uninstall; the workspace uninstall endpoint revokes every capability grant for that workspace before removing the installation. | Implemented | | Plugin events | `plugin.enabled`/`plugin.disabled`/`plugin.uninstalled` are workspace-scoped events mapped to `WORKSPACE_READ` (dispatch-time enforcement applies). | Implemented | | Plugin CLI | `flask plugins …` acts as the operator on persisted rows; installs are local-path only (URLs refused) and never grant capabilities or load code. | Implemented | +| Publisher signatures (detached) | Ed25519 signatures are verified over the canonical manifest body (compact sorted JSON excluding `signature`/trust fields), whether embedded in the manifest or detached in a `.sig`/catalog field. Forged, malformed, and unknown-publisher signatures are refused under **every** policy; unsigned entries are refused when `PLUGIN_TRUST_POLICY=required`. Verification never echoes key material. | Implemented | +| Trusted publisher key management | `PLUGIN_TRUSTED_KEYS` must be a JSON object of canonical publisher ids to valid 32-byte Ed25519 keys (base64 raw or PEM); malformed keys/policies abort startup with a clear message. The API (`GET /plugins/api/trusted-publishers`) and CLI (`flask plugins publishers`) expose publisher ids and the policy only — never key bytes or fingerprints. | Implemented | +| Catalog-pinned remote install | `POST .../plugins/install-from-catalog` is owner-only and resolves `(plugin_id, version)` server-side against `PLUGIN_CATALOG_FILE`; clients cannot supply a manifest, artifact URL, or key. Entries are validated (canonical id/version, https source, sha256 integrity, manifest id/version match) and the trust policy is enforced before any write. It registers metadata only: no download, no code execution, no implicit capability grants; refusals are audited with no artifact contents. | Implemented | +| Plugin lock entries | `PluginLockEntry` rows are scoped to `(workspace_id, owner_id)` with a unique `(workspace_id, plugin_id)` constraint, store only resolved metadata (id, version, source, sha256 integrity, dependencies), and imply no capability grant. Canonical serialization omits volatile timestamps for reproducible diffs. | Implemented | | Stellar security findings | `stellar_security` findings are evidence-labelled rows owned by a single project; reads are owner-scoped (404 for non-owners, no existence oracle). | Implemented | | XDR decoding | Transaction/envelope + contract-data decoding is bounded and read-only; decoded values are never guessed, malformed/unsupported XDR is explicit, and no signing/submission/keys exist. | Implemented | @@ -139,6 +143,16 @@ configuration (see "Known limitations / planned hardening"). - `tests/test_plugin_lifecycle.py` — lifecycle hooks and plugin events. - `tests/test_plugins_cli.py` — CLI commands, `--json`, exit codes, URL refusal, and no implicit grants. +- `tests/test_plugin_trust.py` — trusted publisher key decoding/validation, + startup rejection of malformed keys or policies, and secret-free listing. +- `tests/test_plugin_signatures.py` — canonical signing payload, detached + signature verification, forged/unsigned refusal, and embedded-signature + compatibility. +- `tests/test_plugin_catalog_install.py` — catalog validation plus the + owner-only remote-install endpoint (trust policy, failure codes, audit of + refusals, no code execution, no implicit grants). +- `tests/test_plugin_lock_entries.py` — lock-entry scoping, uniqueness, and + canonical lockfile serialization. ## Known limitations / planned hardening @@ -149,6 +163,10 @@ configuration (see "Known limitations / planned hardening"). points, subscribing their handlers) is intentionally not wired to any install flow yet — plugins are subscribed in-process by code, and a future runtime must keep capability enforcement and review in front of execution. +- Catalog installs register **metadata only**: the catalog's `integrity` hash + is validated as a sha256 digest and returned/recorded, but the artifact is + not downloaded yet. A future download step (bounded transport + hash + verification) must re-verify the digest before any artifact is used. - Plugins have **no outbound network capability wired up yet**. The guard in `app/services/plugin_network.py` (and its allowlist policy) is the required path for any future plugin-triggered request; plugin code must call diff --git a/tests/test_plugin_catalog_install.py b/tests/test_plugin_catalog_install.py new file mode 100644 index 0000000..be6d753 --- /dev/null +++ b/tests/test_plugin_catalog_install.py @@ -0,0 +1,560 @@ +"""Tests for the catalog-backed remote install endpoint (#389). + +The endpoint is owner-only, installs only catalog-pinned metadata, enforces the +publisher trust policy before writing anything, and never creates capability +grants or executes plugin code. +""" + +import base64 +import json + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from app.extensions import db +from app.models import ( + ActivityEvent, + CapabilityGrant, + Plugin, + PluginInstallation, + Workspace, + WorkspaceMember, +) +from app.models.activity_event import EVENT_PLUGIN_DENIED +from app.services.plugin_catalog import ( + CatalogEntryNotFoundError, + CatalogError, + CatalogUnavailableError, + load_catalog_file, + load_plugin_catalog, + parse_catalog, +) +from app.services.plugin_signatures import canonical_manifest_body + +PUBLISHER = "acme-labs" +HASH = "a" * 64 + + +def _manifest(plugin_id="catalog-plugin", version="0.1.0", **overrides): + manifest = { + "id": plugin_id, + "name": "Catalog Plugin", + "version": version, + "description": "A catalogued test plugin", + "author": "Tester", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + manifest.update(overrides) + return manifest + + +def _entry(manifest=None, *, signature=None, **overrides): + manifest = manifest or _manifest() + entry = { + "id": manifest["id"], + "version": manifest["version"], + "source": "https://plugins.example.com/catalog-plugin-0.1.0.zip", + "integrity": f"sha256:{HASH}", + "manifest": manifest, + } + entry.update(overrides) + if signature is not None: + entry["signature"] = signature + return entry + + +def _publisher(): + private_key = Ed25519PrivateKey.generate() + raw = private_key.public_key().public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + return private_key, base64.b64encode(raw).decode() + + +def _signature(private_key, manifest, publisher=PUBLISHER): + return { + "publisher": publisher, + "algorithm": "ed25519", + "value": base64.b64encode(private_key.sign(canonical_manifest_body(manifest))).decode(), + } + + +def _signed_entry(private_key, manifest=None, publisher=PUBLISHER, **overrides): + manifest = manifest or _manifest() + return _entry( + manifest=manifest, + signature=_signature(private_key, manifest, publisher), + **overrides, + ) + + +def _write_catalog(tmp_path, entries, *, version=1): + path = tmp_path / "catalog.json" + path.write_text( + json.dumps({"version": version, "plugins": entries}), + encoding="utf-8", + ) + return path + + +def _workspace(owner, name="WS"): + ws = Workspace(user_id=owner.id, name=name) + db.session.add(ws) + db.session.commit() + return ws + + +def _member(ws, user, role="contributor"): + membership = WorkspaceMember(workspace_id=ws.id, user_id=user.id, role=role) + db.session.add(membership) + db.session.commit() + return membership + + +def _install(client, ws_id, payload): + return client.post( + f"/plugins/api/workspaces/{ws_id}/plugins/install-from-catalog", + json=payload, + ) + + +class TestCatalogParsing: + def test_parses_and_normalizes_entries(self): + catalog = parse_catalog( + { + "version": 1, + "plugins": [ + _entry(id="zeta", manifest=_manifest("zeta")), + _entry( + id="alpha", + manifest=_manifest("alpha"), + integrity=HASH.upper(), + ), + ], + } + ) + + assert len(catalog) == 2 + assert [entry.plugin_id for entry in catalog.list_entries()] == ["alpha", "zeta"] + alpha = catalog.get("alpha") + assert alpha.integrity == f"sha256:{HASH}" + assert alpha.publisher is None + + def test_selects_the_newest_version_by_default(self): + catalog = parse_catalog( + { + "plugins": [ + _entry(manifest=_manifest(version="0.1.0")), + _entry(manifest=_manifest(version="0.10.0")), + _entry(manifest=_manifest(version="0.9.0")), + ] + } + ) + + assert catalog.get("catalog-plugin").version == "0.10.0" + assert catalog.get("catalog-plugin", "0.9.0").version == "0.9.0" + assert catalog.get("catalog-plugin", "0.10").version == "0.10.0" + + def test_unknown_id_and_version_raise(self): + catalog = parse_catalog({"plugins": [_entry()]}) + + with pytest.raises(CatalogEntryNotFoundError, match="ghost"): + catalog.get("ghost") + with pytest.raises(CatalogEntryNotFoundError, match=r"9\.9\.9"): + catalog.get("catalog-plugin", "9.9.9") + + @pytest.mark.parametrize( + "document", + [ + [], + {"version": 2, "plugins": []}, + {"plugins": {}}, + {"version": 1}, + ], + ) + def test_rejects_malformed_documents(self, document): + with pytest.raises(CatalogError): + parse_catalog(document) + + def test_rejects_duplicate_entries(self): + with pytest.raises(CatalogError, match="duplicate"): + parse_catalog({"plugins": [_entry(), _entry()]}) + + @pytest.mark.parametrize( + "entry", + [ + _entry(id="BadId"), + _entry(version="not-a-version"), + _entry(source="http://plugins.example.com/p.zip"), + _entry(integrity="sha512:" + "a" * 128), + _entry(signature={"publisher": PUBLISHER, "algorithm": "rsa", "value": "x"}), + _entry(id="different-id"), + _entry(version="2.0.0", manifest=_manifest(version="0.1.0")), + { + "id": "catalog-plugin", + "version": "0.1.0", + "source": "https://plugins.example.com/p.zip", + "integrity": f"sha256:{HASH}", + "manifest": "not-an-object", + }, + {"id": "catalog-plugin"}, + ], + ) + def test_rejects_invalid_entries_with_the_index(self, entry): + with pytest.raises(CatalogError, match="catalog entry 0"): + parse_catalog({"plugins": [entry]}) + + def test_load_catalog_file_errors(self, tmp_path): + with pytest.raises(CatalogUnavailableError): + load_catalog_file(tmp_path / "missing.json") + + broken = tmp_path / "broken.json" + broken.write_text("{not json", encoding="utf-8") + with pytest.raises(CatalogError, match="JSON"): + load_catalog_file(broken) + + oversized = tmp_path / "big.json" + oversized.write_bytes(b"{" + b" " * (1024 * 1024)) + with pytest.raises(CatalogUnavailableError): + load_catalog_file(oversized) + + def test_unconfigured_catalog_is_unavailable(self): + with pytest.raises(CatalogUnavailableError, match="PLUGIN_CATALOG_FILE"): + load_plugin_catalog({"PLUGIN_CATALOG_FILE": ""}) + + +class TestCatalogInstallAuthorization: + def test_owner_can_install(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 201 + + def test_member_is_forbidden(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user(username="owner", email="owner@example.com") + member = make_user(username="member", email="member@example.com") + ws = _workspace(owner) + _member(ws, member) + login(email="member@example.com") + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 403 + assert db.session.get(Plugin, "catalog-plugin") is None + + def test_outsider_gets_404(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user(username="owner", email="owner@example.com") + make_user(username="outsider", email="outsider@example.com") + ws = _workspace(owner) + login(email="outsider@example.com") + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 404 + + def test_anonymous_is_not_allowed(self, app, client, make_user, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code in (302, 401) + + +class TestCatalogInstallSuccess: + def test_installs_metadata_only_without_grants_or_code_execution( + self, app, client, make_user, login, tmp_path, monkeypatch + ): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + + def _fail_load(*args, **kwargs): # pragma: no cover - only runs on regression + raise AssertionError("catalog install must never load or execute plugin code") + + monkeypatch.setattr("app.services.plugins.Plugin.load", _fail_load) + monkeypatch.setattr("app.services.plugins.Plugin.get_instance", _fail_load) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 201 + data = resp.get_json() + assert data["id"] == "catalog-plugin" + assert data["installed"] is True + assert data["trust_state"] == "Unverified" + assert data["source"].startswith("https://") + assert data["integrity"] == f"sha256:{HASH}" + + plugin = db.session.get(Plugin, "catalog-plugin") + assert plugin is not None + assert plugin.entry_point == "plugins.test:TestPlugin" + installation = PluginInstallation.query.filter_by( + plugin_id="catalog-plugin", workspace_id=ws.id + ).one() + assert installation.enabled is True + assert installation.granted_capabilities == [] + assert CapabilityGrant.query.filter_by(plugin_id="catalog-plugin").count() == 0 + + def test_pins_the_requested_version(self, app, client, make_user, login, tmp_path): + catalog_path = _write_catalog( + tmp_path, + [ + _entry(manifest=_manifest(version="0.1.0")), + _entry(manifest=_manifest(version="0.2.0")), + ], + ) + app.config["PLUGIN_CATALOG_FILE"] = str(catalog_path) + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin", "version": "0.1.0"}) + + assert resp.status_code == 201 + assert db.session.get(Plugin, "catalog-plugin").version == "0.1.0" + + +class TestCatalogInstallTrustPolicy: + def test_required_policy_refuses_unsigned_entries( + self, app, client, make_user, login, tmp_path + ): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + app.config["PLUGIN_TRUST_POLICY"] = "required" + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 403 + assert "required" in resp.get_json()["error"] + assert db.session.get(Plugin, "catalog-plugin") is None + denied = ActivityEvent.query.filter_by( + workspace_id=ws.id, event_type=EVENT_PLUGIN_DENIED + ).one() + assert denied.event_metadata["plugin_id"] == "catalog-plugin" + assert denied.event_metadata["reason"] == "publisher signature refused" + assert HASH not in json.dumps(denied.event_metadata) + + def test_valid_signature_installs_under_required_policy( + self, app, client, make_user, login, tmp_path + ): + private_key, public_key = _publisher() + manifest = _manifest() + app.config["PLUGIN_CATALOG_FILE"] = str( + _write_catalog(tmp_path, [_signed_entry(private_key, manifest)]) + ) + app.config["PLUGIN_TRUSTED_KEYS"] = {PUBLISHER: public_key} + app.config["PLUGIN_TRUST_POLICY"] = "required" + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 201 + data = resp.get_json() + assert data["trust_state"] == "Trusted" + assert data["trust_publisher"] == PUBLISHER + assert db.session.get(Plugin, "catalog-plugin").trust_state == "Trusted" + + def test_forged_signature_is_refused_even_under_if_present( + self, app, client, make_user, login, tmp_path + ): + private_key, public_key = _publisher() + manifest = _manifest() + entry = _signed_entry(private_key, manifest) + entry["manifest"]["name"] = "Tampered Plugin" + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [entry])) + app.config["PLUGIN_TRUSTED_KEYS"] = {PUBLISHER: public_key} + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 403 + assert db.session.get(Plugin, "catalog-plugin") is None + assert ( + ActivityEvent.query.filter_by( + workspace_id=ws.id, event_type=EVENT_PLUGIN_DENIED + ).count() + == 1 + ) + + def test_signature_from_an_unconfigured_publisher_is_refused( + self, app, client, make_user, login, tmp_path + ): + private_key, _ = _publisher() + app.config["PLUGIN_CATALOG_FILE"] = str( + _write_catalog(tmp_path, [_signed_entry(private_key, _manifest())]) + ) + app.config["PLUGIN_TRUSTED_KEYS"] = {} + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 403 + assert PUBLISHER in resp.get_json()["error"] + + def test_embedded_manifest_signature_satisfies_required_policy( + self, app, client, make_user, login, tmp_path + ): + private_key, public_key = _publisher() + manifest = _manifest() + signed_manifest = {**manifest, "signature": _signature(private_key, manifest)} + app.config["PLUGIN_CATALOG_FILE"] = str( + _write_catalog(tmp_path, [_entry(manifest=signed_manifest)]) + ) + app.config["PLUGIN_TRUSTED_KEYS"] = {PUBLISHER: public_key} + app.config["PLUGIN_TRUST_POLICY"] = "required" + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 201 + assert resp.get_json()["trust_state"] == "Trusted" + + +class TestCatalogInstallFailures: + def test_unknown_plugin_is_404(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "ghost"}) + + assert resp.status_code == 404 + assert "ghost" in resp.get_json()["error"] + + def test_unknown_version_is_404(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin", "version": "9.9.9"}) + + assert resp.status_code == 404 + + def test_invalid_version_is_400(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin", "version": "not-a-version"}) + + assert resp.status_code == 400 + assert "Invalid plugin version" in resp.get_json()["error"] + + def test_missing_plugin_id_is_400(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + + assert _install(client, ws.id, {}).status_code == 400 + assert _install(client, ws.id, {"plugin_id": " "}).status_code == 400 + assert ( + _install(client, ws.id, {"plugin_id": "catalog-plugin", "version": 1}).status_code + == 400 + ) + + def test_unconfigured_catalog_is_404(self, app, client, make_user, login): + app.config["PLUGIN_CATALOG_FILE"] = "" + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 404 + assert "PLUGIN_CATALOG_FILE" in resp.get_json()["error"] + + def test_invalid_manifest_in_entry_is_400(self, app, client, make_user, login, tmp_path): + manifest = _manifest(capabilities=[]) + app.config["PLUGIN_CATALOG_FILE"] = str( + _write_catalog(tmp_path, [_entry(manifest=manifest)]) + ) + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 400 + assert "Invalid plugin manifest" in resp.get_json()["error"] + + def test_incompatible_range_is_400(self, app, client, make_user, login, tmp_path): + manifest = _manifest(compatibility=">=99.0.0") + app.config["PLUGIN_CATALOG_FILE"] = str( + _write_catalog(tmp_path, [_entry(manifest=manifest)]) + ) + owner = make_user() + login() + ws = _workspace(owner) + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 400 + assert "99.0.0" in resp.get_json()["error"] + + def test_duplicate_install_is_409(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + assert _install(client, ws.id, {"plugin_id": "catalog-plugin"}).status_code == 201 + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 409 + assert ( + PluginInstallation.query.filter_by( + plugin_id="catalog-plugin", workspace_id=ws.id + ).count() + == 1 + ) + + def test_entry_point_mismatch_is_409(self, app, client, make_user, login, tmp_path): + app.config["PLUGIN_CATALOG_FILE"] = str(_write_catalog(tmp_path, [_entry()])) + owner = make_user() + login() + ws = _workspace(owner) + db.session.add( + Plugin( + id="catalog-plugin", + name="Catalog Plugin", + version="0.1.0", + entry_point="other.module:OtherPlugin", + capabilities=["PROJECT_READ"], + ) + ) + db.session.commit() + + resp = _install(client, ws.id, {"plugin_id": "catalog-plugin"}) + + assert resp.status_code == 409 + assert ( + PluginInstallation.query.filter_by( + plugin_id="catalog-plugin", workspace_id=ws.id + ).count() + == 0 + ) diff --git a/tests/test_plugin_lock_entries.py b/tests/test_plugin_lock_entries.py new file mode 100644 index 0000000..6bff467 --- /dev/null +++ b/tests/test_plugin_lock_entries.py @@ -0,0 +1,145 @@ +"""Tests for the ``PluginLockEntry`` model (#394). + +The model is the persistence half of the deterministic lockfile: rows are +owner/workspace scoped, serialize to the canonical lockfile shape (no volatile +timestamps), and never imply capability grants. +""" + +import json + +import pytest +from sqlalchemy.exc import IntegrityError + +from app.extensions import db +from app.models import CapabilityGrant, PluginLockEntry, Workspace +from app.services.plugin_lockfile import LockfileEntry, dumps_lockfile, normalize_integrity + +HASH_A = "A" * 64 +HASH_B = "b" * 64 + + +def _workspace(owner, name="WS"): + ws = Workspace(user_id=owner.id, name=name) + db.session.add(ws) + db.session.commit() + return ws + + +def _lock( + workspace, + owner, + plugin_id="catalog-plugin", + *, + version="0.1.0", + source="https://plugins.example.com/p.zip", + integrity=HASH_A, + dependencies=None, +): + row = PluginLockEntry( + workspace_id=workspace.id, + owner_id=owner.id, + plugin_id=plugin_id, + version=version, + source=source, + integrity=integrity, + dependencies=dependencies if dependencies is not None else ["dep>=1.0"], + ) + db.session.add(row) + db.session.commit() + return row + + +class TestSerialization: + def test_to_dict_contains_scoping_and_audit_fields(self, db, make_user): + owner = make_user() + ws = _workspace(owner) + row = _lock(ws, owner) + + data = row.to_dict() + + assert data["id"] == row.id + assert data["workspace_id"] == ws.id + assert data["owner_id"] == owner.id + assert data["plugin_id"] == "catalog-plugin" + assert data["version"] == "0.1.0" + assert data["source"] == "https://plugins.example.com/p.zip" + assert data["integrity"] == HASH_A + assert data["dependencies"] == ["dep>=1.0"] + assert data["resolved_at"] == row.resolved_at.isoformat() + + def test_to_lockfile_entry_normalizes_integrity(self, db, make_user): + owner = make_user() + ws = _workspace(owner) + row = _lock(ws, owner) + + entry = row.to_lockfile_entry() + + assert isinstance(entry, LockfileEntry) + assert entry.integrity == normalize_integrity(HASH_A) + assert entry.dependencies == ("dep>=1.0",) + + def test_lockfile_is_byte_stable_and_hides_volatile_fields(self, db, make_user): + owner = make_user() + ws = _workspace(owner) + first = _lock(ws, owner, "zeta") + second = _lock(ws, owner, "alpha", integrity=HASH_B) + + output = dumps_lockfile([first, second]) + document = json.loads(output) + + assert [entry["plugin_id"] for entry in document["plugins"]] == ["alpha", "zeta"] + assert document["plugins"][1]["integrity"] == normalize_integrity(HASH_A) + assert "resolved_at" not in output + assert "workspace_id" not in output + assert "owner_id" not in output + assert output == dumps_lockfile([second, first]) + assert output.endswith("\n") + + +class TestScoping: + def test_unique_per_workspace(self, db, make_user): + owner = make_user() + ws = _workspace(owner) + _lock(ws, owner) + + db.session.add( + PluginLockEntry( + workspace_id=ws.id, + owner_id=owner.id, + plugin_id="catalog-plugin", + version="0.2.0", + source="https://plugins.example.com/p.zip", + integrity=HASH_B, + ) + ) + with pytest.raises(IntegrityError): + db.session.commit() + db.session.rollback() + + def test_same_plugin_allowed_in_another_workspace(self, db, make_user): + owner = make_user() + first = _workspace(owner, "One") + second = _workspace(owner, "Two") + _lock(first, owner) + _lock(second, owner) + + assert PluginLockEntry.query.filter_by(plugin_id="catalog-plugin").count() == 2 + + def test_rows_are_owner_scoped(self, db, make_user): + owner = make_user() + other = make_user(username="other", email="other@example.com") + ws = _workspace(owner) + row = _lock(ws, owner) + + assert PluginLockEntry.query.filter_by(owner_id=other.id).count() == 0 + assert PluginLockEntry.query.filter_by(owner_id=owner.id).one().id == row.id + + +class TestNoImplicitCapabilities: + def test_lock_entries_never_grant_capabilities(self, db, make_user): + owner = make_user() + ws = _workspace(owner) + _lock(ws, owner, dependencies=[]) + + assert CapabilityGrant.query.filter_by(plugin_id="catalog-plugin").count() == 0 + assert CapabilityGrant.query.count() == 0 diff --git a/tests/test_plugin_signatures.py b/tests/test_plugin_signatures.py new file mode 100644 index 0000000..1e2a4ac --- /dev/null +++ b/tests/test_plugin_signatures.py @@ -0,0 +1,255 @@ +"""Tests for detached publisher signatures (#385). + +A valid signature verifies; a forged signature always fails (even in lenient +mode); a missing signature is tolerated only when the trust policy allows it. +""" + +import base64 + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from app.services.plugin_signatures import ( + INVALID, + TRUSTED, + UNVERIFIED, + InvalidPublisherSignatureError, + MissingPublisherSignatureError, + canonical_manifest_body, + enforce_publisher_trust, + evaluate_signature, + validate_signature_metadata, +) +from app.services.plugin_trust import PluginTrustConfigError +from app.services.plugins import ManifestValidationError, PluginManifest + +PUBLISHER = "acme-labs" + + +def _manifest(): + return { + "id": "signed-plugin", + "name": "Signed Plugin", + "version": "0.1.0", + "description": "A signed test plugin", + "author": "Tester", + "entry_point": "plugins.test:TestPlugin", + "capabilities": ["PROJECT_READ"], + } + + +def _publisher(): + private_key = Ed25519PrivateKey.generate() + raw = private_key.public_key().public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + return private_key, base64.b64encode(raw).decode() + + +def _sign(private_key, manifest, publisher=PUBLISHER): + return { + "publisher": publisher, + "algorithm": "ed25519", + "value": base64.b64encode(private_key.sign(canonical_manifest_body(manifest))).decode(), + } + + +class TestCanonicalBody: + def test_is_stable_regardless_of_key_order(self): + manifest = _manifest() + reordered = dict(reversed(list(manifest.items()))) + assert canonical_manifest_body(manifest) == canonical_manifest_body(reordered) + + def test_excludes_signature_and_runtime_trust_fields(self): + manifest = _manifest() + decorated = { + **manifest, + "signature": {"publisher": PUBLISHER, "algorithm": "ed25519", "value": "x"}, + "trust_state": TRUSTED, + "trust_publisher": PUBLISHER, + } + assert canonical_manifest_body(decorated) == canonical_manifest_body(manifest) + + def test_rejects_non_mapping(self): + with pytest.raises(TypeError): + canonical_manifest_body(["not", "a", "manifest"]) + + +class TestMetadataValidation: + def test_accepts_well_formed_signature(self): + private_key, _ = _publisher() + assert validate_signature_metadata(_sign(private_key, _manifest())) is None + + @pytest.mark.parametrize( + "signature", + [ + ["not", "an", "object"], + {"publisher": PUBLISHER}, + {"publisher": PUBLISHER, "algorithm": "rsa", "value": "x"}, + {"publisher": PUBLISHER, "algorithm": "ed25519", "value": ""}, + {"publisher": "", "algorithm": "ed25519", "value": "x"}, + { + "publisher": PUBLISHER, + "algorithm": "ed25519", + "value": "x", + "extra": "nope", + }, + ], + ) + def test_rejects_malformed_metadata(self, signature): + assert validate_signature_metadata(signature) is not None + + +class TestEvaluateSignature: + def test_valid_detached_signature_is_trusted(self): + private_key, public_key = _publisher() + manifest = _manifest() + + trust = evaluate_signature(manifest, _sign(private_key, manifest), {PUBLISHER: public_key}) + + assert trust.state == TRUSTED + assert trust.publisher == PUBLISHER + assert trust.signature_present is True + assert trust.to_dict() == {"trust_state": TRUSTED, "trust_publisher": PUBLISHER} + + def test_forged_signature_is_invalid(self): + private_key, public_key = _publisher() + manifest = _manifest() + signature = _sign(private_key, manifest) + manifest["name"] = "Tampered Plugin" + + trust = evaluate_signature(manifest, signature, {PUBLISHER: public_key}) + + assert trust.state == INVALID + assert trust.publisher == PUBLISHER + + def test_unknown_publisher_is_invalid(self): + private_key, _ = _publisher() + manifest = _manifest() + + trust = evaluate_signature(manifest, _sign(private_key, manifest), {}) + + assert trust.state == INVALID + assert trust.publisher == PUBLISHER + + def test_missing_signature_is_unverified(self): + trust = evaluate_signature(_manifest(), None, {}) + assert trust.state == UNVERIFIED + assert trust.publisher is None + assert trust.signature_present is False + + def test_pem_key_is_accepted(self): + private_key, _ = _publisher() + pem = ( + private_key.public_key() + .public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + .decode() + ) + manifest = _manifest() + + trust = evaluate_signature(manifest, _sign(private_key, manifest), {PUBLISHER: pem}) + + assert trust.state == TRUSTED + + +class TestEnforcePublisherTrust: + def test_valid_signature_passes_both_policies(self): + private_key, public_key = _publisher() + manifest = _manifest() + signature = _sign(private_key, manifest) + keys = {PUBLISHER: public_key} + + for policy in ("if-present", "required"): + trust = enforce_publisher_trust( + manifest, + signature, + trusted_publishers=keys, + policy=policy, + ) + assert trust.state == TRUSTED + + def test_missing_signature_fails_under_required_policy(self): + with pytest.raises(MissingPublisherSignatureError, match="unsigned"): + enforce_publisher_trust( + _manifest(), + None, + trusted_publishers={}, + policy="required", + ) + + def test_missing_signature_is_allowed_under_if_present(self): + trust = enforce_publisher_trust( + _manifest(), + None, + trusted_publishers={}, + policy="if-present", + ) + assert trust.state == UNVERIFIED + + def test_forged_signature_fails_even_if_present(self): + private_key, public_key = _publisher() + manifest = _manifest() + signature = _sign(private_key, manifest) + manifest["name"] = "Tampered Plugin" + + with pytest.raises(InvalidPublisherSignatureError, match="Invalid"): + enforce_publisher_trust( + manifest, + signature, + trusted_publishers={PUBLISHER: public_key}, + policy="if-present", + ) + + def test_untrusted_publisher_fails_even_if_present(self): + private_key, _ = _publisher() + manifest = _manifest() + + with pytest.raises(InvalidPublisherSignatureError, match=PUBLISHER): + enforce_publisher_trust( + manifest, + _sign(private_key, manifest), + trusted_publishers={}, + policy="if-present", + ) + + def test_unknown_policy_is_rejected(self): + with pytest.raises(PluginTrustConfigError): + enforce_publisher_trust( + _manifest(), + None, + trusted_publishers={}, + policy="sometimes", + ) + + +class TestEmbeddedManifestSignatureRegression: + """The shared verifier must keep the existing embedded-signature behaviour.""" + + def test_embedded_signature_is_trusted_after_refactor(self): + private_key, public_key = _publisher() + manifest = _manifest() + signed = {**manifest, "signature": _sign(private_key, manifest)} + + parsed = PluginManifest.from_dict(signed, trusted_publishers={PUBLISHER: public_key}) + + assert parsed.trust_state == TRUSTED + assert parsed.trust_publisher == PUBLISHER + + def test_embedded_forgery_is_invalid_in_lenient_mode(self): + private_key, public_key = _publisher() + manifest = _manifest() + signed = {**manifest, "signature": _sign(private_key, manifest)} + signed["name"] = "Tampered Plugin" + + parsed = PluginManifest.from_dict(signed, trusted_publishers={PUBLISHER: public_key}) + + assert parsed.trust_state == INVALID + + def test_strict_policy_still_rejects_unsigned_manifest(self): + with pytest.raises(ManifestValidationError, match="Unverified"): + PluginManifest.from_dict(_manifest(), trust_policy="required") diff --git a/tests/test_plugin_trust.py b/tests/test_plugin_trust.py new file mode 100644 index 0000000..c5404e5 --- /dev/null +++ b/tests/test_plugin_trust.py @@ -0,0 +1,225 @@ +"""Tests for trusted publisher key management (#386). + +The interesting properties are: valid key material in every supported encoding +is accepted, malformed configuration fails loudly (never silently untrusted), +and the read-only surfaces expose publisher ids only — never key bytes. +""" + +import base64 +import json + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from app.config import TestingConfig +from app.services.plugin_trust import ( + PluginTrustConfigError, + decode_ed25519_public_key, + load_trusted_publishers, + trusted_publisher_ids, + validate_trust_config, + validate_trust_policy, +) + +PUBLISHER = "acme-labs" + + +def _keypair(): + private_key = Ed25519PrivateKey.generate() + raw = private_key.public_key().public_bytes( + serialization.Encoding.Raw, + serialization.PublicFormat.Raw, + ) + return private_key, raw + + +def _base64_key(): + _, raw = _keypair() + return base64.b64encode(raw).decode() + + +class TestDecodeEd25519PublicKey: + def test_accepts_base64_raw_key(self): + _, raw = _keypair() + assert decode_ed25519_public_key(base64.b64encode(raw).decode()) == raw + + def test_accepts_raw_bytes(self): + _, raw = _keypair() + assert decode_ed25519_public_key(raw) == raw + + def test_accepts_pem_subject_public_key_info(self): + private_key, raw = _keypair() + pem = private_key.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + assert decode_ed25519_public_key(pem.decode()) == raw + + @pytest.mark.parametrize("value", ["", " ", "not-base64!!", "a" * 10]) + def test_rejects_malformed_keys(self, value): + with pytest.raises(PluginTrustConfigError): + decode_ed25519_public_key(value) + + def test_rejects_non_ed25519_pem(self): + other = ec.generate_private_key(ec.SECP256R1()) + pem = other.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + with pytest.raises(PluginTrustConfigError, match="Ed25519"): + decode_ed25519_public_key(pem.decode()) + + def test_rejects_no_bytes(self): + with pytest.raises(PluginTrustConfigError): + decode_ed25519_public_key(123) + + def test_error_never_echoes_key_material(self): + secret = base64.b64encode(b"x" * 31).decode() + with pytest.raises(PluginTrustConfigError) as exc_info: + decode_ed25519_public_key(secret) + assert secret not in str(exc_info.value) + + +class TestLoadTrustedPublishers: + def test_empty_configuration(self): + assert load_trusted_publishers(None) == {} + assert load_trusted_publishers({}) == {} + assert load_trusted_publishers("") == {} + + def test_parses_json_string(self): + key = _base64_key() + loaded = load_trusted_publishers(json.dumps({PUBLISHER: key})) + assert list(loaded) == [PUBLISHER] + assert len(loaded[PUBLISHER]) == 32 + + def test_parses_mapping(self): + key = _base64_key() + assert list(load_trusted_publishers({PUBLISHER: key})) == [PUBLISHER] + + def test_rejects_invalid_json(self): + with pytest.raises(PluginTrustConfigError, match="JSON object"): + load_trusted_publishers("{not json") + + def test_rejects_non_object_json(self): + with pytest.raises(PluginTrustConfigError, match="JSON object"): + load_trusted_publishers("[1, 2, 3]") + + @pytest.mark.parametrize("publisher", ["Acme", "acme labs", "", "-acme", "a" * 65]) + def test_rejects_non_canonical_publisher_ids(self, publisher): + with pytest.raises(PluginTrustConfigError, match="publisher"): + load_trusted_publishers({publisher: _base64_key()}) + + def test_invalid_key_names_the_publisher_but_not_the_key(self): + bad = base64.b64encode(b"short").decode() + with pytest.raises(PluginTrustConfigError) as exc_info: + load_trusted_publishers({PUBLISHER: bad}) + message = str(exc_info.value) + assert PUBLISHER in message + assert bad not in message + + +class TestTrustPolicy: + @pytest.mark.parametrize("policy", ["if-present", "required"]) + def test_accepts_known_policies(self, policy): + assert validate_trust_policy(policy) == policy + + @pytest.mark.parametrize("policy", ["", "lenient", None, "REQUIRED"]) + def test_rejects_unknown_policies(self, policy): + with pytest.raises(PluginTrustConfigError, match="PLUGIN_TRUST_POLICY"): + validate_trust_policy(policy) + + +class TestValidateTrustConfig: + def test_valid_configuration(self): + publishers = validate_trust_config( + { + "PLUGIN_TRUSTED_KEYS": {PUBLISHER: _base64_key()}, + "PLUGIN_TRUST_POLICY": "required", + } + ) + assert list(publishers) == [PUBLISHER] + + def test_invalid_key_is_reported_with_the_setting_name(self): + with pytest.raises(PluginTrustConfigError, match="PLUGIN_TRUSTED_KEYS"): + validate_trust_config({"PLUGIN_TRUSTED_KEYS": {PUBLISHER: "bogus"}}) + + def test_invalid_policy_is_reported_with_the_setting_name(self): + with pytest.raises(PluginTrustConfigError, match="PLUGIN_TRUST_POLICY"): + validate_trust_config({"PLUGIN_TRUSTED_KEYS": {}, "PLUGIN_TRUST_POLICY": "maybe"}) + + +class TestReadOnlySurfaces: + def test_ids_are_sorted_and_secret_free(self): + key_one = _base64_key() + key_two = _base64_key() + raw = {"zeta": key_one, "alpha": key_two} + + ids = trusted_publisher_ids(raw) + + assert ids == ["alpha", "zeta"] + assert key_one not in json.dumps(ids) + assert key_two not in json.dumps(ids) + + def test_api_lists_publisher_ids_only(self, app, client, make_user, login): + key = _base64_key() + app.config["PLUGIN_TRUSTED_KEYS"] = {PUBLISHER: key} + app.config["PLUGIN_TRUST_POLICY"] = "required" + make_user() + login() + + resp = client.get("/plugins/api/trusted-publishers") + + assert resp.status_code == 200 + data = resp.get_json() + assert data == {"publishers": [PUBLISHER], "count": 1, "policy": "required"} + assert key not in resp.get_data(as_text=True) + + def test_api_requires_login(self, app, client): + app.config["PLUGIN_TRUSTED_KEYS"] = {PUBLISHER: _base64_key()} + resp = client.get("/plugins/api/trusted-publishers") + assert resp.status_code in (302, 401) + + def test_cli_json_and_human_output(self, app): + key = _base64_key() + app.config["PLUGIN_TRUSTED_KEYS"] = {PUBLISHER: key} + runner = app.test_cli_runner() + + as_json = runner.invoke(args=["plugins", "publishers", "--json"]) + assert as_json.exit_code == 0 + assert json.loads(as_json.output) == { + "publishers": [PUBLISHER], + "count": 1, + "policy": "if-present", + } + + human = runner.invoke(args=["plugins", "publishers"]) + assert human.exit_code == 0 + assert PUBLISHER in human.output + assert key not in human.output + + def test_cli_empty_configuration(self, app): + app.config["PLUGIN_TRUSTED_KEYS"] = {} + result = app.test_cli_runner().invoke(args=["plugins", "publishers"]) + assert result.exit_code == 0 + assert "No trusted publishers configured." in result.output + + +class TestStartupValidation: + def test_create_app_rejects_invalid_key_material(self, monkeypatch): + from app import create_app + + monkeypatch.setattr(TestingConfig, "PLUGIN_TRUSTED_KEYS", {PUBLISHER: "bogus"}) + + with pytest.raises(RuntimeError, match="Invalid plugin trust configuration"): + create_app("testing") + + def test_create_app_rejects_invalid_policy(self, monkeypatch): + from app import create_app + + monkeypatch.setattr(TestingConfig, "PLUGIN_TRUSTED_KEYS", {}) + monkeypatch.setattr(TestingConfig, "PLUGIN_TRUST_POLICY", "lenient") + + with pytest.raises(RuntimeError, match="Invalid plugin trust configuration"): + create_app("testing")