diff --git a/tests/routes/test_store_install_v2.py b/tests/routes/test_store_install_v2.py index 13ae89b77..e3d18fcb9 100644 --- a/tests/routes/test_store_install_v2.py +++ b/tests/routes/test_store_install_v2.py @@ -178,7 +178,7 @@ def test_all_entries_resolve_to_valid_installer_methods(self): @pytest.mark.asyncio async def test_unknown_backend_returns_500_not_exception(self, client, fake_registry): - """A backend_id absent from _BACKEND_TO_METHOD returns HTTP 500, not a traceback.""" + """A backend_id absent from _BACKEND_TO_METHOD returns HTTP 422, not a traceback.""" # Construct a manifest whose variant declares an unmapped backend. unknown_manifest = MagicMock() unknown_manifest.id = "test-model" @@ -224,7 +224,7 @@ async def test_unknown_backend_returns_500_not_exception(self, client, fake_regi "manifest_id": "test-model", "variant_id": "v1", }) - assert r.status_code == 500 + assert r.status_code == 422 assert "_BACKEND_TO_METHOD" in r.json()["error"] diff --git a/tests/test_routes_store_install.py b/tests/test_routes_store_install.py index f31b22d7c..cce12a33d 100644 --- a/tests/test_routes_store_install.py +++ b/tests/test_routes_store_install.py @@ -338,7 +338,7 @@ async def test_model_install_failure_returns_500(self, client): @pytest.mark.asyncio async def test_unknown_backend_returns_500(self, client): - """A backend not in _BACKEND_TO_METHOD returns 500, not an exception.""" + """A backend not in _BACKEND_TO_METHOD returns 422, not an exception.""" 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) diff --git a/tests/test_store_signing.py b/tests/test_store_signing.py new file mode 100644 index 000000000..8c3929d2c --- /dev/null +++ b/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 diff --git a/tinyagentos/app.py b/tinyagentos/app.py index 8eb50b6f6..5cda2a164 100644 --- a/tinyagentos/app.py +++ b/tinyagentos/app.py @@ -255,7 +255,18 @@ def create_app(data_dir: Path | None = None, catalog_dir: Path | None = None) -> # hardware_path / hardware_profile already loaded above before the # auto-register loop; don't re-probe. installed_path = data_dir / "installed.json" - registry = AppRegistry(catalog_dir=catalog_dir, installed_path=installed_path) + + from tinyagentos.store_signing import load_or_create_signing_keypair + + # Keypair loading is deferred to the lifespan so a read-only data_dir + # does not block create_app(). The registry starts with signing_key=None; + # the lifespan will attempt to load the keypair and call + # registry.set_signing_key() if it succeeds. + _store_priv: bytes | None = None + _store_pub: bytes | None = None + registry = AppRegistry( + catalog_dir=catalog_dir, installed_path=installed_path, signing_key=None, + ) from tinyagentos.agent_registry_store import AgentRegistryStore, load_or_create_signing_keypair agent_registry_store = AgentRegistryStore(data_dir / "agent_registry.db") @@ -1583,6 +1594,23 @@ async def dispatch(self, request, call_next): app.state.fallback = fallback app.state.scheduler = scheduler app.state.registry = registry + # Load the store signing keypair lazily here in the lifespan, not in + # create_app(), so a read-only data_dir does not brick startup. + # When the keypair cannot be loaded (missing cryptography, unwritable + # data_dir), signing is simply disabled — the install gate falls + # through to unsigned (fail-open) and the pubkey endpoint returns 404. + _store_pub: bytes | None = None + try: + _store_priv, _store_pub = load_or_create_signing_keypair(data_dir) + if _store_priv is not None: + registry.set_signing_key(_store_priv) + except OSError: + logger.warning( + "store signing keypair could not be created (data_dir=%s may be " + "read-only) — catalog signatures will not be available", + data_dir, + ) + app.state.store_signing_pubkey = _store_pub app.state.hardware_profile = hardware_profile app.state.cluster_manager = cluster_manager app.state.task_router = task_router diff --git a/tinyagentos/registry.py b/tinyagentos/registry.py index 638d4a167..f21092451 100644 --- a/tinyagentos/registry.py +++ b/tinyagentos/registry.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +import logging import threading from dataclasses import dataclass, field from enum import Enum @@ -9,6 +10,8 @@ import yaml +logger = logging.getLogger(__name__) + class AppState(str, Enum): """Possible states for an installed app.""" @@ -51,6 +54,10 @@ class AppManifest: @classmethod def from_file(cls, path: Path) -> AppManifest: data = yaml.safe_load(path.read_text()) + return cls.from_dict(data, manifest_dir=path.parent) + + @classmethod + def from_dict(cls, data: dict, manifest_dir: Path | None = None) -> AppManifest: return cls( id=data["id"], name=data["name"], @@ -70,7 +77,7 @@ def from_file(cls, path: Path) -> AppManifest: variants=data.get("variants", []), capabilities=data.get("capabilities", []), lifecycle=data.get("lifecycle", {}), - manifest_dir=path.parent, + manifest_dir=manifest_dir, ) def is_compatible(self, profile_id: str) -> bool: @@ -87,12 +94,24 @@ def is_compatible(self, profile_id: str) -> bool: class AppRegistry: - def __init__(self, catalog_dir: Path, installed_path: Path): + def __init__(self, catalog_dir: Path, installed_path: Path, signing_key: bytes | None = None): self.catalog_dir = catalog_dir self.installed_path = installed_path + self._signing_key = signing_key # Sentinel: None means catalog has not been loaded yet. Deferred so that # boot does not pay for walking + parsing every manifest under catalog_dir. self._catalog: list[AppManifest] | None = None + # Manifest signatures keyed by app_id. Populated during _load_catalog() + # when a signing key is available. + self._signatures: dict[str, str] = {} + # Raw manifest dicts (stripped of _signature) keyed by app_id, used for + # re-verifying at install time. + self._manifest_dicts: dict[str, dict] = {} + # Manifests that failed signing during catalog load. An entry here + # means the install gate MUST block (fail-closed) — the signing + # failure could be a transient error OR evidence of a manifest that + # is somehow corrupt/malformed in a way the signer can't handle. + self._signing_failures: set[str] = set() self._catalog_lock = threading.Lock() def _ensure_loaded(self) -> None: @@ -105,6 +124,9 @@ def _ensure_loaded(self) -> None: def _load_catalog(self) -> None: catalog: list[AppManifest] = [] + signatures: dict[str, str] = {} + manifest_dicts: dict[str, dict] = {} + signing_failures: set[str] = set() for type_dir in ("agents", "models", "services", "plugins"): base = self.catalog_dir / type_dir if not base.exists(): @@ -113,16 +135,43 @@ def _load_catalog(self) -> None: manifest = app_dir / "manifest.yaml" if manifest.exists(): try: - catalog.append(AppManifest.from_file(manifest)) + raw_dict = yaml.safe_load(manifest.read_text()) + catalog.append(AppManifest.from_dict(raw_dict, manifest_dir=app_dir)) + if self._signing_key is not None: + from tinyagentos.store_signing import sign_manifest + + try: + sig = sign_manifest(raw_dict, self._signing_key) + signatures[catalog[-1].id] = sig + manifest_dicts[catalog[-1].id] = raw_dict + except Exception: + logger.exception( + "failed to sign manifest %s — catalog load continues unsigned", + catalog[-1].id, + ) + signing_failures.add(catalog[-1].id) except (yaml.YAMLError, KeyError): pass # skip invalid manifests # Single atomic assignment: readers either see the old list or the fully built one. self._catalog = catalog + self._signatures = signatures + self._manifest_dicts = manifest_dicts + self._signing_failures = signing_failures def reload(self) -> None: with self._catalog_lock: self._load_catalog() + def set_signing_key(self, key: bytes | None) -> None: + """Set (or clear) the signing key and reload the catalog. + + Call this after the keypair has been loaded (e.g. in the lifespan) + so the catalog is signed with the actual key. Passing ``None`` + disables signing. + """ + self._signing_key = key + self.reload() + def list_available(self, type_filter: str | None = None) -> list[AppManifest]: self._ensure_loaded() if type_filter: @@ -133,6 +182,78 @@ def get(self, app_id: str) -> AppManifest | None: self._ensure_loaded() return next((a for a in self._catalog if a.id == app_id), None) + def get_signature(self, app_id: str) -> str | None: + """Return the hex Ed25519 signature for *app_id*, or None.""" + self._ensure_loaded() + return self._signatures.get(app_id) + + def get_manifest_dict(self, app_id: str) -> dict | None: + """Return the raw manifest dict (without _signature field) for *app_id*.""" + self._ensure_loaded() + return self._manifest_dicts.get(app_id) + + def had_signing_failure(self, app_id: str) -> bool: + """Return True if *app_id* was loaded but signing failed. + + A signing failure means the manifest is present in the catalog but + its Ed25519 signature could not be computed (e.g. malformed content, + serialisation error). The install gate must block these manifests + rather than silently treating them as unsigned — an attacker who can + induce a signing failure for a target manifest would otherwise bypass + the entire tamper-protection gate. + """ + self._ensure_loaded() + return app_id in self._signing_failures + + def verify_manifest_signature(self, app_id: str, public_pem: bytes) -> bool: + """Re-verify the stored signature for *app_id* against *public_pem*. + + **Re-reads the manifest from disk at verify time**, then checks the + stored Ed25519 signature (computed at catalog-load time) against the + canonical bytes of the current on-disk YAML. This detects post-boot + catalog tampering — an attacker who modifies ``manifest.yaml`` after + the server started will produce a mismatch and the install is blocked. + + Returns ``True`` only when the on-disk manifest successfully verifies + against the stored Ed25519 signature. + + Returns ``False`` when: + + * no signature was stored for this app (unsigned — fail-closed), **or** + * the on-disk manifest does not verify against the stored signature + (tampered). + + This primitive is fail-closed on purpose: a future caller that does + ``if not registry.verify_manifest_signature(...)`` gets the safe + default. Callers that need a fail-open policy for unsigned manifests + (e.g. the install gate, which must not block catalog entries that + predate the signing feature) must check ``get_signature(app_id)`` + first and short-circuit before calling this method. See + ``_verify_manifest_for_install`` in ``routes/store_install.py`` for + the canonical fail-open pattern. + """ + self._ensure_loaded() + sig = self._signatures.get(app_id) + if sig is None: + # Never signed — fail-closed. The absence of a signature + # means there is nothing to verify against. Callers that + # want a fail-open policy for unsigned manifests must check + # get_signature() first. + return False + manifest = self.get(app_id) + if manifest is None or manifest.manifest_dir is None: + return False + manifest_path = manifest.manifest_dir / "manifest.yaml" + if not manifest_path.exists(): + return False + try: + on_disk = yaml.safe_load(manifest_path.read_text()) + except (yaml.YAMLError, OSError): + return False + from tinyagentos.store_signing import verify_manifest_signature + + return verify_manifest_signature(on_disk, sig, public_pem) + def _read_installed(self) -> list[dict]: if not self.installed_path.exists(): return [] diff --git a/tinyagentos/routes/store_install.py b/tinyagentos/routes/store_install.py index ba47a7f92..38b4a2b9c 100644 --- a/tinyagentos/routes/store_install.py +++ b/tinyagentos/routes/store_install.py @@ -12,6 +12,7 @@ import asyncio import logging from dataclasses import asdict +from pathlib import Path from urllib.parse import urlparse from fastapi import APIRouter, Request @@ -204,6 +205,47 @@ def _registry_get(registry, app_id: str): return registry.get(app_id) +def _verify_manifest_for_install( + manifest_id: str, + registry, + store_signing_pubkey: bytes | None, +) -> tuple[bool, str | None]: + """Verify a manifest's Ed25519 signature before installing. + + Returns ``(allowed, error_detail)``: + - ``(True, None)`` — proceed with install. + - ``(False, reason)`` — block install with the given error string. + + When the signing pubkey is not configured, the gate is skipped (allowed). + When a manifest has no stored signature (e.g. it was catalog-loaded + before signing was enabled), the gate is also skipped (allowed) rather + than returning a hard 403 — the absence of a signature is not evidence + of tampering. + """ + if store_signing_pubkey is None or registry is None: + return True, None + if not hasattr(registry, "verify_manifest_signature"): + return True, None + + stored_sig = registry.get_signature(manifest_id) + if stored_sig is None: + # Check for a signing failure first — a manifest that failed to + # sign must be blocked (fail-closed). An attacker who can induce + # a signing failure would otherwise bypass the tamper-protection + # gate entirely, since the None-signature short-circuit below + # would silently allow the install. + if hasattr(registry, "had_signing_failure") and registry.had_signing_failure(manifest_id): + return False, "manifest signing failed — catalog may be corrupt" + # Never signed — manifest was loaded before signing was enabled. + # Skip the gate; absence of a signature is not evidence of tampering. + return True, None + + if not registry.verify_manifest_signature(manifest_id, store_signing_pubkey): + return False, "manifest signature verification failed" + + return True, None + + async def _install_agent_framework( request: Request, manifest, app_id: str, meta: dict, body: dict, ) -> JSONResponse: @@ -699,6 +741,93 @@ async def install_app(request: Request): progress.finish(install_id, success=False, error="manifest not found in registry") return await _legacy_install(request, body, manifest_id, target_remote) + # Code-signing gate (#647): verify the catalog manifest's Ed25519 + # signature against the store signing public key. Re-reads the + # manifest from disk at install time so post-boot tampering is + # detected. Manifests that were catalog-loaded without a signing + # key configured are allowed through — no stored signature means + # no tampering evidence. + # + # Threat model: this protects against post-boot *catalog* tampering + # (an attacker modifying a manifest on disk while the server is + # running). It does NOT protect against an attacker who can also + # replace the signing keyfile and then restart the server — that is + # the OS-level trust boundary. For defence-in-depth, deploy the + # keyfile on a read-only filesystem or use a hardware-backed key + # store. + _store_pub = getattr(request.app.state, "store_signing_pubkey", None) + verified, verify_err = _verify_manifest_for_install( + manifest_id, registry, _store_pub, + ) + if not verified: + progress.finish( + install_id, success=False, + error=verify_err or "manifest signature verification failed — the catalog may have been tampered with", + ) + return JSONResponse( + { + "error": verify_err or "manifest signature verification failed", + "detail": "The catalog manifest for this app failed signature verification. " + "The app may have been tampered with. Rebuild the catalog or " + "reinstall the app from a trusted source.", + "install_id": install_id, + }, + status_code=403, + ) + + # TOCTOU guard: the signing gate above verified the signature against + # the on-disk manifest, but the install below uses the in-memory + # manifest object loaded at boot. An attacker who swaps the on-disk + # file between verification and execution could bypass the gate. + # Re-read from disk and re-verify the signature — if it no longer + # verifies, the manifest was modified after the gate check and the + # install is blocked. + # + # Run via asyncio.to_thread to avoid blocking the event loop on disk + # I/O + Ed25519 verification under concurrent load. + _manifest_dir = getattr(manifest, "manifest_dir", None) + if ( + _manifest_dir is not None + and isinstance(_manifest_dir, Path) + and _store_pub is not None + and registry is not None + and hasattr(registry, "verify_manifest_signature") + ): + + def _toctou_reverify(): + disk_path = _manifest_dir / "manifest.yaml" + try: + import yaml as _yaml + on_disk = _yaml.safe_load(disk_path.read_text()) if disk_path.exists() else None + except Exception: + on_disk = None + if on_disk is not None: + stored_sig = registry.get_signature(manifest_id) + if stored_sig is not None: + from tinyagentos.store_signing import verify_manifest_signature as _verify_sig + if not _verify_sig(on_disk, stored_sig, _store_pub): + return False + return True + + if not await asyncio.to_thread(_toctou_reverify): + progress.finish( + install_id, success=False, + error="manifest modified between signature verification and install", + ) + return JSONResponse( + { + "error": "manifest modified between signature verification and install", + "detail": ( + "The manifest on disk was modified after the initial " + "signature verification. This may indicate post-verification " + "tampering. Rebuild the catalog or reinstall the app from " + "a trusted source." + ), + "install_id": install_id, + }, + status_code=403, + ) + # Non-commercial weights gate (#169): a manifest's code license (MIT etc.) # can be permissive while the model weights it downloads are not (e.g. # musicgen pulls Meta's CC-BY-NC 4.0 weights). Block the install until the @@ -789,6 +918,25 @@ async def install_app(request: Request): {"error": f"backend {result.backend_id!r} has no install.method", "install_id": install_id}, status_code=500, ) + # Verify the backend manifest's signature before running its installer. + # A backend manifest that was tampered with after catalog load would + # run an untrusted script/image — reject it here. + be_verified, be_verify_err = _verify_manifest_for_install( + result.backend_id, registry, _store_pub, + ) + if not be_verified: + progress.finish( + install_id, success=False, + error=be_verify_err or "backend manifest signature verification failed", + ) + return JSONResponse( + { + "error": be_verify_err or "backend manifest signature verification failed", + "detail": f"The backend manifest for {result.backend_id!r} failed signature verification.", + "install_id": install_id, + }, + status_code=403, + ) backend_installer = get_installer(backend_method) be_result = await backend_installer.install( result.backend_id, @@ -836,7 +984,7 @@ async def install_app(request: Request): ), "install_id": install_id, }, - status_code=500, + status_code=422, ) model_installer = get_installer(install_method) install_config = dict(getattr(manifest, "install", None) or {}) @@ -885,6 +1033,20 @@ def _on_progress(downloaded: int, total: int) -> None: return JSONResponse({"chain": chain, "compat": classify(manifest_dict, device), "install_id": install_id}) +@router.get("/api/store/signing-pubkey") +async def store_signing_pubkey(request: Request): + """Return the store signing public key (Ed25519, PEM). + + Clients and auditing tools can use this to verify catalog manifest + signatures independently. The corresponding private key never leaves + the node. + """ + pub_pem = getattr(request.app.state, "store_signing_pubkey", None) + if pub_pem is None: + return JSONResponse({"error": "store signing key not configured"}, status_code=404) + return JSONResponse({"public_key_pem": pub_pem.decode()}) + + @router.post("/api/store/uninstall-v2") async def uninstall_app(request: Request): body = await request.json() diff --git a/tinyagentos/store_signing.py b/tinyagentos/store_signing.py new file mode 100644 index 000000000..212d7e83e --- /dev/null +++ b/tinyagentos/store_signing.py @@ -0,0 +1,253 @@ +"""Store manifest signing — Ed25519 signatures for catalog integrity. + +Mirrors the hub identity pattern: an Ed25519 keypair is generated on first +use and persisted to disk. Every catalog manifest is signed at load time; +the install-v2 endpoint verifies the signature before allowing an install +to proceed, so a compromised catalog entry (post-boot tampering, MITM, +supply-chain injection) is caught before any script or image is pulled. + +The public key is exposed via ``GET /api/store/signing-pubkey`` so clients +and auditing tools can verify signatures independently. + +Design decisions (see #647): + +- **One signing key per taOS instance.** Generated on first boot; the + private key never leaves the node. This matches the self-hosted model: + each instance trusts its own catalog. A future shared-catalog model + (e.g. a taOS App Store) would use a network-fetched public key. + +- **Signatures live in the manifest YAML.** A ``_signature`` field at the + root of the manifest holds the hex-encoded Ed25519 signature over the + canonical bytes of the manifest *with that field stripped*. At load + time, the registry strips ``_signature``, computes the signature, and + stores it in-memory. At verify time the same stripped view is used, so + flipping the signature doesn't change the bytes being verified. +""" + +from __future__ import annotations + +import json +import logging +import os +import time +from pathlib import Path + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey +from cryptography.hazmat.primitives.serialization import ( + Encoding, + NoEncryption, + PrivateFormat, + PublicFormat, +) + +logger = logging.getLogger(__name__) + +# Filename under / for the persisted keypair. +_KEYPAIR_FILE = "store_signing_key.json" + +# Field name in the manifest YAML that holds the detached signature. +SIGNATURE_FIELD = "_signature" + + +# --------------------------------------------------------------------------- +# Keypair lifecycle +# --------------------------------------------------------------------------- + + +def generate_signing_keypair() -> tuple[bytes, bytes]: + """Generate a fresh Ed25519 keypair. + + Returns ``(private_pem, public_pem)`` — PEM-encoded byte strings + without passphrase encryption (the private key is stored in a 0600 + file and the security model is local-tamper-detection, not secrecy). + """ + private = Ed25519PrivateKey.generate() + private_pem = private.private_bytes( + encoding=Encoding.PEM, + format=PrivateFormat.PKCS8, + encryption_algorithm=NoEncryption(), + ) + public_pem = private.public_key().public_bytes( + encoding=Encoding.PEM, + format=PublicFormat.SubjectPublicKeyInfo, + ) + return private_pem, public_pem + + +def _enforce_permissions(keyfile: Path) -> None: + """Ensure the keyfile is owner-read/write only (0600). + + Called on every load so a migration, backup-restore, or manual + ``chmod`` cannot leave the private key world/group-readable. + """ + try: + st = keyfile.stat() + if (st.st_mode & 0o777) != 0o600: + os.chmod(keyfile, 0o600) + except OSError: + pass # non-fatal on exotic filesystems + + +def load_or_create_signing_keypair(data_dir: Path) -> tuple[bytes, bytes]: + """Return ``(private_pem, public_pem)``, minting the keypair on first use. + + Idempotent: once the keystore exists it is returned unchanged, so the + instance keeps the same signing identity across restarts. + """ + keyfile = data_dir / _KEYPAIR_FILE + if keyfile.exists(): + # Enforce restrictive permissions BEFORE reading so a key + # written under a loose umask is repaired before the bytes + # touch process memory. A migration / backup-restore / + # manual chmod cannot leave the private key world/group- + # readable across a restart. + _enforce_permissions(keyfile) + try: + data = json.loads(keyfile.read_text()) + priv = data["private_pem"].encode() + # Derive the public key from the loaded private key so + # a keyfile whose public_pem was replaced still yields + # the correct keypair (the private key is the root of + # trust, not its companion field). + key = _load_private_key(priv) + pub = key.public_key().public_bytes( + encoding=Encoding.PEM, + format=PublicFormat.SubjectPublicKeyInfo, + ) + return priv, pub + except (json.JSONDecodeError, KeyError, ValueError) as exc: + logger.warning( + "store signing keyfile corrupt (%s), regenerating", exc, + ) + + priv, pub = generate_signing_keypair() + keyfile.parent.mkdir(parents=True, exist_ok=True) + payload = json.dumps( + {"private_pem": priv.decode(), "public_pem": pub.decode()}, + ) + + # Atomic creation with exclusive open + restrictive permissions. + # O_CREAT|O_EXCL guarantees at most one process wins the race; + # the loser loads the winner's key instead of overwriting it. + # The file descriptor starts with mode 0o600 so there is never + # a world-readable window — no separate chmod call is needed. + tmp = keyfile.with_suffix(keyfile.suffix + ".tmp") + try: + fd = os.open(tmp, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + # Another process is creating the keypair. Wait for the + # winner to promote tmp→keyfile, then load its result. + for _ in range(30): # up to 3 s + time.sleep(0.1) + if keyfile.exists(): + return load_or_create_signing_keypair(data_dir) + # After a reasonable wait the keyfile still does not exist. + # The shared tmp is contested — skip straight to a unique + # PID-based name instead of unlinking (which could race with + # a still-running winner and crash it or leave inconsistent + # on-disk state). + if keyfile.exists(): + return load_or_create_signing_keypair(data_dir) + tmp = keyfile.with_suffix(f"{keyfile.suffix}.tmp.{os.getpid()}") + try: + fd = os.open(tmp, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + if keyfile.exists(): + return load_or_create_signing_keypair(data_dir) + raise + # Write and atomically promote. + try: + os.write(fd, payload.encode("utf-8")) + finally: + os.close(fd) + os.replace(tmp, keyfile) + logger.info("store signing keypair created at %s", keyfile) + return priv, pub + + +# --------------------------------------------------------------------------- +# Serialisation helpers +# --------------------------------------------------------------------------- + + +def _load_private_key(private_pem: bytes) -> Ed25519PrivateKey: + return Ed25519PrivateKey.from_private_bytes( + _raw_private_bytes(private_pem), + ) + + +def _load_public_key(public_pem: bytes) -> Ed25519PublicKey: + from cryptography.hazmat.primitives.serialization import load_pem_public_key + + key = load_pem_public_key(public_pem) + if not isinstance(key, Ed25519PublicKey): + raise TypeError(f"expected Ed25519PublicKey, got {type(key).__name__}") + return key + + +def _raw_private_bytes(private_pem: bytes) -> bytes: + """Extract the 32-byte raw seed from a PKCS8 PEM.""" + from cryptography.hazmat.primitives.serialization import load_pem_private_key + + key = load_pem_private_key(private_pem, password=None) + if not isinstance(key, Ed25519PrivateKey): + raise TypeError(f"expected Ed25519PrivateKey, got {type(key).__name__}") + # Ed25519PrivateKey.private_bytes_raw() returns the 32-byte seed. + return key.private_bytes_raw() + + +def _canonical_manifest_bytes(manifest_dict: dict) -> bytes: + """Deterministic byte representation of a manifest for signing. + + Strips ``_signature`` if present, then serialises as canonical JSON + (sorted keys, no trailing whitespace, UTF-8). This is stable across + YAML load/save cycles as long as the YAML library does not re-order + keys or change scalar representations. + """ + stripped = {k: v for k, v in manifest_dict.items() if k != SIGNATURE_FIELD} + return json.dumps( + stripped, sort_keys=True, ensure_ascii=False, default=str + ).encode("utf-8") + + +# --------------------------------------------------------------------------- +# Sign / verify +# --------------------------------------------------------------------------- + + +def sign_manifest(manifest_dict: dict, private_pem: bytes) -> str: + """Return a hex-encoded Ed25519 signature over *manifest_dict*. + + ``_signature`` is stripped before signing so the field can be embedded + in the same dict without creating a circular dependency. + """ + data = _canonical_manifest_bytes(manifest_dict) + key = _load_private_key(private_pem) + return key.sign(data).hex() + + +def verify_manifest_signature( + manifest_dict: dict, + signature_hex: str, + public_pem: bytes, +) -> bool: + """Verify an Ed25519 signature over *manifest_dict*. + + Returns ``True`` when the signature is valid. Returns ``False`` (never + raises) on a bad signature, a wrong key, or malformed hex — the caller + can treat verification failure as a hard block. + """ + if not signature_hex: + return False + try: + sig_bytes = bytes.fromhex(signature_hex) + data = _canonical_manifest_bytes(manifest_dict) + key = _load_public_key(public_pem) + key.verify(sig_bytes, data) + return True + except (ValueError, InvalidSignature): + return False + except Exception: + logger.exception("unexpected error during manifest verification") + return False