"""SCIM 2.0 路由 (U4 最小子集) — handler 内联，不拆 handlers.py。 端点： - 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 import hmac import logging import os import secrets import uuid from pathlib import Path import aiosqlite from fastapi import APIRouter, Depends, HTTPException, Query, Request, status 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, user_row_to_dict, _now_iso, ) from ...auth.password import hash_password from .models import SCIMListResponse, SCIMPatchRequest, SCIMUser logger = logging.getLogger(__name__) # STAND-3: prefix 由挂载点控制（app.py / test fixture 通过 include_router(prefix="/scim/v2") 注入）。 # router 内部不设 prefix，避免双重前缀 /scim/v2/scim/v2/Users。 router = APIRouter(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: path = getattr(request.app.state, "auth_db_path", None) # API-3: 用 resolve_auth_db_path() 在调用时读取 env var， # 让测试 monkeypatch.setenv("AGENTKIT_AUTH_DB", tmp) 可见。 return Path(path) if path else _resolve_default_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() with tracer.start_span("scim.push.failure") as span: 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 # --------------------------------------------------------------------------- # 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): # STAND-3: SCIM 挂载在 /scim/v2，非 SCIM 路由走 FastAPI 默认格式 if not request.url.path.startswith("/scim/v2"): 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): if not request.url.path.startswith("/scim/v2"): 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": 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), ) -> 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 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, ), ) # 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 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) return _scim_error_response(409, "userName 或 email 已存在") except Exception as exc: # noqa: BLE001 _alert_scim_failure("create", str(exc), user_id) return _scim_error_response(500, "SCIM 创建失败") @router.patch("/Users/{user_id}") async def patch_user( user_id: str, payload: SCIMPatchRequest, request: Request, _token: str = Depends(_verify_scim_token), ) -> 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: return _scim_error_response(404, "用户不存在") prev_active = bool(row["is_active"]) # API-12: 收集未处理的 op/path，循环结束后返回 400 noTarget _skipped_ops: list[str] = [] for op in payload.Operations: op_lower = op.op.lower() if op.op else "" if op_lower not in ("replace", "add", "remove"): # API-12: 仅 replace/add/remove 受支持（当前仅 replace 有副作用） _skipped_ops.append(f"op={op.op}") continue if op_lower != "replace": # add/remove 暂未实现 — 收集而非静默忽略 _skipped_ops.append(f"op={op.op}(未实现)") continue # API-14: path==None 时遍历 value 的所有 keys（原来仅更新 active 丢数据） if op.path is None and isinstance(op.value, dict): for field_name, field_value in op.value.items(): if field_name == "active": active = bool(field_value) await db.execute( "UPDATE users SET is_active = ?, updated_at = ? WHERE id = ?", (1 if active else 0, now, user_id), ) if prev_active and not active: _pending_audit = { "target_user_id": user_id, "target_username": str(row["username"]), } elif field_name == "userName": try: await db.execute( "UPDATE users SET username = ?, updated_at = ? WHERE id = ?", (str(field_value)[:64], now, user_id), ) except aiosqlite.IntegrityError: return _scim_error_response(409, "userName 已存在") else: # API-12: 不支持的字段收集 _skipped_ops.append(f"path={field_name}") continue # path 明确指定的情况 if op.path in ("active", None): 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), ) # 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": 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 已存在") else: # API-12: 不支持的 path 收集 _skipped_ops.append(f"path={op.path}") # API-12: 有未处理的 op/path → 400 noTarget if _skipped_ops: return _scim_error_response( 400, f"SCIM PATCH 包含不支持的操作（scimType=noTarget）: {', '.join(_skipped_ops)}", ) 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) return _scim_error_response(500, "SCIM 更新失败") @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, object]: """SCIM GET /Users — 列表 / 过滤。 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[object] = [] if filter: 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_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) 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 # 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=total_results, startIndex=startIndex, itemsPerPage=len(resources), Resources=resources, ).model_dump()