387 lines
14 KiB
Python
387 lines
14 KiB
Python
from sqlalchemy import func, or_, select
|
|
from typing import Set, Optional, List
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from app.core.request_cache import request_cached
|
|
from app.modules.auth.models.user_model import User
|
|
from app.modules.auth.models.access_model import Access
|
|
|
|
|
|
class PermissionService:
|
|
"""
|
|
Centralized permission engine for checking access codes.
|
|
|
|
Handles:
|
|
- Retrieving user's access codes from their role
|
|
- Checking if user has specific access code(s)
|
|
- Enforcing access restrictions (raises HTTP 403 on denial)
|
|
|
|
**This is the only correct answer to "what may this user do".** There used to
|
|
be a second one — the `User.access_codes` property — which read the single
|
|
legacy `users.role_id` and nothing else. Because `/api/me/profile` reported
|
|
that answer while `require_access` used this one, a user holding a role
|
|
through `user_roles` had authority the product would not show them: no menu
|
|
entry, no route, no button. `User.access_codes` now defers to whatever
|
|
`get_current_user` resolved through here, and
|
|
`tests/probes/test_multi_role.py` asserts the two agree.
|
|
"""
|
|
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def user_access_codes(self, user: Optional[User]) -> Set[str]:
|
|
"""
|
|
Get all access codes for a user.
|
|
Includes both local role-based permissions and SaaS-assigned permissions.
|
|
|
|
Memoised for the life of the request: `require_access` asks once per
|
|
dependency and `get_current_user` asks once more, and the answer cannot
|
|
change beneath a half-finished operation.
|
|
"""
|
|
if not user:
|
|
return set()
|
|
|
|
return request_cached(
|
|
f"access_codes:{user.id}", lambda: self._resolve_codes(user)
|
|
)
|
|
|
|
def _resolve_codes(self, user: User) -> Set[str]:
|
|
all_access_codes = set()
|
|
|
|
saas_permissions = getattr(user, "saas_permissions", set())
|
|
all_access_codes.update(saas_permissions)
|
|
|
|
direct_ids: Set[object] = set()
|
|
direct_codes: Set[str] = set()
|
|
|
|
if user.role:
|
|
for ra in user.role.role_accesses:
|
|
if ra.access:
|
|
direct_ids.add(ra.access.id)
|
|
direct_codes.add(ra.access.access_code)
|
|
|
|
for access in self._accesses_from_scoped_roles(user):
|
|
direct_ids.add(access.id)
|
|
direct_codes.add(access.access_code)
|
|
|
|
if direct_ids:
|
|
all_access_codes.update(
|
|
self._expand_with_descendants(direct_ids, direct_codes)
|
|
)
|
|
|
|
if any(code.startswith("document.conversion") for code in all_access_codes):
|
|
all_access_codes.add("document.conversion")
|
|
|
|
return all_access_codes
|
|
|
|
def _group_ids_subquery(self, user):
|
|
"""
|
|
The groups this user belongs to, as a **subquery** rather than a round trip.
|
|
|
|
Filtered by tenant on both sides: a membership row naming another
|
|
tenant's group must confer nothing, and `user_access_groups` carries its
|
|
own `tenant_id` precisely so a forged or stale row cannot reach across.
|
|
|
|
A subquery and not a `list` because this used to be one extra SELECT on
|
|
every permission check, and permission checks are the hottest path in
|
|
the application.
|
|
"""
|
|
from app.modules.org.models.group_model import AccessGroup, UserAccessGroup
|
|
|
|
return (
|
|
select(UserAccessGroup.group_id)
|
|
.join(AccessGroup, AccessGroup.id == UserAccessGroup.group_id)
|
|
.where(
|
|
UserAccessGroup.user_id == user.id,
|
|
UserAccessGroup.tenant_id == user.tenant_id,
|
|
AccessGroup.tenant_id == user.tenant_id,
|
|
)
|
|
.scalar_subquery()
|
|
)
|
|
|
|
def _group_ids_for(self, user) -> list:
|
|
"""Groups the user belongs to. Retained for callers that want the ids."""
|
|
from app.modules.org.models.group_model import AccessGroup, UserAccessGroup
|
|
|
|
rows = (
|
|
self.db.query(UserAccessGroup.group_id)
|
|
.join(AccessGroup, AccessGroup.id == UserAccessGroup.group_id)
|
|
.filter(
|
|
UserAccessGroup.user_id == user.id,
|
|
UserAccessGroup.tenant_id == user.tenant_id,
|
|
AccessGroup.tenant_id == user.tenant_id,
|
|
)
|
|
.all()
|
|
)
|
|
return [r[0] for r in rows]
|
|
|
|
def _accesses_from_scoped_roles(self, user: User) -> list:
|
|
"""
|
|
Every `Access` reachable through this user's `user_roles` rows.
|
|
|
|
Deliberately ignores `org_unit_id`: holding a role in one team still
|
|
means holding its codes. Narrowing to a subtree is `ScopeService`'s job,
|
|
and conflating the two here would mean an endpoint that has not been
|
|
converted to ask about scope silently refuses people who legitimately
|
|
hold the permission somewhere.
|
|
|
|
**The three predicates below are not optional.** This query originally
|
|
filtered on the principal alone, which meant the coarse gate disagreed
|
|
with `ScopeService` in two ways that both fail open:
|
|
|
|
- an **expired** grant still passed `require_access`, so temporary
|
|
elevation did not expire on any endpoint that had not been converted
|
|
to ask about scope — which is almost all of them;
|
|
- a grant naming **another tenant's role** conferred that role's codes.
|
|
This is the defect the C-series found in `ScopeService._resolve` and
|
|
fixed there; it lived on here for a release.
|
|
|
|
They are expressed in SQL rather than filtered afterwards for the same
|
|
reason `ScopeService` gives: an expired grant must be invisible to every
|
|
reader, and a post-filter is how one caller ends up honouring a dead
|
|
grant.
|
|
"""
|
|
from app.modules.auth.models.role_access_model import RoleAccess
|
|
from app.modules.auth.models.role_model import Role
|
|
from app.modules.auth.models.user_role_model import UserRole
|
|
|
|
return (
|
|
self.db.query(Access)
|
|
.join(RoleAccess, RoleAccess.access_id == Access.id)
|
|
.join(UserRole, UserRole.role_id == RoleAccess.role_id)
|
|
.join(Role, Role.id == UserRole.role_id)
|
|
.filter(
|
|
or_(
|
|
UserRole.user_id == user.id,
|
|
UserRole.group_id.in_(self._group_ids_subquery(user)),
|
|
),
|
|
or_(
|
|
Role.tenant_id == user.tenant_id,
|
|
Role.tenant_id.is_(None),
|
|
),
|
|
or_(
|
|
UserRole.expires_at.is_(None),
|
|
UserRole.expires_at > func.now(),
|
|
),
|
|
)
|
|
.distinct()
|
|
.all()
|
|
)
|
|
|
|
def _access_tree(self):
|
|
"""
|
|
`(code_by_id, children_by_parent)` for the whole catalogue.
|
|
|
|
One query, memoised per request. The catalogue is the same for every
|
|
user, so resolving a page of twenty users used to read this table twenty
|
|
times.
|
|
"""
|
|
|
|
def _load():
|
|
rows = self.db.query(
|
|
Access.id, Access.parent_id, Access.access_code
|
|
).all()
|
|
code_by_id = {row.id: row.access_code for row in rows}
|
|
children_by_parent: dict = {}
|
|
for row in rows:
|
|
if row.parent_id is None:
|
|
continue
|
|
children_by_parent.setdefault(row.parent_id, set()).add(row.id)
|
|
return code_by_id, children_by_parent
|
|
|
|
return request_cached("access_tree", _load)
|
|
|
|
def _expand_with_descendants(
|
|
self, direct_ids: Set[object], direct_codes: Set[str]
|
|
) -> Set[str]:
|
|
"""
|
|
Expand assigned access codes with all descendants so parent access grants
|
|
full module access.
|
|
"""
|
|
code_by_id, children_by_parent = self._access_tree()
|
|
|
|
visited = set(direct_ids)
|
|
stack = list(direct_ids)
|
|
|
|
while stack:
|
|
current = stack.pop()
|
|
for child_id in children_by_parent.get(current, set()):
|
|
if child_id in visited:
|
|
continue
|
|
visited.add(child_id)
|
|
stack.append(child_id)
|
|
|
|
expanded_codes = {code_by_id[access_id] for access_id in visited if access_id in code_by_id}
|
|
return direct_codes.union(expanded_codes)
|
|
|
|
def has_access(self, user: Optional[User], code: str) -> bool:
|
|
"""
|
|
Check if user has a specific access code.
|
|
|
|
Args:
|
|
user: Current user
|
|
code: Access code to check (e.g. "project.create")
|
|
|
|
Returns:
|
|
True if user has the access code, False otherwise
|
|
"""
|
|
return code in self.user_access_codes(user)
|
|
|
|
def require_access(self, user: Optional[User], code: str) -> None:
|
|
"""
|
|
Assert user has a specific access code.
|
|
Raises HTTPException 403 if not authorized.
|
|
|
|
Args:
|
|
user: Current user
|
|
code: Required access code
|
|
|
|
Raises:
|
|
HTTPException: 403 if user not authenticated or lacks access
|
|
"""
|
|
if not user:
|
|
raise HTTPException(status_code=403, detail="Authentication required")
|
|
|
|
if not self.has_access(user, code):
|
|
raise HTTPException(
|
|
status_code=403, detail=f"Access denied: insufficient permissions"
|
|
)
|
|
|
|
def require_any_access(self, user: Optional[User], codes: List[str]) -> None:
|
|
"""
|
|
Assert user has at least one of the provided access codes.
|
|
Raises HTTPException 403 if none match.
|
|
|
|
Args:
|
|
user: Current user
|
|
codes: List of access codes (user needs at least one)
|
|
|
|
Raises:
|
|
HTTPException: 403 if user lacks all codes
|
|
"""
|
|
if not user:
|
|
raise HTTPException(status_code=403, detail="Authentication required")
|
|
|
|
if not any(self.has_access(user, code) for code in codes):
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: insufficient permissions"
|
|
)
|
|
|
|
def require_all_access(self, user: Optional[User], codes: List[str]) -> None:
|
|
"""
|
|
Assert user has ALL of the provided access codes.
|
|
Raises HTTPException 403 if any are missing.
|
|
|
|
Args:
|
|
user: Current user
|
|
codes: List of access codes (user must have all)
|
|
|
|
Raises:
|
|
HTTPException: 403 if user lacks any code
|
|
"""
|
|
if not user:
|
|
raise HTTPException(status_code=403, detail="Authentication required")
|
|
|
|
missing = [code for code in codes if not self.has_access(user, code)]
|
|
if missing:
|
|
raise HTTPException(
|
|
status_code=403, detail="Access denied: insufficient permissions"
|
|
)
|
|
|
|
def get_all_system_accesses(self) -> List[dict]:
|
|
"""
|
|
Get all defined access codes in the system grouped by category.
|
|
"""
|
|
accesses = self.db.query(Access).all()
|
|
return [
|
|
{
|
|
"id": a.id,
|
|
"access_code": a.access_code,
|
|
"category": a.category,
|
|
"name": a.name,
|
|
"parent_id": a.parent_id,
|
|
}
|
|
for a in accesses
|
|
]
|
|
|
|
def granted_by(self, user: Optional[User]) -> dict:
|
|
"""
|
|
`{access_code: [attribution, ...]}` — which role conferred each code.
|
|
|
|
The answer to "why can this person do that", which
|
|
`/api/admin/users/{id}/effective-access` exposes. With one role per user
|
|
the question answered itself; with several it does not, and support
|
|
cannot read the database.
|
|
|
|
Attribution names the role, the scope, and whether it arrived through
|
|
the primary role or a grant. Descendant expansion is attributed to the
|
|
role that holds the parent code, because that is the row an
|
|
administrator would edit to take it away.
|
|
"""
|
|
if not user:
|
|
return {}
|
|
|
|
attributions: dict = {}
|
|
|
|
def _add(code: str, entry: dict) -> None:
|
|
existing = attributions.setdefault(code, [])
|
|
if entry not in existing:
|
|
existing.append(entry)
|
|
|
|
from app.modules.auth.services.user_role_service import UserRoleReader
|
|
|
|
for assignment in UserRoleReader(self.db).assignments(user):
|
|
direct_ids = set()
|
|
direct_codes = set()
|
|
for access in self._accesses_of_role(assignment["role_id"]):
|
|
direct_ids.add(access.id)
|
|
direct_codes.add(access.access_code)
|
|
if not direct_ids:
|
|
continue
|
|
entry = {
|
|
"role_id": str(assignment["role_id"]),
|
|
"role_name": assignment["role_name"],
|
|
"source": assignment["source"],
|
|
"org_unit_id": (
|
|
str(assignment["org_unit_id"])
|
|
if assignment["org_unit_id"]
|
|
else None
|
|
),
|
|
"org_unit_name": assignment["org_unit_name"],
|
|
"expires_at": (
|
|
assignment["expires_at"].isoformat()
|
|
if assignment["expires_at"]
|
|
else None
|
|
),
|
|
}
|
|
for code in self._expand_with_descendants(direct_ids, direct_codes):
|
|
_add(code, entry)
|
|
|
|
for code in getattr(user, "saas_permissions", set()) or set():
|
|
_add(
|
|
code,
|
|
{
|
|
"role_id": None,
|
|
"role_name": "Subscription platform",
|
|
"source": "saas",
|
|
"org_unit_id": None,
|
|
"org_unit_name": None,
|
|
"expires_at": None,
|
|
},
|
|
)
|
|
|
|
return attributions
|
|
|
|
def _accesses_of_role(self, role_id) -> list:
|
|
from app.modules.auth.models.role_access_model import RoleAccess
|
|
|
|
def _load():
|
|
return (
|
|
self.db.query(Access)
|
|
.join(RoleAccess, RoleAccess.access_id == Access.id)
|
|
.filter(RoleAccess.role_id == role_id)
|
|
.all()
|
|
)
|
|
|
|
return request_cached(f"role_accesses:{role_id}", _load)
|