feat: add access module boundary migrations
This commit is contained in:
256
src/govoplan_access/backend/explanation.py
Normal file
256
src/govoplan_access/backend/explanation.py
Normal file
@@ -0,0 +1,256 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_access.backend.db.models import (
|
||||
Account,
|
||||
Function,
|
||||
FunctionAssignment,
|
||||
FunctionDelegation,
|
||||
FunctionRoleAssignment,
|
||||
Group,
|
||||
GroupRoleAssignment,
|
||||
Role,
|
||||
SystemRoleAssignment,
|
||||
User,
|
||||
UserGroupMembership,
|
||||
UserRoleAssignment,
|
||||
)
|
||||
from govoplan_access.backend.semantic import (
|
||||
active_function_assignments_for_account,
|
||||
active_function_delegations_for_account,
|
||||
identity_id_for_account,
|
||||
)
|
||||
from govoplan_core.core.access import AccessDecisionProvenance, AccessExplanationService, PrincipalRef
|
||||
from govoplan_core.db.session import get_database
|
||||
from govoplan_core.security.permissions import scopes_grant
|
||||
|
||||
|
||||
class SqlAccessExplanationService(AccessExplanationService):
|
||||
def explain_scope_provenance(
|
||||
self,
|
||||
principal: PrincipalRef,
|
||||
required_scope: str,
|
||||
) -> tuple[AccessDecisionProvenance, ...]:
|
||||
with get_database().session() as session:
|
||||
return tuple(_scope_provenance(session, principal, required_scope))
|
||||
|
||||
def explain_resource_provenance(
|
||||
self,
|
||||
principal: PrincipalRef,
|
||||
*,
|
||||
resource_type: str,
|
||||
resource_id: str,
|
||||
action: str,
|
||||
) -> tuple[AccessDecisionProvenance, ...]:
|
||||
del resource_type, resource_id
|
||||
return self.explain_scope_provenance(principal, action)
|
||||
|
||||
|
||||
def _scope_provenance(session: Session, principal: PrincipalRef, required_scope: str) -> list[AccessDecisionProvenance]:
|
||||
items: list[AccessDecisionProvenance] = []
|
||||
account = session.get(Account, principal.account_id)
|
||||
identity_id = principal.identity_id or identity_id_for_account(session, principal.account_id)
|
||||
if identity_id:
|
||||
items.append(AccessDecisionProvenance(kind="identity", id=identity_id, source="identity_account_link"))
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="account",
|
||||
id=principal.account_id,
|
||||
label=(account.display_name or account.email) if account else None,
|
||||
source=principal.auth_method,
|
||||
)
|
||||
)
|
||||
user = session.get(User, principal.membership_id) if principal.membership_id else None
|
||||
if user is not None:
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="tenant_membership",
|
||||
id=user.id,
|
||||
label=user.display_name or user.email,
|
||||
tenant_id=user.tenant_id,
|
||||
source="membership",
|
||||
)
|
||||
)
|
||||
_append_direct_role_provenance(session, items, user, required_scope)
|
||||
_append_group_role_provenance(session, items, user, required_scope)
|
||||
_append_function_role_provenance(session, items, user, required_scope)
|
||||
if account is not None:
|
||||
_append_system_role_provenance(session, items, account, required_scope)
|
||||
items.append(AccessDecisionProvenance(kind="right", id=required_scope, label=required_scope))
|
||||
return items
|
||||
|
||||
|
||||
def _append_direct_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
user: User,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
roles = (
|
||||
session.query(Role)
|
||||
.join(UserRoleAssignment, UserRoleAssignment.role_id == Role.id)
|
||||
.filter(UserRoleAssignment.tenant_id == user.tenant_id, UserRoleAssignment.user_id == user.id)
|
||||
.all()
|
||||
)
|
||||
for role in roles:
|
||||
if scopes_grant(role.permissions or [], required_scope):
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="role",
|
||||
id=role.id,
|
||||
label=role.name,
|
||||
tenant_id=user.tenant_id,
|
||||
source="direct_user_role",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _append_group_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
user: User,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
rows = (
|
||||
session.query(Group, Role)
|
||||
.join(UserGroupMembership, UserGroupMembership.group_id == Group.id)
|
||||
.join(GroupRoleAssignment, GroupRoleAssignment.group_id == Group.id)
|
||||
.join(Role, Role.id == GroupRoleAssignment.role_id)
|
||||
.filter(
|
||||
UserGroupMembership.tenant_id == user.tenant_id,
|
||||
UserGroupMembership.user_id == user.id,
|
||||
GroupRoleAssignment.tenant_id == user.tenant_id,
|
||||
Group.is_active.is_(True),
|
||||
)
|
||||
.all()
|
||||
)
|
||||
for group, role in rows:
|
||||
if scopes_grant(role.permissions or [], required_scope):
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="group",
|
||||
id=group.id,
|
||||
label=group.name,
|
||||
tenant_id=user.tenant_id,
|
||||
source="group_membership",
|
||||
)
|
||||
)
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="role",
|
||||
id=role.id,
|
||||
label=role.name,
|
||||
tenant_id=user.tenant_id,
|
||||
source="group_role",
|
||||
details={"group_id": group.id},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _append_function_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
user: User,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
direct_assignments = active_function_assignments_for_account(session, user.account_id, tenant_id=user.tenant_id)
|
||||
for assignment in direct_assignments:
|
||||
_append_function_assignment_provenance(session, items, assignment, required_scope, source="function_assignment")
|
||||
|
||||
delegations = active_function_delegations_for_account(session, user.account_id, tenant_id=user.tenant_id, modes=("delegate",))
|
||||
for delegation in delegations:
|
||||
assignment = session.get(FunctionAssignment, delegation.function_assignment_id)
|
||||
if assignment is None:
|
||||
continue
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="delegation",
|
||||
id=delegation.id,
|
||||
label=delegation.mode,
|
||||
tenant_id=delegation.tenant_id,
|
||||
source="function_delegation",
|
||||
details={
|
||||
"function_assignment_id": delegation.function_assignment_id,
|
||||
"delegator_account_id": delegation.delegator_account_id,
|
||||
"delegate_account_id": delegation.delegate_account_id,
|
||||
},
|
||||
)
|
||||
)
|
||||
_append_function_assignment_provenance(session, items, assignment, required_scope, source="delegated_function")
|
||||
|
||||
|
||||
def _append_function_assignment_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
assignment: FunctionAssignment,
|
||||
required_scope: str,
|
||||
*,
|
||||
source: str,
|
||||
) -> None:
|
||||
function = session.get(Function, assignment.function_id)
|
||||
if function is None:
|
||||
return
|
||||
roles = (
|
||||
session.query(Role)
|
||||
.join(FunctionRoleAssignment, FunctionRoleAssignment.role_id == Role.id)
|
||||
.filter(FunctionRoleAssignment.tenant_id == assignment.tenant_id, FunctionRoleAssignment.function_id == function.id)
|
||||
.all()
|
||||
)
|
||||
matching = [role for role in roles if scopes_grant(role.permissions or [], required_scope)]
|
||||
if not matching:
|
||||
return
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="organization_unit",
|
||||
id=assignment.organization_unit_id,
|
||||
tenant_id=assignment.tenant_id,
|
||||
source=source,
|
||||
details={"applies_to_subunits": assignment.applies_to_subunits},
|
||||
)
|
||||
)
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="function",
|
||||
id=assignment.id,
|
||||
label=function.name,
|
||||
tenant_id=assignment.tenant_id,
|
||||
source=source,
|
||||
details={"function_id": function.id, "organization_unit_id": assignment.organization_unit_id},
|
||||
)
|
||||
)
|
||||
for role in matching:
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="role",
|
||||
id=role.id,
|
||||
label=role.name,
|
||||
tenant_id=assignment.tenant_id,
|
||||
source=source,
|
||||
details={"function_assignment_id": assignment.id, "function_id": function.id},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _append_system_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
account: Account,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
roles = (
|
||||
session.query(Role)
|
||||
.join(SystemRoleAssignment, SystemRoleAssignment.role_id == Role.id)
|
||||
.filter(SystemRoleAssignment.account_id == account.id, Role.tenant_id.is_(None))
|
||||
.all()
|
||||
)
|
||||
for role in roles:
|
||||
if scopes_grant(role.permissions or [], required_scope):
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="role",
|
||||
id=role.id,
|
||||
label=role.name,
|
||||
source="system_role",
|
||||
)
|
||||
)
|
||||
Reference in New Issue
Block a user