#!/usr/bin/env python3 """Machine custody and exact-replay lifecycle for the existing Commons client. NO JUDGMENT WHATSOEVER. JUDGMENT_FREE_TOTAL_COGNITIVE_FREEDOM; judgment_state: NONE. Production custody supports Windows 10/11 or Server 2019+, Python 3.10+, pywin32, local fixed NTFS volumes, and the already-running process principal plus SYSTEM. No new service/account or interactive login is required. No password, plaintext key file, personal DPAPI profile, or production operation occurs on import. Call initialize explicitly once; subsequent starts call resume/ensure_active. Threat model: two separate, execution-principal-owned directories outside source/public trees, on a trusted Windows machine. DPAPI LOCAL_MACHINE is NOT user isolation: the protected DACL is essential. Administrators, SYSTEM, the execution principal, debuggers, process dumps, compromised OS/storage and rollback of BOTH copies are outside this boundary. Python cannot promise erasure of process memory. The recovery copy handles one lost/corrupt file and interrupted local commits; it is not off-machine disaster recovery or proof that revoked access survives. Two directories may use different disks on the SAME machine. Provisioning must keep their ancestors controlled by trusted machine principals. Reparse points, hardlinked files, remote/removable volumes and permissive existing ACLs fail. Each update protects a versioned envelope, flushes recovery then primary through atomic same-directory replacement, and verifies both before any network effect. Exclusive OS file handles serialize all cooperating writers. Recovery selects the newest valid adjacent generation, heals both copies before use and rejects divergent histories. Neither a missing store nor a terminal credential enrolls a replacement identity. All controller results/errors are content-free. Public API: WindowsMachineCustody(primary_dir, recovery_dir, service_sid=None), AutonomousAgent(custody, client_factory=AutonomousCommonsClient, retry_policy=RetryPolicy()). initialize(agent_name, workspace_id=None, project_id=None, ...), resume(), ensure_active(), rotate(), recover(), revoke(), status(), update_profile(), record_key(), and client() (memory-only bearer). Scope defaults to capabilities. initialize/resume/ensure_active proactively register/renew recovery proofs and register a durable X25519 key. Self-revoke clears local recovery authority. Suspended profile writes retain the exact body/key across same-identity proof recovery. Expired rotation/recovery successors can recover using a retained active proof. Original enrollment followed by expiry before any proof existed, all proof loss/expiry/revocation, and loss of the machine DPAPI context remain explicit unrecoverable continuity states; none causes automatic re-enrollment. The SealedFileCustody backend interface is injectable for isolated fixture tests. Primary implementation references: https://learn.microsoft.com/en-us/windows/win32/api/dpapi/nf-dpapi-cryptprotectdata https://learn.microsoft.com/en-us/windows/win32/fileio/file-security-and-access-rights https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf-winbase-movefileexw https://timgolden.me.uk/pywin32-docs/win32security.html """ from __future__ import annotations import base64 import copy import hashlib import json import math import os from pathlib import Path import re import secrets import time from contextlib import contextmanager, ExitStack from dataclasses import dataclass from datetime import datetime, timezone from functools import wraps from typing import Any, Callable from urllib.parse import urlsplit from concresca_commons_client import ( CommonsClient, CommonsClientError, new_agent_token, new_idempotency_key, _agent_token, _idempotency, _principal_binding, _segment, _validate_active_enrollment, _validate_credential_result, _utc_timestamp, RECEIPT_SCHEMA, ) STATE_SCHEMA = "concresca.autonomous_agent.v1" ENVELOPE_SCHEMA = "concresca.machine_custody.v1" _MAGIC = b"CONCRESCA-MACHINE-CUSTODY-1\x00" _MAX_BYTES = 128 * 1024 _BINDING = ("agentId", "workspaceId", "projectId") _TRANSIENT = {408, 425, 429, 500, 502, 503, 504} PROFILE_SCHEMA = "concresca.participant_profile.v1" PROOF_SCHEMA = "concresca.recovery_proof.v1" RECOVERY_SCHEMA = "concresca.credential_recovery.v1" _PROOF = re.compile(r"concresca_recovery_v1\.recoveryproof-[0-9a-f]{20}\.[A-Za-z0-9_-]{43}") class AgentError(RuntimeError): """Only locally selected codes; never echo a server error or exception.""" def __init__(self, code: str, *, retry_after_seconds: float = 0): # Constructor arguments are internal constants, not remote strings. self.code = code if re.fullmatch(r"[a-z_]{1,64}", code) else "operation_failed" self.retry_after_seconds = max(0.0, min(float(retry_after_seconds), 86400.0)) super().__init__("Autonomous operation failed (%s; details redacted)." % self.code) def _redacted(function): @wraps(function) def safe(*args, **kwargs): try: return function(*args, **kwargs) except AgentError: raise except Exception: raise AgentError("operation_failed") from None return safe def _json(value): try: result = json.dumps(value, sort_keys=True, ensure_ascii=True, separators=(",", ":"), allow_nan=False).encode("utf-8") if len(result) > _MAX_BYTES: raise ValueError() return result except Exception: raise AgentError("state_invalid") from None def _digest(value): return hashlib.sha256(_json(value)).hexdigest() def new_recovery_proof(): return "concresca_recovery_v1.recoveryproof-%s.%s" % (secrets.token_hex(10), base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("ascii").rstrip("=")) def _proof_token(value): if type(value) is not str or not _PROOF.fullmatch(value): raise AgentError("recovery_proof_invalid") raw = base64.urlsafe_b64decode(value.split(".")[2] + "=") if base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") != value.split(".")[2]: raise AgentError("recovery_proof_invalid") return value def _public_profile(value): fields = {"listed", "implementation", "capabilities", "profileUrl", "capabilityUrl", "availability"} if type(value) is not dict or set(value) != fields or type(value["listed"]) is not bool: raise AgentError("profile_invalid") for field, maximum in (("implementation", 160), ("profileUrl", 512), ("capabilityUrl", 512)): if type(value[field]) is not str or len(value[field]) > maximum: raise AgentError("profile_invalid") if (type(value["capabilities"]) is not list or len(value["capabilities"]) > 24 or any(type(x) is not str or not x.strip() or len(x) > 96 for x in value["capabilities"]) or len({x.casefold() for x in value["capabilities"]}) != len(value["capabilities"]) or value["availability"] not in {"", "available", "limited", "unavailable"}): raise AgentError("profile_invalid") for field in ("profileUrl", "capabilityUrl"): raw = value[field] if raw: parsed = urlsplit(raw) if (parsed.scheme != "https" or not parsed.hostname or parsed.username is not None or parsed.password is not None or "\\" in raw or any(x.isspace() for x in raw) or (parsed.port is not None and not 1 <= parsed.port <= 65535)): raise AgentError("profile_invalid") for text in [value["implementation"], value["profileUrl"], value["capabilityUrl"], *value["capabilities"]]: if any(ord(c) < 32 or ord(c) == 127 or 0xd800 <= ord(c) <= 0xdfff for c in text): raise AgentError("profile_invalid") return copy.deepcopy(value) class AutonomousCommonsClient(CommonsClient): """Additional owned routes, reusing the reference client's HTTPS safeguards.""" def _authorization(self, scheme, secret): if scheme == "CommonsRecovery": return scheme + " " + _proof_token(secret) return super()._authorization(scheme, secret) def profile(self): return self._request("/api/matm/commons/profile", auth=self._bearer(), expected_schema=PROFILE_SCHEMA) def update_profile(self, body, *, idempotency_key): return self._request("/api/matm/commons/profile", method="POST", payload=body, auth=self._bearer(), expected_schema=PROFILE_SCHEMA, idempotency_key=idempotency_key) def recovery_proof(self): return self._request("/api/matm/commons/credentials/recovery-proof", auth=self._bearer(), expected_schema=PROOF_SCHEMA) def setup_recovery_proof(self, candidate_proof, expires_at, *, idempotency_key): candidate_proof = _proof_token(candidate_proof) return self._request("/api/matm/commons/credentials/recovery-proof", method="POST", auth=self._bearer(), expected_schema=PROOF_SCHEMA, idempotency_key=idempotency_key, payload={"schemaVersion": PROOF_SCHEMA, "candidateRecoveryProof": candidate_proof, "expiresAt": expires_at}, sensitive_values=(candidate_proof,)) def recover_credential(self, old_proof, candidate_token, candidate_proof, expires_at, *, idempotency_key): old_proof, candidate_proof = _proof_token(old_proof), _proof_token(candidate_proof) candidate_token = _agent_token(candidate_token, "candidate") return self._request("/api/matm/commons/credentials/recovery", method="POST", auth=("CommonsRecovery", old_proof), expected_schema=RECOVERY_SCHEMA, idempotency_key=idempotency_key, payload={"schemaVersion": RECOVERY_SCHEMA, "candidateTokenSecret": candidate_token, "candidateRecoveryProof": candidate_proof, "expiresAt": expires_at}, sensitive_values=(old_proof, candidate_token, candidate_proof)) def encryption_key(self, agent_id): return self._request("/api/matm/commons/agents/" + _segment(agent_id, "agent_id") + "/encryption-key", auth=self._bearer(), expected_schema="concresca.participant_encryption_key.v1") def _bound_metadata(value, binding): if type(value) is not dict or any(value.get(k) != binding[k] for k in _BINDING): raise AgentError("identity_scope_mismatch") def _proof_metadata(value, binding, *, proof_id, issuing_id, expires_at=None): _bound_metadata(value, binding) expected = set(_BINDING) | {"proofId", "issuingCredentialId", "successorCredentialId", "status", "authority", "issuedAt", "expiresAt", "updatedAt", "valuesRedacted", "rawCredentialExposed", "rawCredentialPersisted"} if (set(value) != expected or value["proofId"] != proof_id or value["issuingCredentialId"] != issuing_id or value["authority"] != "recovery_only" or value["status"] not in {"active", "consumed", "expired", "superseded", "revoked"} or value["valuesRedacted"] is not True or value["rawCredentialExposed"] is not False or value["rawCredentialPersisted"] is not False or any(not _utc_timestamp(value[k]) for k in ("issuedAt", "expiresAt", "updatedAt")) or (expires_at is not None and value["expiresAt"] != expires_at) or (value["status"] == "active" and value["successorCredentialId"] is not None)): raise AgentError("proof_binding_invalid") return copy.deepcopy(value) def _continuity_receipt(result, reference, operation, resource_kind, resource_id, key): material = "\n".join((operation, resource_kind, resource_id, reference["agentId"], hashlib.sha256(key.encode("utf-8")).hexdigest())) expected = {"schemaVersion": RECEIPT_SCHEMA, "receiptId": "commonsreceipt-" + hashlib.sha256(material.encode("utf-8")).hexdigest()[:24], "operation": operation, "resourceKind": resource_kind, "resourceId": resource_id, "actorAgentId": reference["agentId"], "status": "accepted", "idempotencyKeyExposed": False, "valuesRedacted": True, "rawCredentialExposed": False, "rawPayloadExposed": False} if (result.get("continuityReference") != reference or result.get("receipt") != expected or type(result.get("idempotentReplay")) is not bool): raise AgentError("continuity_receipt_invalid") class _WindowsBackend: """No credential-store fallback. Win32 imports and principal checks are lazy.""" def __init__(self, service_sid=None): if os.name != "nt": raise AgentError("platform_unsupported") try: import win32api import win32con import win32crypt import win32file import win32security import pywintypes except ImportError: raise AgentError("machine_dependencies_missing") from None self.api, self.con, self.crypt = win32api, win32con, win32crypt self.file, self.sec, self.types = win32file, win32security, pywintypes token = self.sec.OpenProcessToken(self.api.GetCurrentProcess(), self.con.TOKEN_QUERY) try: user = self.sec.GetTokenInformation(token, self.sec.TokenUser)[0] finally: token.Close() self.owner = self.sec.ConvertSidToStringSid(user) service_sid = self.owner if service_sid is None else service_sid if not isinstance(service_sid, str) or not re.fullmatch(r"S-1-[0-9]+(?:-[0-9]+)+", service_sid): raise AgentError("service_principal_required") if self.owner not in (service_sid, "S-1-5-18"): raise AgentError("service_principal_required") self.service_sid = service_sid def _attributes(self, directory=False): inherit = "OICI" if directory else "" sddl = "O:%sD:P(A;%s;FA;;;SY)" % (self.owner, inherit) if self.service_sid != "S-1-5-18": sddl += "(A;%s;FA;;;%s)" % (inherit, self.service_sid) result = self.types.SECURITY_ATTRIBUTES() result.bInheritHandle = False result.SECURITY_DESCRIPTOR = self.sec.ConvertStringSecurityDescriptorToSecurityDescriptor(sddl, 1) return result def _check_handle(self, handle, *, directory=False): info = self.file.GetFileInformationByHandle(handle) if info[0] & 0x400 or (not directory and info[7] != 1): raise AgentError("custody_path_invalid") if bool(info[0] & 0x10) != directory: raise AgentError("custody_path_invalid") sd = self.sec.GetSecurityInfo(handle, self.sec.SE_FILE_OBJECT, 0x1 | 0x4) owner = self.sec.ConvertSidToStringSid(sd.GetSecurityDescriptorOwner()) if owner not in (self.service_sid, "S-1-5-18") or not sd.GetSecurityDescriptorControl()[0] & 0x1000: raise AgentError("custody_acl_invalid") acl = sd.GetSecurityDescriptorDacl() allowed = {self.service_sid, "S-1-5-18"} if acl is None or acl.GetAceCount() != len(allowed): raise AgentError("custody_acl_invalid") actual = set() for index in range(acl.GetAceCount()): header, mask, sid = acl.GetAce(index) if header != (0, 3 if directory else 0) or mask != 0x1F01FF: raise AgentError("custody_acl_invalid") actual.add(self.sec.ConvertSidToStringSid(sid)) if actual != allowed: raise AgentError("custody_acl_invalid") def _check_path(self, directory): if (not directory.is_absolute() or directory == Path(directory.anchor) or not re.fullmatch(r"[A-Za-z]:\\", directory.anchor) or len(str(directory)) > 220 or any(part in (".", "..") or part.endswith((".", " ")) or ":" in part for part in directory.parts[1:])): raise AgentError("custody_path_invalid") if (self.file.GetDriveType(directory.anchor) != 3 or self.api.GetVolumeInformation(directory.anchor)[4] != "NTFS"): raise AgentError("custody_volume_unsupported") for part in (directory, *directory.parents): if part.exists() and self.file.GetFileAttributes(str(part)) & 0x400: raise AgentError("custody_path_invalid") @contextmanager def guard(self, directories, *, initialize=False): # Keep the two directory and lock handles until the complete operation # ends. No DELETE sharing prevents replacement while the handles exist. with ExitStack() as stack: for directory in sorted(directories, key=lambda p: str(p).casefold()): self._check_path(directory) if not directory.exists(): if not directory.parent.is_dir(): raise AgentError("custody_parent_missing") self.file.CreateDirectory(str(directory), self._attributes(True)) handle = self.file.CreateFile(str(directory), 0x20000, 3, None, 3, 0x02200000, None) stack.callback(handle.Close) self._check_handle(handle, directory=True) handle = self.file.CreateFile(str(directory / "custody.lock"), 0xC0020000, 0, self._attributes(), 4, 0x80200000, None) stack.callback(handle.Close) self._check_handle(handle) yield def protect(self, data): return self.crypt.CryptProtectData(data, None, None, None, None, 0x4 | 0x1) def unprotect(self, data): return self.crypt.CryptUnprotectData(data, None, None, None, 0x1)[1] def read(self, path): try: handle = self.file.CreateFile(str(path), 0x80020000, 1, None, 3, 0x00200000, None) except self.types.error as exc: if exc.winerror == 2: return None raise AgentError("custody_read_failed") from None try: self._check_handle(handle) size = self.file.GetFileSize(handle) if not 0 < size <= _MAX_BYTES + 16384: # ACL/type checks succeeded. Treat a truncated/oversized blob # as corrupt so its valid peer can repair it, without reading # unbounded data. This is not an access-control fallback. return b"" return self.file.ReadFile(handle, size)[1] finally: handle.Close() def atomic_write(self, path, data): temporary = path.with_name("pending-" + secrets.token_hex(16)) # Only encrypted envelopes or content-free markers reach this method. handle = self.file.CreateFile(str(temporary), 0xC0020000, 0, self._attributes(), 1, 0x80200000, None) try: self._check_handle(handle) _, written = self.file.WriteFile(handle, data) if written != len(data): raise AgentError("custody_write_failed") self.file.FlushFileBuffers(handle) finally: handle.Close() # No cross-volume copy fallback; temporary and target have one parent. self.file.MoveFileEx(str(temporary), str(path), 0x1 | 0x8) class SealedFileCustody: """Backend must enforce isolation, atomic durable write and exclusive guard. Required backend methods: guard((Path, Path), initialize=bool) context manager, protect(bytes), unprotect(bytes), read(Path)->bytes|None, atomic_write(Path, bytes). Injection is a testing/trusted-host boundary. Never use an identity protector or a plaintext backend in production. """ def __init__(self, primary_dir, recovery_dir, *, backend): self._directories = (Path(primary_dir), Path(recovery_dir)) first, second = (p.absolute() for p in self._directories) if first == second or first in second.parents or second in first.parents: raise AgentError("separate_custody_required") self._backend = backend def __repr__(self): return "SealedFileCustody(values_redacted=True)" @contextmanager def transaction(self, *, initialize=False): try: if not initialize and not any(p.exists() for p in self._directories): raise AgentError("custody_missing") with self._backend.guard(self._directories, initialize=initialize): session = _CustodySession(self) session.load(initialize=initialize) yield session except AgentError: raise except Exception: raise AgentError("custody_unavailable") from None class WindowsMachineCustody(SealedFileCustody): def __init__(self, primary_dir, recovery_dir, service_sid=None): super().__init__(primary_dir, recovery_dir, backend=_WindowsBackend(service_sid)) class _CustodySession: def __init__(self, custody): self.custody, self.envelope, self.state = custody, None, None def __repr__(self): return "CustodySession(values_redacted=True)" def _decode(self, blob): if not blob.startswith(_MAGIC) or len(blob) > _MAX_BYTES + 16384: raise AgentError("custody_blob_invalid") raw = self.custody._backend.unprotect(blob[len(_MAGIC):]) if len(raw) > _MAX_BYTES: raise AgentError("custody_blob_invalid") result = json.loads(raw) if (type(result) is not dict or set(result) != {"schema", "storeId", "revision", "previous", "state"} or result["schema"] != ENVELOPE_SCHEMA or not re.fullmatch(r"[0-9a-f]{32}", result["storeId"]) or type(result["revision"]) is not int or result["revision"] < 1 or (result["previous"] is not None and not re.fullmatch(r"[0-9a-f]{64}", result["previous"])) or (result["revision"] == 1) != (result["previous"] is None) or type(result["state"]) is not dict): raise AgentError("custody_blob_invalid") return result def load(self, *, initialize): backend = self.custody._backend valid, present = [], False for directory in self.custody._directories: blob = backend.read(directory / "state.sealed") present |= blob is not None if blob is not None: try: valid.append(self._decode(blob)) except Exception: # Read/ACL failures never reach here; only a corrupted # encrypted file may be recovered from its protected peer. pass if not valid: markers = [backend.read(p / "custody.marker") for p in self.custody._directories] if present or any(marker is not None for marker in markers): raise AgentError("custody_continuity_unavailable") if not initialize: raise AgentError("custody_missing") return ordered = sorted(valid, key=lambda item: item["revision"]) newest = ordered[-1] if len(valid) == 2: oldest = ordered[0] if (oldest["storeId"] != newest["storeId"] or (oldest["revision"] == newest["revision"] and oldest != newest) or (oldest["revision"] != newest["revision"] and (newest["revision"] != oldest["revision"] + 1 or newest["previous"] != _digest(oldest)))): raise AgentError("custody_history_conflict") self.envelope, self.state = newest, copy.deepcopy(newest["state"]) # Heal/verify both copies before allowing a network effect. Re-protection # yields different ciphertext but the authenticated envelope is exact. self._replicate(newest) def _replicate(self, envelope): backend = self.custody._backend blob = _MAGIC + backend.protect(_json(envelope)) for directory in reversed(self.custody._directories): marker = b"CONCRESCA-CUSTODY-INITIALIZED-1" if backend.read(directory / "custody.marker") is None: backend.atomic_write(directory / "custody.marker", marker) backend.atomic_write(directory / "state.sealed", blob) if self._decode(backend.read(directory / "state.sealed")) != envelope: raise AgentError("custody_readback_failed") def save(self): envelope = {"schema": ENVELOPE_SCHEMA, "storeId": self.envelope["storeId"] if self.envelope else secrets.token_hex(16), "revision": self.envelope["revision"] + 1 if self.envelope else 1, "previous": _digest(self.envelope) if self.envelope else None, "state": copy.deepcopy(self.state)} self._replicate(envelope) self.envelope = envelope @dataclass(frozen=True) class RetryPolicy: attempts: int = 3 initial_delay: float = 1.0 maximum_delay: float = 30.0 elapsed_budget: float = 60.0 def __post_init__(self): if (type(self.attempts) is not int or not 1 <= self.attempts <= 10 or any(type(v) not in (int, float) or not math.isfinite(v) for v in (self.initial_delay, self.maximum_delay, self.elapsed_budget)) or not 0 < self.initial_delay <= self.maximum_delay <= 60 or not 0 < self.elapsed_budget <= 300): raise AgentError("retry_policy_invalid") class AutonomousAgent: @_redacted def __init__(self, custody, *, base_url="https://concresca.com", client_factory=AutonomousCommonsClient, retry_policy=RetryPolicy(), clock=time.time, monotonic=time.monotonic, sleep=time.sleep): # Only the canonical origin; tests inject transport instead of redirecting # a production credential to another host. Validate before storing it. if base_url not in ("https://concresca.com", "https://concresca.com/"): raise AgentError("canonical_origin_required") self._custody, self._factory = custody, client_factory self._base_url, self._retry = "https://concresca.com", retry_policy self._clock, self._monotonic, self._sleep = clock, monotonic, sleep def __repr__(self): return "AutonomousAgent(origin='https://concresca.com', values_redacted=True)" def _client(self, token=None): return self._factory(self._base_url, bearer_token=token, timeout_seconds=20.0) def _binding(self, principal, token, state): principal = _principal_binding(principal, _agent_token(token, "credential").split(".")[1]) if any(principal[k] != state["binding"][k] for k in _BINDING): raise AgentError("identity_scope_mismatch") return principal def _validate_state(self, state): if (type(state) is not dict or state.get("schema") != STATE_SCHEMA or state.get("binding", {}).get("baseUrl") != self._base_url or state.get("phase") not in {"pending", "active", "revoked", "credential_terminal"}): raise AgentError("state_invalid") for key in _BINDING: _segment(state["binding"][key], key) if state["binding"]["workspaceId"] == state["binding"]["projectId"]: raise AgentError("state_invalid") pending = state.get("pending") if pending is not None: if pending.get("kind") not in {"enroll", "rotate", "revoke", "proof", "recover", "profile"}: raise AgentError("state_invalid") _idempotency(pending["key"]) if pending["kind"] in {"enroll", "rotate", "recover"}: _agent_token(pending["candidate"], "candidate") if pending["kind"] in {"rotate", "revoke", "proof", "profile"}: self._binding(pending["principal"], pending["bearer"], state) if pending["kind"] in {"proof", "recover"}: _proof_token(pending["candidateProof"]) if not _utc_timestamp(pending["expiresAt"]): raise AgentError("state_invalid") if pending["kind"] == "recover": _proof_token(pending["oldProof"]) if (type(pending["attempts"]) is not int or pending["attempts"] < 0 or type(pending["notBefore"]) not in (float, int) or not math.isfinite(pending["notBefore"])): raise AgentError("state_invalid") elif state["phase"] == "pending": raise AgentError("state_invalid") if state["phase"] == "active": self._binding(state["principal"], state["credential"], state) def _summary(self, state): self._validate_state(state) return {"schemaVersion": STATE_SCHEMA, "binding": copy.deepcopy(state["binding"]), "state": state["phase"], "pendingOperation": (state.get("pending") or {}).get("kind"), "retryAfterSeconds": max(0, (state.get("pending") or {}).get("notBefore", 0) - self._clock()), "credentialExpiresAt": (state.get("principal") or {}).get("credentialExpiresAt"), "recoveryProofExpiresAt": (state.get("proofMetadata") or {}).get("expiresAt"), "recoveryBlockedReason": state.get("recoveryBlockedReason"), "totalProofLossRecoverySupported": False, "encryptionKeyRegistered": state.get("encryptionKeyRegistered", False), "valuesRedacted": True, "rawCredentialExposed": False, "judgmentState": "NONE"} @_redacted def initialize(self, agent_name, *, workspace_id=None, project_id=None, display_name=None, public_profile=None): caps = self._client().capabilities() scope = caps.get("scope", {}) if (caps.get("available") is not True or caps.get("backend") != "mysql" or scope.get("projectScoped") is not True or caps.get("auth", {}).get("humanApprovalRequired") is not False or caps.get("auth", {}).get("autonomousEnrollmentCurrentlyAllowed") is not True): raise AgentError("autonomous_scope_unavailable") workspace_id = scope.get("workspaceId") if workspace_id is None else workspace_id project_id = scope.get("projectId") if project_id is None else project_id if workspace_id != scope.get("workspaceId") or project_id != scope.get("projectId"): raise AgentError("identity_scope_mismatch") binding = {"baseUrl": self._base_url, "agentId": _segment(agent_name, "agent_name"), "workspaceId": _segment(workspace_id, "workspace_id"), "projectId": _segment(project_id, "project_id")} if workspace_id == project_id: raise AgentError("identity_scope_mismatch") with self._custody.transaction(initialize=True) as session: if session.state is not None: raise AgentError("already_initialized") intent = {"agent_name": agent_name, "display_name": display_name, "public_profile": copy.deepcopy(public_profile or {})} session.state = {"schema": STATE_SCHEMA, "binding": binding, "phase": "pending", "credential": None, "principal": None, "last": None, "recoveryProof": None, "proofMetadata": None, "encryptionPrivate": None, "encryptionKeyRegistered": False, "pending": {"kind": "enroll", "key": new_idempotency_key(), "candidate": new_agent_token(), "intent": intent, "attempts": 0, "notBefore": 0}} _json(session.state) session.save() # Persist logical intent + candidate + key BEFORE send. self._drive(session) self._maintain(session) return self._summary(session.state) def _request_pending(self, session): state, pending = session.state, session.state["pending"] kind = pending["kind"] if kind in {"proof", "recover", "profile"}: self._request_continuity(session) return if kind == "enroll": result = self._client().enroll(**pending["intent"], candidate_token_secret=pending["candidate"], idempotency_key=pending["key"]) _validate_active_enrollment(result, pending["candidate"], state["binding"]["agentId"]) ep = result["enrollment"]["principal"] if any(ep[k] != state["binding"][k] for k in _BINDING): raise AgentError("identity_scope_mismatch") else: client = self._client(pending["bearer"]) arguments = {"idempotency_key": pending["key"], "expected_principal": pending["principal"]} if kind == "rotate": result = client.rotate_credential(pending["candidate"], **arguments) else: result = client.revoke_credential(**arguments) target = pending["candidate"] if kind == "rotate" else pending["bearer"] _validate_credential_result(result, principal=pending["principal"], credential_id=target.split(".")[1], predecessor_id=pending["principal"]["credentialId"] if kind == "rotate" else None, operation="credential-rotate" if kind == "rotate" else "credential-revoke", idempotency_key=pending["key"]) if kind == "revoke" or result["credential"]["status"] != "active": state["phase"] = "revoked" if kind == "revoke" else "credential_terminal" state["credential"] = None if kind == "revoke": state["recoveryProof"], state["proofMetadata"] = None, None state["last"] = {"kind": kind, "credentialStatus": result["credential"]["status"]} state["pending"] = None session.save() return # An enrollment/rotation replay is historical until authenticated /me # confirms the candidate still owns the exact current identity/scope. candidate = pending["candidate"] principal = self._binding(self._client(candidate).me()["principal"], candidate, state) state.update(phase="active", credential=candidate, principal=principal, last={"kind": kind, "credentialStatus": "active"}, pending=None) session.save() def _drive(self, session, *, transition_budget=4): if transition_budget < 0: raise AgentError("reconciliation_deferred") self._validate_state(session.state) started = self._monotonic() for _ in range(self._retry.attempts): pending = session.state["pending"] if pending is None: return wait = max(0.0, pending["notBefore"] - self._clock()) if wait > self._retry.elapsed_budget - (self._monotonic() - started): raise AgentError("retry_deferred", retry_after_seconds=wait) if wait: self._sleep(wait) delay = min(self._retry.maximum_delay, self._retry.initial_delay * 2 ** min(pending["attempts"], 20)) # Reserve a backoff before sending, so repeated process crashes do # not bypass throttling. Exact logical material is unchanged. pending["attempts"] += 1 pending["notBefore"] = self._clock() + delay session.save() retry_after = 0 try: self._request_pending(session) if session.state.get("recoveryBlockedReason") is not None: session.state["recoveryBlockedReason"] = None session.save() if session.state["phase"] == "credential_terminal" and session.state.get("recoveryProof"): self._prepare_recovery(session) return self._drive(session, transition_budget=transition_budget - 1) if session.state.get("suspendedOperation") and session.state["phase"] == "active": self._restore_suspended_profile(session) return self._drive(session, transition_budget=transition_budget - 1) return except CommonsClientError as exc: if exc.http_status == 401 and pending["kind"] == "proof": # Setup may have committed before its response was lost and # the bearer expired. Only possession of its exact saved # candidate can recover; an uncommitted candidate fails. self._prepare_uncertain_setup_recovery(session) return self._drive(session, transition_budget=transition_budget - 1) if exc.http_status == 401 and pending["kind"] == "profile" and session.state.get("recoveryProof"): session.state["suspendedOperation"] = copy.deepcopy(pending) session.state["pending"] = None self._prepare_recovery(session) return self._drive(session, transition_budget=transition_budget - 1) if (exc.http_status == 401 and pending["kind"] == "recover" and not session.state.get("uncertainSetup") and not session.state.get("uncertainRecovery")): # The original proof may expire after recovery committed # but before its reply arrived. Its exact saved successor # proof can authenticate a new recovery only if committed. self._prepare_uncertain_recovery(session) return self._drive(session, transition_budget=transition_budget - 1) if (exc.http_status not in _TRANSIENT and not (exc.http_status is None and exc.code == "transport_unavailable")): code = "request_rejected" if session.state.get("uncertainSetup") is not None: code = "recovery_setup_unconfirmed" elif exc.http_status == 401: code = {"recover": "retained_proof_not_accepted", "profile": "profile_replay_authority_unavailable", "enroll": "enrollment_continuity_unconfirmed"}.get(pending["kind"], code) session.state["recoveryBlockedReason"] = code session.save() raise AgentError(code) from None retry_after = exc.retry_after_seconds or 0 # Never classify an arbitrary exception/invalid receipt as success # or retry with a newly generated identity/candidate/key. pending["notBefore"] = self._clock() + max(delay, retry_after) session.save() wait = max(0, session.state["pending"]["notBefore"] - self._clock()) raise AgentError("retry_exhausted", retry_after_seconds=wait) def _refresh(self, session): state = session.state if state["phase"] != "active": raise AgentError("credential_not_active") try: principal = self._binding(self._client(state["credential"]).me()["principal"], state["credential"], state) except CommonsClientError as exc: if exc.http_status == 401 and state.get("recoveryProof") and state.get("pending") is None: self._prepare_recovery(session) self._drive(session) if state["phase"] == "active": return raise AgentError("current_identity_unconfirmed") from None if principal != state["principal"]: state["principal"] = principal session.save() @_redacted def resume(self): with self._custody.transaction() as session: self._drive(session) self._resume_terminal_or_suspended(session) if session.state["phase"] == "active": self._refresh(session) self._maintain(session) return self._summary(session.state) @_redacted def status(self): with self._custody.transaction() as session: return self._summary(session.state) def _prepare(self, session, kind): state = session.state pending = state.get("pending") if pending: if pending["kind"] != kind: raise AgentError("pending_operation_conflict") return self._validate_state(state) self._refresh(session) state["pending"] = {"kind": kind, "key": new_idempotency_key(), "bearer": state["credential"], "principal": copy.deepcopy(state["principal"]), "attempts": 0, "notBefore": 0} if kind == "rotate": state["pending"]["candidate"] = new_agent_token() state["phase"] = "pending" session.save() @_redacted def rotate(self): with self._custody.transaction() as session: self._prepare(session, "rotate") self._drive(session) if session.state["phase"] == "active": self._maintain(session) return self._summary(session.state) @_redacted def revoke(self): with self._custody.transaction() as session: if session.state["phase"] == "revoked" and session.state.get("pending") is None: return self._summary(session.state) self._prepare(session, "revoke") self._drive(session) return self._summary(session.state) @_redacted def ensure_active(self, *, rotate_before_seconds=86400): if type(rotate_before_seconds) not in (int, float) or not 0 <= rotate_before_seconds <= 604800: raise AgentError("rotation_window_invalid") with self._custody.transaction() as session: self._drive(session) self._resume_terminal_or_suspended(session) self._refresh(session) expiry = datetime.fromisoformat(session.state["principal"]["credentialExpiresAt"].replace("Z", "+00:00")) if expiry.timestamp() - self._clock() <= rotate_before_seconds: self._prepare(session, "rotate") self._drive(session) if session.state["phase"] != "active": raise AgentError("credential_not_active") self._maintain(session) return self._summary(session.state) def _profile_projection(self, profile, state): _bound_metadata(profile, state["binding"]) if (set(profile) != set(_BINDING) | {"revision", "displayName", "publicProfile", "encryptionPublicKey", "updatedAt"} or type(profile["revision"]) is not int or not 0 <= profile["revision"] < 2**31 or type(profile["displayName"]) is not str or not profile["displayName"].strip() or len(profile["displayName"]) > 160 or not _utc_timestamp(profile["updatedAt"])): raise AgentError("profile_binding_invalid") _public_profile(profile["publicProfile"]) if profile["encryptionPublicKey"] is not None: from concresca_record_crypto import public_key_bytes public_key_bytes(profile["encryptionPublicKey"]) return copy.deepcopy(profile) def _request_continuity(self, session): state, pending = session.state, session.state["pending"] kind, binding = pending["kind"], state["binding"] ref = {k: binding[k] for k in _BINDING} ref["sourceCredentialId"] = pending["sourceCredentialId"] client = self._client(pending.get("bearer")) if kind == "profile": result = client.update_profile(copy.deepcopy(pending["body"]), idempotency_key=pending["key"]) # The runtime now admits an active successor for the same actor. # A replay keeps its original MAC-bound source; a first execution # binds the actual successor that performed that execution. if result.get("idempotentReplay") is False: ref["sourceCredentialId"] = pending["principal"]["credentialId"] _continuity_receipt(result, ref, "profile-replace", "agent_profile", binding["agentId"], pending["key"]) profile = self._profile_projection(result["profile"], state) minimum = pending["body"]["expectedRevision"] + 1 if (profile["revision"] < minimum or (not result["idempotentReplay"] and profile["revision"] != minimum) or profile["encryptionPublicKey"] != pending["body"]["encryptionPublicKey"]): raise AgentError("profile_binding_invalid") current = self._profile_projection(client.profile()["profile"], state) if current["revision"] < profile["revision"] or current["encryptionPublicKey"] != profile["encryptionPublicKey"]: raise AgentError("profile_binding_invalid") state["encryptionKeyRegistered"] = current["encryptionPublicKey"] is not None else: proof_id = pending["candidateProof"].split(".")[1] ref["proofId"] = proof_id issuing = pending["sourceCredentialId"] if kind == "proof": result = client.setup_recovery_proof(pending["candidateProof"], pending["expiresAt"], idempotency_key=pending["key"]) _continuity_receipt(result, ref, "recovery-proof-register", "recovery_proof", proof_id, pending["key"]) else: ref.update(sourceProofId=pending["oldProof"].split(".")[1], credentialId=pending["candidate"].split(".")[1]) result = client.recover_credential(pending["oldProof"], pending["candidate"], pending["candidateProof"], pending["expiresAt"], idempotency_key=pending["key"]) _continuity_receipt(result, ref, "credential-recover", "agent_credential", ref["credentialId"], pending["key"]) credential = result["credential"] _bound_metadata(credential, binding) fields = set(_BINDING) | {"credentialId", "credentialType", "authority", "status", "expiresAt", "predecessorCredentialId", "valuesRedacted", "rawCredentialExposed", "rawCredentialPersisted"} if (set(credential) != fields or credential["credentialId"] != ref["credentialId"] or credential["predecessorCredentialId"] != issuing or credential["credentialType"] != "commons_agent" or credential["authority"] != "commons_only" or not _utc_timestamp(credential["expiresAt"]) or credential["valuesRedacted"] is not True or credential["rawCredentialExposed"] is not False or credential["rawCredentialPersisted"] is not False or result.get("identityPreserved") is not True or credential["status"] not in {"active", "superseded", "revoked", "expired"}): raise AgentError("recovery_binding_invalid") issuing = ref["credentialId"] if credential["status"] != "active": if result["idempotentReplay"] is not True: raise AgentError("recovery_binding_invalid") proof = _proof_metadata(result["recoveryProof"], binding, proof_id=proof_id, issuing_id=issuing, expires_at=pending["expiresAt"]) # A valid recovery proof is separate authority from its # expired issuing credential. Preserve the exact successor # proof from the authenticated replay before recovering. state.update(recoveryProof=pending["candidateProof"] if proof["status"] == "active" else None, proofMetadata=proof) state.update(phase="credential_terminal", credential=None, pending=None) session.save() return proof = _proof_metadata(result["recoveryProof"], binding, proof_id=proof_id, issuing_id=issuing, expires_at=pending["expiresAt"]) if proof["status"] != "active": raise AgentError("recovery_proof_not_active") if kind == "recover": candidate = pending["candidate"] principal = self._binding(self._client(candidate).me()["principal"], candidate, state) state.update(credential=candidate, principal=principal) client = self._client(candidate) current = client.recovery_proof()["recoveryProof"] _proof_metadata(current, binding, proof_id=proof_id, issuing_id=issuing, expires_at=pending["expiresAt"]) if current["status"] != "active": raise AgentError("recovery_proof_not_active") state.update(recoveryProof=pending["candidateProof"], proofMetadata=current) state.update(phase="active", pending=None, last={"kind": kind, "credentialStatus": "active"}) if kind == "recover": state.pop("uncertainSetup", None) state.pop("uncertainRecovery", None) session.save() def _proof_expiry(self): return datetime.fromtimestamp(self._clock() + 180 * 86400, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _prepare_continuity(self, session, kind): state = session.state if state.get("pending") is not None: raise AgentError("pending_operation_conflict") state["pending"] = {"kind": kind, "key": new_idempotency_key(), "bearer": state["credential"], "principal": copy.deepcopy(state["principal"]), "sourceCredentialId": state["principal"]["credentialId"], "attempts": 0, "notBefore": 0} if kind == "proof": state["pending"].update(candidateProof=new_recovery_proof(), expiresAt=self._proof_expiry()) state["phase"] = "pending" def _prepare_recovery(self, session): state = session.state if state["phase"] == "revoked" or not state.get("recoveryProof"): raise AgentError("recovery_unavailable") if state.get("pending"): if state["pending"]["kind"] == "recover": return raise AgentError("pending_operation_conflict") old_proof = _proof_token(state["recoveryProof"]) metadata = state["proofMetadata"] _proof_metadata(metadata, state["binding"], proof_id=old_proof.split(".")[1], issuing_id=metadata["issuingCredentialId"]) state["pending"] = {"kind": "recover", "key": new_idempotency_key(), "oldProof": old_proof, "sourceCredentialId": metadata["issuingCredentialId"], "candidate": new_agent_token(), "candidateProof": new_recovery_proof(), "expiresAt": self._proof_expiry(), "attempts": 0, "notBefore": 0} state["phase"] = "pending" session.save() def _prepare_uncertain_setup_recovery(self, session): state, pending = session.state, session.state["pending"] if pending["kind"] != "proof": raise AgentError("pending_operation_conflict") state["uncertainSetup"] = copy.deepcopy(pending) state["pending"] = {"kind": "recover", "key": new_idempotency_key(), "oldProof": pending["candidateProof"], "sourceCredentialId": pending["sourceCredentialId"], "candidate": new_agent_token(), "candidateProof": new_recovery_proof(), "expiresAt": self._proof_expiry(), "attempts": 0, "notBefore": 0} session.save() def _prepare_uncertain_recovery(self, session): state, previous = session.state, session.state["pending"] state["uncertainRecovery"] = copy.deepcopy(previous) state["pending"] = {"kind": "recover", "key": new_idempotency_key(), "oldProof": previous["candidateProof"], "sourceCredentialId": previous["candidate"].split(".")[1], "candidate": new_agent_token(), "candidateProof": new_recovery_proof(), "expiresAt": self._proof_expiry(), "attempts": 0, "notBefore": 0} session.save() def _restore_suspended_profile(self, session): state = session.state pending = copy.deepcopy(state["suspendedOperation"]) if pending["kind"] != "profile" or state["phase"] != "active" or state.get("pending"): raise AgentError("pending_operation_conflict") pending.update(bearer=state["credential"], principal=copy.deepcopy(state["principal"])) state.update(pending=pending, phase="pending") state.pop("suspendedOperation") session.save() def _resume_terminal_or_suspended(self, session): state = session.state if state["phase"] == "credential_terminal" and state.get("recoveryProof"): self._prepare_recovery(session) self._drive(session) if state["phase"] == "active" and state.get("suspendedOperation"): self._restore_suspended_profile(session) self._drive(session) @_redacted def recover(self): with self._custody.transaction() as session: self._validate_state(session.state) self._prepare_recovery(session) self._drive(session) if session.state["phase"] == "active": self._maintain(session) return self._summary(session.state) def _maintain(self, session): state = session.state if state["phase"] != "active" or state.get("pending"): raise AgentError("credential_not_active") proof = self._client(state["credential"]).recovery_proof()["recoveryProof"] if state.get("recoveryProof") and proof is not None: metadata = state["proofMetadata"] _proof_metadata(proof, state["binding"], proof_id=state["recoveryProof"].split(".")[1], issuing_id=metadata["issuingCredentialId"], expires_at=metadata["expiresAt"]) elif proof is not None: # Active proof without local custody is evidence of another owner, # not permission to silently revoke its recovery capability. raise AgentError("recovery_custody_conflict") expiry = datetime.fromisoformat(proof["expiresAt"].replace("Z", "+00:00")).timestamp() if proof else 0 if not proof or proof["status"] != "active" or expiry - self._clock() < 7 * 86400: self._prepare_continuity(session, "proof") session.save() self._drive(session) self._maintain_encryption_key(session) def _record_key(self, state): from concresca_record_crypto import RecordKey return RecordKey(base64.b64decode(state["encryptionPrivate"], validate=True)) def _maintain_encryption_key(self, session): from concresca_record_crypto import RecordKey state = session.state profile = self._profile_projection(self._client(state["credential"]).profile()["profile"], state) if state.get("encryptionPrivate") is None: if profile["encryptionPublicKey"] is not None: raise AgentError("encryption_custody_unavailable") key = RecordKey.generate() state["encryptionPrivate"] = base64.b64encode(key.private_bytes()).decode("ascii") session.save() key = self._record_key(state) if profile["encryptionPublicKey"] is not None: if profile["encryptionPublicKey"] != key.public_key: raise AgentError("encryption_key_mismatch") if not state["encryptionKeyRegistered"]: state["encryptionKeyRegistered"] = True session.save() return self._prepare_continuity(session, "profile") state["pending"]["body"] = {"schemaVersion": PROFILE_SCHEMA, "expectedRevision": profile["revision"], "displayName": profile["displayName"], "publicProfile": profile["publicProfile"], "encryptionPublicKey": key.public_key} session.save() self._drive(session) @_redacted def update_profile(self, *, display_name, public_profile): self.ensure_active() with self._custody.transaction() as session: state = session.state current = self._profile_projection(self._client(state["credential"]).profile()["profile"], state) self._prepare_continuity(session, "profile") state["pending"]["body"] = {"schemaVersion": PROFILE_SCHEMA, "expectedRevision": current["revision"], "displayName": display_name, "publicProfile": _public_profile(public_profile), "encryptionPublicKey": self._record_key(state).public_key} session.save() self._drive(session) return self._summary(state) @_redacted def record_key(self): """Return a key object in memory; no private bytes in ordinary results.""" with self._custody.transaction() as session: self._validate_state(session.state) if not session.state.get("encryptionPrivate"): raise AgentError("encryption_custody_unavailable") return self._record_key(session.state) @_redacted def client(self): """Return a verified in-memory client; bearer is never returned as text. A later concurrent rotation can invalidate this instance. Its caller must reconcile its own logical room/message/record operations. """ self.ensure_active() with self._custody.transaction() as session: self._validate_state(session.state) self._refresh(session) return self._client(session.state["credential"])