Move access permissions to access module
This commit is contained in:
@@ -1,9 +1,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable
|
||||
from dataclasses import dataclass
|
||||
from typing import Literal
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_access.backend.db.models import (
|
||||
Account,
|
||||
ExternalFunctionRoleAssignment,
|
||||
Function,
|
||||
FunctionAssignment,
|
||||
FunctionDelegation,
|
||||
@@ -22,18 +27,159 @@ from govoplan_access.backend.semantic import (
|
||||
identity_id_for_account,
|
||||
)
|
||||
from govoplan_core.core.access import AccessDecisionProvenance, AccessExplanationService, PrincipalRef
|
||||
from govoplan_core.core.idm import IdmDirectory, OrganizationFunctionAssignmentRef
|
||||
from govoplan_core.core.organizations import OrganizationDirectory
|
||||
from govoplan_core.db.session import get_database
|
||||
from govoplan_core.security.permissions import scopes_grant
|
||||
from govoplan_access.backend.permissions.catalog import expand_scopes, scopes_grant
|
||||
|
||||
|
||||
AccessRoleSourceType = Literal[
|
||||
"direct_role",
|
||||
"group_role",
|
||||
"legacy_function_role",
|
||||
"idm_function_role",
|
||||
"system_role",
|
||||
]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AccessRoleSourceExplanation:
|
||||
source_type: AccessRoleSourceType
|
||||
role_id: str
|
||||
role_slug: str
|
||||
role_name: str
|
||||
permissions: tuple[str, ...]
|
||||
tenant_id: str | None = None
|
||||
group_id: str | None = None
|
||||
group_name: str | None = None
|
||||
function_assignment_id: str | None = None
|
||||
function_id: str | None = None
|
||||
function_name: str | None = None
|
||||
organization_unit_id: str | None = None
|
||||
organization_unit_name: str | None = None
|
||||
identity_id: str | None = None
|
||||
account_id: str | None = None
|
||||
source_module: str | None = None
|
||||
assignment_source: str | None = None
|
||||
applies_to_subunits: bool = False
|
||||
delegated_from_assignment_id: str | None = None
|
||||
delegation_id: str | None = None
|
||||
acting_for_account_id: str | None = None
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"source_type": self.source_type,
|
||||
"role_id": self.role_id,
|
||||
"role_slug": self.role_slug,
|
||||
"role_name": self.role_name,
|
||||
"permissions": list(self.permissions),
|
||||
"tenant_id": self.tenant_id,
|
||||
"group_id": self.group_id,
|
||||
"group_name": self.group_name,
|
||||
"function_assignment_id": self.function_assignment_id,
|
||||
"function_id": self.function_id,
|
||||
"function_name": self.function_name,
|
||||
"organization_unit_id": self.organization_unit_id,
|
||||
"organization_unit_name": self.organization_unit_name,
|
||||
"identity_id": self.identity_id,
|
||||
"account_id": self.account_id,
|
||||
"source_module": self.source_module,
|
||||
"assignment_source": self.assignment_source,
|
||||
"applies_to_subunits": self.applies_to_subunits,
|
||||
"delegated_from_assignment_id": self.delegated_from_assignment_id,
|
||||
"delegation_id": self.delegation_id,
|
||||
"acting_for_account_id": self.acting_for_account_id,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AccessScopeExplanation:
|
||||
scope: str
|
||||
sources: tuple[AccessRoleSourceExplanation, ...]
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {"scope": self.scope, "sources": [source.to_dict() for source in self.sources]}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class FunctionFactExplanation:
|
||||
source_module: str
|
||||
assignment_id: str
|
||||
tenant_id: str
|
||||
identity_id: str | None
|
||||
account_id: str | None
|
||||
function_id: str
|
||||
function_name: str | None
|
||||
organization_unit_id: str
|
||||
organization_unit_name: str | None
|
||||
applies_to_subunits: bool
|
||||
assignment_source: str
|
||||
status: str
|
||||
delegated_from_assignment_id: str | None = None
|
||||
acting_for_account_id: str | None = None
|
||||
role_ids: tuple[str, ...] = ()
|
||||
role_names: tuple[str, ...] = ()
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"source_module": self.source_module,
|
||||
"assignment_id": self.assignment_id,
|
||||
"tenant_id": self.tenant_id,
|
||||
"identity_id": self.identity_id,
|
||||
"account_id": self.account_id,
|
||||
"function_id": self.function_id,
|
||||
"function_name": self.function_name,
|
||||
"organization_unit_id": self.organization_unit_id,
|
||||
"organization_unit_name": self.organization_unit_name,
|
||||
"applies_to_subunits": self.applies_to_subunits,
|
||||
"assignment_source": self.assignment_source,
|
||||
"status": self.status,
|
||||
"delegated_from_assignment_id": self.delegated_from_assignment_id,
|
||||
"acting_for_account_id": self.acting_for_account_id,
|
||||
"role_ids": list(self.role_ids),
|
||||
"role_names": list(self.role_names),
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class UserAccessExplanation:
|
||||
role_sources: tuple[AccessRoleSourceExplanation, ...] = ()
|
||||
scopes: tuple[AccessScopeExplanation, ...] = ()
|
||||
function_facts: tuple[FunctionFactExplanation, ...] = ()
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"role_sources": [source.to_dict() for source in self.role_sources],
|
||||
"scopes": [scope.to_dict() for scope in self.scopes],
|
||||
"function_facts": [fact.to_dict() for fact in self.function_facts],
|
||||
}
|
||||
|
||||
|
||||
class SqlAccessExplanationService(AccessExplanationService):
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
idm_directory: IdmDirectory | None = None,
|
||||
organization_directory: OrganizationDirectory | None = None,
|
||||
) -> None:
|
||||
self._idm_directory = idm_directory
|
||||
self._organization_directory = organization_directory
|
||||
|
||||
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))
|
||||
return tuple(
|
||||
_scope_provenance(
|
||||
session,
|
||||
principal,
|
||||
required_scope,
|
||||
idm_directory=self._idm_directory,
|
||||
organization_directory=self._organization_directory,
|
||||
)
|
||||
)
|
||||
|
||||
def explain_resource_provenance(
|
||||
self,
|
||||
@@ -47,7 +193,44 @@ class SqlAccessExplanationService(AccessExplanationService):
|
||||
return self.explain_scope_provenance(principal, action)
|
||||
|
||||
|
||||
def _scope_provenance(session: Session, principal: PrincipalRef, required_scope: str) -> list[AccessDecisionProvenance]:
|
||||
def build_user_access_explanation(
|
||||
session: Session,
|
||||
user: User,
|
||||
*,
|
||||
idm_directory: IdmDirectory | None = None,
|
||||
organization_directory: OrganizationDirectory | None = None,
|
||||
include_system: bool = False,
|
||||
) -> UserAccessExplanation:
|
||||
role_sources: list[AccessRoleSourceExplanation] = []
|
||||
role_sources.extend(_direct_role_sources(session, user))
|
||||
role_sources.extend(_group_role_sources(session, user))
|
||||
role_sources.extend(_legacy_function_role_sources(session, user))
|
||||
idm_sources, function_facts = _idm_function_role_sources(
|
||||
session,
|
||||
user,
|
||||
idm_directory=idm_directory,
|
||||
organization_directory=organization_directory,
|
||||
)
|
||||
role_sources.extend(idm_sources)
|
||||
if include_system:
|
||||
account = session.get(Account, user.account_id)
|
||||
if account is not None:
|
||||
role_sources.extend(_system_role_sources(session, account))
|
||||
return UserAccessExplanation(
|
||||
role_sources=tuple(role_sources),
|
||||
scopes=_scope_explanations(role_sources),
|
||||
function_facts=tuple(function_facts),
|
||||
)
|
||||
|
||||
|
||||
def _scope_provenance(
|
||||
session: Session,
|
||||
principal: PrincipalRef,
|
||||
required_scope: str,
|
||||
*,
|
||||
idm_directory: IdmDirectory | None = None,
|
||||
organization_directory: OrganizationDirectory | None = None,
|
||||
) -> 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)
|
||||
@@ -72,46 +255,46 @@ def _scope_provenance(session: Session, principal: PrincipalRef, required_scope:
|
||||
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)
|
||||
explanation = build_user_access_explanation(
|
||||
session,
|
||||
user,
|
||||
idm_directory=idm_directory,
|
||||
organization_directory=organization_directory,
|
||||
include_system=False,
|
||||
)
|
||||
_append_source_provenance(
|
||||
items,
|
||||
[source for source in explanation.role_sources if scopes_grant(source.permissions, required_scope)],
|
||||
)
|
||||
if account is not None:
|
||||
_append_system_role_provenance(session, items, account, required_scope)
|
||||
for source in _system_role_sources(session, account):
|
||||
if scopes_grant(source.permissions, required_scope):
|
||||
_append_source_provenance(items, [source])
|
||||
items.append(AccessDecisionProvenance(kind="right", id=required_scope, label=required_scope))
|
||||
return items
|
||||
return _dedupe_provenance(items)
|
||||
|
||||
|
||||
def _append_direct_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
user: User,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
def _direct_role_sources(session: Session, user: User) -> list[AccessRoleSourceExplanation]:
|
||||
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",
|
||||
)
|
||||
)
|
||||
return [
|
||||
AccessRoleSourceExplanation(
|
||||
source_type="direct_role",
|
||||
role_id=role.id,
|
||||
role_slug=role.slug,
|
||||
role_name=role.name,
|
||||
permissions=tuple(role.permissions or ()),
|
||||
tenant_id=user.tenant_id,
|
||||
)
|
||||
for role in roles
|
||||
]
|
||||
|
||||
|
||||
def _append_group_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
user: User,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
def _group_role_sources(session: Session, user: User) -> list[AccessRoleSourceExplanation]:
|
||||
rows = (
|
||||
session.query(Group, Role)
|
||||
.join(UserGroupMembership, UserGroupMembership.group_id == Group.id)
|
||||
@@ -125,132 +308,296 @@ def _append_group_role_provenance(
|
||||
)
|
||||
.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},
|
||||
)
|
||||
)
|
||||
return [
|
||||
AccessRoleSourceExplanation(
|
||||
source_type="group_role",
|
||||
role_id=role.id,
|
||||
role_slug=role.slug,
|
||||
role_name=role.name,
|
||||
permissions=tuple(role.permissions or ()),
|
||||
tenant_id=user.tenant_id,
|
||||
group_id=group.id,
|
||||
group_name=group.name,
|
||||
)
|
||||
for group, role in rows
|
||||
]
|
||||
|
||||
|
||||
def _append_function_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
user: User,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
def _legacy_function_role_sources(session: Session, user: User) -> list[AccessRoleSourceExplanation]:
|
||||
sources: list[AccessRoleSourceExplanation] = []
|
||||
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")
|
||||
sources.extend(_legacy_function_assignment_sources(session, assignment, assignment_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,
|
||||
},
|
||||
sources.extend(
|
||||
_legacy_function_assignment_sources(
|
||||
session,
|
||||
assignment,
|
||||
assignment_source="delegated_function",
|
||||
delegation=delegation,
|
||||
)
|
||||
)
|
||||
_append_function_assignment_provenance(session, items, assignment, required_scope, source="delegated_function")
|
||||
return sources
|
||||
|
||||
|
||||
def _append_function_assignment_provenance(
|
||||
def _legacy_function_assignment_sources(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
assignment: FunctionAssignment,
|
||||
required_scope: str,
|
||||
*,
|
||||
source: str,
|
||||
) -> None:
|
||||
assignment_source: str,
|
||||
delegation: FunctionDelegation | None = None,
|
||||
) -> list[AccessRoleSourceExplanation]:
|
||||
function = session.get(Function, assignment.function_id)
|
||||
if function is None:
|
||||
return
|
||||
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,
|
||||
return [
|
||||
AccessRoleSourceExplanation(
|
||||
source_type="legacy_function_role",
|
||||
role_id=role.id,
|
||||
role_slug=role.slug,
|
||||
role_name=role.name,
|
||||
permissions=tuple(role.permissions or ()),
|
||||
tenant_id=assignment.tenant_id,
|
||||
source=source,
|
||||
details={"applies_to_subunits": assignment.applies_to_subunits},
|
||||
function_assignment_id=assignment.id,
|
||||
function_id=function.id,
|
||||
function_name=function.name,
|
||||
organization_unit_id=assignment.organization_unit_id,
|
||||
identity_id=assignment.identity_id,
|
||||
account_id=assignment.account_id,
|
||||
assignment_source=assignment_source,
|
||||
applies_to_subunits=assignment.applies_to_subunits,
|
||||
delegated_from_assignment_id=assignment.delegated_from_assignment_id,
|
||||
delegation_id=delegation.id if delegation else None,
|
||||
acting_for_account_id=assignment.acting_for_account_id,
|
||||
)
|
||||
)
|
||||
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 roles
|
||||
]
|
||||
|
||||
|
||||
def _idm_function_role_sources(
|
||||
session: Session,
|
||||
user: User,
|
||||
*,
|
||||
idm_directory: IdmDirectory | None,
|
||||
organization_directory: OrganizationDirectory | None,
|
||||
) -> tuple[list[AccessRoleSourceExplanation], list[FunctionFactExplanation]]:
|
||||
if idm_directory is None:
|
||||
return [], []
|
||||
assignments = [
|
||||
assignment
|
||||
for assignment in idm_directory.organization_function_assignments_for_account(user.account_id, tenant_id=user.tenant_id)
|
||||
if assignment.tenant_id == user.tenant_id and assignment.status == "active"
|
||||
]
|
||||
if not assignments:
|
||||
return [], []
|
||||
function_ids = sorted({assignment.function_id for assignment in assignments})
|
||||
rows = (
|
||||
session.query(ExternalFunctionRoleAssignment, Role)
|
||||
.join(Role, ExternalFunctionRoleAssignment.role_id == Role.id)
|
||||
.filter(
|
||||
ExternalFunctionRoleAssignment.tenant_id == user.tenant_id,
|
||||
ExternalFunctionRoleAssignment.function_id.in_(function_ids),
|
||||
Role.tenant_id == user.tenant_id,
|
||||
)
|
||||
.order_by(Role.name.asc())
|
||||
.all()
|
||||
)
|
||||
for role in matching:
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="role",
|
||||
id=role.id,
|
||||
label=role.name,
|
||||
mappings_by_function: dict[str, list[tuple[ExternalFunctionRoleAssignment, Role]]] = {}
|
||||
for mapping, role in rows:
|
||||
mappings_by_function.setdefault(mapping.function_id, []).append((mapping, role))
|
||||
|
||||
role_sources: list[AccessRoleSourceExplanation] = []
|
||||
function_facts: list[FunctionFactExplanation] = []
|
||||
for assignment in assignments:
|
||||
function_name, organization_unit_name = _organization_labels(organization_directory, assignment)
|
||||
mappings = mappings_by_function.get(assignment.function_id, [])
|
||||
function_facts.append(
|
||||
FunctionFactExplanation(
|
||||
source_module="organizations",
|
||||
assignment_id=assignment.id,
|
||||
tenant_id=assignment.tenant_id,
|
||||
source=source,
|
||||
details={"function_assignment_id": assignment.id, "function_id": function.id},
|
||||
identity_id=assignment.identity_id,
|
||||
account_id=assignment.account_id,
|
||||
function_id=assignment.function_id,
|
||||
function_name=function_name,
|
||||
organization_unit_id=assignment.organization_unit_id,
|
||||
organization_unit_name=organization_unit_name,
|
||||
applies_to_subunits=assignment.applies_to_subunits,
|
||||
assignment_source=assignment.source,
|
||||
status=assignment.status,
|
||||
delegated_from_assignment_id=assignment.delegated_from_assignment_id,
|
||||
acting_for_account_id=assignment.acting_for_account_id,
|
||||
role_ids=tuple(role.id for _, role in mappings),
|
||||
role_names=tuple(role.name for _, role in mappings),
|
||||
)
|
||||
)
|
||||
for mapping, role in mappings:
|
||||
role_sources.append(
|
||||
AccessRoleSourceExplanation(
|
||||
source_type="idm_function_role",
|
||||
role_id=role.id,
|
||||
role_slug=role.slug,
|
||||
role_name=role.name,
|
||||
permissions=tuple(role.permissions or ()),
|
||||
tenant_id=assignment.tenant_id,
|
||||
function_assignment_id=assignment.id,
|
||||
function_id=assignment.function_id,
|
||||
function_name=function_name,
|
||||
organization_unit_id=assignment.organization_unit_id,
|
||||
organization_unit_name=organization_unit_name,
|
||||
identity_id=assignment.identity_id,
|
||||
account_id=assignment.account_id,
|
||||
source_module=mapping.source_module,
|
||||
assignment_source=assignment.source,
|
||||
applies_to_subunits=assignment.applies_to_subunits,
|
||||
delegated_from_assignment_id=assignment.delegated_from_assignment_id,
|
||||
acting_for_account_id=assignment.acting_for_account_id,
|
||||
)
|
||||
)
|
||||
return role_sources, function_facts
|
||||
|
||||
|
||||
def _append_system_role_provenance(
|
||||
session: Session,
|
||||
items: list[AccessDecisionProvenance],
|
||||
account: Account,
|
||||
required_scope: str,
|
||||
) -> None:
|
||||
def _organization_labels(
|
||||
organization_directory: OrganizationDirectory | None,
|
||||
assignment: OrganizationFunctionAssignmentRef,
|
||||
) -> tuple[str | None, str | None]:
|
||||
if organization_directory is None:
|
||||
return None, None
|
||||
function = organization_directory.get_function(assignment.function_id)
|
||||
unit = organization_directory.get_organization_unit(assignment.organization_unit_id)
|
||||
return (function.name if function is not None else None, unit.name if unit is not None else None)
|
||||
|
||||
|
||||
def _system_role_sources(session: Session, account: Account) -> list[AccessRoleSourceExplanation]:
|
||||
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):
|
||||
return [
|
||||
AccessRoleSourceExplanation(
|
||||
source_type="system_role",
|
||||
role_id=role.id,
|
||||
role_slug=role.slug,
|
||||
role_name=role.name,
|
||||
permissions=tuple(role.permissions or ()),
|
||||
)
|
||||
for role in roles
|
||||
]
|
||||
|
||||
|
||||
def _scope_explanations(role_sources: Iterable[AccessRoleSourceExplanation]) -> tuple[AccessScopeExplanation, ...]:
|
||||
sources_by_scope: dict[str, list[AccessRoleSourceExplanation]] = {}
|
||||
for source in role_sources:
|
||||
for scope in expand_scopes(source.permissions):
|
||||
sources_by_scope.setdefault(scope, []).append(source)
|
||||
return tuple(
|
||||
AccessScopeExplanation(scope=scope, sources=tuple(sources))
|
||||
for scope, sources in sorted(sources_by_scope.items(), key=lambda item: item[0])
|
||||
)
|
||||
|
||||
|
||||
def _append_source_provenance(
|
||||
items: list[AccessDecisionProvenance],
|
||||
sources: Iterable[AccessRoleSourceExplanation],
|
||||
) -> None:
|
||||
for source in sources:
|
||||
if source.group_id:
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="role",
|
||||
id=role.id,
|
||||
label=role.name,
|
||||
source="system_role",
|
||||
kind="group",
|
||||
id=source.group_id,
|
||||
label=source.group_name,
|
||||
tenant_id=source.tenant_id,
|
||||
source="group_membership",
|
||||
)
|
||||
)
|
||||
if source.delegation_id:
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="delegation",
|
||||
id=source.delegation_id,
|
||||
label=source.assignment_source,
|
||||
tenant_id=source.tenant_id,
|
||||
source=source.source_type,
|
||||
details={
|
||||
"function_assignment_id": source.function_assignment_id,
|
||||
"delegated_from_assignment_id": source.delegated_from_assignment_id,
|
||||
"acting_for_account_id": source.acting_for_account_id,
|
||||
},
|
||||
)
|
||||
)
|
||||
if source.function_id:
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="organization_unit",
|
||||
id=source.organization_unit_id,
|
||||
label=source.organization_unit_name,
|
||||
tenant_id=source.tenant_id,
|
||||
source=source.source_type,
|
||||
details={"applies_to_subunits": source.applies_to_subunits},
|
||||
)
|
||||
)
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="function",
|
||||
id=source.function_assignment_id,
|
||||
label=source.function_name,
|
||||
tenant_id=source.tenant_id,
|
||||
source=source.source_type,
|
||||
details={
|
||||
"function_id": source.function_id,
|
||||
"organization_unit_id": source.organization_unit_id,
|
||||
"identity_id": source.identity_id,
|
||||
"account_id": source.account_id,
|
||||
"source_module": source.source_module,
|
||||
"assignment_source": source.assignment_source,
|
||||
},
|
||||
)
|
||||
)
|
||||
items.append(
|
||||
AccessDecisionProvenance(
|
||||
kind="role",
|
||||
id=source.role_id,
|
||||
label=source.role_name,
|
||||
tenant_id=source.tenant_id,
|
||||
source=source.source_type,
|
||||
details={
|
||||
"role_slug": source.role_slug,
|
||||
"group_id": source.group_id,
|
||||
"function_assignment_id": source.function_assignment_id,
|
||||
"function_id": source.function_id,
|
||||
"source_module": source.source_module,
|
||||
},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _dedupe_provenance(items: Iterable[AccessDecisionProvenance]) -> list[AccessDecisionProvenance]:
|
||||
seen: set[tuple[object, ...]] = set()
|
||||
deduped: list[AccessDecisionProvenance] = []
|
||||
for item in items:
|
||||
key = (
|
||||
item.kind,
|
||||
item.id,
|
||||
item.tenant_id,
|
||||
item.source,
|
||||
tuple(sorted((str(key), str(value)) for key, value in item.details.items())),
|
||||
)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
deduped.append(item)
|
||||
return deduped
|
||||
|
||||
Reference in New Issue
Block a user