248 lines
8.5 KiB
Python
248 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.dsar import (
|
|
DsarErasureActionRef,
|
|
DsarExecutionResultRef,
|
|
DsarRecordRef,
|
|
DsarSubjectRef,
|
|
dsar_capability_name,
|
|
)
|
|
from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
|
|
|
|
|
|
IDENTITY_DSAR_CAPABILITY = dsar_capability_name("identity")
|
|
_CONFLICT = object()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _SubjectSelectors:
|
|
identity_id: str
|
|
account_id: str | None
|
|
link_id: str | None
|
|
|
|
|
|
class IdentityDsarProvider:
|
|
provider_id = "identity"
|
|
module_id = "identity"
|
|
|
|
def search_subject(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
) -> Sequence[DsarRecordRef]:
|
|
del tenant_id
|
|
db = _session(session)
|
|
selectors = _subject_selectors(subject)
|
|
if selectors is None:
|
|
return ()
|
|
query = (
|
|
db.query(Identity, IdentityAccountLink)
|
|
.join(
|
|
IdentityAccountLink,
|
|
IdentityAccountLink.identity_id == Identity.id,
|
|
)
|
|
.filter(Identity.id == selectors.identity_id)
|
|
)
|
|
if selectors.account_id:
|
|
query = query.filter(
|
|
IdentityAccountLink.account_id == selectors.account_id
|
|
)
|
|
if selectors.link_id:
|
|
query = query.filter(IdentityAccountLink.id == selectors.link_id)
|
|
matches = query.limit(2).all()
|
|
if len(matches) != 1:
|
|
return ()
|
|
identity, link = matches[0]
|
|
return (_identity_record(identity, link),)
|
|
|
|
def plan_erasure(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
records: Sequence[DsarRecordRef],
|
|
) -> Sequence[DsarErasureActionRef]:
|
|
del tenant_id
|
|
_session(session)
|
|
selectors = _subject_selectors(subject)
|
|
if selectors is None:
|
|
raise ValueError("Identity DSAR subject selectors conflict or are incomplete.")
|
|
actions: list[DsarErasureActionRef] = []
|
|
for record in records:
|
|
_validate_record(record)
|
|
if record.resource_id != selectors.identity_id:
|
|
raise ValueError("Identity DSAR record does not match the subject.")
|
|
actions.append(
|
|
DsarErasureActionRef(
|
|
action_id=f"identity:manual_review:canonical_identity:{record.resource_id}",
|
|
provider_id=self.provider_id,
|
|
module_id=self.module_id,
|
|
kind="manual_review",
|
|
resource_type=record.resource_type,
|
|
resource_id=record.resource_id,
|
|
title=f"Review {record.title}",
|
|
rationale=(
|
|
"Canonical identities and account links are system-scoped and may "
|
|
"support authentication or memberships in more than one tenant. "
|
|
"Identity, Access, and tenancy owners must review deactivation, "
|
|
"unlinking, or minimization together."
|
|
),
|
|
executable=False,
|
|
)
|
|
)
|
|
return tuple(actions)
|
|
|
|
def execute_erasure(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
actions: Sequence[DsarErasureActionRef],
|
|
request_id: str,
|
|
) -> Sequence[DsarExecutionResultRef]:
|
|
del tenant_id
|
|
_session(session)
|
|
if _subject_selectors(subject) is None:
|
|
raise ValueError("Identity DSAR subject selectors conflict or are incomplete.")
|
|
results: list[DsarExecutionResultRef] = []
|
|
for action in actions:
|
|
_validate_action(action)
|
|
if action.executable or action.kind != "manual_review":
|
|
raise ValueError("Identity DSAR publishes manual-review actions only.")
|
|
results.append(
|
|
DsarExecutionResultRef(
|
|
action_id=action.action_id,
|
|
status="blocked",
|
|
summary=(
|
|
"The system identity and account link remain unchanged pending "
|
|
"cross-tenant identity, authentication, and retention review."
|
|
),
|
|
evidence={"request_id": request_id},
|
|
)
|
|
)
|
|
return tuple(results)
|
|
|
|
|
|
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
|
|
references = subject.external_references
|
|
identity_id = _coalesce(
|
|
subject.identity_id,
|
|
references.get("identity.id"),
|
|
references.get("identity.identity"),
|
|
)
|
|
account_id = _coalesce(
|
|
subject.account_id,
|
|
references.get("identity.account"),
|
|
references.get("access.account"),
|
|
)
|
|
link_id = _coalesce(
|
|
references.get("identity.link"),
|
|
references.get("identity.account_link"),
|
|
)
|
|
if _CONFLICT in (identity_id, account_id, link_id):
|
|
return None
|
|
normalized_identity = _optional_string(identity_id)
|
|
normalized_account = _optional_string(account_id)
|
|
normalized_link = _optional_string(link_id)
|
|
if not normalized_identity or not (normalized_account or normalized_link):
|
|
return None
|
|
return _SubjectSelectors(
|
|
identity_id=normalized_identity,
|
|
account_id=normalized_account,
|
|
link_id=normalized_link,
|
|
)
|
|
|
|
|
|
def _identity_record(
|
|
identity: Identity,
|
|
link: IdentityAccountLink,
|
|
) -> DsarRecordRef:
|
|
observed = max(
|
|
value for value in (identity.updated_at, link.updated_at) if value is not None
|
|
)
|
|
return DsarRecordRef(
|
|
provider_id="identity",
|
|
module_id="identity",
|
|
resource_type="canonical_identity",
|
|
resource_id=identity.id,
|
|
category="system_identity_and_account_link",
|
|
title="Canonical identity and corroborated account link",
|
|
data={
|
|
"identity_id": identity.id,
|
|
"display_name": (identity.display_name or "")[:255] or None,
|
|
"external_subject": (identity.external_subject or "")[:255] or None,
|
|
"source": identity.source,
|
|
"is_active": identity.is_active,
|
|
"created_at": _iso(identity.created_at),
|
|
"updated_at": _iso(identity.updated_at),
|
|
"matching_account_link": {
|
|
"id": link.id,
|
|
"account_id": link.account_id,
|
|
"is_primary": link.is_primary,
|
|
"source": link.source,
|
|
"created_at": _iso(link.created_at),
|
|
"updated_at": _iso(link.updated_at),
|
|
},
|
|
},
|
|
observed_at=_aware(observed),
|
|
retention_reason=(
|
|
"The canonical identity and account link are system-scoped and require "
|
|
"cross-tenant lifecycle review before alteration."
|
|
),
|
|
)
|
|
|
|
|
|
def _coalesce(*values: str | None) -> str | None | object:
|
|
normalized = {str(value).strip() for value in values if str(value or "").strip()}
|
|
if len(normalized) > 1:
|
|
return _CONFLICT
|
|
return next(iter(normalized), None)
|
|
|
|
|
|
def _optional_string(value: object) -> str | None:
|
|
return value if isinstance(value, str) and value else None
|
|
|
|
|
|
def _iso(value: datetime | None) -> str | None:
|
|
aware = _aware(value)
|
|
return aware.isoformat() if aware else None
|
|
|
|
|
|
def _aware(value: datetime | None) -> datetime | None:
|
|
if value is None or value.tzinfo is not None:
|
|
return value
|
|
return value.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not isinstance(value, Session):
|
|
raise TypeError("Identity DSAR requires a SQLAlchemy Session.")
|
|
return value
|
|
|
|
|
|
def _validate_record(record: DsarRecordRef) -> None:
|
|
if record.provider_id != "identity" or record.module_id != "identity":
|
|
raise ValueError("Identity DSAR cannot plan a foreign provider record.")
|
|
if record.resource_type != "canonical_identity" or not record.resource_id:
|
|
raise ValueError("Identity DSAR record identity is invalid.")
|
|
|
|
|
|
def _validate_action(action: DsarErasureActionRef) -> None:
|
|
if action.provider_id != "identity" or action.module_id != "identity":
|
|
raise ValueError("Identity DSAR cannot execute a foreign provider action.")
|
|
if not action.action_id.startswith("identity:manual_review:"):
|
|
raise ValueError("Identity DSAR action identity is invalid.")
|
|
|
|
|
|
__all__ = ["IDENTITY_DSAR_CAPABILITY", "IdentityDsarProvider"]
|