266 lines
9.3 KiB
Python
266 lines
9.3 KiB
Python
"""RBAC declared in code: system roles, permission bundles, and the SQL that syncs them.
|
|
|
|
`PermissionTag` (users/permissions.py) is the tag vocabulary. `SYSTEM_ROLES` and
|
|
`RBAC_BUNDLES` below declare the rest. Adding a module means adding its enum values
|
|
and a bundle entry here — no hand-written migrations/manual/*.sql.
|
|
|
|
`build_rbac_sql` turns the difference between this code and a database into
|
|
idempotent SQL. Admins curate roles in Access Control (a role's matrix replaces its
|
|
bundle list), so every code-declared grant — a tag inside a bundle, a bundle on a
|
|
role — is applied once per database and recorded in a ledger. A grant an admin later
|
|
removes stays removed; only grants the ledger has never seen are applied.
|
|
|
|
A database that already has roles but no ledger predates the sync. Its grants were
|
|
applied by the old manual migrations and may since have been curated, so the first
|
|
sync only records them. Bundles that do not exist yet are still created and filled.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import NamedTuple
|
|
|
|
from role.models import EnumRoles
|
|
from users.permissions import PermissionModule, PermissionTag
|
|
|
|
RBAC_LEDGER_TABLE = "rbac_sync_ledger"
|
|
|
|
|
|
class SystemRole(NamedTuple):
|
|
id: int # pinned: the app hardcodes candidate = 8 and hiring_manager = 4
|
|
name: EnumRoles
|
|
description: str
|
|
retired: bool = False # soft-deleted by 026; created retired on a fresh database
|
|
|
|
|
|
class Bundle(NamedTuple):
|
|
description: str
|
|
modules: tuple[PermissionModule, ...] = ()
|
|
tags: tuple[PermissionTag, ...] = ()
|
|
roles: tuple[EnumRoles, ...] = ()
|
|
|
|
|
|
SYSTEM_ROLES: tuple[SystemRole, ...] = (
|
|
SystemRole(1, EnumRoles.SYSTEM_ADMINISTRATOR, "Full system access"),
|
|
SystemRole(2, EnumRoles.HR_ADMINISTRATOR, "HR administration", retired=True),
|
|
SystemRole(3, EnumRoles.RECRUITER, "Recruiting staff"),
|
|
SystemRole(4, EnumRoles.HIRING_MANAGER, "Hiring manager for own requisitions"),
|
|
SystemRole(5, EnumRoles.DEPARTMENT_HEAD, "Head of a department"),
|
|
SystemRole(6, EnumRoles.INTERVIEWER, "Interview panel member", retired=True),
|
|
SystemRole(7, EnumRoles.CEO, "Chief executive", retired=True),
|
|
SystemRole(8, EnumRoles.CANDIDATE, "Applicant account"),
|
|
)
|
|
|
|
_R = EnumRoles
|
|
_T = PermissionTag
|
|
_M = PermissionModule
|
|
_STAFF = (
|
|
_R.SYSTEM_ADMINISTRATOR,
|
|
_R.HR_ADMINISTRATOR,
|
|
_R.RECRUITER,
|
|
_R.HIRING_MANAGER,
|
|
_R.DEPARTMENT_HEAD,
|
|
_R.CEO,
|
|
)
|
|
|
|
RBAC_BUNDLES: dict[str, Bundle] = {
|
|
"all_access": Bundle(
|
|
"Every permission in every module",
|
|
modules=tuple(PermissionModule),
|
|
roles=(_R.SYSTEM_ADMINISTRATOR,),
|
|
),
|
|
"analytics_dashboard": Bundle(
|
|
"Dashboard KPI tiles, analytics charts, offers, and interview list",
|
|
modules=(_M.DASHBOARD, _M.ANALYTICS, _M.OFFERS),
|
|
tags=(_T.INTERVIEWS_VIEW,),
|
|
roles=_STAFF,
|
|
),
|
|
"tasks_management": Bundle(
|
|
"Recruiting task list: view, create, complete and manage tasks",
|
|
modules=(_M.TASKS,),
|
|
roles=(_R.SYSTEM_ADMINISTRATOR, _R.HR_ADMINISTRATOR, _R.RECRUITER),
|
|
),
|
|
"tasks_viewer": Bundle(
|
|
"Recruiting task list: read-only access",
|
|
tags=(_T.TASKS_VIEW, _T.TASKS_EXPORT),
|
|
roles=(_R.HIRING_MANAGER, _R.DEPARTMENT_HEAD, _R.CEO),
|
|
),
|
|
"talent_sourcing": Bundle(
|
|
"LinkedIn talent sourcing: run Apify searches and view sourced profiles",
|
|
modules=(_M.TALENT,),
|
|
roles=_STAFF,
|
|
),
|
|
"hiring_forms": Bundle(
|
|
"Fill and amend candidate hiring forms (requisition, interview analysis, cultural fit)",
|
|
tags=(_T.INTERVIEWS_CREATE, _T.INTERVIEWS_EDIT, _T.INTERVIEWS_DELETE),
|
|
roles=_STAFF,
|
|
),
|
|
"requisitions_management": Bundle(
|
|
"Employee requisition forms: view, create, edit and manage requisitions",
|
|
modules=(_M.REQUISITIONS,),
|
|
roles=_STAFF,
|
|
),
|
|
"manager_candidates": Bundle(
|
|
"Hiring manager: list candidates on own requisition jobs, view profiles, write notes",
|
|
tags=(_T.CANDIDATES_VIEW, _T.CANDIDATES_CREATE, _T.CANDIDATES_EDIT),
|
|
roles=(_R.HIRING_MANAGER,),
|
|
),
|
|
# Unattached: admins tick these on custom roles in Access Control.
|
|
"requisitions_self": Bundle(
|
|
"Own employee requisition forms: view, create, edit (not org-wide manage)",
|
|
tags=(_T.REQUISITIONS_VIEW, _T.REQUISITIONS_CREATE, _T.REQUISITIONS_EDIT),
|
|
),
|
|
"interviews_tab": Bundle(
|
|
"Interviews and Calendar tabs: list, schedule, reschedule",
|
|
tags=(_T.INTERVIEWS_VIEW, _T.INTERVIEWS_CREATE, _T.INTERVIEWS_EDIT),
|
|
),
|
|
"department_management": Bundle(
|
|
"Departments: view, create, edit and manage departments",
|
|
modules=(_M.DEPARTMENT,),
|
|
roles=(_R.SYSTEM_ADMINISTRATOR, _R.HR_ADMINISTRATOR),
|
|
),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RbacState:
|
|
"""What the database already holds, read before planning."""
|
|
|
|
tags: frozenset[str]
|
|
roles: frozenset[str]
|
|
bundles: frozenset[str]
|
|
ledger: frozenset[tuple[str, str]] # (kind, key)
|
|
|
|
|
|
def bundle_tag_names(bundle: Bundle) -> list[str]:
|
|
"""The bundle's tags in vocabulary order: whole modules plus explicit tags."""
|
|
modules = {m.value for m in bundle.modules}
|
|
explicit = set(bundle.tags)
|
|
return [t.value for t in PermissionTag if t.value.split(".")[0] in modules or t in explicit]
|
|
|
|
|
|
def _lit(value: str) -> str:
|
|
return "'" + value.replace("'", "''") + "'"
|
|
|
|
|
|
def _in(values: list[str]) -> str:
|
|
return "(" + ", ".join(_lit(v) for v in values) + ")"
|
|
|
|
|
|
def _rows(rows: list[str]) -> str:
|
|
return ",\n ".join(rows)
|
|
|
|
|
|
def build_rbac_sql(state: RbacState, *, schema: str | None) -> list[str]:
|
|
"""Idempotent statements that bring the database up to the code. Empty when in sync."""
|
|
|
|
def table(name: str) -> str:
|
|
return f'"{schema}".{name}' if schema else name
|
|
|
|
tags_t = table("permission_tags")
|
|
roles_t = table("roles")
|
|
perms_t = table("permissions")
|
|
ledger_t = table(RBAC_LEDGER_TABLE)
|
|
adopt = not state.ledger and bool(state.roles)
|
|
sql: list[str] = []
|
|
recorded: list[tuple[str, str]] = []
|
|
|
|
missing_tags = [t.value for t in PermissionTag if t.value not in state.tags]
|
|
if missing_tags:
|
|
values = _rows([
|
|
f"({_lit(tag)}, {_lit(tag.split('.')[0])}, {_lit(tag.split('.')[1])}, "
|
|
"NOW(), NOW(), true, false)"
|
|
for tag in missing_tags
|
|
])
|
|
sql.append(f"""\
|
|
INSERT INTO {tags_t}
|
|
(tag_name, module, action, created_at, updated_at, is_active, is_deleted)
|
|
VALUES
|
|
{values}
|
|
ON CONFLICT (tag_name) DO NOTHING;""")
|
|
|
|
missing_roles = [r for r in SYSTEM_ROLES if r.name.value not in state.roles]
|
|
if missing_roles:
|
|
values = _rows([
|
|
f"({r.id}, {_lit(r.name.value)}, {_lit(r.description)}, '[]'::jsonb, true, "
|
|
f"NOW(), NOW(), {str(not r.retired).lower()}, {str(r.retired).lower()})"
|
|
for r in missing_roles
|
|
])
|
|
sql.append(f"""\
|
|
INSERT INTO {roles_t}
|
|
(id, role_name, description, permissions, is_system,
|
|
created_at, updated_at, is_active, is_deleted)
|
|
VALUES
|
|
{values}
|
|
ON CONFLICT DO NOTHING;""")
|
|
# Pinned ids bypass the sequence; move it past them so new roles don't collide.
|
|
sql.append(
|
|
f"SELECT setval(pg_get_serial_sequence('{roles_t}', 'id'), "
|
|
f"GREATEST((SELECT MAX(id) FROM {roles_t}), 1));"
|
|
)
|
|
|
|
for name, bundle in RBAC_BUNDLES.items():
|
|
created = name not in state.bundles
|
|
if created:
|
|
sql.append(f"""\
|
|
INSERT INTO {perms_t}
|
|
(name, description, permission_tags, is_system,
|
|
created_at, updated_at, is_active, is_deleted)
|
|
VALUES
|
|
({_lit(name)}, {_lit(bundle.description)}, '[]'::jsonb, true, NOW(), NOW(), true, false)
|
|
ON CONFLICT (name) DO NOTHING;""")
|
|
apply = created or not adopt
|
|
|
|
tags = [
|
|
t for t in bundle_tag_names(bundle)
|
|
if ("bundle_tag", f"{name}:{t}") not in state.ledger
|
|
]
|
|
if tags:
|
|
recorded += [("bundle_tag", f"{name}:{t}") for t in tags]
|
|
if apply:
|
|
sql.append(f"""\
|
|
UPDATE {perms_t} p
|
|
SET permission_tags = COALESCE(p.permission_tags, '[]'::jsonb) || (
|
|
SELECT COALESCE(jsonb_agg(t.id ORDER BY t.id), '[]'::jsonb)
|
|
FROM {tags_t} t
|
|
WHERE t.tag_name IN {_in(tags)}
|
|
AND NOT (COALESCE(p.permission_tags, '[]'::jsonb) @> jsonb_build_array(t.id))
|
|
),
|
|
updated_at = NOW()
|
|
WHERE p.name = {_lit(name)};""")
|
|
|
|
roles = [
|
|
r.value for r in bundle.roles
|
|
if ("role_bundle", f"{r.value}:{name}") not in state.ledger
|
|
]
|
|
if roles:
|
|
recorded += [("role_bundle", f"{r}:{name}") for r in roles]
|
|
if apply:
|
|
sql.append(f"""\
|
|
UPDATE {roles_t} r
|
|
SET permissions = COALESCE(r.permissions, '[]'::jsonb) || jsonb_build_array(p.id),
|
|
updated_at = NOW()
|
|
FROM {perms_t} p
|
|
WHERE p.name = {_lit(name)}
|
|
AND r.role_name IN {_in(roles)}
|
|
AND NOT (COALESCE(r.permissions, '[]'::jsonb) @> jsonb_build_array(p.id));""")
|
|
|
|
if not sql and not recorded:
|
|
return []
|
|
|
|
ledger_sql = [f"""\
|
|
CREATE TABLE IF NOT EXISTS {ledger_t} (
|
|
kind text NOT NULL,
|
|
key text NOT NULL,
|
|
applied_at timestamptz NOT NULL DEFAULT now(),
|
|
PRIMARY KEY (kind, key)
|
|
);"""]
|
|
if recorded:
|
|
values = _rows([f"({_lit(kind)}, {_lit(key)})" for kind, key in recorded])
|
|
ledger_sql.append(f"""\
|
|
INSERT INTO {ledger_t} (kind, key)
|
|
VALUES
|
|
{values}
|
|
ON CONFLICT DO NOTHING;""")
|
|
return ledger_sql + sql
|