tegwick
0e66a33389
A registered submitter with a present_existing profile may pose a memo that pins an approval it created itself in approval-engine. Intake verifies the object read-only (requested, no entries, human_control, actor == profile actor, digest == memo, binding == intent, required_count 1, unexpired, one memo per approval) and never creates, enters or consumes. The recorded requester reaches review and the access-engine context. Existing submitters are unchanged. Answer doc accepts S1 (profile, not route; packet carries the admission), S4 in principle (waits on gate-house classes and policy) and S3 (exists behind INFD_HUB_WRITEBACK). Workplan INFD-WP-0010 proposed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 5502@bnt-lap001 Assistant-Session: 68b3b02f-5d0f-4e71-8421-796f62e50196
69 lines
3.3 KiB
Python
69 lines
3.3 KiB
Python
"""Create-only approval adapter and the present_existing read-only adapter.
|
|
|
|
Token custody/exchange belongs to workload setup.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
import re
|
|
|
|
from .http_transport import JSONTransport, fixed_origin
|
|
from .intake import canonical
|
|
from .intake_service import IntakeError
|
|
|
|
|
|
class IntakeApprovalClient:
|
|
def __init__(self, origin, token_provider, *, transport=None, allow_internal_http=False):
|
|
self.origin = fixed_origin(origin, allow_internal_http=allow_internal_http)
|
|
self.token_provider = token_provider
|
|
self.transport = transport or JSONTransport(allow_internal_http=allow_internal_http)
|
|
|
|
def __call__(self, request):
|
|
if (set(request) != {'binding', 'human_control', 'required_count'}
|
|
or request['human_control'] is not True or type(request['required_count']) is not int
|
|
or request['required_count'] != 1 or request['binding'].get('actor') != 'informed-decision'):
|
|
raise IntakeError(400, 'invalid_approval_intent')
|
|
try:
|
|
token = self.token_provider()
|
|
if (not isinstance(token, str) or not token or len(token) > 16384
|
|
or not token.isascii() or any(c.isspace() for c in token)):
|
|
raise ValueError()
|
|
now = datetime.now(timezone.utc)
|
|
body = {**request, 'pdp_path': False, 'validity': {
|
|
'not_before': now.isoformat(), 'expires_at': (now + timedelta(hours=24)).isoformat()}}
|
|
status, result = self.transport.request('POST', self.origin + '/v1/approvals',
|
|
headers={'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}, body=canonical(body))
|
|
if status != 201:
|
|
raise ValueError()
|
|
return result
|
|
except Exception:
|
|
raise IntakeError(503, 'approval_create_unresolved') from None
|
|
|
|
|
|
|
|
class IntakeApprovalReader:
|
|
"""GET-only adapter for present_existing (INFD-WP-0010-T01).
|
|
|
|
It has no create, entry or consume route. Any failure is ``None``-equivalent
|
|
for the caller: intake refuses to present what it could not read.
|
|
"""
|
|
|
|
def __init__(self, origin, token_provider, *, transport=None, allow_internal_http=False):
|
|
self.origin = fixed_origin(origin, allow_internal_http=allow_internal_http)
|
|
self.token_provider = token_provider
|
|
self.transport = transport or JSONTransport(allow_internal_http=allow_internal_http)
|
|
|
|
def __call__(self, approval_id):
|
|
if not isinstance(approval_id, str) or not re.fullmatch(
|
|
r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", approval_id):
|
|
raise IntakeError(400, "invalid_approval_id")
|
|
try:
|
|
token = self.token_provider()
|
|
if (not isinstance(token, str) or not token or len(token) > 16384
|
|
or not token.isascii() or any(c.isspace() for c in token)):
|
|
raise ValueError()
|
|
status, result = self.transport.request('GET', self.origin + '/v1/approvals/' + approval_id,
|
|
headers={'Authorization': 'Bearer ' + token})
|
|
if status != 200 or not isinstance(result, dict):
|
|
raise ValueError()
|
|
return result
|
|
except Exception:
|
|
raise IntakeError(503, 'approval_unreadable') from None
|