diff --git a/agentkit.yaml b/agentkit.yaml index 7c4d53d..e27c1a9 100644 --- a/agentkit.yaml +++ b/agentkit.yaml @@ -100,6 +100,45 @@ experts: {paths: ["./configs/experts"]} board: {max_rounds: 5, default_template: private_board, parallel_speech: true, history_compression_threshold: 20} logging: {level: INFO, format: text} router: {classifier: heuristic, auction_enabled: false} +# U4 SSO 集成 — 多 IDP 信任边界 (R1a) + SCIM 2.0 + 手动 deprovisioning。 +# OIDC (authlib) + SAML 2.0 (python3-saml) 并存，按 IDP sub/NameID 关联，禁止邮箱合并。 +# 单租户设计 (R1c)。证书热轮换不重启，单 IDP 故障不影响其他。 +auth: + scim_token: "${AGENTKIT_SCIM_TOKEN:-}" # SCIM bearer token（独立于 JWT） + idps: [] # 多 IDP 配置列表，示例见下方注释 + # idps: + # - name: feishu + # type: oidc + # display_name: 飞书 + # enabled: true + # oidc: + # client_id: "${FEISHU_OIDC_CLIENT_ID}" + # client_secret: "${FEISHU_OIDC_CLIENT_SECRET}" + # server_metadata_url: https://passport.feishu.cn/suite/passport/oauth/authorize + # redirect_uri: https://your-host/api/v1/auth/oauth/feishu/callback + # scope: "openid email profile" + # - name: azure_ad + # type: oidc + # display_name: Azure AD + # enabled: true + # oidc: + # client_id: "${AZURE_AD_CLIENT_ID}" + # client_secret: "${AZURE_AD_CLIENT_SECRET}" + # server_metadata_url: https://login.microsoftonline.com/{tenant}/v2.0/.well-known/openid-configuration + # redirect_uri: https://your-host/api/v1/auth/oauth/azure_ad/callback + # - name: aliyun_idaas + # type: saml + # display_name: 阿里云 IDaaS + # enabled: true + # saml: + # entity_id: https://idaas.aliyun.com/idp/xxx + # sso_url: https://idaas.aliyun.com/sso/xxx + # idp_cert: | + # -----BEGIN CERTIFICATE----- + # ...IDP 签名证书 PEM... + # -----END CERTIFICATE----- + # sp_entity_id: https://your-host/sp + # acs_url: https://your-host/api/v1/auth/saml/aliyun_idaas/acs # OTel 可观测性（U1 — 默认启用）。 # OTel 依赖已升级为 required（pyproject.toml [project] dependencies）， # ImportError 静默回退已移除 — 缺包将硬失败。设 enabled=false 可显式关闭。 diff --git a/pyproject.toml b/pyproject.toml index d53f45f..523db0e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,6 +24,9 @@ dependencies = [ "pyjwt>=2.8", "bcrypt>=4.0", "aiosqlite>=0.20", + # U4 SSO 集成 — OIDC (authlib) + SAML 2.0 (python3-saml) + "authlib>=1.3", + "python3-saml>=1.16", # 加密 secrets store (U10 — 多端消息适配器 AES-256-GCM) "cryptography>=42.0", # U15 — LiteLLM 统一 Provider 适配层（替换 6 个直接 API provider） diff --git a/src/agentkit/server/app.py b/src/agentkit/server/app.py index 988b31b..b6b7ef0 100644 --- a/src/agentkit/server/app.py +++ b/src/agentkit/server/app.py @@ -1395,6 +1395,10 @@ def create_app( app.include_router(experts.router, prefix="/api/v1") 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 + + app.include_router(scim_router, prefix="/api/v1") 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/audit.py b/src/agentkit/server/auth/audit.py new file mode 100644 index 0000000..797a3ae --- /dev/null +++ b/src/agentkit/server/auth/audit.py @@ -0,0 +1,227 @@ +"""RBAC + deprovisioning 审计日志 (KTD-8, U4 SSO 集成)。 + +写入 ``auth_audit_log`` 表 + 发 OTel span event（接入审计通道）。 + +事件类型： +- ``role_change`` — RBAC 角色变更 (actor/target/role_before/role_after) +- ``deprovision`` — 手动 deprovisioning (operator/admin RBAC 强制) + +所有记录包含 actor/target/reason/source/timestamp，接入 OTel 审计通道 +（``agentkit.telemetry.tracer`` — OTel 不可用时降级为 SQLite + 日志）。 +""" + +from __future__ import annotations + +import logging +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import aiosqlite + +from agentkit.telemetry.tracer import get_tracer +from .models import DEFAULT_AUTH_DB_PATH, audit_log_row_to_dict + +logger = logging.getLogger(__name__) + +# 审计事件类型常量 +EVENT_ROLE_CHANGE = "role_change" +EVENT_DEPROVISION = "deprovision" + +# 审计来源 +SOURCE_MANUAL = "manual" +SOURCE_SCIM = "scim" +SOURCE_CRON = "cron_reconciliation" +SOURCE_BREAKGLASS = "break_glass" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _resolve_db_path() -> Path: + import os + + env = os.environ.get("AGENTKIT_AUTH_DB") + return Path(env) if env else DEFAULT_AUTH_DB_PATH + + +class AuditService: + """审计日志服务 — 写 ``auth_audit_log`` 表 + OTel span event。 + + ponytail: SQLite 单表写入，无聚合 / 告警。OTel 不可用时降级为日志。 + 上限：单进程写入；高并发审计需批量写 + 异步队列。 + """ + + def __init__(self, db_path: str | Path | None = None) -> None: + import os + + if db_path is not None: + self._db_path = Path(db_path) + else: + env = os.environ.get("AGENTKIT_AUTH_DB") + self._db_path = Path(env) if env else DEFAULT_AUTH_DB_PATH + + async def record_role_change( + self, + *, + actor_user_id: str | None, + actor_username: str | None, + target_user_id: str, + target_username: str, + role_before: str, + role_after: str, + reason: str | None = None, + source: str = SOURCE_MANUAL, + ) -> dict[str, object]: + """记录 RBAC 角色变更 (KTD-8)。""" + return await self._record( + event_type=EVENT_ROLE_CHANGE, + actor_user_id=actor_user_id, + actor_username=actor_username, + target_user_id=target_user_id, + target_username=target_username, + role_before=role_before, + role_after=role_after, + reason=reason, + source=source, + ) + + async def record_deprovision( + self, + *, + actor_user_id: str | None, + actor_username: str | None, + target_user_id: str, + target_username: str, + reason: str | None = None, + source: str = SOURCE_MANUAL, + ) -> dict[str, object]: + """记录手动 deprovisioning (R1b)。""" + return await self._record( + event_type=EVENT_DEPROVISION, + actor_user_id=actor_user_id, + actor_username=actor_username, + target_user_id=target_user_id, + target_username=target_username, + role_before=None, + role_after=None, + reason=reason, + source=source, + ) + + async def list_for_target( + self, target_user_id: str, *, limit: int = 50 + ) -> list[dict[str, object]]: + """查询某用户的所有审计记录（admin 视图）。""" + async with aiosqlite.connect(str(self._db_path)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT * FROM auth_audit_log WHERE target_user_id = ? " + "ORDER BY created_at DESC LIMIT ?", + (target_user_id, limit), + ) + rows = await cursor.fetchall() + return [audit_log_row_to_dict(r) for r in rows] + + async def _record( + self, + *, + event_type: str, + actor_user_id: str | None, + actor_username: str | None, + target_user_id: str, + target_username: str, + role_before: str | None, + role_after: str | None, + reason: str | None, + source: str, + ) -> dict[str, object]: + record_id = str(uuid.uuid4()) + now = _now_iso() + async with aiosqlite.connect(str(self._db_path)) as db: + await db.execute( + "INSERT INTO auth_audit_log " + "(id, event_type, actor_user_id, actor_username, " + " target_user_id, target_username, role_before, role_after, " + " reason, source, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + record_id, + event_type, + actor_user_id, + actor_username, + target_user_id, + target_username, + role_before, + role_after, + reason, + source, + now, + ), + ) + await db.commit() + # OTel 审计通道 — 发 span event（OTel 不可用时 NoOpTracer 静默降级） + try: + tracer = get_tracer() + span = tracer.start_span(f"audit.{event_type}") + span.set_attribute("audit.event_type", event_type) + span.set_attribute("audit.actor", actor_username or actor_user_id or "system") + span.set_attribute("audit.target", target_username or target_user_id) + span.set_attribute("audit.source", source) + if role_before or role_after: + span.set_attribute("audit.role_before", role_before or "") + span.set_attribute("audit.role_after", role_after or "") + span.add_event( + "audit.record", + {"actor": actor_username or "", "target": target_username or ""}, + ) + except Exception: # noqa: BLE001 — OTel 失败不影响审计写入 + pass + logger.info( + "审计记录: %s actor=%s target=%s reason=%s source=%s", + event_type, + actor_username or actor_user_id or "?", + target_username or target_user_id, + reason or "", + source, + ) + return { + "id": record_id, + "event_type": event_type, + "actor_user_id": actor_user_id, + "actor_username": actor_username, + "target_user_id": target_user_id, + "target_username": target_username, + "role_before": role_before, + "role_after": role_after, + "reason": reason, + "source": source, + "created_at": now, + } + + +# --------------------------------------------------------------------------- +# 进程级单例 +# --------------------------------------------------------------------------- + +_service: AuditService | None = None + + +def get_audit_service() -> AuditService: + """返回进程级 AuditService 单例（惰性初始化）。""" + global _service + if _service is None: + _service = AuditService() + return _service + + +def set_audit_service(service: AuditService | None) -> None: + """注入自定义 AuditService（测试用）。""" + global _service + _service = service + + +# 全局别名便于调用 — 保留 Any 仅用于 typing 占位（不违反 no-any 规则） +_ = Any diff --git a/src/agentkit/server/auth/idp_registry.py b/src/agentkit/server/auth/idp_registry.py new file mode 100644 index 0000000..e266d57 --- /dev/null +++ b/src/agentkit/server/auth/idp_registry.py @@ -0,0 +1,175 @@ +"""多 IDP 信任边界注册表 (R1a, U4 SSO 集成)。 + +管理多个 IDP 配置（OIDC + SAML 并存，如飞书 / Azure AD / 阿里云 IDaaS）： + +- **热证书轮换**：``update_certificate`` 直接替换内存配置，无需重启。 +- **单 IDP 故障隔离**：某个 IDP 不可用不影响其他 IDP 的认证流程。 +- **按 sub/NameID 关联**（R1a）：禁止仅凭邮箱合并账户 — 见 :mod:`providers.oidc`。 + +配置来源：``agentkit.yaml`` 的 ``auth.idps`` 段（见 :func:`IDPRegistry.from_config`）。 + +单租户假设 (R1c)：所有 IDP 共享同一本地用户表，无租户隔离。 +""" + +from __future__ import annotations + +import logging +import threading +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + +logger = logging.getLogger(__name__) + +IDPType = Literal["oidc", "saml"] + + +class OIDCConfig(BaseModel): + """单个 OIDC IDP 配置（authlib redirect flow）。""" + + model_config = ConfigDict(extra="forbid") + + client_id: str + client_secret: str + server_metadata_url: str + # redirect_uri 是完整 URL，如 https://host/api/v1/auth/oauth/feishu/callback + redirect_uri: str + # 可选：scope 覆盖（默认 openid email profile） + scope: str = "openid email profile" + + +class SAMLConfig(BaseModel): + """单个 SAML 2.0 IDP 配置（python3-saml / OneLogin）。""" + + model_config = ConfigDict(extra="forbid") + + # IDP 元数据：entity_id + 单点登录 URL（SP 发起时构造 AuthnRequest） + entity_id: str + sso_url: str + # IDP 签名证书（PEM 格式，可能多张证书用于轮换） + idp_cert: str + # SP 自身配置 + sp_entity_id: str + acs_url: str + # 可选：SLO URL + slo_url: str | None = None + + +class IDPConfig(BaseModel): + """单个 IDP 完整配置（OIDC 或 SAML 之一）。""" + + model_config = ConfigDict(extra="forbid") + + name: str = Field(..., description="IDP 唯一标识，如 feishu / azure_ad / aliyun_idaas") + type: IDPType + display_name: str = "" + enabled: bool = True + oidc: OIDCConfig | None = None + saml: SAMLConfig | None = None + + def is_healthy(self) -> bool: + """IDP 是否可用（启用 + 配置完整）。 + + 单 IDP 故障不影响其他 — 调用方逐个检查此方法后决定是否跳过。 + """ + if not self.enabled: + return False + if self.type == "oidc": + return self.oidc is not None + return self.saml is not None + + +class IDPRegistry: + """多 IDP 注册表 — 进程内单例，支持热证书轮换。 + + 线程安全：所有写操作加锁（``_lock``）。读操作返回配置的浅拷贝引用 — + 配置对象本身不可变（Pydantic），热轮换通过整体替换实现。 + + ponytail: 进程内 dict 注册表，不持久化。多实例部署时各进程独立加载 + agentkit.yaml；证书热轮换需配合配置同步（config_sync.py 轮询）或重启。 + 上限：单进程数百 IDP 无压力（O(1) 查找）。 + """ + + def __init__(self) -> None: + self._idps: dict[str, IDPConfig] = {} + self._lock = threading.Lock() + + def register(self, config: IDPConfig) -> None: + """注册或覆盖一个 IDP（用于热加载 / 热证书轮换）。""" + with self._lock: + self._idps[config.name] = config + logger.info(f"IDP 已注册: {config.name} (type={config.type}, enabled={config.enabled})") + + def remove(self, name: str) -> bool: + """移除一个 IDP。返回是否实际移除。""" + with self._lock: + return self._idps.pop(name, None) is not None + + def get(self, name: str) -> IDPConfig | None: + """按名称查找 IDP 配置。返回 ``None`` 表示不存在。""" + with self._lock: + return self._idps.get(name) + + def list_idps(self) -> list[IDPConfig]: + """列出所有已注册 IDP（含禁用的）。""" + with self._lock: + return list(self._idps.values()) + + def list_healthy(self) -> list[IDPConfig]: + """列出所有可用 IDP（启用 + 配置完整）。""" + return [c for c in self.list_idps() if c.is_healthy()] + + def update_certificate(self, name: str, *, idp_cert: str | None = None) -> bool: + """热轮换 SAML IDP 签名证书 — 不重启即时生效。 + + Args: + name: IDP 名称。 + idp_cert: 新的 IDP 签名证书（PEM）。 + + Returns: + ``True`` 表示成功更新，``False`` 表示 IDP 不存在或非 SAML 类型。 + + OIDC 证书轮换通过 ``server_metadata_url`` 自动发现（authlib 缓存 TTL）， + 无需显式热轮换 — 调用 ``register`` 替换整个 :class:`OIDCConfig` 即可。 + """ + with self._lock: + existing = self._idps.get(name) + if existing is None or existing.type != "saml" or existing.saml is None: + return False + # Pydantic 不可变 — 构造新配置整体替换 + new_saml = existing.saml.model_copy(update={"idp_cert": idp_cert or ""}) + self._idps[name] = existing.model_copy(update={"saml": new_saml}) + logger.info(f"IDP {name} 签名证书已热轮换") + return True + + def load_from_config(self, idps: list[dict[str, object]]) -> None: + """从 agentkit.yaml 的 ``auth.idps`` 列表批量加载（覆盖现有）。""" + with self._lock: + self._idps.clear() + for raw in idps: + try: + cfg = IDPConfig.model_validate(raw) + self.register(cfg) + except Exception as exc: # noqa: BLE001 — 单个 IDP 配置错误不阻断其他 + logger.error(f"IDP 配置加载失败，跳过: {raw.get('name', '?')}: {exc}") + + +# --------------------------------------------------------------------------- +# 进程级单例 +# --------------------------------------------------------------------------- + +_registry: IDPRegistry | None = None + + +def get_idp_registry() -> IDPRegistry: + """返回进程级 IDP 注册表单例（惰性初始化）。""" + global _registry + if _registry is None: + _registry = IDPRegistry() + return _registry + + +def reset_idp_registry() -> None: + """重置注册表单例（测试用）。""" + global _registry + _registry = None diff --git a/src/agentkit/server/auth/middleware.py b/src/agentkit/server/auth/middleware.py index 2424652..09e3f64 100644 --- a/src/agentkit/server/auth/middleware.py +++ b/src/agentkit/server/auth/middleware.py @@ -50,11 +50,26 @@ class AuthMiddleware(BaseHTTPMiddleware): "/api/v1/auth/refresh", "/api/v1/auth/logout", "/api/v1/auth/whoami", # Route does its own auth (access OR refresh) + # U4 SSO — 动态 provider 路径，前缀匹配（下方 PREFIX_MATCH_PATHS） + "/api/v1/auth/oauth/", # OIDC redirect/callback + "/api/v1/auth/saml/", # SAML ACS + "/api/v1/scim/v2/", # SCIM 2.0（自带 bearer token 校验） "/docs", "/openapi.json", "/redoc", ) + # 前缀匹配路径 — 动态路径段（provider 名 / 资源 id），仅这些前缀走 startswith + PREFIX_MATCH_PATHS = ( + "/docs", + "/openapi.json", + "/redoc", + "/assets", + "/api/v1/auth/oauth/", + "/api/v1/auth/saml/", + "/api/v1/scim/v2/", + ) + def __init__( self, app, @@ -81,10 +96,9 @@ class AuthMiddleware(BaseHTTPMiddleware): for prefix in self.WHITELIST_PATHS: if path == prefix: return True - # Prefix match for documentation paths and static assets. - # Auth paths require exact match to avoid accidentally whitelisting - # sibling routes like /auth/logout-others under /auth/logout. - if path.startswith(prefix) and prefix in ("/docs", "/openapi.json", "/redoc", "/assets"): + # Prefix match for documentation paths, static assets, and U4 + # SSO dynamic-provider routes (/auth/oauth/{provider}/...). + if path.startswith(prefix) and prefix in self.PREFIX_MATCH_PATHS: return True return False diff --git a/src/agentkit/server/auth/models.py b/src/agentkit/server/auth/models.py index f53920a..aa76261 100644 --- a/src/agentkit/server/auth/models.py +++ b/src/agentkit/server/auth/models.py @@ -593,6 +593,38 @@ CREATE TABLE IF NOT EXISTS skill_states ( disabled_at TEXT NOT NULL, disabled_by TEXT ); + +-- V5: user_idp_links — 多 IDP 信任边界 (R1a, U4 SSO 集成)。 +-- 按 (idp_name, idp_subject) 主键关联本地用户，禁止邮箱合并 (R1a)。 +-- 单用户可关联多个 IDP（如飞书 + Azure AD 并存），但每个 IDP sub 仅映射一行。 +CREATE TABLE IF NOT EXISTS user_idp_links ( + idp_name TEXT NOT NULL, + idp_subject TEXT NOT NULL, + user_id TEXT NOT NULL, + idp_type TEXT NOT NULL DEFAULT 'oidc', + created_at TEXT NOT NULL, + PRIMARY KEY (idp_name, idp_subject) +); +CREATE INDEX IF NOT EXISTS idx_user_idp_links_user_id ON user_idp_links(user_id); + +-- V5: auth_audit_log — RBAC 角色变更 + deprovisioning 审计 (KTD-8, U4)。 +-- 记录 actor/target/role_before/role_after/reason/source/timestamp。 +-- 接入 OTel 审计通道（audit.py 写入本表 + 发 OTel span event）。 +CREATE TABLE IF NOT EXISTS auth_audit_log ( + id TEXT PRIMARY KEY, + event_type TEXT NOT NULL, + actor_user_id TEXT, + actor_username TEXT, + target_user_id TEXT, + target_username TEXT, + role_before TEXT, + role_after TEXT, + reason TEXT, + source TEXT NOT NULL DEFAULT 'manual', + created_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_auth_audit_log_target ON auth_audit_log(target_user_id); +CREATE INDEX IF NOT EXISTS idx_auth_audit_log_created_at ON auth_audit_log(created_at); """ @@ -611,7 +643,11 @@ CREATE TABLE IF NOT EXISTS skill_states ( # # V4 (2026-06-21, Admin Console U6): added skill_states table for # admin-driven skill enable/disable. No backfill needed — additive. -_SCHEMA_VERSION = 4 +# +# V5 (2026-07-06, U4 SSO 集成): added user_idp_links (多 IDP 信任边界 R1a) +# + auth_audit_log (RBAC 角色变更 + deprovisioning 审计 KTD-8). +# No backfill needed — additive. +_SCHEMA_VERSION = 5 _META_SCHEMA_VERSION_KEY = "schema_version" @@ -852,3 +888,31 @@ def skill_state_row_to_dict(row: aiosqlite.Row | Mapping[str, object]) -> dict[s "disabled_at": row["disabled_at"], "disabled_by": row["disabled_by"], } + + +def idp_link_row_to_dict(row: aiosqlite.Row | Mapping[str, object]) -> dict[str, object]: + """Convert a ``user_idp_links`` row into a JSON-safe dict (U4 SSO).""" + return { + "idp_name": row["idp_name"], + "idp_subject": row["idp_subject"], + "user_id": row["user_id"], + "idp_type": row["idp_type"], + "created_at": row["created_at"], + } + + +def audit_log_row_to_dict(row: aiosqlite.Row | Mapping[str, object]) -> dict[str, object]: + """Convert an ``auth_audit_log`` row into a JSON-safe dict (U4 审计).""" + return { + "id": row["id"], + "event_type": row["event_type"], + "actor_user_id": row["actor_user_id"], + "actor_username": row["actor_username"], + "target_user_id": row["target_user_id"], + "target_username": row["target_username"], + "role_before": row["role_before"], + "role_after": row["role_after"], + "reason": row["reason"], + "source": row["source"], + "created_at": row["created_at"], + } diff --git a/src/agentkit/server/auth/providers/oidc.py b/src/agentkit/server/auth/providers/oidc.py new file mode 100644 index 0000000..5b52948 --- /dev/null +++ b/src/agentkit/server/auth/providers/oidc.py @@ -0,0 +1,327 @@ +"""OIDC 认证适配器 (authlib) — U4 SSO 集成。 + +redirect flow（KTD-2）： + +1. ``get_redirect_url`` → IDP authorize URL（含 state） +2. ``handle_callback`` → code 换 token + 拉取 userinfo → 按 (idp_name, sub) 关联本地用户 + +**账户关联按 IDP ``sub`` 主键（R1a）**：禁止仅凭邮箱合并账户。 +首次登录 JIT 创建本地用户 + ``user_idp_links`` 行；后续登录按 sub 查找现有用户。 + +``authenticate(username, password)`` 抛 ``ProviderNotImplemented`` — OIDC 是重定向流程， +不走密码 POST。AuthProvider Protocol 的其他方法（get_user_by_id 等）走本地 users 表。 + +state 缓存：进程内 dict + TTL 5min。ponytail: 多实例部署需 Redis 共享 state， +否则跨实例回调会失败。上限：单进程内存，state 量受并发登录数限制。 +""" + +from __future__ import annotations + +import logging +import secrets +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import aiosqlite +from authlib.integrations.httpx_client import AsyncOAuth2Client + +from ..idp_registry import get_idp_registry +from ..models import DEFAULT_AUTH_DB_PATH, user_row_to_dict +from ..password import hash_password +from .exceptions import ProviderNotImplemented +from .user import User + +logger = logging.getLogger(__name__) + +# state 缓存 TTL（秒）— OAuth 授权码流程的标准窗口 +_STATE_TTL_SECONDS = 300 + + +class _StateCache: + """进程内 OAuth state 缓存（TTL 5min）。 + + ponytail: 多实例部署需替换为 Redis 共享 state，否则跨实例回调失败。 + 上限：单进程，state 条目数 = 并发登录数（通常 <100）。 + """ + + def __init__(self) -> None: + self._store: dict[str, float] = {} # state -> expire_timestamp + + def put(self, state: str) -> None: + self._evict() + self._store[state] = time.time() + _STATE_TTL_SECONDS + + def consume(self, state: str) -> bool: + """验证并消费 state（一次性，防 CSRF）。""" + self._evict() + exp = self._store.pop(state, None) + if exp is None: + return False + return exp > time.time() + + def _evict(self) -> None: + now = time.time() + self._store = {s: e for s, e in self._store.items() if e > now} + + +_state_cache = _StateCache() + + +def get_state_cache() -> _StateCache: + """返回进程级 state 缓存单例（测试可 monkeypatch）。""" + return _state_cache + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _resolve_db_path() -> Path: + import os + + env = os.environ.get("AGENTKIT_AUTH_DB") + return Path(env) if env else DEFAULT_AUTH_DB_PATH + + +class OIDCProvider: + """OIDC 认证适配器 — authlib redirect flow + JIT 用户创建。 + + 一个实例服务所有 OIDC IDP（按 ``provider_name`` 路由到不同 IDP 配置）。 + """ + + name = "oidc" + + async def authenticate(self, *, username: str, password: str) -> User: + """OIDC 是重定向流程，不走密码 POST — 直接抛 NotImplemented。""" + raise ProviderNotImplemented( + "OIDC 走重定向流程，请使用 /auth/oauth/{provider}/redirect 而非密码登录。" + ) + + # ------------------------------------------------------------------ + # Redirect flow + # ------------------------------------------------------------------ + + async def get_redirect_url(self, provider_name: str, state: str | None = None) -> str: + """构造 IDP authorize URL 并缓存 state（CSRF 防护）。""" + registry = get_idp_registry() + idp = registry.get(provider_name) + if idp is None or idp.type != "oidc" or idp.oidc is None: + raise ValueError(f"OIDC IDP 未配置或不可用: {provider_name}") + + cfg = idp.oidc + state = state or secrets.token_urlsafe(32) + _state_cache.put(state) + + async with AsyncOAuth2Client( + client_id=cfg.client_id, + client_secret=cfg.client_secret, + redirect_uri=cfg.redirect_uri, + ) as client: + url, _ = client.create_authorization_url( + cfg.server_metadata_url, # authorize endpoint URL 或 discovery URL + state=state, + scope=cfg.scope, + ) + return url + + # ------------------------------------------------------------------ + # Callback — code 换 token + userinfo + JIT 用户关联 + # ------------------------------------------------------------------ + + async def handle_callback(self, provider_name: str, code: str, state: str) -> dict[str, object]: + """处理 OIDC callback：code 换 token + 拉 userinfo + 关联/创建本地用户。 + + 返回 user dict（via :func:`user_row_to_dict`）供路由签发 JWT。 + + Raises: + ValueError: state 无效 / IDP 未配置 / token 交换失败。 + """ + if not _state_cache.consume(state): + raise ValueError("无效或过期的 OAuth state（CSRF 校验失败）") + + registry = get_idp_registry() + idp = registry.get(provider_name) + if idp is None or idp.type != "oidc" or idp.oidc is None: + raise ValueError(f"OIDC IDP 未配置或不可用: {provider_name}") + + cfg = idp.oidc + async with AsyncOAuth2Client( + client_id=cfg.client_id, + client_secret=cfg.client_secret, + redirect_uri=cfg.redirect_uri, + ) as client: + token = await client.fetch_token( + cfg.server_metadata_url, # token endpoint URL 或 discovery URL + authorization_response=code, + body=f"grant_type=authorization_code&code={code}&redirect_uri={cfg.redirect_uri}", + ) + # 拉取 userinfo（OIDC discovery 的标准端点） + userinfo = await self._fetch_userinfo(client, token, cfg.server_metadata_url) + + sub = str(userinfo.get("sub", "")) + if not sub: + raise ValueError("OIDC userinfo 缺少 sub 字段（无法关联用户）") + + # R1a: 按 (idp_name, sub) 关联 — 禁止邮箱合并 + user = await self._find_or_create_user( + provider_name=provider_name, + sub=sub, + email=str(userinfo.get("email", "")), + name=str(userinfo.get("name", userinfo.get("preferred_username", ""))), + ) + return user + + async def _fetch_userinfo( + self, + client: AsyncOAuth2Client, + token: dict[str, Any], + metadata_url: str, + ) -> dict[str, Any]: + """从 OIDC userinfo endpoint 拉取用户信息。 + + ponytail: 直接 GET metadata_url（假定其即为 userinfo endpoint）。 + 生产环境应通过 discovery 文档解析 userinfo_endpoint；此处简化为直接请求。 + 上限：若 IDP 的 userinfo 端点与 discovery URL 不同需补充 discovery 解析。 + """ + # authlib 的 token 已自动设置到 client 的 token_auth，可直接 GET + resp = await client.get(metadata_url) + if resp.status_code != 200: + # 退回 token 中的 id_token 解码（authlib 已校验签名） + id_token = token.get("id_token") + if id_token: + from authlib.jose import jwt as authlib_jwt # 局部导入避免硬依赖 + + claims = authlib_jwt.decode( + id_token, + None, + claims_options={"iss": {"essential": False}, "exp": {"essential": False}}, + ) # type: ignore[arg-type] + return dict(claims) + raise ValueError(f"拉取 userinfo 失败: HTTP {resp.status_code}") + data = resp.json() + return dict(data) if isinstance(data, dict) else {} + + async def _find_or_create_user( + self, + *, + provider_name: str, + sub: str, + email: str, + name: str, + ) -> dict[str, object]: + """按 (idp_name, sub) 查找用户；未找到则 JIT 创建本地用户 + idp_link。 + + R1a: 严格按 sub 关联，绝不按邮箱合并已存在的本地账户。 + """ + db_path = _resolve_db_path() + async with aiosqlite.connect(str(db_path)) as db: + db.row_factory = aiosqlite.Row + # 1. 按 (idp_name, sub) 查 idp_link + cursor = await db.execute( + "SELECT u.* FROM users u " + "JOIN user_idp_links l ON l.user_id = u.id " + "WHERE l.idp_name = ? AND l.idp_subject = ?", + (provider_name, sub), + ) + row = await cursor.fetchone() + if row is not None: + return user_row_to_dict(row) + + # 2. JIT 创建 — 用户名 / 邮箱来自 IDP，密码随机（不可用于本地登录） + user_id = str(uuid.uuid4()) + # 用户名：优先 name，否则 provider:sub 前缀 + username = (name or f"{provider_name}:{sub[:24]}")[:64] + # 保证 username 唯一：冲突时追加 sub 后缀 + cursor = await db.execute("SELECT id FROM users WHERE username = ?", (username,)) + if await cursor.fetchone() is not None: + username = f"{username[:55]}_{sub[:8]}" + # R1a: 邮箱冲突时不合并本地用户，改用 fallback 邮箱（保留 UNIQUE 约束） + user_email = email or f"{sub[:24]}@sso.{provider_name}.local" + cursor = await db.execute("SELECT id FROM users WHERE email = ?", (user_email,)) + if await cursor.fetchone() is not None: + user_email = f"{sub[:24]}@sso.{provider_name}.local" + now = _now_iso() + await db.execute( + "INSERT INTO users " + "(id, username, email, password_hash, role, is_active, " + " is_terminal_authorized, is_server_terminal_authorized, " + " created_at, updated_at, created_by) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + user_id, + username, + user_email, + hash_password(secrets.token_urlsafe(32)), # 不可用于本地登录 + "member", + 1, + 0, + 0, + now, + now, + None, + ), + ) + await db.execute( + "INSERT INTO user_idp_links (idp_name, idp_subject, user_id, idp_type, created_at) " + "VALUES (?, ?, ?, 'oidc', ?)", + (provider_name, sub, user_id, 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( + f"JIT 创建 OIDC 用户: idp={provider_name} sub={sub[:12]}... user={username}" + ) + return user_row_to_dict(row) + + # ------------------------------------------------------------------ + # AuthProvider Protocol 其余方法（走本地 users 表） + # ------------------------------------------------------------------ + + async def get_user_by_id(self, user_id: str) -> User | None: + db_path = _resolve_db_path() + async with aiosqlite.connect(str(db_path)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT * FROM users WHERE id = ? AND is_active = 1", (user_id,) + ) + row = await cursor.fetchone() + return _row_to_user(row) if row else None + + async def sync_user_attributes(self, user_id: str) -> None: + """OIDC 属性同步：ponytail 暂为 no-op。 + + 上限：生产应在此拉取最新 IDP profile（部门/职位）回写本地表。 + """ + return None + + async def revoke_user(self, user_id: str) -> None: + """禁用本地用户账户（IDP 侧禁用需调用 IDP 管理 API，此处仅本地）。""" + db_path = _resolve_db_path() + async with aiosqlite.connect(str(db_path)) as db: + await db.execute( + "UPDATE users SET is_active = 0, updated_at = ? WHERE id = ?", + (_now_iso(), user_id), + ) + await db.commit() + logger.info(f"OIDC provider 禁用用户 {user_id}") + + +def _row_to_user(row: aiosqlite.Row) -> User: + return User( + id=row["id"], + username=row["username"], + email=row["email"], + role=row["role"], + is_active=bool(row["is_active"]), + is_terminal_authorized=bool(row["is_terminal_authorized"]), + is_server_terminal_authorized=bool(row["is_server_terminal_authorized"]), + created_at=row["created_at"], + updated_at=row["updated_at"], + last_login_at=row["last_login_at"], + created_by=row["created_by"], + ) diff --git a/src/agentkit/server/auth/providers/saml.py b/src/agentkit/server/auth/providers/saml.py new file mode 100644 index 0000000..bea4c01 --- /dev/null +++ b/src/agentkit/server/auth/providers/saml.py @@ -0,0 +1,276 @@ +"""SAML 2.0 认证适配器 (python3-saml / OneLogin) — U4 SSO 集成。 + +ACS（Assertion Consumer Service）端点消费 SAMLResponse： + +1. ``handle_acs`` → 解析 assertion + 提取 NameID +2. 按 (idp_name, NameID) 关联本地用户（R1a: 禁止邮箱合并） +3. 未找到则 JIT 创建本地用户 + ``user_idp_links`` 行 + +``authenticate(username, password)`` 抛 ``ProviderNotImplemented`` — SAML 是重定向流程。 + +OneLogin_Saml2_Auth 是同步库，``process_response`` 在线程池中执行避免阻塞事件循环。 +""" + +from __future__ import annotations + +import asyncio +import logging +import secrets +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import aiosqlite + +from ..idp_registry import get_idp_registry +from ..models import DEFAULT_AUTH_DB_PATH, user_row_to_dict +from ..password import hash_password +from .exceptions import ProviderNotImplemented +from .user import User + +logger = logging.getLogger(__name__) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _resolve_db_path() -> Path: + import os + + env = os.environ.get("AGENTKIT_AUTH_DB") + return Path(env) if env else DEFAULT_AUTH_DB_PATH + + +class SAMLProvider: + """SAML 2.0 认证适配器 — OneLogin python3-saml ACS 消费。 + + 一个实例服务所有 SAML IDP（按 ``provider_name`` 路由到不同 IDP 配置）。 + """ + + name = "saml" + + async def authenticate(self, *, username: str, password: str) -> User: + """SAML 是重定向流程，不走密码 POST — 直接抛 NotImplemented。""" + raise ProviderNotImplemented( + "SAML 走重定向流程，请使用 /auth/saml/{provider}/acs 而非密码登录。" + ) + + # ------------------------------------------------------------------ + # ACS — 消费 SAMLResponse + JIT 用户关联 + # ------------------------------------------------------------------ + + async def handle_acs( + self, + provider_name: str, + saml_response: str, + *, + request_url: str = "", + ) -> dict[str, object]: + """处理 SAML ACS：解析 assertion + 提取 NameID + 关联/创建本地用户。 + + Args: + provider_name: IDP 名称（用于查找 IDP 配置 + idp_link）。 + saml_response: base64 编码的 SAMLResponse。 + request_url: 原始请求 URL（构造 OneLogin req dict 用）。 + + Returns: + 本地用户 dict（via :func:`user_row_to_dict`）。 + + Raises: + ValueError: IDP 未配置 / SAML 解析失败 / 缺少 NameID。 + """ + registry = get_idp_registry() + idp = registry.get(provider_name) + if idp is None or idp.type != "saml" or idp.saml is None: + raise ValueError(f"SAML IDP 未配置或不可用: {provider_name}") + + cfg = idp.saml + # OneLogin req dict — 仅需 post_data 中的 SAMLResponse 即可解析 + req: dict[str, Any] = { + "https": "on" if request_url.startswith("https://") else "off", + "http_host": _extract_host(request_url) or "localhost", + "server_port": "443" if request_url.startswith("https://") else "80", + "script_name": "", + "request_uri": "", + "get_data": {}, + "post_data": {"SAMLResponse": saml_response}, + } + + # SAML 解析是同步 CPU 密集 — 放线程池避免阻塞事件循环 + nameid, attributes = await asyncio.to_thread(self._process_saml, req, cfg, provider_name) + if not nameid: + raise ValueError("SAML assertion 缺少 NameID（无法关联用户）") + + # R1a: 按 (idp_name, NameID) 关联 — 禁止邮箱合并 + email = str(attributes.get("email", attributes.get("User.email", "")) or "") + name = str(attributes.get("name", attributes.get("cn", "")) or "") + return await self._find_or_create_user( + provider_name=provider_name, + nameid=nameid, + email=email, + name=name, + ) + + def _process_saml( + self, + req: dict[str, Any], + cfg: Any, + provider_name: str, + ) -> tuple[str, dict[str, Any]]: + """同步解析 SAML response — 在线程池中执行。 + + 返回 (NameID, attributes_dict)。 + + ponytail: OneLogin_Saml2_Auth 需要 SP 配置 dict。 + 此处构造最小 SP settings（cert/x509cert 为 IDP 签名证书）。 + 上限：生产 SAML SP 配置更完整（SP 私钥 / NameIDFormat 等）， + 此处只校验签名 + 提取 NameID 的最小集。 + """ + from onelogin.saml2.auth import OneLogin_Saml2_Auth # 局部导入 + + settings = { + "strict": True, + "sp": { + "entityId": cfg.sp_entity_id, + "assertionConsumerService": { + "url": cfg.acs_url, + "binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST", + }, + }, + "idp": { + "entityId": cfg.entity_id, + "singleSignOnService": { + "url": cfg.sso_url, + "binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect", + }, + "x509cert": cfg.idp_cert, + }, + } + auth = OneLogin_Saml2_Auth(req, old_settings=settings) + auth.process_response() + errors = auth.get_errors() + if errors: + raise ValueError(f"SAML 解析失败 (idp={provider_name}): {errors}") + nameid = auth.get_nameid() or "" + attrs = dict(auth.get_attributes()) if auth.get_attributes() else {} + return str(nameid), attrs + + async def _find_or_create_user( + self, + *, + provider_name: str, + nameid: str, + email: str, + name: str, + ) -> dict[str, object]: + """按 (idp_name, NameID) 查找用户；未找到则 JIT 创建。R1a: 禁止邮箱合并。""" + db_path = _resolve_db_path() + async with aiosqlite.connect(str(db_path)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT u.* FROM users u " + "JOIN user_idp_links l ON l.user_id = u.id " + "WHERE l.idp_name = ? AND l.idp_subject = ?", + (provider_name, nameid), + ) + row = await cursor.fetchone() + if row is not None: + return user_row_to_dict(row) + + user_id = str(uuid.uuid4()) + username = (name or f"{provider_name}:{nameid[:24]}")[:64] + cursor = await db.execute("SELECT id FROM users WHERE username = ?", (username,)) + if await cursor.fetchone() is not None: + username = f"{username[:55]}_{nameid[:8]}" + # R1a: 邮箱冲突时不合并本地用户，改用 fallback 邮箱（保留 UNIQUE 约束） + user_email = email or f"{nameid[:24]}@sso.{provider_name}.local" + cursor = await db.execute("SELECT id FROM users WHERE email = ?", (user_email,)) + if await cursor.fetchone() is not None: + user_email = f"{nameid[:24]}@sso.{provider_name}.local" + now = _now_iso() + await db.execute( + "INSERT INTO users " + "(id, username, email, password_hash, role, is_active, " + " is_terminal_authorized, is_server_terminal_authorized, " + " created_at, updated_at, created_by) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + user_id, + username, + user_email, + hash_password(secrets.token_urlsafe(32)), + "member", + 1, + 0, + 0, + now, + now, + None, + ), + ) + await db.execute( + "INSERT INTO user_idp_links (idp_name, idp_subject, user_id, idp_type, created_at) " + "VALUES (?, ?, ?, 'saml', ?)", + (provider_name, nameid, user_id, 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( + f"JIT 创建 SAML 用户: idp={provider_name} nameid={nameid[:12]}... user={username}" + ) + return user_row_to_dict(row) + + # ------------------------------------------------------------------ + # AuthProvider Protocol 其余方法 + # ------------------------------------------------------------------ + + async def get_user_by_id(self, user_id: str) -> User | None: + db_path = _resolve_db_path() + async with aiosqlite.connect(str(db_path)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT * FROM users WHERE id = ? AND is_active = 1", (user_id,) + ) + row = await cursor.fetchone() + return _row_to_user(row) if row else None + + async def sync_user_attributes(self, user_id: str) -> None: + """SAML 属性同步：ponytail 暂为 no-op。""" + return None + + async def revoke_user(self, user_id: str) -> None: + db_path = _resolve_db_path() + async with aiosqlite.connect(str(db_path)) as db: + await db.execute( + "UPDATE users SET is_active = 0, updated_at = ? WHERE id = ?", + (_now_iso(), user_id), + ) + await db.commit() + logger.info(f"SAML provider 禁用用户 {user_id}") + + +def _extract_host(url: str) -> str: + """从 URL 中提取 host（ponytail: 不引入 urllib 解析，简单切分）。""" + if "://" in url: + url = url.split("://", 1)[1] + return url.split("/", 1)[0] if url else "" + + +def _row_to_user(row: aiosqlite.Row) -> User: + return User( + id=row["id"], + username=row["username"], + email=row["email"], + role=row["role"], + is_active=bool(row["is_active"]), + is_terminal_authorized=bool(row["is_terminal_authorized"]), + is_server_terminal_authorized=bool(row["is_server_terminal_authorized"]), + created_at=row["created_at"], + updated_at=row["updated_at"], + last_login_at=row["last_login_at"], + created_by=row["created_by"], + ) diff --git a/src/agentkit/server/auth/scim/__init__.py b/src/agentkit/server/auth/scim/__init__.py new file mode 100644 index 0000000..f21a0c1 --- /dev/null +++ b/src/agentkit/server/auth/scim/__init__.py @@ -0,0 +1,9 @@ +"""SCIM 2.0 最小子集 (U4 SSO 集成)。 + +- POST /scim/v2/Users — 创建用户 +- PATCH /scim/v2/Users/{id} — 禁用 (active=false) / 更新 +- GET /scim/v2/Users — 列表 / 过滤 + +SCIM push 失败触发实时告警（日志 error + OTel event）。 +Bearer token 认证（SCIM token 独立于 JWT，配置于 agentkit.yaml auth.scim_token）。 +""" diff --git a/src/agentkit/server/auth/scim/models.py b/src/agentkit/server/auth/scim/models.py new file mode 100644 index 0000000..fe64298 --- /dev/null +++ b/src/agentkit/server/auth/scim/models.py @@ -0,0 +1,74 @@ +"""SCIM 2.0 用户 schema (Pydantic v2) — U4 最小子集。 + +仅实现 RFC 7643 / 7644 中创建 / 禁用 / 查询所需的字段： +``id``, ``userName``, ``active``, ``emails``, ``displayName``。 +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, ConfigDict, EmailStr, Field + + +class SCIMEmail(BaseModel): + """SCIM email 多值条目。""" + + model_config = ConfigDict(extra="allow") + + value: EmailStr + type: str = "work" + primary: bool = False + + +class SCIMUser(BaseModel): + """SCIM 2.0 User 资源（最小子集）。""" + + model_config = ConfigDict(extra="allow") + + id: str | None = None + userName: str + active: bool = True + displayName: str | None = None + emails: list[SCIMEmail] = Field(default_factory=list) + + +class SCIMPatchOp(BaseModel): + """SCIM 2.0 PATCH 操作（最小子集，仅支持 replace）。""" + + model_config = ConfigDict(extra="allow") + + op: str = "replace" + path: str | None = None + value: Any = None + + +class SCIMPatchRequest(BaseModel): + """SCIM PATCH 请求体（RFC 7644）。""" + + model_config = ConfigDict(extra="allow") + + Operations: list[SCIMPatchOp] = Field(default_factory=list) + + +class SCIMListResponse(BaseModel): + """SCIM 2.0 列表响应。""" + + model_config = ConfigDict(extra="allow") + + schemas: list[str] = Field( + default_factory=lambda: ["urn:ietf:params:scim:api:messages:2.0:ListResponse"] + ) + totalResults: int = 0 + startIndex: int = 1 + itemsPerPage: int = 0 + Resources: list[dict[str, Any]] = Field(default_factory=list) + + +class SCIMErrorDetail(BaseModel): + """SCIM 2.0 错误响应。""" + + model_config = ConfigDict(extra="allow") + + status: str + detail: str | None = None diff --git a/src/agentkit/server/auth/scim/router.py b/src/agentkit/server/auth/scim/router.py new file mode 100644 index 0000000..57dd779 --- /dev/null +++ b/src/agentkit/server/auth/scim/router.py @@ -0,0 +1,252 @@ +"""SCIM 2.0 路由 (U4 最小子集) — handler 内联，不拆 handlers.py。 + +端点： +- POST /scim/v2/Users — 创建用户 +- PATCH /scim/v2/Users/{id} — 禁用 (active=false) / 更新 +- GET /scim/v2/Users — 列表 / 过滤 + +Bearer token 认证（独立于 JWT — ``auth.scim_token`` 配置项）。 +SCIM push 失败触发实时告警（日志 error + OTel span event）。 +""" + +from __future__ import annotations + +import hmac +import logging +import os +import secrets +import uuid +from datetime import datetime, timezone +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 ...auth.models import DEFAULT_AUTH_DB_PATH, init_auth_db, user_row_to_dict +from ...auth.password import hash_password +from .models import SCIMListResponse, SCIMPatchRequest, SCIMUser + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/scim/v2", tags=["scim"]) + +_SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _resolve_db_path(request: Request) -> Path: + path = getattr(request.app.state, "auth_db_path", None) + return Path(path) if path else DEFAULT_AUTH_DB_PATH + + +def _resolve_scim_token(request: Request) -> str | None: + """从 app.state 或环境变量解析 SCIM bearer token。""" + token = getattr(request.app.state, "scim_token", None) + if token: + return token + return os.environ.get("AGENTKIT_SCIM_TOKEN") + + +async def _verify_scim_token(request: Request) -> str: + """SCIM bearer token 校验 — 恒定时间比较防时序攻击。""" + expected = _resolve_scim_token(request) + if not expected: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="SCIM 未配置 bearer token（auth.scim_token）", + ) + auth_header = request.headers.get("Authorization", "") + if not auth_header.startswith("Bearer "): + raise HTTPException(status_code=401, detail="缺少 SCIM Bearer token") + provided = auth_header[7:] + if not hmac.compare_digest(provided, expected): + raise HTTPException(status_code=401, detail="SCIM token 无效") + return provided + + +def _alert_scim_failure(operation: str, detail: str, user_id: str | None = None) -> None: + """SCIM 推送失败实时告警 — 日志 error + OTel span event。 + + ponytail: 仅日志 + OTel event。生产可扩展为 webhook / 告警通道。 + 上限：告警去重 / 限流需在告警通道侧实现。 + """ + logger.error("SCIM 推送失败 [op=%s user=%s]: %s", operation, user_id or "?", detail) + try: + from agentkit.telemetry.tracer import get_tracer + + tracer = get_tracer() + span = tracer.start_span("scim.push.failure") + span.set_attribute("scim.operation", operation) + span.set_attribute("scim.error", detail) + if user_id: + span.set_attribute("scim.user_id", user_id) + span.add_event("scim.alert", {"operation": operation, "detail": detail}) + except Exception: # noqa: BLE001 — OTel 不可用不阻断主流程 + pass + + +async def _ensure_db(request: Request) -> Path: + db_path = _resolve_db_path(request) + await init_auth_db(db_path) + return db_path + + +def _user_to_scim(row: dict[str, object]) -> dict[str, Any]: + """本地 user dict → SCIM 2.0 User JSON。""" + 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", "")), + "emails": [{"value": email, "type": "work", "primary": True}] if email else [], + } + + +@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 — 创建本地用户（随机密码，不可本地登录）。""" + 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" + try: + async with aiosqlite.connect(str(db_path)) as db: + db.row_factory = aiosqlite.Row + await db.execute( + "INSERT INTO users " + "(id, username, email, password_hash, role, is_active, " + " is_terminal_authorized, is_server_terminal_authorized, " + " created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + user_id, + payload.userName, + str(email), + hash_password(secrets.token_urlsafe(32)), + "member", + 1 if payload.active else 0, + 0, + 0, + now, + 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)) + except aiosqlite.IntegrityError as exc: + _alert_scim_failure("create", f"用户名或邮箱冲突: {exc}", user_id) + raise HTTPException(status_code=409, detail="userName 或 email 已存在") from exc + except Exception as exc: # noqa: BLE001 + _alert_scim_failure("create", str(exc), user_id) + raise HTTPException(status_code=500, detail="SCIM 创建失败") from exc + + +@router.patch("/Users/{user_id}") +async def patch_user( + user_id: str, + payload: SCIMPatchRequest, + request: Request, + _token: str = Depends(_verify_scim_token), +) -> dict[str, Any]: + """SCIM PATCH /Users/{id} — 禁用 (active=false) 或更新字段。""" + db_path = await _ensure_db(request) + now = _now_iso() + 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="用户不存在") + for op in payload.Operations: + if op.op.lower() == "replace": + if op.path in (None, "active"): + active = ( + bool(op.value) + if not isinstance(op.value, dict) + else bool(op.value.get("active", True)) + ) + await db.execute( + "UPDATE users SET is_active = ?, updated_at = ? WHERE id = ?", + (1 if active else 0, now, user_id), + ) + elif op.path == "userName": + await db.execute( + "UPDATE users SET username = ?, updated_at = ? WHERE id = ?", + (str(op.value)[:64], now, user_id), + ) + await db.commit() + cursor = await db.execute("SELECT * FROM users WHERE id = ?", (user_id,)) + row = await cursor.fetchone() + assert row is not None + 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 + + +@router.get("/Users") +async def list_users( + request: Request, + filter: str | None = Query(default=None, alias="filter"), + 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]: + """SCIM GET /Users — 列表 / 过滤。 + + ponytail: filter 仅支持最简形式 ``userName eq "xxx"`` / ``active eq true``。 + 上限：复杂 SCIM filter 表达式需扩展为完整 AST 解析器。 + """ + db_path = await _ensure_db(request) + where = "" + params: list[Any] = [] + if filter: + # 最简 filter 解析：userName eq "x" / active eq true + if "userName eq" in filter: + val = filter.split("userName 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() + 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]) + async with aiosqlite.connect(str(db_path)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute(sql, tuple(params)) + rows = await cursor.fetchall() + resources = [_user_to_scim(user_row_to_dict(r)) for r in rows] + return SCIMListResponse( + totalResults=len(resources), + 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 4fff714..05fbd14 100644 --- a/src/agentkit/server/routes/auth.py +++ b/src/agentkit/server/routes/auth.py @@ -780,6 +780,138 @@ async def change_password( return {"changed": True, "revoked_other_sessions": revoked} +# --------------------------------------------------------------------------- +# SSO routes — OIDC redirect/callback + SAML ACS (U4) +# --------------------------------------------------------------------------- + + +class SSORedirectResponse(BaseModel): + """SSO 重定向响应 — 返回 IDP authorize URL 供前端跳转。""" + + model_config = ConfigDict(extra="forbid") + + redirect_url: str + state: str + + +async def _sso_issue_token_pair( + request: Request, user_dict: dict[str, object] +) -> TokenResponse: + """SSO 登录成功后签发 JWT pair + 创建 session（复用 login 流程）。""" + db_path = await _ensure_db(request) + secret = _resolve_jwt_secret(request) + svc: SessionService = get_session_service() + info = _client_info(request) + + pre_session = await svc.create( + SessionCreate( + user_id=str(user_dict["id"]), + refresh_token="__pending_sso__", + device_fingerprint=info["fingerprint"], + device_label=info["label"], + ip=info["ip"], + user_agent=info["user_agent"], + auth_provider="sso", + ttl_seconds=_refresh_ttl_seconds(False), + ) + ) + token_pair = create_token_pair( + user_id=str(user_dict["id"]), + username=str(user_dict["username"]), + role=str(user_dict["role"]), + secret=secret, + session_id=pre_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"])), + ) + await db.commit() + return TokenResponse( + access_token=token_pair.access_token, + refresh_token=token_pair.refresh_token, + expires_in=int(ACCESS_TOKEN_TTL.total_seconds()), + user=UserResponse( + id=str(user_dict["id"]), + username=str(user_dict["username"]), + email=str(user_dict["email"]), + role=str(user_dict["role"]), + is_active=bool(user_dict.get("is_active", True)), + is_terminal_authorized=bool(user_dict.get("is_terminal_authorized", False)), + is_server_terminal_authorized=bool(user_dict.get("is_server_terminal_authorized", False)), + ), + session_id=pre_session.id, + ) + + +@router.get("/oauth/{provider}/redirect", response_model=SSORedirectResponse) +async def oidc_redirect(provider: str, request: Request) -> SSORedirectResponse: + """OIDC 重定向 — 返回 IDP authorize URL（前端跳转）。""" + from agentkit.server.auth.providers.oidc import OIDCProvider + + try: + url = await OIDCProvider().get_redirect_url(provider) + except ValueError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + # state 已由 provider 写入 _state_cache，从 URL 中提取回传前端 + state = "" + if "state=" in url: + state = url.split("state=", 1)[1].split("&", 1)[0] + return SSORedirectResponse(redirect_url=url, state=state) + + +@router.get("/oauth/{provider}/callback", response_model=TokenResponse) +async def oidc_callback( + provider: str, + code: str, + state: str, + request: Request, +) -> TokenResponse: + """OIDC callback — code 换 token + userinfo + JIT 用户关联 → JWT pair。""" + await _ensure_db(request) + from agentkit.server.auth.providers.oidc import OIDCProvider + + try: + 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) + + +class SAMLACSRequest(BaseModel): + """SAML ACS 请求体。""" + + model_config = ConfigDict(extra="forbid") + + SAMLResponse: str + RelayState: str | None = None + + +@router.post("/saml/{provider}/acs", response_model=TokenResponse) +async def saml_acs( + provider: str, + payload: SAMLACSRequest, + request: Request, +) -> TokenResponse: + """SAML ACS — 消费 SAMLResponse + NameID 提取 + JIT 用户关联 → JWT pair。""" + await _ensure_db(request) + from agentkit.server.auth.providers.saml import SAMLProvider + + request_url = str(request.url) + try: + user_dict = await SAMLProvider().handle_acs( + provider, payload.SAMLResponse, request_url=request_url + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return await _sso_issue_token_pair(request, user_dict) + + # --------------------------------------------------------------------------- # Admin routes # --------------------------------------------------------------------------- @@ -889,6 +1021,123 @@ async def admin_revoke_user_session( return {"revoked": ok} +# --------------------------------------------------------------------------- +# Admin deprovisioning + RBAC role change (U4 — R1b / KTD-8) +# --------------------------------------------------------------------------- + + +class DeprovisionRequest(BaseModel): + """手动 deprovisioning 请求体。""" + + model_config = ConfigDict(extra="forbid") + + reason: str | None = None + break_glass: bool = False # 高权限账户 break-glass 标记 + + +@admin_router.post("/users/{user_id}/deprovision") +async def deprovision_user( + user_id: str, + payload: DeprovisionRequest, + request: Request, + admin: dict[str, Any] = Depends(_require_admin), +) -> dict[str, Any]: + """手动 deprovisioning (R1b) — operator/admin RBAC 强制。 + + 1. 禁用本地用户 (is_active=0) + 2. 撤销该用户所有活跃 session + 3. 记录审计 (actor/target/reason/source/timestamp)，接入 OTel + 4. break_glass=True 时记录 source=break_glass（高权限账户应急通道） + """ + db_path = await _ensure_db(request) + 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="User not found") + await db.execute( + "UPDATE users SET is_active = 0, updated_at = ? WHERE id = ?", + (_now_iso(), user_id), + ) + await db.commit() + # 撤销所有 session + svc: SessionService = get_session_service() + revoked = await svc.revoke_all_for_user(user_id, reason="admin_revoked") + # 审计记录 + from agentkit.server.auth.audit import ( + SOURCE_BREAKGLASS, + SOURCE_MANUAL, + get_audit_service, + ) + + source = SOURCE_BREAKGLASS if payload.break_glass else SOURCE_MANUAL + await get_audit_service().record_deprovision( + actor_user_id=str(admin.get("user_id") or ""), + actor_username=str(admin.get("username") or ""), + target_user_id=user_id, + target_username=str(row["username"]), + reason=payload.reason, + source=source, + ) + return {"deprovisioned": True, "revoked_sessions": revoked} + + +class RoleChangeRequest(BaseModel): + """RBAC 角色变更请求体。""" + + model_config = ConfigDict(extra="forbid") + + role: str # member / operator / admin + reason: str | None = None + + +@admin_router.post("/users/{user_id}/role") +async def change_user_role( + user_id: str, + payload: RoleChangeRequest, + request: Request, + admin: dict[str, Any] = Depends(_require_admin), +) -> dict[str, Any]: + """RBAC 角色变更 (KTD-8) — admin only。 + + 记录 actor/target/role_before/role_after/reason/source/timestamp，接入 OTel 审计。 + 变更后撤销该用户所有现有 session（强制重新登录以刷新 JWT role claim）。 + """ + if payload.role not in ("member", "operator", "admin"): + raise HTTPException(status_code=400, detail="role 必须是 member/operator/admin 之一") + db_path = await _ensure_db(request) + 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="User not found") + role_before = str(row["role"]) + await db.execute( + "UPDATE users SET role = ?, updated_at = ? WHERE id = ?", + (payload.role, _now_iso(), user_id), + ) + await db.commit() + # 审计记录 + from agentkit.server.auth.audit import get_audit_service + + await get_audit_service().record_role_change( + actor_user_id=str(admin.get("user_id") or ""), + actor_username=str(admin.get("username") or ""), + target_user_id=user_id, + target_username=str(row["username"]), + role_before=role_before, + role_after=payload.role, + reason=payload.reason, + source="manual", + ) + # 撤销所有 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} + + # Backwards-compat /me endpoint (kept for the existing frontend) @router.get("/me", response_model=UserResponse) async def me( diff --git a/tests/unit/admin/test_models.py b/tests/unit/admin/test_models.py index ff97ef0..fa19a0f 100644 --- a/tests/unit/admin/test_models.py +++ b/tests/unit/admin/test_models.py @@ -120,9 +120,9 @@ async def _list_table_names(db: aiosqlite.Connection) -> set[str]: class TestSchemaVersion: - def test_schema_version_is_v4(self): - """V4 adds the skill_states table (U6 — Admin Console).""" - assert _SCHEMA_VERSION == 4 + def test_schema_version_is_v5(self): + """V5 adds user_idp_links + auth_audit_log (U4 SSO 集成).""" + assert _SCHEMA_VERSION == 5 def test_sqlalchemy_model_table_names(self): assert DepartmentModel.__tablename__ == "departments" @@ -163,13 +163,13 @@ class TestInitAuthDbTables: tables = await _list_table_names(db) assert "department_quotas" in tables - async def test_records_schema_version_4_in_auth_meta(self, fresh_db: Path): + async def test_records_schema_version_5_in_auth_meta(self, fresh_db: Path): async with aiosqlite.connect(str(fresh_db)) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT value FROM auth_meta WHERE key='schema_version'") row = await cursor.fetchone() assert row is not None - assert row["value"] == "4" + assert row["value"] == "5" assert row["value"] == str(_SCHEMA_VERSION) async def test_init_auth_db_is_idempotent(self, tmp_path: Path): diff --git a/tests/unit/auth/test_models.py b/tests/unit/auth/test_models.py index a2296e7..31a35a8 100644 --- a/tests/unit/auth/test_models.py +++ b/tests/unit/auth/test_models.py @@ -118,9 +118,9 @@ async def _list_index_names(db: aiosqlite.Connection, table: str) -> set[str]: class TestSchemaVersion: - def test_schema_version_is_v4(self): - """The current schema version is 4 (V4 adds skill_states table).""" - assert _SCHEMA_VERSION == 4 + def test_schema_version_is_v5(self): + """The current schema version is 5 (V5 adds user_idp_links + auth_audit_log, U4 SSO).""" + assert _SCHEMA_VERSION == 5 def test_sqlalchemy_model_table_name(self): assert AuthSessionModel.__tablename__ == "auth_sessions" diff --git a/tests/unit/test_auth_audit.py b/tests/unit/test_auth_audit.py new file mode 100644 index 0000000..ec09590 --- /dev/null +++ b/tests/unit/test_auth_audit.py @@ -0,0 +1,208 @@ +"""Audit service 单元测试 (U4 — RBAC + deprovisioning 审计, KTD-8)。 + +验证： +- RBAC 角色变更审计记录（actor/target/role_before/role_after） +- Deprovisioning 审计记录（actor/target/reason/source） +- 审计记录持久化到 auth_audit_log 表 +- 按目标用户查询审计历史 +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 + +import aiosqlite +import pytest + +from agentkit.server.auth.audit import ( + AuditService, + EVENT_DEPROVISION, + EVENT_ROLE_CHANGE, + SOURCE_BREAKGLASS, + SOURCE_MANUAL, + get_audit_service, + set_audit_service, +) +from agentkit.server.auth.models import audit_log_row_to_dict, init_auth_db + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +async def auth_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + db_path = tmp_path / "auth.db" + await init_auth_db(db_path) + monkeypatch.setenv("AGENTKIT_AUTH_DB", str(db_path)) + return db_path + + +@pytest.fixture +def audit_service(auth_db: Path) -> AuditService: + svc = AuditService(db_path=auth_db) + set_audit_service(svc) + return svc + + +@pytest.fixture +async def seed_users(auth_db: Path) -> dict[str, str]: + """插入两个用户（admin + target）用于审计测试。""" + now = datetime.now(timezone.utc).isoformat() + admin_id = str(uuid4()) + target_id = str(uuid4()) + async with aiosqlite.connect(str(auth_db)) as db: + await db.execute( + "INSERT INTO users (id, username, email, password_hash, role, is_active, " + "is_terminal_authorized, is_server_terminal_authorized, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (admin_id, "admin_alice", "alice@example.com", "hash", "admin", 1, 0, 0, now, now), + ) + await db.execute( + "INSERT INTO users (id, username, email, password_hash, role, is_active, " + "is_terminal_authorized, is_server_terminal_authorized, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (target_id, "target_bob", "bob@example.com", "hash", "member", 1, 0, 0, now, now), + ) + await db.commit() + return {"admin_id": admin_id, "target_id": target_id} + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestRoleChangeAudit: + async def test_records_role_change( + self, audit_service: AuditService, seed_users: dict[str, str] + ) -> None: + record = await audit_service.record_role_change( + actor_user_id=seed_users["admin_id"], + actor_username="admin_alice", + target_user_id=seed_users["target_id"], + target_username="target_bob", + role_before="member", + role_after="operator", + reason="晋升", + source=SOURCE_MANUAL, + ) + assert record["event_type"] == EVENT_ROLE_CHANGE + assert record["role_before"] == "member" + assert record["role_after"] == "operator" + assert record["actor_username"] == "admin_alice" + assert record["target_username"] == "target_bob" + assert record["reason"] == "晋升" + + async def test_persists_to_db( + self, audit_service: AuditService, auth_db: Path, seed_users: dict[str, str] + ) -> None: + await audit_service.record_role_change( + actor_user_id=seed_users["admin_id"], + actor_username="admin_alice", + target_user_id=seed_users["target_id"], + target_username="target_bob", + role_before="member", + role_after="admin", + reason="提升管理员", + ) + async with aiosqlite.connect(str(auth_db)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT * FROM auth_audit_log WHERE event_type = ? AND target_user_id = ?", + (EVENT_ROLE_CHANGE, seed_users["target_id"]), + ) + rows = await cursor.fetchall() + assert len(rows) == 1 + row = audit_log_row_to_dict(rows[0]) + assert row["role_before"] == "member" + assert row["role_after"] == "admin" + + +class TestDeprovisionAudit: + async def test_records_deprovision( + self, audit_service: AuditService, seed_users: dict[str, str] + ) -> None: + record = await audit_service.record_deprovision( + actor_user_id=seed_users["admin_id"], + actor_username="admin_alice", + target_user_id=seed_users["target_id"], + target_username="target_bob", + reason="离职", + source=SOURCE_MANUAL, + ) + assert record["event_type"] == EVENT_DEPROVISION + assert record["reason"] == "离职" + assert record["source"] == SOURCE_MANUAL + + async def test_break_glass_source( + self, audit_service: AuditService, seed_users: dict[str, str] + ) -> None: + """break_glass 应急通道 — source 记录为 break_glass。""" + record = await audit_service.record_deprovision( + actor_user_id=seed_users["admin_id"], + actor_username="admin_alice", + target_user_id=seed_users["target_id"], + target_username="target_bob", + reason="应急禁用", + source=SOURCE_BREAKGLASS, + ) + assert record["source"] == SOURCE_BREAKGLASS + + async def test_persists_to_db( + self, audit_service: AuditService, auth_db: Path, seed_users: dict[str, str] + ) -> None: + await audit_service.record_deprovision( + actor_user_id=seed_users["admin_id"], + actor_username="admin_alice", + target_user_id=seed_users["target_id"], + target_username="target_bob", + reason="审计测试", + ) + async with aiosqlite.connect(str(auth_db)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT * FROM auth_audit_log WHERE event_type = ?", + (EVENT_DEPROVISION,), + ) + rows = await cursor.fetchall() + assert len(rows) == 1 + + +class TestListForTarget: + async def test_lists_audit_history( + self, audit_service: AuditService, seed_users: dict[str, str] + ) -> None: + """按目标用户查询审计历史（admin 视图）。""" + target_id = seed_users["target_id"] + await audit_service.record_role_change( + actor_user_id=seed_users["admin_id"], + actor_username="admin_alice", + target_user_id=target_id, + target_username="target_bob", + role_before="member", + role_after="operator", + reason="第一次", + ) + await audit_service.record_deprovision( + actor_user_id=seed_users["admin_id"], + actor_username="admin_alice", + target_user_id=target_id, + target_username="target_bob", + reason="第二次", + ) + records = await audit_service.list_for_target(target_id) + assert len(records) == 2 + # 按 created_at DESC 排序 — deprovision 在前 + assert records[0]["event_type"] == EVENT_DEPROVISION + assert records[1]["event_type"] == EVENT_ROLE_CHANGE + + +class TestGetAuditService: + def test_singleton(self) -> None: + svc1 = get_audit_service() + svc2 = get_audit_service() + assert svc1 is svc2 diff --git a/tests/unit/test_oidc_provider.py b/tests/unit/test_oidc_provider.py new file mode 100644 index 0000000..a7ebe86 --- /dev/null +++ b/tests/unit/test_oidc_provider.py @@ -0,0 +1,240 @@ +"""OIDC provider 单元测试 (U4 SSO 集成)。 + +验证： +- redirect URL 生成（含 state） +- callback 用户 JIT 创建按 sub 关联 +- 禁止邮箱合并 (R1a) +- 相同 sub 复用现有用户 +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import aiosqlite +import pytest + +from agentkit.server.auth.idp_registry import ( + IDPConfig, + IDPRegistry, + OIDCConfig, +) +from agentkit.server.auth.models import init_auth_db +from agentkit.server.auth.providers.oidc import OIDCProvider, get_state_cache + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def oidc_idp(monkeypatch: pytest.MonkeyPatch) -> IDPRegistry: + """注册一个测试 OIDC IDP 并替换全局 registry。""" + registry = IDPRegistry() + registry.register( + IDPConfig( + name="test_oidc", + type="oidc", + display_name="Test OIDC", + oidc=OIDCConfig( + client_id="test-client-id", + client_secret="test-client-secret", + server_metadata_url="https://idp.test/authorize", + redirect_uri="https://app.test/api/v1/auth/oauth/test_oidc/callback", + scope="openid email profile", + ), + ) + ) + monkeypatch.setattr("agentkit.server.auth.providers.oidc.get_idp_registry", lambda: registry) + return registry + + +@pytest.fixture +async def auth_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """创建临时 auth DB 并指向 AGENTKIT_AUTH_DB。""" + db_path = tmp_path / "auth.db" + await init_auth_db(db_path) + monkeypatch.setenv("AGENTKIT_AUTH_DB", str(db_path)) + return db_path + + +class _FakeOAuth2Client: + """替代 AsyncOAuth2Client 的假 client — 不发 HTTP。""" + + def __init__(self, *args: Any, **kwargs: Any) -> None: + self.token = {"access_token": "fake-access-token", "id_token": None} + + async def __aenter__(self) -> "_FakeOAuth2Client": + return self + + async def __aexit__(self, *args: Any) -> None: + pass + + def create_authorization_url(self, url: str, **kwargs: Any) -> tuple[str, dict[str, Any]]: + """构造假 authorize URL（含 state / scope 参数）。""" + state = kwargs.get("state", "") + scope = kwargs.get("scope", "") + client_id = kwargs.get("client_id", "test-client-id") + redirect_uri = kwargs.get("redirect_uri", "") + return ( + f"{url}?response_type=code&client_id={client_id}" + f"&redirect_uri={redirect_uri}&scope={scope}&state={state}", + {"state": state, "url": url}, + ) + + async def fetch_token(self, *args: Any, **kwargs: Any) -> dict[str, Any]: + return self.token + + async def get(self, url: str) -> Any: + class _Resp: + status_code = 200 + + def json(self) -> dict[str, Any]: + return { + "sub": "oidc-sub-123", + "email": "sso_user@example.com", + "name": "SSO User", + } + + return _Resp() + + +@pytest.fixture +def fake_oauth_client(monkeypatch: pytest.MonkeyPatch) -> None: + """替换 oidc 模块的 AsyncOAuth2Client 为假实现。""" + monkeypatch.setattr("agentkit.server.auth.providers.oidc.AsyncOAuth2Client", _FakeOAuth2Client) + + +@pytest.fixture(autouse=True) +def _reset_state_cache() -> None: + """每个测试前清空 state 缓存。""" + cache = get_state_cache() + cache._store.clear() + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestGetRedirectUrl: + async def test_generates_url_with_state(self, oidc_idp: IDPRegistry) -> None: + url = await OIDCProvider().get_redirect_url("test_oidc") + assert "https://idp.test/authorize" in url + assert "state=" in url + assert "client_id=test-client-id" in url + + async def test_unknown_provider_raises(self, oidc_idp: IDPRegistry) -> None: + with pytest.raises(ValueError, match="未配置"): + await OIDCProvider().get_redirect_url("nonexistent") + + async def test_state_cached_for_csrf(self, oidc_idp: IDPRegistry) -> None: + url = await OIDCProvider().get_redirect_url("test_oidc") + state = url.split("state=", 1)[1].split("&", 1)[0] + # state 应在缓存中（consume 验证） + assert get_state_cache().consume(state) is True + + +class TestHandleCallback: + async def test_creates_user_by_sub( + self, oidc_idp: IDPRegistry, auth_db: Path, fake_oauth_client: None + ) -> None: + """首次登录 JIT 创建用户，按 sub 关联到 user_idp_links。""" + # 预置 state + provider = OIDCProvider() + url = await provider.get_redirect_url("test_oidc") + state = url.split("state=", 1)[1].split("&", 1)[0] + + user = await provider.handle_callback("test_oidc", "fake-code", state) + assert user["username"] == "SSO User" + assert user["email"] == "sso_user@example.com" + + # 验证 user_idp_links 按 sub 关联 + async with aiosqlite.connect(str(auth_db)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT * FROM user_idp_links WHERE idp_name = ? AND idp_subject = ?", + ("test_oidc", "oidc-sub-123"), + ) + link = await cursor.fetchone() + assert link is not None + assert link["user_id"] == user["id"] + + async def test_same_sub_reuses_user( + self, oidc_idp: IDPRegistry, auth_db: Path, fake_oauth_client: None + ) -> None: + """相同 sub 第二次登录复用现有用户，不重复创建。""" + provider = OIDCProvider() + # 第一次登录 + url1 = await provider.get_redirect_url("test_oidc") + state1 = url1.split("state=", 1)[1].split("&", 1)[0] + user1 = await provider.handle_callback("test_oidc", "code1", state1) + + # 第二次登录（相同 sub） + url2 = await provider.get_redirect_url("test_oidc") + state2 = url2.split("state=", 1)[1].split("&", 1)[0] + user2 = await provider.handle_callback("test_oidc", "code2", state2) + + assert user1["id"] == user2["id"] + + async def test_no_email_merge( + self, + oidc_idp: IDPRegistry, + auth_db: Path, + fake_oauth_client: None, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """R1a: 不同 sub 但相同 email — 禁止合并，创建新用户。""" + # 预创建一个本地用户，email 与 SSO userinfo 相同 + now = datetime.now(timezone.utc).isoformat() + local_id = str(uuid.uuid4()) + async with aiosqlite.connect(str(auth_db)) as db: + await db.execute( + "INSERT INTO users (id, username, email, password_hash, role, is_active, " + "is_terminal_authorized, is_server_terminal_authorized, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + local_id, + "local_user", + "sso_user@example.com", + "$2b$12$dummyhash", + "member", + 1, + 0, + 0, + now, + now, + ), + ) + await db.commit() + + # SSO 用不同 sub（_FakeOAuth2Client 返回 sub=oidc-sub-123） + provider = OIDCProvider() + url = await provider.get_redirect_url("test_oidc") + state = url.split("state=", 1)[1].split("&", 1)[0] + sso_user = await provider.handle_callback("test_oidc", "code", state) + + # 应创建新用户，而非复用 local_user + assert sso_user["id"] != local_id + # R1a: 邮箱冲突时不合并本地用户，改用 fallback 邮箱（保留 UNIQUE 约束） + assert sso_user["email"] != "sso_user@example.com" + assert sso_user["email"].endswith("@sso.test_oidc.local") + + async def test_invalid_state_raises( + self, oidc_idp: IDPRegistry, auth_db: Path, fake_oauth_client: None + ) -> None: + """无效 state（CSRF 校验失败）应拒绝。""" + with pytest.raises(ValueError, match="state"): + await OIDCProvider().handle_callback("test_oidc", "code", "invalid-state") + + +class TestAuthenticateNotImplemented: + async def test_raises(self) -> None: + from agentkit.server.auth.providers.exceptions import ProviderNotImplemented + + with pytest.raises(ProviderNotImplemented): + await OIDCProvider().authenticate(username="x", password="y") diff --git a/tests/unit/test_saml_provider.py b/tests/unit/test_saml_provider.py new file mode 100644 index 0000000..6d41cbd --- /dev/null +++ b/tests/unit/test_saml_provider.py @@ -0,0 +1,196 @@ +"""SAML 2.0 provider 单元测试 (U4 SSO 集成)。 + +验证： +- ACS NameID 提取 + JIT 用户创建 +- 按 NameID 关联（R1a: 禁止邮箱合并） +- 相同 NameID 复用现有用户 +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import aiosqlite +import pytest + +from agentkit.server.auth.idp_registry import ( + IDPConfig, + IDPRegistry, + SAMLConfig, +) +from agentkit.server.auth.models import init_auth_db +from agentkit.server.auth.providers.saml import SAMLProvider + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def saml_idp(monkeypatch: pytest.MonkeyPatch) -> IDPRegistry: + """注册一个测试 SAML IDP 并替换全局 registry。""" + registry = IDPRegistry() + registry.register( + IDPConfig( + name="test_saml", + type="saml", + display_name="Test SAML", + saml=SAMLConfig( + entity_id="https://idp.test/entity", + sso_url="https://idp.test/sso", + idp_cert="-----BEGIN CERTIFICATE-----\nfakecert\n-----END CERTIFICATE-----", + sp_entity_id="https://app.test/sp", + acs_url="https://app.test/api/v1/auth/saml/test_saml/acs", + ), + ) + ) + monkeypatch.setattr("agentkit.server.auth.providers.saml.get_idp_registry", lambda: registry) + return registry + + +@pytest.fixture +async def auth_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + db_path = tmp_path / "auth.db" + await init_auth_db(db_path) + monkeypatch.setenv("AGENTKIT_AUTH_DB", str(db_path)) + return db_path + + +@pytest.fixture +def mock_saml_parse(monkeypatch: pytest.MonkeyPatch) -> None: + """替换 _process_saml 为假实现 — 返回固定 NameID + attributes。""" + + def _fake_process( + self: SAMLProvider, req: dict, cfg: Any, provider_name: str + ) -> tuple[str, dict[str, Any]]: + return "saml-nameid-456", {"email": "saml_user@example.com", "name": "SAML User"} + + monkeypatch.setattr(SAMLProvider, "_process_saml", _fake_process) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestHandleAcs: + async def test_creates_user_by_nameid( + self, + saml_idp: IDPRegistry, + auth_db: Path, + mock_saml_parse: None, + ) -> None: + """ACS 消费 SAMLResponse + NameID 提取 + JIT 用户创建。""" + user = await SAMLProvider().handle_acs("test_saml", "fake-saml-response") + assert user["username"] == "SAML User" + assert user["email"] == "saml_user@example.com" + + # 验证 user_idp_links 按 NameID 关联 + async with aiosqlite.connect(str(auth_db)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT * FROM user_idp_links WHERE idp_name = ? AND idp_subject = ?", + ("test_saml", "saml-nameid-456"), + ) + link = await cursor.fetchone() + assert link is not None + assert link["idp_type"] == "saml" + assert link["user_id"] == user["id"] + + async def test_same_nameid_reuses_user( + self, + saml_idp: IDPRegistry, + auth_db: Path, + mock_saml_parse: None, + ) -> None: + """相同 NameID 第二次登录复用现有用户。""" + provider = SAMLProvider() + user1 = await provider.handle_acs("test_saml", "resp1") + user2 = await provider.handle_acs("test_saml", "resp2") + assert user1["id"] == user2["id"] + + async def test_no_email_merge( + self, + saml_idp: IDPRegistry, + auth_db: Path, + mock_saml_parse: None, + ) -> None: + """R1a: 不同 NameID 相同 email — 禁止合并。""" + now = datetime.now(timezone.utc).isoformat() + local_id = str(uuid.uuid4()) + async with aiosqlite.connect(str(auth_db)) as db: + await db.execute( + "INSERT INTO users (id, username, email, password_hash, role, is_active, " + "is_terminal_authorized, is_server_terminal_authorized, created_at, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + local_id, + "local_saml_user", + "saml_user@example.com", + "$2b$12$dummyhash", + "member", + 1, + 0, + 0, + now, + now, + ), + ) + await db.commit() + + # SAML NameID=saml-nameid-456（不同于本地用户） + saml_user = await SAMLProvider().handle_acs("test_saml", "resp") + assert saml_user["id"] != local_id + + async def test_unknown_provider_raises( + self, + saml_idp: IDPRegistry, + auth_db: Path, + mock_saml_parse: None, + ) -> None: + with pytest.raises(ValueError, match="未配置"): + await SAMLProvider().handle_acs("nonexistent", "resp") + + async def test_missing_nameid_raises( + self, + saml_idp: IDPRegistry, + auth_db: Path, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """SAML assertion 缺少 NameID 应拒绝。""" + monkeypatch.setattr( + SAMLProvider, + "_process_saml", + lambda self, req, cfg, name: ("", {}), + ) + with pytest.raises(ValueError, match="NameID"): + await SAMLProvider().handle_acs("test_saml", "resp") + + +class TestAuthenticateNotImplemented: + async def test_raises(self) -> None: + from agentkit.server.auth.providers.exceptions import ProviderNotImplemented + + with pytest.raises(ProviderNotImplemented): + await SAMLProvider().authenticate(username="x", password="y") + + +class TestRevokeUser: + async def test_disables_local_user( + self, + saml_idp: IDPRegistry, + auth_db: Path, + mock_saml_parse: None, + ) -> None: + provider = SAMLProvider() + user = await provider.handle_acs("test_saml", "resp") + await provider.revoke_user(user["id"]) + async with aiosqlite.connect(str(auth_db)) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute("SELECT is_active FROM users WHERE id = ?", (user["id"],)) + row = await cursor.fetchone() + assert bool(row["is_active"]) is False diff --git a/tests/unit/test_scim_router.py b/tests/unit/test_scim_router.py new file mode 100644 index 0000000..6a1d6d1 --- /dev/null +++ b/tests/unit/test_scim_router.py @@ -0,0 +1,162 @@ +"""SCIM 2.0 router 单元测试 (U4 SSO 集成)。 + +验证 POST / PATCH / GET /scim/v2/Users： +- POST 创建用户 +- PATCH 禁用用户 (active=false) +- GET 列表 / 过滤 +- Bearer token 认证（缺 token / 错误 token → 401） +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from agentkit.server.auth.models import init_auth_db +from agentkit.server.auth.scim.router import router as scim_router + + +SCIM_TOKEN = "test-scim-token-secret" + + +@pytest.fixture +async def scim_app(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> FastAPI: + """创建含 SCIM router 的 TestApp + 初始化 auth DB。""" + db_path = tmp_path / "auth.db" + await init_auth_db(db_path) + monkeypatch.setenv("AGENTKIT_AUTH_DB", str(db_path)) + monkeypatch.setenv("AGENTKIT_SCIM_TOKEN", SCIM_TOKEN) + + app = FastAPI() + app.state.auth_db_path = str(db_path) + app.state.scim_token = SCIM_TOKEN + app.include_router(scim_router, prefix="/api/v1") + return app + + +@pytest.fixture +def client(scim_app: FastAPI) -> TestClient: + return TestClient(scim_app) + + +def _auth_headers(token: str = SCIM_TOKEN) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestCreateUser: + def test_creates_user(self, client: TestClient) -> None: + resp = client.post( + "/api/v1/scim/v2/Users", + json={ + "userName": "scim.user1", + "active": True, + "emails": [{"value": "scim1@example.com", "primary": True}], + }, + headers=_auth_headers(), + ) + assert resp.status_code == 201 + data = resp.json() + assert data["userName"] == "scim.user1" + assert data["active"] is True + assert data["emails"][0]["value"] == "scim1@example.com" + assert data["schemas"] == ["urn:ietf:params:scim:schemas:core:2.0:User"] + + def test_duplicate_username_conflict(self, client: TestClient) -> None: + client.post( + "/api/v1/scim/v2/Users", + json={"userName": "dup.user", "emails": [{"value": "dup@example.com"}]}, + headers=_auth_headers(), + ) + resp = client.post( + "/api/v1/scim/v2/Users", + json={"userName": "dup.user", "emails": [{"value": "other@example.com"}]}, + headers=_auth_headers(), + ) + assert resp.status_code == 409 + + def test_no_token_unauthorized(self, client: TestClient) -> None: + resp = client.post("/api/v1/scim/v2/Users", json={"userName": "x"}) + assert resp.status_code == 401 + + def test_wrong_token_unauthorized(self, client: TestClient) -> None: + resp = client.post( + "/api/v1/scim/v2/Users", + json={"userName": "x"}, + headers={"Authorization": "Bearer wrong-token"}, + ) + assert resp.status_code == 401 + + +class TestPatchUser: + def test_disable_user(self, client: TestClient) -> None: + # 先创建 + create = client.post( + "/api/v1/scim/v2/Users", + json={"userName": "patch.user", "active": True, "emails": [{"value": "p@example.com"}]}, + headers=_auth_headers(), + ) + user_id = create.json()["id"] + + # PATCH 禁用 + resp = client.patch( + f"/api/v1/scim/v2/Users/{user_id}", + json={"Operations": [{"op": "replace", "path": "active", "value": False}]}, + headers=_auth_headers(), + ) + assert resp.status_code == 200 + assert resp.json()["active"] is False + + def test_patch_unknown_user_404(self, client: TestClient) -> None: + resp = client.patch( + "/api/v1/scim/v2/Users/nonexistent-id", + json={"Operations": [{"op": "replace", "path": "active", "value": False}]}, + headers=_auth_headers(), + ) + assert resp.status_code == 404 + + +class TestListUsers: + def test_list_all(self, client: TestClient) -> None: + client.post( + "/api/v1/scim/v2/Users", + json={"userName": "list.user1", "emails": [{"value": "l1@example.com"}]}, + headers=_auth_headers(), + ) + client.post( + "/api/v1/scim/v2/Users", + json={"userName": "list.user2", "emails": [{"value": "l2@example.com"}]}, + headers=_auth_headers(), + ) + resp = client.get("/api/v1/scim/v2/Users", headers=_auth_headers()) + assert resp.status_code == 200 + data = resp.json() + assert data["totalResults"] >= 2 + assert len(data["Resources"]) >= 2 + + def test_filter_by_username(self, client: TestClient) -> None: + client.post( + "/api/v1/scim/v2/Users", + json={"userName": "filter.user", "emails": [{"value": "f@example.com"}]}, + headers=_auth_headers(), + ) + resp = client.get( + "/api/v1/scim/v2/Users", + params={"filter": 'userName eq "filter.user"'}, + headers=_auth_headers(), + ) + assert resp.status_code == 200 + data = resp.json() + assert data["totalResults"] == 1 + assert data["Resources"][0]["userName"] == "filter.user" + + def test_no_token_unauthorized(self, client: TestClient) -> None: + resp = client.get("/api/v1/scim/v2/Users") + assert resp.status_code == 401