feat(plugins): publisher trust, signed catalog installs, and lock entries

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

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)

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

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",

app/models/plugin.py

@@ -176,3 +176,99 @@ def __repr__(self) -> str:
             f"<CapabilityGrant plugin_id={self.plugin_id} workspace_id={self.workspace_id} "
             f"capability={self.capability}>"
         )
+
+
+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"<PluginLockEntry plugin={self.plugin_id} version={self.version} "
+            f"workspace={self.workspace_id}>"
+        )