tegwick
93b6ba2aae
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
59 lines
2.7 KiB
Python
59 lines
2.7 KiB
Python
"""Check the packaged caller module and transition without credentials or mail."""
|
|
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import email_connect.callers as callers
|
|
from email_connect.transactional import SQLiteDeliveryStore
|
|
|
|
assert os.getuid() == 10001
|
|
assert hashlib.sha256(Path(callers.__file__).read_bytes()).hexdigest() == sys.argv[1]
|
|
assert len(Path('/proc/net/route').read_text().splitlines()) == 1
|
|
token = "synthetic-packaged-caller-token"
|
|
Path('/tmp/token').write_text(token)
|
|
document = json.loads(Path('/review/config/callers.user-engine.draft.json').read_text())
|
|
document['callers'][0]['token_file'] = '/tmp/token'
|
|
Path('/tmp/policy.json').write_text(json.dumps(document))
|
|
policies = callers.load_callers('/tmp/policy.json')
|
|
assert ('tenant:synthetic-family', 'coulomb-social') in policies[0]
|
|
assert ('tenant:platform', 'coulomb-social') not in policies[0]
|
|
store = SQLiteDeliveryStore('/tmp/ledger.sqlite3')
|
|
|
|
|
|
class ForbiddenProvider:
|
|
def send(self, *args):
|
|
raise AssertionError('provider_call_forbidden')
|
|
|
|
|
|
app = callers.ScopedTransactionalApplication(store, ForbiddenProvider(),
|
|
'https://issuer.example.test', policies)
|
|
old = {'id': 'evt-synthetic', 'source': 'user-engine', 'type': 'family_member.invited',
|
|
'tenant': 'tenant:synthetic-family',
|
|
'data': {'primary_email': 'member@example.test', 'invitation_id': 'inv-synthetic'}}
|
|
fingerprint = hashlib.sha256(json.dumps(old, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
|
store.reserve_send(old['id'], fingerprint)
|
|
|
|
|
|
def invoke(payload):
|
|
raw = json.dumps(payload).encode()
|
|
response = {}
|
|
body = json.loads(b''.join(app({'PATH_INFO': '/v1/send', 'REQUEST_METHOD': 'POST',
|
|
'CONTENT_LENGTH': str(len(raw)), 'wsgi.input': io.BytesIO(raw),
|
|
'HTTP_AUTHORIZATION': 'Bearer ' + token, 'HTTP_IDEMPOTENCY_KEY': payload['id']},
|
|
lambda status, headers: response.update(status=int(status.split()[0])))))
|
|
return response['status'], body
|
|
|
|
|
|
bound = dict(old, client_id='coulomb-social')
|
|
assert invoke(bound)[1]['error'] == 'delivery_outcome_unknown'
|
|
store.record(old['id'], 'synthetic-old-reference')
|
|
assert invoke(bound)[1]['status'] == 'duplicate'
|
|
assert invoke(dict(bound, tenant='tenant:platform'))[1]['error'] == 'caller_scope_denied'
|
|
assert invoke(dict(bound, data=dict(old['data'], primary_email='changed@example.test')))[1]['error'] == 'idempotency_payload_mismatch'
|
|
print(json.dumps({'status': 'packaged_caller_contract_verified', 'uid': os.getuid(),
|
|
'mail_sent': False, 'legacy_unknown_blocked': True,
|
|
'legacy_accepted_preserved': True, 'glob_denials_verified': True}))
|