597 lines
22 KiB
Python
597 lines
22 KiB
Python
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,
|
|
"tenant_ids": principal.tenant_ids,
|
|
"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}
|