651 lines
25 KiB
Python
651 lines
25 KiB
Python
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,
|
|
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, ResourceAccessExplanationProvider
|
|
from govoplan_core.core.identity import IdentityDirectory
|
|
from govoplan_core.core.idm import IdmDirectory, OrganizationFunctionAssignmentRef
|
|
from govoplan_core.core.organizations import ORGANIZATIONS_MODULE_ID, OrganizationDirectory
|
|
from govoplan_core.db.session import get_database
|
|
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,
|
|
*,
|
|
identity_directory: IdentityDirectory | None = None,
|
|
idm_directory: IdmDirectory | None = None,
|
|
organization_directory: OrganizationDirectory | None = None,
|
|
resource_explanation_providers: Iterable[ResourceAccessExplanationProvider] = (),
|
|
) -> None:
|
|
self._identity_directory = identity_directory
|
|
self._idm_directory = idm_directory
|
|
self._organization_directory = organization_directory
|
|
self._resource_explanation_providers = tuple(resource_explanation_providers)
|
|
|
|
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,
|
|
identity_directory=self._identity_directory,
|
|
idm_directory=self._idm_directory,
|
|
organization_directory=self._organization_directory,
|
|
)
|
|
)
|
|
|
|
def explain_resource_provenance(
|
|
self,
|
|
principal: PrincipalRef,
|
|
*,
|
|
resource_type: str,
|
|
resource_id: str,
|
|
action: str,
|
|
) -> tuple[AccessDecisionProvenance, ...]:
|
|
with get_database().session() as session:
|
|
items = _scope_provenance(
|
|
session,
|
|
principal,
|
|
action,
|
|
identity_directory=self._identity_directory,
|
|
idm_directory=self._idm_directory,
|
|
organization_directory=self._organization_directory,
|
|
)
|
|
for provider in self._resource_explanation_providers:
|
|
provider_items = tuple(
|
|
provider.explain_resource_provenance(
|
|
session,
|
|
principal,
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
action=action,
|
|
)
|
|
)
|
|
if provider_items:
|
|
items.extend(provider_items)
|
|
break
|
|
return tuple(_dedupe_provenance(items))
|
|
|
|
|
|
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,
|
|
*,
|
|
identity_directory: IdentityDirectory | None = None,
|
|
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,
|
|
identity_directory=identity_directory,
|
|
)
|
|
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",
|
|
)
|
|
)
|
|
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:
|
|
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 _dedupe_provenance(items)
|
|
|
|
|
|
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()
|
|
)
|
|
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 _group_role_sources(session: Session, user: User) -> list[AccessRoleSourceExplanation]:
|
|
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()
|
|
)
|
|
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 _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:
|
|
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
|
|
sources.extend(
|
|
_legacy_function_assignment_sources(
|
|
session,
|
|
assignment,
|
|
assignment_source="delegated_function",
|
|
delegation=delegation,
|
|
)
|
|
)
|
|
return sources
|
|
|
|
|
|
def _legacy_function_assignment_sources(
|
|
session: Session,
|
|
assignment: FunctionAssignment,
|
|
*,
|
|
assignment_source: str,
|
|
delegation: FunctionDelegation | None = None,
|
|
) -> list[AccessRoleSourceExplanation]:
|
|
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()
|
|
)
|
|
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,
|
|
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,
|
|
)
|
|
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 organization_directory is not None:
|
|
assignments = [
|
|
assignment
|
|
for assignment in assignments
|
|
if _organization_function_active(organization_directory, assignment)
|
|
]
|
|
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.source_module == ORGANIZATIONS_MODULE_ID,
|
|
ExternalFunctionRoleAssignment.function_id.in_(function_ids),
|
|
Role.tenant_id == user.tenant_id,
|
|
)
|
|
.order_by(Role.name.asc())
|
|
.all()
|
|
)
|
|
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_MODULE_ID,
|
|
assignment_id=assignment.id,
|
|
tenant_id=assignment.tenant_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 _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 _organization_function_active(
|
|
organization_directory: OrganizationDirectory,
|
|
assignment: OrganizationFunctionAssignmentRef,
|
|
) -> bool:
|
|
function = organization_directory.get_function(assignment.function_id)
|
|
return function is not None and function.tenant_id == assignment.tenant_id and function.status == "active"
|
|
|
|
|
|
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()
|
|
)
|
|
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="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
|