feat(identity-trust): add governed DSAR coverage
This commit is contained in:
@@ -0,0 +1,461 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.dsar import (
|
||||
DsarErasureActionRef,
|
||||
DsarExecutionResultRef,
|
||||
DsarRecordRef,
|
||||
DsarSubjectRef,
|
||||
dsar_capability_name,
|
||||
)
|
||||
from govoplan_identity_trust.backend.db.models import (
|
||||
AssuranceEvidence,
|
||||
DevicePublicKey,
|
||||
KeyAccessDecisionRecord,
|
||||
TrustKeyEpoch,
|
||||
)
|
||||
|
||||
|
||||
IDENTITY_TRUST_DSAR_CAPABILITY = dsar_capability_name("identity_trust")
|
||||
_MAX_RECORDS = 5_000
|
||||
_CONFLICT = object()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _SubjectSelectors:
|
||||
account_id: str | None
|
||||
identity_id: str | None
|
||||
device_id: str | None
|
||||
key_id: str | None
|
||||
|
||||
|
||||
class IdentityTrustDsarProvider:
|
||||
provider_id = "identity_trust"
|
||||
module_id = "identity_trust"
|
||||
|
||||
def search_subject(
|
||||
self,
|
||||
session: object,
|
||||
*,
|
||||
tenant_id: str,
|
||||
subject: DsarSubjectRef,
|
||||
) -> Sequence[DsarRecordRef]:
|
||||
db = _session(session)
|
||||
selectors = _subject_selectors(subject)
|
||||
if selectors is None:
|
||||
return ()
|
||||
|
||||
devices = db.query(DevicePublicKey).filter(
|
||||
DevicePublicKey.tenant_id == tenant_id
|
||||
)
|
||||
if selectors.account_id:
|
||||
devices = devices.filter(
|
||||
DevicePublicKey.account_id == selectors.account_id
|
||||
)
|
||||
if selectors.identity_id:
|
||||
devices = devices.filter(
|
||||
DevicePublicKey.identity_id == selectors.identity_id
|
||||
)
|
||||
if selectors.device_id:
|
||||
devices = devices.filter(DevicePublicKey.device_id == selectors.device_id)
|
||||
if selectors.key_id:
|
||||
devices = devices.filter(DevicePublicKey.key_id == selectors.key_id)
|
||||
records: list[DsarRecordRef] = [
|
||||
_device_record(row)
|
||||
for row in _limited(
|
||||
devices,
|
||||
DevicePublicKey.registered_at,
|
||||
DevicePublicKey.id,
|
||||
label="device key",
|
||||
)
|
||||
]
|
||||
|
||||
if selectors.account_id:
|
||||
assurances = db.query(AssuranceEvidence).filter(
|
||||
AssuranceEvidence.tenant_id == tenant_id,
|
||||
AssuranceEvidence.account_id == selectors.account_id,
|
||||
)
|
||||
if selectors.key_id:
|
||||
assurances = assurances.filter(
|
||||
AssuranceEvidence.device_key_id == selectors.key_id
|
||||
)
|
||||
records.extend(
|
||||
_assurance_record(row)
|
||||
for row in _limited(
|
||||
assurances,
|
||||
AssuranceEvidence.verified_at,
|
||||
AssuranceEvidence.id,
|
||||
label="assurance evidence",
|
||||
)
|
||||
)
|
||||
decisions = db.query(KeyAccessDecisionRecord).filter(
|
||||
KeyAccessDecisionRecord.tenant_id == tenant_id,
|
||||
KeyAccessDecisionRecord.account_id == selectors.account_id,
|
||||
)
|
||||
if selectors.key_id:
|
||||
decisions = decisions.filter(
|
||||
KeyAccessDecisionRecord.device_key_id == selectors.key_id
|
||||
)
|
||||
records.extend(
|
||||
_access_decision_record(row)
|
||||
for row in _limited(
|
||||
decisions,
|
||||
KeyAccessDecisionRecord.created_at,
|
||||
KeyAccessDecisionRecord.id,
|
||||
label="key-access decision",
|
||||
)
|
||||
)
|
||||
|
||||
subject_conditions = []
|
||||
if selectors.device_id:
|
||||
subject_conditions.append(
|
||||
(TrustKeyEpoch.subject_kind == "device")
|
||||
& (TrustKeyEpoch.subject_id == selectors.device_id)
|
||||
)
|
||||
elif not selectors.key_id:
|
||||
if selectors.account_id:
|
||||
subject_conditions.append(
|
||||
(TrustKeyEpoch.subject_kind == "account")
|
||||
& (TrustKeyEpoch.subject_id == selectors.account_id)
|
||||
)
|
||||
if selectors.identity_id:
|
||||
subject_conditions.append(
|
||||
(TrustKeyEpoch.subject_kind == "identity")
|
||||
& (TrustKeyEpoch.subject_id == selectors.identity_id)
|
||||
)
|
||||
if subject_conditions:
|
||||
epochs = db.query(TrustKeyEpoch).filter(
|
||||
TrustKeyEpoch.tenant_id == tenant_id,
|
||||
or_(*subject_conditions),
|
||||
)
|
||||
records.extend(
|
||||
_epoch_record(row)
|
||||
for row in _limited(
|
||||
epochs,
|
||||
TrustKeyEpoch.effective_at,
|
||||
TrustKeyEpoch.id,
|
||||
label="key epoch",
|
||||
)
|
||||
)
|
||||
|
||||
if len(records) > _MAX_RECORDS:
|
||||
raise ValueError(
|
||||
"Identity Trust DSAR result limit exceeded; narrow the selectors."
|
||||
)
|
||||
return tuple(
|
||||
sorted(records, key=lambda item: (item.resource_type, item.resource_id))
|
||||
)
|
||||
|
||||
def plan_erasure(
|
||||
self,
|
||||
session: object,
|
||||
*,
|
||||
tenant_id: str,
|
||||
subject: DsarSubjectRef,
|
||||
records: Sequence[DsarRecordRef],
|
||||
) -> Sequence[DsarErasureActionRef]:
|
||||
del tenant_id
|
||||
_session(session)
|
||||
if _subject_selectors(subject) is None:
|
||||
raise ValueError("Identity Trust DSAR subject selectors conflict.")
|
||||
actions: list[DsarErasureActionRef] = []
|
||||
for record in records:
|
||||
_validate_record(record)
|
||||
device = record.resource_type == "device_public_key"
|
||||
actions.append(
|
||||
DsarErasureActionRef(
|
||||
action_id=(
|
||||
f"identity_trust:{'manual_review' if device else 'retain'}:"
|
||||
f"{record.resource_type}:{record.resource_id}"
|
||||
),
|
||||
provider_id=self.provider_id,
|
||||
module_id=self.module_id,
|
||||
kind="manual_review" if device else "retain",
|
||||
resource_type=record.resource_type,
|
||||
resource_id=record.resource_id,
|
||||
title=("Review " if device else "Retain ") + record.title,
|
||||
rationale=(
|
||||
"Revocation or removal must account for encrypted-resource "
|
||||
"recovery, active sessions, and the current key epoch."
|
||||
if device
|
||||
else record.retention_reason
|
||||
or "Trust and access evidence remains immutable."
|
||||
),
|
||||
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 Trust DSAR subject selectors conflict.")
|
||||
results: list[DsarExecutionResultRef] = []
|
||||
for action in actions:
|
||||
_validate_action(action)
|
||||
if action.executable or action.kind not in {"manual_review", "retain"}:
|
||||
raise ValueError(
|
||||
"Identity Trust DSAR publishes non-executable actions only."
|
||||
)
|
||||
results.append(
|
||||
DsarExecutionResultRef(
|
||||
action_id=action.action_id,
|
||||
status="blocked",
|
||||
summary=(
|
||||
"The device key remains unchanged pending cryptographic "
|
||||
"recovery and revocation review."
|
||||
if action.kind == "manual_review"
|
||||
else "Trust and key-access evidence remains immutable."
|
||||
),
|
||||
evidence={"request_id": request_id},
|
||||
)
|
||||
)
|
||||
return tuple(results)
|
||||
|
||||
|
||||
def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None:
|
||||
references = subject.external_references
|
||||
values = {
|
||||
"account_id": _coalesce(
|
||||
subject.account_id,
|
||||
references.get("identity_trust.account"),
|
||||
references.get("access.account"),
|
||||
),
|
||||
"identity_id": _coalesce(
|
||||
subject.identity_id,
|
||||
references.get("identity_trust.identity"),
|
||||
references.get("identity.id"),
|
||||
),
|
||||
"device_id": _coalesce(
|
||||
references.get("identity_trust.device"),
|
||||
references.get("identity_trust.device_id"),
|
||||
),
|
||||
"key_id": _coalesce(
|
||||
references.get("identity_trust.key"),
|
||||
references.get("identity_trust.key_id"),
|
||||
),
|
||||
}
|
||||
if any(value is _CONFLICT for value in values.values()):
|
||||
return None
|
||||
account_id = _optional_string(values["account_id"])
|
||||
identity_id = _optional_string(values["identity_id"])
|
||||
if not (account_id or identity_id):
|
||||
return None
|
||||
return _SubjectSelectors(
|
||||
account_id=account_id,
|
||||
identity_id=identity_id,
|
||||
device_id=_optional_string(values["device_id"]),
|
||||
key_id=_optional_string(values["key_id"]),
|
||||
)
|
||||
|
||||
|
||||
def _limited(query, first, second, *, label: str):
|
||||
rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all()
|
||||
if len(rows) > _MAX_RECORDS:
|
||||
raise ValueError(
|
||||
f"Identity Trust DSAR {label} limit exceeded; narrow the selectors."
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def _device_record(row: DevicePublicKey) -> DsarRecordRef:
|
||||
return DsarRecordRef(
|
||||
provider_id="identity_trust",
|
||||
module_id="identity_trust",
|
||||
resource_type="device_public_key",
|
||||
resource_id=row.id,
|
||||
category="personal_device_trust",
|
||||
title=f"Device public key {row.key_id[:255]}",
|
||||
data={
|
||||
"identity_id": row.identity_id,
|
||||
"account_id": row.account_id,
|
||||
"device_id": row.device_id,
|
||||
"key_id": row.key_id,
|
||||
"algorithm": row.algorithm,
|
||||
"public_jwk": _public_jwk(row.public_jwk),
|
||||
"purpose": row.purpose,
|
||||
"assurance_level": row.assurance_level,
|
||||
"attestation_ref": (row.attestation_ref or "")[:1_000] or None,
|
||||
"status": row.status,
|
||||
"epoch": row.epoch,
|
||||
"registered_at": _iso(row.registered_at),
|
||||
"expires_at": _iso(row.expires_at),
|
||||
"revoked_at": _iso(row.revoked_at),
|
||||
"revocation_reason": (row.revocation_reason or "")[:4_000] or None,
|
||||
},
|
||||
observed_at=_aware(row.updated_at),
|
||||
retention_reason=(
|
||||
"Device-key changes require recovery and active-resource review."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _assurance_record(row: AssuranceEvidence) -> DsarRecordRef:
|
||||
return DsarRecordRef(
|
||||
provider_id="identity_trust",
|
||||
module_id="identity_trust",
|
||||
resource_type="assurance_evidence",
|
||||
resource_id=row.id,
|
||||
category="identity_assurance_evidence",
|
||||
title="Identity assurance evidence",
|
||||
data={
|
||||
"account_id": row.account_id,
|
||||
"device_key_id": row.device_key_id,
|
||||
"evidence_ref": row.evidence_ref[:1_000],
|
||||
"assurance_level": row.assurance_level,
|
||||
"provider_id": row.provider_id,
|
||||
"verified_at": _iso(row.verified_at),
|
||||
"expires_at": _iso(row.expires_at),
|
||||
},
|
||||
observed_at=_aware(row.created_at),
|
||||
immutable_evidence=True,
|
||||
retention_reason=(
|
||||
"Assurance verification is immutable security and accountability evidence."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _epoch_record(row: TrustKeyEpoch) -> DsarRecordRef:
|
||||
return DsarRecordRef(
|
||||
provider_id="identity_trust",
|
||||
module_id="identity_trust",
|
||||
resource_type="key_epoch",
|
||||
resource_id=row.id,
|
||||
category="cryptographic_access_history",
|
||||
title="Subject key epoch",
|
||||
data={
|
||||
"subject_kind": row.subject_kind,
|
||||
"subject_id": row.subject_id,
|
||||
"epoch": row.epoch,
|
||||
"previous_epoch": row.previous_epoch,
|
||||
"state": row.state,
|
||||
"history_policy": row.history_policy,
|
||||
"reason": row.reason[:4_000],
|
||||
"access_decision_ref": row.access_decision_ref[:1_000],
|
||||
"effective_at": _iso(row.effective_at),
|
||||
},
|
||||
observed_at=_aware(row.effective_at),
|
||||
immutable_evidence=True,
|
||||
retention_reason="Key-epoch history is immutable cryptographic-access evidence.",
|
||||
)
|
||||
|
||||
|
||||
def _access_decision_record(row: KeyAccessDecisionRecord) -> DsarRecordRef:
|
||||
return DsarRecordRef(
|
||||
provider_id="identity_trust",
|
||||
module_id="identity_trust",
|
||||
resource_type="key_access_decision",
|
||||
resource_id=row.id,
|
||||
category="cryptographic_access_decision",
|
||||
title="Key-access decision",
|
||||
data={
|
||||
"decision_ref": row.decision_ref,
|
||||
"account_id": row.account_id,
|
||||
"device_key_id": row.device_key_id,
|
||||
"subject_kind": row.subject_kind,
|
||||
"subject_id": row.subject_id,
|
||||
"key_epoch": row.key_epoch,
|
||||
"access_decision_ref": row.access_decision_ref[:1_000],
|
||||
"purpose": row.purpose[:255],
|
||||
"allowed": row.allowed,
|
||||
"reason": row.reason[:4_000],
|
||||
"resource_ref": (row.resource_ref or "")[:1_000] or None,
|
||||
"function_assignment_id": row.function_assignment_id,
|
||||
"delegation_id": row.delegation_id,
|
||||
"recorded_at": _iso(row.created_at),
|
||||
},
|
||||
observed_at=_aware(row.created_at),
|
||||
immutable_evidence=True,
|
||||
retention_reason=(
|
||||
"Key-access decisions are immutable authorization and security evidence."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _public_jwk(value: object) -> dict[str, object]:
|
||||
if not isinstance(value, Mapping):
|
||||
raise ValueError("Identity Trust public JWK is invalid.")
|
||||
projection: dict[str, object] = {}
|
||||
for key in ("kty", "crv", "x", "y", "use", "alg"):
|
||||
if key in value:
|
||||
projection[key] = str(value[key])[:2_000]
|
||||
key_ops = value.get("key_ops")
|
||||
if key_ops is not None:
|
||||
if not isinstance(key_ops, list) or len(key_ops) > 20:
|
||||
raise ValueError("Identity Trust public JWK key operations are invalid.")
|
||||
projection["key_ops"] = [str(item)[:120] for item in key_ops]
|
||||
return projection
|
||||
|
||||
|
||||
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 Trust DSAR requires a SQLAlchemy Session.")
|
||||
return value
|
||||
|
||||
|
||||
_RESOURCE_TYPES = {
|
||||
"device_public_key",
|
||||
"assurance_evidence",
|
||||
"key_epoch",
|
||||
"key_access_decision",
|
||||
}
|
||||
|
||||
|
||||
def _validate_record(record: DsarRecordRef) -> None:
|
||||
if (
|
||||
record.provider_id != "identity_trust"
|
||||
or record.module_id != "identity_trust"
|
||||
):
|
||||
raise ValueError("Identity Trust DSAR cannot plan a foreign provider record.")
|
||||
if record.resource_type not in _RESOURCE_TYPES or not record.resource_id:
|
||||
raise ValueError("Identity Trust DSAR record identity is invalid.")
|
||||
|
||||
|
||||
def _validate_action(action: DsarErasureActionRef) -> None:
|
||||
if (
|
||||
action.provider_id != "identity_trust"
|
||||
or action.module_id != "identity_trust"
|
||||
):
|
||||
raise ValueError("Identity Trust DSAR cannot execute a foreign action.")
|
||||
if not action.action_id.startswith("identity_trust:"):
|
||||
raise ValueError("Identity Trust DSAR action identity is invalid.")
|
||||
|
||||
|
||||
__all__ = ["IDENTITY_TRUST_DSAR_CAPABILITY", "IdentityTrustDsarProvider"]
|
||||
@@ -26,6 +26,10 @@ from govoplan_core.core.modules import (
|
||||
from govoplan_core.core.provider_governance import declared_module_architecture
|
||||
from govoplan_core.db.base import Base
|
||||
from govoplan_identity_trust.backend.db import models
|
||||
from govoplan_identity_trust.backend.dsar_provider import (
|
||||
IDENTITY_TRUST_DSAR_CAPABILITY,
|
||||
IdentityTrustDsarProvider,
|
||||
)
|
||||
from govoplan_identity_trust.backend.service import SqlIdentityTrustService
|
||||
|
||||
|
||||
@@ -64,6 +68,10 @@ def _service(_context: ModuleContext) -> SqlIdentityTrustService:
|
||||
return SqlIdentityTrustService()
|
||||
|
||||
|
||||
def _dsar_provider(_context: ModuleContext) -> IdentityTrustDsarProvider:
|
||||
return IdentityTrustDsarProvider()
|
||||
|
||||
|
||||
manifest = ModuleManifest(
|
||||
id=MODULE_ID,
|
||||
name=MODULE_NAME,
|
||||
@@ -72,6 +80,10 @@ manifest = ModuleManifest(
|
||||
provides_interfaces=(
|
||||
ModuleInterfaceProvider(name="identity_trust.directory", version="1.0.0"),
|
||||
ModuleInterfaceProvider(name="identity_trust.assurance", version="1.0.0"),
|
||||
ModuleInterfaceProvider(
|
||||
name=IDENTITY_TRUST_DSAR_CAPABILITY,
|
||||
version="0.1.0",
|
||||
),
|
||||
),
|
||||
permissions=(
|
||||
_permission(
|
||||
@@ -165,6 +177,7 @@ manifest = ModuleManifest(
|
||||
capability_factories={
|
||||
CAPABILITY_IDENTITY_TRUST_DIRECTORY: _service,
|
||||
CAPABILITY_IDENTITY_TRUST_ASSURANCE: _service,
|
||||
IDENTITY_TRUST_DSAR_CAPABILITY: _dsar_provider,
|
||||
},
|
||||
capability_documentation={
|
||||
CAPABILITY_IDENTITY_TRUST_DIRECTORY: CapabilityDocumentation(
|
||||
@@ -177,6 +190,14 @@ manifest = ModuleManifest(
|
||||
summary="Verifies bounded, recent assurance evidence for high-risk cryptographic operations.",
|
||||
contract_version="1.0.0",
|
||||
),
|
||||
IDENTITY_TRUST_DSAR_CAPABILITY: CapabilityDocumentation(
|
||||
label="Identity Trust data-subject request provider",
|
||||
summary=(
|
||||
"Exports bounded device-key and assurance records while keeping "
|
||||
"cryptographic history under governed retention and review."
|
||||
),
|
||||
contract_version="0.1.0",
|
||||
),
|
||||
},
|
||||
migration_spec=MigrationSpec(
|
||||
module_id=MODULE_ID,
|
||||
@@ -202,6 +223,47 @@ manifest = ModuleManifest(
|
||||
),
|
||||
),
|
||||
documentation=(
|
||||
DocumentationTopic(
|
||||
id="identity-trust.data-subject-requests",
|
||||
title="Identity Trust data-subject requests",
|
||||
summary=(
|
||||
"Export tenant-scoped device trust, assurance, epoch, and key-access "
|
||||
"evidence without private or operational key material."
|
||||
),
|
||||
body=(
|
||||
"Identity Trust correlates exact account and identity identifiers in "
|
||||
"the active tenant and can narrow results to an exact device or key. "
|
||||
"The access package includes bounded public-key registration fields, "
|
||||
"assurance state, matching key epochs, and key-access decisions. It "
|
||||
"never exports private JWK parameters, request digests, idempotency "
|
||||
"keys, or arbitrary provenance payloads. Device-key revocation requires "
|
||||
"manual review of recovery, active encrypted resources, and the current "
|
||||
"epoch. Assurance, epoch, and access-decision records remain immutable "
|
||||
"security evidence."
|
||||
),
|
||||
layer="configured",
|
||||
documentation_types=("admin", "user"),
|
||||
audience=("user", "administrator", "security_officer", "auditor"),
|
||||
related_modules=("core", "access", "encryption", "postbox"),
|
||||
metadata={
|
||||
"help_contexts": [
|
||||
"identity_trust.settings.devices",
|
||||
"identity_trust.admin.trust",
|
||||
"privacy.data-subject-requests",
|
||||
],
|
||||
"consequence_classes": {
|
||||
"export_trust_state": (
|
||||
"Returns bounded public trust and assurance evidence only."
|
||||
),
|
||||
"review_device_revocation": (
|
||||
"Requires recovery and encrypted-resource impact review."
|
||||
),
|
||||
"retain_security_evidence": (
|
||||
"Preserves epoch, assurance, and access-decision history."
|
||||
),
|
||||
},
|
||||
},
|
||||
),
|
||||
DocumentationTopic(
|
||||
id="identity-trust.device-keys",
|
||||
title="Device keys and key epochs",
|
||||
|
||||
@@ -0,0 +1,297 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
|
||||
from govoplan_core.db.base import Base
|
||||
from govoplan_core.privacy.dsar_workflow import (
|
||||
create_data_subject_request,
|
||||
search_data_subject_request,
|
||||
)
|
||||
from govoplan_identity_trust.backend.db.models import (
|
||||
AssuranceEvidence,
|
||||
DevicePublicKey,
|
||||
KeyAccessDecisionRecord,
|
||||
TrustKeyEpoch,
|
||||
)
|
||||
from govoplan_identity_trust.backend.dsar_provider import (
|
||||
IDENTITY_TRUST_DSAR_CAPABILITY,
|
||||
IdentityTrustDsarProvider,
|
||||
)
|
||||
from govoplan_identity_trust.backend.manifest import manifest
|
||||
|
||||
|
||||
NOW = datetime(2026, 8, 21, 12, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
class _Registry:
|
||||
def __init__(self, provider: IdentityTrustDsarProvider) -> None:
|
||||
self.provider = provider
|
||||
|
||||
def capability_names(self):
|
||||
return (IDENTITY_TRUST_DSAR_CAPABILITY,)
|
||||
|
||||
def capability_owner(self, name):
|
||||
if name != IDENTITY_TRUST_DSAR_CAPABILITY:
|
||||
raise KeyError(name)
|
||||
return "identity_trust"
|
||||
|
||||
def tenant_entitlement_resolver(self):
|
||||
class _Resolver:
|
||||
@staticmethod
|
||||
def resolve(session, tenant_id):
|
||||
del session, tenant_id
|
||||
return type(
|
||||
"State", (), {"effective_modules": ("identity_trust",)}
|
||||
)()
|
||||
|
||||
return _Resolver()
|
||||
|
||||
def require_tenant_capability(self, name, session, **kwargs):
|
||||
del session, kwargs
|
||||
if name != IDENTITY_TRUST_DSAR_CAPABILITY:
|
||||
raise KeyError(name)
|
||||
return self.provider
|
||||
|
||||
def manifests(self):
|
||||
return (type("Manifest", (), {"id": "identity_trust"})(),)
|
||||
|
||||
|
||||
class IdentityTrustDsarProviderTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
||||
Base.metadata.create_all(self.engine)
|
||||
self.session = Session(self.engine)
|
||||
self.provider = IdentityTrustDsarProvider()
|
||||
self.assertIsInstance(self.provider, DsarProvider)
|
||||
self._seed()
|
||||
self.session.commit()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.session.close()
|
||||
self.engine.dispose()
|
||||
|
||||
def _seed(self) -> None:
|
||||
self.session.add_all(
|
||||
(
|
||||
DevicePublicKey(
|
||||
id="device-key-row-1",
|
||||
tenant_id="tenant-1",
|
||||
identity_id="identity-1",
|
||||
account_id="account-1",
|
||||
device_id="device-1",
|
||||
key_id="key-1",
|
||||
algorithm="EdDSA",
|
||||
public_jwk={
|
||||
"kty": "OKP",
|
||||
"crv": "Ed25519",
|
||||
"x": "public-coordinate",
|
||||
"d": "private-jwk-do-not-export",
|
||||
"unknown": "arbitrary-jwk-do-not-export",
|
||||
},
|
||||
purpose="postbox",
|
||||
assurance_level="substantial",
|
||||
attestation_ref="evidence:attestation-1",
|
||||
status="active",
|
||||
epoch=1,
|
||||
registration_digest="registration-digest-do-not-export",
|
||||
idempotency_key="device-idempotency-do-not-export",
|
||||
registered_at=NOW,
|
||||
created_by="account-1",
|
||||
updated_by="account-1",
|
||||
),
|
||||
DevicePublicKey(
|
||||
id="device-key-other-tenant",
|
||||
tenant_id="tenant-2",
|
||||
identity_id="identity-1",
|
||||
account_id="account-1",
|
||||
device_id="device-other-tenant",
|
||||
key_id="key-other-tenant",
|
||||
algorithm="EdDSA",
|
||||
public_jwk={"kty": "OKP", "x": "other-tenant-public"},
|
||||
purpose="postbox",
|
||||
assurance_level="substantial",
|
||||
status="active",
|
||||
epoch=1,
|
||||
registration_digest="other-digest",
|
||||
idempotency_key="other-key",
|
||||
registered_at=NOW,
|
||||
),
|
||||
)
|
||||
)
|
||||
self.session.add(
|
||||
AssuranceEvidence(
|
||||
id="assurance-1",
|
||||
tenant_id="tenant-1",
|
||||
account_id="account-1",
|
||||
device_key_id="key-1",
|
||||
evidence_ref="assurance:evidence-1",
|
||||
assurance_level="substantial",
|
||||
provider_id="bund-id",
|
||||
verified_at=NOW,
|
||||
expires_at=NOW,
|
||||
provenance={"secret": "assurance-provenance-do-not-export"},
|
||||
recorded_by="security-officer",
|
||||
)
|
||||
)
|
||||
self.session.add(
|
||||
TrustKeyEpoch(
|
||||
id="epoch-1",
|
||||
tenant_id="tenant-1",
|
||||
subject_kind="account",
|
||||
subject_id="account-1",
|
||||
epoch=2,
|
||||
previous_epoch=1,
|
||||
state="active",
|
||||
history_policy="forward_only",
|
||||
reason="Device rotation",
|
||||
access_decision_ref="access:decision-1",
|
||||
idempotency_key="epoch-idempotency-do-not-export",
|
||||
request_digest="epoch-request-digest-do-not-export",
|
||||
effective_at=NOW,
|
||||
created_by="security-officer",
|
||||
)
|
||||
)
|
||||
self.session.add(
|
||||
KeyAccessDecisionRecord(
|
||||
id="access-decision-1",
|
||||
tenant_id="tenant-1",
|
||||
decision_ref="trust-decision-1",
|
||||
request_digest="access-request-digest-do-not-export",
|
||||
account_id="account-1",
|
||||
device_key_id="key-1",
|
||||
subject_kind="postbox",
|
||||
subject_id="postbox-1",
|
||||
key_epoch=2,
|
||||
access_decision_ref="access:authorization-1",
|
||||
purpose="read-message",
|
||||
allowed=True,
|
||||
reason="Current device and epoch",
|
||||
resource_ref="postbox:message-1",
|
||||
provenance={"secret": "access-provenance-do-not-export"},
|
||||
)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _subject() -> DsarSubjectRef:
|
||||
return DsarSubjectRef(account_id="account-1", identity_id="identity-1")
|
||||
|
||||
def test_search_exports_bounded_trust_records(self) -> None:
|
||||
records = self.provider.search_subject(
|
||||
self.session, tenant_id="tenant-1", subject=self._subject()
|
||||
)
|
||||
self.assertEqual(
|
||||
{
|
||||
"device_public_key",
|
||||
"assurance_evidence",
|
||||
"key_epoch",
|
||||
"key_access_decision",
|
||||
},
|
||||
{record.resource_type for record in records},
|
||||
)
|
||||
exported = json.dumps([record.to_dict() for record in records])
|
||||
self.assertIn("public-coordinate", exported)
|
||||
self.assertIn("assurance:evidence-1", exported)
|
||||
self.assertIn("postbox:message-1", exported)
|
||||
for excluded in (
|
||||
"private-jwk-do-not-export",
|
||||
"arbitrary-jwk-do-not-export",
|
||||
"registration-digest-do-not-export",
|
||||
"device-idempotency-do-not-export",
|
||||
"assurance-provenance-do-not-export",
|
||||
"epoch-request-digest-do-not-export",
|
||||
"access-request-digest-do-not-export",
|
||||
"access-provenance-do-not-export",
|
||||
"other-tenant-public",
|
||||
):
|
||||
self.assertNotIn(excluded, exported)
|
||||
|
||||
def test_key_narrowing_and_conflicts_fail_closed(self) -> None:
|
||||
narrowed = self.provider.search_subject(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
subject=DsarSubjectRef(
|
||||
account_id="account-1",
|
||||
external_references={"identity_trust.key": "key-1"},
|
||||
),
|
||||
)
|
||||
conflict = self.provider.search_subject(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
subject=DsarSubjectRef(
|
||||
account_id="account-1",
|
||||
external_references={"identity_trust.account": "account-other"},
|
||||
),
|
||||
)
|
||||
reference_only = self.provider.search_subject(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
subject=DsarSubjectRef(
|
||||
external_references={"identity_trust.key": "key-1"}
|
||||
),
|
||||
)
|
||||
self.assertEqual(
|
||||
{"device_public_key", "assurance_evidence", "key_access_decision"},
|
||||
{record.resource_type for record in narrowed},
|
||||
)
|
||||
self.assertEqual((), conflict)
|
||||
self.assertEqual((), reference_only)
|
||||
|
||||
def test_erasure_is_review_or_retain_only(self) -> None:
|
||||
records = self.provider.search_subject(
|
||||
self.session, tenant_id="tenant-1", subject=self._subject()
|
||||
)
|
||||
actions = self.provider.plan_erasure(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
subject=self._subject(),
|
||||
records=records,
|
||||
)
|
||||
self.assertEqual(
|
||||
{"manual_review", "retain"}, {action.kind for action in actions}
|
||||
)
|
||||
results = self.provider.execute_erasure(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
subject=self._subject(),
|
||||
actions=actions,
|
||||
request_id="dsar-trust-1",
|
||||
)
|
||||
self.assertTrue(all(result.status == "blocked" for result in results))
|
||||
self.assertEqual(
|
||||
"active",
|
||||
self.session.get(DevicePublicKey, "device-key-row-1").status,
|
||||
)
|
||||
|
||||
def test_manifest_and_core_workflow_discover_provider(self) -> None:
|
||||
self.assertIn(
|
||||
IDENTITY_TRUST_DSAR_CAPABILITY, manifest.capability_factories
|
||||
)
|
||||
row = create_data_subject_request(
|
||||
self.session,
|
||||
tenant_id="tenant-1",
|
||||
reference="DSAR-TRUST-1",
|
||||
request_kind="access",
|
||||
subject=self._subject(),
|
||||
purpose="Trust access request",
|
||||
legal_basis=None,
|
||||
due_at=None,
|
||||
requested_by_account_id="operator-1",
|
||||
)
|
||||
search_data_subject_request(
|
||||
self.session,
|
||||
registry=_Registry(self.provider),
|
||||
row=row,
|
||||
expected_revision=row.resource_revision,
|
||||
)
|
||||
self.assertEqual("searched", row.status)
|
||||
self.assertEqual(4, row.search_result["record_count"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user