Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
80dabde81d | ||
|
|
8d1ca493bb | ||
|
|
b9a158c2cb |
+2
-2
@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "govoplan-identity-trust"
|
||||
version = "0.1.18"
|
||||
version = "0.1.19"
|
||||
description = "Public device-key, assurance, and key-epoch trust services for GovOPlaN."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.12"
|
||||
license = "AGPL-3.0-or-later"
|
||||
authors = [{ name = "GovOPlaN" }]
|
||||
dependencies = ["govoplan-core>=0.1.18"]
|
||||
dependencies = ["govoplan-core>=0.1.37"]
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
|
||||
@@ -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"]
|
||||
@@ -12,6 +12,7 @@ from govoplan_core.core.module_guards import (
|
||||
)
|
||||
from govoplan_core.core.modules import (
|
||||
CapabilityDocumentation,
|
||||
DocumentationCondition,
|
||||
DocumentationLink,
|
||||
DocumentationTopic,
|
||||
FrontendModule,
|
||||
@@ -26,12 +27,16 @@ 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
|
||||
|
||||
|
||||
MODULE_ID = "identity_trust"
|
||||
MODULE_NAME = "Identity Trust"
|
||||
MODULE_VERSION = "0.1.18"
|
||||
MODULE_VERSION = "0.1.19"
|
||||
DEVICE_READ_SCOPE = "identity_trust:device:read"
|
||||
DEVICE_WRITE_SCOPE = "identity_trust:device:write"
|
||||
KEY_ACCESS_SCOPE = "identity_trust:key_access:approve"
|
||||
@@ -64,6 +69,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 +81,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 +178,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 +191,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 +224,84 @@ 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={
|
||||
"kind": "reference",
|
||||
"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."
|
||||
),
|
||||
},
|
||||
},
|
||||
translations={
|
||||
"de": {
|
||||
"title": "Datenschutzanfragen zu Identity Trust",
|
||||
"summary": (
|
||||
"Mandantenbezogene Gerätevertrauens-, Assurance-, Epochen- und "
|
||||
"Schlüsselzugriffsnachweise ohne private oder operative Schlüsselmaterialien exportieren."
|
||||
),
|
||||
"body": (
|
||||
"Identity Trust gleicht im aktiven Mandanten exakte Konto- und Identitätskennungen ab "
|
||||
"und kann Ergebnisse auf ein bestimmtes Gerät oder einen bestimmten Schlüssel eingrenzen. "
|
||||
"Das Auskunftspaket enthält begrenzte Registrierungsfelder öffentlicher Schlüssel, den "
|
||||
"Assurance-Status, passende Schlüsselepochen und Schlüsselzugriffsentscheidungen. Private "
|
||||
"JWK-Parameter, Anforderungsprüfsummen, Idempotenzschlüssel und beliebige "
|
||||
"Provenienzinhalte werden niemals exportiert. Der Widerruf eines Geräteschlüssels erfordert "
|
||||
"eine manuelle Prüfung der Wiederherstellung, aktiver verschlüsselter Ressourcen und der "
|
||||
"aktuellen Epoche. Assurance-, Epochen- und Zugriffsentscheidungsdatensätze bleiben "
|
||||
"unveränderliche Sicherheitsnachweise."
|
||||
),
|
||||
}
|
||||
},
|
||||
structured_translation_version="1",
|
||||
structured_translations={
|
||||
"de": {
|
||||
"consequence_classes": {
|
||||
"export_trust_state": (
|
||||
"Gibt ausschließlich begrenzte öffentliche Vertrauens- und Assurance-Nachweise zurück."
|
||||
),
|
||||
"review_device_revocation": (
|
||||
"Erfordert eine Prüfung der Wiederherstellung und der Auswirkungen auf verschlüsselte Ressourcen."
|
||||
),
|
||||
"retain_security_evidence": (
|
||||
"Bewahrt die Historie von Epochen, Assurance und Zugriffsentscheidungen auf."
|
||||
),
|
||||
}
|
||||
}
|
||||
},
|
||||
),
|
||||
DocumentationTopic(
|
||||
id="identity-trust.device-keys",
|
||||
title="Device keys and key epochs",
|
||||
@@ -212,6 +312,102 @@ manifest = ModuleManifest(
|
||||
layer="available",
|
||||
documentation_types=("admin", "user"),
|
||||
audience=("user", "administrator", "security_officer", "auditor"),
|
||||
conditions=(
|
||||
DocumentationCondition(
|
||||
any_scopes=(DEVICE_READ_SCOPE, ASSURANCE_READ_SCOPE)
|
||||
),
|
||||
),
|
||||
related_modules=("access", "audit", "policy", "encryption", "postbox"),
|
||||
metadata={
|
||||
"kind": "workflow",
|
||||
"help_contexts": [
|
||||
"identity_trust.settings.devices",
|
||||
"identity_trust.admin.trust",
|
||||
],
|
||||
"purpose": (
|
||||
"Review public device-key trust and rotate or revoke trust evidence without handling private keys."
|
||||
),
|
||||
"prerequisites": [
|
||||
"The actor has device or assurance read access; consequential actions require their dedicated scopes.",
|
||||
"Access has already authorized the account and protected resource independently.",
|
||||
],
|
||||
"steps": [
|
||||
"Review registered public device keys and their assurance provenance in Settings.",
|
||||
"Inspect the current key epoch and any recorded key-access decisions before changing trust state.",
|
||||
"Assess recovery and encrypted-resource impact before revoking a device key.",
|
||||
"Security officers may rotate a subject epoch or record assurance only with the corresponding authority.",
|
||||
"Reload stale evidence before retrying a revision-bound revoke or rotation action.",
|
||||
],
|
||||
"limitations": [
|
||||
"Identity Trust stores public trust metadata only and provides neither private-key custody nor content encryption.",
|
||||
"Trust metadata is not device certification or proof that plaintext was never obtained.",
|
||||
],
|
||||
"operational_consequences": {
|
||||
"revoke_device": "Blocks future trust decisions for the device but cannot erase plaintext already obtained.",
|
||||
"rotate_epoch": "Changes the epoch accepted by future key-access decisions and requires impact review.",
|
||||
"record_assurance": "Appends immutable provenance evidence; it does not replace Access authorization.",
|
||||
},
|
||||
"verification": [
|
||||
"The displayed key contains public parameters only and names its current status and revision.",
|
||||
"Every consequential action records the actor, revision, reason, and resulting trust state.",
|
||||
"Access authorization and Identity Trust decisions remain separately auditable.",
|
||||
],
|
||||
},
|
||||
translations={
|
||||
"de": {
|
||||
"title": "Geräteschlüssel und Schlüsselepochen",
|
||||
"summary": (
|
||||
"Anmeldeberechtigung von öffentlichen Geräteschlüsseln und dem Vertrauen für "
|
||||
"kryptografische Zugriffe trennen."
|
||||
),
|
||||
"body": (
|
||||
"Identity Trust speichert ausschließlich öffentliche Schlüssel. Benutzer können ihre "
|
||||
"Geräteschlüssel in den Einstellungen prüfen und widerrufen sowie die Herkunft von "
|
||||
"Assurance-Nachweisen einsehen. Sicherheitsverantwortliche können ein berechtigtes Konto "
|
||||
"auswählen, Nachweise zu widerrufenen oder kompromittierten Geräten prüfen, "
|
||||
"Schlüsselepochen einer betroffenen Person rotieren und Schlüsselzugriffsentscheidungen "
|
||||
"in der Administration nachvollziehen. Access entscheidet zuerst, ob ein Konto eine "
|
||||
"geschützte Ressource erreichen darf; Identity Trust prüft anschließend das aktuelle Gerät "
|
||||
"und die Schlüsselepoche und zeichnet eine nachvollziehbare Entscheidung auf. Historische "
|
||||
"Freigaben für Funktionen und Postbox sind ausdrückliche Epochenrichtlinien. Ein Widerruf "
|
||||
"kann bereits erhaltenen Klartext nicht löschen. Jeder Widerruf und jede Rotation ist an "
|
||||
"eine Revision gebunden; veraltete Aktionen müssen neu geladen werden."
|
||||
),
|
||||
}
|
||||
},
|
||||
structured_translation_version="1",
|
||||
structured_translations={
|
||||
"de": {
|
||||
"purpose": (
|
||||
"Das Vertrauen in öffentliche Geräteschlüssel prüfen und Vertrauensnachweise rotieren oder widerrufen, ohne private Schlüssel zu verarbeiten."
|
||||
),
|
||||
"prerequisites": [
|
||||
"Die handelnde Person darf Geräte oder Assurance lesen; folgenreiche Aktionen erfordern ihre jeweils eigenen Berechtigungen.",
|
||||
"Access hat das Konto und die geschützte Ressource bereits unabhängig autorisiert.",
|
||||
],
|
||||
"steps": [
|
||||
"Registrierte öffentliche Geräteschlüssel und die Herkunft ihrer Assurance-Nachweise in den Einstellungen prüfen.",
|
||||
"Vor einer Änderung des Vertrauensstatus die aktuelle Schlüsselepoche und aufgezeichnete Schlüsselzugriffsentscheidungen prüfen.",
|
||||
"Vor dem Widerruf eines Geräteschlüssels Wiederherstellung und Auswirkungen auf verschlüsselte Ressourcen bewerten.",
|
||||
"Sicherheitsverantwortliche dürfen eine Epoche rotieren oder Assurance nur mit der jeweiligen Berechtigung aufzeichnen.",
|
||||
"Veraltete Nachweise neu laden, bevor eine revisionsgebundene Widerrufs- oder Rotationsaktion wiederholt wird.",
|
||||
],
|
||||
"limitations": [
|
||||
"Identity Trust speichert nur öffentliche Vertrauensmetadaten und bietet weder private Schlüsselverwahrung noch Inhaltsverschlüsselung.",
|
||||
"Vertrauensmetadaten sind keine Gerätezertifizierung und kein Nachweis dafür, dass niemals Klartext erhalten wurde.",
|
||||
],
|
||||
"operational_consequences": {
|
||||
"revoke_device": "Blockiert künftige Vertrauensentscheidungen für das Gerät, kann aber bereits erhaltenen Klartext nicht löschen.",
|
||||
"rotate_epoch": "Ändert die für künftige Schlüsselzugriffsentscheidungen akzeptierte Epoche und erfordert eine Folgenprüfung.",
|
||||
"record_assurance": "Fügt unveränderliche Herkunftsnachweise an und ersetzt nicht die Autorisierung durch Access.",
|
||||
},
|
||||
"verification": [
|
||||
"Der angezeigte Schlüssel enthält nur öffentliche Parameter und nennt aktuellen Status und Revision.",
|
||||
"Jede folgenreiche Aktion zeichnet Akteur, Revision, Begründung und resultierenden Vertrauensstatus auf.",
|
||||
"Access-Autorisierung und Identity-Trust-Entscheidung bleiben getrennt nachvollziehbar.",
|
||||
],
|
||||
}
|
||||
},
|
||||
links=(
|
||||
DocumentationLink(
|
||||
label="Device-key trust and recovery boundary",
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
from govoplan_core.core.modules import (
|
||||
documentation_structured_translation_issues,
|
||||
user_workflow_scope_condition_issues,
|
||||
)
|
||||
from govoplan_identity_trust.backend.manifest import manifest
|
||||
|
||||
|
||||
class IdentityTrustDocumentationTests(unittest.TestCase):
|
||||
def test_public_topics_have_complete_german_reference_content(self) -> None:
|
||||
self.assertEqual(2, len(manifest.documentation))
|
||||
for topic in manifest.documentation:
|
||||
translation = topic.translations.get("de", {})
|
||||
self.assertTrue(
|
||||
all(translation.get(key) for key in ("title", "summary", "body"))
|
||||
)
|
||||
self.assertEqual((), documentation_structured_translation_issues(topic))
|
||||
|
||||
def test_documentation_has_scope_conditioned_workflow_and_reference(self) -> None:
|
||||
kinds = {topic.metadata.get("kind") for topic in manifest.documentation}
|
||||
self.assertIn("workflow", kinds)
|
||||
self.assertIn("reference", kinds)
|
||||
for topic in manifest.documentation:
|
||||
self.assertEqual((), user_workflow_scope_condition_issues(topic))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -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()
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@govoplan/identity-trust-webui",
|
||||
"version": "0.1.18",
|
||||
"version": "0.1.19",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "src/index.ts",
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { MetricGrid } from "@govoplan/core-webui";
|
||||
import { useCallback, useEffect, useMemo, useState } from "react";
|
||||
import { Eye, RefreshCw, RotateCw, ShieldOff } from "lucide-react";
|
||||
import {
|
||||
@@ -247,12 +248,12 @@ export default function IdentityTrustPanel({ settings, auth, administrative = fa
|
||||
</Card>}
|
||||
|
||||
<LoadingFrame loading={loading} label="Loading identity trust state">
|
||||
<div className="metric-grid compact">
|
||||
<MetricGrid density="compact">
|
||||
<MetricCard label="Active device keys" value={keys.filter((key) => key.status === "active").length} tone="good" />
|
||||
<MetricCard label="Revoked or expired" value={keys.filter((key) => key.status !== "active").length} tone="warning" />
|
||||
<MetricCard label="Active assurance evidence" value={activeEvidence.length} tone={activeEvidence.length ? "good" : "warning"} />
|
||||
<MetricCard label="Highest assurance" value={humanize(highestAssurance)} tone={highestAssurance === "None" ? "warning" : "info"} />
|
||||
</div>
|
||||
</MetricGrid>
|
||||
|
||||
<Card title="Device keys" actions={<ToggleSwitch label="Show revoked and expired" checked={showRevoked} onChange={setShowRevoked} />}>
|
||||
<p className="muted small-note">Only public key and trust metadata are stored. Revocation blocks future server-mediated use but cannot erase plaintext or key material already obtained by a device.</p>
|
||||
|
||||
Reference in New Issue
Block a user