Python Première admission et confiance historique

Python Première admission et confiance historique

Les couches Python AdmissionService ont authentifié la première admission et la confiance historique au-dessus du vérificateur Signed Document en six étapes (MWP-ADM-005). Le flux de contrôle indépendant de la langue est la page d’exécution locale de première admission et de confiance historique, et les exigences exactes sont les clauses locales de première admission et de confiance historique.

Types publics

Public API	Responsabilité
AdmissionCurrentKeyResolver.resolve_current	Renvoie la preuve complète Organization Registry que le déploiement affirme être à jour pour une nouvelle décision de première admission.
TrustedAdmissionContext.issue	Émettez l’ID d’enregistrement de confiance, l’acceptation de confiance instantanée et l’acceptation du service uniquement après une absence faisant autorité.
AdmissionLog.lookup	Renvoie AdmissionLookupStatus.FOUND avec des octets authentifiés ou AdmissionLookupStatus.AUTHORITATIVE_ABSENCE. Échec du transport, échec du cache, état indéterminé et échec d’absence non authentifié fermé sous MWP-ADM-003.
AdmissionLog.append_or_return_existing	Ajoutez atomiquement les octets candidats ou renvoyez le gagnant faisant autorité simultané.
AuthenticatedAdmissionRecord	Bind a renvoyé des octets d’enregistrement à l’identité de service acceptante authentifiée par l’adaptateur pour la validation MWP-ADM-009.
AdmissionService.prepare_first_admission	Créez et validez des octets candidats à partir d’un document déjà vérifié et d’un contexte fiable. Cela n’ajoute ni n’implique une admission.
AdmissionService.admit_first	Implémentez le flux de preuves actuelles, d’absence faisant autorité, d’ajout atomique et d’enregistrement renvoyé dans MWP-ADM-006.
AdmissionService.verify_historical_admission	Réexécutez la vérification avec les preuves historiques Registry et exigez un enregistrement trouvé sans émettre de contexte ni ajouter, comme requis par MWP-ADM-008.

Le résultat de l’adaptateur est une preuve typée. Un booléen de confiance fourni par l’appelant ne constitue pas un résultat de recherche faisant autorité et ne doit jamais sélectionner le chemin de réussite.

Appel de première admission

Informative example

admitted = AdmissionService().admit_first(
    SignedDocumentKind.COMMAND,
    document_bytes,
    current_registry,
    admission_log,
    trusted_context,
)

admit_first vérifie avant d’appeler le journal. Lorsque la recherche renvoie AdmissionLookupStatus.AUTHORITATIVE_ABSENCE, elle appelle prepare_first_admission, appelle append_or_return_existing, puis valide l’enregistrement réellement renvoyé. Un gagnant simultané n’est acceptable que lorsque ses octets, son service authentifié, sa liaison de document et son heure de confiance sont tous validés (MWP-ADM-009, MWP-ADM-010).

Appel en rediffusion historique

Informative example

historical = AdmissionService().verify_historical_admission(
    SignedDocumentKind.COMMAND,
    document_bytes,
    historical_registry,
    admission_log,
)

La relecture historique utilise l’historique Registry conservé, nécessite un enregistrement d’admission trouvé et n’appelle jamais TrustedAdmissionContext.issue ou AdmissionLog.append_or_return_existing. Les ID d’enregistrement correspondants dans l’exemple démontrent que la relecture a récupéré le premier enregistrement faisant autorité ; ils ne prouvent pas par eux-mêmes la fraîcheur de Command, l’autorisation du signataire, l’acceptation de la machine d’état ou la vérification portable de la preuve de journal. La fraîcheur Command et l’autorisation du signataire restent des exigences distinctes sous MWP-ADM-013 et MWP-ADM-014.

Exemple d’exécution exacte épinglée

Informative example

L’exemple utilise uniquement les API publiques SDK et la dépendance d’exécution cryptography déclarée. Il crée de manière déterministe une clé Ed25519, un Organization complet Registry, un Command signé et des adaptateurs en mémoire saisis. La graine privée fixe est un matériel de démonstration et ne doit jamais être utilisée en dehors de l’exemple.

"""Run First Admission and Historical Trust with typed in-memory adapters."""

from __future__ import annotations

import base64
import hashlib
import json
from dataclasses import dataclass
from threading import Lock

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from missionweaveprotocol import (
    AdmissionContextValue,
    AdmissionCurrentKeyResolver,
    AdmissionLog,
    AdmissionLookup,
    AdmissionLookupStatus,
    AdmissionService,
    AuthenticatedAdmissionRecord,
    KeyRegistryCompleteness,
    KeyRegistrySnapshot,
    KeyResolutionRequest,
    KeyResolver,
    PrincipalEvidence,
    SignedDocumentCodec,
    SignedDocumentKind,
    SigningKey,
    TrustedAdmissionContext,
)

AGENT_ID = "urn:missionweaveprotocol:agent:website-example"
KEY_ID = "urn:missionweaveprotocol:key:website-example"
ORGANIZATION_ID = "urn:missionweaveprotocol:organization:website-example"


class DeterministicSigningKey:
    """Demonstration-only Ed25519 signing adapter with a fixed seed."""

    algorithm = "Ed25519"

    def __init__(self, key_id: str) -> None:
        self.key_id = key_id
        self._private_key = Ed25519PrivateKey.from_private_bytes(bytes(range(1, 33)))
        self.public_key_bytes = self._private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )

    def sign(self, message: bytes) -> bytes:
        return self._private_key.sign(message)


@dataclass(frozen=True, slots=True)
class CurrentRegistryFixture:
    """Expose one complete Registry as both current and historical evidence."""

    registry_bytes: bytes

    def _snapshot(self) -> KeyRegistrySnapshot:
        return KeyRegistrySnapshot(
            completeness=KeyRegistryCompleteness.ORGANIZATION_WIDE,
            registry_bytes=self.registry_bytes,
        )

    def resolve_current(self, request: KeyResolutionRequest) -> KeyRegistrySnapshot:
        del request
        return self._snapshot()

    def resolve(self, request: KeyResolutionRequest) -> KeyRegistrySnapshot:
        del request
        return self._snapshot()


@dataclass(frozen=True, slots=True)
class FixedTrustedAdmissionContext:
    """Issue trusted acceptance metadata after authoritative log absence."""

    trusted_accepted_at: str
    accepted_by: PrincipalEvidence

    def issue(self, organization_id: str, signing_hash: str) -> AdmissionContextValue:
        logical_key_digest = hashlib.sha256(
            f"{organization_id}\0{signing_hash}".encode()
        ).hexdigest()
        return AdmissionContextValue(
            admission_record_id=(f"urn:missionweaveprotocol:admission-record:{logical_key_digest}"),
            trusted_accepted_at=self.trusted_accepted_at,
            accepted_by=self.accepted_by,
        )


class InMemoryAdmissionLog:
    """Model authenticated lookup and atomic append-or-return-existing behavior."""

    def __init__(self, authenticated_service: PrincipalEvidence) -> None:
        self._authenticated_service = authenticated_service
        self._records: dict[tuple[str, str], AuthenticatedAdmissionRecord] = {}
        self._lock = Lock()
        self.append_count = 0

    def lookup(self, organization_id: str, signing_hash: str) -> AdmissionLookup:
        with self._lock:
            record = self._records.get((organization_id, signing_hash))
        if record is None:
            return AdmissionLookup(AdmissionLookupStatus.AUTHORITATIVE_ABSENCE)
        return AdmissionLookup(AdmissionLookupStatus.FOUND, record)

    def append_or_return_existing(
        self,
        organization_id: str,
        signing_hash: str,
        candidate_bytes: bytes,
    ) -> AuthenticatedAdmissionRecord:
        key = (organization_id, signing_hash)
        with self._lock:
            existing = self._records.get(key)
            if existing is not None:
                return existing

            candidate = AuthenticatedAdmissionRecord(
                record_bytes=candidate_bytes,
                authenticated_service=self._authenticated_service,
            )
            self._records[key] = candidate
            self.append_count += 1
            return candidate


def main() -> None:
    signing_key: SigningKey = DeterministicSigningKey(KEY_ID)
    registry_bytes = json.dumps(
        {
            "organizationId": ORGANIZATION_ID,
            "bindings": [
                {
                    "keyId": KEY_ID,
                    "principal": {"type": "agent", "id": AGENT_ID},
                    "algorithm": "Ed25519",
                    "publicKey": base64.urlsafe_b64encode(signing_key.public_key_bytes)
                    .rstrip(b"=")
                    .decode("ascii"),
                    "validFrom": "2026-07-01T00:00:00Z",
                    "validityHistory": [],
                }
            ],
        },
        separators=(",", ":"),
    ).encode("utf-8")
    registry = CurrentRegistryFixture(registry_bytes)
    current_registry: AdmissionCurrentKeyResolver = registry
    historical_registry: KeyResolver = registry

    document_bytes = (
        SignedDocumentCodec()
        .sign(
            SignedDocumentKind.COMMAND,
            {
                "protocolVersion": "0.1",
                "actionId": "urn:uuid:11111111-2222-4333-8444-555555555555",
                "actor": {"type": "agent", "id": AGENT_ID},
                "sessionEpoch": 1,
                "membershipEpoch": 1,
                "groupId": "urn:missionweaveprotocol:group:website-example",
                "kind": "message.post",
                "correlationId": "urn:uuid:aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee",
                "issuedAt": "2026-07-15T00:00:00Z",
                "payload": {"text": "Admission example"},
            },
            signing_key,
        )
        .canonical_document_bytes
    )

    admission_service = PrincipalEvidence(
        type="service",
        id="urn:missionweaveprotocol:service:admission",
    )
    trusted_context: TrustedAdmissionContext = FixedTrustedAdmissionContext(
        trusted_accepted_at="2026-07-15T00:05:00Z",
        accepted_by=admission_service,
    )
    admission_log_impl = InMemoryAdmissionLog(admission_service)
    admission_log: AdmissionLog = admission_log_impl
    service = AdmissionService()

    first = service.admit_first(
        SignedDocumentKind.COMMAND,
        document_bytes,
        current_registry,
        admission_log,
        trusted_context,
    )
    historical = service.verify_historical_admission(
        SignedDocumentKind.COMMAND,
        document_bytes,
        historical_registry,
        admission_log,
    )

    assert first.record.admission_record_id == historical.record.admission_record_id
    assert first.verified.signing_hash == historical.verified.signing_hash
    assert first.record.raw_bytes == historical.record.raw_bytes
    assert admission_log_impl.append_count == 1
    print(f"first admission: {first.record.admission_record_id}")
    print(f"historical replay: {historical.record.admission_record_id}")


if __name__ == "__main__":
    main()

Exécutez-le à partir d’un checkout exact après avoir installé tous les extras SDK :

uv sync --all-extras
uv run python /absolute/path/to/examples/sdk/python/admission.py

La sortie attendue contient le même ID First-Admission Record pour first admission et historical replay. Le programme affirme également que les hachages de signature et les octets d’enregistrement correspondent, et que le journal est ajouté exactement une fois.

Implementation note

Remplacez les exemples d’adaptateurs dans un déploiement :

• CurrentRegistryFixture avec un résolveur courant-Registry authentifié ;
• InMemoryAdmissionLog avec un Admission Log atomique authentifié, en ajout uniquement ; et
• FixedTrustedAdmissionContext avec une heure de confiance contrôlée par Organization et émetteur de services acceptant.

Remplacez DeterministicSigningKey par un service de signature ou un adaptateur de clé appartenant au déploiement. N’intégrez jamais de graine de signature privée dans le code de production.

Ne réduisez pas ces limites en un seul indicateur de disponibilité ou de confiance.

Le rejet d’admission post-cryptographique reste protégé par l’étape admission et le code électronique AUTH_INVALID_SIGNATURE (MWP-ADM-012).