308 lines
10 KiB
Python
308 lines
10 KiB
Python
"""HTTP Bearer scheme, current-user dependency, and RBAC enforcement.
|
|
|
|
PermissionTag is a str Enum of every "module.action" tag. With class PermissionTag(str, Enum),
|
|
f"{PermissionTag.JOBS_VIEW}" renders "PermissionTag.JOBS_VIEW" — always use .value in JSON
|
|
and HTTPException details.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from enum import Enum
|
|
from typing import Annotated
|
|
|
|
import jwt
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db_setup import get_session
|
|
from role.models import EnumRoles, Roles
|
|
from users.models import Users
|
|
from users.plugins import decode_token
|
|
from users.serializers import serialize_user
|
|
|
|
bearer_scheme = HTTPBearer()
|
|
|
|
|
|
class PermissionModule(str, Enum):
|
|
DASHBOARD = "dashboard"
|
|
INBOX = "inbox"
|
|
JOBS = "jobs"
|
|
CANDIDATES = "candidates"
|
|
PIPELINE = "pipeline"
|
|
INTERVIEWS = "interviews"
|
|
ASSESSMENTS = "assessments"
|
|
OFFERS = "offers"
|
|
REPORTS = "reports"
|
|
ANALYTICS = "analytics"
|
|
JOB_BOARD = "job_board"
|
|
SETTINGS = "settings"
|
|
RBAC_USERS = "rbac_users"
|
|
TASKS = "tasks"
|
|
TALENT = "talent"
|
|
REQUISITIONS = "requisitions"
|
|
|
|
class PermissionAction(str, Enum):
|
|
VIEW = "view"
|
|
CREATE = "create"
|
|
EDIT = "edit"
|
|
DELETE = "delete"
|
|
APPROVE = "approve"
|
|
EXPORT = "export"
|
|
MANAGE = "manage"
|
|
CONFIGURE = "configure"
|
|
|
|
|
|
class PermissionTag(str, Enum):
|
|
REQUISITIONS_VIEW = "requisitions.view"
|
|
REQUISITIONS_CREATE = "requisitions.create"
|
|
REQUISITIONS_EDIT = "requisitions.edit"
|
|
REQUISITIONS_DELETE = "requisitions.delete"
|
|
REQUISITIONS_APPROVE = "requisitions.approve"
|
|
REQUISITIONS_EXPORT = "requisitions.export"
|
|
REQUISITIONS_MANAGE = "requisitions.manage"
|
|
REQUISITIONS_CONFIGURE = "requisitions.configure"
|
|
|
|
DASHBOARD_VIEW = "dashboard.view"
|
|
DASHBOARD_CREATE = "dashboard.create"
|
|
DASHBOARD_EDIT = "dashboard.edit"
|
|
DASHBOARD_DELETE = "dashboard.delete"
|
|
DASHBOARD_APPROVE = "dashboard.approve"
|
|
DASHBOARD_EXPORT = "dashboard.export"
|
|
DASHBOARD_MANAGE = "dashboard.manage"
|
|
DASHBOARD_CONFIGURE = "dashboard.configure"
|
|
INBOX_VIEW = "inbox.view"
|
|
INBOX_CREATE = "inbox.create"
|
|
INBOX_EDIT = "inbox.edit"
|
|
INBOX_DELETE = "inbox.delete"
|
|
INBOX_APPROVE = "inbox.approve"
|
|
INBOX_EXPORT = "inbox.export"
|
|
INBOX_MANAGE = "inbox.manage"
|
|
INBOX_CONFIGURE = "inbox.configure"
|
|
JOBS_VIEW = "jobs.view"
|
|
JOBS_CREATE = "jobs.create"
|
|
JOBS_EDIT = "jobs.edit"
|
|
JOBS_DELETE = "jobs.delete"
|
|
JOBS_APPROVE = "jobs.approve"
|
|
JOBS_EXPORT = "jobs.export"
|
|
JOBS_MANAGE = "jobs.manage"
|
|
JOBS_CONFIGURE = "jobs.configure"
|
|
CANDIDATES_VIEW = "candidates.view"
|
|
CANDIDATES_CREATE = "candidates.create"
|
|
CANDIDATES_EDIT = "candidates.edit"
|
|
CANDIDATES_DELETE = "candidates.delete"
|
|
CANDIDATES_APPROVE = "candidates.approve"
|
|
CANDIDATES_EXPORT = "candidates.export"
|
|
CANDIDATES_MANAGE = "candidates.manage"
|
|
CANDIDATES_CONFIGURE = "candidates.configure"
|
|
PIPELINE_VIEW = "pipeline.view"
|
|
PIPELINE_CREATE = "pipeline.create"
|
|
PIPELINE_EDIT = "pipeline.edit"
|
|
PIPELINE_DELETE = "pipeline.delete"
|
|
PIPELINE_APPROVE = "pipeline.approve"
|
|
PIPELINE_EXPORT = "pipeline.export"
|
|
PIPELINE_MANAGE = "pipeline.manage"
|
|
PIPELINE_CONFIGURE = "pipeline.configure"
|
|
INTERVIEWS_VIEW = "interviews.view"
|
|
INTERVIEWS_CREATE = "interviews.create"
|
|
INTERVIEWS_EDIT = "interviews.edit"
|
|
INTERVIEWS_DELETE = "interviews.delete"
|
|
INTERVIEWS_APPROVE = "interviews.approve"
|
|
INTERVIEWS_EXPORT = "interviews.export"
|
|
INTERVIEWS_MANAGE = "interviews.manage"
|
|
INTERVIEWS_CONFIGURE = "interviews.configure"
|
|
ASSESSMENTS_VIEW = "assessments.view"
|
|
ASSESSMENTS_CREATE = "assessments.create"
|
|
ASSESSMENTS_EDIT = "assessments.edit"
|
|
ASSESSMENTS_DELETE = "assessments.delete"
|
|
ASSESSMENTS_APPROVE = "assessments.approve"
|
|
ASSESSMENTS_EXPORT = "assessments.export"
|
|
ASSESSMENTS_MANAGE = "assessments.manage"
|
|
ASSESSMENTS_CONFIGURE = "assessments.configure"
|
|
OFFERS_VIEW = "offers.view"
|
|
OFFERS_CREATE = "offers.create"
|
|
OFFERS_EDIT = "offers.edit"
|
|
OFFERS_DELETE = "offers.delete"
|
|
OFFERS_APPROVE = "offers.approve"
|
|
OFFERS_EXPORT = "offers.export"
|
|
OFFERS_MANAGE = "offers.manage"
|
|
OFFERS_CONFIGURE = "offers.configure"
|
|
REPORTS_VIEW = "reports.view"
|
|
REPORTS_CREATE = "reports.create"
|
|
REPORTS_EDIT = "reports.edit"
|
|
REPORTS_DELETE = "reports.delete"
|
|
REPORTS_APPROVE = "reports.approve"
|
|
REPORTS_EXPORT = "reports.export"
|
|
REPORTS_MANAGE = "reports.manage"
|
|
REPORTS_CONFIGURE = "reports.configure"
|
|
ANALYTICS_VIEW = "analytics.view"
|
|
ANALYTICS_CREATE = "analytics.create"
|
|
ANALYTICS_EDIT = "analytics.edit"
|
|
ANALYTICS_DELETE = "analytics.delete"
|
|
ANALYTICS_APPROVE = "analytics.approve"
|
|
ANALYTICS_EXPORT = "analytics.export"
|
|
ANALYTICS_MANAGE = "analytics.manage"
|
|
ANALYTICS_CONFIGURE = "analytics.configure"
|
|
JOB_BOARD_VIEW = "job_board.view"
|
|
JOB_BOARD_CREATE = "job_board.create"
|
|
JOB_BOARD_EDIT = "job_board.edit"
|
|
JOB_BOARD_DELETE = "job_board.delete"
|
|
JOB_BOARD_APPROVE = "job_board.approve"
|
|
JOB_BOARD_EXPORT = "job_board.export"
|
|
JOB_BOARD_MANAGE = "job_board.manage"
|
|
JOB_BOARD_CONFIGURE = "job_board.configure"
|
|
SETTINGS_VIEW = "settings.view"
|
|
SETTINGS_CREATE = "settings.create"
|
|
SETTINGS_EDIT = "settings.edit"
|
|
SETTINGS_DELETE = "settings.delete"
|
|
SETTINGS_APPROVE = "settings.approve"
|
|
SETTINGS_EXPORT = "settings.export"
|
|
SETTINGS_MANAGE = "settings.manage"
|
|
SETTINGS_CONFIGURE = "settings.configure"
|
|
RBAC_USERS_VIEW = "rbac_users.view"
|
|
RBAC_USERS_CREATE = "rbac_users.create"
|
|
RBAC_USERS_EDIT = "rbac_users.edit"
|
|
RBAC_USERS_DELETE = "rbac_users.delete"
|
|
RBAC_USERS_APPROVE = "rbac_users.approve"
|
|
RBAC_USERS_EXPORT = "rbac_users.export"
|
|
RBAC_USERS_MANAGE = "rbac_users.manage"
|
|
RBAC_USERS_CONFIGURE = "rbac_users.configure"
|
|
TASKS_VIEW = "tasks.view"
|
|
TASKS_CREATE = "tasks.create"
|
|
TASKS_EDIT = "tasks.edit"
|
|
TASKS_DELETE = "tasks.delete"
|
|
TASKS_APPROVE = "tasks.approve"
|
|
TASKS_EXPORT = "tasks.export"
|
|
TASKS_MANAGE = "tasks.manage"
|
|
TASKS_CONFIGURE = "tasks.configure"
|
|
TALENT_VIEW = "talent.view"
|
|
TALENT_CREATE = "talent.create"
|
|
TALENT_EDIT = "talent.edit"
|
|
TALENT_DELETE = "talent.delete"
|
|
TALENT_APPROVE = "talent.approve"
|
|
TALENT_EXPORT = "talent.export"
|
|
TALENT_MANAGE = "talent.manage"
|
|
TALENT_CONFIGURE = "talent.configure"
|
|
|
|
|
|
def _assert_vocabulary_complete() -> None:
|
|
expected = {
|
|
f"{m.value}.{a.value}"
|
|
for m in PermissionModule
|
|
for a in PermissionAction
|
|
}
|
|
actual = {t.value for t in PermissionTag}
|
|
if expected != actual:
|
|
missing = sorted(expected - actual)
|
|
extra = sorted(actual - expected)
|
|
raise RuntimeError(
|
|
f"PermissionTag vocabulary drift: missing={missing!r} extra={extra!r}"
|
|
)
|
|
|
|
|
|
_assert_vocabulary_complete()
|
|
|
|
|
|
def is_hiring_manager(current_user: dict | None) -> bool:
|
|
"""Hiring-manager portal: seeded hiring_manager, or a custom Manager role.
|
|
|
|
Ahmed Baig's Access Control role is named Manager (not hiring_manager).
|
|
Matching is case-insensitive so the sidebar and API scope agree.
|
|
"""
|
|
name = ((current_user or {}).get("role_name") or "").strip().lower()
|
|
return name in {EnumRoles.HIRING_MANAGER.value, "manager"}
|
|
|
|
|
|
_ADMIN_ROLES = {
|
|
EnumRoles.SYSTEM_ADMINISTRATOR.value,
|
|
EnumRoles.HR_ADMINISTRATOR.value,
|
|
"admin",
|
|
}
|
|
|
|
|
|
def is_admin(current_user: dict | None) -> bool:
|
|
"""Org-wide staff: seeded admin roles, a custom Admin role, or requisitions.manage.
|
|
|
|
Managers keep a created_by-scoped requisition list. Admins see every
|
|
non-deleted requisition, linked to a job post or not.
|
|
"""
|
|
user = current_user or {}
|
|
name = (user.get("role_name") or "").strip().lower()
|
|
if name in _ADMIN_ROLES:
|
|
return True
|
|
granted = user.get("permissions") or []
|
|
return PermissionTag.REQUISITIONS_MANAGE.value in granted
|
|
|
|
|
|
def has_permission(
|
|
granted: set[str] | list[str] | tuple[str, ...],
|
|
*required: PermissionTag,
|
|
require_all: bool = True,
|
|
) -> bool:
|
|
needed = {t.value for t in required}
|
|
have = set(granted or ())
|
|
if require_all:
|
|
return needed.issubset(have)
|
|
return bool(needed & have)
|
|
|
|
|
|
def require_permission(*required: PermissionTag, require_all: bool = True):
|
|
"""FastAPI dependency: enforce one or more PermissionTag values (AND by default)."""
|
|
|
|
async def dependency(current_user: CurrentUser) -> dict:
|
|
if current_user.get("role_id") is None:
|
|
raise HTTPException(status_code=403, detail="User has no role assigned")
|
|
|
|
granted = current_user.get("permissions") or []
|
|
if not has_permission(granted, *required, require_all=require_all):
|
|
if require_all and len(required) == 1:
|
|
detail = f"Missing required permission: {required[0].value}"
|
|
elif require_all:
|
|
detail = (
|
|
"Missing required permissions: "
|
|
+ ", ".join(t.value for t in required)
|
|
)
|
|
else:
|
|
detail = (
|
|
"Missing any of required permissions: "
|
|
+ ", ".join(t.value for t in required)
|
|
)
|
|
raise HTTPException(status_code=403, detail=detail)
|
|
return current_user
|
|
|
|
return dependency
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: Annotated[HTTPAuthorizationCredentials, Depends(bearer_scheme)],
|
|
session: Annotated[AsyncSession, Depends(get_session)],
|
|
) -> dict:
|
|
credentials_exception = HTTPException(
|
|
status_code=401,
|
|
detail="Could not validate credentials",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
try:
|
|
payload = decode_token(credentials.credentials, expected_type="access")
|
|
except jwt.PyJWTError:
|
|
raise credentials_exception
|
|
|
|
user = await Users.get_user_by_id(session, payload.get("sub"))
|
|
if not user or user.is_deleted or not user.is_active:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="User is inactive or does not exist",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
if not user.is_approved:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="Your Approval is at Pending",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
permissions = await Roles.resolve_tags(session, user.role)
|
|
return serialize_user(user, with_permissions=True, permissions=permissions)
|
|
|
|
|
|
CurrentUser = Annotated[dict, Depends(get_current_user)]
|