feat(auth)!: migrate customer metadata auth

Remove local auth and user-management endpoints in favor of Keycloak-backed metadata users, project context, admin metadata APIs, and agent auth context.
2026-06-13 15:00:58 +08:00
parent 4b02118286
commit c7947a7481
28 changed files with 1562 additions and 1155 deletions
@@ -4,17 +4,13 @@
 ENVIRONMENT="production"
 NETWORK_NAME="tjwater"
 # ============================================
-# 安全配置 (必填)
+# 敏感配置加密 (必填)
 # ============================================

-# JWT 密钥 - 用于生成和验证 Token
-# 生成方式: openssl rand -hex 32
-SECRET_KEY=your-secret-key-here-change-in-production-use-openssl-rand-hex-32
-
-# 数据加密密钥 - 用于敏感数据加密
+# Fernet 格式，生产环境必须替换为独立密钥
 # 生成方式: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
-ENCRYPTION_KEY=
-DATABASE_ENCRYPTION_KEY="rJC2VqLg4KrlSq+DGJcYm869q4v5KB2dFAeuQTe0I50="
+# 用于项目数据库 DSN、GeoServer 管理密码等敏感配置
+DATABASE_ENCRYPTION_KEY="replace-with-generated-fernet-key"

 # ============================================
 # 数据库配置 (PostgreSQL)
@@ -0,0 +1,90 @@
+# TJWater Authentication and Metadata Management
+
+## Ownership
+
+Keycloak owns login identity, credentials, token issuance, and token expiry.
+TJWater metadata stores only business snapshots and authorization data:
+
+- `users.keycloak_id` is the stable identity binding.
+- `users.username`, `users.email`, and `users.last_login_at` are Keycloak claim caches.
+- `users.role`, `users.is_active`, and `users.is_superuser` control TJWater system access.
+- `user_project_membership.project_role` controls project access.
+
+The backend does not accept passwords, does not issue local JWTs, and does not
+trust frontend-supplied user IDs.
+
+## Login Snapshot Refresh
+
+Every authenticated metadata-user resolution validates the Keycloak access token
+and reads `sub`, `preferred_username` or `username`, and `email` claims. The
+backend finds `users` by `keycloak_id = sub`, rejects inactive or missing users,
+then refreshes `username`, `email`, and `last_login_at`.
+
+This keeps local display data current without changing the identity binding.
+There is no Keycloak webhook requirement; second-level user or permission sync is
+out of scope unless explicitly requested later.
+
+## Admin APIs
+
+All admin APIs require metadata admin access: `users.is_superuser = true` or
+`users.role = 'admin'`.
+
+User and membership management:
+
+- `GET /api/v1/admin/me`
+- `POST /api/v1/admin/users/sync`
+- `POST /api/v1/admin/users/sync/batch`
+- `GET /api/v1/admin/users`
+- `GET /api/v1/admin/users/{user_id}`
+- `PATCH /api/v1/admin/users/{user_id}`
+- `GET /api/v1/admin/projects/{project_id}/members`
+- `POST /api/v1/admin/projects/{project_id}/members`
+- `PATCH /api/v1/admin/projects/{project_id}/members/{user_id}`
+- `DELETE /api/v1/admin/projects/{project_id}/members/{user_id}`
+
+Project configuration:
+
+- `GET /api/v1/admin/projects`
+- `POST /api/v1/admin/projects`
+- `PATCH /api/v1/admin/projects/{project_id}`
+- `GET /api/v1/admin/projects/{project_id}/databases`
+- `PUT /api/v1/admin/projects/{project_id}/databases`
+- `DELETE /api/v1/admin/projects/{project_id}/databases/{db_role}`
+- `POST /api/v1/admin/projects/{project_id}/databases/{db_role}/health`
+
+## Secret Handling
+
+Admins submit plaintext DSNs only through HTTPS admin APIs. Operators should not
+write encrypted columns manually.
+
+- `project_databases.dsn_encrypted` is encrypted with `DATABASE_ENCRYPTION_KEY`.
+- Admin responses return only `has_dsn`.
+- Audit logs record whether a secret was updated, but never store plaintext DSNs
+  or other secrets.
+
+Generate the database encryption key with:
+
+```bash
+python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
+```
+
+Keep keys stable for the lifetime of encrypted metadata. Rotating a key requires
+decrypting with the old key and re-encrypting with the new key.
+
+## Metadata Schema Patches
+
+Apply metadata patches in order:
+
+1. `resources/sql/004_metadata_auth_management.sql`
+2. `resources/sql/005_metadata_project_configuration.sql`
+
+`004` creates Keycloak-backed metadata users and project memberships. `005`
+creates project and project database routing tables with uniqueness, role/type,
+and pool-size constraints.
+
+## Frontend System Management
+
+`/system-admin` is shown only after `GET /api/v1/admin/me` confirms metadata
+admin access. The page lets admins maintain metadata users, project members,
+projects, project database routing for `biz_data` and `iot_data`, connection
+health checks. This replaces direct SQL editing for normal project onboarding.
@@ -0,0 +1,695 @@
+from typing import List
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Path, Query, Response, status
+from sqlalchemy import text
+from sqlalchemy.engine.url import make_url
+from sqlalchemy.exc import IntegrityError, SQLAlchemyError
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from app.auth.metadata_dependencies import (
+    get_current_metadata_admin,
+    get_metadata_repository,
+)
+from app.core.audit import AuditAction, log_audit_event
+from app.domain.schemas.admin_metadata import (
+    AdminProjectCreateRequest,
+    AdminProjectResponse,
+    AdminProjectUpdateRequest,
+    MetadataUsersBatchSyncRequest,
+    MetadataUserResponse,
+    MetadataUserSyncRequest,
+    MetadataUserSyncResult,
+    MetadataUserUpdateRequest,
+    ProjectDatabaseHealthResponse,
+    ProjectDatabaseHealthRequest,
+    ProjectDatabaseResponse,
+    ProjectDatabaseUpsertRequest,
+    ProjectDbRole,
+    ProjectMemberCreateRequest,
+    ProjectMemberResponse,
+    ProjectMemberUpdateRequest,
+)
+from app.infra.db.metadb import models
+from app.infra.db.metadb.repositories.metadata_repository import (
+    MetadataRepository,
+    ProjectDbRouting,
+)
+
+router = APIRouter()
+
+
+def _project_response(project: models.Project) -> AdminProjectResponse:
+    return AdminProjectResponse(
+        project_id=project.id,
+        name=project.name,
+        code=project.code,
+        description=project.description,
+        gs_workspace=project.gs_workspace,
+        map_extent=project.map_extent,
+        status=project.status,
+        created_at=project.created_at,
+        updated_at=project.updated_at,
+    )
+
+
+def _project_database_response(
+    record: models.ProjectDatabase,
+) -> ProjectDatabaseResponse:
+    return ProjectDatabaseResponse(
+        id=record.id,
+        project_id=record.project_id,
+        db_role=record.db_role,
+        db_type=record.db_type,
+        pool_min_size=record.pool_min_size,
+        pool_max_size=record.pool_max_size,
+        has_dsn=bool(record.dsn_encrypted),
+    )
+
+
+def _database_audit_payload(payload: ProjectDatabaseUpsertRequest) -> dict:
+    return {
+        "db_role": payload.db_role,
+        "db_type": _db_type_for_role(payload.db_role),
+        "pool_min_size": payload.pool_min_size,
+        "pool_max_size": payload.pool_max_size,
+        "dsn_updated": payload.dsn is not None,
+    }
+
+
+def _to_async_sqlalchemy_url(dsn: str) -> str:
+    parsed = make_url(dsn)
+    if parsed.drivername in {"postgresql", "postgres"}:
+        parsed = parsed.set(drivername="postgresql+psycopg")
+    return parsed.render_as_string(hide_password=False)
+
+
+def _db_type_for_role(db_role: str) -> str:
+    if db_role == "iot_data":
+        return "timescaledb"
+    return "postgresql"
+
+
+def _status_for_config_value_error(exc: ValueError) -> int:
+    if "DATABASE_ENCRYPTION_KEY" in str(exc):
+        return status.HTTP_503_SERVICE_UNAVAILABLE
+    return status.HTTP_400_BAD_REQUEST
+
+
+async def _check_database_connection(routing: ProjectDbRouting) -> None:
+    engine = create_async_engine(
+        _to_async_sqlalchemy_url(routing.dsn),
+        pool_size=1,
+        max_overflow=0,
+        pool_pre_ping=True,
+    )
+    try:
+        async with engine.connect() as conn:
+            await conn.execute(text("SELECT 1"))
+    finally:
+        await engine.dispose()
+
+
+def _database_health_error_detail(exc: Exception) -> str:
+    message = str(exc)
+    lower_message = message.lower()
+    if "password authentication failed" in lower_message:
+        return "连通性测试失败：用户名或密码错误，请检查 DSN 中的账号密码。"
+    if "connection refused" in lower_message:
+        return "连通性测试失败：目标主机或端口拒绝连接，请检查地址、端口和服务状态。"
+    if "timeout" in lower_message or "timed out" in lower_message:
+        return "连通性测试失败：连接超时，请检查网络、防火墙和数据库服务状态。"
+    if "could not translate host name" in lower_message or "name or service not known" in lower_message:
+        return "连通性测试失败：数据库主机名无法解析，请检查 DSN 中的主机地址。"
+    first_line = message.splitlines()[0] if message else exc.__class__.__name__
+    return f"连通性测试失败：{first_line}"
+
+
+async def _upsert_and_audit_metadata_user(
+    payload: MetadataUserSyncRequest,
+    *,
+    current_user,
+    metadata_repo: MetadataRepository,
+    response_status: int,
+) -> MetadataUserResponse:
+    user = await metadata_repo.upsert_user_from_keycloak(
+        keycloak_id=payload.keycloak_id,
+        username=payload.username,
+        email=str(payload.email),
+        role=payload.role,
+        is_active=payload.is_active,
+    )
+    await log_audit_event(
+        action=AuditAction.UPDATE,
+        user_id=current_user.id,
+        resource_type="metadata_user",
+        resource_id=str(user.id),
+        request_data=payload.model_dump(mode="json"),
+        response_status=response_status,
+        session=metadata_repo.session,
+    )
+    return MetadataUserResponse.model_validate(user)
+
+
+@router.get("/me", response_model=MetadataUserResponse)
+async def get_metadata_admin_me(
+    current_user=Depends(get_current_metadata_admin),
+) -> MetadataUserResponse:
+    return MetadataUserResponse.model_validate(current_user)
+
+
+@router.post("/users/sync", response_model=MetadataUserResponse)
+async def sync_metadata_user(
+    payload: MetadataUserSyncRequest,
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> MetadataUserResponse:
+    try:
+        return await _upsert_and_audit_metadata_user(
+            payload,
+            current_user=current_user,
+            metadata_repo=metadata_repo,
+            response_status=status.HTTP_200_OK,
+        )
+    except IntegrityError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="User keycloak_id, username, or email conflicts with an existing user",
+        ) from exc
+    except SQLAlchemyError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"Metadata database error: {exc}",
+        ) from exc
+
+
+
+@router.post("/users/sync/batch", response_model=List[MetadataUserSyncResult])
+async def sync_metadata_users_batch(
+    payload: MetadataUsersBatchSyncRequest,
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> List[MetadataUserSyncResult]:
+    results: list[MetadataUserSyncResult] = []
+    for item in payload.users:
+        try:
+            user = await _upsert_and_audit_metadata_user(
+                item,
+                current_user=current_user,
+                metadata_repo=metadata_repo,
+                response_status=status.HTTP_200_OK,
+            )
+        except IntegrityError as exc:
+            results.append(
+                MetadataUserSyncResult(
+                    keycloak_id=item.keycloak_id,
+                    success=False,
+                    error="User keycloak_id, username, or email conflicts with an existing user",
+                )
+            )
+            await metadata_repo.session.rollback()
+        except SQLAlchemyError as exc:
+            results.append(
+                MetadataUserSyncResult(
+                    keycloak_id=item.keycloak_id,
+                    success=False,
+                    error=f"Metadata database error: {exc}",
+                )
+            )
+            await metadata_repo.session.rollback()
+        else:
+            results.append(
+                MetadataUserSyncResult(
+                    keycloak_id=item.keycloak_id,
+                    success=True,
+                    user=user,
+                )
+            )
+    return results
+
+
+@router.get("/users", response_model=List[MetadataUserResponse])
+async def list_metadata_users(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> List[MetadataUserResponse]:
+    users = await metadata_repo.list_users(skip=skip, limit=limit)
+    return [MetadataUserResponse.model_validate(user) for user in users]
+
+
+@router.get("/projects", response_model=List[AdminProjectResponse])
+async def list_admin_projects(
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> List[AdminProjectResponse]:
+    projects = await metadata_repo.list_project_records()
+    return [_project_response(project) for project in projects]
+
+
+@router.post(
+    "/projects",
+    response_model=AdminProjectResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+async def create_admin_project(
+    payload: AdminProjectCreateRequest,
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> AdminProjectResponse:
+    try:
+        project = await metadata_repo.create_project(
+            name=payload.name,
+            code=payload.code,
+            description=payload.description,
+            gs_workspace=payload.gs_workspace,
+            map_extent=payload.map_extent,
+            status=payload.status,
+        )
+    except IntegrityError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Project code or workspace conflicts with an existing project",
+        ) from exc
+    except SQLAlchemyError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"Metadata database error: {exc}",
+        ) from exc
+
+    await log_audit_event(
+        action=AuditAction.CREATE,
+        user_id=current_user.id,
+        project_id=project.id,
+        resource_type="project",
+        resource_id=str(project.id),
+        request_data=payload.model_dump(mode="json"),
+        response_status=status.HTTP_201_CREATED,
+        session=metadata_repo.session,
+    )
+    return _project_response(project)
+
+
+@router.patch(
+    "/projects/{project_id}",
+    response_model=AdminProjectResponse,
+)
+async def update_admin_project(
+    payload: AdminProjectUpdateRequest,
+    project_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> AdminProjectResponse:
+    updates = payload.model_dump(mode="json", exclude_unset=True)
+    try:
+        project = await metadata_repo.update_project(project_id, updates=updates)
+    except IntegrityError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Project code or workspace conflicts with an existing project",
+        ) from exc
+    except SQLAlchemyError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"Metadata database error: {exc}",
+        ) from exc
+    if project is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
+
+    await log_audit_event(
+        action=AuditAction.UPDATE,
+        user_id=current_user.id,
+        project_id=project.id,
+        resource_type="project",
+        resource_id=str(project.id),
+        request_data=updates,
+        response_status=status.HTTP_200_OK,
+        session=metadata_repo.session,
+    )
+    return _project_response(project)
+
+
+@router.get(
+    "/projects/{project_id}/databases",
+    response_model=List[ProjectDatabaseResponse],
+)
+async def list_project_databases(
+    project_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> List[ProjectDatabaseResponse]:
+    project = await metadata_repo.get_project_by_id(project_id)
+    if project is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
+    records = await metadata_repo.list_project_databases(project_id)
+    return [_project_database_response(record) for record in records]
+
+
+@router.put(
+    "/projects/{project_id}/databases",
+    response_model=ProjectDatabaseResponse,
+)
+async def upsert_project_database(
+    payload: ProjectDatabaseUpsertRequest,
+    project_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> ProjectDatabaseResponse:
+    project = await metadata_repo.get_project_by_id(project_id)
+    if project is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
+    try:
+        routing = (
+            ProjectDbRouting(
+                project_id=project_id,
+                db_role=payload.db_role,
+                db_type=_db_type_for_role(payload.db_role),
+                dsn=payload.dsn,
+                pool_min_size=payload.pool_min_size,
+                pool_max_size=payload.pool_max_size,
+            )
+            if payload.dsn
+            else await metadata_repo.get_project_db_routing(project_id, payload.db_role)
+        )
+        if routing is None:
+            raise ValueError("dsn is required when creating project database config")
+        await _check_database_connection(routing)
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=_status_for_config_value_error(exc),
+            detail=str(exc),
+        ) from exc
+    except Exception as exc:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=_database_health_error_detail(exc),
+        ) from exc
+
+    try:
+        record = await metadata_repo.upsert_project_database_config(
+            project_id,
+            db_role=payload.db_role,
+            db_type=_db_type_for_role(payload.db_role),
+            dsn=payload.dsn,
+            pool_min_size=payload.pool_min_size,
+            pool_max_size=payload.pool_max_size,
+        )
+    except IntegrityError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="Project database role conflicts with an existing config",
+        ) from exc
+    except SQLAlchemyError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"Metadata database error: {exc}",
+        ) from exc
+
+    await log_audit_event(
+        action=AuditAction.CONFIG_CHANGE,
+        user_id=current_user.id,
+        project_id=project_id,
+        resource_type="project_database",
+        resource_id=payload.db_role,
+        request_data=_database_audit_payload(payload),
+        response_status=status.HTTP_200_OK,
+        session=metadata_repo.session,
+    )
+    return _project_database_response(record)
+
+
+@router.delete(
+    "/projects/{project_id}/databases/{db_role}",
+    status_code=status.HTTP_204_NO_CONTENT,
+)
+async def delete_project_database(
+    project_id: UUID = Path(...),
+    db_role: ProjectDbRole = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> None:
+    removed = await metadata_repo.delete_project_database_config(project_id, db_role)
+    if not removed:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Project database config not found",
+        )
+    await log_audit_event(
+        action=AuditAction.CONFIG_CHANGE,
+        user_id=current_user.id,
+        project_id=project_id,
+        resource_type="project_database",
+        resource_id=db_role,
+        request_data={"deleted": True},
+        response_status=status.HTTP_204_NO_CONTENT,
+        session=metadata_repo.session,
+    )
+
+
+@router.post(
+    "/projects/{project_id}/databases/{db_role}/health",
+    response_model=ProjectDatabaseHealthResponse,
+)
+async def check_project_database_health(
+    response: Response,
+    project_id: UUID = Path(...),
+    db_role: ProjectDbRole = Path(...),
+    payload: ProjectDatabaseHealthRequest | None = None,
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> ProjectDatabaseHealthResponse:
+    dsn_to_test = payload.dsn if payload and payload.dsn else None
+    if dsn_to_test:
+        routing = ProjectDbRouting(
+            project_id=project_id,
+            db_role=db_role,
+            db_type=_db_type_for_role(db_role),
+            dsn=dsn_to_test,
+            pool_min_size=1,
+            pool_max_size=1,
+        )
+    else:
+        try:
+            routing = await metadata_repo.get_project_db_routing(project_id, db_role)
+        except ValueError as exc:
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail=f"Project database routing DSN is invalid: {exc}",
+            ) from exc
+        if routing is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="Project database config not found",
+            )
+
+    try:
+        await _check_database_connection(routing)
+    except Exception as exc:  # health endpoint should return diagnostic status
+        response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
+        return ProjectDatabaseHealthResponse(
+            project_id=project_id,
+            db_role=db_role,
+            db_type=routing.db_type,
+            ok=False,
+            detail=_database_health_error_detail(exc),
+        )
+    return ProjectDatabaseHealthResponse(
+        project_id=project_id,
+        db_role=db_role,
+        db_type=routing.db_type,
+        ok=True,
+        detail="连通性测试通过",
+    )
+
+
+@router.get("/users/{user_id}", response_model=MetadataUserResponse)
+async def get_metadata_user(
+    user_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> MetadataUserResponse:
+    user = await metadata_repo.get_user_by_id(user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+    return MetadataUserResponse.model_validate(user)
+
+
+@router.patch("/users/{user_id}", response_model=MetadataUserResponse)
+async def update_metadata_user(
+    payload: MetadataUserUpdateRequest,
+    user_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> MetadataUserResponse:
+    updates = payload.model_dump(mode="json", exclude_unset=True)
+    if user_id == current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Users cannot modify themselves",
+        )
+    user = await metadata_repo.update_user_admin(
+        user_id,
+        updates=updates,
+    )
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
+    await log_audit_event(
+        action=AuditAction.UPDATE,
+        user_id=current_user.id,
+        resource_type="metadata_user",
+        resource_id=str(user.id),
+        request_data=updates,
+        response_status=status.HTTP_200_OK,
+        session=metadata_repo.session,
+    )
+    return MetadataUserResponse.model_validate(user)
+
+
+@router.get(
+    "/projects/{project_id}/members",
+    response_model=List[ProjectMemberResponse],
+)
+async def list_project_members(
+    project_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> List[ProjectMemberResponse]:
+    project = await metadata_repo.get_project_by_id(project_id)
+    if project is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Project not found"
+        )
+    members = await metadata_repo.list_project_members(project_id)
+    return [ProjectMemberResponse(**member.__dict__) for member in members]
+
+
+@router.post(
+    "/projects/{project_id}/members",
+    response_model=ProjectMemberResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+async def add_project_member(
+    payload: ProjectMemberCreateRequest,
+    project_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> ProjectMemberResponse:
+    if payload.user_id == current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Users cannot modify their own project membership",
+        )
+    project = await metadata_repo.get_project_by_id(project_id)
+    if project is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Project not found"
+        )
+    user = await metadata_repo.get_user_by_id(payload.user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+    existing = await metadata_repo.get_project_membership(project_id, payload.user_id)
+    if existing is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail="User is already a project member",
+        )
+
+    membership = await metadata_repo.add_project_member(
+        project_id, payload.user_id, payload.project_role
+    )
+    await log_audit_event(
+        action=AuditAction.PERMISSION_CHANGE,
+        user_id=current_user.id,
+        project_id=project_id,
+        resource_type="project_member",
+        resource_id=str(payload.user_id),
+        request_data=payload.model_dump(mode="json"),
+        response_status=status.HTTP_201_CREATED,
+        session=metadata_repo.session,
+    )
+    return ProjectMemberResponse(
+        id=membership.id,
+        user_id=membership.user_id,
+        project_id=membership.project_id,
+        project_role=membership.project_role,
+        username=user.username,
+        email=user.email,
+        is_active=user.is_active,
+    )
+
+
+@router.patch(
+    "/projects/{project_id}/members/{user_id}",
+    response_model=ProjectMemberResponse,
+)
+async def update_project_member(
+    payload: ProjectMemberUpdateRequest,
+    project_id: UUID = Path(...),
+    user_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> ProjectMemberResponse:
+    if user_id == current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Users cannot modify their own project membership",
+        )
+    user = await metadata_repo.get_user_by_id(user_id)
+    if user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+    membership = await metadata_repo.update_project_member_role(
+        project_id, user_id, payload.project_role
+    )
+    if membership is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Project member not found"
+        )
+    await log_audit_event(
+        action=AuditAction.PERMISSION_CHANGE,
+        user_id=current_user.id,
+        project_id=project_id,
+        resource_type="project_member",
+        resource_id=str(user_id),
+        request_data=payload.model_dump(mode="json"),
+        response_status=status.HTTP_200_OK,
+        session=metadata_repo.session,
+    )
+    return ProjectMemberResponse(
+        id=membership.id,
+        user_id=membership.user_id,
+        project_id=membership.project_id,
+        project_role=membership.project_role,
+        username=user.username,
+        email=user.email,
+        is_active=user.is_active,
+    )
+
+
+@router.delete("/projects/{project_id}/members/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def remove_project_member(
+    project_id: UUID = Path(...),
+    user_id: UUID = Path(...),
+    current_user=Depends(get_current_metadata_admin),
+    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
+) -> None:
+    if user_id == current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Users cannot modify their own project membership",
+        )
+    removed = await metadata_repo.remove_project_member(project_id, user_id)
+    if not removed:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Project member not found"
+        )
+    await log_audit_event(
+        action=AuditAction.PERMISSION_CHANGE,
+        user_id=current_user.id,
+        project_id=project_id,
+        resource_type="project_member",
+        resource_id=str(user_id),
+        response_status=status.HTTP_204_NO_CONTENT,
+        session=metadata_repo.session,
+    )
@@ -0,0 +1,50 @@
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+
+from app.auth.keycloak_dependencies import get_current_keycloak_payload
+from app.auth.metadata_dependencies import get_current_metadata_user
+from app.auth.project_dependencies import (
+    ProjectContext,
+    get_project_context,
+)
+
+router = APIRouter()
+
+
+class AgentAuthContextResponse(BaseModel):
+    user_id: str
+    keycloak_sub: str
+    username: str
+    role: str
+    is_superuser: bool
+    project_id: str
+    network: str
+    project_role: str
+    token_expires_at: str | None = None
+
+
+@router.get("/agent/auth/context", response_model=AgentAuthContextResponse)
+async def get_agent_auth_context(
+    ctx: ProjectContext = Depends(get_project_context),
+    current_user=Depends(get_current_metadata_user),
+    keycloak_payload: dict = Depends(get_current_keycloak_payload),
+) -> AgentAuthContextResponse:
+    exp = keycloak_payload.get("exp")
+    token_expires_at = (
+        datetime.fromtimestamp(exp, tz=timezone.utc).isoformat()
+        if isinstance(exp, int)
+        else None
+    )
+    return AgentAuthContextResponse(
+        user_id=str(current_user.id),
+        keycloak_sub=str(current_user.keycloak_id),
+        username=current_user.username,
+        role=current_user.role,
+        is_superuser=current_user.is_superuser,
+        project_id=str(ctx.project_id),
+        network=ctx.project_code,
+        project_role=ctx.project_role,
+        token_expires_at=token_expires_at,
+    )
@@ -1,190 +0,0 @@
-from typing import Annotated
-from datetime import timedelta
-from fastapi import APIRouter, Depends, HTTPException, status
-from fastapi.security import OAuth2PasswordRequestForm
-from app.core.config import settings
-from app.core.security import create_access_token, create_refresh_token, verify_password
-from app.domain.schemas.user import UserCreate, UserResponse, UserLogin, Token
-from app.infra.db.metadb.repositories.user_repository import UserRepository
-from app.auth.dependencies import get_user_repository, get_current_active_user
-from app.domain.schemas.user import UserInDB
-import logging
-
-logger = logging.getLogger(__name__)
-
-router = APIRouter()
-
-
-@router.post(
-    "/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED
-)
-async def register(
-    user_data: UserCreate, user_repo: UserRepository = Depends(get_user_repository)
-) -> UserResponse:
-    """
-    用户注册
-
-    创建新用户账号
-    """
-    # 检查用户名和邮箱是否已存在
-    if await user_repo.user_exists(username=user_data.username):
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Username already registered",
-        )
-
-    if await user_repo.user_exists(email=user_data.email):
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered"
-        )
-
-    # 创建用户
-    try:
-        user = await user_repo.create_user(user_data)
-        if not user:
-            raise HTTPException(
-                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                detail="Failed to create user",
-            )
-        return UserResponse.model_validate(user)
-    except Exception as e:
-        logger.error(f"Error during user registration: {e}")
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Registration failed",
-        )
-
-
-@router.post("/login", response_model=Token)
-async def login(
-    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> Token:
-    """
-    用户登录（OAuth2 标准格式）
-
-    返回 JWT Access Token 和 Refresh Token
-    """
-    # 验证用户（支持用户名或邮箱登录）
-    user = await user_repo.get_user_by_username(form_data.username)
-    if not user:
-        # 尝试用邮箱登录
-        user = await user_repo.get_user_by_email(form_data.username)
-
-    if not user or not verify_password(form_data.password, user.hashed_password):
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Incorrect username or password",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-
-    if not user.is_active:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user account"
-        )
-
-    # 生成 Token
-    access_token = create_access_token(subject=user.username)
-    refresh_token = create_refresh_token(subject=user.username)
-
-    return Token(
-        access_token=access_token,
-        refresh_token=refresh_token,
-        token_type="bearer",
-        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
-    )
-
-
-@router.post("/login/simple", response_model=Token)
-async def login_simple(
-    username: str,
-    password: str,
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> Token:
-    """
-    简化版登录接口（保持向后兼容）
-
-    直接使用 username 和 password 参数
-    """
-    # 验证用户
-    user = await user_repo.get_user_by_username(username)
-    if not user:
-        user = await user_repo.get_user_by_email(username)
-
-    if not user or not verify_password(password, user.hashed_password):
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Incorrect username or password",
-        )
-
-    if not user.is_active:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user account"
-        )
-
-    # 生成 Token
-    access_token = create_access_token(subject=user.username)
-    refresh_token = create_refresh_token(subject=user.username)
-
-    return Token(
-        access_token=access_token,
-        refresh_token=refresh_token,
-        token_type="bearer",
-        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
-    )
-
-
-@router.get("/me", response_model=UserResponse)
-async def get_current_user_info(
-    current_user: UserInDB = Depends(get_current_active_user),
-) -> UserResponse:
-    """
-    获取当前登录用户信息
-    """
-    return UserResponse.model_validate(current_user)
-
-
-@router.post("/refresh", response_model=Token)
-async def refresh_token(
-    refresh_token: str, user_repo: UserRepository = Depends(get_user_repository)
-) -> Token:
-    """
-    刷新 Access Token
-
-    使用 Refresh Token 获取新的 Access Token
-    """
-    from jose import jwt, JWTError
-
-    credentials_exception = HTTPException(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Could not validate refresh token",
-        headers={"WWW-Authenticate": "Bearer"},
-    )
-
-    try:
-        payload = jwt.decode(
-            refresh_token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
-        )
-        username: str = payload.get("sub")
-        token_type: str = payload.get("type")
-
-        if username is None or token_type != "refresh":
-            raise credentials_exception
-
-    except JWTError:
-        raise credentials_exception
-
-    # 验证用户仍然存在且激活
-    user = await user_repo.get_user_by_username(username)
-    if not user or not user.is_active:
-        raise credentials_exception
-
-    # 生成新的 Access Token
-    new_access_token = create_access_token(subject=user.username)
-
-    return Token(
-        access_token=new_access_token,
-        refresh_token=refresh_token,  # 保持原 refresh token
-        token_type="bearer",
-        expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
-    )
@@ -10,7 +10,7 @@ from app.services.tjnetwork import (
    get_network_node_coords,
    get_node_coord,
 )
-from app.auth.dependencies import get_current_user as verify_token
+from app.auth.metadata_dependencies import get_current_metadata_user
 from app.infra.cache.redis_client import redis_client, encode_datetime, decode_datetime
 import msgpack

@@ -64,7 +64,7 @@ async def fastapi_get_network_in_extent(

@router.get(
    "/getnetworkgeometries/",
-    dependencies=[Depends(verify_token)],
+    dependencies=[Depends(get_current_metadata_user)],
    summary="获取完整网络几何信息",
    description="获取整个水网的所有节点、管线和SCADA点的几何信息(需要身份验证)"
 )
@@ -1,215 +0,0 @@
-"""
-用户管理 API 接口
-
-演示权限控制的使用
-"""
-
-from typing import List
-from fastapi import APIRouter, Depends, HTTPException, status, Path, Query
-from app.domain.schemas.user import UserResponse, UserUpdate, UserCreate
-from app.domain.models.role import UserRole
-from app.domain.schemas.user import UserInDB
-from app.infra.db.metadb.repositories.user_repository import UserRepository
-from app.auth.dependencies import get_user_repository, get_current_active_user
-from app.auth.permissions import get_current_admin, require_role, check_resource_owner
-
-router = APIRouter()
-
-
-@router.get(
-    "/",
-    summary="列出所有用户",
-    description="获取用户列表（仅管理员）",
-    response_model=List[UserResponse],
-)
-async def list_users(
-    skip: int = Query(0, ge=0, description="跳过的用户数"),
-    limit: int = Query(100, ge=1, le=1000, description="返回的最大用户数"),
-    current_user: UserInDB = Depends(require_role(UserRole.ADMIN)),
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> List[UserResponse]:
-    """
-    获取用户列表
-
-    获取系统中所有的用户信息（需要管理员权限）
-    """
-    users = await user_repo.get_all_users(skip=skip, limit=limit)
-    return [UserResponse.model_validate(user) for user in users]
-
-
-@router.get(
-    "/{user_id}",
-    summary="获取用户详情",
-    description="获取指定用户的详细信息",
-    response_model=UserResponse,
-)
-async def get_user(
-    user_id: int = Path(..., gt=0, description="用户ID"),
-    current_user: UserInDB = Depends(get_current_active_user),
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> UserResponse:
-    """
-    获取用户详情
-
-    管理员可查看所有用户，普通用户只能查看自己
-    """
-    # 检查权限
-    if not check_resource_owner(user_id, current_user):
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="You don't have permission to view this user",
-        )
-
-    user = await user_repo.get_user_by_id(user_id)
-    if not user:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
-        )
-
-    return UserResponse.model_validate(user)
-
-
-@router.put(
-    "/{user_id}",
-    summary="更新用户信息",
-    description="更新指定用户的信息",
-    response_model=UserResponse,
-)
-async def update_user(
-    user_id: int = Path(..., gt=0, description="用户ID"),
-    user_update: UserUpdate = None,
-    current_user: UserInDB = Depends(get_current_active_user),
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> UserResponse:
-    """
-    更新用户信息
-
-    管理员可更新所有用户，普通用户只能更新自己（且不能修改角色）
-    """
-    # 检查用户是否存在
-    target_user = await user_repo.get_user_by_id(user_id)
-    if not target_user:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
-        )
-
-    # 权限检查
-    is_owner = current_user.id == user_id
-    is_admin = UserRole(current_user.role).has_permission(UserRole.ADMIN)
-
-    if not is_owner and not is_admin:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="You don't have permission to update this user",
-        )
-
-    # 非管理员不能修改角色和激活状态
-    if not is_admin:
-        if user_update.role is not None:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="Only admins can change user roles",
-            )
-        if user_update.is_active is not None:
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="Only admins can change user active status",
-            )
-
-    # 更新用户
-    updated_user = await user_repo.update_user(user_id, user_update)
-    if not updated_user:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="Failed to update user",
-        )
-
-    return UserResponse.model_validate(updated_user)
-
-
-@router.delete("/{user_id}", summary="删除用户", description="删除指定用户（仅管理员）")
-async def delete_user(
-    user_id: int = Path(..., gt=0, description="用户ID"),
-    current_user: UserInDB = Depends(get_current_admin),
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> dict:
-    """
-    删除用户
-
-    删除指定用户（需要管理员权限，不能删除自己）
-    """
-    # 不能删除自己
-    if current_user.id == user_id:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="You cannot delete your own account",
-        )
-
-    success = await user_repo.delete_user(user_id)
-    if not success:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
-        )
-
-    return {"message": "User deleted successfully"}
-
-
-@router.post(
-    "/{user_id}/activate",
-    summary="激活用户",
-    description="激活指定用户账户（仅管理员）",
-    response_model=UserResponse,
-)
-async def activate_user(
-    user_id: int = Path(..., gt=0, description="用户ID"),
-    current_user: UserInDB = Depends(get_current_admin),
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> UserResponse:
-    """
-    激活用户
-
-    激活指定用户的账户（需要管理员权限）
-    """
-    user_update = UserUpdate(is_active=True)
-    updated_user = await user_repo.update_user(user_id, user_update)
-
-    if not updated_user:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
-        )
-
-    return UserResponse.model_validate(updated_user)
-
-
-@router.post(
-    "/{user_id}/deactivate",
-    summary="停用用户",
-    description="停用指定用户账户（仅管理员）",
-    response_model=UserResponse,
-)
-async def deactivate_user(
-    user_id: int = Path(..., gt=0, description="用户ID"),
-    current_user: UserInDB = Depends(get_current_admin),
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> UserResponse:
-    """
-    停用用户
-
-    停用指定用户的账户（需要管理员权限，不能停用自己）
-    """
-    # 不能停用自己
-    if current_user.id == user_id:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="You cannot deactivate your own account",
-        )
-
-    user_update = UserUpdate(is_active=False)
-    updated_user = await user_repo.update_user(user_id, user_update)
-
-    if not updated_user:
-        raise HTTPException(
-            status_code=status.HTTP_404_NOT_FOUND, detail="User not found"
-        )
-
-    return UserResponse.model_validate(updated_user)
@@ -1,6 +1,7 @@
 from fastapi import APIRouter
 from app.api.v1.endpoints import (
-    auth,
+    admin_metadata,
+    agent_auth,
    project,
    simulation,
    scada,
@@ -15,7 +16,6 @@ from app.api.v1.endpoints import (
    leakage,
    burst_detection,
    burst_location,
-    user_management,  # 新增：用户管理
    audit,  # 新增：审计日志
    meta,
    web_search,
@@ -54,10 +54,10 @@ from app.api.v1.endpoints.timeseries import (
 api_router = APIRouter()

 # Core Services
-api_router.include_router(auth.router, prefix="/auth", tags=["Auth"])
+api_router.include_router(agent_auth.router, tags=["Agent Auth"])
 api_router.include_router(
-    user_management.router, prefix="/users", tags=["User Management"]
-)  # 新增
+    admin_metadata.router, prefix="/admin", tags=["Metadata Admin"]
+)
 api_router.include_router(audit.router, prefix="/audit", tags=["Audit Logs"])  # 新增
 api_router.include_router(meta.router, tags=["Metadata"])
 api_router.include_router(project.router, tags=["Project"])
@@ -1,100 +0,0 @@
-from typing import Annotated, Optional
-from fastapi import Depends, HTTPException, status, Request
-from fastapi.security import OAuth2PasswordBearer
-from jose import jwt, JWTError
-from app.core.config import settings
-from app.domain.schemas.user import UserInDB, TokenPayload
-from app.infra.db.metadb.repositories.user_repository import UserRepository
-from app.infra.db.postgresql.database import Database
-
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
-
-
-# 数据库依赖
-async def get_db(request: Request) -> Database:
-    """
-    获取数据库实例
-
-    从 FastAPI app.state 中获取在启动时初始化的数据库连接
-    """
-    if not hasattr(request.app.state, "db"):
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail="Database not initialized",
-        )
-    return request.app.state.db
-
-
-async def get_user_repository(db: Database = Depends(get_db)) -> UserRepository:
-    """获取用户仓储实例"""
-    return UserRepository(db)
-
-
-async def get_current_user(
-    token: str = Depends(oauth2_scheme),
-    user_repo: UserRepository = Depends(get_user_repository),
-) -> UserInDB:
-    """
-    获取当前登录用户
-
-    从 JWT Token 中解析用户信息，并从数据库验证
-    """
-    credentials_exception = HTTPException(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Could not validate credentials",
-        headers={"WWW-Authenticate": "Bearer"},
-    )
-
-    try:
-        payload = jwt.decode(
-            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
-        )
-        username: str = payload.get("sub")
-        token_type: str = payload.get("type", "access")
-
-        if username is None:
-            raise credentials_exception
-
-        if token_type != "access":
-            raise HTTPException(
-                status_code=status.HTTP_401_UNAUTHORIZED,
-                detail="Invalid token type. Access token required.",
-                headers={"WWW-Authenticate": "Bearer"},
-            )
-
-    except JWTError:
-        raise credentials_exception
-
-    # 从数据库获取用户
-    user = await user_repo.get_user_by_username(username)
-    if user is None:
-        raise credentials_exception
-
-    return user
-
-
-async def get_current_active_user(
-    current_user: UserInDB = Depends(get_current_user),
-) -> UserInDB:
-    """
-    获取当前活跃用户（必须是激活状态）
-    """
-    if not current_user.is_active:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user"
-        )
-    return current_user
-
-
-async def get_current_superuser(
-    current_user: UserInDB = Depends(get_current_user),
-) -> UserInDB:
-    """
-    获取当前超级管理员用户
-    """
-    if not current_user.is_superuser:
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail="Not enough privileges. Superuser access required.",
-        )
-    return current_user
@@ -8,35 +8,41 @@ from jose import JWTError, jwt
 from app.core.config import settings

 oauth2_optional = OAuth2PasswordBearer(
-    tokenUrl=f"{settings.API_V1_STR}/auth/login", auto_error=False
+    tokenUrl="keycloak", auto_error=False
 )

 # logger = logging.getLogger(__name__)


-async def get_current_keycloak_sub(
+def _decode_keycloak_token(token: str) -> dict:
+    if not settings.KEYCLOAK_PUBLIC_KEY:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Keycloak public key is not configured",
+        )
+
+    key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
+
+    return jwt.decode(
+        token,
+        key,
+        algorithms=[settings.KEYCLOAK_ALGORITHM],
+        audience=settings.KEYCLOAK_AUDIENCE or None,
+    )
+
+
+async def get_current_keycloak_payload(
    token: str | None = Depends(oauth2_optional),
-) -> UUID:
+) -> dict:
    if not token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Not authenticated",
            headers={"WWW-Authenticate": "Bearer"},
        )
-    if settings.KEYCLOAK_PUBLIC_KEY:
-        key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
-        algorithms = [settings.KEYCLOAK_ALGORITHM]
-    else:
-        key = settings.SECRET_KEY
-        algorithms = [settings.ALGORITHM]

    try:
-        payload = jwt.decode(
-            token,
-            key,
-            algorithms=algorithms,
-            audience=settings.KEYCLOAK_AUDIENCE or None,
-        )
+        return _decode_keycloak_token(token)
    except JWTError as exc:
        # logger.warning("Keycloak token validation failed: %s", exc)
        raise HTTPException(
@@ -45,6 +51,10 @@ async def get_current_keycloak_sub(
            headers={"WWW-Authenticate": "Bearer"},
        ) from exc

+
+async def get_current_keycloak_sub(
+    payload: dict = Depends(get_current_keycloak_payload),
+) -> UUID:
    sub = payload.get("sub")
    if not sub:
        raise HTTPException(
@@ -64,35 +74,8 @@ async def get_current_keycloak_sub(


 async def get_current_keycloak_username(
-    token: str | None = Depends(oauth2_optional),
+    payload: dict = Depends(get_current_keycloak_payload),
 ) -> str:
-    if not token:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Not authenticated",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    if settings.KEYCLOAK_PUBLIC_KEY:
-        key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
-        algorithms = [settings.KEYCLOAK_ALGORITHM]
-    else:
-        key = settings.SECRET_KEY
-        algorithms = [settings.ALGORITHM]
-
-    try:
-        payload = jwt.decode(
-            token,
-            key,
-            algorithms=algorithms,
-            audience=settings.KEYCLOAK_AUDIENCE or None,
-        )
-    except JWTError as exc:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid token",
-            headers={"WWW-Authenticate": "Bearer"},
-        ) from exc
-
    username = payload.get("preferred_username") or payload.get("username")
    if not username:
        raise HTTPException(
@@ -6,8 +6,7 @@ from fastapi import Depends, HTTPException, status
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession

-from app.auth.keycloak_dependencies import get_current_keycloak_sub
-from app.core.config import settings
+from app.auth.keycloak_dependencies import get_current_keycloak_payload
 from app.infra.db.metadb.database import get_metadata_session
 from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository

@@ -20,10 +19,40 @@ async def get_metadata_repository(
    return MetadataRepository(session)


+def _keycloak_sub_from_payload(payload: dict) -> UUID:
+    sub = payload.get("sub")
+    if not sub:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing subject claim",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    try:
+        return UUID(str(sub))
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid subject claim",
+            headers={"WWW-Authenticate": "Bearer"},
+        ) from exc
+
+
+def _username_from_payload(payload: dict) -> str | None:
+    username = payload.get("preferred_username") or payload.get("username")
+    return str(username) if username else None
+
+
+def _email_from_payload(payload: dict) -> str | None:
+    email = payload.get("email")
+    return str(email) if email else None
+
+
 async def get_current_metadata_user(
-    keycloak_sub: UUID = Depends(get_current_keycloak_sub),
+    keycloak_payload: dict = Depends(get_current_keycloak_payload),
    metadata_repo: MetadataRepository = Depends(get_metadata_repository),
 ):
+    keycloak_sub = _keycloak_sub_from_payload(keycloak_payload)
    try:
        user = await metadata_repo.get_user_by_keycloak_id(keycloak_sub)
    except SQLAlchemyError as exc:
@@ -39,6 +68,21 @@ async def get_current_metadata_user(
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Inactive user"
        )
+    try:
+        user = await metadata_repo.refresh_user_keycloak_snapshot(
+            user,
+            username=_username_from_payload(keycloak_payload),
+            email=_email_from_payload(keycloak_payload),
+        )
+    except SQLAlchemyError as exc:
+        logger.error(
+            "Metadata DB error while refreshing current user snapshot",
+            exc_info=True,
+        )
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail=f"Metadata database error: {exc}",
+        ) from exc
    return user


@@ -1,106 +0,0 @@
-"""
-权限控制依赖项和装饰器
-
-基于角色的访问控制（RBAC）
-"""
-from typing import Callable
-from fastapi import Depends, HTTPException, status
-from app.domain.models.role import UserRole
-from app.domain.schemas.user import UserInDB
-from app.auth.dependencies import get_current_active_user
-
-def require_role(required_role: UserRole):
-    """
-    要求特定角色或更高权限
-    
-    用法:
-        @router.get("/admin-only")
-        async def admin_endpoint(user: UserInDB = Depends(require_role(UserRole.ADMIN))):
-            ...
-    
-    Args:
-        required_role: 需要的最低角色
-        
-    Returns:
-        依赖函数
-    """
-    async def role_checker(
-        current_user: UserInDB = Depends(get_current_active_user)
-    ) -> UserInDB:
-        user_role = UserRole(current_user.role)
-        
-        if not user_role.has_permission(required_role):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail=f"Insufficient permissions. Required role: {required_role.value}, "
-                       f"Your role: {user_role.value}"
-            )
-        
-        return current_user
-    
-    return role_checker
-
-# 预定义的权限检查依赖
-require_admin = require_role(UserRole.ADMIN)
-require_operator = require_role(UserRole.OPERATOR)
-require_user = require_role(UserRole.USER)
-
-def get_current_admin(
-    current_user: UserInDB = Depends(require_admin)
-) -> UserInDB:
-    """
-    获取当前管理员用户
-    
-    等同于 Depends(require_role(UserRole.ADMIN))
-    """
-    return current_user
-
-def get_current_operator(
-    current_user: UserInDB = Depends(require_operator)
-) -> UserInDB:
-    """
-    获取当前操作员用户（或更高权限）
-    
-    等同于 Depends(require_role(UserRole.OPERATOR))
-    """
-    return current_user
-
-def check_resource_owner(user_id: int, current_user: UserInDB) -> bool:
-    """
-    检查是否是资源拥有者或管理员
-    
-    Args:
-        user_id: 资源拥有者ID
-        current_user: 当前用户
-        
-    Returns:
-        是否有权限
-    """
-    # 管理员可以访问所有资源
-    if UserRole(current_user.role).has_permission(UserRole.ADMIN):
-        return True
-    
-    # 检查是否是资源拥有者
-    return current_user.id == user_id
-
-def require_owner_or_admin(user_id: int):
-    """
-    要求是资源拥有者或管理员
-    
-    Args:
-        user_id: 资源拥有者ID
-        
-    Returns:
-        依赖函数
-    """
-    async def owner_or_admin_checker(
-        current_user: UserInDB = Depends(get_current_active_user)
-    ) -> UserInDB:
-        if not check_resource_owner(user_id, current_user):
-            raise HTTPException(
-                status_code=status.HTTP_403_FORBIDDEN,
-                detail="You don't have permission to access this resource"
-            )
-        return current_user
-    
-    return owner_or_admin_checker
@@ -25,6 +25,7 @@ logger = logging.getLogger(__name__)
@dataclass(frozen=True)
 class ProjectContext:
    project_id: UUID
+    project_code: str
    user_id: UUID
    project_role: str

@@ -85,6 +86,7 @@ async def get_project_context(

    return ProjectContext(
        project_id=project.id,
+        project_code=project.code,
        user_id=user.id,
        project_role=membership_role,
    )
@@ -11,17 +11,8 @@ class Settings(BaseSettings):

    NETWORK_NAME: str = "default_network"

-    # JWT 配置
-    SECRET_KEY: str = (
-        "your-secret-key-here-change-in-production-use-openssl-rand-hex-32"
-    )
-    ALGORITHM: str = "HS256"
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
-    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
-
-    # 数据加密密钥 (使用 Fernet)
-    ENCRYPTION_KEY: str = ""  # 必须从环境变量设置
-    DATABASE_ENCRYPTION_KEY: str = ""  # project_databases.dsn_encrypted 专用
+    # 敏感配置加密密钥 (Fernet)
+    DATABASE_ENCRYPTION_KEY: str = ""

    # Database Config (PostgreSQL)
    DB_NAME: str = "tjwater"
@@ -36,6 +27,11 @@ class Settings(BaseSettings):
    TIMESCALEDB_DB_PORT: str = "5433"
    TIMESCALEDB_DB_USER: str = "postgres"
    TIMESCALEDB_DB_PASSWORD: str = "password"
+    # InfluxDB
+    INFLUXDB_URL: str = "http://localhost:8086"
+    INFLUXDB_TOKEN: str = "token"
+    INFLUXDB_ORG: str = "org"
+    INFLUXDB_BUCKET: str = "bucket"

    # Metadata Database Config (PostgreSQL)
    METADATA_DB_NAME: str = "system_hub"
@@ -54,7 +50,7 @@ class Settings(BaseSettings):
    PROJECT_TS_POOL_MIN_SIZE: int = 1
    PROJECT_TS_POOL_MAX_SIZE: int = 10

-    # Keycloak JWT (optional override)
+    # Keycloak access token verification
    KEYCLOAK_PUBLIC_KEY: str = ""
    KEYCLOAK_ALGORITHM: str = "RS256"
    KEYCLOAK_AUDIENCE: str = ""
@@ -20,10 +20,10 @@ class Encryptor:
            key: 加密密钥，如果为 None 则从环境变量读取
        """
        if key is None:
-            key_str = os.getenv("ENCRYPTION_KEY") or settings.ENCRYPTION_KEY
+            key_str = os.getenv("DATABASE_ENCRYPTION_KEY") or settings.DATABASE_ENCRYPTION_KEY
            if not key_str:
                raise ValueError(
-                    "ENCRYPTION_KEY not found in environment variables or .env. "
+                    "DATABASE_ENCRYPTION_KEY not found in environment variables or .env. "
                    "Generate one using: Encryptor.generate_key()"
                )
            key = key_str.encode()
@@ -80,15 +80,13 @@ _database_encryptor: Optional[Encryptor] = None


 def is_encryption_configured() -> bool:
-    return bool(os.getenv("ENCRYPTION_KEY") or settings.ENCRYPTION_KEY)
+    return is_database_encryption_configured()


 def is_database_encryption_configured() -> bool:
    return bool(
        os.getenv("DATABASE_ENCRYPTION_KEY")
        or settings.DATABASE_ENCRYPTION_KEY
-        or os.getenv("ENCRYPTION_KEY")
-        or settings.ENCRYPTION_KEY
    )


@@ -107,8 +105,6 @@ def get_database_encryptor() -> Encryptor:
        key_str = (
            os.getenv("DATABASE_ENCRYPTION_KEY")
            or settings.DATABASE_ENCRYPTION_KEY
-            or os.getenv("ENCRYPTION_KEY")
-            or settings.ENCRYPTION_KEY
        )
        if not key_str:
            raise ValueError(
@@ -1,95 +0,0 @@
-from datetime import datetime, timedelta, timezone
-from typing import Optional, Union, Any
-
-from jose import jwt
-from passlib.context import CryptContext
-from app.core.config import settings
-
-pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
-
-
-def _utc_now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def create_access_token(
-    subject: Union[str, Any], expires_delta: Optional[timedelta] = None
-) -> str:
-    """
-    创建 JWT Access Token
-
-    Args:
-        subject: 用户标识（通常是用户名或用户ID）
-        expires_delta: 过期时间增量
-
-    Returns:
-        JWT token 字符串
-    """
-    if expires_delta:
-        expire = _utc_now() + expires_delta
-    else:
-        expire = _utc_now() + timedelta(
-            minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
-        )
-
-    to_encode = {
-        "exp": expire,
-        "sub": str(subject),
-        "type": "access",
-        "iat": _utc_now(),
-    }
-    encoded_jwt = jwt.encode(
-        to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
-    )
-    return encoded_jwt
-
-
-def create_refresh_token(subject: Union[str, Any]) -> str:
-    """
-    创建 JWT Refresh Token（长期有效）
-
-    Args:
-        subject: 用户标识
-
-    Returns:
-        JWT refresh token 字符串
-    """
-    expire = _utc_now() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
-
-    to_encode = {
-        "exp": expire,
-        "sub": str(subject),
-        "type": "refresh",
-        "iat": _utc_now(),
-    }
-    encoded_jwt = jwt.encode(
-        to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
-    )
-    return encoded_jwt
-
-
-def verify_password(plain_password: str, hashed_password: str) -> bool:
-    """
-    验证密码
-
-    Args:
-        plain_password: 明文密码
-        hashed_password: 密码哈希
-
-    Returns:
-        是否匹配
-    """
-    return pwd_context.verify(plain_password, hashed_password)
-
-
-def get_password_hash(password: str) -> str:
-    """
-    生成密码哈希
-
-    Args:
-        password: 明文密码
-
-    Returns:
-        bcrypt 哈希字符串
-    """
-    return pwd_context.hash(password)
@@ -1,7 +1,9 @@
 """
 This module is reserved for future implementation of advanced cryptographic operations.

-Current basic encryption (Fernet) and password hashing are implemented in `app.core.encryption` and `app.core.security`.
+Current Fernet encryption helpers are implemented in `app.core.encryption`.
+Login credentials are owned by Keycloak; this backend does not hash or store
+local passwords.
 Future expansion may include:
 - Asymmetric encryption (RSA/ECC) for secure communication
 - Key management and rotation services
@@ -1,36 +0,0 @@
-from enum import Enum
-
-class UserRole(str, Enum):
-    """用户角色枚举"""
-    ADMIN = "ADMIN"        # 管理员 - 完全权限
-    OPERATOR = "OPERATOR"  # 操作员 - 可修改数据
-    USER = "USER"          # 普通用户 - 读写权限
-    VIEWER = "VIEWER"      # 观察者 - 仅查询权限
-    
-    def __str__(self):
-        return self.value
-    
-    @classmethod
-    def get_hierarchy(cls) -> dict:
-        """
-        获取角色层级（数字越大权限越高）
-        """
-        return {
-            cls.VIEWER: 1,
-            cls.USER: 2,
-            cls.OPERATOR: 3,
-            cls.ADMIN: 4,
-        }
-    
-    def has_permission(self, required_role: 'UserRole') -> bool:
-        """
-        检查当前角色是否有足够权限
-        
-        Args:
-            required_role: 需要的最低角色
-            
-        Returns:
-            True if has permission
-        """
-        hierarchy = self.get_hierarchy()
-        return hierarchy[self] >= hierarchy[required_role]
@@ -0,0 +1,134 @@
+from datetime import datetime
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+
+BusinessRole = Literal["admin", "user", "operator", "viewer"]
+ProjectRole = Literal["owner", "admin", "member", "viewer"]
+ProjectStatus = Literal["active", "inactive", "archived"]
+ProjectDbRole = Literal["biz_data", "iot_data"]
+
+
+class MetadataUserSyncRequest(BaseModel):
+    keycloak_id: UUID
+    username: str = Field(..., min_length=1, max_length=50)
+    email: str = Field(..., min_length=1, max_length=100)
+    role: BusinessRole = "user"
+    is_active: bool = True
+
+
+class MetadataUsersBatchSyncRequest(BaseModel):
+    users: list[MetadataUserSyncRequest] = Field(..., min_length=1, max_length=500)
+
+
+class MetadataUserUpdateRequest(BaseModel):
+    role: BusinessRole | None = None
+    is_active: bool | None = None
+
+
+class MetadataUserResponse(BaseModel):
+    id: UUID
+    keycloak_id: UUID
+    username: str
+    email: str
+    role: str
+    is_active: bool
+    is_superuser: bool
+    created_at: datetime
+    updated_at: datetime
+    last_login_at: datetime | None = None
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class MetadataUserSyncResult(BaseModel):
+    keycloak_id: UUID
+    user: MetadataUserResponse | None = None
+    success: bool
+    error: str | None = None
+
+
+class ProjectMemberCreateRequest(BaseModel):
+    user_id: UUID
+    project_role: ProjectRole = "viewer"
+
+
+class ProjectMemberUpdateRequest(BaseModel):
+    project_role: ProjectRole
+
+
+class ProjectMemberResponse(BaseModel):
+    id: UUID
+    user_id: UUID
+    project_id: UUID
+    project_role: str
+    username: str
+    email: str
+    is_active: bool
+
+
+class AdminProjectCreateRequest(BaseModel):
+    name: str = Field(..., min_length=1, max_length=100)
+    code: str = Field(..., min_length=1, max_length=50)
+    description: str | None = None
+    gs_workspace: str = Field(..., min_length=1, max_length=100)
+    map_extent: dict | None = None
+    status: ProjectStatus = "active"
+
+
+class AdminProjectUpdateRequest(BaseModel):
+    name: str | None = Field(default=None, min_length=1, max_length=100)
+    code: str | None = Field(default=None, min_length=1, max_length=50)
+    description: str | None = None
+    gs_workspace: str | None = Field(default=None, min_length=1, max_length=100)
+    map_extent: dict | None = None
+    status: ProjectStatus | None = None
+
+
+class AdminProjectResponse(BaseModel):
+    project_id: UUID
+    name: str
+    code: str
+    description: str | None = None
+    gs_workspace: str
+    map_extent: dict | None = None
+    status: str
+    created_at: datetime
+    updated_at: datetime
+
+
+class ProjectDatabaseUpsertRequest(BaseModel):
+    db_role: ProjectDbRole
+    dsn: str | None = Field(default=None, min_length=1)
+    pool_min_size: int = Field(default=2, ge=1)
+    pool_max_size: int = Field(default=10, ge=1)
+
+    @model_validator(mode="after")
+    def validate_pool_bounds(self):
+        if self.pool_max_size < self.pool_min_size:
+            raise ValueError("pool_max_size must be greater than or equal to pool_min_size")
+        return self
+
+
+class ProjectDatabaseResponse(BaseModel):
+    id: UUID
+    project_id: UUID
+    db_role: str
+    db_type: str
+    pool_min_size: int
+    pool_max_size: int
+    has_dsn: bool
+
+
+class ProjectDatabaseHealthRequest(BaseModel):
+    dsn: str | None = Field(default=None, min_length=1)
+
+
+class ProjectDatabaseHealthResponse(BaseModel):
+    project_id: UUID
+    db_role: str
+    db_type: str
+    ok: bool
+    detail: str
@@ -1,68 +0,0 @@
-from datetime import datetime
-from typing import Optional
-from pydantic import BaseModel, EmailStr, Field, ConfigDict
-from app.domain.models.role import UserRole
-
-# ============================================
-# Request Schemas (输入)
-# ============================================
-
-class UserCreate(BaseModel):
-    """用户注册"""
-    username: str = Field(..., min_length=3, max_length=50, 
-                         description="用户名，3-50个字符")
-    email: EmailStr = Field(..., description="邮箱地址")
-    password: str = Field(..., min_length=6, max_length=100, 
-                         description="密码，至少6个字符")
-    role: UserRole = Field(default=UserRole.USER, description="用户角色")
-
-class UserLogin(BaseModel):
-    """用户登录"""
-    username: str = Field(..., description="用户名或邮箱")
-    password: str = Field(..., description="密码")
-
-class UserUpdate(BaseModel):
-    """用户信息更新"""
-    email: Optional[EmailStr] = None
-    password: Optional[str] = Field(None, min_length=6, max_length=100)
-    role: Optional[UserRole] = None
-    is_active: Optional[bool] = None
-
-# ============================================
-# Response Schemas (输出)
-# ============================================
-
-class UserResponse(BaseModel):
-    """用户信息响应（不含密码）"""
-    id: int
-    username: str
-    email: str
-    role: UserRole
-    is_active: bool
-    is_superuser: bool
-    created_at: datetime
-    updated_at: datetime
-    
-    model_config = ConfigDict(from_attributes=True)
-
-class UserInDB(UserResponse):
-    """数据库中的用户（含密码哈希）"""
-    hashed_password: str
-
-# ============================================
-# Token Schemas
-# ============================================
-
-class Token(BaseModel):
-    """JWT Token 响应"""
-    access_token: str
-    refresh_token: Optional[str] = None
-    token_type: str = "bearer"
-    expires_in: int = Field(..., description="过期时间（秒）")
-
-class TokenPayload(BaseModel):
-    """JWT Token Payload"""
-    sub: str = Field(..., description="用户ID或用户名")
-    exp: Optional[int] = None
-    iat: Optional[int] = None
-    type: str = Field(default="access", description="token类型: access 或 refresh")
@@ -33,8 +33,6 @@ class AuditMiddleware(BaseHTTPMiddleware):

    # 需要审计的路径前缀
    AUDIT_PATHS = [
-        # "/api/v1/auth/",
-        # "/api/v1/users/",
        # "/api/v1/projects/",
        # "/api/v1/networks/",
    ]
@@ -193,20 +191,14 @@ class AuditMiddleware(BaseHTTPMiddleware):
            return None
        sub = None
        try:
-            key = (
-                settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
-                if settings.KEYCLOAK_PUBLIC_KEY
-                else settings.SECRET_KEY
-            )
-            algorithms = (
-                [settings.KEYCLOAK_ALGORITHM]
-                if settings.KEYCLOAK_PUBLIC_KEY
-                else [settings.ALGORITHM]
-            )
+            if not settings.KEYCLOAK_PUBLIC_KEY:
+                return None
+
+            key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
            payload = jwt.decode(
                token,
                key,
-                algorithms=algorithms,
+                algorithms=[settings.KEYCLOAK_ALGORITHM],
                audience=settings.KEYCLOAK_AUDIENCE or None,
            )
            sub = payload.get("sub")
@@ -221,7 +213,7 @@ class AuditMiddleware(BaseHTTPMiddleware):
                keycloak_id = UUID(sub)
                user = await repo.get_user_by_keycloak_id(keycloak_id)
            except ValueError:
-                user = await repo.get_user_by_username(sub)
+                return None
            if user and user.is_active:
                return user.id
        return None
@@ -54,9 +54,9 @@ class ProjectConnectionManager:

    def _normalize_pg_url(self, url: str) -> str:
        parsed = make_url(url)
-        if parsed.drivername == "postgresql":
+        if parsed.drivername in {"postgresql", "postgres"}:
            parsed = parsed.set(drivername="postgresql+psycopg")
-        return str(parsed)
+        return parsed.render_as_string(hide_password=False)

    async def get_pg_sessionmaker(
        self,
@@ -1,9 +1,10 @@
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from typing import Optional, List
-from uuid import UUID
+from uuid import UUID, uuid4

 from cryptography.fernet import InvalidToken
-from sqlalchemy import select
+from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession

 from app.core.encryption import (
@@ -64,6 +65,27 @@ class ProjectDetail:
    status: str


+@dataclass(frozen=True)
+class ProjectMemberSummary:
+    id: UUID
+    user_id: UUID
+    project_id: UUID
+    project_role: str
+    username: str
+    email: str
+    is_active: bool
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _encrypt_database_secret(value: str) -> str:
+    if not is_database_encryption_configured():
+        raise ValueError("DATABASE_ENCRYPTION_KEY is not configured")
+    return get_database_encryptor().encrypt(value)
+
+
 class MetadataRepository:
    """元数据访问层（system_hub）"""

@@ -82,6 +104,86 @@ class MetadataRepository:
        )
        return result.scalar_one_or_none()

+    async def get_user_by_id(self, user_id: UUID) -> Optional[models.User]:
+        result = await self.session.execute(
+            select(models.User).where(models.User.id == user_id)
+        )
+        return result.scalar_one_or_none()
+
+    async def list_users(self, skip: int = 0, limit: int = 100) -> List[models.User]:
+        result = await self.session.execute(
+            select(models.User)
+            .order_by(models.User.created_at.desc())
+            .offset(skip)
+            .limit(limit)
+        )
+        return list(result.scalars().all())
+
+    async def upsert_user_from_keycloak(
+        self,
+        *,
+        keycloak_id: UUID,
+        username: str,
+        email: str,
+        role: str,
+        is_active: bool,
+    ) -> models.User:
+        user = await self.get_user_by_keycloak_id(keycloak_id)
+        if user is None:
+            user = models.User(
+                id=uuid4(),
+                keycloak_id=keycloak_id,
+                username=username,
+                email=email,
+                role=role,
+                is_active=is_active,
+                is_superuser=False,
+            )
+            self.session.add(user)
+        else:
+            user.username = username
+            user.email = email
+            user.role = role
+            user.is_active = is_active
+        await self.session.commit()
+        await self.session.refresh(user)
+        return user
+
+    async def refresh_user_keycloak_snapshot(
+        self,
+        user: models.User,
+        *,
+        username: str | None,
+        email: str | None,
+        last_login_at: datetime | None = None,
+    ) -> models.User:
+        if username:
+            user.username = username
+        if email:
+            user.email = email
+        user.last_login_at = last_login_at or _utcnow()
+        user.updated_at = _utcnow()
+        await self.session.commit()
+        await self.session.refresh(user)
+        return user
+
+    async def update_user_admin(
+        self,
+        user_id: UUID,
+        *,
+        updates: dict,
+    ) -> Optional[models.User]:
+        user = await self.get_user_by_id(user_id)
+        if user is None:
+            return None
+        if "role" in updates:
+            user.role = updates["role"]
+        if "is_active" in updates:
+            user.is_active = updates["is_active"]
+        await self.session.commit()
+        await self.session.refresh(user)
+        return user
+
    async def get_project_by_id(self, project_id: UUID) -> Optional[models.Project]:
        result = await self.session.execute(
            select(models.Project).where(models.Project.id == project_id)
@@ -94,11 +196,66 @@ class MetadataRepository:
        )
        return result.scalar_one_or_none()

+    async def list_project_records(self) -> List[models.Project]:
+        result = await self.session.execute(
+            select(models.Project).order_by(models.Project.name)
+        )
+        return list(result.scalars().all())
+
+    async def create_project(
+        self,
+        *,
+        name: str,
+        code: str,
+        description: str | None,
+        gs_workspace: str,
+        map_extent: dict | None,
+        status: str,
+    ) -> models.Project:
+        project = models.Project(
+            id=uuid4(),
+            name=name,
+            code=code,
+            description=description,
+            gs_workspace=gs_workspace,
+            map_extent=map_extent,
+            status=status,
+            created_at=_utcnow(),
+            updated_at=_utcnow(),
+        )
+        self.session.add(project)
+        await self.session.commit()
+        await self.session.refresh(project)
+        return project
+
+    async def update_project(
+        self,
+        project_id: UUID,
+        *,
+        updates: dict,
+    ) -> Optional[models.Project]:
+        project = await self.get_project_by_id(project_id)
+        if project is None:
+            return None
+        for field in (
+            "name",
+            "code",
+            "description",
+            "gs_workspace",
+            "map_extent",
+            "status",
+        ):
+            if field in updates:
+                setattr(project, field, updates[field])
+        project.updated_at = _utcnow()
+        await self.session.commit()
+        await self.session.refresh(project)
+        return project
+
    async def get_project_detail_by_code(self, code: str) -> Optional[ProjectDetail]:
        project = await self.get_project_by_code(code)
        if not project:
            return None
-
        return ProjectDetail(
            project_id=project.id,
            name=project.name,
@@ -120,6 +277,142 @@ class MetadataRepository:
        )
        return result.scalar_one_or_none()

+    async def list_project_members(
+        self, project_id: UUID
+    ) -> List[ProjectMemberSummary]:
+        stmt = (
+            select(models.UserProjectMembership, models.User)
+            .join(models.User, models.User.id == models.UserProjectMembership.user_id)
+            .where(models.UserProjectMembership.project_id == project_id)
+            .order_by(models.User.username)
+        )
+        result = await self.session.execute(stmt)
+        return [
+            ProjectMemberSummary(
+                id=membership.id,
+                user_id=membership.user_id,
+                project_id=membership.project_id,
+                project_role=membership.project_role,
+                username=user.username,
+                email=user.email,
+                is_active=user.is_active,
+            )
+            for membership, user in result.all()
+        ]
+
+    async def get_project_membership(
+        self, project_id: UUID, user_id: UUID
+    ) -> Optional[models.UserProjectMembership]:
+        result = await self.session.execute(
+            select(models.UserProjectMembership).where(
+                models.UserProjectMembership.project_id == project_id,
+                models.UserProjectMembership.user_id == user_id,
+            )
+        )
+        return result.scalar_one_or_none()
+
+    async def add_project_member(
+        self, project_id: UUID, user_id: UUID, project_role: str
+    ) -> models.UserProjectMembership:
+        membership = models.UserProjectMembership(
+            id=uuid4(),
+            user_id=user_id,
+            project_id=project_id,
+            project_role=project_role,
+        )
+        self.session.add(membership)
+        await self.session.commit()
+        await self.session.refresh(membership)
+        return membership
+
+    async def update_project_member_role(
+        self, project_id: UUID, user_id: UUID, project_role: str
+    ) -> Optional[models.UserProjectMembership]:
+        membership = await self.get_project_membership(project_id, user_id)
+        if membership is None:
+            return None
+        membership.project_role = project_role
+        await self.session.commit()
+        await self.session.refresh(membership)
+        return membership
+
+    async def remove_project_member(self, project_id: UUID, user_id: UUID) -> bool:
+        result = await self.session.execute(
+            delete(models.UserProjectMembership).where(
+                models.UserProjectMembership.project_id == project_id,
+                models.UserProjectMembership.user_id == user_id,
+            )
+        )
+        await self.session.commit()
+        return bool(result.rowcount)
+
+    async def list_project_databases(
+        self, project_id: UUID
+    ) -> List[models.ProjectDatabase]:
+        result = await self.session.execute(
+            select(models.ProjectDatabase)
+            .where(models.ProjectDatabase.project_id == project_id)
+            .order_by(models.ProjectDatabase.db_role)
+        )
+        return list(result.scalars().all())
+
+    async def get_project_database_config(
+        self, project_id: UUID, db_role: str
+    ) -> Optional[models.ProjectDatabase]:
+        result = await self.session.execute(
+            select(models.ProjectDatabase).where(
+                models.ProjectDatabase.project_id == project_id,
+                models.ProjectDatabase.db_role == db_role,
+            )
+        )
+        return result.scalar_one_or_none()
+
+    async def upsert_project_database_config(
+        self,
+        project_id: UUID,
+        *,
+        db_role: str,
+        db_type: str,
+        dsn: str | None,
+        pool_min_size: int,
+        pool_max_size: int,
+    ) -> models.ProjectDatabase:
+        record = await self.get_project_database_config(project_id, db_role)
+        if record is None:
+            if dsn is None:
+                raise ValueError("dsn is required when creating project database config")
+            record = models.ProjectDatabase(
+                id=uuid4(),
+                project_id=project_id,
+                db_role=db_role,
+                db_type=db_type,
+                dsn_encrypted=_encrypt_database_secret(dsn),
+                pool_min_size=pool_min_size,
+                pool_max_size=pool_max_size,
+            )
+            self.session.add(record)
+        else:
+            record.db_type = db_type
+            if dsn is not None:
+                record.dsn_encrypted = _encrypt_database_secret(dsn)
+            record.pool_min_size = pool_min_size
+            record.pool_max_size = pool_max_size
+        await self.session.commit()
+        await self.session.refresh(record)
+        return record
+
+    async def delete_project_database_config(
+        self, project_id: UUID, db_role: str
+    ) -> bool:
+        result = await self.session.execute(
+            delete(models.ProjectDatabase).where(
+                models.ProjectDatabase.project_id == project_id,
+                models.ProjectDatabase.db_role == db_role,
+            )
+        )
+        await self.session.commit()
+        return bool(result.rowcount)
+
    async def get_project_db_routing(
        self, project_id: UUID, db_role: str
    ) -> Optional[ProjectDbRouting]:
@@ -1,235 +0,0 @@
-from typing import Optional, List
-from datetime import datetime
-from app.infra.db.postgresql.database import Database
-from app.domain.schemas.user import UserCreate, UserUpdate, UserInDB
-from app.domain.models.role import UserRole
-from app.core.security import get_password_hash
-import logging
-
-logger = logging.getLogger(__name__)
-
-class UserRepository:
-    """用户数据访问层"""
-    
-    def __init__(self, db: Database):
-        self.db = db
-    
-    async def create_user(self, user: UserCreate) -> Optional[UserInDB]:
-        """
-        创建新用户
-        
-        Args:
-            user: 用户创建数据
-            
-        Returns:
-            创建的用户对象
-        """
-        hashed_password = get_password_hash(user.password)
-        
-        query = """
-            INSERT INTO users (username, email, hashed_password, role, is_active, is_superuser)
-            VALUES (%(username)s, %(email)s, %(hashed_password)s, %(role)s, TRUE, FALSE)
-            RETURNING id, username, email, hashed_password, role, is_active, is_superuser, 
-                      created_at, updated_at
-        """
-        
-        try:
-            async with self.db.get_connection() as conn:
-                async with conn.cursor() as cur:
-                    await cur.execute(query, {
-                        'username': user.username,
-                        'email': user.email,
-                        'hashed_password': hashed_password,
-                        'role': user.role.value
-                    })
-                    row = await cur.fetchone()
-                    if row:
-                        return UserInDB(**row)
-        except Exception as e:
-            logger.error(f"Error creating user: {e}")
-            raise
-        
-        return None
-    
-    async def get_user_by_id(self, user_id: int) -> Optional[UserInDB]:
-        """根据ID获取用户"""
-        query = """
-            SELECT id, username, email, hashed_password, role, is_active, is_superuser,
-                   created_at, updated_at
-            FROM users
-            WHERE id = %(user_id)s
-        """
-        
-        async with self.db.get_connection() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(query, {'user_id': user_id})
-                row = await cur.fetchone()
-                if row:
-                    return UserInDB(**row)
-        
-        return None
-    
-    async def get_user_by_username(self, username: str) -> Optional[UserInDB]:
-        """根据用户名获取用户"""
-        query = """
-            SELECT id, username, email, hashed_password, role, is_active, is_superuser,
-                   created_at, updated_at
-            FROM users
-            WHERE username = %(username)s
-        """
-        
-        async with self.db.get_connection() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(query, {'username': username})
-                row = await cur.fetchone()
-                if row:
-                    return UserInDB(**row)
-        
-        return None
-    
-    async def get_user_by_email(self, email: str) -> Optional[UserInDB]:
-        """根据邮箱获取用户"""
-        query = """
-            SELECT id, username, email, hashed_password, role, is_active, is_superuser,
-                   created_at, updated_at
-            FROM users
-            WHERE email = %(email)s
-        """
-        
-        async with self.db.get_connection() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(query, {'email': email})
-                row = await cur.fetchone()
-                if row:
-                    return UserInDB(**row)
-        
-        return None
-    
-    async def get_all_users(self, skip: int = 0, limit: int = 100) -> List[UserInDB]:
-        """获取所有用户（分页）"""
-        query = """
-            SELECT id, username, email, hashed_password, role, is_active, is_superuser,
-                   created_at, updated_at
-            FROM users
-            ORDER BY created_at DESC
-            LIMIT %(limit)s OFFSET %(skip)s
-        """
-        
-        async with self.db.get_connection() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(query, {'skip': skip, 'limit': limit})
-                rows = await cur.fetchall()
-                return [UserInDB(**row) for row in rows]
-    
-    async def update_user(self, user_id: int, user_update: UserUpdate) -> Optional[UserInDB]:
-        """
-        更新用户信息
-        
-        Args:
-            user_id: 用户ID
-            user_update: 更新数据
-            
-        Returns:
-            更新后的用户对象
-        """
-        # 构建动态更新语句
-        update_fields = []
-        params = {'user_id': user_id}
-        
-        if user_update.email is not None:
-            update_fields.append("email = %(email)s")
-            params['email'] = user_update.email
-        
-        if user_update.password is not None:
-            update_fields.append("hashed_password = %(hashed_password)s")
-            params['hashed_password'] = get_password_hash(user_update.password)
-        
-        if user_update.role is not None:
-            update_fields.append("role = %(role)s")
-            params['role'] = user_update.role.value
-        
-        if user_update.is_active is not None:
-            update_fields.append("is_active = %(is_active)s")
-            params['is_active'] = user_update.is_active
-        
-        if not update_fields:
-            return await self.get_user_by_id(user_id)
-        
-        query = f"""
-            UPDATE users
-            SET {', '.join(update_fields)}, updated_at = CURRENT_TIMESTAMP
-            WHERE id = %(user_id)s
-            RETURNING id, username, email, hashed_password, role, is_active, is_superuser,
-                      created_at, updated_at
-        """
-        
-        try:
-            async with self.db.get_connection() as conn:
-                async with conn.cursor() as cur:
-                    await cur.execute(query, params)
-                    row = await cur.fetchone()
-                    if row:
-                        return UserInDB(**row)
-        except Exception as e:
-            logger.error(f"Error updating user {user_id}: {e}")
-            raise
-        
-        return None
-    
-    async def delete_user(self, user_id: int) -> bool:
-        """
-        删除用户
-        
-        Args:
-            user_id: 用户ID
-            
-        Returns:
-            是否成功删除
-        """
-        query = "DELETE FROM users WHERE id = %(user_id)s"
-        
-        try:
-            async with self.db.get_connection() as conn:
-                async with conn.cursor() as cur:
-                    await cur.execute(query, {'user_id': user_id})
-                    return cur.rowcount > 0
-        except Exception as e:
-            logger.error(f"Error deleting user {user_id}: {e}")
-            return False
-    
-    async def user_exists(self, username: str = None, email: str = None) -> bool:
-        """
-        检查用户是否存在
-        
-        Args:
-            username: 用户名
-            email: 邮箱
-            
-        Returns:
-            是否存在
-        """
-        conditions = []
-        params = {}
-        
-        if username:
-            conditions.append("username = %(username)s")
-            params['username'] = username
-        
-        if email:
-            conditions.append("email = %(email)s")
-            params['email'] = email
-        
-        if not conditions:
-            return False
-        
-        query = f"""
-            SELECT EXISTS(
-                SELECT 1 FROM users WHERE {' OR '.join(conditions)}
-            )
-        """
-        
-        async with self.db.get_connection() as conn:
-            async with conn.cursor() as cur:
-                await cur.execute(query, params)
-                result = await cur.fetchone()
-                return result['exists'] if result else False
@@ -0,0 +1,62 @@
+-- Metadata auth management schema patch.
+-- Keycloak owns login credentials; TJWater stores only business identity and access.
+
+CREATE EXTENSION IF NOT EXISTS pgcrypto;
+
+DO $$
+DECLARE
+    users_id_type text;
+BEGIN
+    SELECT data_type INTO users_id_type
+    FROM information_schema.columns
+    WHERE table_schema = 'public'
+      AND table_name = 'users'
+      AND column_name = 'id';
+
+    IF users_id_type IS NULL THEN
+        CREATE TABLE users (
+            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+            keycloak_id UUID UNIQUE NOT NULL,
+            username VARCHAR(50) UNIQUE NOT NULL,
+            email VARCHAR(100) UNIQUE NOT NULL,
+            role VARCHAR(20) DEFAULT 'user' NOT NULL,
+            is_active BOOLEAN DEFAULT TRUE NOT NULL,
+            is_superuser BOOLEAN DEFAULT FALSE NOT NULL,
+            attributes JSONB,
+            created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL,
+            updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL,
+            last_login_at TIMESTAMP WITH TIME ZONE
+        );
+    ELSIF users_id_type <> 'uuid' THEN
+        RAISE EXCEPTION
+            'Existing public.users.id is %, not uuid. Export old local users, create Keycloak accounts, then migrate to metadata UUID users before applying this patch.',
+            users_id_type;
+    END IF;
+END $$;
+
+ALTER TABLE users
+    ADD COLUMN IF NOT EXISTS keycloak_id UUID,
+    ADD COLUMN IF NOT EXISTS attributes JSONB,
+    ADD COLUMN IF NOT EXISTS last_login_at TIMESTAMP WITH TIME ZONE;
+
+ALTER TABLE users
+    ALTER COLUMN role SET DEFAULT 'user';
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_users_keycloak_id ON users(keycloak_id);
+CREATE INDEX IF NOT EXISTS idx_users_role ON users(role);
+CREATE INDEX IF NOT EXISTS idx_users_is_active ON users(is_active);
+
+CREATE TABLE IF NOT EXISTS user_project_membership (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    user_id UUID NOT NULL,
+    project_id UUID NOT NULL,
+    project_role VARCHAR(20) DEFAULT 'viewer' NOT NULL,
+    CONSTRAINT user_project_membership_role_check
+        CHECK (project_role IN ('owner', 'admin', 'member', 'viewer')),
+    CONSTRAINT user_project_membership_unique UNIQUE (user_id, project_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_user_project_membership_user_id
+    ON user_project_membership(user_id);
+CREATE INDEX IF NOT EXISTS idx_user_project_membership_project_id
+    ON user_project_membership(project_id);
@@ -0,0 +1,55 @@
+-- Metadata project configuration schema patch.
+-- Admin APIs write these tables; operators should not hand-edit encrypted values.
+
+CREATE EXTENSION IF NOT EXISTS pgcrypto;
+
+CREATE OR REPLACE FUNCTION update_updated_at_column()
+RETURNS TRIGGER AS $$
+BEGIN
+    NEW.updated_at = CURRENT_TIMESTAMP;
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE TABLE IF NOT EXISTS projects (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    name VARCHAR(100) NOT NULL,
+    code VARCHAR(50) UNIQUE NOT NULL,
+    description TEXT,
+    gs_workspace VARCHAR(100) UNIQUE NOT NULL,
+    map_extent JSONB,
+    status VARCHAR(20) DEFAULT 'active' NOT NULL,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    CONSTRAINT projects_status_check CHECK (status IN ('active', 'inactive', 'archived'))
+);
+
+CREATE INDEX IF NOT EXISTS idx_projects_status ON projects(status);
+CREATE INDEX IF NOT EXISTS idx_projects_code ON projects(code);
+
+DROP TRIGGER IF EXISTS update_projects_updated_at ON projects;
+CREATE TRIGGER update_projects_updated_at
+    BEFORE UPDATE ON projects
+    FOR EACH ROW
+    EXECUTE FUNCTION update_updated_at_column();
+
+CREATE TABLE IF NOT EXISTS project_databases (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    project_id UUID NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
+    db_role VARCHAR(20) NOT NULL,
+    db_type VARCHAR(20) NOT NULL,
+    dsn_encrypted TEXT NOT NULL,
+    pool_min_size INTEGER DEFAULT 2 NOT NULL,
+    pool_max_size INTEGER DEFAULT 10 NOT NULL,
+    CONSTRAINT project_databases_unique_role UNIQUE (project_id, db_role),
+    CONSTRAINT project_databases_role_check CHECK (db_role IN ('biz_data', 'iot_data')),
+    CONSTRAINT project_databases_type_check CHECK (db_type IN ('postgresql', 'timescaledb')),
+    CONSTRAINT project_databases_pool_check CHECK (
+        pool_min_size >= 1 AND pool_max_size >= pool_min_size
+    )
+);
+
+CREATE INDEX IF NOT EXISTS idx_project_databases_project_id
+    ON project_databases(project_id);
+CREATE INDEX IF NOT EXISTS idx_project_databases_role
+    ON project_databases(db_role);
@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+"""Build metadata user sync payloads from an old-user to Keycloak mapping CSV."""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+from pathlib import Path
+
+
+REQUIRED_COLUMNS = {"keycloak_id", "username", "email"}
+
+
+def parse_bool(value: str | None) -> bool:
+    if value is None or value == "":
+        return True
+    return value.strip().lower() not in {"0", "false", "no", "n", "disabled"}
+
+
+def build_payload(mapping_csv: Path) -> dict:
+    with mapping_csv.open(newline="", encoding="utf-8") as handle:
+        reader = csv.DictReader(handle)
+        missing = REQUIRED_COLUMNS.difference(reader.fieldnames or [])
+        if missing:
+            raise SystemExit(f"missing required CSV columns: {', '.join(sorted(missing))}")
+
+        users = []
+        for row in reader:
+            users.append(
+                {
+                    "keycloak_id": row["keycloak_id"].strip(),
+                    "username": row["username"].strip(),
+                    "email": row["email"].strip(),
+                    "role": (row.get("role") or "user").strip().lower(),
+                    "is_active": parse_bool(row.get("is_active")),
+                }
+            )
+
+    return {"users": users}
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Convert old local user mappings into a JSON body for "
+            "POST /api/v1/admin/users/sync/batch. Passwords are never migrated."
+        )
+    )
+    parser.add_argument("mapping_csv", type=Path)
+    parser.add_argument("-o", "--output", type=Path)
+    args = parser.parse_args()
+
+    payload = build_payload(args.mapping_csv)
+    content = json.dumps(payload, ensure_ascii=False, indent=2)
+    if args.output:
+        args.output.write_text(content + "\n", encoding="utf-8")
+    else:
+        print(content)
+
+
+if __name__ == "__main__":
+    main()
@@ -2,7 +2,7 @@
 """
 测试新增 API 集成

-验证新的认证、用户管理和审计日志接口是否正确集成
+验证 Keycloak metadata auth、管理接口和审计日志接口是否正确集成
 """

 import sys
@@ -17,16 +17,15 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../.
    "module_name, desc",
    [
        ("app.core.encryption", "加密模块"),
-        ("app.core.security", "安全模块"),
        ("app.core.audit", "审计模块"),
-        ("app.domain.models.role", "角色模型"),
-        ("app.domain.schemas.user", "用户Schema"),
+        ("app.domain.schemas.admin_metadata", "管理元数据Schema"),
        ("app.domain.schemas.audit", "审计Schema"),
-        ("app.auth.permissions", "权限控制"),
-        ("app.api.v1.endpoints.auth", "认证接口"),
-        ("app.api.v1.endpoints.user_management", "用户管理接口"),
+        ("app.auth.keycloak_dependencies", "Keycloak认证依赖"),
+        ("app.auth.metadata_dependencies", "元数据认证依赖"),
+        ("app.api.v1.endpoints.agent_auth", "Agent认证上下文接口"),
+        ("app.api.v1.endpoints.admin_metadata", "元数据管理接口"),
        ("app.api.v1.endpoints.audit", "审计日志接口"),
-        ("app.infra.db.metadb.repositories.user_repository", "用户仓储"),
+        ("app.infra.db.metadb.repositories.metadata_repository", "元数据仓储"),
        ("app.infra.db.metadb.repositories.audit_repository", "审计仓储"),
        ("app.infra.audit.middleware", "审计中间件"),
    ],
@@ -49,8 +48,8 @@ def test_router_configuration():
        routes = [r.path for r in api_router.routes if hasattr(r, "path")]

        # 验证基础路径是否存在
-        assert any("/auth" in r for r in routes), "缺少认证相关路由 (/auth)"
-        assert any("/users" in r for r in routes), "缺少用户管理路由 (/users)"
+        assert any("/agent/auth/context" in r for r in routes), "缺少 Agent 认证上下文路由"
+        assert any("/admin/me" in r for r in routes), "缺少元数据管理认证路由"
        assert any("/audit" in r for r in routes), "缺少审计日志路由 (/audit)"

    except Exception as e: