113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.models import (
|
|
Account,
|
|
Group,
|
|
GroupRoleAssignment,
|
|
Role,
|
|
SystemRoleAssignment,
|
|
User,
|
|
UserGroupMembership,
|
|
UserRoleAssignment,
|
|
)
|
|
from govoplan_access.backend.permissions.evaluator import expand_scopes
|
|
from govoplan_core.core.modules import PermissionDefinition, SubjectType
|
|
|
|
|
|
def membership_group_ids(session: Session, *, tenant_id: str, membership_id: str) -> frozenset[str]:
|
|
rows = (
|
|
session.query(Group.id)
|
|
.join(UserGroupMembership, UserGroupMembership.group_id == Group.id)
|
|
.filter(
|
|
UserGroupMembership.tenant_id == tenant_id,
|
|
UserGroupMembership.user_id == membership_id,
|
|
Group.is_active.is_(True),
|
|
)
|
|
.all()
|
|
)
|
|
return frozenset(row[0] for row in rows)
|
|
|
|
|
|
def roles_for_subject(
|
|
session: Session,
|
|
*,
|
|
subject_type: SubjectType,
|
|
subject_id: str,
|
|
tenant_id: str | None,
|
|
) -> list[Role]:
|
|
if tenant_id is None and subject_type == "account":
|
|
return (
|
|
session.query(Role)
|
|
.join(SystemRoleAssignment, SystemRoleAssignment.role_id == Role.id)
|
|
.filter(SystemRoleAssignment.account_id == subject_id, Role.tenant_id.is_(None))
|
|
.order_by(Role.name.asc())
|
|
.all()
|
|
)
|
|
if tenant_id is None:
|
|
return []
|
|
if subject_type == "membership":
|
|
return (
|
|
session.query(Role)
|
|
.join(UserRoleAssignment, UserRoleAssignment.role_id == Role.id)
|
|
.filter(
|
|
UserRoleAssignment.tenant_id == tenant_id,
|
|
UserRoleAssignment.user_id == subject_id,
|
|
Role.tenant_id == tenant_id,
|
|
)
|
|
.order_by(Role.name.asc())
|
|
.all()
|
|
)
|
|
if subject_type == "group":
|
|
return (
|
|
session.query(Role)
|
|
.join(GroupRoleAssignment, GroupRoleAssignment.role_id == Role.id)
|
|
.filter(
|
|
GroupRoleAssignment.tenant_id == tenant_id,
|
|
GroupRoleAssignment.group_id == subject_id,
|
|
Role.tenant_id == tenant_id,
|
|
)
|
|
.order_by(Role.name.asc())
|
|
.all()
|
|
)
|
|
return []
|
|
|
|
|
|
def membership_roles(session: Session, membership: User) -> list[Role]:
|
|
roles_by_id = {
|
|
role.id: role
|
|
for role in roles_for_subject(
|
|
session,
|
|
subject_type="membership",
|
|
subject_id=membership.id,
|
|
tenant_id=membership.tenant_id,
|
|
)
|
|
}
|
|
for group_id in membership_group_ids(session, tenant_id=membership.tenant_id, membership_id=membership.id):
|
|
for role in roles_for_subject(session, subject_type="group", subject_id=group_id, tenant_id=membership.tenant_id):
|
|
roles_by_id[role.id] = role
|
|
return list(roles_by_id.values())
|
|
|
|
|
|
def account_system_roles(session: Session, account: Account) -> list[Role]:
|
|
return roles_for_subject(session, subject_type="account", subject_id=account.id, tenant_id=None)
|
|
|
|
|
|
def principal_scopes(
|
|
session: Session,
|
|
*,
|
|
account: Account,
|
|
membership: User | None,
|
|
include_system: bool,
|
|
catalog: dict[str, PermissionDefinition],
|
|
) -> list[str]:
|
|
scopes: set[str] = set()
|
|
if membership is not None:
|
|
for role in membership_roles(session, membership):
|
|
scopes.update(role.permissions or [])
|
|
if include_system:
|
|
for role in account_system_roles(session, account):
|
|
scopes.update(role.permissions or [])
|
|
return expand_scopes(scopes, catalog=catalog)
|