fix(security): address 8 CodeRabbit findings — aliasing, permissions, race, snapshot, signing-failure gate, 422 test

tests/test_routes_store_install.py

@@ -337,8 +337,8 @@ async def test_model_install_failure_returns_500(self, client):
         assert "model install failed" in body["error"]
 
     @pytest.mark.asyncio
-    async def test_unknown_backend_returns_500(self, client):
-        """A backend not in _BACKEND_TO_METHOD returns 500, not an exception."""
+    async def test_unknown_backend_returns_422(self, client):
+        """A backend not in _BACKEND_TO_METHOD returns 422."""
         manifest = _make_model_manifest(backend_id="totally-unknown-backend")
         reg = _make_registry(manifest)
         client._transport.app.state.registry = reg
@@ -353,7 +353,7 @@ async def test_unknown_backend_returns_500(self, client):
                 "manifest_id": "test-model",
                 "variant_id": "v1",
             })
-        assert resp.status_code == 500
+        assert resp.status_code == 422
         assert "_BACKEND_TO_METHOD" in resp.json()["error"]
 
     @pytest.mark.asyncio
@@ -430,6 +430,210 @@ async def test_response_includes_compat(self, client):
         assert "compat" in body
         assert "chain" in body
 
+    # ------------------------------------------------------------------
+    # Code-signing tests (#647)
+    # ------------------------------------------------------------------
+
+    @pytest.mark.asyncio
+    async def test_tampered_manifest_rejected_403(self, client):
+        """When registry.verify_manifest_signature returns False, the
+        install is rejected with 403."""
+        from tinyagentos.store_signing import generate_signing_keypair
+
+        manifest = _make_model_manifest()
+        reg = _make_registry(manifest)
+        reg.verify_manifest_signature = MagicMock(return_value=False)
+        client._transport.app.state.registry = reg
+
+        # Provide a signing keypair so the code-signing gate is active.
+        _, pub = generate_signing_keypair()
+        client._transport.app.state.store_signing_pubkey = pub
+
+        resp = await client.post("/api/store/install-v2", json={
+            "manifest_id": "test-model",
+            "variant_id": "v1",
+        })
+        assert resp.status_code == 403
+        body = resp.json()
+        assert "manifest signature verification failed" == body["error"]
+        assert "install_id" in body
+
+    @pytest.mark.asyncio
+    async def test_valid_signature_allows_install(self, client):
+        """When registry.verify_manifest_signature returns True, the
+        install proceeds past the signing gate."""
+        from tinyagentos.store_signing import generate_signing_keypair
+
+        manifest = _make_model_manifest()
+        reg = _make_registry(manifest)
+        reg.verify_manifest_signature = MagicMock(return_value=True)
+        reg.mark_installed = MagicMock()
+        client._transport.app.state.registry = reg
+        client._transport.app.state.installed_apps = _make_installed_apps()
+
+        _, pub = generate_signing_keypair()
+        client._transport.app.state.store_signing_pubkey = pub
+
+        cap = _cpu_cap(installed_backends=("llama-cpp",))
+        with patch(
+            "tinyagentos.routes.store_install.get_device_capability",
+            new=AsyncMock(return_value=cap),
+        ), patch(
+            "tinyagentos.routes.store_install.get_installer",
+        ) as mock_get:
+            model_inst = MagicMock()
+            model_inst.install = AsyncMock(return_value={"success": True})
+            mock_get.return_value = model_inst
+
+            resp = await client.post("/api/store/install-v2", json={
+                "manifest_id": "test-model",
+                "variant_id": "v1",
+            })
+        assert resp.status_code == 200
+        body = resp.json()
+        assert "chain" in body
+
+    @pytest.mark.asyncio
+    async def test_real_registry_detects_post_load_tampering(self, client):
+        """End-to-end test: a real AppRegistry re-reads the manifest from
+        disk at install time and rejects a manifest that was tampered with
+        after catalog load with 403.
+
+        Unlike the mocked tests above that stub verify_manifest_signature,
+        this exercises the full path: sign-at-load, mutate-the-file,
+        verify-at-install.
+        """
+        import tempfile
+        from pathlib import Path
+
+        from tinyagentos.registry import AppRegistry
+        from tinyagentos.store_signing import generate_signing_keypair
+
+        # 1. Create a catalog directory with one service manifest on disk.
+        catalog_dir = Path(tempfile.mkdtemp())
+        svc_dir = catalog_dir / "services" / "test-svc"
+        svc_dir.mkdir(parents=True)
+        manifest_path = svc_dir / "manifest.yaml"
+        manifest_path.write_text(
+            "id: test-svc\n"
+            "name: Test Service\n"
+            "type: service\n"
+            "version: \"1.0\"\n"
+            "install:\n"
+            "  method: download\n",
+        )
+
+        # 2. Build a real AppRegistry with a signing key.
+        priv, pub = generate_signing_keypair()
+        installed_path = Path(tempfile.mkstemp(suffix=".json")[1])
+        installed_path.write_text("[]")  # initialise with valid JSON
+        reg = AppRegistry(
+            catalog_dir=catalog_dir,
+            installed_path=installed_path,
+            signing_key=priv,
+        )
+        # Load the catalog to populate _signatures.
+        reg._ensure_loaded()
+        assert reg.get_signature("test-svc") is not None, (
+            "expected a stored signature for test-svc"
+        )
+
+        # 3. Wire the real registry into the app state.
+        client._transport.app.state.registry = reg
+        client._transport.app.state.store_signing_pubkey = pub
+        client._transport.app.state.installed_apps = _make_installed_apps()
+
+        # 4. Before tampering: the signature should verify and the install
+        # proceeds past the gate.  (The legacy installer may fail because
+        # there is no real download URL, but the HTTP status is NOT 403.)
+        resp = await client.post("/api/store/install-v2", json={
+            "manifest_id": "test-svc",
+        })
+        assert resp.status_code != 403, (
+            f"expected install to pass the signing gate, got 403: {resp.json()}"
+        )
+
+        # 5. Tamper with the manifest on disk.
+        manifest_path.write_text(
+            "id: test-svc\n"
+            "name: EVIL Service\n"
+            "type: service\n"
+            "version: \"1.0\"\n"
+            "install:\n"
+            "  method: download\n",
+        )
+
+        # 6. Now the install MUST be rejected with 403.
+        resp = await client.post("/api/store/install-v2", json={
+            "manifest_id": "test-svc",
+        })
+        assert resp.status_code == 403, (
+            f"expected 403 after tampering, got {resp.status_code}: {resp.json()}"
+        )
+        body = resp.json()
+        assert body["error"] == "manifest signature verification failed"
+        assert "install_id" in body
+
+    @pytest.mark.asyncio
+    async def test_no_signing_key_skips_verification(self, client):
+        """When store_signing_pubkey is not set, the signing gate is
+        skipped and the install proceeds normally."""
+        manifest = _make_model_manifest()
+        reg = _make_registry(manifest)
+        reg.mark_installed = MagicMock()
+        client._transport.app.state.registry = reg
+        client._transport.app.state.installed_apps = _make_installed_apps()
+
+        # No store_signing_pubkey — simulates a taOS instance without
+        # signing configured (graceful degradation).
+        client._transport.app.state.store_signing_pubkey = None
+
+        cap = _cpu_cap(installed_backends=("llama-cpp",))
+        with patch(
+            "tinyagentos.routes.store_install.get_device_capability",
+            new=AsyncMock(return_value=cap),
+        ), patch(
+            "tinyagentos.routes.store_install.get_installer",
+        ) as mock_get:
+            model_inst = MagicMock()
+            model_inst.install = AsyncMock(return_value={"success": True})
+            mock_get.return_value = model_inst
+
+            resp = await client.post("/api/store/install-v2", json={
+                "manifest_id": "test-model",
+                "variant_id": "v1",
+            })
+        assert resp.status_code == 200
+
+
+# ---------------------------------------------------------------------------
+# GET /api/store/signing-pubkey
+# ---------------------------------------------------------------------------
+
+
+class TestSigningPubkey:
+    @pytest.mark.asyncio
+    async def test_returns_public_key_when_configured(self, client):
+        from tinyagentos.store_signing import generate_signing_keypair
+
+        _, pub = generate_signing_keypair()
+        client._transport.app.state.store_signing_pubkey = pub
+
+        resp = await client.get("/api/store/signing-pubkey")
+        assert resp.status_code == 200
+        body = resp.json()
+        assert "public_key_pem" in body
+        assert body["public_key_pem"] == pub.decode()
+
+    @pytest.mark.asyncio
+    async def test_returns_404_when_not_configured(self, client):
+        client._transport.app.state.store_signing_pubkey = None
+
+        resp = await client.get("/api/store/signing-pubkey")
+        assert resp.status_code == 404
+        body = resp.json()
+        assert "error" in body
+
 
 # ---------------------------------------------------------------------------
 # POST /api/store/install-v2 -- agent-framework manifests (method: script) (#1582)

tests/test_store_signing.py

@@ -0,0 +1,115 @@
+"""Unit tests for tinyagentos/store_signing.py."""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+from tinyagentos.store_signing import (
+    generate_signing_keypair,
+    load_or_create_signing_keypair,
+    sign_manifest,
+    verify_manifest_signature,
+)
+
+
+class TestGenerateKeypair:
+    def test_generates_valid_keypair(self):
+        priv, pub = generate_signing_keypair()
+        assert len(priv) > 0
+        assert len(pub) > 0
+        assert priv.startswith(b"-----BEGIN PRIVATE KEY-----")
+        assert pub.startswith(b"-----BEGIN PUBLIC KEY-----")
+
+    def test_keypair_can_sign_and_verify(self):
+        priv, pub = generate_signing_keypair()
+        manifest = {"id": "test", "name": "Test", "type": "model", "version": "1.0.0"}
+        sig = sign_manifest(manifest, priv)
+        assert len(sig) == 128  # Ed25519 signature is 64 bytes → 128 hex chars
+        assert verify_manifest_signature(manifest, sig, pub)
+
+
+class TestSignAndVerify:
+    def setup_method(self):
+        self.priv, self.pub = generate_signing_keypair()
+        self.manifest = {
+            "id": "ollama",
+            "name": "Ollama",
+            "type": "service",
+            "version": "latest",
+            "install": {"method": "script"},
+        }
+
+    def test_verify_valid_signature(self):
+        sig = sign_manifest(self.manifest, self.priv)
+        assert verify_manifest_signature(self.manifest, sig, self.pub)
+
+    def test_verify_tampered_signature(self):
+        sig = sign_manifest(self.manifest, self.priv)
+        # Flip the last byte so it ALWAYS differs from the original.
+        # sig[:-2]+"ff" fails ~1/256 of the time when the last byte
+        # already happens to be "ff".
+        last_byte = int(sig[-2:], 16) ^ 0x01
+        bad_sig = sig[:-2] + f"{last_byte:02x}"
+        assert not verify_manifest_signature(self.manifest, bad_sig, self.pub)
+
+    def test_verify_tampered_manifest(self):
+        sig = sign_manifest(self.manifest, self.priv)
+        tampered = {**self.manifest, "name": "Evil Ollama"}
+        assert not verify_manifest_signature(tampered, sig, self.pub)
+
+    def test_verify_empty_signature(self):
+        assert not verify_manifest_signature(self.manifest, "", self.pub)
+
+    def test_verify_wrong_key(self):
+        sig = sign_manifest(self.manifest, self.priv)
+        _, other_pub = generate_signing_keypair()
+        assert not verify_manifest_signature(self.manifest, sig, other_pub)
+
+    def test_signature_is_deterministic_for_same_input(self):
+        """Ed25519 is deterministic — same input + key = same signature."""
+        sig1 = sign_manifest(self.manifest, self.priv)
+        sig2 = sign_manifest(self.manifest, self.priv)
+        assert sig1 == sig2
+
+    def test_different_manifests_produce_different_signatures(self):
+        sig1 = sign_manifest(self.manifest, self.priv)
+        sig2 = sign_manifest({**self.manifest, "id": "other"}, self.priv)
+        assert sig1 != sig2
+
+    def test_signature_field_stripped(self):
+        """The _signature field is stripped before signing so embedding it
+        doesn't create a circular dependency."""
+        manifest_with_sig = {**self.manifest, "_signature": "should-be-ignored"}
+        sig = sign_manifest(manifest_with_sig, self.priv)
+        # Verify against the same manifest (with _signature field still there)
+        assert verify_manifest_signature(manifest_with_sig, sig, self.pub)
+        # Verify against clean manifest
+        assert verify_manifest_signature(self.manifest, sig, self.pub)
+
+
+class TestLoadOrCreateKeypair:
+    def test_creates_keypair_on_first_call(self):
+        with tempfile.TemporaryDirectory() as td:
+            td = Path(td)
+            priv, pub = load_or_create_signing_keypair(td)
+            assert (td / "store_signing_key.json").exists()
+            assert len(priv) > 0
+            assert len(pub) > 0
+
+    def test_returns_same_keypair_on_second_call(self):
+        with tempfile.TemporaryDirectory() as td:
+            td = Path(td)
+            priv1, pub1 = load_or_create_signing_keypair(td)
+            priv2, pub2 = load_or_create_signing_keypair(td)
+            assert priv1 == priv2
+            assert pub1 == pub2
+
+    def test_file_permissions_are_restrictive(self):
+        with tempfile.TemporaryDirectory() as td:
+            td = Path(td)
+            load_or_create_signing_keypair(td)
+            keyfile = td / "store_signing_key.json"
+            stat = keyfile.stat()
+            # 0o600 = owner read+write only
+            assert (stat.st_mode & 0o777) == 0o600