226 lines
7.2 KiB
Python
226 lines
7.2 KiB
Python
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 (
|
|
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",
|
|
}
|
|
|
|
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|