tegwick
c1cf6cf54c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029c-c003-7283-acc9-930b434e6cae
34 lines
1.8 KiB
Python
34 lines
1.8 KiB
Python
"""Create-only approval adapter; token custody/exchange belongs to workload setup."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
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
|