from fastapi import APIRouter, Depends, HTTPException from urllib.parse import quote from pydantic import BaseModel, Field from sqlalchemy import func, or_ from sqlalchemy.orm import Session from app.core.auth import SsoPrincipal, get_principal from app.db.models import ( SupportRole, SupportRoleAssignment, SupportUser, UserWorkspaceAccess, Workspace, WorkspaceChannelMapping, ) from app.db.session import get_db from app.core.config import ABCProjectTarget, settings from app.services.access_service import AccessService from app.services.workspace_mapping_service import workspace_mapping_service router = APIRouter(prefix="/access", tags=["access"]) service = AccessService() def abc_project_target( db: Session, workspace_id: int, ) -> ABCProjectTarget | None: mapping = ( db.query(WorkspaceChannelMapping) .filter( WorkspaceChannelMapping.workspace_id == workspace_id, WorkspaceChannelMapping.is_active.is_(True), WorkspaceChannelMapping.abc_project_id.is_not(None), ) .order_by(WorkspaceChannelMapping.id.asc()) .first() ) if mapping is None or mapping.abc_project_id is None: return None try: project_id = int(mapping.abc_project_id) channel_id = int(mapping.abc_channel_id) except (TypeError, ValueError): return None if project_id <= 0 or channel_id <= 0: return None return ABCProjectTarget(project_id=project_id, channel_id=channel_id) def candidate_tenant_ids(value: str | None = None) -> list[str]: raw_value = settings.admin_candidate_tenant_id if value is None else value return [ tenant_id.strip() for tenant_id in raw_value.split(",") if tenant_id.strip() ] def candidate_scope_filter(tenant_ids: str | list[str]): configured_tenant_ids = ( candidate_tenant_ids(tenant_ids) if isinstance(tenant_ids, str) else tenant_ids ) return or_( *[ condition for tenant_id in configured_tenant_ids for condition in ( SupportUser.tenant_id == tenant_id, func.json_contains( func.coalesce(SupportUser.tenant_ids, "[]"), func.json_quote(tenant_id), ) == 1, ) ], ) def normalize_phone_number(value: str | None) -> str: if not value: return "" digits = "".join(character for character in value if character.isdigit()) if digits.startswith("82") and len(digits) >= 11: return "0" + digits[2:] return digits def is_initial_super_admin(principal: SsoPrincipal) -> bool: configured_phone = normalize_phone_number( settings.initial_super_admin_phone_number, ) return bool(configured_phone) and normalize_phone_number( principal.phone_number, ) == configured_phone def is_candidate_tenant_user(principal: SsoPrincipal) -> bool: configured_tenant_ids = candidate_tenant_ids() return bool(configured_tenant_ids) and any( principal.tenant_id == candidate_tenant_id or candidate_tenant_id in principal.tenant_ids for candidate_tenant_id in configured_tenant_ids ) def has_global_admin(db: Session) -> bool: return ( db.query(SupportRoleAssignment) .join(SupportUser) .filter( SupportUser.status == "ACTIVE", SupportRoleAssignment.role_code.in_(service.GLOBAL_ADMIN_ROLES), ) .first() is not None ) def can_manage_admins(db: Session, principal: SsoPrincipal) -> bool: service.ensure_support_user(db, principal) db.commit() if service.is_system_admin(db, principal): return True return not has_global_admin(db) and is_initial_super_admin(principal) and is_candidate_tenant_user(principal) class WorkspaceAccessRequest(BaseModel): user_id: str = Field(min_length=1, max_length=100) tenant_id: str = Field(min_length=1, max_length=100) workspace_role: str = Field(default="END_USER", max_length=30) can_read: bool = True can_write: bool = False can_manage: bool = False can_approve: bool = False page_scope: str | None = Field(default=None, max_length=50) class AdminAssignmentRequest(BaseModel): user_id: str = Field(min_length=1, max_length=100) tenant_id: str = Field(min_length=1, max_length=100) email: str | None = Field(default=None, max_length=320) name: str | None = Field(default=None, max_length=100) department: str | None = Field(default=None, max_length=100) phone_number: str | None = Field(default=None, max_length=50) tenant_ids: list[str] | None = None role_code: str = Field(min_length=1, max_length=40) workspace_code: str | None = Field(default=None, max_length=50) @router.get("/me") def get_my_access( principal: SsoPrincipal = Depends(get_principal), db: Session = Depends(get_db), ) -> dict[str, object]: workspace_mapping_service.sync_project_workspaces(db) workspace_mapping_service.sync_missing_mappings(db) entries = service.list_accessible_workspaces(db, principal) is_system_admin = service.is_system_admin(db, principal) support_entries = [ (workspace, access) for workspace, access in entries if access.can_read or access.workspace_role in {"SYSTEM_ADMIN", "SUPER_ADMIN", "PROJECT_MANAGER"} ] if not support_entries and not is_system_admin: entries = service.ensure_default_support_access(db, principal) manager_entries = [ (workspace, access) for workspace, access in entries if access.workspace_role in {"SYSTEM_ADMIN", "SUPER_ADMIN", "PROJECT_MANAGER"} or access.can_manage ] support_entries = [ (workspace, access) for workspace, access in entries if access.can_read or access.workspace_role in {"SYSTEM_ADMIN", "SUPER_ADMIN", "PROJECT_MANAGER"} ] default_admin_path = None admin_workspace_code = None for workspace, access in manager_entries: target = abc_project_target(db, workspace.id) if admin_workspace_code is None: admin_workspace_code = workspace.workspace_code if target is None: continue admin_workspace_code = workspace.workspace_code default_admin_path = ( f"/main/project/{target.project_id}/feedback" f"?channelId={target.channel_id}" ) break default_support_path = None default_support_create_path = None if support_entries: workspace_code = support_entries[0][0].workspace_code encoded_workspace_code = quote(workspace_code, safe='') default_support_path = f"/support/{encoded_workspace_code}/list" default_support_create_path = f"/support/{encoded_workspace_code}/new" return { "user_id": principal.user_id, "tenant_id": principal.tenant_id, "email": principal.email, "name": principal.name, "department": principal.department, "phone_number": principal.phone_number, "is_system_admin": is_system_admin, "is_admin": bool(manager_entries), "roles": sorted({access.workspace_role for _, access in entries}), "workspaces": [ { "workspace_code": workspace.workspace_code, "workspace_name": workspace.workspace_name, "workspace_role": access.workspace_role, "can_read": access.can_read, "can_write": access.can_write, "can_manage": access.can_manage, "can_approve": access.can_approve, "page_scope": access.page_scope, "abc_project_id": ( abc_project_target(db, workspace.id).project_id if abc_project_target(db, workspace.id) else None ), "abc_channel_id": ( abc_project_target(db, workspace.id).channel_id if abc_project_target(db, workspace.id) else None ), } for workspace, access in entries ], "default_admin_path": default_admin_path, "admin_workspace_code": admin_workspace_code, "default_support_path": default_support_path, "default_support_create_path": default_support_create_path, } @router.post("/workspaces/{workspace_code}/users") def upsert_workspace_access( workspace_code: str, payload: WorkspaceAccessRequest, principal: SsoPrincipal = Depends(get_principal), db: Session = Depends(get_db), ) -> dict[str, object]: if not service.is_system_admin(db, principal): raise HTTPException(status_code=403, detail="시스템 관리자 권한이 필요합니다.") workspace = db.query(Workspace).filter_by( workspace_code=workspace_code, is_active=True, ).one_or_none() if workspace is None: raise HTTPException(status_code=404, detail="워크스페이스를 찾을 수 없습니다.") if payload.workspace_role not in service.SUPPORTED_WORKSPACE_ROLES: raise HTTPException(status_code=400, detail="지원하지 않는 workspace 역할입니다.") support_user = db.query(SupportUser).filter_by( sso_subject=payload.user_id, tenant_id=payload.tenant_id, ).one_or_none() if support_user is None: support_user = SupportUser( sso_subject=payload.user_id, tenant_id=payload.tenant_id, ) db.add(support_user) db.flush() assignment = db.query(SupportRoleAssignment).filter_by( support_user_id=support_user.id, workspace_id=workspace.id, ).filter(SupportRoleAssignment.role_code.in_(service.SUPPORTED_WORKSPACE_ROLES)).first() if assignment is None: assignment = SupportRoleAssignment( support_user_id=support_user.id, workspace_id=workspace.id, role_code=payload.workspace_role, ) db.add(assignment) else: assignment.role_code = payload.workspace_role access = db.query(UserWorkspaceAccess).filter_by( user_id=payload.user_id, tenant_id=payload.tenant_id, workspace_id=workspace.id, ).one_or_none() if access is None: access = UserWorkspaceAccess( user_id=payload.user_id, tenant_id=payload.tenant_id, workspace_id=workspace.id, ) db.add(access) service._apply_role_permissions(access, payload.workspace_role) access.page_scope = payload.page_scope db.commit() db.refresh(access) return { "id": access.id, "user_id": access.user_id, "tenant_id": access.tenant_id, "workspace_code": workspace.workspace_code, "workspace_role": access.workspace_role, } @router.get("/candidates") def list_admin_candidates( principal: SsoPrincipal = Depends(get_principal), db: Session = Depends(get_db), ) -> dict[str, object]: if not can_manage_admins(db, principal): raise HTTPException(status_code=403, detail="시스템 관리자 권한이 필요합니다.") configured_tenant_ids = candidate_tenant_ids() users_query = db.query(SupportUser).filter(SupportUser.status == "ACTIVE") if configured_tenant_ids: users_query = users_query.filter(candidate_scope_filter(configured_tenant_ids)) else: # System administrators are already authorized to manage the current # tenant. Do not make the assignment dialog unusable just because the # optional initial-admin candidate scope is not configured. The POST # endpoint applies the same current-tenant boundary when this setting # is empty. if not service.is_system_admin(db, principal): return {"items": []} users_query = users_query.filter(SupportUser.tenant_id == principal.tenant_id) users = users_query.order_by(SupportUser.id.asc()).all() return { "items": [ { "email": user.email or user.sso_subject, "isRegistered": True, "user": { "id": user.id, "email": user.email, "name": user.name, "department": user.department, "phone_number": user.phone_number, "oauthSubject": user.sso_subject, "oauthTenantId": user.tenant_id, "tenantIds": user.tenant_ids or [], }, } for user in users ], } @router.get("/admins") def list_admin_assignments( principal: SsoPrincipal = Depends(get_principal), db: Session = Depends(get_db), ) -> dict[str, object]: if not can_manage_admins(db, principal): raise HTTPException(status_code=403, detail="시스템 관리자 권한이 필요합니다.") project_workspace_codes = workspace_mapping_service.sync_project_workspaces(db) roles = db.query(SupportRole).order_by(SupportRole.code.asc()).all() workspace_query = db.query(Workspace).filter( Workspace.is_active.is_(True), Workspace.workspace_type == "SOFTWARE_APP", ) # Limit the selector to projects currently returned by ABC. This avoids # exposing legacy software_app seed rows that are not ABC projects. if project_workspace_codes is not None: workspace_query = workspace_query.filter( Workspace.workspace_code.in_(project_workspace_codes), ) workspaces = workspace_query.order_by(Workspace.id.asc()).all() configured_tenant_ids = candidate_tenant_ids() assignment_scope = ( candidate_scope_filter(configured_tenant_ids) if configured_tenant_ids else SupportUser.tenant_id == principal.tenant_id ) assignments = ( db.query(SupportRoleAssignment) .join(SupportUser) .filter( SupportUser.status == "ACTIVE", assignment_scope, SupportRoleAssignment.role_code.in_( service.GLOBAL_ADMIN_ROLES | {"PROJECT_MANAGER", "FEEDBACK_PROVIDER"}, ), ) .order_by(SupportRoleAssignment.created_at.asc()) .all() ) users: dict[int, dict[str, object]] = {} for assignment in assignments: user = assignment.user item = users.setdefault( user.id, { "id": user.id, "user_id": user.sso_subject, "tenant_id": user.tenant_id, "email": user.email, "name": user.name, "department": user.department, "phone_number": user.phone_number, "assignments": [], }, ) item["assignments"].append( { "id": assignment.id, "role_code": assignment.role_code, "role_name": assignment.role.name if assignment.role else assignment.role_code, "workspace_id": assignment.workspace_id, "workspace_code": assignment.workspace.workspace_code if assignment.workspace else None, "workspace_name": assignment.workspace.workspace_name if assignment.workspace else None, }, ) return { "roles": [ { "code": role.code, "name": role.name, "description": role.description, } for role in roles ], "workspaces": [ { "id": workspace.id, "code": workspace.workspace_code, "name": workspace.workspace_name, } for workspace in workspaces ], "users": list(users.values()), } @router.post("/admins") def upsert_admin_assignment( payload: AdminAssignmentRequest, principal: SsoPrincipal = Depends(get_principal), db: Session = Depends(get_db), ) -> dict[str, object]: if not can_manage_admins(db, principal): raise HTTPException(status_code=403, detail="시스템 관리자 권한이 필요합니다.") configured_tenant_ids = candidate_tenant_ids() if configured_tenant_ids: target_user = db.query(SupportUser).filter_by( sso_subject=payload.user_id, tenant_id=payload.tenant_id, status="ACTIVE", ).one_or_none() if target_user is None or not ( target_user.tenant_id in configured_tenant_ids or any(tenant_id in (target_user.tenant_ids or []) for tenant_id in configured_tenant_ids) ): raise HTTPException(status_code=403, detail="지정된 후보 테넌트의 사용자만 관리할 수 있습니다.") elif payload.tenant_id != principal.tenant_id: raise HTTPException(status_code=403, detail="현재 tenant의 사용자만 관리할 수 있습니다.") role_code = payload.role_code.upper() if role_code not in service.ROLE_PERMISSIONS: raise HTTPException(status_code=400, detail="지원하지 않는 역할입니다.") workspace = None if role_code in service.GLOBAL_ADMIN_ROLES: if payload.workspace_code: raise HTTPException(status_code=400, detail="전체 관리자 역할에는 workspace를 지정할 수 없습니다.") else: if not payload.workspace_code: raise HTTPException(status_code=400, detail="workspace 역할에는 workspace가 필요합니다.") workspace = db.query(Workspace).filter_by( workspace_code=payload.workspace_code, is_active=True, ).one_or_none() if workspace is None: raise HTTPException(status_code=404, detail="워크스페이스를 찾을 수 없습니다.") user = service.ensure_support_user_identity( db, user_id=payload.user_id, tenant_id=payload.tenant_id, email=payload.email, name=payload.name, department=payload.department, phone_number=payload.phone_number, tenant_ids=payload.tenant_ids, ) if workspace is not None: existing_workspace_assignments = db.query(SupportRoleAssignment).filter( SupportRoleAssignment.support_user_id == user.id, SupportRoleAssignment.workspace_id == workspace.id, SupportRoleAssignment.role_code.in_(service.SUPPORTED_WORKSPACE_ROLES), ).all() for existing in existing_workspace_assignments: if existing.role_code != role_code: db.delete(existing) assignment = db.query(SupportRoleAssignment).filter_by( support_user_id=user.id, role_code=role_code, workspace_id=workspace.id if workspace else None, ).one_or_none() if assignment is None: assignment = SupportRoleAssignment( support_user_id=user.id, role_code=role_code, workspace_id=workspace.id if workspace else None, ) db.add(assignment) if workspace is not None: access = db.query(UserWorkspaceAccess).filter_by( user_id=user.sso_subject, tenant_id=user.tenant_id, workspace_id=workspace.id, ).one_or_none() if access is None: access = UserWorkspaceAccess( user_id=user.sso_subject, tenant_id=user.tenant_id, workspace_id=workspace.id, ) db.add(access) service._apply_role_permissions(access, role_code) db.commit() db.refresh(assignment) return { "id": assignment.id, "user_id": user.sso_subject, "tenant_id": user.tenant_id, "role_code": role_code, "workspace_code": workspace.workspace_code if workspace else None, } @router.delete("/admins/{assignment_id}") def delete_admin_assignment( assignment_id: int, principal: SsoPrincipal = Depends(get_principal), db: Session = Depends(get_db), ) -> dict[str, bool]: if not service.is_system_admin(db, principal): raise HTTPException(status_code=403, detail="시스템 관리자 권한이 필요합니다.") assignment = db.query(SupportRoleAssignment).filter_by(id=assignment_id).one_or_none() if assignment is None: raise HTTPException(status_code=404, detail="권한 할당을 찾을 수 없습니다.") if assignment.role_code in service.GLOBAL_ADMIN_ROLES: remaining = db.query(SupportRoleAssignment).filter( SupportRoleAssignment.role_code.in_(service.GLOBAL_ADMIN_ROLES), SupportRoleAssignment.id != assignment.id, ).count() if remaining == 0: raise HTTPException(status_code=400, detail="전체 관리자는 최소 1명 이상 유지해야 합니다.") workspace_id = assignment.workspace_id support_user_id = assignment.support_user_id db.delete(assignment) if workspace_id is not None: remaining_role = db.query(SupportRoleAssignment).filter( SupportRoleAssignment.support_user_id == support_user_id, SupportRoleAssignment.workspace_id == workspace_id, SupportRoleAssignment.role_code.in_(service.SUPPORTED_WORKSPACE_ROLES), ).order_by(SupportRoleAssignment.created_at.desc()).first() access = db.query(UserWorkspaceAccess).filter_by( workspace_id=workspace_id, user_id=assignment.user.sso_subject, tenant_id=assignment.user.tenant_id, ).one_or_none() if access is not None: if remaining_role is None: db.delete(access) else: service._apply_role_permissions(access, remaining_role.role_code) db.commit() return {"deleted": True}