diff --git a/.env.example b/.env.example index 9b3b90c..0f63d17 100644 --- a/.env.example +++ b/.env.example @@ -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) diff --git a/AUTHENTICATION_AND_USER_MANAGEMENT.md b/AUTHENTICATION_AND_USER_MANAGEMENT.md new file mode 100644 index 0000000..5cad69f --- /dev/null +++ b/AUTHENTICATION_AND_USER_MANAGEMENT.md @@ -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. diff --git a/app/api/v1/endpoints/admin_metadata.py b/app/api/v1/endpoints/admin_metadata.py new file mode 100644 index 0000000..9ecd0ee --- /dev/null +++ b/app/api/v1/endpoints/admin_metadata.py @@ -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, + ) diff --git a/app/api/v1/endpoints/agent_auth.py b/app/api/v1/endpoints/agent_auth.py new file mode 100644 index 0000000..c2637de --- /dev/null +++ b/app/api/v1/endpoints/agent_auth.py @@ -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, + ) diff --git a/app/api/v1/endpoints/auth.py b/app/api/v1/endpoints/auth.py deleted file mode 100644 index 819a094..0000000 --- a/app/api/v1/endpoints/auth.py +++ /dev/null @@ -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, - ) diff --git a/app/api/v1/endpoints/network/geometry.py b/app/api/v1/endpoints/network/geometry.py index 8a99743..5adb575 100644 --- a/app/api/v1/endpoints/network/geometry.py +++ b/app/api/v1/endpoints/network/geometry.py @@ -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点的几何信息(需要身份验证)" ) diff --git a/app/api/v1/endpoints/user_management.py b/app/api/v1/endpoints/user_management.py deleted file mode 100644 index 72e40f0..0000000 --- a/app/api/v1/endpoints/user_management.py +++ /dev/null @@ -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) diff --git a/app/api/v1/router.py b/app/api/v1/router.py index 52c5431..99da712 100644 --- a/app/api/v1/router.py +++ b/app/api/v1/router.py @@ -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"]) diff --git a/app/auth/dependencies.py b/app/auth/dependencies.py deleted file mode 100644 index 3524f0a..0000000 --- a/app/auth/dependencies.py +++ /dev/null @@ -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 diff --git a/app/auth/keycloak_dependencies.py b/app/auth/keycloak_dependencies.py index 6b34936..ac43799 100644 --- a/app/auth/keycloak_dependencies.py +++ b/app/auth/keycloak_dependencies.py @@ -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( diff --git a/app/auth/metadata_dependencies.py b/app/auth/metadata_dependencies.py index 8424429..021063e 100644 --- a/app/auth/metadata_dependencies.py +++ b/app/auth/metadata_dependencies.py @@ -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 diff --git a/app/auth/permissions.py b/app/auth/permissions.py deleted file mode 100644 index 0fb8d1c..0000000 --- a/app/auth/permissions.py +++ /dev/null @@ -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 diff --git a/app/auth/project_dependencies.py b/app/auth/project_dependencies.py index 6513c93..0bcaa43 100644 --- a/app/auth/project_dependencies.py +++ b/app/auth/project_dependencies.py @@ -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, ) diff --git a/app/core/config.py b/app/core/config.py index bc51014..3aca912 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -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 = "" diff --git a/app/core/encryption.py b/app/core/encryption.py index 9b5f6c2..a14ca62 100644 --- a/app/core/encryption.py +++ b/app/core/encryption.py @@ -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( diff --git a/app/core/security.py b/app/core/security.py deleted file mode 100644 index a99e69f..0000000 --- a/app/core/security.py +++ /dev/null @@ -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) diff --git a/app/crypto/__init__.py b/app/crypto/__init__.py index 868dc25..9914618 100644 --- a/app/crypto/__init__.py +++ b/app/crypto/__init__.py @@ -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 diff --git a/app/domain/models/role.py b/app/domain/models/role.py deleted file mode 100644 index 1870bf8..0000000 --- a/app/domain/models/role.py +++ /dev/null @@ -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] diff --git a/app/domain/schemas/admin_metadata.py b/app/domain/schemas/admin_metadata.py new file mode 100644 index 0000000..98d2d57 --- /dev/null +++ b/app/domain/schemas/admin_metadata.py @@ -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 diff --git a/app/domain/schemas/user.py b/app/domain/schemas/user.py deleted file mode 100644 index 864035a..0000000 --- a/app/domain/schemas/user.py +++ /dev/null @@ -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") diff --git a/app/infra/audit/middleware.py b/app/infra/audit/middleware.py index d8f1e3e..d0774d4 100644 --- a/app/infra/audit/middleware.py +++ b/app/infra/audit/middleware.py @@ -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 diff --git a/app/infra/db/dynamic_manager.py b/app/infra/db/dynamic_manager.py index e444a6e..c78a2e5 100644 --- a/app/infra/db/dynamic_manager.py +++ b/app/infra/db/dynamic_manager.py @@ -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, diff --git a/app/infra/db/metadb/repositories/metadata_repository.py b/app/infra/db/metadb/repositories/metadata_repository.py index 27e528b..f35ef71 100644 --- a/app/infra/db/metadb/repositories/metadata_repository.py +++ b/app/infra/db/metadb/repositories/metadata_repository.py @@ -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]: diff --git a/app/infra/db/metadb/repositories/user_repository.py b/app/infra/db/metadb/repositories/user_repository.py deleted file mode 100644 index 4d975ec..0000000 --- a/app/infra/db/metadb/repositories/user_repository.py +++ /dev/null @@ -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 diff --git a/resources/sql/004_metadata_auth_management.sql b/resources/sql/004_metadata_auth_management.sql new file mode 100644 index 0000000..e3d682b --- /dev/null +++ b/resources/sql/004_metadata_auth_management.sql @@ -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); diff --git a/resources/sql/005_metadata_project_configuration.sql b/resources/sql/005_metadata_project_configuration.sql new file mode 100644 index 0000000..d3f9e12 --- /dev/null +++ b/resources/sql/005_metadata_project_configuration.sql @@ -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); diff --git a/scripts/migrate_local_users_to_metadata.py b/scripts/migrate_local_users_to_metadata.py new file mode 100644 index 0000000..55edbc3 --- /dev/null +++ b/scripts/migrate_local_users_to_metadata.py @@ -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() diff --git a/tests/api/test_api_integration.py b/tests/api/test_api_integration.py index a5afbb9..41e4e75 100755 --- a/tests/api/test_api_integration.py +++ b/tests/api/test_api_integration.py @@ -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: