MMD
2747f9d4bb
The PM panel (PmPanelPage -> iframe /pm/) needs the project-management platform served at /pm/ on the ERP domain. Source: org repo Aramesh/PM, branch feature/project-management-platform. Update via: git subtree pull --prefix=pm aramesh-pm feature/project-management-platform --squash
297 lines
9.8 KiB
Python
297 lines
9.8 KiB
Python
"""Central authorization (RBAC) — roles, permissions and per-project scoping.
|
|
|
|
Everything authorization-related lives in this module so that adding a role is
|
|
one entry in `ROLE_PERMISSIONS` (plus a nav entry on the frontend), not a hunt
|
|
through the routers.
|
|
|
|
Two independent questions, answered separately:
|
|
|
|
* **Role → permissions** (`Perm`, `ROLE_PERMISSIONS`) decides *which operations*
|
|
a user may call at all.
|
|
* **Project scope** decides *which projects* those operations may touch. A role
|
|
listed in `SCOPED_ROLES` is limited to the rows in `project_assignments`
|
|
(an empty set means "no project at all"); every other role is unrestricted.
|
|
|
|
Routers must go through the dependencies and helpers below — hiding a button in
|
|
the UI is a convenience, never the enforcement.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import enum
|
|
import logging
|
|
import uuid
|
|
from collections.abc import Awaitable, Callable
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import Depends
|
|
from sqlalchemy import Select, select, text
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.db import get_db
|
|
from app.errors import ForbiddenError, NotFoundError
|
|
from app.models import PMUser, ProjectAssignment, Task
|
|
from app.security import get_current_user
|
|
|
|
logger = logging.getLogger("pm.authz")
|
|
|
|
|
|
# --- roles --------------------------------------------------------------------
|
|
|
|
|
|
class Role(str, enum.Enum):
|
|
project_manager = "project_manager"
|
|
project_supervisor = "project_supervisor"
|
|
|
|
|
|
ROLE_LABELS: dict[str, str] = {
|
|
Role.project_manager.value: "مدیر پروژه",
|
|
Role.project_supervisor.value: "سرپرست پروژه",
|
|
}
|
|
|
|
# Users created before RBAC existed keep their previous (full) access.
|
|
DEFAULT_ROLE = Role.project_manager
|
|
|
|
# Roles whose access is limited to their assigned projects.
|
|
SCOPED_ROLES: frozenset[Role] = frozenset({Role.project_supervisor})
|
|
|
|
|
|
# --- permissions --------------------------------------------------------------
|
|
|
|
|
|
class Perm(str, enum.Enum):
|
|
"""One value per authorizable operation — keep it coarse and stable."""
|
|
|
|
dashboard_view = "dashboard.view"
|
|
projects_view = "projects.view"
|
|
projects_create = "projects.create"
|
|
projects_update = "projects.update"
|
|
tasks_view = "tasks.view"
|
|
tasks_create = "tasks.create"
|
|
tasks_update = "tasks.update"
|
|
tasks_delete = "tasks.delete"
|
|
tasks_progress = "tasks.progress"
|
|
tasks_dates = "tasks.dates"
|
|
qc_view = "qc.view"
|
|
# executing QC: recording answers and raising/resolving defects
|
|
qc_execute = "qc.execute"
|
|
# QC structure: global templates + per-task definitions
|
|
qc_define = "qc.define"
|
|
materials_view = "materials.view"
|
|
materials_manage = "materials.manage"
|
|
users_manage = "users.manage"
|
|
files_upload = "files.upload"
|
|
|
|
|
|
# Project manager: full access to everything, across every project.
|
|
_MANAGER_PERMISSIONS: frozenset[str] = frozenset(p.value for p in Perm)
|
|
|
|
# Project supervisor: executes the assigned projects — reads the plan, records
|
|
# progress and QC results, needs the material list; never edits the QC
|
|
# structure, templates, procurement or integration settings.
|
|
_SUPERVISOR_PERMISSIONS: frozenset[str] = frozenset(
|
|
{
|
|
Perm.dashboard_view.value,
|
|
Perm.projects_view.value,
|
|
Perm.tasks_view.value,
|
|
Perm.tasks_progress.value,
|
|
Perm.qc_view.value,
|
|
Perm.qc_execute.value,
|
|
Perm.materials_view.value,
|
|
Perm.files_upload.value,
|
|
}
|
|
)
|
|
|
|
ROLE_PERMISSIONS: dict[Role, frozenset[str]] = {
|
|
Role.project_manager: _MANAGER_PERMISSIONS,
|
|
Role.project_supervisor: _SUPERVISOR_PERMISSIONS,
|
|
}
|
|
|
|
|
|
def permissions_for(role: Role) -> frozenset[str]:
|
|
return ROLE_PERMISSIONS.get(role, frozenset())
|
|
|
|
|
|
def role_of(user: PMUser) -> Role:
|
|
try:
|
|
return Role(user.role)
|
|
except ValueError:
|
|
logger.warning(
|
|
"unknown role %r on user %s — falling back to %s", user.role, user.username, DEFAULT_ROLE.value
|
|
)
|
|
return DEFAULT_ROLE
|
|
|
|
|
|
# --- request context ----------------------------------------------------------
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AuthContext:
|
|
"""Who is calling, what they may do, and which projects they may touch.
|
|
|
|
`project_uuids` — native projects (`projects.id`).
|
|
None for an unscoped role (project manager). For a scoped role a set — possibly empty — means "only these".
|
|
"""
|
|
|
|
user: PMUser
|
|
role: Role
|
|
project_uuids: frozenset[uuid.UUID] | None = None
|
|
|
|
@property
|
|
def permissions(self) -> frozenset[str]:
|
|
return permissions_for(self.role)
|
|
|
|
@property
|
|
def role_label(self) -> str:
|
|
return ROLE_LABELS.get(self.role.value, self.role.value)
|
|
|
|
@property
|
|
def all_projects(self) -> bool:
|
|
return self.project_uuids is None
|
|
|
|
def can(self, permission: Perm | str) -> bool:
|
|
value = permission.value if isinstance(permission, Perm) else permission
|
|
return value in self.permissions
|
|
|
|
def require(self, permission: Perm | str) -> None:
|
|
if not self.can(permission):
|
|
raise ForbiddenError(f"نقش «{self.role_label}» به این عملیات دسترسی ندارد")
|
|
|
|
def allows_project(self, project_id) -> bool:
|
|
"""Check whether the user is allowed to access the specified project."""
|
|
if self.project_uuids is None:
|
|
return True
|
|
try:
|
|
pid = uuid.UUID(str(project_id))
|
|
except (ValueError, AttributeError):
|
|
return False
|
|
return pid in self.project_uuids
|
|
|
|
def ensure_project(self, project_id) -> None:
|
|
"""Raise 403 when the project is outside this user's scope."""
|
|
if not self.allows_project(project_id):
|
|
raise ForbiddenError("این پروژه به شما اختصاص داده نشده است")
|
|
|
|
allows_native_project = allows_project
|
|
ensure_native_project = ensure_project
|
|
|
|
def scope(self, stmt: Select, column) -> Select:
|
|
"""Restrict a query over project UUIDs."""
|
|
if self.project_uuids is None:
|
|
return stmt
|
|
return stmt.where(column.in_(self.project_uuids))
|
|
|
|
scope_native = scope
|
|
|
|
@property
|
|
def scoped_project_count(self) -> int:
|
|
"""How many projects a scoped user may see (0 when nothing is assigned)."""
|
|
return len(self.project_uuids or ())
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"role": self.role.value,
|
|
"role_label": self.role_label,
|
|
"permissions": sorted(self.permissions),
|
|
"scope": {
|
|
"all_projects": self.all_projects,
|
|
"project_count": None if self.all_projects else self.scoped_project_count,
|
|
},
|
|
}
|
|
|
|
|
|
async def assigned_scope(db: AsyncSession, user_id) -> frozenset[uuid.UUID]:
|
|
"""Load assigned project UUIDs for a scoped user."""
|
|
rows = (
|
|
await db.execute(
|
|
select(ProjectAssignment.project_id).where(
|
|
ProjectAssignment.user_id == user_id
|
|
)
|
|
)
|
|
).scalars().all()
|
|
assigned = set(r for r in rows if r is not None)
|
|
|
|
# Also resolve projects assigned via projects.pm_user_id or user_project_members in the shared DB
|
|
pm_user = await db.get(PMUser, user_id)
|
|
if pm_user and pm_user.username:
|
|
erp_rows = (
|
|
await db.execute(
|
|
text("""
|
|
SELECT p.id FROM projects p JOIN users u ON p.pm_user_id = u.id WHERE u.username = :username
|
|
UNION
|
|
SELECT upm.project_id FROM user_project_members upm JOIN users u ON upm.user_id = u.id WHERE u.username = :username
|
|
"""),
|
|
{"username": pm_user.username},
|
|
)
|
|
).scalars().all()
|
|
for r in erp_rows:
|
|
if r:
|
|
try:
|
|
assigned.add(uuid.UUID(str(r)))
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
return frozenset(assigned)
|
|
|
|
|
|
async def build_context(db: AsyncSession, user: PMUser) -> AuthContext:
|
|
"""Resolve role + project scope for a user."""
|
|
role = role_of(user)
|
|
if role not in SCOPED_ROLES:
|
|
return AuthContext(user=user, role=role)
|
|
native = await assigned_scope(db, user.id)
|
|
return AuthContext(user=user, role=role, project_uuids=native)
|
|
|
|
|
|
async def get_auth_context(
|
|
user: PMUser = Depends(get_current_user), db: AsyncSession = Depends(get_db)
|
|
) -> AuthContext:
|
|
return await build_context(db, user)
|
|
|
|
|
|
def require_permission(permission: Perm) -> Callable[..., Awaitable[AuthContext]]:
|
|
"""Dependency: authenticated **and** allowed to perform `permission`."""
|
|
|
|
async def _guard(ctx: AuthContext = Depends(get_auth_context)) -> AuthContext:
|
|
ctx.require(permission)
|
|
return ctx
|
|
|
|
return _guard
|
|
|
|
|
|
# --- data-access guards -------------------------------------------------------
|
|
|
|
|
|
async def resolve_task(db: AsyncSession, task_ref: str | uuid.UUID) -> Task | None:
|
|
"""Find a native task by UUID."""
|
|
task_ref = task_ref.strip() if isinstance(task_ref, str) else task_ref
|
|
try:
|
|
task_uuid = uuid.UUID(str(task_ref))
|
|
except (ValueError, AttributeError):
|
|
return None
|
|
return await db.scalar(select(Task).where(Task.id == task_uuid))
|
|
|
|
|
|
async def scoped_task(db: AsyncSession, ctx: AuthContext, task_ref: str | uuid.UUID) -> Task:
|
|
"""Load a native task: 404 unknown, 403 out of scope."""
|
|
task = await resolve_task(db, task_ref)
|
|
if task is None:
|
|
raise NotFoundError("تسک یافت نشد")
|
|
ctx.ensure_project(task.project_id)
|
|
return task
|
|
|
|
|
|
scoped_native_task = scoped_task
|
|
resolve_native_task = resolve_task
|
|
|
|
|
|
# --- development role switching ----------------------------------------------
|
|
|
|
def dev_switch_enabled() -> bool:
|
|
return bool(settings.DEV_ROLE_SWITCH)
|
|
|
|
|
|
def dev_login_usernames() -> list[str]:
|
|
"""Demo accounts the dev switcher may log in as (never arbitrary users)."""
|
|
return [name for name in settings.dev_demo_usernames() if name]
|