13 Commits
Author SHA1 Message Date
zemion 909e41b7ec fix(webui): bind key revocation to help
Module Package Release / publish-packages (push) Successful in 11s
2026-08-24 11:36:40 +02:00
zemion 80dabde81d docs(identity-trust): add complete German trust guidance
Module Package Release / publish-packages (push) Successful in 12s
2026-08-23 01:54:20 +02:00
zemion 8d1ca493bb feat(identity-trust): add governed DSAR coverage 2026-08-21 12:15:56 +02:00
zemion b9a158c2cb Adopt shared WebUI layout primitives 2026-08-18 11:30:39 +02:00
zemion 9b0b1112a9 Release v0.1.18
Module Package Release / publish-packages (push) Successful in 12s
2026-08-05 21:07:51 +02:00
zemion ab30da6e79 Release v0.1.17
Module Package Release / publish-packages (push) Successful in 12s
2026-08-05 20:34:13 +02:00
zemion e46a915507 Release v0.1.16
Module Package Release / publish-packages (push) Successful in 12s
2026-08-05 19:52:29 +02:00
zemion 4161a642d8 Release v0.1.15
Module Package Release / publish-packages (push) Successful in 11s
2026-08-04 15:10:27 +02:00
zemion e91820af09 Make package publication retries hash-safe 2026-08-04 14:32:19 +02:00
zemion 54f7a94f13 Harden module package publication 2026-08-04 14:02:40 +02:00
zemion 103c6f5aaf Add protected package release workflow 2026-08-04 04:14:05 +02:00
zemion 9640ec29dd Add identity trust administration surfaces 2026-08-04 01:04:39 +02:00
zemion 0ffc8b6b0f Implement identity trust module 2026-08-01 20:57:27 +02:00
32 changed files with 4452 additions and 5 deletions
+270
View File
@@ -0,0 +1,270 @@
name: Module Package Release
on:
push:
tags:
- "v*"
workflow_dispatch:
inputs:
release_tag:
description: Existing protected version tag to publish
required: true
type: string
jobs:
publish-packages:
runs-on: ubuntu-latest
env:
GITEA_REPOSITORY: ${{ gitea.repository }}
steps:
- uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
with:
fetch-depth: 0
- uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
with:
python-version: "3.12"
- uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
with:
node-version: "22"
- name: Select and validate protected release tag
shell: bash
env:
REQUESTED_TAG: ${{ inputs.release_tag }}
TRIGGER_TAG: ${{ gitea.ref_name }}
run: |
set -euo pipefail
tag="${REQUESTED_TAG:-$TRIGGER_TAG}"
case "$tag" in
v[0-9]*.[0-9]*.[0-9]*) ;;
*) echo "Release tag must start with a SemVer-shaped vX.Y.Z value" >&2; exit 1 ;;
esac
git fetch --force origin "refs/tags/$tag:refs/tags/$tag" refs/heads/main:refs/remotes/origin/main
tag_commit="$(git rev-list -n 1 "$tag")"
git merge-base --is-ancestor "$tag_commit" refs/remotes/origin/main || {
echo "Release tag is not contained in main" >&2
exit 1
}
git checkout --detach "$tag"
printf 'RELEASE_TAG=%s\n' "$tag" >> "$GITEA_ENV"
printf 'SOURCE_DATE_EPOCH=%s\n' "$(git show -s --format=%ct HEAD)" >> "$GITEA_ENV"
- name: Validate package versions
run: |
python - <<'PY'
import json
from pathlib import Path
import os
import re
import tomllib
tag = os.environ["RELEASE_TAG"]
expected = tag.removeprefix("v")
project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
if project.get("version") != expected:
raise SystemExit(f"pyproject version {project.get('version')!r} does not match {tag}")
if re.fullmatch(r"govoplan-[a-z0-9-]+", str(project.get("name", ""))) is None:
raise SystemExit("Python distribution name must use the govoplan-* namespace")
webui = Path("webui/package.json")
if webui.is_file():
package = json.loads(webui.read_text(encoding="utf-8"))
if package.get("version") != expected:
raise SystemExit(f"WebUI version {package.get('version')!r} does not match {tag}")
if re.fullmatch(r"@govoplan/[a-z0-9-]+-webui", str(package.get("name", ""))) is None:
raise SystemExit("WebUI package name must use the @govoplan/*-webui namespace")
release = Path("webui/package.release.json")
if release.is_file():
release_package = json.loads(release.read_text(encoding="utf-8"))
if (
release_package.get("name") != package.get("name")
or release_package.get("version") != expected
):
raise SystemExit("WebUI release package identity does not match package.json and the release tag")
PY
- name: Build immutable package artifacts
shell: bash
run: |
set -euo pipefail
python -m pip install --disable-pip-version-check build==1.5.0 twine==7.0.0
rm -rf dist .package-webui
python -m build --wheel --outdir dist
python -m twine check dist/*.whl
if [[ -f webui/package.json ]]; then
mkdir .package-webui
cp -a webui/. .package-webui/
rm -rf .package-webui/node_modules .package-webui/dist
if [[ -f .package-webui/package.release.json ]]; then
cp .package-webui/package.release.json .package-webui/package.json
fi
node <<'NODE'
const fs = require("node:fs");
const path = ".package-webui/package.json";
const packageJson = JSON.parse(fs.readFileSync(path, "utf8"));
const groups = ["dependencies", "optionalDependencies", "peerDependencies"];
for (const group of groups) {
for (const [name, specifier] of Object.entries(packageJson[group] || {})) {
if (!name.startsWith("@govoplan/")) continue;
if (typeof specifier !== "string") {
throw new Error(`${group}.${name} must use a string version`);
}
const packageSlug = name.slice("@govoplan/".length);
if (!packageSlug.endsWith("-webui")) {
throw new Error(`${group}.${name} is outside the WebUI package namespace`);
}
const repository = `govoplan-${packageSlug.slice(0, -"-webui".length)}`;
const escapedRepository = repository.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
const gitTag = specifier.match(
new RegExp(
`^git\\+(?:ssh://git@|https://)git\\.add-ideas\\.de/(?:GovOPlaN|add-ideas)/${escapedRepository}\\.git#v([0-9]+\\.[0-9]+\\.[0-9]+)$`,
),
);
if (gitTag) {
packageJson[group][name] = gitTag[1];
continue;
}
if (specifier.startsWith("file:") || specifier.startsWith("git+")) {
throw new Error(
`${group}.${name} must resolve to an exact registry version for publication`,
);
}
}
}
delete packageJson.private;
fs.writeFileSync(path, `${JSON.stringify(packageJson, null, 2)}\n`);
NODE
npm pkg delete private --prefix .package-webui
(cd .package-webui && npm pack --ignore-scripts --pack-destination ../dist)
fi
python - <<'PY'
import hashlib
import json
from pathlib import Path
import os
import subprocess
artifacts = []
for path in sorted(Path("dist").iterdir()):
if path.suffix not in {".whl", ".tgz"}:
continue
digest = hashlib.sha256(path.read_bytes()).hexdigest()
artifacts.append({"filename": path.name, "sha256": digest, "size": path.stat().st_size})
payload = {
"schema_version": "1",
"repository": os.environ["GITEA_REPOSITORY"],
"tag": os.environ["RELEASE_TAG"],
"commit": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(),
"artifacts": artifacts,
}
Path("dist/package-artifacts.json").write_text(
json.dumps(payload, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
PY
- name: Retain package hash evidence
uses: actions/upload-artifact@a8a3f3ad30e3422c9c7b888a15615d19a852ae32
with:
name: module-packages-${{ gitea.ref_name }}
path: dist/package-artifacts.json
- name: Check immutable registry state
shell: bash
env:
PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
run: |
set -euo pipefail
test -n "$PACKAGE_TOKEN"
python - <<'PY'
import hashlib
import json
import os
from pathlib import Path
import tomllib
from urllib.error import HTTPError
from urllib.parse import quote
from urllib.request import Request, urlopen
api_root = "https://git.add-ideas.de/api/v1/packages/GovOPlaN"
token = os.environ["PACKAGE_TOKEN"]
def should_publish(kind, name, version, path):
package_url = "/".join(
(api_root, kind, quote(name, safe=""), quote(version, safe=""), "files")
)
request = Request(
package_url,
headers={"Accept": "application/json", "Authorization": f"token {token}"},
)
try:
with urlopen(request, timeout=30) as response:
files = json.load(response)
except HTTPError as exc:
if exc.code == 404:
print(f"{kind} package {name}=={version} is not published yet")
return True
raise
if not isinstance(files, list) or len(files) != 1:
raise SystemExit(
f"immutable {kind} package {name}=={version} has an unexpected file set"
)
expected_sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
if files[0].get("sha256") != expected_sha256:
raise SystemExit(
f"immutable {kind} package {name}=={version} already exists with a different SHA-256"
)
print(f"verified existing {kind} package {name}=={version} ({expected_sha256})")
return False
project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
wheels = tuple(Path("dist").glob("*.whl"))
if len(wheels) != 1:
raise SystemExit("release build must contain exactly one wheel")
publish_pypi = should_publish(
"pypi", str(project["name"]), str(project["version"]), wheels[0]
)
tarballs = tuple(Path("dist").glob("*.tgz"))
if len(tarballs) > 1:
raise SystemExit("release build must contain at most one npm package")
publish_npm = False
if tarballs:
webui = json.loads(
Path(".package-webui/package.json").read_text(encoding="utf-8")
)
publish_npm = should_publish(
"npm", str(webui["name"]), str(webui["version"]), tarballs[0]
)
with Path(os.environ["GITEA_ENV"]).open("a", encoding="utf-8") as env_file:
env_file.write(f"PUBLISH_PYPI={int(publish_pypi)}\n")
env_file.write(f"PUBLISH_NPM={int(publish_npm)}\n")
PY
- name: Publish wheel and WebUI package
shell: bash
env:
PACKAGE_USERNAME: ${{ secrets.GOVOPLAN_PACKAGE_USERNAME }}
PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
run: |
set -euo pipefail
test -n "$PACKAGE_USERNAME"
test -n "$PACKAGE_TOKEN"
if [[ "$PUBLISH_PYPI" == 1 ]]; then
TWINE_USERNAME="$PACKAGE_USERNAME" TWINE_PASSWORD="$PACKAGE_TOKEN" \
python -m twine upload --non-interactive \
--repository-url https://git.add-ideas.de/api/packages/GovOPlaN/pypi \
dist/*.whl
else
echo "Exact wheel is already present; skipping immutable retry."
fi
shopt -s nullglob
webui_packages=(dist/*.tgz)
if (( ${#webui_packages[@]} )) && [[ "$PUBLISH_NPM" == 1 ]]; then
npmrc="$(mktemp)"
trap 'rm -f "$npmrc"' EXIT
chmod 600 "$npmrc"
printf '%s\n' \
'@govoplan:registry=https://git.add-ideas.de/api/packages/GovOPlaN/npm/' \
"//git.add-ideas.de/api/packages/GovOPlaN/npm/:_authToken=$PACKAGE_TOKEN" \
> "$npmrc"
NPM_CONFIG_USERCONFIG="$npmrc" npm publish "./${webui_packages[0]}" \
--ignore-scripts --access public \
--registry https://git.add-ideas.de/api/packages/GovOPlaN/npm/
elif (( ${#webui_packages[@]} )); then
echo "Exact WebUI package is already present; skipping immutable retry."
fi
+14
View File
@@ -0,0 +1,14 @@
# GovOPlaN Identity Trust Codex Guide
## Scope
This module owns public device keys, assurance evidence, trust epochs, and
auditable key-access decisions. Access owns authentication and authorization;
Encryption owns key custody and rewrapping. Never persist private keys or
plaintext content here.
## Documentation Contract
- Update manifest-driven user/admin documentation with behavior changes.
- Keep Gitea issues as the delivery source of truth.
- Do not claim cryptographic protection or certification from trust metadata.
+36 -1
View File
@@ -1,5 +1,40 @@
# govoplan-identity-trust
# GovOPlaN Identity Trust
<!-- govoplan-repository-type:start -->
**Repository type:** module (platform).
<!-- govoplan-repository-type:end -->
`govoplan-identity-trust` owns public device keys, key epochs, bounded
authentication-assurance evidence, and auditable key-access trust decisions.
It deliberately does not own login sessions, resource authorization, private
keys, encryption, or plaintext.
The backend exposes `identity_trust.directory` and
`identity_trust.assurance`. Access or Policy must approve resource access
first; Identity Trust then verifies the acting account, active public device
key, current subject epoch, and assurance evidence. Encryption providers may
consume that decision to rewrap a key, but no key material is returned by this
module.
The WebUI contributes two optional surfaces:
- **Settings > Device trust** lets an account inspect active/revoked public
device keys, revoke a current key with its expected revision, and inspect
assurance level, provider, device binding, expiry, and provenance.
- **Administration > Identity trust** lets an authorized security officer use
an Access-backed account selector when Access is available, inspect the same
bounded projections, rotate subject epochs with an upstream Access decision,
and review immutable key-access decisions. Explicit account references remain
usable when the optional Access directory is absent.
Revocation and rotation are not retroactive: neither can recall plaintext,
exports, or key material already obtained by an endpoint. Stale revisions fail
closed and must be reloaded. The API and UI expose public JWK metadata only;
private JWK parameters are rejected by the capability contract.
Focused verification:
```bash
PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src \
/mnt/DATA/git/govoplan/.venv/bin/python -m unittest discover -s tests
```
+62 -4
View File
@@ -9,15 +9,15 @@ recipient trust.
## Responsibilities
The module should eventually own:
The module owns:
- public key directory
- per-device encryption keys
- account or identity signing-key references
- account or identity public signing-key references
- device registration and revocation
- key rotation and key epochs
- assurance metadata from OIDC, WebAuthn, SAML, or other providers
- recovery and lost-device policy hooks
- bounded assurance evidence and recovery/lost-device policy hooks
- audited key-access and rewrap decisions
It should not own login sessions, tenant membership, groups, functions, roles,
@@ -34,4 +34,62 @@ For encrypted role/function postboxes, identity trust should be able to:
- emit audit events for key fetch or rewrap operations
The module must never require an identity provider or directory service to see
message plaintext or private postbox keys.
message plaintext or private Postbox keys.
## Implemented Boundary
The SQL-backed module now provides:
- tenant-bound public JWK registration with replay safety and explicit
revocation;
- effective key epochs for identities, accounts, functions, Postboxes, and
external recipients, including an explicit retained-history policy;
- immutable allow/deny key-access decisions bound to the upstream Access
decision, acting account, device, epoch, purpose, assignment/delegation, and
resource;
- bounded assurance evidence with provider, level, device, verification,
expiry, and maximum-age checks;
- migrations, uninstall guards, permissions, APIs, capability contracts, and
manifest-driven user/admin documentation.
- bounded user and security-officer projections for public device keys,
assurance provenance, subject epoch history, and immutable key-access
decisions;
- user key revocation and security-officer epoch rotation surfaces with stale
revision rejection and explicit non-retroactivity consequences;
- optional Access-backed account selection without making Access a hard module
dependency.
Only public keys are accepted. JWK private parameters are rejected by the Core
contract. Key-access decisions state explicitly that no cryptographic material
was released. A concrete Encryption provider performs any later key rewrap.
## Epoch And History Semantics
A function or Postbox has one active epoch. Rotation supersedes, but does not
rewrite, the previous epoch. The grant records one explicit history policy,
with `all_retained` as the accepted baseline for a new incumbent. Other bounded
policies may be selected by Policy. Rotation does not itself grant resource
access and ownership transfer does not transfer a private key.
Revocation stops future server-mediated release or rewrap decisions. It cannot
erase plaintext, exports, printouts, or keys already obtained by an endpoint.
That limitation must remain visible to users and operators.
## Recovery
Database recovery restores public keys, epochs, assurance references, and
decision evidence. It cannot restore private device keys. Lost-device recovery
therefore registers a new device key, rotates affected epochs under a separate
authorized workflow, and leaves the old key revoked. Encryption key custody
and quorum recovery remain with `govoplan-encryption` and its selected provider.
## Current Limits
- There is no private-key custody or browser/device key generator.
- Attestation references are retained but no WebAuthn/OIDC attestation adapter
is selected yet.
- Device key generation and registration remains provider/browser driven; the
administration UI deliberately does not ask an operator to paste private or
manually generated key material.
- The module decides trust eligibility; it does not perform encryption,
decryption, signing, or rewrapping.
+22
View File
@@ -0,0 +1,22 @@
[build-system]
requires = ["setuptools>=69", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "govoplan-identity-trust"
version = "0.1.20"
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.37"]
[tool.setuptools.packages.find]
where = ["src"]
[tool.setuptools.package-data]
govoplan_identity_trust = ["py.typed"]
[project.entry-points."govoplan.modules"]
identity_trust = "govoplan_identity_trust.backend.manifest:get_manifest"
+5
View File
@@ -0,0 +1,5 @@
"""GovOPlaN identity-trust module."""
from govoplan_identity_trust.backend.manifest import get_manifest
__all__ = ["get_manifest"]
@@ -0,0 +1 @@
"""Backend integration surface for govoplan-identity-trust."""
@@ -0,0 +1,13 @@
from govoplan_identity_trust.backend.db.models import (
AssuranceEvidence,
DevicePublicKey,
KeyAccessDecisionRecord,
TrustKeyEpoch,
)
__all__ = [
"AssuranceEvidence",
"DevicePublicKey",
"KeyAccessDecisionRecord",
"TrustKeyEpoch",
]
@@ -0,0 +1,190 @@
from __future__ import annotations
from datetime import datetime
from typing import Any
import uuid
from sqlalchemy import DateTime, Index, Integer, JSON, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from govoplan_core.db.base import Base, TimestampMixin
def new_uuid() -> str:
return str(uuid.uuid4())
class DevicePublicKey(Base, TimestampMixin):
__tablename__ = "identity_trust_device_keys"
__table_args__ = (
UniqueConstraint("tenant_id", "key_id", name="uq_identity_trust_device_key"),
UniqueConstraint(
"tenant_id",
"idempotency_key",
name="uq_identity_trust_device_idempotency",
),
Index(
"ix_identity_trust_device_account",
"tenant_id",
"account_id",
"status",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
identity_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
account_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
device_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
key_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
algorithm: Mapped[str] = mapped_column(String(120), nullable=False)
public_jwk: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
purpose: Mapped[str] = mapped_column(String(40), nullable=False)
assurance_level: Mapped[str] = mapped_column(String(80), nullable=False)
attestation_ref: Mapped[str | None] = mapped_column(String(1000), nullable=True)
status: Mapped[str] = mapped_column(
String(30), default="active", nullable=False, index=True
)
epoch: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
registration_digest: Mapped[str] = mapped_column(String(64), nullable=False)
idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
registered_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False
)
expires_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, index=True
)
revoked_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True, index=True
)
revocation_reason: Mapped[str | None] = mapped_column(Text, nullable=True)
created_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
class TrustKeyEpoch(Base, TimestampMixin):
__tablename__ = "identity_trust_key_epochs"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"subject_kind",
"subject_id",
"epoch",
name="uq_identity_trust_key_epoch",
),
UniqueConstraint(
"tenant_id",
"idempotency_key",
name="uq_identity_trust_epoch_idempotency",
),
Index(
"ix_identity_trust_epoch_current",
"tenant_id",
"subject_kind",
"subject_id",
"state",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
subject_kind: Mapped[str] = mapped_column(String(40), nullable=False, index=True)
subject_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
epoch: Mapped[int] = mapped_column(Integer, nullable=False)
previous_epoch: Mapped[int | None] = mapped_column(Integer, nullable=True)
state: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
history_policy: Mapped[str] = mapped_column(String(80), nullable=False)
reason: Mapped[str] = mapped_column(Text, nullable=False)
access_decision_ref: Mapped[str] = mapped_column(String(1000), nullable=False)
idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False)
request_digest: Mapped[str] = mapped_column(String(64), nullable=False)
effective_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False
)
created_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
class AssuranceEvidence(Base, TimestampMixin):
__tablename__ = "identity_trust_assurance_evidence"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"evidence_ref",
name="uq_identity_trust_assurance_ref",
),
Index(
"ix_identity_trust_assurance_account",
"tenant_id",
"account_id",
"verified_at",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
account_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
device_key_id: Mapped[str | None] = mapped_column(
String(255), nullable=True, index=True
)
evidence_ref: Mapped[str] = mapped_column(String(1000), nullable=False)
assurance_level: Mapped[str] = mapped_column(String(80), nullable=False)
provider_id: Mapped[str] = mapped_column(String(120), nullable=False)
verified_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False
)
expires_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, index=True
)
provenance: Mapped[dict[str, Any]] = mapped_column(
JSON, default=dict, nullable=False
)
recorded_by: Mapped[str | None] = mapped_column(String(255), nullable=True)
class KeyAccessDecisionRecord(Base, TimestampMixin):
__tablename__ = "identity_trust_key_access_decisions"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"decision_ref",
name="uq_identity_trust_key_access_decision",
),
Index(
"ix_identity_trust_key_access_subject",
"tenant_id",
"subject_kind",
"subject_id",
"created_at",
),
)
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
decision_ref: Mapped[str] = mapped_column(String(255), nullable=False)
request_digest: Mapped[str] = mapped_column(String(64), nullable=False)
account_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
device_key_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
subject_kind: Mapped[str] = mapped_column(String(40), nullable=False, index=True)
subject_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
key_epoch: Mapped[int] = mapped_column(Integer, nullable=False)
access_decision_ref: Mapped[str] = mapped_column(String(1000), nullable=False)
purpose: Mapped[str] = mapped_column(String(255), nullable=False)
allowed: Mapped[bool] = mapped_column(nullable=False)
reason: Mapped[str] = mapped_column(Text, nullable=False)
resource_ref: Mapped[str | None] = mapped_column(String(1000), nullable=True)
function_assignment_id: Mapped[str | None] = mapped_column(
String(255), nullable=True
)
delegation_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
provenance: Mapped[dict[str, Any]] = mapped_column(
JSON, default=dict, nullable=False
)
__all__ = [
"AssuranceEvidence",
"DevicePublicKey",
"KeyAccessDecisionRecord",
"TrustKeyEpoch",
"new_uuid",
]
@@ -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"]
@@ -0,0 +1,464 @@
from __future__ import annotations
from pathlib import Path
from govoplan_core.core.identity_trust import (
CAPABILITY_IDENTITY_TRUST_ASSURANCE,
CAPABILITY_IDENTITY_TRUST_DIRECTORY,
)
from govoplan_core.core.module_guards import (
drop_table_retirement_provider,
persistent_table_uninstall_guard,
)
from govoplan_core.core.modules import (
CapabilityDocumentation,
DocumentationCondition,
DocumentationLink,
DocumentationTopic,
FrontendModule,
MigrationSpec,
ModuleContext,
ModuleInterfaceProvider,
ModuleManifest,
PermissionDefinition,
RoleTemplate,
ViewSurface,
)
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.20"
DEVICE_READ_SCOPE = "identity_trust:device:read"
DEVICE_WRITE_SCOPE = "identity_trust:device:write"
KEY_ACCESS_SCOPE = "identity_trust:key_access:approve"
ASSURANCE_SCOPE = "identity_trust:assurance:record"
ASSURANCE_READ_SCOPE = "identity_trust:assurance:read"
ADMIN_SCOPE = "identity_trust:device:admin"
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
scope=scope,
label=label,
description=description,
category="Identity Trust",
level="tenant",
module_id=module_id,
resource=resource,
action=action,
)
def _router(_context: ModuleContext):
from govoplan_identity_trust.backend.router import router
return router
def _service(_context: ModuleContext) -> SqlIdentityTrustService:
return SqlIdentityTrustService()
def _dsar_provider(_context: ModuleContext) -> IdentityTrustDsarProvider:
return IdentityTrustDsarProvider()
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
version=MODULE_VERSION,
optional_dependencies=("access", "audit", "policy", "encryption", "postbox"),
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(
DEVICE_READ_SCOPE,
"View device keys",
"View public device-key and trust state.",
),
_permission(
DEVICE_WRITE_SCOPE,
"Manage own device keys",
"Register and revoke public keys for the acting account.",
),
_permission(
KEY_ACCESS_SCOPE,
"Evaluate key access",
"Evaluate device and key-epoch trust after Access has approved a resource action.",
),
_permission(
ASSURANCE_READ_SCOPE,
"View assurance evidence",
"View bounded assurance state and provenance for the acting account or, with administrative authority, another account.",
),
_permission(
ASSURANCE_SCOPE,
"Record assurance evidence",
"Record bounded assurance evidence from a trusted authentication provider.",
),
_permission(
ADMIN_SCOPE,
"Administer identity trust",
"Administer device keys, trust epochs, and assurance evidence.",
),
),
role_templates=(
RoleTemplate(
slug="identity_trust_user",
name="Identity trust user",
description="Manage own public device keys.",
permissions=(DEVICE_READ_SCOPE, DEVICE_WRITE_SCOPE, ASSURANCE_READ_SCOPE),
),
RoleTemplate(
slug="identity_trust_officer",
name="Identity trust officer",
description="Administer trust epochs and assurance evidence.",
permissions=(
DEVICE_READ_SCOPE,
ASSURANCE_READ_SCOPE,
KEY_ACCESS_SCOPE,
ASSURANCE_SCOPE,
ADMIN_SCOPE,
),
),
),
route_factory=_router,
frontend=FrontendModule(
module_id=MODULE_ID,
package_name="@govoplan/identity-trust-webui",
view_surfaces=(
ViewSurface(
id="identity_trust.settings.devices",
module_id=MODULE_ID,
kind="section",
label="Device trust",
order=10,
),
ViewSurface(
id="identity_trust.admin.trust",
module_id=MODULE_ID,
kind="section",
label="Identity trust administration",
order=20,
),
ViewSurface(
id="identity_trust.admin.epochs",
module_id=MODULE_ID,
kind="section",
label="Key epoch administration",
parent_id="identity_trust.admin.trust",
order=30,
),
ViewSurface(
id="identity_trust.admin.decisions",
module_id=MODULE_ID,
kind="section",
label="Key-access decisions",
parent_id="identity_trust.admin.trust",
order=40,
),
),
),
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(
label="Identity Trust directory",
summary="Resolves public device keys, key epochs, and auditable release decisions without private key material.",
contract_version="1.0.0",
),
CAPABILITY_IDENTITY_TRUST_ASSURANCE: CapabilityDocumentation(
label="Identity Trust assurance",
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,
metadata=Base.metadata,
script_location=str(Path(__file__).with_name("migrations") / "versions"),
retirement_supported=True,
retirement_provider=drop_table_retirement_provider(
models.KeyAccessDecisionRecord,
models.AssuranceEvidence,
models.TrustKeyEpoch,
models.DevicePublicKey,
label="Identity Trust",
),
retirement_notes="Destructive retirement removes public-key, epoch, assurance, and key-access evidence after a database snapshot.",
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
models.DevicePublicKey,
models.TrustKeyEpoch,
models.AssuranceEvidence,
models.KeyAccessDecisionRecord,
label="Identity Trust",
),
),
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",
summary="Separate login authority from public device-key and cryptographic-access trust.",
body=(
"Identity Trust stores public keys only. Users can review and revoke their device keys and inspect assurance provenance in Settings. Security officers can select an authorized account, inspect revoked or compromised-device evidence, rotate subject key epochs, and review key-access decisions in Administration. Access first decides whether an account may reach a protected resource; Identity Trust then verifies the current device and key epoch and records an auditable decision. Function and Postbox history grants are explicit epoch policy, and revocation cannot erase plaintext already obtained. Every revoke and rotation is revision-bound and stale actions must be reloaded."
),
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",
href="govoplan-identity-trust/docs/DEVICE_KEY_TRUST_CONCEPT.md",
kind="repository",
),
),
),
),
architecture=declared_module_architecture(
layer="institutional_foundation",
kind="foundation",
maturity="vertical_slice",
documentation_ref="docs/DEVICE_KEY_TRUST_CONCEPT.md",
test_ref="tests/test_identity_trust.py",
known_limits=(
"No private key custody, device attestation verifier, or cryptographic rewrap implementation is included.",
),
owned_concepts=(
"public device key",
"key epoch",
"assurance evidence",
"key-access trust decision",
),
non_owned_concepts=(
"login session",
"resource authorization",
"private key",
"content encryption",
),
migration_docs=("docs/DEVICE_KEY_TRUST_CONCEPT.md",),
recovery_docs=("docs/DEVICE_KEY_TRUST_CONCEPT.md",),
security_docs=("docs/DEVICE_KEY_TRUST_CONCEPT.md",),
operations_docs=("README.md",),
),
)
def get_manifest() -> ModuleManifest:
return manifest
__all__ = [
"ADMIN_SCOPE",
"ASSURANCE_READ_SCOPE",
"ASSURANCE_SCOPE",
"DEVICE_READ_SCOPE",
"DEVICE_WRITE_SCOPE",
"KEY_ACCESS_SCOPE",
"MODULE_ID",
"MODULE_VERSION",
"get_manifest",
"manifest",
]
@@ -0,0 +1 @@
"""Identity Trust Alembic migrations."""
@@ -0,0 +1 @@
"""Identity Trust migration versions."""
@@ -0,0 +1,190 @@
"""v0.1.14 identity-trust public keys and epochs
Revision ID: c3f5a7b9d1e2
Revises: None
Create Date: 2026-08-01 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = "c3f5a7b9d1e2"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"identity_trust_device_keys",
sa.Column("id", sa.String(36), primary_key=True),
sa.Column("tenant_id", sa.String(36), nullable=False),
sa.Column("identity_id", sa.String(255), nullable=False),
sa.Column("account_id", sa.String(255), nullable=False),
sa.Column("device_id", sa.String(255), nullable=False),
sa.Column("key_id", sa.String(255), nullable=False),
sa.Column("algorithm", sa.String(120), nullable=False),
sa.Column("public_jwk", sa.JSON(), nullable=False),
sa.Column("purpose", sa.String(40), nullable=False),
sa.Column("assurance_level", sa.String(80), nullable=False),
sa.Column("attestation_ref", sa.String(1000), nullable=True),
sa.Column("status", sa.String(30), nullable=False),
sa.Column("epoch", sa.Integer(), nullable=False),
sa.Column("registration_digest", sa.String(64), nullable=False),
sa.Column("idempotency_key", sa.String(255), nullable=False),
sa.Column("registered_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("revocation_reason", sa.Text(), nullable=True),
sa.Column("created_by", sa.String(255), nullable=True),
sa.Column("updated_by", sa.String(255), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint("tenant_id", "key_id", name="uq_identity_trust_device_key"),
sa.UniqueConstraint(
"tenant_id", "idempotency_key", name="uq_identity_trust_device_idempotency"
),
)
op.create_index(
"ix_identity_trust_device_account",
"identity_trust_device_keys",
["tenant_id", "account_id", "status"],
)
for name in (
"tenant_id",
"identity_id",
"account_id",
"device_id",
"key_id",
"status",
"expires_at",
"revoked_at",
):
op.create_index(
f"ix_identity_trust_device_keys_{name}",
"identity_trust_device_keys",
[name],
)
op.create_table(
"identity_trust_key_epochs",
sa.Column("id", sa.String(36), primary_key=True),
sa.Column("tenant_id", sa.String(36), nullable=False),
sa.Column("subject_kind", sa.String(40), nullable=False),
sa.Column("subject_id", sa.String(255), nullable=False),
sa.Column("epoch", sa.Integer(), nullable=False),
sa.Column("previous_epoch", sa.Integer(), nullable=True),
sa.Column("state", sa.String(30), nullable=False),
sa.Column("history_policy", sa.String(80), nullable=False),
sa.Column("reason", sa.Text(), nullable=False),
sa.Column("access_decision_ref", sa.String(1000), nullable=False),
sa.Column("idempotency_key", sa.String(255), nullable=False),
sa.Column("request_digest", sa.String(64), nullable=False),
sa.Column("effective_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("created_by", sa.String(255), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint(
"tenant_id",
"subject_kind",
"subject_id",
"epoch",
name="uq_identity_trust_key_epoch",
),
sa.UniqueConstraint(
"tenant_id", "idempotency_key", name="uq_identity_trust_epoch_idempotency"
),
)
op.create_index(
"ix_identity_trust_epoch_current",
"identity_trust_key_epochs",
["tenant_id", "subject_kind", "subject_id", "state"],
)
for name in ("tenant_id", "subject_kind", "subject_id", "state"):
op.create_index(
f"ix_identity_trust_key_epochs_{name}", "identity_trust_key_epochs", [name]
)
op.create_table(
"identity_trust_assurance_evidence",
sa.Column("id", sa.String(36), primary_key=True),
sa.Column("tenant_id", sa.String(36), nullable=False),
sa.Column("account_id", sa.String(255), nullable=False),
sa.Column("device_key_id", sa.String(255), nullable=True),
sa.Column("evidence_ref", sa.String(1000), nullable=False),
sa.Column("assurance_level", sa.String(80), nullable=False),
sa.Column("provider_id", sa.String(120), nullable=False),
sa.Column("verified_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("provenance", sa.JSON(), nullable=False),
sa.Column("recorded_by", sa.String(255), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint(
"tenant_id", "evidence_ref", name="uq_identity_trust_assurance_ref"
),
)
op.create_index(
"ix_identity_trust_assurance_account",
"identity_trust_assurance_evidence",
["tenant_id", "account_id", "verified_at"],
)
for name in ("tenant_id", "account_id", "device_key_id", "expires_at"):
op.create_index(
f"ix_identity_trust_assurance_evidence_{name}",
"identity_trust_assurance_evidence",
[name],
)
op.create_table(
"identity_trust_key_access_decisions",
sa.Column("id", sa.String(36), primary_key=True),
sa.Column("tenant_id", sa.String(36), nullable=False),
sa.Column("decision_ref", sa.String(255), nullable=False),
sa.Column("request_digest", sa.String(64), nullable=False),
sa.Column("account_id", sa.String(255), nullable=False),
sa.Column("device_key_id", sa.String(255), nullable=False),
sa.Column("subject_kind", sa.String(40), nullable=False),
sa.Column("subject_id", sa.String(255), nullable=False),
sa.Column("key_epoch", sa.Integer(), nullable=False),
sa.Column("access_decision_ref", sa.String(1000), nullable=False),
sa.Column("purpose", sa.String(255), nullable=False),
sa.Column("allowed", sa.Boolean(), nullable=False),
sa.Column("reason", sa.Text(), nullable=False),
sa.Column("resource_ref", sa.String(1000), nullable=True),
sa.Column("function_assignment_id", sa.String(255), nullable=True),
sa.Column("delegation_id", sa.String(255), nullable=True),
sa.Column("provenance", sa.JSON(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint(
"tenant_id", "decision_ref", name="uq_identity_trust_key_access_decision"
),
)
op.create_index(
"ix_identity_trust_key_access_subject",
"identity_trust_key_access_decisions",
["tenant_id", "subject_kind", "subject_id", "created_at"],
)
for name in (
"tenant_id",
"account_id",
"device_key_id",
"subject_kind",
"subject_id",
):
op.create_index(
f"ix_identity_trust_key_access_decisions_{name}",
"identity_trust_key_access_decisions",
[name],
)
def downgrade() -> None:
op.drop_table("identity_trust_key_access_decisions")
op.drop_table("identity_trust_assurance_evidence")
op.drop_table("identity_trust_key_epochs")
op.drop_table("identity_trust_device_keys")
@@ -0,0 +1,406 @@
from __future__ import annotations
from dataclasses import asdict
from datetime import UTC, datetime
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from govoplan_core.audit.logging import audit_event
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.core.identity_trust import (
AssuranceCheckRequest,
DeviceKeyRegistration,
KeyAccessRequest,
KeyEpochRotationRequest,
)
from govoplan_core.core.references import (
access_scope_reference_options,
access_scope_reference_provider_available,
)
from govoplan_core.core.runtime import get_registry
from govoplan_core.db.session import get_session
from govoplan_identity_trust.backend.manifest import (
ADMIN_SCOPE,
ASSURANCE_READ_SCOPE,
ASSURANCE_SCOPE,
DEVICE_READ_SCOPE,
DEVICE_WRITE_SCOPE,
KEY_ACCESS_SCOPE,
)
from govoplan_identity_trust.backend.schemas import (
AssuranceCheckPayload,
AssuranceEvidenceListResponse,
AssuranceEvidencePayload,
AssuranceEvidenceResponse,
AssuranceResponse,
DeviceKeyListResponse,
DeviceKeyRegisterPayload,
DeviceKeyResponse,
DeviceKeyRevokePayload,
EpochResponse,
EpochListResponse,
EpochRotatePayload,
KeyAccessPayload,
KeyAccessDecisionItem,
KeyAccessDecisionListResponse,
KeyAccessResponse,
ReferenceOptionsResponse,
)
from govoplan_identity_trust.backend.service import (
IdentityTrustAccessDenied,
IdentityTrustError,
SqlIdentityTrustService,
record_assurance_evidence,
)
router = APIRouter(prefix="/identity-trust", tags=["identity-trust"])
service = SqlIdentityTrustService()
def _require(principal: ApiPrincipal, *scopes: str) -> None:
if any(has_scope(principal, scope) for scope in scopes):
return
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Missing one of: {', '.join(scopes)}",
)
def _error(exc: IdentityTrustError) -> HTTPException:
return HTTPException(
status_code=(
status.HTTP_403_FORBIDDEN
if isinstance(exc, IdentityTrustAccessDenied)
else status.HTTP_409_CONFLICT
),
detail=str(exc),
)
def _device_response(value) -> DeviceKeyResponse:
return DeviceKeyResponse(**asdict(value))
def _epoch_response(value) -> EpochResponse:
return EpochResponse(**asdict(value))
def _assurance_response(value) -> AssuranceEvidenceResponse:
expires_at = value.expires_at
if expires_at.tzinfo is None:
expires_at = expires_at.replace(tzinfo=UTC)
return AssuranceEvidenceResponse(
id=value.id,
tenant_id=value.tenant_id,
account_id=value.account_id,
device_key_id=value.device_key_id,
evidence_ref=value.evidence_ref,
assurance_level=value.assurance_level,
provider_id=value.provider_id,
verified_at=value.verified_at,
expires_at=value.expires_at,
active=expires_at >= datetime.now(UTC),
provenance=dict(value.provenance or {}),
)
@router.post("/device-keys", response_model=DeviceKeyResponse)
def api_register_device_key(
payload: DeviceKeyRegisterPayload,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DeviceKeyResponse:
_require(principal, DEVICE_WRITE_SCOPE, ADMIN_SCOPE)
try:
value = service.register_device_key(
session,
principal,
request=DeviceKeyRegistration(
tenant_id=principal.tenant_id,
**payload.model_dump(),
),
)
except (IdentityTrustError, ValueError) as exc:
raise _error(IdentityTrustError(str(exc))) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="identity_trust.device_key.registered",
object_type="device_public_key",
object_id=value.key_id,
details={"algorithm": value.algorithm, "private_material": False},
)
session.commit()
return _device_response(value)
@router.get("/device-keys", response_model=DeviceKeyListResponse)
def api_list_device_keys(
account_id: str,
active_only: bool = Query(default=True),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DeviceKeyListResponse:
_require(principal, DEVICE_READ_SCOPE, ADMIN_SCOPE)
try:
values = service.list_device_keys(
session,
principal,
tenant_id=principal.tenant_id,
account_id=account_id,
active_only=active_only,
)
except IdentityTrustError as exc:
raise _error(exc) from exc
return DeviceKeyListResponse(keys=[_device_response(value) for value in values])
@router.get(
"/account-options",
response_model=ReferenceOptionsResponse,
)
def api_account_options(
q: str = Query(default="", max_length=200),
selected: list[str] = Query(default=[]),
limit: int = Query(default=50, ge=1, le=200),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> ReferenceOptionsResponse:
_require(principal, ADMIN_SCOPE)
registry = get_registry()
options = access_scope_reference_options(
registry,
principal,
scope_type="user",
reference_kind="user",
query=q,
selected_values=selected,
limit=limit,
administrative=True,
session=session,
)
return ReferenceOptionsResponse(
options=[option.to_dict() for option in options],
provider_available=access_scope_reference_provider_available(registry),
)
@router.get(
"/assurance/evidence",
response_model=AssuranceEvidenceListResponse,
)
def api_list_assurance_evidence(
account_id: str,
active_only: bool = Query(default=False),
limit: int = Query(default=200, ge=1, le=500),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> AssuranceEvidenceListResponse:
_require(principal, ASSURANCE_READ_SCOPE, ADMIN_SCOPE)
try:
values = service.list_assurance_evidence(
session,
principal,
tenant_id=principal.tenant_id,
account_id=account_id,
active_only=active_only,
limit=limit,
)
except IdentityTrustError as exc:
raise _error(exc) from exc
return AssuranceEvidenceListResponse(
evidence=[_assurance_response(value) for value in values]
)
@router.post("/device-keys/{key_id}/revoke", response_model=DeviceKeyResponse)
def api_revoke_device_key(
key_id: str,
payload: DeviceKeyRevokePayload,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> DeviceKeyResponse:
_require(principal, DEVICE_WRITE_SCOPE, ADMIN_SCOPE)
try:
value = service.revoke_device_key(
session,
principal,
tenant_id=principal.tenant_id,
key_id=key_id,
expected_epoch=payload.expected_epoch,
reason=payload.reason,
)
except IdentityTrustError as exc:
raise _error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="identity_trust.device_key.revoked",
object_type="device_public_key",
object_id=key_id,
details={"reason": payload.reason, "epoch": value.epoch},
)
session.commit()
return _device_response(value)
@router.post("/epochs/rotate", response_model=EpochResponse)
def api_rotate_epoch(
payload: EpochRotatePayload,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> EpochResponse:
_require(principal, ADMIN_SCOPE)
try:
value = service.rotate_epoch(
session,
principal,
request=KeyEpochRotationRequest(
tenant_id=principal.tenant_id,
**payload.model_dump(),
),
)
except (IdentityTrustError, ValueError) as exc:
raise _error(IdentityTrustError(str(exc))) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="identity_trust.key_epoch.rotated",
object_type=payload.subject_kind,
object_id=payload.subject_id,
details={"epoch": value.epoch, "history_policy": value.history_policy},
)
session.commit()
return _epoch_response(value)
@router.get("/epochs", response_model=EpochListResponse)
def api_list_epochs(
subject_kind: str,
subject_id: str,
limit: int = Query(default=200, ge=1, le=500),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> EpochListResponse:
_require(principal, ADMIN_SCOPE)
try:
values = service.list_epochs(
session,
principal,
tenant_id=principal.tenant_id,
subject_kind=subject_kind,
subject_id=subject_id,
limit=limit,
)
except IdentityTrustError as exc:
raise _error(exc) from exc
return EpochListResponse(epochs=[_epoch_response(value) for value in values])
@router.post("/key-access/decide", response_model=KeyAccessResponse)
def api_decide_key_access(
payload: KeyAccessPayload,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> KeyAccessResponse:
_require(principal, KEY_ACCESS_SCOPE, ADMIN_SCOPE)
try:
value = service.decide_key_access(
session,
principal,
request=KeyAccessRequest(
tenant_id=principal.tenant_id,
**payload.model_dump(),
),
)
except (IdentityTrustError, ValueError) as exc:
raise _error(IdentityTrustError(str(exc))) from exc
session.commit()
data = asdict(value)
data["requirements"] = list(value.requirements)
return KeyAccessResponse(**data)
@router.get(
"/key-access/decisions",
response_model=KeyAccessDecisionListResponse,
)
def api_list_key_access_decisions(
account_id: str,
limit: int = Query(default=200, ge=1, le=500),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> KeyAccessDecisionListResponse:
_require(principal, KEY_ACCESS_SCOPE, ADMIN_SCOPE)
try:
values = service.list_key_access_decisions(
session,
principal,
tenant_id=principal.tenant_id,
account_id=account_id,
limit=limit,
)
except IdentityTrustError as exc:
raise _error(exc) from exc
return KeyAccessDecisionListResponse(
decisions=[
KeyAccessDecisionItem.model_validate(value, from_attributes=True)
for value in values
]
)
@router.post("/assurance/evidence", response_model=dict[str, object])
def api_record_assurance(
payload: AssuranceEvidencePayload,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> dict[str, object]:
_require(principal, ASSURANCE_SCOPE, ADMIN_SCOPE)
try:
value = record_assurance_evidence(
session,
principal,
tenant_id=principal.tenant_id,
**payload.model_dump(),
)
except IdentityTrustError as exc:
raise _error(exc) from exc
audit_event(
session,
tenant_id=principal.tenant_id,
user_id=getattr(principal.user, "id", None),
api_key_id=principal.api_key_id,
action="identity_trust.assurance.recorded",
object_type="assurance_evidence",
object_id=value.id,
details={"provider_id": value.provider_id, "level": value.assurance_level},
)
session.commit()
return {"id": value.id, "evidence_ref": value.evidence_ref}
@router.post("/assurance/check", response_model=AssuranceResponse)
def api_check_assurance(
payload: AssuranceCheckPayload,
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> AssuranceResponse:
_require(principal, KEY_ACCESS_SCOPE, ADMIN_SCOPE)
value = service.verify_assurance(
session,
principal,
request=AssuranceCheckRequest(
tenant_id=principal.tenant_id,
**payload.model_dump(),
),
)
return AssuranceResponse(**asdict(value))
@@ -0,0 +1,183 @@
from __future__ import annotations
from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, Field
class DeviceKeyRegisterPayload(BaseModel):
identity_id: str = Field(min_length=1, max_length=255)
account_id: str = Field(min_length=1, max_length=255)
device_id: str = Field(min_length=1, max_length=255)
key_id: str = Field(min_length=1, max_length=255)
algorithm: str = Field(min_length=1, max_length=120)
public_jwk: dict[str, Any]
purpose: Literal["encryption", "signing", "encryption_and_signing"] = "encryption"
assurance_level: str = Field(default="software", min_length=1, max_length=80)
attestation_ref: str | None = Field(default=None, max_length=1000)
expires_at: datetime | None = None
idempotency_key: str = Field(min_length=1, max_length=255)
class DeviceKeyRevokePayload(BaseModel):
expected_epoch: int = Field(ge=1)
reason: str = Field(min_length=1, max_length=4000)
class DeviceKeyResponse(BaseModel):
tenant_id: str
identity_id: str
account_id: str
device_id: str
key_id: str
algorithm: str
public_jwk: dict[str, Any]
purpose: str
assurance_level: str
status: str
epoch: int
registered_at: datetime
attestation_ref: str | None = None
expires_at: datetime | None = None
revoked_at: datetime | None = None
revocation_reason: str | None = None
provenance: dict[str, Any] = Field(default_factory=dict)
class DeviceKeyListResponse(BaseModel):
keys: list[DeviceKeyResponse]
class AssuranceEvidenceResponse(BaseModel):
id: str
tenant_id: str
account_id: str
device_key_id: str | None = None
evidence_ref: str
assurance_level: str
provider_id: str
verified_at: datetime
expires_at: datetime
active: bool
provenance: dict[str, Any] = Field(default_factory=dict)
class AssuranceEvidenceListResponse(BaseModel):
evidence: list[AssuranceEvidenceResponse]
class EpochListResponse(BaseModel):
epochs: list[EpochResponse]
class EpochRotatePayload(BaseModel):
subject_kind: Literal[
"identity", "account", "function", "postbox", "external_recipient"
]
subject_id: str = Field(min_length=1, max_length=255)
reason: str = Field(min_length=1, max_length=4000)
access_decision_ref: str = Field(min_length=1, max_length=1000)
idempotency_key: str = Field(min_length=1, max_length=255)
history_policy: str = Field(default="all_retained", min_length=1, max_length=80)
previous_epoch: int | None = Field(default=None, ge=1)
class EpochResponse(BaseModel):
tenant_id: str
subject_kind: str
subject_id: str
epoch: int
state: str
history_policy: str
effective_at: datetime
previous_epoch: int | None = None
reason: str | None = None
access_decision_ref: str | None = None
provenance: dict[str, Any] = Field(default_factory=dict)
class KeyAccessPayload(BaseModel):
account_id: str = Field(min_length=1, max_length=255)
device_key_id: str = Field(min_length=1, max_length=255)
subject_kind: Literal[
"identity", "account", "function", "postbox", "external_recipient"
]
subject_id: str = Field(min_length=1, max_length=255)
key_epoch: int = Field(ge=1)
access_decision_ref: str = Field(min_length=1, max_length=1000)
purpose: str = Field(min_length=1, max_length=255)
requested_at: datetime
function_assignment_id: str | None = Field(default=None, max_length=255)
delegation_id: str | None = Field(default=None, max_length=255)
resource_ref: str | None = Field(default=None, max_length=1000)
class KeyAccessResponse(BaseModel):
allowed: bool
decision_ref: str
reason: str
device_key: DeviceKeyResponse | None = None
epoch: EpochResponse | None = None
audit_event_ref: str | None = None
requirements: list[str] = Field(default_factory=list)
provenance: dict[str, Any] = Field(default_factory=dict)
class KeyAccessDecisionItem(BaseModel):
id: str
decision_ref: str
account_id: str
device_key_id: str
subject_kind: str
subject_id: str
key_epoch: int
access_decision_ref: str
purpose: str
allowed: bool
reason: str
resource_ref: str | None = None
function_assignment_id: str | None = None
delegation_id: str | None = None
provenance: dict[str, Any] = Field(default_factory=dict)
created_at: datetime
class KeyAccessDecisionListResponse(BaseModel):
decisions: list[KeyAccessDecisionItem]
class ReferenceOptionsResponse(BaseModel):
options: list[dict[str, Any]] = Field(default_factory=list)
provider_available: bool = False
class AssuranceEvidencePayload(BaseModel):
account_id: str = Field(min_length=1, max_length=255)
evidence_ref: str = Field(min_length=1, max_length=1000)
assurance_level: str = Field(min_length=1, max_length=80)
provider_id: str = Field(min_length=1, max_length=120)
verified_at: datetime
expires_at: datetime
device_key_id: str | None = Field(default=None, max_length=255)
provenance: dict[str, Any] = Field(default_factory=dict)
class AssuranceCheckPayload(BaseModel):
account_id: str = Field(min_length=1, max_length=255)
purpose: str = Field(min_length=1, max_length=255)
minimum_level: str = Field(min_length=1, max_length=80)
evidence_ref: str = Field(min_length=1, max_length=1000)
evaluated_at: datetime
maximum_age_seconds: int = Field(default=300, ge=1, le=86400)
device_key_id: str | None = Field(default=None, max_length=255)
class AssuranceResponse(BaseModel):
allowed: bool
reason: str
assurance_level: str | None = None
evidence_ref: str | None = None
verified_at: datetime | None = None
expires_at: datetime | None = None
provenance: dict[str, Any] = Field(default_factory=dict)
@@ -0,0 +1,743 @@
from __future__ import annotations
from collections.abc import Mapping
from datetime import datetime, timezone
import hashlib
import json
from sqlalchemy import select
from sqlalchemy.orm import Session
from govoplan_core.core.identity_trust import (
AssuranceCheckRequest,
AssuranceDecision,
DeviceKeyRef,
DeviceKeyRegistration,
KeyAccessDecision,
KeyAccessRequest,
KeyEpochRef,
KeyEpochRotationRequest,
)
from govoplan_core.db.base import utcnow
from govoplan_identity_trust.backend.db.models import (
AssuranceEvidence,
DevicePublicKey,
KeyAccessDecisionRecord,
TrustKeyEpoch,
)
class IdentityTrustError(ValueError):
pass
class IdentityTrustAccessDenied(IdentityTrustError):
pass
class SqlIdentityTrustService:
def register_device_key(
self,
session: object,
principal: object,
*,
request: DeviceKeyRegistration,
) -> DeviceKeyRef:
db = _session(session)
_require_tenant(principal, request.tenant_id)
actor_id = _account_id(principal)
if request.account_id != actor_id and not _has_scope(
principal, "identity_trust:device:admin"
):
raise IdentityTrustError(
"A device key can only be registered for the acting account."
)
digest = _digest(_registration_payload(request))
existing = db.scalar(
select(DevicePublicKey).where(
DevicePublicKey.tenant_id == request.tenant_id,
DevicePublicKey.key_id == request.key_id,
)
)
if existing is not None:
if (
existing.registration_digest != digest
or existing.idempotency_key != request.idempotency_key
):
raise IdentityTrustError(
"The public key id already exists with different evidence."
)
return _device_ref(existing)
replay = db.scalar(
select(DevicePublicKey).where(
DevicePublicKey.tenant_id == request.tenant_id,
DevicePublicKey.idempotency_key == request.idempotency_key,
)
)
if replay is not None:
if replay.registration_digest != digest:
raise IdentityTrustError(
"The idempotency key was used for different public-key data."
)
return _device_ref(replay)
item = DevicePublicKey(
tenant_id=request.tenant_id,
identity_id=request.identity_id,
account_id=request.account_id,
device_id=request.device_id,
key_id=request.key_id,
algorithm=request.algorithm,
public_jwk=dict(request.public_jwk),
purpose=request.purpose,
assurance_level=request.assurance_level,
attestation_ref=request.attestation_ref,
status="active",
epoch=1,
registration_digest=digest,
idempotency_key=request.idempotency_key,
registered_at=_as_utc(utcnow()),
expires_at=request.expires_at,
created_by=actor_id,
updated_by=actor_id,
)
db.add(item)
db.flush()
return _device_ref(item)
def revoke_device_key(
self,
session: object,
principal: object,
*,
tenant_id: str,
key_id: str,
expected_epoch: int,
reason: str,
) -> DeviceKeyRef:
db = _session(session)
_require_tenant(principal, tenant_id)
item = _device_key(db, tenant_id=tenant_id, key_id=key_id, lock=True)
if item.account_id != _account_id(principal) and not _has_scope(
principal, "identity_trust:device:admin"
):
raise IdentityTrustError(
"The acting account cannot revoke this device key."
)
if expected_epoch != item.epoch:
raise IdentityTrustError(
"The device key changed; reload before revoking it."
)
if item.status == "revoked":
if item.revocation_reason != reason.strip():
raise IdentityTrustError(
"The device key is already revoked for another reason."
)
return _device_ref(item)
item.status = "revoked"
item.revoked_at = _as_utc(utcnow())
item.revocation_reason = _required(reason, "revocation reason")
item.epoch += 1
item.updated_by = _account_id(principal)
db.flush()
return _device_ref(item)
def list_device_keys(
self,
session: object,
principal: object,
*,
tenant_id: str,
account_id: str,
active_only: bool = True,
) -> tuple[DeviceKeyRef, ...]:
db = _session(session)
_require_tenant(principal, tenant_id)
if not _may_read_account(principal, account_id):
raise IdentityTrustAccessDenied(
"The acting account cannot inspect these device keys."
)
statement = select(DevicePublicKey).where(
DevicePublicKey.tenant_id == tenant_id,
DevicePublicKey.account_id == account_id,
)
if active_only:
statement = statement.where(DevicePublicKey.status == "active")
return tuple(
_device_ref(item)
for item in db.scalars(
statement.order_by(
DevicePublicKey.registered_at.desc(),
DevicePublicKey.id,
)
)
)
def list_assurance_evidence(
self,
session: object,
principal: object,
*,
tenant_id: str,
account_id: str,
active_only: bool = False,
limit: int = 200,
) -> tuple[AssuranceEvidence, ...]:
db = _session(session)
_require_tenant(principal, tenant_id)
if not _may_read_account(principal, account_id):
raise IdentityTrustAccessDenied(
"The acting account cannot inspect this assurance evidence."
)
statement = select(AssuranceEvidence).where(
AssuranceEvidence.tenant_id == tenant_id,
AssuranceEvidence.account_id == account_id,
)
if active_only:
statement = statement.where(
AssuranceEvidence.expires_at >= _as_utc(utcnow())
)
return tuple(
db.scalars(
statement.order_by(
AssuranceEvidence.verified_at.desc(),
AssuranceEvidence.id,
).limit(max(1, min(int(limit), 500)))
)
)
def list_epochs(
self,
session: object,
principal: object,
*,
tenant_id: str,
subject_kind: str,
subject_id: str,
limit: int = 200,
) -> tuple[KeyEpochRef, ...]:
db = _session(session)
_require_tenant(principal, tenant_id)
if not _has_scope(principal, "identity_trust:device:admin"):
raise IdentityTrustAccessDenied(
"Identity Trust administration is required to inspect key epochs."
)
values = db.scalars(
select(TrustKeyEpoch)
.where(
TrustKeyEpoch.tenant_id == tenant_id,
TrustKeyEpoch.subject_kind == subject_kind,
TrustKeyEpoch.subject_id == subject_id,
)
.order_by(TrustKeyEpoch.epoch.desc())
.limit(max(1, min(int(limit), 500)))
)
return tuple(_epoch_ref(value) for value in values)
def list_key_access_decisions(
self,
session: object,
principal: object,
*,
tenant_id: str,
account_id: str,
limit: int = 200,
) -> tuple[KeyAccessDecisionRecord, ...]:
db = _session(session)
_require_tenant(principal, tenant_id)
if not (
_has_scope(principal, "identity_trust:key_access:approve")
or _has_scope(principal, "identity_trust:device:admin")
):
raise IdentityTrustAccessDenied(
"Key-access decision authority is required to inspect decisions."
)
return tuple(
db.scalars(
select(KeyAccessDecisionRecord)
.where(
KeyAccessDecisionRecord.tenant_id == tenant_id,
KeyAccessDecisionRecord.account_id == account_id,
)
.order_by(
KeyAccessDecisionRecord.created_at.desc(),
KeyAccessDecisionRecord.id,
)
.limit(max(1, min(int(limit), 500)))
)
)
def rotate_epoch(
self,
session: object,
principal: object,
*,
request: KeyEpochRotationRequest,
) -> KeyEpochRef:
db = _session(session)
_require_tenant(principal, request.tenant_id)
digest = _digest(
{
"tenant_id": request.tenant_id,
"subject_kind": request.subject_kind,
"subject_id": request.subject_id,
"reason": request.reason,
"access_decision_ref": request.access_decision_ref,
"history_policy": request.history_policy,
"previous_epoch": request.previous_epoch,
}
)
replay = db.scalar(
select(TrustKeyEpoch).where(
TrustKeyEpoch.tenant_id == request.tenant_id,
TrustKeyEpoch.idempotency_key == request.idempotency_key,
)
)
if replay is not None:
if replay.request_digest != digest:
raise IdentityTrustError(
"The epoch idempotency key was used with another request."
)
return _epoch_ref(replay)
current = db.scalar(
select(TrustKeyEpoch)
.where(
TrustKeyEpoch.tenant_id == request.tenant_id,
TrustKeyEpoch.subject_kind == request.subject_kind,
TrustKeyEpoch.subject_id == request.subject_id,
TrustKeyEpoch.state == "active",
)
.order_by(TrustKeyEpoch.epoch.desc())
.with_for_update()
)
current_epoch = current.epoch if current else None
if request.previous_epoch != current_epoch:
raise IdentityTrustError(
"The key epoch changed; reload before rotating it."
)
if current is not None:
current.state = "superseded"
item = TrustKeyEpoch(
tenant_id=request.tenant_id,
subject_kind=request.subject_kind,
subject_id=request.subject_id,
epoch=(current_epoch or 0) + 1,
previous_epoch=current_epoch,
state="active",
history_policy=_required(request.history_policy, "history policy"),
reason=request.reason.strip(),
access_decision_ref=request.access_decision_ref.strip(),
idempotency_key=request.idempotency_key,
request_digest=digest,
effective_at=_as_utc(utcnow()),
created_by=_account_id(principal),
)
db.add(item)
db.flush()
return _epoch_ref(item)
def resolve_epoch(
self,
session: object,
*,
tenant_id: str,
subject_kind: str,
subject_id: str,
epoch: int | None = None,
) -> KeyEpochRef | None:
db = _session(session)
statement = select(TrustKeyEpoch).where(
TrustKeyEpoch.tenant_id == tenant_id,
TrustKeyEpoch.subject_kind == subject_kind,
TrustKeyEpoch.subject_id == subject_id,
)
if epoch is None:
statement = statement.where(TrustKeyEpoch.state == "active")
else:
statement = statement.where(TrustKeyEpoch.epoch == epoch)
item = db.scalar(statement.order_by(TrustKeyEpoch.epoch.desc()))
return _epoch_ref(item) if item else None
def decide_key_access(
self,
session: object,
principal: object,
*,
request: KeyAccessRequest,
) -> KeyAccessDecision:
db = _session(session)
_require_tenant(principal, request.tenant_id)
request_payload = {
"tenant_id": request.tenant_id,
"account_id": request.account_id,
"device_key_id": request.device_key_id,
"subject_kind": request.subject_kind,
"subject_id": request.subject_id,
"key_epoch": request.key_epoch,
"access_decision_ref": request.access_decision_ref,
"purpose": request.purpose,
"function_assignment_id": request.function_assignment_id,
"delegation_id": request.delegation_id,
"resource_ref": request.resource_ref,
}
digest = _digest(request_payload)
decision_ref = f"identity-trust:key-access:{digest}"
existing = db.scalar(
select(KeyAccessDecisionRecord).where(
KeyAccessDecisionRecord.tenant_id == request.tenant_id,
KeyAccessDecisionRecord.decision_ref == decision_ref,
)
)
if existing is not None:
return _decision_ref(db, existing)
device = db.scalar(
select(DevicePublicKey).where(
DevicePublicKey.tenant_id == request.tenant_id,
DevicePublicKey.key_id == request.device_key_id,
)
)
epoch = db.scalar(
select(TrustKeyEpoch).where(
TrustKeyEpoch.tenant_id == request.tenant_id,
TrustKeyEpoch.subject_kind == request.subject_kind,
TrustKeyEpoch.subject_id == request.subject_id,
TrustKeyEpoch.epoch == request.key_epoch,
)
)
allowed = True
reason = "Current device, epoch, and upstream access evidence are valid."
if request.account_id != _account_id(principal):
allowed = False
reason = "The access request does not belong to the acting account."
elif device is None or device.account_id != request.account_id:
allowed = False
reason = "The requested device key is not registered for this account."
elif _device_status(device) != "active":
allowed = False
reason = "The requested device key is not active."
elif epoch is None or epoch.state != "active":
allowed = False
reason = "The requested key epoch is not active."
item = KeyAccessDecisionRecord(
tenant_id=request.tenant_id,
decision_ref=decision_ref,
request_digest=digest,
account_id=request.account_id,
device_key_id=request.device_key_id,
subject_kind=request.subject_kind,
subject_id=request.subject_id,
key_epoch=request.key_epoch,
access_decision_ref=request.access_decision_ref,
purpose=request.purpose,
allowed=allowed,
reason=reason,
resource_ref=request.resource_ref,
function_assignment_id=request.function_assignment_id,
delegation_id=request.delegation_id,
provenance={
"upstream_access_decision_ref": request.access_decision_ref,
"requested_at": request.requested_at.isoformat(),
"cryptographic_material_released": False,
},
)
db.add(item)
db.flush()
return _decision_ref(db, item)
def verify_assurance(
self,
session: object,
principal: object,
*,
request: AssuranceCheckRequest,
) -> AssuranceDecision:
db = _session(session)
_require_tenant(principal, request.tenant_id)
if request.account_id != _account_id(principal):
return AssuranceDecision(
allowed=False,
reason="Assurance evidence belongs to another account.",
)
evidence = db.scalar(
select(AssuranceEvidence).where(
AssuranceEvidence.tenant_id == request.tenant_id,
AssuranceEvidence.account_id == request.account_id,
AssuranceEvidence.evidence_ref == request.evidence_ref,
)
)
now = _as_utc(request.evaluated_at)
if evidence is None:
return AssuranceDecision(
allowed=False,
reason="Assurance evidence was not found.",
)
age = (now - _as_utc(evidence.verified_at)).total_seconds()
allowed = (
_as_utc(evidence.expires_at) >= now
and age <= request.maximum_age_seconds
and _assurance_rank(evidence.assurance_level)
>= _assurance_rank(request.minimum_level)
and (
request.device_key_id is None
or evidence.device_key_id == request.device_key_id
)
)
return AssuranceDecision(
allowed=allowed,
reason=(
"Assurance evidence satisfies the requested level and age."
if allowed
else "Assurance evidence is stale, insufficient, expired, or for another device."
),
assurance_level=evidence.assurance_level,
evidence_ref=evidence.evidence_ref,
verified_at=evidence.verified_at,
expires_at=evidence.expires_at,
provenance={"provider_id": evidence.provider_id},
)
def record_assurance_evidence(
session: Session,
principal: object,
*,
tenant_id: str,
account_id: str,
evidence_ref: str,
assurance_level: str,
provider_id: str,
verified_at: datetime,
expires_at: datetime,
device_key_id: str | None = None,
provenance: Mapping[str, object] | None = None,
) -> AssuranceEvidence:
_require_tenant(principal, tenant_id)
if _as_utc(expires_at) <= _as_utc(verified_at):
raise IdentityTrustError("Assurance expiry must follow verification.")
existing = session.scalar(
select(AssuranceEvidence).where(
AssuranceEvidence.tenant_id == tenant_id,
AssuranceEvidence.evidence_ref == evidence_ref,
)
)
payload = {
"account_id": account_id,
"device_key_id": device_key_id,
"assurance_level": assurance_level,
"provider_id": provider_id,
"verified_at": verified_at.isoformat(),
"expires_at": expires_at.isoformat(),
"provenance": dict(provenance or {}),
}
if existing is not None:
current = {
"account_id": existing.account_id,
"device_key_id": existing.device_key_id,
"assurance_level": existing.assurance_level,
"provider_id": existing.provider_id,
"verified_at": existing.verified_at.isoformat(),
"expires_at": existing.expires_at.isoformat(),
"provenance": dict(existing.provenance),
}
if _digest(current) != _digest(payload):
raise IdentityTrustError(
"The assurance reference already exists with different evidence."
)
return existing
item = AssuranceEvidence(
tenant_id=tenant_id,
account_id=account_id,
device_key_id=device_key_id,
evidence_ref=_required(evidence_ref, "evidence reference"),
assurance_level=_required(assurance_level, "assurance level"),
provider_id=_required(provider_id, "provider id"),
verified_at=_as_utc(verified_at),
expires_at=_as_utc(expires_at),
provenance=dict(provenance or {}),
recorded_by=_account_id(principal),
)
session.add(item)
session.flush()
return item
def _device_key(
session: Session,
*,
tenant_id: str,
key_id: str,
lock: bool = False,
) -> DevicePublicKey:
statement = select(DevicePublicKey).where(
DevicePublicKey.tenant_id == tenant_id,
DevicePublicKey.key_id == key_id,
)
if lock:
statement = statement.with_for_update()
item = session.scalar(statement)
if item is None:
raise IdentityTrustError("Device key not found.")
return item
def _device_ref(item: DevicePublicKey) -> DeviceKeyRef:
return DeviceKeyRef(
tenant_id=item.tenant_id,
identity_id=item.identity_id,
account_id=item.account_id,
device_id=item.device_id,
key_id=item.key_id,
algorithm=item.algorithm,
public_jwk=dict(item.public_jwk),
purpose=item.purpose, # type: ignore[arg-type]
assurance_level=item.assurance_level,
status=_device_status(item), # type: ignore[arg-type]
epoch=item.epoch,
registered_at=item.registered_at,
attestation_ref=item.attestation_ref,
expires_at=item.expires_at,
revoked_at=item.revoked_at,
revocation_reason=item.revocation_reason,
provenance={"registration_digest": item.registration_digest},
)
def _epoch_ref(item: TrustKeyEpoch) -> KeyEpochRef:
return KeyEpochRef(
tenant_id=item.tenant_id,
subject_kind=item.subject_kind, # type: ignore[arg-type]
subject_id=item.subject_id,
epoch=item.epoch,
state=item.state, # type: ignore[arg-type]
history_policy=item.history_policy,
effective_at=item.effective_at,
previous_epoch=item.previous_epoch,
reason=item.reason,
access_decision_ref=item.access_decision_ref,
provenance={"request_digest": item.request_digest},
)
def _decision_ref(
session: Session,
item: KeyAccessDecisionRecord,
) -> KeyAccessDecision:
device = session.scalar(
select(DevicePublicKey).where(
DevicePublicKey.tenant_id == item.tenant_id,
DevicePublicKey.key_id == item.device_key_id,
)
)
epoch = session.scalar(
select(TrustKeyEpoch).where(
TrustKeyEpoch.tenant_id == item.tenant_id,
TrustKeyEpoch.subject_kind == item.subject_kind,
TrustKeyEpoch.subject_id == item.subject_id,
TrustKeyEpoch.epoch == item.key_epoch,
)
)
return KeyAccessDecision(
allowed=item.allowed,
decision_ref=item.decision_ref,
reason=item.reason,
device_key=_device_ref(device) if device else None,
epoch=_epoch_ref(epoch) if epoch else None,
audit_event_ref=f"identity-trust-decision:{item.id}",
requirements=() if item.allowed else ("current_device", "current_epoch"),
provenance=dict(item.provenance),
)
def _registration_payload(request: DeviceKeyRegistration) -> dict[str, object]:
return {
"tenant_id": request.tenant_id,
"identity_id": request.identity_id,
"account_id": request.account_id,
"device_id": request.device_id,
"key_id": request.key_id,
"algorithm": request.algorithm,
"public_jwk": dict(request.public_jwk),
"purpose": request.purpose,
"assurance_level": request.assurance_level,
"attestation_ref": request.attestation_ref,
"expires_at": request.expires_at.isoformat() if request.expires_at else None,
}
def _device_status(item: DevicePublicKey) -> str:
if item.status == "active" and item.expires_at is not None:
if _as_utc(item.expires_at) < _as_utc(utcnow()):
return "expired"
return item.status
def _assurance_rank(value: str) -> int:
return {
"none": 0,
"software": 1,
"mfa": 2,
"hardware": 3,
"high": 4,
}.get(value.strip().lower(), 0)
def _session(value: object) -> Session:
if not isinstance(value, Session):
raise TypeError("Identity Trust requires a SQLAlchemy Session.")
return value
def _require_tenant(principal: object, tenant_id: str) -> None:
if str(getattr(principal, "tenant_id", "")) != tenant_id:
raise IdentityTrustAccessDenied(
"Cross-tenant identity-trust access is denied."
)
def _account_id(principal: object) -> str:
return _required(str(getattr(principal, "account_id", "")), "account id")
def _has_scope(principal: object, scope: str) -> bool:
if hasattr(principal, "has"):
return bool(principal.has(scope))
return scope in set(getattr(principal, "scopes", ()))
def _may_read_account(principal: object, account_id: str) -> bool:
return (
account_id == _account_id(principal)
or _has_scope(principal, "identity_trust:device:read_all")
or _has_scope(principal, "identity_trust:device:admin")
)
def _required(value: str, label: str) -> str:
cleaned = value.strip()
if not cleaned:
raise IdentityTrustError(f"{label.capitalize()} is required.")
return cleaned
def _digest(value: Mapping[str, object]) -> str:
encoded = json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=True,
default=str,
)
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
def _as_utc(value: datetime) -> datetime:
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
__all__ = [
"IdentityTrustAccessDenied",
"IdentityTrustError",
"SqlIdentityTrustService",
"record_assurance_evidence",
]
View File
+31
View File
@@ -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()
+297
View File
@@ -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()
+318
View File
@@ -0,0 +1,318 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.core.identity_trust import (
AssuranceCheckRequest,
DeviceKeyRegistration,
KeyAccessRequest,
KeyEpochRotationRequest,
)
from govoplan_identity_trust.backend.db.models import (
AssuranceEvidence,
DevicePublicKey,
KeyAccessDecisionRecord,
TrustKeyEpoch,
)
from govoplan_identity_trust.backend.service import (
IdentityTrustAccessDenied,
IdentityTrustError,
SqlIdentityTrustService,
record_assurance_evidence,
)
NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
class Principal:
tenant_id = "tenant-1"
account_id = "account-1"
def has(self, scope: str) -> bool:
return scope in {
"identity_trust:device:admin",
"identity_trust:device:read_all",
"identity_trust:key_access:approve",
}
class RestrictedPrincipal:
tenant_id = "tenant-1"
account_id = "account-2"
def has(self, scope: str) -> bool:
return False
def registration(**changes) -> DeviceKeyRegistration:
values = {
"tenant_id": "tenant-1",
"identity_id": "identity-1",
"account_id": "account-1",
"device_id": "device-1",
"key_id": "key-1",
"algorithm": "X25519",
"public_jwk": {"kty": "OKP", "crv": "X25519", "x": "public"},
"purpose": "encryption",
"assurance_level": "hardware",
"idempotency_key": "register-1",
}
values.update(changes)
return DeviceKeyRegistration(**values)
class IdentityTrustTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite+pysqlite:///:memory:")
self.tables = [
DevicePublicKey.__table__,
TrustKeyEpoch.__table__,
AssuranceEvidence.__table__,
KeyAccessDecisionRecord.__table__,
]
for table in self.tables:
table.create(self.engine)
self.session = Session(self.engine)
self.service = SqlIdentityTrustService()
self.principal = Principal()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def test_public_key_registration_replay_revoke_and_private_rejection(self) -> None:
first = self.service.register_device_key(
self.session,
self.principal,
request=registration(),
)
replay = self.service.register_device_key(
self.session,
self.principal,
request=registration(),
)
self.assertEqual(first.key_id, replay.key_id)
with self.assertRaisesRegex(ValueError, "public JWK"):
registration(
key_id="private",
idempotency_key="private",
public_jwk={"kty": "OKP", "x": "public", "d": "private"},
)
revoked = self.service.revoke_device_key(
self.session,
self.principal,
tenant_id="tenant-1",
key_id="key-1",
expected_epoch=1,
reason="Device lost.",
)
self.assertEqual("revoked", revoked.status)
with self.assertRaisesRegex(IdentityTrustError, "changed"):
self.service.revoke_device_key(
self.session,
self.principal,
tenant_id="tenant-1",
key_id="key-1",
expected_epoch=1,
reason="Device lost.",
)
def test_epoch_history_and_key_access_decision(self) -> None:
self.service.register_device_key(
self.session,
self.principal,
request=registration(),
)
epoch = self.service.rotate_epoch(
self.session,
self.principal,
request=KeyEpochRotationRequest(
tenant_id="tenant-1",
subject_kind="postbox",
subject_id="postbox-1",
reason="Initial incumbent.",
access_decision_ref="access:grant-1",
idempotency_key="epoch-1",
previous_epoch=None,
history_policy="all_retained",
),
)
decision = self.service.decide_key_access(
self.session,
self.principal,
request=KeyAccessRequest(
tenant_id="tenant-1",
account_id="account-1",
device_key_id="key-1",
subject_kind="postbox",
subject_id="postbox-1",
key_epoch=epoch.epoch,
access_decision_ref="access:grant-1",
purpose="postbox.message.read",
requested_at=NOW,
resource_ref="postbox-message:1",
),
)
self.assertTrue(decision.allowed)
self.assertFalse(decision.provenance["cryptographic_material_released"])
next_epoch = self.service.rotate_epoch(
self.session,
self.principal,
request=KeyEpochRotationRequest(
tenant_id="tenant-1",
subject_kind="postbox",
subject_id="postbox-1",
reason="Incumbency changed.",
access_decision_ref="access:grant-2",
idempotency_key="epoch-2",
previous_epoch=1,
history_policy="all_retained",
),
)
self.assertEqual(2, next_epoch.epoch)
self.assertEqual(
"superseded",
self.service.resolve_epoch(
self.session,
tenant_id="tenant-1",
subject_kind="postbox",
subject_id="postbox-1",
epoch=1,
).state,
)
def test_assurance_must_be_recent_sufficient_and_device_bound(self) -> None:
record_assurance_evidence(
self.session,
self.principal,
tenant_id="tenant-1",
account_id="account-1",
evidence_ref="webauthn:assertion-1",
assurance_level="hardware",
provider_id="webauthn",
verified_at=NOW,
expires_at=NOW + timedelta(minutes=10),
device_key_id="key-1",
)
allowed = self.service.verify_assurance(
self.session,
self.principal,
request=AssuranceCheckRequest(
tenant_id="tenant-1",
account_id="account-1",
purpose="encryption.recovery.approve",
minimum_level="mfa",
evidence_ref="webauthn:assertion-1",
evaluated_at=NOW + timedelta(minutes=2),
maximum_age_seconds=300,
device_key_id="key-1",
),
)
stale = self.service.verify_assurance(
self.session,
self.principal,
request=AssuranceCheckRequest(
tenant_id="tenant-1",
account_id="account-1",
purpose="encryption.recovery.approve",
minimum_level="mfa",
evidence_ref="webauthn:assertion-1",
evaluated_at=NOW + timedelta(minutes=6),
maximum_age_seconds=300,
device_key_id="key-1",
),
)
self.assertTrue(allowed.allowed)
self.assertFalse(stale.allowed)
def test_bounded_projections_enforce_subject_access_and_keep_provenance(self) -> None:
self.service.register_device_key(
self.session,
self.principal,
request=registration(),
)
record_assurance_evidence(
self.session,
self.principal,
tenant_id="tenant-1",
account_id="account-1",
evidence_ref="webauthn:assertion-2",
assurance_level="hardware",
provider_id="webauthn",
verified_at=NOW,
expires_at=NOW + timedelta(minutes=10),
device_key_id="key-1",
provenance={"ceremony": "uv", "policy_ref": "policy:assurance:1"},
)
epoch = self.service.rotate_epoch(
self.session,
self.principal,
request=KeyEpochRotationRequest(
tenant_id="tenant-1",
subject_kind="postbox",
subject_id="postbox-1",
reason="Initial epoch.",
access_decision_ref="access:grant-1",
idempotency_key="epoch-list-1",
previous_epoch=None,
history_policy="all_retained",
),
)
self.service.decide_key_access(
self.session,
self.principal,
request=KeyAccessRequest(
tenant_id="tenant-1",
account_id="account-1",
device_key_id="key-1",
subject_kind="postbox",
subject_id="postbox-1",
key_epoch=epoch.epoch,
access_decision_ref="access:grant-1",
purpose="postbox.message.read",
requested_at=NOW,
),
)
evidence = self.service.list_assurance_evidence(
self.session,
self.principal,
tenant_id="tenant-1",
account_id="account-1",
)
epochs = self.service.list_epochs(
self.session,
self.principal,
tenant_id="tenant-1",
subject_kind="postbox",
subject_id="postbox-1",
)
decisions = self.service.list_key_access_decisions(
self.session,
self.principal,
tenant_id="tenant-1",
account_id="account-1",
)
self.assertEqual(1, len(evidence))
self.assertEqual("policy:assurance:1", evidence[0].provenance["policy_ref"])
self.assertEqual((1,), tuple(item.epoch for item in epochs))
self.assertEqual(1, len(decisions))
self.assertFalse(decisions[0].provenance["cryptographic_material_released"])
with self.assertRaises(IdentityTrustAccessDenied):
self.service.list_assurance_evidence(
self.session,
RestrictedPrincipal(),
tenant_id="tenant-1",
account_id="account-1",
)
if __name__ == "__main__":
unittest.main()
+43
View File
@@ -0,0 +1,43 @@
from __future__ import annotations
import unittest
from govoplan_core.core.identity_trust import (
CAPABILITY_IDENTITY_TRUST_ASSURANCE,
CAPABILITY_IDENTITY_TRUST_DIRECTORY,
)
from govoplan_identity_trust.backend.manifest import get_manifest
class IdentityTrustManifestTests(unittest.TestCase):
def test_manifest_exposes_headless_trust_capabilities(self) -> None:
manifest = get_manifest()
self.assertEqual("identity_trust", manifest.id)
self.assertEqual((), manifest.dependencies)
self.assertIn(
CAPABILITY_IDENTITY_TRUST_DIRECTORY,
manifest.capability_factories,
)
self.assertIn(
CAPABILITY_IDENTITY_TRUST_ASSURANCE,
manifest.capability_factories,
)
self.assertIsNotNone(manifest.frontend)
self.assertEqual(
"@govoplan/identity-trust-webui",
manifest.frontend.package_name,
)
self.assertIn(
"identity_trust.settings.devices",
{surface.id for surface in manifest.frontend.view_surfaces},
)
self.assertIn(
"identity_trust:assurance:read",
{permission.scope for permission in manifest.permissions},
)
self.assertIsNotNone(manifest.migration_spec)
self.assertEqual("vertical_slice", manifest.architecture.maturity)
if __name__ == "__main__":
unittest.main()
+46
View File
@@ -0,0 +1,46 @@
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from alembic.runtime.migration import MigrationContext
from sqlalchemy import create_engine, inspect
from govoplan_core.db.migrations import migrate_database
from govoplan_identity_trust.backend.manifest import get_manifest
class IdentityTrustMigrationTests(unittest.TestCase):
def test_migration_creates_all_trust_tables(self) -> None:
with tempfile.TemporaryDirectory(
prefix="govoplan-identity-trust-"
) as directory:
url = f"sqlite:///{Path(directory) / 'trust.db'}"
migrate_database(
database_url=url,
enabled_modules=("identity_trust",),
manifest_factories=(get_manifest,),
)
engine = create_engine(url)
try:
tables = set(inspect(engine).get_table_names())
self.assertTrue(
{
"identity_trust_device_keys",
"identity_trust_key_epochs",
"identity_trust_assurance_evidence",
"identity_trust_key_access_decisions",
}.issubset(tables)
)
with engine.connect() as connection:
self.assertIn(
"c3f5a7b9d1e2",
set(MigrationContext.configure(connection).get_current_heads()),
)
finally:
engine.dispose()
if __name__ == "__main__":
unittest.main()
+30
View File
@@ -0,0 +1,30 @@
{
"name": "@govoplan/identity-trust-webui",
"version": "0.1.20",
"private": true,
"type": "module",
"main": "src/index.ts",
"module": "src/index.ts",
"types": "src/index.ts",
"exports": {
".": {
"types": "./src/index.ts",
"import": "./src/index.ts"
},
"./styles/identity-trust.css": "./src/styles/identity-trust.css"
},
"peerDependencies": {
"@govoplan/core-webui": "^0.1.18",
"lucide-react": "^1.23.0",
"react": ">=19.2.7 <20",
"react-dom": ">=19.2.7 <20"
},
"peerDependenciesMeta": {
"@govoplan/core-webui": {
"optional": true
}
},
"scripts": {
"test:identity-trust-ui": "node tests/identity-trust-ui-structure.test.mjs"
}
}
+123
View File
@@ -0,0 +1,123 @@
import {
apiFetch,
apiReferenceOptionProvider,
type ApiSettings,
type ReferenceOptionProvider
} from "@govoplan/core-webui";
export type DeviceKey = {
tenant_id: string;
identity_id: string;
account_id: string;
device_id: string;
key_id: string;
algorithm: string;
purpose: string;
assurance_level: string;
status: string;
epoch: number;
registered_at: string;
attestation_ref?: string | null;
expires_at?: string | null;
revoked_at?: string | null;
revocation_reason?: string | null;
provenance: Record<string, unknown>;
};
export type AssuranceEvidence = {
id: string;
account_id: string;
device_key_id?: string | null;
evidence_ref: string;
assurance_level: string;
provider_id: string;
verified_at: string;
expires_at: string;
active: boolean;
provenance: Record<string, unknown>;
};
export type KeyEpoch = {
tenant_id: string;
subject_kind: string;
subject_id: string;
epoch: number;
state: string;
history_policy: string;
effective_at: string;
previous_epoch?: number | null;
reason?: string | null;
access_decision_ref?: string | null;
provenance: Record<string, unknown>;
};
export type KeyAccessDecision = {
id: string;
decision_ref: string;
account_id: string;
device_key_id: string;
subject_kind: string;
subject_id: string;
key_epoch: number;
access_decision_ref: string;
purpose: string;
allowed: boolean;
reason: string;
resource_ref?: string | null;
function_assignment_id?: string | null;
delegation_id?: string | null;
provenance: Record<string, unknown>;
created_at: string;
};
export type EpochRotatePayload = {
subject_kind: "identity" | "account" | "function" | "postbox" | "external_recipient";
subject_id: string;
reason: string;
access_decision_ref: string;
idempotency_key: string;
history_policy: string;
previous_epoch?: number | null;
};
export async function listDeviceKeys(settings: ApiSettings, accountId: string, activeOnly = false): Promise<DeviceKey[]> {
const params = new URLSearchParams({ account_id: accountId, active_only: String(activeOnly) });
const response = await apiFetch<{ keys: DeviceKey[] }>(settings, `/api/v1/identity-trust/device-keys?${params}`);
return response.keys;
}
export async function revokeDeviceKey(settings: ApiSettings, keyId: string, expectedEpoch: number, reason: string): Promise<DeviceKey> {
return apiFetch<DeviceKey>(settings, `/api/v1/identity-trust/device-keys/${encodeURIComponent(keyId)}/revoke`, {
method: "POST",
body: JSON.stringify({ expected_epoch: expectedEpoch, reason })
});
}
export async function listAssuranceEvidence(settings: ApiSettings, accountId: string, activeOnly = false): Promise<AssuranceEvidence[]> {
const params = new URLSearchParams({ account_id: accountId, active_only: String(activeOnly) });
const response = await apiFetch<{ evidence: AssuranceEvidence[] }>(settings, `/api/v1/identity-trust/assurance/evidence?${params}`);
return response.evidence;
}
export async function listEpochs(settings: ApiSettings, subjectKind: string, subjectId: string): Promise<KeyEpoch[]> {
const params = new URLSearchParams({ subject_kind: subjectKind, subject_id: subjectId });
const response = await apiFetch<{ epochs: KeyEpoch[] }>(settings, `/api/v1/identity-trust/epochs?${params}`);
return response.epochs;
}
export async function rotateEpoch(settings: ApiSettings, payload: EpochRotatePayload): Promise<KeyEpoch> {
return apiFetch<KeyEpoch>(settings, "/api/v1/identity-trust/epochs/rotate", {
method: "POST",
body: JSON.stringify(payload)
});
}
export async function listKeyAccessDecisions(settings: ApiSettings, accountId: string): Promise<KeyAccessDecision[]> {
const params = new URLSearchParams({ account_id: accountId });
const response = await apiFetch<{ decisions: KeyAccessDecision[] }>(settings, `/api/v1/identity-trust/key-access/decisions?${params}`);
return response.decisions;
}
export function identityTrustAccountProvider(settings: ApiSettings): ReferenceOptionProvider {
return apiReferenceOptionProvider(settings, "/api/v1/identity-trust/account-options");
}
+333
View File
@@ -0,0 +1,333 @@
import { MetricGrid } from "@govoplan/core-webui";
import { useCallback, useEffect, useMemo, useState } from "react";
import { Eye, RefreshCw, RotateCw, ShieldOff } from "lucide-react";
import {
AdminPageLayout,
Button,
Card,
DataGrid,
Dialog,
DismissibleAlert,
FormField,
LoadingFrame,
MetricCard,
ReferenceSelect,
StatusBadge,
TableActionGroup,
ToggleSwitch,
hasScope,
type ApiSettings,
type AuthInfo,
type DataGridColumn
} from "@govoplan/core-webui";
import {
identityTrustAccountProvider,
listAssuranceEvidence,
listDeviceKeys,
listEpochs,
listKeyAccessDecisions,
revokeDeviceKey,
rotateEpoch,
type AssuranceEvidence,
type DeviceKey,
type KeyAccessDecision,
type KeyEpoch
} from "../api/identityTrust";
type IdentityTrustPanelProps = {
settings: ApiSettings;
auth: AuthInfo;
administrative?: boolean;
};
type EpochDraft = {
subjectKind: "identity" | "account" | "function" | "postbox" | "external_recipient";
subjectId: string;
reason: string;
accessDecisionRef: string;
};
const EMPTY_EPOCH_DRAFT: EpochDraft = {
subjectKind: "postbox",
subjectId: "",
reason: "",
accessDecisionRef: ""
};
export default function IdentityTrustPanel({ settings, auth, administrative = false }: IdentityTrustPanelProps) {
const ownAccountId = auth.principal?.account_id || auth.user.account_id;
const canRevokeDevice = hasScope(auth, "identity_trust:device:write")
|| hasScope(auth, "identity_trust:device:admin");
const [accountId, setAccountId] = useState(ownAccountId);
const [keys, setKeys] = useState<DeviceKey[]>([]);
const [evidence, setEvidence] = useState<AssuranceEvidence[]>([]);
const [decisions, setDecisions] = useState<KeyAccessDecision[]>([]);
const [epochs, setEpochs] = useState<KeyEpoch[]>([]);
const [showRevoked, setShowRevoked] = useState(false);
const [loading, setLoading] = useState(false);
const [busy, setBusy] = useState(false);
const [error, setError] = useState("");
const [success, setSuccess] = useState("");
const [revoking, setRevoking] = useState<DeviceKey | null>(null);
const [revocationReason, setRevocationReason] = useState("");
const [selectedEvidence, setSelectedEvidence] = useState<AssuranceEvidence | null>(null);
const [selectedDecision, setSelectedDecision] = useState<KeyAccessDecision | null>(null);
const [epochDraft, setEpochDraft] = useState<EpochDraft>(EMPTY_EPOCH_DRAFT);
const accountProvider = useMemo(() => identityTrustAccountProvider(settings), [settings]);
const loadAccount = useCallback(async () => {
if (!accountId) return;
setLoading(true);
setError("");
try {
const [nextKeys, nextEvidence, nextDecisions] = await Promise.all([
listDeviceKeys(settings, accountId, false),
listAssuranceEvidence(settings, accountId, false),
administrative ? listKeyAccessDecisions(settings, accountId) : Promise.resolve([])
]);
setKeys(nextKeys);
setEvidence(nextEvidence);
setDecisions(nextDecisions);
} catch (caught) {
setError(errorMessage(caught));
setKeys([]);
setEvidence([]);
setDecisions([]);
} finally {
setLoading(false);
}
}, [accountId, administrative, settings]);
useEffect(() => {
void loadAccount();
}, [loadAccount]);
const visibleKeys = showRevoked ? keys : keys.filter((key) => key.status === "active");
const activeEvidence = evidence.filter((item) => item.active);
const highestAssurance = activeEvidence
.map((item) => item.assurance_level)
.sort((left, right) => assuranceRank(right) - assuranceRank(left))[0] ?? "None";
const keyColumns = useMemo<DataGridColumn<DeviceKey>[]>(() => [
{ id: "device", header: "Device", width: 180, sortable: true, filterable: true, render: (row) => row.device_id, value: (row) => row.device_id },
{ id: "key", header: "Public key", width: 220, sortable: true, filterable: true, render: (row) => row.key_id, value: (row) => row.key_id },
{ id: "purpose", header: "Purpose", width: 170, sortable: true, filterable: true, render: (row) => humanize(row.purpose), value: (row) => row.purpose },
{ id: "algorithm", header: "Algorithm", width: 130, sortable: true, filterable: true, render: (row) => row.algorithm, value: (row) => row.algorithm },
{ id: "assurance", header: "Assurance", width: 140, sortable: true, filterable: true, render: (row) => humanize(row.assurance_level), value: (row) => row.assurance_level },
{ id: "status", header: "Status", width: 130, sortable: true, filterable: true, render: (row) => <StatusBadge status={row.status} />, value: (row) => row.status },
{ id: "epoch", header: "Revision", width: 100, sortable: true, filterable: true, filterType: "integer", render: (row) => row.epoch, value: (row) => row.epoch },
{ id: "expiry", header: "Expiry", width: 180, sortable: true, filterable: true, filterType: "date", render: (row) => formatDateTime(row.expires_at), value: (row) => row.expires_at ?? "" },
{
id: "actions",
header: "Actions",
width: 90,
sticky: "end",
align: "right",
render: (row) => <TableActionGroup actions={[
{
id: "revoke",
label: "Revoke device key",
icon: <ShieldOff aria-hidden="true" />,
variant: "danger",
applicable: row.status === "active",
disabled: !canRevokeDevice,
disabledReason: !canRevokeDevice
? "Device-key write permission is required."
: undefined,
onClick: () => {
setRevocationReason("");
setRevoking(row);
}
}
]} />
}
], [canRevokeDevice]);
const evidenceColumns = useMemo<DataGridColumn<AssuranceEvidence>[]>(() => [
{ id: "level", header: "Level", width: 130, sortable: true, filterable: true, render: (row) => humanize(row.assurance_level), value: (row) => row.assurance_level },
{ id: "provider", header: "Provider", width: 160, sortable: true, filterable: true, render: (row) => row.provider_id, value: (row) => row.provider_id },
{ id: "evidence", header: "Evidence reference", width: 260, sortable: true, filterable: true, render: (row) => row.evidence_ref, value: (row) => row.evidence_ref },
{ id: "device", header: "Device key", width: 190, sortable: true, filterable: true, render: (row) => row.device_key_id || "Any registered device", value: (row) => row.device_key_id ?? "" },
{ id: "verified", header: "Verified", width: 180, sortable: true, filterable: true, filterType: "date", render: (row) => formatDateTime(row.verified_at), value: (row) => row.verified_at },
{ id: "expires", header: "Expires", width: 180, sortable: true, filterable: true, filterType: "date", render: (row) => formatDateTime(row.expires_at), value: (row) => row.expires_at },
{ id: "state", header: "State", width: 120, sortable: true, filterable: true, render: (row) => <StatusBadge status={row.active ? "active" : "expired"} />, value: (row) => row.active ? "active" : "expired" },
{ id: "actions", header: "Actions", width: 80, sticky: "end", align: "right", render: (row) => <TableActionGroup actions={[{ id: "details", label: "View provenance", icon: <Eye aria-hidden="true" />, onClick: () => setSelectedEvidence(row) }]} /> }
], []);
const decisionColumns = useMemo<DataGridColumn<KeyAccessDecision>[]>(() => [
{ id: "time", header: "Recorded", width: 180, sortable: true, filterable: true, filterType: "date", render: (row) => formatDateTime(row.created_at), value: (row) => row.created_at },
{ id: "purpose", header: "Purpose", width: 220, sortable: true, filterable: true, render: (row) => row.purpose, value: (row) => row.purpose },
{ id: "subject", header: "Subject", width: 230, sortable: true, filterable: true, render: (row) => `${humanize(row.subject_kind)}: ${row.subject_id}`, value: (row) => `${row.subject_kind}:${row.subject_id}` },
{ id: "device", header: "Device key", width: 180, sortable: true, filterable: true, render: (row) => row.device_key_id, value: (row) => row.device_key_id },
{ id: "decision", header: "Decision", width: 120, sortable: true, filterable: true, render: (row) => <StatusBadge status={row.allowed ? "allowed" : "denied"} />, value: (row) => row.allowed ? "allowed" : "denied" },
{ id: "reason", header: "Reason", width: 320, render: (row) => row.reason, value: (row) => row.reason },
{ id: "actions", header: "Actions", width: 80, sticky: "end", align: "right", render: (row) => <TableActionGroup actions={[{ id: "details", label: "View decision provenance", icon: <Eye aria-hidden="true" />, onClick: () => setSelectedDecision(row) }]} /> }
], []);
const epochColumns = useMemo<DataGridColumn<KeyEpoch>[]>(() => [
{ id: "epoch", header: "Epoch", width: 90, sortable: true, filterable: true, filterType: "integer", render: (row) => row.epoch, value: (row) => row.epoch },
{ id: "state", header: "State", width: 120, sortable: true, filterable: true, render: (row) => <StatusBadge status={row.state} />, value: (row) => row.state },
{ id: "history", header: "History access", width: 170, sortable: true, filterable: true, render: (row) => humanize(row.history_policy), value: (row) => row.history_policy },
{ id: "effective", header: "Effective", width: 180, sortable: true, filterable: true, filterType: "date", render: (row) => formatDateTime(row.effective_at), value: (row) => row.effective_at },
{ id: "access", header: "Access decision", width: 260, render: (row) => row.access_decision_ref || "-", value: (row) => row.access_decision_ref ?? "" },
{ id: "reason", header: "Reason", width: 320, render: (row) => row.reason || "-", value: (row) => row.reason ?? "" }
], []);
async function applyRevoke() {
if (!revoking || !revocationReason.trim() || busy) return;
setBusy(true);
setError("");
try {
await revokeDeviceKey(settings, revoking.key_id, revoking.epoch, revocationReason.trim());
setRevoking(null);
setSuccess("The device key was revoked. Existing plaintext or exported keys cannot be recalled.");
await loadAccount();
} catch (caught) {
setError(errorMessage(caught));
} finally {
setBusy(false);
}
}
async function loadEpochHistory() {
if (!epochDraft.subjectId.trim()) return;
setBusy(true);
setError("");
try {
setEpochs(await listEpochs(settings, epochDraft.subjectKind, epochDraft.subjectId.trim()));
} catch (caught) {
setError(errorMessage(caught));
setEpochs([]);
} finally {
setBusy(false);
}
}
async function applyEpochRotation() {
if (!epochDraft.subjectId.trim() || !epochDraft.reason.trim() || !epochDraft.accessDecisionRef.trim() || busy) return;
setBusy(true);
setError("");
try {
const current = epochs.find((epoch) => epoch.state === "active");
await rotateEpoch(settings, {
subject_kind: epochDraft.subjectKind,
subject_id: epochDraft.subjectId.trim(),
reason: epochDraft.reason.trim(),
access_decision_ref: epochDraft.accessDecisionRef.trim(),
idempotency_key: crypto.randomUUID(),
history_policy: "all_retained",
previous_epoch: current?.epoch ?? null
});
setSuccess("The key epoch was rotated. Existing device copies and previously obtained plaintext cannot be revoked retroactively.");
setEpochDraft((currentDraft) => ({ ...currentDraft, reason: "", accessDecisionRef: "" }));
await loadEpochHistory();
} catch (caught) {
setError(errorMessage(caught));
} finally {
setBusy(false);
}
}
const content = <>
{error && <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert>}
{success && <DismissibleAlert tone="success" resetKey={success}>{success}</DismissibleAlert>}
{administrative && <Card title="Account" compact>
<div className="identity-trust-account-selector">
<FormField label="Account">
<ReferenceSelect
value={accountId}
provider={accountProvider}
onChange={(value) => setAccountId(value)}
createCustomOption={(value) => value.trim() ? { value: value.trim(), label: value.trim(), description: "Explicit account reference" } : null}
placeholder="Select or enter an account"
searchPlaceholder="Search accounts" />
</FormField>
<Button onClick={() => void loadAccount()} disabled={loading || busy}><RefreshCw aria-hidden="true" /> Reload</Button>
</div>
</Card>}
<LoadingFrame loading={loading} label="Loading identity trust state">
<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"} />
</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>
<div className="admin-table-surface"><DataGrid id={`identity-trust-device-keys-${administrative ? "admin" : "self"}`} rows={visibleKeys} columns={keyColumns} initialFit="container" getRowKey={(row) => row.key_id} emptyText="No device keys found." /></div>
</Card>
<Card title="Assurance evidence">
<p className="muted small-note">Evidence is bounded by provider, assurance level, device, verification time, and expiry. It does not grant resource access on its own.</p>
<div className="admin-table-surface"><DataGrid id={`identity-trust-assurance-${administrative ? "admin" : "self"}`} rows={evidence} columns={evidenceColumns} initialFit="container" getRowKey={(row) => row.id} emptyText="No assurance evidence found." /></div>
</Card>
{administrative && <>
<Card title="Key epoch administration">
<p className="muted small-note">Rotation supersedes the active epoch and retains history for newly authorized incumbents. It does not grant Access permission, recall exported material, or transfer private keys.</p>
<div className="identity-trust-epoch-form">
<FormField label="Subject type"><select value={epochDraft.subjectKind} disabled={busy} onChange={(event) => { setEpochDraft({ ...epochDraft, subjectKind: event.target.value as EpochDraft["subjectKind"] }); setEpochs([]); }}><option value="identity">Identity</option><option value="account">Account</option><option value="function">Function</option><option value="postbox">Postbox</option><option value="external_recipient">External recipient</option></select></FormField>
<FormField label="Subject reference"><input value={epochDraft.subjectId} disabled={busy} onChange={(event) => { setEpochDraft({ ...epochDraft, subjectId: event.target.value }); setEpochs([]); }} /></FormField>
<Button onClick={() => void loadEpochHistory()} disabled={busy || !epochDraft.subjectId.trim()}><RefreshCw aria-hidden="true" /> Load history</Button>
<FormField label="History access"><input value="All retained history" disabled /></FormField>
<FormField label="Authorizing Access decision"><input value={epochDraft.accessDecisionRef} disabled={busy} onChange={(event) => setEpochDraft({ ...epochDraft, accessDecisionRef: event.target.value })} /></FormField>
<FormField label="Rotation reason"><input value={epochDraft.reason} disabled={busy} onChange={(event) => setEpochDraft({ ...epochDraft, reason: event.target.value })} /></FormField>
<Button variant="danger" onClick={() => void applyEpochRotation()} disabled={busy || !epochDraft.subjectId.trim() || !epochDraft.accessDecisionRef.trim() || !epochDraft.reason.trim()}><RotateCw aria-hidden="true" /> Rotate epoch</Button>
</div>
<div className="admin-table-surface"><DataGrid id="identity-trust-epochs-admin" rows={epochs} columns={epochColumns} initialFit="container" getRowKey={(row) => `${row.subject_kind}:${row.subject_id}:${row.epoch}`} emptyText="Load a subject to inspect its epoch history." /></div>
</Card>
<Card title="Key-access decisions">
<p className="muted small-note">These immutable decisions combine an upstream Access decision with the acting account, current public device key, active epoch, purpose, and resource reference. No cryptographic material is returned by Identity Trust.</p>
<div className="admin-table-surface"><DataGrid id="identity-trust-decisions-admin" rows={decisions} columns={decisionColumns} initialFit="container" getRowKey={(row) => row.id} emptyText="No key-access decisions found for this account." /></div>
</Card>
</>}
</LoadingFrame>
<Dialog open={Boolean(revoking)} title="Revoke device key" helpContextId="identity_trust.settings.devices" helpModuleId="identity_trust" onClose={() => !busy && setRevoking(null)} closeDisabled={busy} footer={<><Button onClick={() => setRevoking(null)} disabled={busy}>Cancel</Button><Button variant="danger" helpContextId="identity_trust.settings.devices" helpModuleId="identity_trust" onClick={() => void applyRevoke()} disabled={busy || !revocationReason.trim()}>Revoke key</Button></>}>
<p>Revoke <strong>{revoking?.key_id}</strong>? Future key-access decisions will reject this device. Plaintext, exports, and keys already obtained by the device cannot be recalled.</p>
<FormField label="Reason"><textarea rows={4} value={revocationReason} disabled={busy} onChange={(event) => setRevocationReason(event.target.value)} /></FormField>
</Dialog>
<ProvenanceDialog title="Assurance evidence provenance" value={selectedEvidence} onClose={() => setSelectedEvidence(null)} />
<ProvenanceDialog title="Key-access decision provenance" value={selectedDecision} onClose={() => setSelectedDecision(null)} />
</>;
if (administrative) {
return <AdminPageLayout title="Identity trust" description="Inspect public device trust, assurance provenance, epoch history, and immutable key-access decisions." loading={false} error="" success="" actions={<Button onClick={() => void loadAccount()} disabled={loading || busy}><RefreshCw aria-hidden="true" /> Reload</Button>}>{content}</AdminPageLayout>;
}
return <div className="identity-trust-panel">{content}</div>;
}
function ProvenanceDialog({ title, value, onClose }: { title: string; value: AssuranceEvidence | KeyAccessDecision | null; onClose: () => void }) {
return <Dialog open={Boolean(value)} title={title} onClose={onClose} footer={<Button onClick={onClose}>Close</Button>}>
{value && <div className="identity-trust-provenance">
<dl>
{"evidence_ref" in value && <><dt>Evidence reference</dt><dd>{value.evidence_ref}</dd><dt>Provider</dt><dd>{value.provider_id}</dd></>}
{"decision_ref" in value && <><dt>Decision reference</dt><dd>{value.decision_ref}</dd><dt>Upstream Access decision</dt><dd>{value.access_decision_ref}</dd><dt>Resource</dt><dd>{value.resource_ref || "Not bound"}</dd></>}
</dl>
<pre>{JSON.stringify(value.provenance, null, 2)}</pre>
</div>}
</Dialog>;
}
function assuranceRank(value: string): number {
return ({ none: 0, software: 1, mfa: 2, hardware: 3, high: 4 } as Record<string, number>)[value.toLowerCase()] ?? 0;
}
function humanize(value: string): string {
return value.replace(/_/g, " ").replace(/\b\w/g, (letter) => letter.toUpperCase());
}
function formatDateTime(value?: string | null): string {
if (!value) return "-";
const parsed = new Date(value);
return Number.isNaN(parsed.getTime()) ? value : parsed.toLocaleString();
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
+19
View File
@@ -0,0 +1,19 @@
import type { PlatformTranslations } from "@govoplan/core-webui";
const en = {
"i18n:govoplan-identity-trust.identity_trust": "Identity trust",
"i18n:govoplan-identity-trust.device_trust": "Device trust",
"i18n:govoplan-identity-trust.identity_trust_administration": "Identity trust administration",
"i18n:govoplan-identity-trust.key_epochs": "Key epochs",
"i18n:govoplan-identity-trust.key_access_decisions": "Key-access decisions"
} as const;
const de: Record<keyof typeof en, string> = {
"i18n:govoplan-identity-trust.identity_trust": "Identitaetsvertrauen",
"i18n:govoplan-identity-trust.device_trust": "Geraetevertrauen",
"i18n:govoplan-identity-trust.identity_trust_administration": "Administration des Identitaetsvertrauens",
"i18n:govoplan-identity-trust.key_epochs": "Schluesselepochen",
"i18n:govoplan-identity-trust.key_access_decisions": "Schluesselzugriffsentscheidungen"
};
export const generatedTranslations: PlatformTranslations = { en, de };
+2
View File
@@ -0,0 +1,2 @@
export { default, identityTrustModule } from "./module";
export * from "./api/identityTrust";
+61
View File
@@ -0,0 +1,61 @@
import { createElement, lazy } from "react";
import type {
AdminSectionsUiCapability,
PlatformWebModule,
SettingsSectionsUiCapability
} from "@govoplan/core-webui";
import { generatedTranslations } from "./i18n/generatedTranslations";
import "./styles/identity-trust.css";
const IdentityTrustPanel = lazy(() => import("./features/IdentityTrustPanel"));
const settingsSections: SettingsSectionsUiCapability = {
sections: [
{
id: "identity-trust",
surfaceId: "identity_trust.settings.devices",
label: "i18n:govoplan-identity-trust.device_trust",
group: "account",
order: 45,
anyOf: ["identity_trust:device:read", "identity_trust:assurance:read"],
render: ({ settings, auth }) => createElement(IdentityTrustPanel, { settings, auth })
}
]
};
const adminSections: AdminSectionsUiCapability = {
sections: [
{
id: "tenant-identity-trust",
moduleId: "identity_trust",
kind: "management",
surfaceId: "identity_trust.admin.trust",
label: "i18n:govoplan-identity-trust.identity_trust",
group: "TENANT",
order: 75,
anyOf: ["identity_trust:device:admin"],
render: ({ settings, auth }) => createElement(IdentityTrustPanel, { settings, auth, administrative: true })
}
]
};
export const identityTrustModule: PlatformWebModule = {
id: "identity_trust",
label: "i18n:govoplan-identity-trust.identity_trust",
version: "0.1.14",
dependencies: [],
optionalDependencies: ["access", "audit", "policy", "encryption", "postbox"],
translations: generatedTranslations,
viewSurfaces: [
{ id: "identity_trust.settings.devices", moduleId: "identity_trust", kind: "section", label: "i18n:govoplan-identity-trust.device_trust", order: 10 },
{ id: "identity_trust.admin.trust", moduleId: "identity_trust", kind: "section", label: "i18n:govoplan-identity-trust.identity_trust_administration", order: 20 },
{ id: "identity_trust.admin.epochs", moduleId: "identity_trust", kind: "section", label: "i18n:govoplan-identity-trust.key_epochs", parentId: "identity_trust.admin.trust", order: 30 },
{ id: "identity_trust.admin.decisions", moduleId: "identity_trust", kind: "section", label: "i18n:govoplan-identity-trust.key_access_decisions", parentId: "identity_trust.admin.trust", order: 40 }
],
uiCapabilities: {
"settings.sections": settingsSections,
"admin.sections": adminSections
}
};
export default identityTrustModule;
+64
View File
@@ -0,0 +1,64 @@
.identity-trust-panel,
.identity-trust-panel > .loading-frame,
.identity-trust-panel .loading-frame-content {
min-width: 0;
}
.identity-trust-panel {
display: flex;
flex-direction: column;
gap: 16px;
}
.identity-trust-account-selector {
display: grid;
grid-template-columns: minmax(280px, 1fr) auto;
gap: 12px;
align-items: end;
}
.identity-trust-epoch-form {
display: grid;
grid-template-columns: repeat(3, minmax(180px, 1fr));
gap: 12px;
align-items: end;
margin-bottom: 14px;
}
.identity-trust-panel .admin-table-surface {
max-height: 380px;
overflow: auto;
}
.identity-trust-provenance dl {
display: grid;
grid-template-columns: minmax(140px, auto) minmax(0, 1fr);
gap: 8px 14px;
margin: 0 0 14px;
}
.identity-trust-provenance dt {
color: var(--text-muted);
}
.identity-trust-provenance dd {
min-width: 0;
margin: 0;
overflow-wrap: anywhere;
}
.identity-trust-provenance pre {
max-height: 280px;
overflow: auto;
padding: 12px;
border: 1px solid var(--line-subtle);
background: var(--surface-muted);
font-size: 0.82rem;
}
@media (max-width: 900px) {
.identity-trust-account-selector,
.identity-trust-epoch-form {
grid-template-columns: 1fr;
}
}
@@ -0,0 +1,23 @@
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
const moduleSource = readFileSync("src/module.ts", "utf8");
const panel = readFileSync("src/features/IdentityTrustPanel.tsx", "utf8");
const api = readFileSync("src/api/identityTrust.ts", "utf8");
assert.match(moduleSource, /"settings.sections": settingsSections/);
assert.match(moduleSource, /"admin.sections": adminSections/);
assert.match(moduleSource, /identity_trust\.settings\.devices/);
assert.match(moduleSource, /identity_trust\.admin\.epochs/);
assert.match(panel, /<ReferenceSelect/);
assert.match(panel, /revoking\.epoch/);
assert.match(panel, /disabled: !canRevokeDevice/);
assert.match(panel, /identity_trust:device:write/);
assert.match(panel, /cannot be recalled/);
assert.match(panel, /listKeyAccessDecisions/);
assert.match(panel, /rotateEpoch/);
assert.doesNotMatch(panel, /public_jwk/);
assert.match(api, /expected_epoch: expectedEpoch/);
assert.match(api, /identity-trust\/account-options/);
console.log("Identity Trust user and administration UI structural contract passed.");