atproto pds in zig pds.zat.dev
pds atproto
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140#!/usr/bin/env python3"""Exercise independent space access using the disposable smoke server/accounts."""import jsonimport osimport subprocessimport urllib.errorimport urllib.parseimport urllib.request
base = os.environ['SMOKE_BASE']owner = os.environ['SMOKE_TOKEN']member = os.environ['SMOKE_BOB_TOKEN']member_did = 'did:plc:spacesmokebob'member_policy = {'$type': 'com.atproto.simplespace.defs#memberListPolicy'}public_policy = {'$type': 'com.atproto.simplespace.defs#publicPolicy'}
def call(method, body=None, token=owner, params=None, headers=None): url = base + '/xrpc/' + method if params: url += '?' + urllib.parse.urlencode(params) request = urllib.request.Request(url, data=json.dumps(body).encode() if body is not None else None, headers={'authorization': 'Bearer ' + token, 'content-type': 'application/json', **(headers or {})}) try: response = urllib.request.urlopen(request, timeout=10) except urllib.error.HTTPError as error: response = error with response: return response.status, json.load(response)
def expect(method, expected, **kwargs): status, body = call(method, **kwargs) assert status == expected, f'{method}: expected {expected}, got {status} ({body.get("error")})' return body
create = {'spaceType': 'test.access.space', 'skey': 'self', 'readPolicy': member_policy, 'writePolicy': member_policy, 'appAccess': {'$type': 'com.atproto.simplespace.defs#open'}}space = expect('com.atproto.simplespace.createSpace', 200, body=create)['uri']for obsolete in ({'policy': member_policy}, {**create, 'policy': member_policy}): expect('com.atproto.simplespace.createSpace', 400, body={**create, **obsolete})expect('com.atproto.simplespace.updateSpace', 400, body={'space': space, 'policy': public_policy})expect('com.atproto.simplespace.addMember', 404, body={'space': space, 'did': member_did})for fields in ({'read': True}, {'read': 'true', 'write': True}, {'read': True, 'write': None}): expect('com.atproto.simplespace.putMember', 400, body={'space': space, 'did': member_did, **fields})expect('com.atproto.simplespace.putMember', 403, token=member, body={'space': space, 'did': member_did, 'read': True, 'write': True})expect('com.atproto.simplespace.listMembers', 403, token=member, params={'space': space})
for index, (read, write) in enumerate(((True, False), (False, True), (False, False), (True, True))): expect('com.atproto.simplespace.putMember', 200, body={'space': space, 'did': member_did, 'read': read, 'write': write}) members = expect('com.atproto.simplespace.listMembers', 200, params={'space': space})['members'] assert len(members) == 2 assert next(m for m in members if m['did'] == member_did) == {'did': member_did, 'read': read, 'write': write} expect('com.atproto.space.createRecord', 200, token=member, body={'space': space, 'repo': member_did, 'collection': 'test.access.record', 'rkey': f'case-{index}', 'record': {'$type': 'test.access.record', 'text': 'disposable'}}) delegation = expect('com.atproto.space.getDelegationToken', 200, token=member, params={'space': space})['token'] proof = subprocess.check_output(['./zig-out/bin/zds-space-signature', 'Bearer ' + delegation], text=True, stderr=subprocess.STDOUT).strip() signature_headers = dict(line.split(': ', 1) for line in proof.splitlines()) expect('com.atproto.space.getSpaceCredential', 200 if read else 403, token=delegation, body={'space': space}, headers=signature_headers)
expect('com.atproto.simplespace.updateSpace', 200, body={'space': space, 'readPolicy': public_policy})view = expect('com.atproto.simplespace.getSpace', 200, params={'space': space})assert view['readPolicy'] == public_policy and view['writePolicy'] == member_policy and 'policy' not in viewexpect('com.atproto.simplespace.updateSpace', 200, body={'space': space, 'writePolicy': public_policy})view = expect('com.atproto.simplespace.getSpace', 200, params={'space': space})assert view['readPolicy'] == public_policy and view['writePolicy'] == public_policyexpect('com.atproto.simplespace.removeMember', 200, body={'space': space, 'did': member_did})delegation = expect('com.atproto.space.getDelegationToken', 200, params={'space': space})['token']def signed_headers(authorization, audience=None): args = ['./zig-out/bin/zds-space-signature', authorization] if audience: args.append(audience) output = subprocess.check_output(args, text=True, stderr=subprocess.STDOUT) return dict(line.split(': ', 1) for line in output.strip().splitlines())
exchange = signed_headers('Bearer ' + delegation)credential = expect('com.atproto.space.getSpaceCredential', 200, token=delegation, body={'space': space}, headers=exchange)['credential']expect('com.atproto.space.getSpaceCredential', 401, token=delegation, body={'space': space}, headers=exchange)import base64claims = json.loads(base64.urlsafe_b64decode(credential.split('.')[1] + '==='))assert claims['exp'] - claims['iat'] == 600 and claims['cnf']['kid'].startswith('did:key:')authority = 'did:plc:permissionsmoke'authorization = 'Atproto-Space ' + credentialheaders = {'authorization': authorization, **signed_headers(authorization, authority)}for _ in range(2): expect('com.atproto.space.listRepos', 200, params={'space': space}, headers=headers)wrong = {'authorization': authorization, **signed_headers(authorization, member_did)}assert expect('com.atproto.space.listRepos', 401, params={'space': space}, headers=wrong)['error'] == 'BadSpaceAudience'repos = expect('com.atproto.space.listRepos', 200, params={'space': space, 'limit': 1}, headers=headers)assert len(repos['repos']) == 1 and repos['cursor'] == repos['repos'][0]['spaceRev']writer = repos['repos'][0]assert 'repoRev' in writer and 'rev' not in writercheckpoint = repos['cursor']assert expect('com.atproto.space.listRepos', 200, params={'space': space, 'cursor': checkpoint}, headers=headers) == {'repos': []}writer_auth = expect('com.atproto.server.getServiceAuth', 200, token=member, params={'aud': authority + '#atproto_space_host', 'lxm': 'com.atproto.space.notifyWrite'})['token']body = {'space': space, 'repo': member_did, 'repoRev': writer['repoRev'], 'hash': writer['hash']}expect('com.atproto.space.notifyWrite', 200, token=writer_auth, body=body)assert expect('com.atproto.space.listRepos', 200, params={'space': space}, headers=headers)['cursor'] == checkpointexpect('com.atproto.space.notifyWrite', 400, token=writer_auth, body={**body, 'repoRev': 'not-a-tid'})assert expect('com.atproto.space.notifyWrite', 400, token=writer_auth, body={**body, 'repoRev': '7zzzzzzzzzzzz'})['error'] == 'FutureRev'expect('com.atproto.space.notifyWrite', 401, token=writer_auth, body={**body, 'repo': authority})import http.clientconnection = http.client.HTTPConnection(urllib.parse.urlparse(base).netloc)connection.putrequest('GET', '/xrpc/com.atproto.space.listRepos?' + urllib.parse.urlencode({'space': space}))for name, value in headers.items(): connection.putheader(name, value)connection.putheader('authorization', authorization)connection.endheaders()response = connection.getresponse()assert response.status == 401response.read()connection.close()attacker = expect('com.atproto.server.getServiceAuth', 200, token=member, params={'aud': authority, 'lxm': 'com.atproto.space.notifyCredentialRevoked'})['token']expect('com.atproto.space.notifyCredentialRevoked', 401, token=attacker, body={'space': space, 'credentials': [claims['jti']]})wrong_method = expect('com.atproto.server.getServiceAuth', 200, params={'aud': authority, 'lxm': 'com.atproto.space.notifyWrite'})['token']expect('com.atproto.space.notifyCredentialRevoked', 401, token=wrong_method, body={'space': space, 'credentials': [claims['jti']]})service = expect('com.atproto.server.getServiceAuth', 200, params={'aud': authority, 'lxm': 'com.atproto.space.notifyCredentialRevoked'})['token']for _ in range(2): expect('com.atproto.space.notifyCredentialRevoked', 200, token=service, body={'space': space, 'credentials': [claims['jti'], claims['jti']]})assert expect('com.atproto.space.listRepos', 401, params={'space': space}, headers=headers)['error'] == 'CredentialRevoked'for ids in ([], [''], [claims['jti']] * 101): expect('com.atproto.space.notifyCredentialRevoked', 400, token=service, body={'space': space, 'credentials': ids})expect('com.atproto.simplespace.deleteSpace', 200, body={'space': space})print('simplespace access: independent policies, member upserts, credentials, writes, and obsolete payloads passed')