# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-handle-credential-rotation-in-the-middle-of-a-delegated-operation.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-handle-credential-rotation-in-the-middle-of-a-delegated-operation.warrant.json # Markdown SHA-256: bb49227161075cfbc718912142e1cb7b917ebacd02f08701522db72a5b6e2218 # Code AST SHA-256: 8a0db76cd019e34b9b8258bb51b1907f3439fb6d0d317f67a07f73f9eaf0d615 # Verified Classes: RotationVerdict, CredentialRotation, RotationContext, RotationManifest, CredentialRotationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class RotationVerdict(Enum): CREDENTIAL_STABLE = auto() ROTATION_DETECTED_RESUMABLE = auto() ROTATION_DETECTED_ABORT = auto() INSUFFICIENT_NEW_CREDENTIAL = auto() @dataclass(frozen=True) class CredentialRotation: original_credential_epoch: str new_credential_epoch: str rotation_epoch: float completed_steps: int pending_steps: int all_completed_steps_idempotent: bool new_credential_has_resumption_rights: bool @dataclass(frozen=True) class RotationContext: rotation_detected: bool @dataclass(frozen=True) class RotationManifest: verdict: RotationVerdict directive: str manifest_digest: str class CredentialRotationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_rotation( cls, rot: CredentialRotation, ctx: RotationContext ) -> RotationManifest: # Gate 1: No rotation if not ctx.rotation_detected: return RotationManifest( verdict=RotationVerdict.CREDENTIAL_STABLE, directive="PROCEED: Credential stable for operation duration.", manifest_digest=cls._hash("STABLE") ) # Gate 2: New credential resumption rights if not rot.new_credential_has_resumption_rights: return RotationManifest( verdict=RotationVerdict.INSUFFICIENT_NEW_CREDENTIAL, directive="HALT: New credential lacks resumption rights. Abort operation and roll back.", manifest_digest=cls._hash(f"NO_RIGHTS:{rot.new_credential_epoch}") ) # Gate 3: Idempotency safety if not rot.all_completed_steps_idempotent: return RotationManifest( verdict=RotationVerdict.ROTATION_DETECTED_ABORT, directive=f"HALT: Non-idempotent completed steps ({rot.completed_steps}). Cannot retry safely. Abort and reconcile.", manifest_digest=cls._hash(f"NON_IDEMPOTENT:{rot.completed_steps}") ) # Gate 4: Resumable digest = cls._hash(json.dumps({ "original": rot.original_credential_epoch, "new": rot.new_credential_epoch, "completed": str(rot.completed_steps), "pending": str(rot.pending_steps) }, sort_keys=True)) return RotationManifest( verdict=RotationVerdict.ROTATION_DETECTED_RESUMABLE, directive=f"PROCEED: Resume operation with new credential epoch {rot.new_credential_epoch}. Completed steps are idempotent.", manifest_digest=digest )