"""Thin wrapper around the canonical REST contract (POST /api/v1/authorize). No decision logic lives here.""" import json import uuid import urllib.error import urllib.request AUTHORIZED = "AUTHORIZED" APPROVAL_REQUIRED = "APPROVAL_REQUIRED" DENIED = "DENIED" NEEDS_INFORMATION = "NEEDS_INFORMATION" class QuattrinoError(Exception): def __init__(self, status, code, message, details=None): super().__init__(f"{status} {code}: {message}") self.status, self.code, self.message, self.details = status, code, message, details or {} class AuthorizationResult(dict): """dict with convenience attributes; the full API payload is preserved.""" @property def decision(self): return self.get("decision") @property def authorization_id(self): return self.get("authorization_id") @property def status(self): return self.get("status") @property def effective_maximum_minor(self): return self.get("effective_maximum_minor") @property def binding_layer(self): return (self.get("effective_authority") or {}).get("binding_layer") @property def execution_package(self): return self.get("execution_plan") @property def reasons(self): return (self.get("denial_reasons") or []) + (self.get("needs_information") or []) + \ (self.get("approval_reasons") or []) @property def authorized(self): return self.decision == AUTHORIZED # ---- Phase 30 ---- @property def missing_authority(self): return self.get("missing_authority") or [] @property def question(self): return self.get("question") @property def question_required(self): return bool(self.get("question_required")) @property def subactions(self): return self.get("subactions") or [] @property def authorized_subactions(self): return self.get("authorized_subactions") or [] @property def compliance(self): return self.get("compliance") @property def envelope(self): return self.get("envelope") @property def expires_at(self): return self.get("expires_at") def verify_proof(base, proof, *, audience=None, expected=None, verifier=None, timeout=30): """Credential-less verifier for an external executor / counterparty that only holds the proof token. ``base`` is the API root (".../api/v1") or the site URL. ALWAYS refuse to act when result["valid"] is not True. Sandbox only.""" base = base.rstrip("/") if not base.endswith("/api/v1"): base += "/api/v1" body = {"proof": proof} for k, v in (("audience", audience), ("expected", expected), ("verifier", verifier)): if v is not None: body[k] = v req = urllib.request.Request(base + "/authorizations/verify", data=json.dumps(body).encode(), headers={"Content-Type": "application/json", "User-Agent": "quattrino-python/0.31.0"}, method="POST") try: with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: payload = json.loads(e.read() or b"{}") except ValueError: payload = {} err = payload.get("error") or {} raise QuattrinoError(e.code, err.get("code") or "http_error", err.get("message") or str(e), err.get("details")) class Quattrino: def __init__(self, api_key, base_url="https://quattrino.io", timeout=30): if not api_key: raise ValueError("api_key (agent credential) is required") self.api_key = api_key self.base = base_url.rstrip("/") + "/api/v1" self.timeout = timeout # ---------------------------------------------------------- transport ---- def _request(self, method, path, body=None, headers=None): h = {"Authorization": "Bearer " + self.api_key, "Content-Type": "application/json", "User-Agent": "quattrino-python/0.31.0"} h.update(headers or {}) data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(self.base + path, data=data, headers=h, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as r: return json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: try: payload = json.loads(e.read() or b"{}") except ValueError: payload = {} err = payload.get("error") or {} raise QuattrinoError(e.code, err.get("code") or "http_error", err.get("message") or payload.get("detail") or str(e), err.get("details")) # ---------------------------------------------------------- authorize ---- def authorize(self, *, amount_minor=None, merchant=None, purpose=None, action_type="purchase", mode="dry_run", currency="USD", task=None, service_id=None, category=None, delegated_authority_id=None, task_id=None, disclosures=None, recurring=False, execution="auto", ttl_seconds=None, idempotency_key=None, envelope_id=None, action_class=None, subactions=None, commitment=None, attributes=None, fulfillment=None, context=None, proof_audience=None, **extra): """Ask whether an economic action is authorized. mode='binding' issues a portable EconomicAuthorization (an Idempotency-Key is generated if none is supplied).""" body = {"mode": mode, "action_type": action_type, "currency": currency, "recurring": bool(recurring), "execution": execution} for k, v in (("amount_minor", amount_minor), ("merchant", merchant), ("purpose", purpose), ("task", task), ("service_id", service_id), ("category", category), ("delegated_authority_id", delegated_authority_id), ("task_id", task_id), ("disclosures", disclosures), ("ttl_seconds", ttl_seconds), ("envelope_id", envelope_id), ("action_class", action_class), ("subactions", subactions), ("commitment", commitment), ("attributes", attributes), ("fulfillment", fulfillment), ("proof_audience", proof_audience)): if v is not None: body[k] = v if isinstance(task, str): # Phase 30: a plain-language task is compiled server-side into an envelope first compiled = self.compile_intent(task) body.pop("task", None) body["envelope_id"] = compiled["envelope"]["envelope_id"] if context: body.update({k: v for k, v in context.items() if k in ("category", "fulfillment", "attributes", "country", "destination")}) body.update(extra) headers = {} if mode == "binding": headers["Idempotency-Key"] = idempotency_key or ("qtrn-" + uuid.uuid4().hex) elif idempotency_key: headers["Idempotency-Key"] = idempotency_key return AuthorizationResult(self._request("POST", "/authorize", body, headers)) def authorize_binding(self, **kwargs): kwargs["mode"] = "binding" return self.authorize(**kwargs) # ---------------------------------------------------------- lifecycle ---- def get_authorization(self, authorization_id): return AuthorizationResult(self._request("GET", "/authorizations/" + authorization_id)) def list_authorizations(self, status=None, limit=50): q = "?limit=%d" % limit + (("&status=" + status) if status else "") return self._request("GET", "/authorizations" + q) def record_outcome(self, authorization_id, *, success, provider=None, provider_reference=None, actual_amount_minor=None, currency=None, merchant=None, material_result=None, protocol=None, receipt=None, side_effects=None, recurring=None, commitment=None, disclosed_fields=None, delegations_performed=None, category=None, action_type=None, attributes=None, proof_id=None, proof_verified=None): """Report what YOUR executor did under an authorization Quattrino did not execute. Phase 30: report commitment / disclosed_fields / delegations for compliance. R4: pass proof_id (and proof_verified=True) to link the outcome to the portable proof the executor verified; single use is enforced by the authorization lifecycle.""" body = {"success": bool(success)} for k, v in (("provider", provider), ("provider_reference", provider_reference), ("actual_amount_minor", actual_amount_minor), ("currency", currency), ("merchant", merchant), ("material_result", material_result), ("protocol", protocol), ("receipt", receipt), ("side_effects", side_effects), ("recurring", recurring), ("commitment", commitment), ("disclosed_fields", disclosed_fields), ("delegations_performed", delegations_performed), ("category", category), ("action_type", action_type), ("attributes", attributes), ("proof_id", proof_id), ("proof_verified", proof_verified)): if v is not None: body[k] = v return AuthorizationResult(self._request("POST", "/authorizations/" + authorization_id + "/outcome", body)) # ------------------------------------------- R4: signed portable proof (sandbox) ---- def get_proof(self, authorization_id): """The signed portable proof (EdDSA/Ed25519 JWS) of a binding AUTHORIZED externally executed authorization. Dry runs never have one. Hand the `token` to the executor / counterparty; they verify it with verify_proof().""" return self._request("GET", "/authorizations/" + authorization_id + "/proof") def verify_proof(self, proof, *, audience=None, expected=None, verifier=None): """Online verification (fail closed): signature, key, version, issuer, expiry, audience, registration, terms, revocation, single-use, live authority. Returns {valid, status, reason_codes, reasons, checks, proof, authorization}.""" return verify_proof(self.base, proof, audience=audience, expected=expected, verifier=verifier, timeout=self.timeout) def proof_keys(self): """Public (JWKS-style) sandbox verification keys - for offline signature checks only.""" return self._request("GET", "/public/proof-keys") # ------------------------------------------- R7: stable reason-code registry ---- def reason_codes(self, family=None, severity=None): q = [] if family: q.append("family=" + family) if severity: q.append("severity=" + severity) return self._request("GET", "/public/reason-codes" + (("?" + "&".join(q)) if q else "")) def describe_reason(self, code): return self._request("GET", "/public/reason-codes/" + code) def revoke(self, authorization_id, reason=None): return AuthorizationResult(self._request("POST", "/authorizations/" + authorization_id + "/revoke", {"reason": reason or ""})) def explain(self, authorization_id): return self._request("GET", "/authorizations/" + authorization_id + "/explain") def schema(self): return self._request("GET", "/authorize/schema") # ------------------------------------------- Phase 30: intent / envelope ---- def compile_intent(self, statement, *, proposal=None, statement_ref=None, create_envelope=True, parent_envelope_id=None): """Turn a human instruction (+ optional structured proposal) into a CompiledIntent and an AuthorityEnvelope v1. Meaning may be inferred; consequential permission never is.""" body = {"statement": statement, "create_envelope": create_envelope} for k, v in (("proposal", proposal), ("statement_ref", statement_ref), ("parent_envelope_id", parent_envelope_id)): if v is not None: body[k] = v return self._request("POST", "/intents/compile", body) def get_envelope(self, envelope_id): return self._request("GET", "/envelopes/" + envelope_id) def list_envelopes(self, status=None, limit=50): q = "?limit=%d" % limit + (("&status=" + status) if status else "") return self._request("GET", "/envelopes" + q) def create_envelope(self, dimensions, *, note=None, compiled_intent_id=None, parent_envelope_id=None): body = {"dimensions": dimensions} for k, v in (("note", note), ("compiled_intent_id", compiled_intent_id), ("parent_envelope_id", parent_envelope_id)): if v is not None: body[k] = v return self._request("POST", "/envelopes", body) def narrow_envelope(self, envelope_id, dimensions, *, note=None): """Self-restrict (an agent may only narrow).""" return self._request("POST", "/envelopes/" + envelope_id + "/narrow", {"dimensions": dimensions, "note": note}) def propose_envelope_change(self, envelope_id, dimensions, *, reason=None): """Propose a change (e.g. widening) - the owner decides; nothing changes until then.""" return self._request("POST", "/envelopes/" + envelope_id + "/propose", {"dimensions": dimensions, "reason": reason}) def discover(self, envelope_id=None): """What may I currently do? (effective capabilities, no policy internals)""" return self._request("GET", "/authority/me" + (("?envelope_id=" + envelope_id) if envelope_id else "")) def gap(self, **action): """What additional authority would I need to do X? (nothing recorded)""" return self._request("POST", "/authority/gap", action) def release_disclosure(self, authorization_id, scope, *, recipient): """Consume a RELEASE / RELEASE_ONCE / RELEASE_REDACTED decision bound to the authorization.""" return self._request("POST", "/authorizations/" + authorization_id + "/disclosures/" + scope + "/release", {"recipient": recipient}) def list_disclosures(self, authorization_id): return self._request("GET", "/authorizations/" + authorization_id + "/disclosures") # ------------------------------------------------ execution adapters ---- def execute(self, authorization, *, idempotency_key=None, **adapter_args): """Consume a binding authorization through the Quattrino execution adapter named in its execution_package (stripe_spt_merchant or credits_marketplace). Authorize-only integrations never call this.""" auth = authorization if isinstance(authorization, dict) else self.get_authorization(authorization) plan = auth.get("execution_plan") or {} if plan.get("execution_mode") != "quattrino": raise QuattrinoError(400, "authorization_external_only", "This authorization has no Quattrino execution adapter; execute externally " "and call record_outcome().") headers = {"Idempotency-Key": idempotency_key or ("qtrn-exec-" + uuid.uuid4().hex)} if plan.get("adapter") == "credits_marketplace": body = {"authorization_id": auth["authorization_id"], "quantity": adapter_args.get("quantity", 1)} return self._request("POST", "/agent/services/" + plan["service_id"] + "/purchase", body, headers) if plan.get("adapter") == "stripe_spt_merchant": required = ("sku", "terms_hash", "expected_total_minor") missing = [k for k in required if k not in adapter_args] if missing: raise QuattrinoError(400, "adapter_args_missing", "stripe_spt_merchant needs a quote first: " + ", ".join(missing)) body = {"merchant_key": plan["merchant_key"], "authorization_id": auth["authorization_id"], "quantity": adapter_args.get("quantity", 1), **{k: adapter_args[k] for k in required}, "category": adapter_args.get("category"), "purpose": auth.get("purpose")} return self._request("POST", "/agent/merchant-payments/purchase", body, headers) raise QuattrinoError(400, "unknown_adapter", "Unknown execution adapter: " + str(plan.get("adapter"))) def quote(self, merchant_key, sku, quantity=1): """Live merchant quote for the stripe_spt_merchant adapter.""" return self._request("POST", "/agent/merchant-payments/quote", {"merchant_key": merchant_key, "sku": sku, "quantity": quantity})