[2/3] feat(memory): 角色感知授权与 Agent 代操作

.gitignore

@@ -39,3 +39,7 @@ deploy/docker/.venv-modelscope/
 .vscode/
 .claude/skills/gitcode-config.json
 CLAUDE.md
+
+# 安全模块
+# 开发子plan
+/security-plans

agent_plugin/jiuwenswarm/agent_memory_provider.py

@@ -631,6 +631,10 @@ class _InProcessClient(_AgentMemoryClient):
 
     def __init__(self, config_path: str | None) -> None:
         from api import build_kernel
+        from common.security import internal_context
+        from common.security.authentication.authentication_impl.dev_authenticator import (
+            DevAuthenticator,
+        )
         from config.config import Config
 
         config = None
@@ -645,6 +649,9 @@ def __init__(self, config_path: str | None) -> None:
         kernel = build_kernel(config=config)
         self._api = kernel.api
         self._kv = kernel.kv
+        # 身份由认证能力产出，不由调用方传入的 scope 充当（F05 §进程内调用）：
+        # scope 说「操作哪个范围」，security 说「谁在操作」。
+        self._security = internal_context(DevAuthenticator())
 
     @staticmethod
     def _to_api_scope(scope):
@@ -663,7 +670,7 @@ async def write(self, content, scope, *, tags=None, metadata=None) -> str | None
 
         units = await self._api.write_async(
             content, api_scope,
-            source=Modality.TEXT, identity=api_scope,
+            source=Modality.TEXT, security=self._security,
             tags=tags, metadata=metadata,
         )
         return units[0].id if units else None
@@ -680,7 +687,7 @@ async def search(
             self._api.recall,
             query,
             Context(scope=api_scope),
-            identity=api_scope,
+            security=self._security,
             filters=filters,
             top_k=top_k,
             disclosure=DisclosureLevel.L2,
@@ -712,7 +719,11 @@ async def evolve_extract(self, scope) -> None:
 
         # evolve 是同步+asyncio.run，必须 to_thread
         await asyncio.to_thread(
-            self._api.evolve, api_scope, EvolveMode.EXTRACT, Channel.BACKGROUND, identity=api_scope
+            self._api.evolve,
+            api_scope,
+            EvolveMode.EXTRACT,
+            Channel.BACKGROUND,
+            security=self._security,
         )
 
     async def close(self) -> None:

bootstrap/cli/__main__.py

@@ -41,14 +41,27 @@ def build_parser() -> argparse.ArgumentParser:
         description="agent-memory memory engine CLI",
     )
     parser.add_argument(
-        "--server", "--base-url", dest="server",
-        metavar="URL", default=os.environ.get("AGENT_MEMORY_SERVER"),
+        "--server",
+        "--base-url",
+        dest="server",
+        metavar="URL",
+        default=os.environ.get("AGENT_MEMORY_SERVER"),
         help="drive a running server over HTTP (Mem0 --base-url; default: in-process)",
     )
     parser.add_argument(
-        "--config", action="append", default=[], metavar="PATH",
+        "--config",
+        action="append",
+        default=[],
+        metavar="PATH",
         help="JSON config layer stacked on OFFLINE (in-process only; repeatable)",
     )
+    parser.add_argument(
+        "--api-key",
+        dest="api_key",
+        metavar="KEY",
+        default=None,
+        help="API key for --server mode (default: $AGENT_MEMORY_API_KEY)",
+    )
 
     sub = parser.add_subparsers(dest="command", required=True)
 
@@ -78,7 +91,7 @@ def main(argv: list[str] | None = None) -> int:
         sys.stderr.write("note: --config is ignored in --server (HTTP) mode\n")
 
     try:
-        client = make_client(args.server, args.config)
+        client = make_client(args.server, args.config, args.api_key)
         if args.command in ("health", "status"):
             return commands.run_health(client, args)
         if args.command == "batch":

bootstrap/cli/client.py

@@ -41,9 +41,11 @@ class EngineClient(Protocol):
     """A backend the CLI can drive: turn a (verb, payload) into (status, body)."""
 
     def call(self, verb: str, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+        """Dispatch one memory-engine verb."""
         ...
 
     def healthz(self) -> tuple[int, dict[str, Any]]:
+        """Return the backend health response."""
         ...
 
 
@@ -74,9 +76,33 @@ def server(self):
         return self._srv
 
     def call(self, verb: str, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]:
+        from auth_middleware import authenticated
         from handler import dispatch
 
-        return dispatch(self._srv, verb, payload)
+        from common.errors import AuthenticationError
+        from common.security.types import Credentials, Surface
+
+        # 进程内直连没有 HTTP header，故过一个空 Credentials。DEV 模式下得到
+        # ROOT，与现状一致（CLI 一直是全权限的）；API_KEY 模式下会认证失败——
+        # 这是**正确的**：没有凭据就不该有权限。要在 API_KEY 模式下用 CLI，
+        # 走 HttpClient 带 --api-key。
+        #
+        # 走的是与 HTTP 完全相同的中间件与 dispatch 签名（迁移计划 §5.4「HTTP、
+        # MCP、CLI、SDK 和进程内调用使用相同安全契约」）——差别只在 surface 标识
+        # 和「没有网络对端故不限流」。
+        #
+        # 认证失败转成 (401, body) 而非抛出：本方法的契约是返回状态码，
+        # 与 HttpClient.call 一致。
+        try:
+            with authenticated(
+                self._srv.authenticator,
+                Credentials(),
+                self._srv.audit,
+                surface=Surface.CLI,
+            ) as security:
+                return dispatch(self._srv, verb, payload, security)
+        except AuthenticationError as exc:
+            return 401, {"error": type(exc).__name__, "message": str(exc)}
 
     def healthz(self) -> tuple[int, dict[str, Any]]:
         return 200, {"status": "ok", "profile": self._srv.config.profile}
@@ -85,17 +111,21 @@ def healthz(self) -> tuple[int, dict[str, Any]]:
 class HttpClient:
     """Drive a running ``bootstrap`` server over HTTP (``POST /v1/<verb>``)."""
 
-    def __init__(self, base_url: str, timeout: float = 30.0) -> None:
+    def __init__(self, base_url: str, timeout: float = 30.0, api_key: str = "") -> None:
         self.base_url = base_url.rstrip("/")
         self.timeout = timeout
+        self.api_key = api_key
 
     def _request(self, method: str, path: str, body: dict | None) -> tuple[int, dict[str, Any]]:
         url = f"{self.base_url}{path}"
         data = json.dumps(body).encode("utf-8") if body is not None else None
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["Authorization"] = f"Bearer {self.api_key}"
         req = urllib.request.Request(
             url,
             data=data,
-            headers={"Content-Type": "application/json"},
+            headers=headers,
             method=method,
         )
         try:
@@ -124,8 +154,16 @@ def _read_json(resp) -> dict[str, Any]:
         return {"error": "BadResponse", "message": raw.decode("utf-8", "replace")}
 
 
-def make_client(server_url: str | None, configs: list[str] | None = None) -> EngineClient:
-    """Pick a backend: HTTP when ``server_url`` is given, else in-process."""
+def make_client(
+    server_url: str | None,
+    configs: list[str] | None = None,
+    api_key: str | None = None,
+) -> EngineClient:
+    """Pick a backend: HTTP when ``server_url`` is given, else in-process.
+
+    ``api_key`` 缺省读环境变量 ``AGENT_MEMORY_API_KEY``——让 key 不必出现在
+    shell history 与 ``ps`` 输出里。
+    """
     if server_url:
-        return HttpClient(server_url)
+        return HttpClient(server_url, api_key=api_key or os.environ.get("AGENT_MEMORY_API_KEY", ""))
     return InProcessClient(configs)

bootstrap/core/auth_middleware.py

@@ -0,0 +1,186 @@
+"""请求作用域的安全上下文——凭据提取 + ``RequestSecurityContext`` 构造。
+
+各 surface（HTTP / MCP / CLI 直连）用同一条中间件：把本形态的凭据材料归一成
+:class:`~common.security.types.Credentials`，交给装配好的 ``Authenticator``，把产出的
+``AuthContext`` 包成 :class:`~common.security.types.RequestSecurityContext` 交给
+``handler.dispatch``——这是 ``MemoryAPI`` 的唯一显式安全输入（迁移计划 §5.2 第 7 项）。
+
+**本模块不决定认证策略**——模式（dev / trusted / api_key）由配置在装配期选定，
+这里只负责「在正确的时机调用它、并保证退出时清理干净」。
+
+上下文经**参数**下传，不经 ContextVar：ContextVar 在本模块仍会设置，但已降级为
+日志/trace 的辅助传播（迁移计划 §5.2 第 10 项），授权判定不得依赖它存在。
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from contextlib import contextmanager
+from importlib import import_module
+from typing import Any, Iterator, Mapping
+
+_SRC = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "src"
+)
+if _SRC not in sys.path:
+    sys.path.append(_SRC)
+
+_security_types = import_module("common.security.types")
+reset_current = _security_types.reset_current
+set_current = _security_types.set_current
+Credentials = _security_types.Credentials
+Surface = _security_types.Surface
+
+# RequestSecurityContext 的构造规则（服务端生成 request_id、服务端时钟、attributes
+# 只由系统组件写）收在 common.security.request_context 一处，本模块只提供本形态的
+# surface 与 peer。
+new_request_context = import_module("common.security.request_context").new_request_context
+
+Scope = import_module("common.type_def").Scope
+AuditEvent = import_module("common.type_def").AuditEvent
+_errors = import_module("common.errors")
+AuthenticationError = _errors.AuthenticationError
+RateLimitedError = _errors.RateLimitedError
+
+_BEARER = "bearer "
+
+_RATE_LIMITED = "too many requests"
+
+
+def credentials_from_headers(headers: Mapping[str, Any], peer_address: str = "") -> Credentials:
+    """从 HTTP header 提取凭据。
+
+    HTTP header 名大小写不敏感（RFC 9110 §5.1）。``http.client.HTTPMessage`` 的
+    ``get`` 自己会做不敏感匹配，但传给 authenticator 的是普通 Mapping——故在这里
+    统一归一成小写键，authenticator 侧按小写常量查，两边不必各写一次 ``.lower()``。
+    """
+    normalized = {str(k).lower(): str(v) for k, v in headers.items()}
+
+    api_key = ""
+    auth = normalized.get("authorization", "")
+    bearer_len = len(_BEARER)
+    if auth[:bearer_len].lower() == _BEARER:
+        api_key = auth[bearer_len:].strip()
+    if not api_key:
+        api_key = normalized.get("x-api-key", "").strip()
+
+    return Credentials(api_key=api_key, headers=normalized, peer_address=peer_address)
+
+
+@contextmanager
+def authenticated(
+    authenticator,
+    credentials,
+    audit=None,
+    limiter=None,
+    *,
+    workload_guard=None,
+    surface=None,
+) -> Iterator[Any]:
+    """在请求作用域内建立可信 :class:`RequestSecurityContext`；退出时**必定** reset。
+
+    产出的上下文由调用方**显式**传给 ``dispatch``——它是 ``MemoryAPI`` 的唯一安全
+    输入。ContextVar 仍在这里设置，但只供日志/trace 关联，授权不读它。
+
+    reset 放 ``finally`` 是硬性要求：``ThreadingHTTPServer`` 每请求一线程，
+    但线程可能被池化复用；漏 reset 会让下一个请求继承上一个请求的身份——
+    最严重的一类越权。
+
+    ``authenticate`` 故意放在 ``try`` 之外：认证失败时没有 token 可 reset，
+    放进 try 会需要一个 ``token = None`` 的分支判断，反而更容易写错。
+
+    ``limiter`` 在 ``authenticate`` **之前**执行（F05 §请求执行流程）：认证本身就是
+    要保护的资源——API_KEY 模式下每次 authenticate 跑一次 Argon2id verify（128 MiB ×
+    time_cost=4），放在认证之后限流就等于「先让攻击者把 CPU 用掉，再告诉他
+    超限了」。``limiter=None`` 表示不限流（进程内直连 / MCP stdio 无网络对端）。
+
+    ``workload_guard`` 是昂贵操作的全局并发预算（F05 §Protection §WorkloadGuard）：
+    IP 桶限请求速率，限不住「同时在跑的 Argon2 verify 数」。耗尽即快速拒绝（429）
+    而不是排队——无界排队只是把资源耗尽从 CPU/内存转移到线程和请求队列。在 limiter
+    之后、authenticate 之前执行；acquire 成功后用 ``finally`` 释放。``None`` 表示
+    该认证实现声明不需要预算保护（见 ``Authenticator.requires_concurrency_guard``）。
+
+    ``surface`` 由适配层写入（迁移计划 §5.2 第 7 项），调用方不能经 payload 声明；
+    缺省 ``INTERNAL`` 对应进程内装配。
+    """
+    if limiter is not None and not limiter.allow(credentials.peer_address):
+        _record_denial(audit, authenticator, credentials, "rate_limit")
+        raise RateLimitedError(_RATE_LIMITED)
+
+    guard_acquired = False
+    if workload_guard is not None:
+        if not workload_guard.acquire():
+            _record_denial(audit, authenticator, credentials, "workload_budget")
+            raise RateLimitedError(_RATE_LIMITED)
+        guard_acquired = True
+
+    try:
+        ctx = authenticator.authenticate(credentials)
+    except AuthenticationError:
+        _record_denial(audit, authenticator, credentials, "authenticate")
+        raise
+    finally:
+        if guard_acquired:
+            workload_guard.release()
+
+    security = new_request_context(
+        ctx,
+        surface=surface if surface is not None else Surface.INTERNAL,
+        peer=_normalized_peer(credentials),
+        # attributes 留空：本层没有可写入的系统属性，而业务 payload 一律不得注入
+        # （迁移计划 §5.2 第 7 项）。将来要加（如可信代理链、mTLS 主体）只能由
+        # 服务端组件在此处写。
+    )
+
+    token = set_current(ctx)
+    try:
+        yield security
+    finally:
+        reset_current(token)
+
+
+def _normalized_peer(credentials) -> str:
+    """规范化连接来源：只采信传输层对端地址。
+
+    刻意**不读** ``X-Forwarded-For`` / ``X-Real-IP``：没有可信代理白名单时采信这类
+    header，等于让调用方自述来源——限流分桶、审计溯源和将来基于 peer 的策略会同时
+    被绕过。要支持反向代理部署，得先有「哪些前置跳是可信的」这项配置，那是独立设计。
+    """
+    return str(credentials.peer_address or "").strip()
+
+
+def _record_denial(audit, authenticator, credentials, action) -> None:
+    """入口拒绝落一条审计（security.md §7.2）：``action`` 区分限流与认证失败。
+
+    每次拒绝都记，无阈值聚合——限流器的计数器目前只用于准入判断，不对外暴露
+    统计；要做「同一 peer 连续失败 N 次告警」还需要一个独立的失败计数维度
+    （限流桶按请求数计，不区分成功与失败），那是可观测性设计，不在本期。
+
+    ``actor`` 是空 ``Scope()``——身份未知，**不可用调用方声明的任何值填充**。
+    ``detail`` 里不放 api_key、不放 key 前缀（§7.5 PII 脱敏），也不放桶余量
+    （那能用来反推限流参数）。
+
+    暂不记录认证失败的细分原因（``missing_credentials`` / ``unknown_principal`` /
+    ``bad_gateway_key``）：三个 authenticator 都刻意只抛同一个笼统消息，要拿到
+    细分原因得在 authenticator 侧另开一条只进审计的通道。那是独立设计，
+    不顺手塞进本期。
+    """
+    if audit is None:
+        return
+    try:
+        mode = authenticator.mode()
+        audit.record(
+            AuditEvent(
+                actor=Scope(),
+                action=action,
+                decision="deny",
+                layer="security",
+                detail={
+                    "mode": str(getattr(mode, "value", mode)),
+                    "peer": credentials.peer_address,
+                },
+            )
+        )
+    except Exception:  # pragma: no cover - 审计后端故障不该把 401/429 变成 500
+        pass