diff --git a/deploy/helm/agentkit/templates/deployment.yaml b/deploy/helm/agentkit/templates/deployment.yaml index d87e3ec..d61bbd8 100644 --- a/deploy/helm/agentkit/templates/deployment.yaml +++ b/deploy/helm/agentkit/templates/deployment.yaml @@ -92,11 +92,10 @@ spec: secretKeyRef: name: {{ include "agentkit.secretName" . }} key: {{ .Values.secrets.redisPassword.key }} - - name: POSTGRES_PASSWORD - valueFrom: - secretKeyRef: - name: {{ include "agentkit.secretName" . }} - key: {{ .Values.secrets.postgresPassword.key }} + # DEPLOY-4: 移除 POSTGRES_PASSWORD env — 应用通过 DATABASE_URL（Slot 3） + # 消费 PG 密码（DSN 内嵌），POSTGRES_PASSWORD 仅用于 PostgreSQL StatefulSet + # 初始化（由独立 PG chart / StatefulSet 引用 secrets.postgresPassword）。 + # 此处注入是死 env，误导运维以为应用读 POSTGRES_PASSWORD。 # --- 非密配置（OTel endpoint、service name）--- {{- if .Values.otel.enabled }} - name: OTEL_EXPORTER_OTLP_ENDPOINT diff --git a/deploy/helm/agentkit/values.yaml b/deploy/helm/agentkit/values.yaml index b473fad..4caed7a 100644 --- a/deploy/helm/agentkit/values.yaml +++ b/deploy/helm/agentkit/values.yaml @@ -194,19 +194,19 @@ secrets: # 9) OTel exporter token — OTLP collector 鉴权 token otelExporterToken: key: otel-exporter-token - description: "OTLP exporter 鉴权 token（collector 启用 bearer auth 时）" + description: "OTLP exporter 鉴权 token（DEPLOY-5: 值格式必须为 'Authorization=Bearer '，符合 OTEL_EXPORTER_OTLP_HEADERS key=value 规范）" rotation: 90d # 10) Redis 密码 — Redis requirepass（基础设施层） redisPassword: key: redis-password - description: "Redis requirepass（基础设施层）" + description: "Redis requirepass（基础设施层，env REDIS_PASSWORD 注入应用）" rotation: 90d - # 11) PostgreSQL 密码 — POSTGRES_PASSWORD（基础设施层） + # 11) PostgreSQL 密码 — POSTGRES_PASSWORD（基础设施层，PG StatefulSet 初始化用） postgresPassword: key: postgres-password - description: "PostgreSQL POSTGRES_PASSWORD（基础设施层）" + description: "PostgreSQL POSTGRES_PASSWORD（DEPLOY-4: 仅 PG StatefulSet 初始化消费，应用通过 DATABASE_URL Slot 3 内嵌密码连接，不读此 env）" rotation: 90d # --------------------------------------------------------------------------- diff --git a/deploy/offline/scripts/install.sh b/deploy/offline/scripts/install.sh index c99ebd2..6b8793b 100755 --- a/deploy/offline/scripts/install.sh +++ b/deploy/offline/scripts/install.sh @@ -55,7 +55,11 @@ helm upgrade --install "${RELEASE}" "${CHART_TGZ}" \ --set image.pullPolicy=Never # 离线本地镜像，Never 强制使用已加载的本地镜像 echo "==> 等待 rollout" -kubectl rollout status deployment/"${RELEASE}-agentkit" -n "${NAMESPACE}" --timeout=300s || \ +# DEPLOY-1: 用标签选择器定位 Deployment — fullname 在 release 名包含 chart 名时 +# 仅返回 release 名（如 release=agentkit → Deployment=agentkit），其他情况返回 +# release-chart（如 release=myprod → myprod-agentkit）。硬编码 "${RELEASE}-agentkit" +# 在默认 release=agentkit 时拼出不存在的 agentkit-agentkit。改用 instance 标签。 +kubectl rollout status deployment -l app.kubernetes.io/instance="${RELEASE}" -n "${NAMESPACE}" --timeout=300s || \ echo "WARN: rollout 未在 300s 内完成，请检查 pod 状态：kubectl get pods -n ${NAMESPACE}" echo "==> 完成。验证：" diff --git a/docs/deployment/k8s-install.md b/docs/deployment/k8s-install.md index 2ef0b38..8714888 100644 --- a/docs/deployment/k8s-install.md +++ b/docs/deployment/k8s-install.md @@ -163,12 +163,14 @@ kubectl get pods -n agentkit kubectl get svc,ingress -n agentkit # 3) 健康检查 -kubectl exec -n agentkit deploy/agentkit-agentkit -- \ +# DEPLOY-1: 默认 release=agentkit 时 Deployment 名为 agentkit（fullname helper +# 在 release 名包含 chart 名时去重）。若改了 release 名，按 helm status 输出替换。 +kubectl exec -n agentkit deploy/agentkit -- \ curl -s http://localhost:8001/api/v1/health # 期望：{"status":"ok"} # 4) Secret 已注入 -kubectl exec -n agentkit deploy/agentkit-agentkit -- env | grep -E 'JWT|API_KEY|DATABASE' | sed 's/=.*/=/' +kubectl exec -n agentkit deploy/agentkit -- env | grep -E 'JWT|API_KEY|DATABASE' | sed 's/=.*/=/' # 期望：所有 env 都有值（不为空） # 5) helm 状态 @@ -181,7 +183,7 @@ helm status agentkit -n agentkit ```bash # 查看日志 -kubectl logs -n agentkit deploy/agentkit-agentkit --tail=50 +kubectl logs -n agentkit deploy/agentkit --tail=50 # 常见原因： # - Secret 未注入：检查 SealedSecret 是否已解封（kubectl get sealedsecret -n agentkit） diff --git a/docs/deployment/key-rotation-runbook.md b/docs/deployment/key-rotation-runbook.md index 9caf123..63df903 100644 --- a/docs/deployment/key-rotation-runbook.md +++ b/docs/deployment/key-rotation-runbook.md @@ -34,10 +34,11 @@ kubectl create secret generic agentkit-secrets-new --dry-run=client -o yaml \ # 3) 应用并等待同步 kubectl apply -f agentkit-sealedsecret.yaml -n agentkit -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +# DEPLOY-1: 用 instance 标签定位 Deployment — 避免硬编码 fullname 拼错。 +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 4) 验证 -kubectl exec -n agentkit deploy/agentkit-agentkit -- \ +kubectl exec -n agentkit deploy/agentkit -- \ python -c "import os; print('jwt key len:', len(os.environ['JWT_SIGNING_KEY']))" # 5) 用户重新登录（旧 token 失效） @@ -62,7 +63,7 @@ NEW_KEY=$(openssl rand -hex 32) # 3) 通知所有使用 API Key 的客户端更新（agentkit pair 重新生成） # 4) 应用并 restart -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 5) 验证：用新 key 调用 /api/v1/auth/me 或 SCIM 端点 ``` @@ -87,7 +88,7 @@ NEW_URL="postgresql+asyncpg://agentkit:new-pg-pwd@postgres:5432/agentkit" # 3) 更新密钥后端（key=database-url） # 4) restart pod（连接池重建） -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 5) 验证：触发一次 DB 操作（如 GET /api/v1/memory） @@ -117,7 +118,7 @@ NEW_CERTS='{"feishu":"-----BEGIN CERTIFICATE-----\n新证书\n-----END CERTIFICA # 3) 更新密钥后端（key=idp-signing-certs） # 4) restart pod（证书热轮换不重启，但 Helm 部署需 restart 加载新 Secret） -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 5) 验证：通过该 IdP 发起一次 SSO 登录 ``` @@ -141,7 +142,7 @@ NEW_KEYS='{"dashscope":"sk-new1","deepseek":"sk-new2","openai":"sk-old"}' # 3) 更新密钥后端（key=third-party-api-keys） # 4) restart pod -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 5) 验证：调用对应 provider 的模型（如 agentkit chat 发一条测试消息） @@ -196,7 +197,7 @@ NEW_PEM="$(cat client.crt client.key ca.crt)" # 3) 更新密钥后端 # 4) restart pod -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 5) 验证：触发一次 mTLS 调用（如 KMS 签名操作） ``` @@ -218,7 +219,7 @@ kubectl rollout restart deployment/agentkit-agentkit -n agentkit NEW_PIN='' # 3) restart pod -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 4) 验证：触发一次 HSM 操作（如 PKCS#11 签名） ``` @@ -238,11 +239,13 @@ kubectl rollout restart deployment/agentkit-agentkit -n agentkit ```bash # 1) 在 OTel collector / 可观测性平台生成新 token -# 2) 更新密钥后端（key=otel-exporter-token，格式 'Authorization: Bearer '） -NEW_TOKEN='Bearer new-otel-token' +# 2) 更新密钥后端（key=otel-exporter-token） +# DEPLOY-5: OTEL_EXPORTER_OTLP_HEADERS 期望 'key=value' 格式（OTel SDK 规范）， +# 不是裸 'Bearer '。secret 值必须形如 'Authorization=Bearer '。 +NEW_TOKEN='Authorization=Bearer new-otel-token' # 3) restart pod -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 4) 验证：在 collector 端确认收到新 trace/metrics ``` @@ -270,10 +273,10 @@ psql -U postgres -c "ALTER USER agentkit WITH PASSWORD 'new-pg-pwd';" # 需同步更新 requirepass 并 restart Redis） # 5) restart agentkit pod -kubectl rollout restart deployment/agentkit-agentkit -n agentkit +kubectl rollout restart deployment -l app.kubernetes.io/instance=agentkit -n agentkit # 6) 验证 -kubectl exec -n agentkit deploy/agentkit-agentkit -- \ +kubectl exec -n agentkit deploy/agentkit -- \ python -c "import redis,os; redis.from_url(os.environ['REDIS_URL'],password=os.environ['REDIS_PASSWORD']).ping()" ``` diff --git a/docs/deployment/secret-inventory.md b/docs/deployment/secret-inventory.md index 371e752..5a41688 100644 --- a/docs/deployment/secret-inventory.md +++ b/docs/deployment/secret-inventory.md @@ -19,9 +19,9 @@ | 6 | TLS 服务器证书 | `tls-server-cert` | （Ingress TLS Secret） | Ingress TLS 终止证书 + 私钥 | 证书签发机构 / cert-manager | 365 天 | 平台运维 | | 7 | mTLS 客户端证书 | `mtls-client-cert` | `MTLS_CLIENT_CERT` | 与下游服务（KMS/HSM、内部 API）双向 TLS 客户端证书 | 内部 PKI | 180 天 | 安全团队 | | 8 | KMS/HSM PKCS#11 PIN | `kms-hsm-pin` | `KMS_HSM_PIN` | 硬件安全模块 PKCS#11 访问 PIN | HSM 管理员 | 365 天 | HSM 运维 | -| 9 | OTel exporter token | `otel-exporter-token` | `OTEL_EXPORTER_OTLP_HEADERS` | OTLP collector 鉴权 bearer token | 可观测性平台 | 90 天 | 可观测性团队 | -| 10 | Redis 密码 | `redis-password` | `REDIS_PASSWORD` | Redis requirepass（基础设施层） | DBA / 缓存运维 | 90 天 | DBA | -| 11 | PostgreSQL 密码 | `postgres-password` | `POSTGRES_PASSWORD` | PostgreSQL POSTGRES_PASSWORD（基础设施层） | DBA | 90 天 | DBA | +| 9 | OTel exporter token | `otel-exporter-token` | `OTEL_EXPORTER_OTLP_HEADERS` | OTLP collector 鉴权（DEPLOY-5: 值格式 `Authorization=Bearer `，非裸 Bearer） | 可观测性平台 | 90 天 | 可观测性团队 | +| 10 | Redis 密码 | `redis-password` | `REDIS_PASSWORD` | Redis requirepass（基础设施层，应用通过 env 消费） | DBA / 缓存运维 | 90 天 | DBA | +| 11 | PostgreSQL 密码 | `postgres-password` | `POSTGRES_PASSWORD`（仅 PG StatefulSet） | PostgreSQL POSTGRES_PASSWORD（DEPLOY-4: 应用不读此 env，通过 DATABASE_URL Slot 3 内嵌密码连接） | DBA | 90 天 | DBA | ## 风险等级 diff --git a/src/agentkit/bus/redis_bus.py b/src/agentkit/bus/redis_bus.py index 4017f25..36432f2 100644 --- a/src/agentkit/bus/redis_bus.py +++ b/src/agentkit/bus/redis_bus.py @@ -9,6 +9,7 @@ from __future__ import annotations import asyncio import json import logging +import os from typing import Callable, Awaitable from agentkit.bus.message import AgentMessage @@ -41,7 +42,13 @@ class RedisMessageBus: """获取 Redis 连接（懒初始化）。""" if self._redis is None: import redis.asyncio as aioredis - self._redis = aioredis.from_url(self._redis_url, decode_responses=True) + + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 + self._redis = aioredis.from_url( + self._redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) return self._redis def _stream_key(self, agent_name: str) -> str: @@ -101,7 +108,10 @@ class RedisMessageBus: # Create consumer group if not exists try: await redis.xgroup_create( - stream_key, self._consumer_group, id="0", mkstream=True, + stream_key, + self._consumer_group, + id="0", + mkstream=True, ) except asyncio.CancelledError: raise @@ -141,7 +151,11 @@ class RedisMessageBus: logger.warning(f"Failed to process message {msg_id}: {e}") # Move to dead letter after max retries await self._handle_failed_message( - redis, stream_key, msg_id, fields, agent_name, + redis, + stream_key, + msg_id, + fields, + agent_name, ) except asyncio.CancelledError: break @@ -194,9 +208,7 @@ class RedisMessageBus: await self.publish(message) return await asyncio.wait_for(future, timeout=timeout) except asyncio.TimeoutError: - raise TimeoutError( - f"Request {message.correlation_id} timed out after {timeout}s" - ) + raise TimeoutError(f"Request {message.correlation_id} timed out after {timeout}s") finally: self._pending_requests.pop(message.correlation_id, None) @@ -256,6 +268,7 @@ def create_message_bus( if backend == "redis": try: import redis.asyncio as aioredis # noqa: F401 + bus = RedisMessageBus( redis_url=redis_url, consumer_group=consumer_group, @@ -267,8 +280,7 @@ def create_message_bus( raise except Exception as exc: # noqa: BLE001 — factory fallback to InMemoryMessageBus; must catch import/init errors broadly logger.warning( - f"Failed to initialise RedisMessageBus ({exc}), " - f"falling back to InMemoryMessageBus" + f"Failed to initialise RedisMessageBus ({exc}), falling back to InMemoryMessageBus" ) bus = InMemoryMessageBus() diff --git a/src/agentkit/core/base.py b/src/agentkit/core/base.py index 75c6629..51036c3 100644 --- a/src/agentkit/core/base.py +++ b/src/agentkit/core/base.py @@ -10,6 +10,7 @@ import asyncio import json import logging +import os import time from abc import ABC, abstractmethod from collections.abc import AsyncGenerator @@ -266,7 +267,12 @@ class BaseAgent(ABC): if redis_url: try: - self._redis = aioredis.from_url(redis_url, decode_responses=True) + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 + self._redis = aioredis.from_url( + redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) await self._redis.ping() logger.info(f"Agent '{self.name}' connected to Redis") except (ConnectionError, OSError, asyncio.TimeoutError, ValueError) as e: diff --git a/src/agentkit/core/config_driven.py b/src/agentkit/core/config_driven.py index 33a762d..0d461aa 100644 --- a/src/agentkit/core/config_driven.py +++ b/src/agentkit/core/config_driven.py @@ -394,7 +394,12 @@ class ConfigDrivenAgent(BaseAgent, EvolutionMixin): import redis.asyncio as aioredis redis_url = config.memory["working"].get("redis_url", "redis://localhost:6379") - redis_client = aioredis.from_url(redis_url, decode_responses=True) + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 + redis_client = aioredis.from_url( + redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) working = WorkingMemory(redis=redis_client) if config.memory.get("episodic", {}).get("enabled"): diff --git a/src/agentkit/core/handoff_transport.py b/src/agentkit/core/handoff_transport.py index 5a2bf83..af093f0 100644 --- a/src/agentkit/core/handoff_transport.py +++ b/src/agentkit/core/handoff_transport.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio import json import logging +import os from typing import AsyncIterator, Awaitable, Callable import redis.asyncio as aioredis @@ -144,7 +145,12 @@ class RedisHandoffTransport: """确保 Redis 连接已建立（懒初始化）""" if self._redis is None: try: - self._redis = aioredis.from_url(self._redis_url, decode_responses=True) + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 + self._redis = aioredis.from_url( + self._redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) await self._redis.ping() logger.info("RedisHandoffTransport connected to Redis") except Exception: @@ -199,9 +205,7 @@ class RedisHandoffTransport: # 启动后台监听任务 if channel not in self._listen_tasks: - self._listen_tasks[channel] = asyncio.create_task( - self._background_listen(channel) - ) + self._listen_tasks[channel] = asyncio.create_task(self._background_listen(channel)) async def _background_listen(self, channel: str) -> None: """后台监听频道消息并调用处理器""" diff --git a/src/agentkit/llm/cache.py b/src/agentkit/llm/cache.py index 0919920..46e3508 100644 --- a/src/agentkit/llm/cache.py +++ b/src/agentkit/llm/cache.py @@ -11,6 +11,7 @@ Design doc: docs/plans/2026-06-14-002-u1-llm-cache-architecture.md import json import logging +import os import time from collections import OrderedDict from dataclasses import dataclass, field @@ -386,7 +387,12 @@ class RedisLLMCache: if self._redis is None: import redis.asyncio as aioredis - self._redis = aioredis.from_url(self._redis_url, decode_responses=True) + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 + self._redis = aioredis.from_url( + self._redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) return self._redis async def aclose(self) -> None: diff --git a/src/agentkit/llm/gateway.py b/src/agentkit/llm/gateway.py index dbc9cf5..8e2299a 100644 --- a/src/agentkit/llm/gateway.py +++ b/src/agentkit/llm/gateway.py @@ -2,6 +2,7 @@ import asyncio import logging +import os import time from datetime import datetime, timezone from pathlib import Path @@ -84,6 +85,12 @@ class LLMGateway: self._usage_tracker = UsageTracker(store=usage_store) if usage_store else UsageTracker() self._config = config or LLMConfig() + # PERF-1: PII 脱敏接入生产路径（env 驱动，默认关闭以保持向后兼容）。 + # AGENTKIT_PII_FILTER_ENABLED=true 启用；filter_messages_pii fail-closed。 + self._pii_filter_enabled = ( + os.environ.get("AGENTKIT_PII_FILTER_ENABLED", "false").lower() == "true" + ) + # Cache (U17 — LiteLLM 缓存管理器，opt-in，默认禁用) self._cache_manager: "LitellmCacheManager | None" = None if self._config.cache and self._config.cache.enabled: @@ -97,6 +104,22 @@ class LLMGateway: f"similarity_threshold={litellm_config.similarity_threshold})" ) + if self._pii_filter_enabled: + logger.info("PII filter enabled (AGENTKIT_PII_FILTER_ENABLED=true, fail-closed)") + + def _apply_pii_filter(self, messages: list[dict[str, str]]) -> list[dict[str, str]]: + """PERF-1: 对 messages 执行 PII 脱敏。 + + fail-closed：脱敏失败抛 PIIFilterError，由调用方转 HTTP 422。 + 禁用时原样返回（零开销 — 仅一次 env 读 + bool 判断）。 + """ + if not self._pii_filter_enabled: + return messages + from agentkit.llm.pii_filter import filter_messages_pii + + redacted, _matches = filter_messages_pii(messages, fail_closed=True) + return redacted + def register_provider(self, name: str, provider: LLMProvider) -> None: """注册 Provider""" self._providers[name] = provider @@ -129,6 +152,9 @@ class LLMGateway: if not self._providers: raise LLMProviderError("", "No provider registered") + # PERF-1: PII 脱敏（fail-closed）— 在 provider 调用前替换 messages。 + messages = self._apply_pii_filter(messages) + # ── Quota enforcement ── # Only enforce when department_ids + db_path are provided # (other call sites pass None — no quota check). @@ -349,6 +375,9 @@ class LLMGateway: if not self._providers: raise LLMProviderError("", "No provider registered") + # PERF-1: PII 脱敏（fail-closed）— 在 provider 调用前替换 messages。 + messages = self._apply_pii_filter(messages) + # ── Quota enforcement ── if department_ids and db_path: await self._enforce_quota(db_path, department_ids, resolved_model) diff --git a/src/agentkit/llm/providers/usage_store.py b/src/agentkit/llm/providers/usage_store.py index 3bd634d..bd0ee0d 100644 --- a/src/agentkit/llm/providers/usage_store.py +++ b/src/agentkit/llm/providers/usage_store.py @@ -16,6 +16,7 @@ Legacy v1 keys (still readable for backward compat): import asyncio import json import logging +import os from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import Protocol, runtime_checkable @@ -304,7 +305,12 @@ class RedisUsageStore: if self._redis is None: import redis.asyncio as aioredis - self._redis = aioredis.from_url(self._redis_url, decode_responses=True) + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 + self._redis = aioredis.from_url( + self._redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) return self._redis def _get_sync_redis(self): @@ -312,7 +318,12 @@ class RedisUsageStore: if self._sync_redis is None: import redis as sync_redis - self._sync_redis = sync_redis.from_url(self._redis_url, decode_responses=True) + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 + self._sync_redis = sync_redis.from_url( + self._redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) return self._sync_redis async def aclose(self) -> None: diff --git a/src/agentkit/orchestrator/pipeline_state.py b/src/agentkit/orchestrator/pipeline_state.py index cb2a6ce..8907bfc 100644 --- a/src/agentkit/orchestrator/pipeline_state.py +++ b/src/agentkit/orchestrator/pipeline_state.py @@ -11,6 +11,7 @@ from __future__ import annotations import json import logging +import os import uuid from datetime import datetime, timezone from typing import Callable, Coroutine @@ -176,9 +177,11 @@ class PipelineStateRedis: if self._redis is None: import redis.asyncio as aioredis + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 self._redis = aioredis.from_url( self._redis_url, decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, ) return self._redis diff --git a/src/agentkit/quality/cascade_state_store.py b/src/agentkit/quality/cascade_state_store.py index 04199ea..6a8b894 100644 --- a/src/agentkit/quality/cascade_state_store.py +++ b/src/agentkit/quality/cascade_state_store.py @@ -10,6 +10,7 @@ Key schema (Redis): """ import logging +import os import time from typing import Protocol, runtime_checkable @@ -140,8 +141,12 @@ class RedisCascadeStateStore: """Get or create a persistent sync Redis client (connection pool backed).""" if self._sync_redis is None: import redis as sync_redis + + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 self._sync_redis = sync_redis.from_url( - self._redis_url, decode_responses=True + self._redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, ) return self._sync_redis @@ -163,7 +168,15 @@ class RedisCascadeStateStore: pipe.expire(key, self._session_ttl) results = pipe.execute() return results[0] - except (ImportError, OSError, _RedisError, ValueError, KeyError, RuntimeError, TypeError) as e: + except ( + ImportError, + OSError, + _RedisError, + ValueError, + KeyError, + RuntimeError, + TypeError, + ) as e: logger.warning(f"Redis cascade increment failed: {e}") self._degrade_to_fallback() if self._fallback is not None: @@ -177,7 +190,15 @@ class RedisCascadeStateStore: r = self._get_sync_redis() val = r.get(f"{self.INTER_PREFIX}{session_id}") return int(val) if val is not None else 0 - except (ImportError, OSError, _RedisError, ValueError, KeyError, RuntimeError, TypeError) as e: + except ( + ImportError, + OSError, + _RedisError, + ValueError, + KeyError, + RuntimeError, + TypeError, + ) as e: logger.warning(f"Redis cascade get failed: {e}") if self._fallback is not None: return self._fallback.get_interaction(session_id) @@ -194,7 +215,15 @@ class RedisCascadeStateStore: pipe.set(key, depth) pipe.expire(key, self._session_ttl) pipe.execute() - except (ImportError, OSError, _RedisError, ValueError, KeyError, RuntimeError, TypeError) as e: + except ( + ImportError, + OSError, + _RedisError, + ValueError, + KeyError, + RuntimeError, + TypeError, + ) as e: logger.warning(f"Redis cascade set_depth failed: {e}") self._degrade_to_fallback() if self._fallback is not None: @@ -207,7 +236,15 @@ class RedisCascadeStateStore: r = self._get_sync_redis() val = r.get(f"{self.DEPTH_PREFIX}{session_id}") return int(val) if val is not None else 0 - except (ImportError, OSError, _RedisError, ValueError, KeyError, RuntimeError, TypeError) as e: + except ( + ImportError, + OSError, + _RedisError, + ValueError, + KeyError, + RuntimeError, + TypeError, + ) as e: logger.warning(f"Redis cascade get_depth failed: {e}") if self._fallback is not None: return self._fallback.get_depth(session_id) @@ -223,7 +260,15 @@ class RedisCascadeStateStore: pipe.delete(f"{self.INTER_PREFIX}{session_id}") pipe.delete(f"{self.DEPTH_PREFIX}{session_id}") pipe.execute() - except (ImportError, OSError, _RedisError, ValueError, KeyError, RuntimeError, TypeError) as e: + except ( + ImportError, + OSError, + _RedisError, + ValueError, + KeyError, + RuntimeError, + TypeError, + ) as e: logger.warning(f"Redis cascade reset failed: {e}") self._degrade_to_fallback() if self._fallback is not None: @@ -250,6 +295,7 @@ def create_cascade_state_store( if backend in ("auto", "redis"): try: import redis # noqa: F401 + return RedisCascadeStateStore(redis_url=redis_url, session_ttl=session_ttl) except ImportError: logger.warning("redis package not available, falling back to in-memory cascade store") diff --git a/src/agentkit/server/app.py b/src/agentkit/server/app.py index b6b7ef0..6efdbbb 100644 --- a/src/agentkit/server/app.py +++ b/src/agentkit/server/app.py @@ -1258,11 +1258,13 @@ def create_app( "redis_url", "redis://localhost:6379" ) # U5c: socket_timeout 防止单点 Redis 故障时网关请求挂死。 + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 redis_client = aioredis.from_url( redis_url, decode_responses=True, socket_timeout=5.0, socket_connect_timeout=5.0, + password=os.environ.get("REDIS_PASSWORD") or None, ) working = WorkingMemory(redis=redis_client) app.state.working_redis_client = redis_client @@ -1396,9 +1398,14 @@ def create_app( app.include_router(auth_routes.router, prefix="/api/v1") app.include_router(auth_routes.admin_router, prefix="/api/v1") # U4 SSO 集成 — SCIM 2.0 最小子集（自带 bearer token 校验） - from agentkit.server.auth.scim.router import router as scim_router + from agentkit.server.auth.scim.router import ( + register_scim_exception_handlers, + router as scim_router, + ) app.include_router(scim_router, prefix="/api/v1") + # API-2: 注册 SCIM 异常处理器（scope 到 /scim/v2，非 SCIM 路由走默认格式） + register_scim_exception_handlers(app) app.include_router(admin_routes_module.admin_router, prefix="/api/v1") app.include_router(documents.router, prefix="/api/v1") app.include_router(calendar_routes.router, prefix="/api/v1") diff --git a/src/agentkit/server/auth/scim/models.py b/src/agentkit/server/auth/scim/models.py index fe64298..55b8f59 100644 --- a/src/agentkit/server/auth/scim/models.py +++ b/src/agentkit/server/auth/scim/models.py @@ -1,13 +1,11 @@ """SCIM 2.0 用户 schema (Pydantic v2) — U4 最小子集。 仅实现 RFC 7643 / 7644 中创建 / 禁用 / 查询所需的字段： -``id``, ``userName``, ``active``, ``emails``, ``displayName``。 +``id``, ``userName``, ``active``, ``emails``, ``displayName``, ``externalId``。 """ from __future__ import annotations -from typing import Any - from pydantic import BaseModel, ConfigDict, EmailStr, Field @@ -22,7 +20,11 @@ class SCIMEmail(BaseModel): class SCIMUser(BaseModel): - """SCIM 2.0 User 资源（最小子集）。""" + """SCIM 2.0 User 资源（最小子集）。 + + STAND-1: 移除 ``Any`` — ``externalId`` 是 RFC 7643 标准 attribute， + 用于 SCIM 预配的用户与 SSO IDP subject 关联（API-5）。 + """ model_config = ConfigDict(extra="allow") @@ -30,17 +32,22 @@ class SCIMUser(BaseModel): userName: str active: bool = True displayName: str | None = None + externalId: str | None = None emails: list[SCIMEmail] = Field(default_factory=list) class SCIMPatchOp(BaseModel): - """SCIM 2.0 PATCH 操作（最小子集，仅支持 replace）。""" + """SCIM 2.0 PATCH 操作（最小子集，仅支持 replace）。 + + STAND-1: ``value`` 用 ``object`` 替代 ``Any``（Pydantic v2 中 ``object`` 等价 + 于旧 ``Any`` 但更明确表达"任意 JSON 值"的语义）。 + """ model_config = ConfigDict(extra="allow") op: str = "replace" path: str | None = None - value: Any = None + value: object = None class SCIMPatchRequest(BaseModel): @@ -52,7 +59,10 @@ class SCIMPatchRequest(BaseModel): class SCIMListResponse(BaseModel): - """SCIM 2.0 列表响应。""" + """SCIM 2.0 列表响应。 + + STAND-1: ``Resources`` 用 ``list[dict[str, object]]`` 替代 ``list[dict[str, Any]]``。 + """ model_config = ConfigDict(extra="allow") @@ -62,7 +72,7 @@ class SCIMListResponse(BaseModel): totalResults: int = 0 startIndex: int = 1 itemsPerPage: int = 0 - Resources: list[dict[str, Any]] = Field(default_factory=list) + Resources: list[dict[str, object]] = Field(default_factory=list) class SCIMErrorDetail(BaseModel): diff --git a/src/agentkit/server/auth/scim/router.py b/src/agentkit/server/auth/scim/router.py index d53057c..b960442 100644 --- a/src/agentkit/server/auth/scim/router.py +++ b/src/agentkit/server/auth/scim/router.py @@ -1,12 +1,13 @@ """SCIM 2.0 路由 (U4 最小子集) — handler 内联，不拆 handlers.py。 端点： -- POST /scim/v2/Users — 创建用户 +- POST /scim/v2/Users — 创建用户（同步写 user_idp_links 供 SSO 联动） - PATCH /scim/v2/Users/{id} — 禁用 (active=false) / 更新 - GET /scim/v2/Users — 列表 / 过滤 Bearer token 认证（独立于 JWT — ``auth.scim_token`` 配置项）。 SCIM push 失败触发实时告警（日志 error + OTel span event）。 +RFC 7643/7644 合规：错误响应用 SCIMErrorResponse 格式 + Location 头 + totalResults 全量总数。 """ from __future__ import annotations @@ -17,12 +18,13 @@ import os import secrets import uuid from pathlib import Path -from typing import Any import aiosqlite from fastapi import APIRouter, Depends, HTTPException, Query, Request, status -from pydantic import BaseModel, ConfigDict +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from ...auth.audit import SOURCE_SCIM, get_audit_service from ...auth.models import ( init_auth_db, resolve_auth_db_path as _resolve_default_db_path, @@ -37,6 +39,7 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/scim/v2", tags=["scim"]) _SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User" +_SCIM_ERROR_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:Error" def _resolve_db_path(request: Request) -> Path: @@ -98,30 +101,108 @@ async def _ensure_db(request: Request) -> Path: return db_path -def _user_to_scim(row: dict[str, object]) -> dict[str, Any]: - """本地 user dict → SCIM 2.0 User JSON。""" +# --------------------------------------------------------------------------- +# RFC 7644 3.12 — SCIM 错误响应格式 +# --------------------------------------------------------------------------- + + +def _scim_error_response(status_code: int, detail: str) -> JSONResponse: + """API-2: 构造 RFC 7644 3.12 规定的 SCIM 错误响应格式。""" + return JSONResponse( + status_code=status_code, + content={ + "schemas": [_SCIM_ERROR_SCHEMA], + "status": str(status_code), + "detail": detail, + }, + ) + + +# API-2: APIRouter 在旧版 FastAPI 无 exception_handler 方法，改为导出注册函数 +# 由 app.py 在挂载 SCIM router 后调用，scope 到 /scim/v2 路径。 + + +def register_scim_exception_handlers(app) -> None: + """注册 SCIM 异常处理器 — 仅对 /scim/v2 路径生效，不影响其他路由。 + + API-2: APIRouter 在旧版 FastAPI 无 exception_handler 方法，改为在 app 上注册 + 并通过 request.url.path scope 到 SCIM 路径。非 SCIM 路由走 FastAPI 默认格式。 + """ + + @app.exception_handler(HTTPException) + async def _scim_http_handler(request: Request, exc: HTTPException): + path = request.url.path + is_scim = path.startswith("/scim/v2") or path.startswith("/api/v1/scim/v2") + if not is_scim: + # 非 SCIM 路由 — 复制 FastAPI 默认 HTTPException handler 行为（含 headers） + headers = getattr(exc, "headers", None) + return JSONResponse( + status_code=exc.status_code, + content={"detail": exc.detail}, + headers=headers, + ) + return _scim_error_response(exc.status_code, str(exc.detail)) + + @app.exception_handler(RequestValidationError) + async def _scim_validation_handler(request: Request, exc: RequestValidationError): + path = request.url.path + is_scim = path.startswith("/scim/v2") or path.startswith("/api/v1/scim/v2") + if not is_scim: + # 非 SCIM 路由 — 复制 FastAPI 默认 RequestValidationError handler 行为 + from fastapi.encoders import jsonable_encoder + + return JSONResponse( + status_code=422, + content={"detail": jsonable_encoder(exc.errors())}, + ) + return _scim_error_response(400, f"请求体校验失败: {exc}") + + +# --------------------------------------------------------------------------- +# SCIM User 资源转换 — STAND-1: 用 dict[str, object] 替代 Any +# --------------------------------------------------------------------------- + + +def _user_to_scim(row: dict[str, object]) -> dict[str, object]: + """本地 user dict → SCIM 2.0 User JSON。 + + API-11: displayName 改为 None（让客户端用 userName fallback）， + 因为 users 表无 name/full_name 列，username 不是 displayName 的合理来源。 + """ email = str(row.get("email", "") or "") return { "schemas": [_SCIM_USER_SCHEMA], "id": str(row["id"]), "userName": str(row["username"]), "active": bool(row.get("is_active", True)), - "displayName": str(row.get("username", "")), + "displayName": None, "emails": [{"value": email, "type": "work", "primary": True}] if email else [], } +# --------------------------------------------------------------------------- +# SCIM 端点 +# --------------------------------------------------------------------------- + + @router.post("/Users", status_code=status.HTTP_201_CREATED) async def create_user( payload: SCIMUser, request: Request, _token: str = Depends(_verify_scim_token), -) -> dict[str, Any]: - """SCIM POST /Users — 创建本地用户（随机密码，不可本地登录）。""" +) -> JSONResponse: + """SCIM POST /Users — 创建本地用户（随机密码，不可本地登录）。 + + API-5: 同时写 user_idp_links 行，让 SCIM 预配的用户能通过 SSO 登录， + 避免 OIDC JIT 创建重复账户。externalId 作为 idp_subject， + 缺省时用 userName。 + """ db_path = await _ensure_db(request) user_id = str(uuid.uuid4()) now = _now_iso() email = payload.emails[0].value if payload.emails else f"{user_id}@scim.local" + # externalId 是 RFC 7643 标准 attribute，作为 IDP subject 关联本地用户 + external_id = getattr(payload, "externalId", None) or payload.userName try: async with aiosqlite.connect(str(db_path)) as db: db.row_factory = aiosqlite.Row @@ -144,18 +225,32 @@ async def create_user( now, ), ) + # API-5: 写 user_idp_links 行，idp_name='scim'，idp_subject=externalId + # 表无 id 列（PK = idp_name + idp_subject），省去 uuid 生成。 + await db.execute( + "INSERT INTO user_idp_links " + "(idp_name, idp_subject, user_id, idp_type, created_at) " + "VALUES (?, ?, ?, ?, ?)", + ("scim", external_id, user_id, "scim", now), + ) await db.commit() cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,)) row = await cursor.fetchone() assert row is not None - logger.info("SCIM 创建用户 userName=%s id=%s", payload.userName, user_id) - return _user_to_scim(user_row_to_dict(row)) + logger.info( + "SCIM 创建用户 userName=%s id=%s externalId=%s", payload.userName, user_id, external_id + ) + # API-6: RFC 7644 3.14 强制要求 Location 头 + body = _user_to_scim(user_row_to_dict(row)) + response = JSONResponse(status_code=status.HTTP_201_CREATED, content=body) + response.headers["Location"] = f"/scim/v2/Users/{user_id}" + return response except aiosqlite.IntegrityError as exc: _alert_scim_failure("create", f"用户名或邮箱冲突: {exc}", user_id) - raise HTTPException(status_code=409, detail="userName 或 email 已存在") from exc + return _scim_error_response(409, "userName 或 email 已存在") except Exception as exc: # noqa: BLE001 _alert_scim_failure("create", str(exc), user_id) - raise HTTPException(status_code=500, detail="SCIM 创建失败") from exc + return _scim_error_response(500, "SCIM 创建失败") @router.patch("/Users/{user_id}") @@ -164,17 +259,24 @@ async def patch_user( payload: SCIMPatchRequest, request: Request, _token: str = Depends(_verify_scim_token), -) -> dict[str, Any]: - """SCIM PATCH /Users/{id} — 禁用 (active=false) 或更新字段。""" +) -> dict[str, object]: + """SCIM PATCH /Users/{id} — 禁用 (active=false) 或更新字段。 + + API-7: userName UNIQUE 冲突时返回 409（而非 500）。 + SEC-3: active true→false 触发审计记录（与 admin deprovision 对齐）。 + """ db_path = await _ensure_db(request) now = _now_iso() + # SEC-3: 收集 deprovision 审计信息，在事务提交后调用（避免 SQLite 写锁竞争） + _pending_audit: dict[str, object] | None = None try: async with aiosqlite.connect(str(db_path)) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,)) row = await cursor.fetchone() if row is None: - raise HTTPException(status_code=404, detail="用户不存在") + return _scim_error_response(404, "用户不存在") + prev_active = bool(row["is_active"]) for op in payload.Operations: if op.op.lower() == "replace": if op.path in (None, "active"): @@ -187,21 +289,41 @@ async def patch_user( "UPDATE users SET is_active = ?, updated_at = ? WHERE id = ?", (1 if active else 0, now, user_id), ) + # SEC-3: active true→false 写审计（延后到事务提交后） + if prev_active and not active: + _pending_audit = { + "target_user_id": user_id, + "target_username": str(row["username"]), + } elif op.path == "userName": - await db.execute( - "UPDATE users SET username = ?, updated_at = ? WHERE id = ?", - (str(op.value)[:64], now, user_id), - ) + try: + await db.execute( + "UPDATE users SET username = ?, updated_at = ? WHERE id = ?", + (str(op.value)[:64], now, user_id), + ) + except aiosqlite.IntegrityError: + # API-7: UNIQUE 冲突返回 409 + return _scim_error_response(409, "userName 已存在") await db.commit() cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,)) row = await cursor.fetchone() assert row is not None + # SEC-3: 事务提交后写审计（避免 SQLite "database is locked"） + if _pending_audit is not None: + await get_audit_service().record_deprovision( + actor_user_id=None, + actor_username="scim", + target_user_id=str(_pending_audit["target_user_id"]), + target_username=str(_pending_audit["target_username"]), + reason="scim_patch_active_false", + source=SOURCE_SCIM, + ) return _user_to_scim(user_row_to_dict(row)) except HTTPException: raise except Exception as exc: # noqa: BLE001 _alert_scim_failure("patch", str(exc), user_id) - raise HTTPException(status_code=500, detail="SCIM 更新失败") from exc + return _scim_error_response(500, "SCIM 更新失败") @router.get("/Users") @@ -211,44 +333,49 @@ async def list_users( startIndex: int = Query(default=1, ge=1), count: int = Query(default=100, ge=1, le=200), _token: str = Depends(_verify_scim_token), -) -> dict[str, Any]: +) -> dict[str, object]: """SCIM GET /Users — 列表 / 过滤。 - ponytail: filter 仅支持最简形式 ``userName eq "xxx"`` / ``active eq true``。 - 上限：复杂 SCIM filter 表达式需扩展为完整 AST 解析器。 + API-1: totalResults 返回全量总数（COUNT 查询），而非当前页条数。 + API-9: 不支持的 filter 语法返回 400 + invalidFilter，不静默全量返回。 + API-13: filter 解析大小写不敏感（RFC 7644 3.4.2.2）。 """ db_path = await _ensure_db(request) where = "" - params: list[Any] = [] + params: list[object] = [] if filter: - # 最简 filter 解析：userName eq "x" / active eq true - if "userName eq" in filter: - val = filter.split("userName eq", 1)[1].strip().strip('"').strip("'") + filter_lower = filter.lower() + # API-13: 大小写不敏感匹配 + if "username eq" in filter_lower: + # 从原 filter 中提取引号内的值（保留原大小写） + val = filter.split(" eq", 1)[1].strip().strip('"').strip("'") where = "WHERE username = ?" params.append(val) - elif "active eq" in filter: - val = filter.split("active eq", 1)[1].strip().strip('"').strip("'").lower() + elif "active eq" in filter_lower: + val = filter.split(" eq", 1)[1].strip().strip('"').strip("'").lower() where = "WHERE is_active = ?" params.append(1 if val in ("true", "1") else 0) - sql = f"SELECT * FROM users {where} ORDER BY created_at ASC LIMIT ? OFFSET ?" - params.extend([count, startIndex - 1]) + else: + # API-9: 不支持的 filter 返回 400 + return _scim_error_response( + 400, f"Unsupported filter syntax (scimType=invalidFilter): {filter}" + ) + # API-1: 先 COUNT 全量总数 + count_sql = f"SELECT COUNT(*) FROM users {where}" + list_sql = f"SELECT * FROM users {where} ORDER BY created_at ASC LIMIT ? OFFSET ?" + list_params = [*params, count, startIndex - 1] async with aiosqlite.connect(str(db_path)) as db: db.row_factory = aiosqlite.Row - cursor = await db.execute(sql, tuple(params)) + # totalResults + cursor = await db.execute(count_sql, tuple(params)) + total_results = (await cursor.fetchone())[0] + # 当前页 + cursor = await db.execute(list_sql, tuple(list_params)) rows = await cursor.fetchall() resources = [_user_to_scim(user_row_to_dict(r)) for r in rows] return SCIMListResponse( - totalResults=len(resources), + totalResults=total_results, startIndex=startIndex, itemsPerPage=len(resources), Resources=resources, ).model_dump() - - -class SCIMErrorResponse(BaseModel): - """SCIM 错误响应包装。""" - - model_config = ConfigDict(extra="allow") - - status: str - detail: str | None = None diff --git a/src/agentkit/server/routes/auth.py b/src/agentkit/server/routes/auth.py index 62871a4..ac77089 100644 --- a/src/agentkit/server/routes/auth.py +++ b/src/agentkit/server/routes/auth.py @@ -23,6 +23,7 @@ from __future__ import annotations import hmac import logging +import secrets from datetime import datetime, timezone from pathlib import Path from typing import Any, Literal @@ -794,8 +795,20 @@ class SSORedirectResponse(BaseModel): state: str -async def _sso_issue_token_pair(request: Request, user_dict: dict[str, object]) -> TokenResponse: - """SSO 登录成功后签发 JWT pair + 创建 session（复用 login 流程）。""" +async def _sso_issue_token_pair( + request: Request, + user_dict: dict[str, object], + *, + provider_type: str = "sso", + provider_name: str | None = None, +) -> TokenResponse: + """SSO 登录成功后签发 JWT pair + 创建 session（复用 login 流程）。 + + API-4: 一次性生成真实 refresh_token 再创建 session（避免占位符并发冲突）。 + API-8: auth_provider 携带 provider_type + provider_name（如 'oidc-keycloak'）， + 供 admin session 列表按 IDP 筛选 + 审计追溯。 + SEC-1: 与本地 login 一致地校验 is_active。 + """ # SEC-1: SSO 流程必须与本地 login（auth.py:377）一致地校验 is_active， # 否则已 deprovision 的用户可通过 IDP 回调重新拿到 JWT + session。 if not bool(user_dict.get("is_active", True)): @@ -805,15 +818,21 @@ async def _sso_issue_token_pair(request: Request, user_dict: dict[str, object]) svc: SessionService = get_session_service() info = _client_info(request) - pre_session = await svc.create( + # API-4: 先生成真实 refresh_token，避免占位符 "__pending_sso__" 在并发 SSO + # 登录时触发 refresh_token_hash UNIQUE 约束冲突。 + refresh_token = secrets.token_urlsafe(32) + # API-8: auth_provider 形如 "oidc-keycloak" / "saml-azuread" / "sso" + auth_provider = f"{provider_type}-{provider_name}" if provider_name else provider_type + + session = await svc.create( SessionCreate( user_id=str(user_dict["id"]), - refresh_token="__pending_sso__", + refresh_token=refresh_token, device_fingerprint=info["fingerprint"], device_label=info["label"], ip=info["ip"], user_agent=info["user_agent"], - auth_provider="sso", + auth_provider=auth_provider, ttl_seconds=_refresh_ttl_seconds(False), ) ) @@ -822,13 +841,9 @@ async def _sso_issue_token_pair(request: Request, user_dict: dict[str, object]) username=str(user_dict["username"]), role=str(user_dict["role"]), secret=secret, - session_id=pre_session.id, + session_id=session.id, ) async with aiosqlite.connect(str(db_path)) as db: - await db.execute( - "UPDATE auth_sessions SET refresh_token_hash = ? WHERE id = ?", - (hash_token(token_pair.refresh_token), pre_session.id), - ) await db.execute( "UPDATE users SET last_login_at = ?, updated_at = ? WHERE id = ?", (_now_iso(), _now_iso(), str(user_dict["id"])), @@ -849,7 +864,7 @@ async def _sso_issue_token_pair(request: Request, user_dict: dict[str, object]) user_dict.get("is_server_terminal_authorized", False) ), ), - session_id=pre_session.id, + session_id=session.id, ) @@ -886,7 +901,9 @@ async def oidc_callback( user_dict = await OIDCProvider().handle_callback(provider, code, state) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - return await _sso_issue_token_pair(request, user_dict) + return await _sso_issue_token_pair( + request, user_dict, provider_type="oidc", provider_name=provider + ) class SAMLACSRequest(BaseModel): @@ -915,7 +932,9 @@ async def saml_acs( ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - return await _sso_issue_token_pair(request, user_dict) + return await _sso_issue_token_pair( + request, user_dict, provider_type="saml", provider_name=provider + ) # --------------------------------------------------------------------------- @@ -1041,13 +1060,33 @@ class DeprovisionRequest(BaseModel): break_glass: bool = False # 高权限账户 break-glass 标记 -@admin_router.post("/users/{user_id}/deprovision") +class DeprovisionResponse(BaseModel): + """API-10: deprovision 响应体（让 OpenAPI schema 可文档化）。""" + + model_config = ConfigDict(extra="forbid") + + deprovisioned: bool + revoked_sessions: int + + +class RoleChangeResponse(BaseModel): + """API-10: 角色变更响应体。""" + + model_config = ConfigDict(extra="forbid") + + role_changed: bool + role_before: str + role_after: str + revoked_sessions: int + + +@admin_router.post("/users/{user_id}/deprovision", response_model=DeprovisionResponse) async def deprovision_user( user_id: str, payload: DeprovisionRequest, request: Request, admin: dict[str, Any] = Depends(_require_admin), -) -> dict[str, Any]: +) -> DeprovisionResponse: """手动 deprovisioning (R1b) — operator/admin RBAC 强制。 1. 禁用本地用户 (is_active=0) @@ -1086,7 +1125,7 @@ async def deprovision_user( reason=payload.reason, source=source, ) - return {"deprovisioned": True, "revoked_sessions": revoked} + return DeprovisionResponse(deprovisioned=True, revoked_sessions=revoked) class RoleChangeRequest(BaseModel): @@ -1098,13 +1137,13 @@ class RoleChangeRequest(BaseModel): reason: str | None = None -@admin_router.post("/users/{user_id}/role") +@admin_router.post("/users/{user_id}/role", response_model=RoleChangeResponse) async def change_user_role( user_id: str, payload: RoleChangeRequest, request: Request, admin: dict[str, Any] = Depends(_require_admin), -) -> dict[str, Any]: +) -> RoleChangeResponse: """RBAC 角色变更 (KTD-8) — admin only。 记录 actor/target/role_before/role_after/reason/source/timestamp，接入 OTel 审计。 @@ -1139,12 +1178,12 @@ async def change_user_role( # 撤销所有 session（强制重新登录刷新 role claim） svc: SessionService = get_session_service() revoked = await svc.revoke_all_for_user(user_id, reason="role_changed") - return { - "role_changed": True, - "role_before": role_before, - "role_after": payload.role, - "revoked_sessions": revoked, - } + return RoleChangeResponse( + role_changed=True, + role_before=role_before, + role_after=payload.role, + revoked_sessions=revoked, + ) # Backwards-compat /me endpoint (kept for the existing frontend) diff --git a/src/agentkit/server/task_store.py b/src/agentkit/server/task_store.py index d1c9d42..69f69c1 100644 --- a/src/agentkit/server/task_store.py +++ b/src/agentkit/server/task_store.py @@ -3,6 +3,7 @@ import asyncio import json import logging +import os from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any @@ -15,6 +16,7 @@ logger = logging.getLogger(__name__) @dataclass class TaskRecord: """Stored task record with full lifecycle data""" + task_id: str agent_name: str skill_name: str | None @@ -57,9 +59,15 @@ class TaskRecord: status=TaskStatus(data.get("status", "pending")), output_data=data.get("output_data"), error_message=data.get("error_message"), - created_at=datetime.fromisoformat(data["created_at"]) if data.get("created_at") else datetime.now(timezone.utc), - started_at=datetime.fromisoformat(data["started_at"]) if data.get("started_at") else None, - completed_at=datetime.fromisoformat(data["completed_at"]) if data.get("completed_at") else None, + created_at=datetime.fromisoformat(data["created_at"]) + if data.get("created_at") + else datetime.now(timezone.utc), + started_at=datetime.fromisoformat(data["started_at"]) + if data.get("started_at") + else None, + completed_at=datetime.fromisoformat(data["completed_at"]) + if data.get("completed_at") + else None, progress=data.get("progress", 0.0), progress_message=data.get("progress_message", ""), metadata=data.get("metadata", {}), @@ -124,14 +132,19 @@ class InMemoryTaskStore: if expired: logger.info(f"TaskStore cleaned up {len(expired)} expired records") - def create(self, task_id: str, agent_name: str, input_data: dict, skill_name: str | None = None) -> TaskRecord: + def create( + self, task_id: str, agent_name: str, input_data: dict, skill_name: str | None = None + ) -> TaskRecord: """Create a new task record""" if len(self._tasks) >= self._max_records: # Remove oldest completed task oldest = None for rec in self._tasks.values(): if rec.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED): - if oldest is None or (rec.completed_at and (oldest.completed_at is None or rec.completed_at < oldest.completed_at)): + if oldest is None or ( + rec.completed_at + and (oldest.completed_at is None or rec.completed_at < oldest.completed_at) + ): oldest = rec if oldest: del self._tasks[oldest.task_id] @@ -223,9 +236,11 @@ class RedisTaskStore: if self._redis is None: import redis.asyncio as aioredis + # DEPLOY-2: 显式传入 REDIS_PASSWORD env var（Helm 注入但 redis-py 不自动读取）。 self._redis = aioredis.from_url( self._redis_url, decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, ) return self._redis @@ -245,7 +260,9 @@ class RedisTaskStore: # ── CRUD ─────────────────────────────────────────────────── - async def create(self, task_id: str, agent_name: str, input_data: dict, skill_name: str | None = None) -> TaskRecord: + async def create( + self, task_id: str, agent_name: str, input_data: dict, skill_name: str | None = None + ) -> TaskRecord: """Create a new task record in Redis.""" redis = await self._get_redis() @@ -305,7 +322,9 @@ end return encoded """ - async def update_status(self, task_id: str, status: TaskStatus, reset_ttl: bool = False, **kwargs) -> TaskRecord: + async def update_status( + self, task_id: str, status: TaskStatus, reset_ttl: bool = False, **kwargs + ) -> TaskRecord: """Update task status and optional fields atomically via Lua script. Args: @@ -322,7 +341,15 @@ return encoded # Build flat list of key-value pairs for the merge fields merge_fields = {"status": status.value} for k, value in kwargs.items(): - if k in ("started_at", "completed_at", "output_data", "error_message", "progress", "progress_message", "metadata"): + if k in ( + "started_at", + "completed_at", + "output_data", + "error_message", + "progress", + "progress_message", + "metadata", + ): if isinstance(value, datetime): merge_fields[k] = value.isoformat() else: @@ -340,7 +367,9 @@ return encoded data = json.loads(result) return TaskRecord.from_dict(data) - async def list_tasks(self, status: TaskStatus | None = None, limit: int = 100) -> list[TaskRecord]: + async def list_tasks( + self, status: TaskStatus | None = None, limit: int = 100 + ) -> list[TaskRecord]: """List tasks, optionally filtered by status, sorted by created_at desc.""" redis = await self._get_redis() tasks: list[TaskRecord] = [] @@ -433,7 +462,11 @@ return encoded await redis.zrem(self.ZSET_KEY, tid) continue record = TaskRecord.from_dict(json.loads(raw)) - if record.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED) and record.completed_at is not None: + if ( + record.status + in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED) + and record.completed_at is not None + ): await redis.delete(self._key(tid)) await redis.zrem(self.ZSET_KEY, tid) return True @@ -452,7 +485,11 @@ return encoded if raw is None: continue record = TaskRecord.from_dict(json.loads(raw)) - if record.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED): + if record.status in ( + TaskStatus.COMPLETED, + TaskStatus.FAILED, + TaskStatus.CANCELLED, + ): tasks.append(record) if cursor == 0: break @@ -505,7 +542,9 @@ def create_task_store( logger.info(f"TaskStore backend: redis ({_sanitize_redis_url(redis_url)})") return store except Exception as exc: - logger.warning(f"Failed to initialise RedisTaskStore ({exc}), falling back to InMemoryTaskStore") + logger.warning( + f"Failed to initialise RedisTaskStore ({exc}), falling back to InMemoryTaskStore" + ) store = InMemoryTaskStore(ttl_seconds=ttl_seconds, max_records=max_records) logger.info("TaskStore backend: memory") diff --git a/src/agentkit/session/store.py b/src/agentkit/session/store.py index 23b5f2e..4628d87 100644 --- a/src/agentkit/session/store.py +++ b/src/agentkit/session/store.py @@ -24,12 +24,18 @@ class SessionStore(Protocol): async def save_session(self, session: Session) -> None: ... async def get_session(self, session_id: str) -> Session | None: ... - async def update_session_status(self, session_id: str, status: SessionStatus) -> Session | None: ... + async def update_session_status( + self, session_id: str, status: SessionStatus + ) -> Session | None: ... async def delete_session(self, session_id: str) -> bool: ... - async def list_sessions(self, agent_name: str | None = None, limit: int = 100) -> list[Session]: ... + async def list_sessions( + self, agent_name: str | None = None, limit: int = 100 + ) -> list[Session]: ... async def append_message(self, message: Message) -> None: ... - async def get_messages(self, session_id: str, limit: int | None = None, offset: int = 0) -> list[Message]: ... + async def get_messages( + self, session_id: str, limit: int | None = None, offset: int = 0 + ) -> list[Message]: ... async def count_messages(self, session_id: str) -> int: ... async def health_check(self) -> bool: ... @@ -91,7 +97,9 @@ class InMemorySessionStore: del msgs[:excess] msgs.append(message) - async def get_messages(self, session_id: str, limit: int | None = None, offset: int = 0) -> list[Message]: + async def get_messages( + self, session_id: str, limit: int | None = None, offset: int = 0 + ) -> list[Message]: msgs = self._messages.get(session_id, []) sliced = msgs[offset:] if limit is not None: @@ -125,7 +133,13 @@ class RedisSessionStore: if self._redis is None: import redis.asyncio as aioredis - self._redis = aioredis.from_url(self._redis_url, decode_responses=True) + # DEPLOY-2: Helm 注入 REDIS_PASSWORD env var，redis-py from_url 不自动读取， + # 必须显式以 password= 传入。env var 未设置时 password=None 保持向后兼容。 + self._redis = aioredis.from_url( + self._redis_url, + decode_responses=True, + password=os.environ.get("REDIS_PASSWORD") or None, + ) return self._redis def _session_key(self, session_id: str) -> str: @@ -192,7 +206,9 @@ class RedisSessionStore: # Set TTL on message list to match session TTL await redis.expire(key, self._ttl_seconds) - async def get_messages(self, session_id: str, limit: int | None = None, offset: int = 0) -> list[Message]: + async def get_messages( + self, session_id: str, limit: int | None = None, offset: int = 0 + ) -> list[Message]: redis = await self._get_redis() key = self._messages_key(session_id) # Use LRANGE for offset-based pagination @@ -304,7 +320,9 @@ class FileSessionStore: data["session"]["updated_at"] = datetime.now(timezone.utc).isoformat() self._write_session_file(message.session_id, data) - async def get_messages(self, session_id: str, limit: int | None = None, offset: int = 0) -> list[Message]: + async def get_messages( + self, session_id: str, limit: int | None = None, offset: int = 0 + ) -> list[Message]: data = self._read_session_file(session_id) if data is None: return [] @@ -347,10 +365,12 @@ def create_session_store( import redis.asyncio as aioredis # noqa: F401 store = RedisSessionStore(redis_url=redis_url, ttl_seconds=ttl_seconds) - logger.info(f"SessionStore backend: redis") + logger.info("SessionStore backend: redis") return store except (ImportError, OSError, _RedisError, ValueError, KeyError, RuntimeError) as exc: - logger.warning(f"Failed to initialise RedisSessionStore ({exc}), falling back to InMemorySessionStore") + logger.warning( + f"Failed to initialise RedisSessionStore ({exc}), falling back to InMemorySessionStore" + ) store = InMemorySessionStore() logger.info("SessionStore backend: memory")