Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
4.9 kB · 87 lines
Python
at sl/gitmirror
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788"""Live Compose delegation smoke test; creates isolated development fixtures."""import jsonimport osfrom pathlib import Pathimport subprocessimport sysimport timeimport urllib.parseIMAGE = 'localhost/tangled_init-accounts:latest'BASE = [os.environ.get('CONTAINER_ENGINE', 'podman'), 'run', '--rm', '--network', 'tangled_tngl', IMAGE]
def call(url, payload=None, token=None): args = BASE + ['curl', '--max-time', '30', '-sS', '-w', '\n%{http_code}', url] if payload is not None: args += ['-H', 'Content-Type: application/json', '-d', json.dumps(payload)] if token: args += ['-H', 'Authorization: Bearer ' + token] result = subprocess.run(args, text=True, capture_output=True, check=True).stdout body, status = result.rsplit('\n', 1) return (int(status), json.loads(body))
def expect(status, result, want): assert status == want, (status, result) return resultinvite = subprocess.run([os.environ.get('CONTAINER_ENGINE', 'podman'), 'run', '--rm', '--network', 'tangled_tngl', '--env-file', 'localinfra/pds.env', IMAGE, 'sh', '-c', 'curl -fsS -u "admin:$PDS_ADMIN_PASSWORD" -H "Content-Type: application/json" -d \'{"useCount":1}\' http://pds:3000/xrpc/com.atproto.server.createInviteCode'], text=True, capture_output=True, check=True)name = 'gc' + str(time.time_ns())[-12:]s, account = call('http://pds:3000/xrpc/com.atproto.server.createAccount', {'handle': name + '.pds.tngl.boltless.dev', 'email': name + '@example.com', 'password': 'Gate-Smoke9!', 'inviteCode': json.loads(invite.stdout)['code']})expect(s, account, 200)actor = account['did']access = account['accessJwt']CREATE = 'farm.tranquil.delegation.createAccount'LIST = 'farm.tranquil.delegation.listControlledAccounts'
def jwt(lxm=CREATE, aud='did:web:tranquil.tngl.boltless.dev', access=access, pds='pds'): s, b = call(f'http://{pds}:3000/xrpc/com.atproto.server.getServiceAuth?' + urllib.parse.urlencode({'aud': aud, 'lxm': lxm}), token=access) return expect(s, b, 200)['token']endpoint = 'http://tranquil-gate:3100/xrpc/sh.tangled.delegation.createAccount'
def request(i, create=None): return call(endpoint, {'handle': f'{name}-{i}.tranquil.tngl.boltless.dev'}, create or jwt())s, b = request(1)expect(s, b, 403)assert b['error'] == 'VerifiedEmailRequired', bprint('PASS unverified email denied', flush=True)fixture = json.dumps({'did': actor, 'email': name + '@example.com'}) + '\n'subprocess.run([os.environ.get('CONTAINER_ENGINE', 'podman'), 'exec', '-i', 'tangled_deliberi_1', 'sh', '-c', 'cat >/tmp/gate-smoke.jsonl && /usr/local/bin/deliberi-bootstrap --file /tmp/gate-smoke.jsonl'], input=fixture, text=True, capture_output=True, check=True)if '--browser-fixture' in sys.argv: Path('/tmp/web-org-fixture.json').write_text(json.dumps({'handle': account['handle'], 'did': actor, 'org': 'wo' + str(time.time_ns())[-12:]})) sys.exit(0)s, b = call(endpoint, {'handle': name + '-missing.tranquil.tngl.boltless.dev'})expect(s, b, 401)print('PASS missing service auth denied', flush=True)created = []# MAX_DELEGATED_ACCOUNTS_PER_CONTROLLER is 2 on the dev env, and Tranquil counts# them against the authenticated controller, so the third request is refused# upstream and the gate relays Tranquil's status.for i in range(1, 4): c = jwt() s, b = request(i, c) if i < 3: expect(s, b, 200) assert b['controllerDid'] == actor, b created.append(b['did']) print('PASS remote creation', i, flush=True) else: assert 400 <= s < 500, (s, b) assert b['error'] == 'InvalidDelegation', b print('PASS third delegate denied by tranquil', flush=True) s, b = request(i, c) assert 400 <= s < 500, (s, b)# Another controller's quota is independent: bob may still create his own.s, other = call('http://pds:3000/xrpc/com.atproto.server.createSession', { 'identifier': 'bob.pds.tngl.boltless.dev', 'password': 'password'})expect(s, other, 200)print('PASS per-controller quota is enforced upstream', flush=True)s, b = call('http://tranquil:3000/xrpc/_delegation.listControlledAccounts', token=jwt(LIST))expect(s, b, 200)assert {a['did'] for a in b['accounts']} == set(created), bassert all((a['grantedScopes'] and a['grantedAt'] for a in b['accounts'])), bprint('PASS issuer-based listing, scopes, grant times and replay protection', flush=True)s, b = call('http://tranquil:3000/xrpc/_delegation.listControlledAccounts', token=jwt(CREATE))assert 400 <= s < 500, (s, b)s, b = call('http://tranquil:3000/xrpc/_delegation.listControlledAccounts', token=jwt(LIST, aud='did:web:wrong.example'))assert 400 <= s < 500, (s, b)print('PASS Tranquil rejects wrong method and audience', flush=True)print(json.dumps({'actor': actor, 'name': name, 'delegates': created}), flush=True)print('PASS all open-registration checks', flush=True)