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