384 lines
13 KiB
Python
384 lines
13 KiB
Python
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.dialects.mysql import insert as mysql_insert
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.auth import SsoPrincipal
|
|
from app.db.models import (
|
|
SupportRoleAssignment,
|
|
SupportUser,
|
|
UserWorkspaceAccess,
|
|
Workspace,
|
|
)
|
|
from app.core.config import settings
|
|
|
|
|
|
class AccessService:
|
|
ROLE_PRIORITY = {
|
|
"SYSTEM_ADMIN": 5,
|
|
"SUPER_ADMIN": 4,
|
|
"PROJECT_MANAGER": 3,
|
|
"FEEDBACK_PROVIDER": 2,
|
|
"END_USER": 1,
|
|
}
|
|
ROLE_PERMISSIONS = {
|
|
"SYSTEM_ADMIN": (True, True, True, True),
|
|
"SUPER_ADMIN": (True, True, True, True),
|
|
"PROJECT_MANAGER": (True, True, True, True),
|
|
"FEEDBACK_PROVIDER": (True, True, False, False),
|
|
"END_USER": (True, True, False, False),
|
|
}
|
|
GLOBAL_ADMIN_ROLES = {"SYSTEM_ADMIN", "SUPER_ADMIN"}
|
|
MANAGER_ROLES = {"SYSTEM_ADMIN", "SUPER_ADMIN", "PROJECT_MANAGER"}
|
|
SUPPORTED_WORKSPACE_ROLES = {
|
|
"PROJECT_MANAGER",
|
|
"END_USER",
|
|
"FEEDBACK_PROVIDER",
|
|
}
|
|
|
|
@staticmethod
|
|
def default_user_workspace_code(db: Session) -> str | None:
|
|
"""Return the configured public feedback workspace.
|
|
|
|
Users without an explicit role assignment receive END_USER access to
|
|
this workspace. A configured code prevents old seed data or a newly
|
|
created project from changing the public writer target.
|
|
"""
|
|
configured_code = settings.default_support_workspace_code.strip()
|
|
query = (
|
|
select(Workspace.workspace_code)
|
|
.where(
|
|
Workspace.is_active.is_(True),
|
|
Workspace.workspace_type == "SOFTWARE_APP",
|
|
)
|
|
)
|
|
if configured_code:
|
|
query = query.where(Workspace.workspace_code == configured_code)
|
|
else:
|
|
query = query.order_by(Workspace.id.asc())
|
|
return db.execute(query).scalar()
|
|
|
|
@staticmethod
|
|
def _upsert_support_user(
|
|
db: Session,
|
|
*,
|
|
user_id: str,
|
|
tenant_id: str,
|
|
email: str | None = None,
|
|
name: str | None = None,
|
|
department: str | None = None,
|
|
phone_number: str | None = None,
|
|
tenant_ids: list[str] | None = None,
|
|
) -> SupportUser:
|
|
"""Create or refresh an SSO identity without concurrent insert races.
|
|
|
|
The access endpoint is called several times in parallel during the
|
|
first SSO redirect. A select-then-insert sequence can deadlock on the
|
|
unique (sso_subject, tenant_id) index, so use MySQL's atomic upsert.
|
|
Empty profile fields never erase values already synchronized.
|
|
"""
|
|
table = SupportUser.__table__
|
|
statement = mysql_insert(table).values(
|
|
sso_subject=user_id,
|
|
tenant_id=tenant_id,
|
|
email=email,
|
|
name=name,
|
|
department=department,
|
|
phone_number=phone_number,
|
|
tenant_ids=tenant_ids or None,
|
|
)
|
|
statement = statement.on_duplicate_key_update(
|
|
email=func.coalesce(statement.inserted.email, table.c.email),
|
|
name=func.coalesce(statement.inserted.name, table.c.name),
|
|
department=func.coalesce(
|
|
statement.inserted.department,
|
|
table.c.department,
|
|
),
|
|
phone_number=func.coalesce(
|
|
statement.inserted.phone_number,
|
|
table.c.phone_number,
|
|
),
|
|
tenant_ids=func.coalesce(
|
|
statement.inserted.tenant_ids,
|
|
table.c.tenant_ids,
|
|
),
|
|
)
|
|
db.execute(statement)
|
|
return db.execute(
|
|
select(SupportUser).where(
|
|
SupportUser.sso_subject == user_id,
|
|
SupportUser.tenant_id == tenant_id,
|
|
),
|
|
).scalar_one()
|
|
|
|
@staticmethod
|
|
def ensure_support_user(db: Session, principal: SsoPrincipal) -> SupportUser:
|
|
return AccessService._upsert_support_user(
|
|
db,
|
|
user_id=principal.user_id,
|
|
tenant_id=principal.tenant_id,
|
|
email=principal.email,
|
|
name=principal.name,
|
|
department=principal.department,
|
|
phone_number=principal.phone_number,
|
|
tenant_ids=principal.tenant_ids,
|
|
)
|
|
|
|
@staticmethod
|
|
def ensure_support_user_identity(
|
|
db: Session,
|
|
*,
|
|
user_id: str,
|
|
tenant_id: str,
|
|
email: str | None = None,
|
|
name: str | None = None,
|
|
department: str | None = None,
|
|
phone_number: str | None = None,
|
|
tenant_ids: list[str] | None = None,
|
|
) -> SupportUser:
|
|
return AccessService._upsert_support_user(
|
|
db,
|
|
user_id=user_id,
|
|
tenant_id=tenant_id,
|
|
email=email,
|
|
name=name,
|
|
department=department,
|
|
phone_number=phone_number,
|
|
tenant_ids=tenant_ids,
|
|
)
|
|
|
|
@staticmethod
|
|
def _role_codes(
|
|
db: Session,
|
|
user: SupportUser,
|
|
workspace_id: int | None = None,
|
|
*,
|
|
global_only: bool = False,
|
|
) -> list[str]:
|
|
conditions = [SupportRoleAssignment.support_user_id == user.id]
|
|
if global_only:
|
|
# Global admin roles are global by role code. Older seed data may
|
|
# still contain a workspace_id on these assignments.
|
|
conditions.append(
|
|
SupportRoleAssignment.role_code.in_(AccessService.GLOBAL_ADMIN_ROLES),
|
|
)
|
|
elif workspace_id is not None:
|
|
# Only global administrator roles apply to every workspace. A
|
|
# project-scoped role with a NULL workspace is not a shortcut to
|
|
# global access.
|
|
conditions.append(
|
|
or_(
|
|
SupportRoleAssignment.workspace_id == workspace_id,
|
|
and_(
|
|
SupportRoleAssignment.workspace_id.is_(None),
|
|
SupportRoleAssignment.role_code.in_(AccessService.GLOBAL_ADMIN_ROLES),
|
|
),
|
|
),
|
|
)
|
|
return list(
|
|
db.execute(
|
|
select(SupportRoleAssignment.role_code).where(*conditions),
|
|
).scalars().all(),
|
|
)
|
|
|
|
def is_system_admin(self, db: Session, principal: SsoPrincipal) -> bool:
|
|
"""Check only Secretary's internal global role assignments.
|
|
|
|
The ABC `users.type` claim is intentionally not consulted here. ABC's
|
|
users/roles/members tables remain responsible for ABC's own console API.
|
|
"""
|
|
user = self.ensure_support_user(db, principal)
|
|
return bool(
|
|
set(self._role_codes(db, user, global_only=True))
|
|
& AccessService.GLOBAL_ADMIN_ROLES,
|
|
)
|
|
|
|
def _workspace_role(
|
|
self,
|
|
db: Session,
|
|
user: SupportUser,
|
|
workspace_id: int,
|
|
) -> str | None:
|
|
roles = self._role_codes(db, user, workspace_id)
|
|
if not roles:
|
|
return None
|
|
return max(roles, key=lambda role: self.ROLE_PRIORITY.get(role, 0))
|
|
|
|
def _apply_role_permissions(
|
|
self,
|
|
access: UserWorkspaceAccess,
|
|
role: str,
|
|
) -> UserWorkspaceAccess:
|
|
can_read, can_write, can_manage, can_approve = self.ROLE_PERMISSIONS.get(
|
|
role,
|
|
(False, False, False, False),
|
|
)
|
|
access.workspace_role = role
|
|
access.can_read = can_read
|
|
access.can_write = can_write
|
|
access.can_manage = can_manage
|
|
access.can_approve = can_approve
|
|
return access
|
|
|
|
@staticmethod
|
|
def _synthetic_access(
|
|
principal: SsoPrincipal,
|
|
workspace: Workspace,
|
|
role: str,
|
|
) -> UserWorkspaceAccess:
|
|
access = UserWorkspaceAccess(
|
|
user_id=principal.user_id,
|
|
tenant_id=principal.tenant_id,
|
|
workspace_id=workspace.id,
|
|
workspace=workspace,
|
|
)
|
|
return AccessService()._apply_role_permissions(access, role)
|
|
|
|
def find_workspace_access(
|
|
self,
|
|
db: Session,
|
|
principal: SsoPrincipal,
|
|
workspace_code: str,
|
|
default_workspace_code: str | None = None,
|
|
) -> UserWorkspaceAccess | None:
|
|
user = self.ensure_support_user(db, principal)
|
|
workspace = db.execute(
|
|
select(Workspace).where(
|
|
Workspace.workspace_code == workspace_code,
|
|
Workspace.is_active.is_(True),
|
|
),
|
|
).scalar_one_or_none()
|
|
if workspace is None or user.status != "ACTIVE":
|
|
return None
|
|
|
|
global_roles = set(self._role_codes(db, user, global_only=True))
|
|
if global_roles & AccessService.GLOBAL_ADMIN_ROLES:
|
|
return self._synthetic_access(principal, workspace, "SYSTEM_ADMIN")
|
|
|
|
role = self._workspace_role(db, user, workspace.id)
|
|
if role is None:
|
|
# A first-time user can submit feedback in the default workspace
|
|
# without creating an END_USER role assignment. Role assignments
|
|
# are reserved for explicit workspace administration.
|
|
if workspace.workspace_code != (
|
|
default_workspace_code or self.default_user_workspace_code(db)
|
|
):
|
|
return None
|
|
role = "END_USER"
|
|
|
|
access = db.execute(
|
|
select(UserWorkspaceAccess).where(
|
|
UserWorkspaceAccess.user_id == principal.user_id,
|
|
UserWorkspaceAccess.tenant_id == principal.tenant_id,
|
|
UserWorkspaceAccess.workspace_id == workspace.id,
|
|
),
|
|
).scalar_one_or_none()
|
|
if access is None:
|
|
access = UserWorkspaceAccess(
|
|
user_id=principal.user_id,
|
|
tenant_id=principal.tenant_id,
|
|
workspace_id=workspace.id,
|
|
workspace=workspace,
|
|
)
|
|
else:
|
|
access.workspace = workspace
|
|
return self._apply_role_permissions(access, role)
|
|
|
|
def assert_workspace_access(
|
|
self,
|
|
db: Session,
|
|
principal: SsoPrincipal,
|
|
workspace_code: str,
|
|
*,
|
|
write: bool = False,
|
|
) -> UserWorkspaceAccess:
|
|
access = self.find_workspace_access(db, principal, workspace_code)
|
|
if access is None:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(status_code=403, detail="이 workspace에 대한 권한이 없습니다.")
|
|
|
|
if write and not access.can_write:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(status_code=403, detail="이 workspace에 작성 권한이 없습니다.")
|
|
if not write and not access.can_read:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(status_code=403, detail="이 workspace를 조회할 권한이 없습니다.")
|
|
return access
|
|
|
|
def list_accessible_workspaces(
|
|
self,
|
|
db: Session,
|
|
principal: SsoPrincipal,
|
|
default_workspace_code: str | None = None,
|
|
) -> list[tuple[Workspace, UserWorkspaceAccess]]:
|
|
default_workspace_code = (
|
|
default_workspace_code or self.default_user_workspace_code(db)
|
|
)
|
|
workspaces = db.execute(
|
|
select(Workspace)
|
|
.where(Workspace.is_active.is_(True))
|
|
.order_by(Workspace.id.asc()),
|
|
).scalars().all()
|
|
entries: list[tuple[Workspace, UserWorkspaceAccess]] = []
|
|
for workspace in workspaces:
|
|
access = self.find_workspace_access(
|
|
db,
|
|
principal,
|
|
workspace.workspace_code,
|
|
default_workspace_code=default_workspace_code,
|
|
)
|
|
if access is not None and (access.can_read or access.can_manage):
|
|
entries.append((workspace, access))
|
|
return entries
|
|
|
|
def ensure_default_support_access(
|
|
self,
|
|
db: Session,
|
|
principal: SsoPrincipal,
|
|
default_workspace_code: str | None = None,
|
|
) -> list[tuple[Workspace, UserWorkspaceAccess]]:
|
|
user = self.ensure_support_user(db, principal)
|
|
if self.is_system_admin(db, principal):
|
|
return self.list_accessible_workspaces(
|
|
db,
|
|
principal,
|
|
default_workspace_code=default_workspace_code,
|
|
)
|
|
|
|
default_workspace_code = (
|
|
default_workspace_code or self.default_user_workspace_code(db)
|
|
)
|
|
if default_workspace_code is None:
|
|
return []
|
|
|
|
workspace = db.execute(
|
|
select(Workspace).where(
|
|
Workspace.workspace_code == default_workspace_code,
|
|
Workspace.is_active.is_(True),
|
|
),
|
|
).scalar_one_or_none()
|
|
if workspace is None:
|
|
return []
|
|
|
|
role = self._workspace_role(db, user, workspace.id) or "END_USER"
|
|
|
|
access = db.execute(
|
|
select(UserWorkspaceAccess).where(
|
|
UserWorkspaceAccess.user_id == principal.user_id,
|
|
UserWorkspaceAccess.tenant_id == principal.tenant_id,
|
|
UserWorkspaceAccess.workspace_id == workspace.id,
|
|
),
|
|
).scalar_one_or_none()
|
|
if access is None:
|
|
access = UserWorkspaceAccess(
|
|
user_id=principal.user_id,
|
|
tenant_id=principal.tenant_id,
|
|
workspace_id=workspace.id,
|
|
)
|
|
db.add(access)
|
|
access.workspace = workspace
|
|
self._apply_role_permissions(access, role)
|
|
db.commit()
|
|
db.refresh(access)
|
|
access.workspace = workspace
|
|
return [(workspace, access)]
|