feat(hardening): build scoped Grant writers per service

active-runtime/app/aios_v2/provenance_operations.py

@@ -1,11 +1,11 @@
-"""AIOS V2 运维操作 Provenance Gate 与生产 Writer 工厂。"""
+"""AIOS V2 运维操作 Provenance Gate 与服务级 Scoped Writer 工厂。"""
 from __future__ import annotations
 
 import json
 import os
 from dataclasses import asdict, dataclass
 from pathlib import Path
-from typing import Any, Mapping, Protocol
+from typing import Any, Mapping, Protocol, Sequence
 
 from .provenance import (
     Ed25519PublicKeyVerifier,
@@ -17,12 +17,8 @@ from .provenance import (
     SQLiteProvenanceStore,
     digest_payload,
 )
-from .provenance_grants import (
-    GrantAuthorization,
-    GrantVerifier,
-    LeastPrivilegeProvenanceWriter,
-    SQLiteGrantStore,
-)
+from .provenance_grants import GrantAuthorization, GrantVerifier, SQLiteGrantStore
+from .provenance_scoped import ScopedLeastPrivilegeProvenanceWriter
 
 
 class AuthorizationWriter(Protocol):
@@ -67,7 +63,7 @@ class OperationReceipt:
 
 
 class ProvenanceOperationGate:
-    """部署、回滚和审计操作的统一授权入口。"""
+    """部署、回滚、审计和运行操作的统一授权入口。"""
 
     def __init__(self, writer: AuthorizationWriter) -> None:
         self.writer = writer
@@ -151,33 +147,45 @@ _GRANT_ENV_BY_PATH: Mapping[str, str] = {
 }
 
 
-def grant_ids_from_environment() -> dict[str, str]:
-    values = {
-        path: os.getenv(variable, "").strip()
-        for path, variable in _GRANT_ENV_BY_PATH.items()
-    }
-    configured = {path: value for path, value in values.items() if value}
-    if configured and len(configured) != len(values):
-        missing = [
-            variable
-            for path, variable in _GRANT_ENV_BY_PATH.items()
-            if not values[path]
-        ]
-        raise RuntimeError(
-            f"partial least-privilege Provenance Grant configuration; missing {missing}"
+def grant_ids_from_environment(
+    *,
+    required_paths: Sequence[str] | None = None,
+) -> dict[str, str]:
+    paths = tuple(
+        sorted(
+            {
+                ExecutionPath(path).value
+                for path in (required_paths or tuple(item.value for item in ExecutionPath))
+            }
         )
-    return configured
+    )
+    values = {
+        path: os.getenv(_GRANT_ENV_BY_PATH[path], "").strip()
+        for path in paths
+    }
+    missing = [
+        _GRANT_ENV_BY_PATH[path]
+        for path, value in values.items()
+        if not value
+    ]
+    if missing:
+        raise RuntimeError(
+            f"missing scoped Provenance Grant configuration: {missing}"
+        )
+    return values
 
 
 def build_provenance_writer_from_env(
     *,
     state_dir: str | Path | None = None,
     required: bool | None = None,
+    required_paths: Sequence[str] | None = None,
+    writer_id: str = "aios-v2-scoped-grant-writer",
 ) -> AuthorizationWriter | None:
-    """从环境构建验证 Writer。
+    """从环境构建服务级最小权限 Writer。
 
-    生产强制模式必须配置七个独立子 Grant。旧全权限 Writer 只允许在显式关闭
-   强制模式的候选/测试环境中使用。
+    生产强制模式必须显式声明当前进程需要的路径；仅加载这些路径对应的 Grant。
+    旧全权限 Writer 只允许在显式关闭强制模式的候选/测试环境中使用。
     """
 
     require_value = (
@@ -195,8 +203,6 @@ def build_provenance_writer_from_env(
     receipt_root = os.getenv(
         "AIOS_PROVENANCE_RECEIPT_DIR", str(base / "provenance-receipts")
     ).strip()
-    grant_ids = grant_ids_from_environment()
-
     configured = bool(public_key_file and key_id)
     if not configured:
         if require_value:
@@ -207,22 +213,33 @@ def build_provenance_writer_from_env(
 
     verifier = Ed25519PublicKeyVerifier.from_file(public_key_file, key_id=key_id)
     parent_store = SQLiteProvenanceStore(store_file)
-    if grant_ids:
+    normalized_paths = tuple(
+        sorted(
+            {
+                ExecutionPath(path).value
+                for path in (required_paths or ())
+            }
+        )
+    )
+
+    if require_value:
+        if not normalized_paths:
+            raise RuntimeError(
+                "production Run Provenance requires explicit service scope"
+            )
+        grant_ids = grant_ids_from_environment(required_paths=normalized_paths)
         grant_store = SQLiteGrantStore(store_file)
         trusted_grants = GrantVerifier(
             parent_store,
             grant_store,
             {key_id: verifier},
         )
-        return LeastPrivilegeProvenanceWriter(
+        return ScopedLeastPrivilegeProvenanceWriter(
             trusted_grants,
             receipt_root,
             grant_ids,
-        )
-
-    if require_value:
-        raise RuntimeError(
-            "production Run Provenance requires seven independent Grant IDs; full-rights provenance is prohibited"
+            required_paths=normalized_paths,
+            writer_id=writer_id,
         )
 
     legacy = ProvenanceVerifier(parent_store, {key_id: verifier})