Files
govoplan-identity-trust/src/govoplan_identity_trust/backend/service.py
T

744 lines
25 KiB
Python

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",
]