atproto pds in zig pds.zat.dev
pds atproto
Something went wrong. Try again.
Python
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576"""Exercise browser authorization against the local smoke server, without tokens in output."""import hashlibimport http.clientimport reimport sqlite3import sysimport urllib.parse
base, database = sys.argv[1:]origin = urllib.parse.urlsplit(base)conn = sqlite3.connect(database)
def seed(name, hint='smoke.test', prompt=None): conn.execute('INSERT INTO oauth_requests (request_id,client_id,redirect_uri,scope,state,code_challenge,code_challenge_method,login_hint,expires_at,prompt) VALUES (?,?,?,?,?,?,?,?,unixepoch()+600,?)', (name, 'http://127.0.0.1:2586/oauth-client-metadata.json', 'http://127.0.0.1/callback', 'atproto', name, 'challenge', 'S256', hint, prompt)) conn.commit() return 'urn:ietf:params:oauth:request_uri:' + name
def call(method, uri, fields=None, cookie=None, request_origin=base): connection = http.client.HTTPConnection(origin.hostname, origin.port) headers = {} if cookie: headers['Cookie'] = cookie body = None if fields is not None: headers['Content-Type'] = 'application/x-www-form-urlencoded' headers['Origin'] = request_origin body = urllib.parse.urlencode(fields) connection.request(method, uri, body, headers) response = connection.getresponse() result = response.status, dict(response.getheaders()), response.read().decode() connection.close() return result
def page(uri, cookie=None): return call('GET', '/oauth/authorize?' + urllib.parse.urlencode({'request_uri': uri}), cookie=cookie)
uri = seed('browser-login')fields = {'request_uri': uri, 'username': 'smoke.test', 'password': 'password', 'remember_browser': '1'}assert call('POST', '/oauth/authorize', fields, request_origin='https://evil.example')[0] == 403status, headers, _ = call('POST', '/oauth/authorize', fields)assert status == 303cookie_header = headers['set-cookie']for attribute in ('HttpOnly', 'Secure', 'SameSite=Lax', 'Max-Age=31536000', 'Path=/'): assert attribute in cookie_headercookie = cookie_header.split(';')[0]token = cookie.split('=', 1)[1]assert conn.execute('SELECT token_hash FROM browser_sessions WHERE token_hash=?', (hashlib.sha256(token.encode()).hexdigest(),)).fetchone()assert not conn.execute('SELECT token_hash FROM browser_sessions WHERE token_hash=?', (token,)).fetchone()uri = seed('browser-consent')status, headers, html = page(uri, cookie)assert status == 200 and 'name="password"' not in html and 'value="continue"' in htmlassert headers['cache-control'] == 'no-store' and headers['x-frame-options'] == 'DENY'# chromium and webkit apply form-action to every hop of a form submission's# redirect chain, so 'self' would block the 303 to the client's redirect_uriassert headers['content-security-policy'] == "frame-ancestors 'none'"csrf = re.search(r'name="csrf" value="([^"]+)"', html)[1]fields = {'request_uri': uri, 'browser_action': 'continue', 'csrf': csrf}assert call('POST', '/oauth/authorize', fields, cookie, 'https://evil.example')[0] == 403assert call('POST', '/oauth/authorize', {**fields, 'csrf': '0'*64}, cookie)[0] == 403assert call('POST', '/oauth/authorize', fields)[0] == 403assert call('POST', '/oauth/authorize', fields, cookie)[0] == 303assert conn.execute('SELECT sub FROM oauth_requests WHERE request_id=?', ('browser-consent',)).fetchone()[0] == 'did:plc:smoketest'assert 'name="password"' in page(seed('browser-forced', prompt='login'), cookie)[2]assert 'name="password"' in page(seed('browser-mismatch', hint='signup-smoke.test'), cookie)[2]assert call('POST', '/oauth/authorize', {**fields, 'request_uri': seed('browser-csrf-other')}, cookie)[0] == 403conn.execute('UPDATE browser_sessions SET expires_at=unixepoch()-1')conn.commit()assert 'name="password"' in page(uri, cookie)[2]conn.execute('UPDATE browser_sessions SET expires_at=unixepoch()+600')conn.commit()assert call('POST', '/oauth/authorize', {**fields, 'browser_action': 'forget'}, cookie)[0] == 303assert 'name="password"' in page(uri, cookie)[2]assert not conn.execute('SELECT 1 FROM browser_sessions').fetchone()print('remembered browser login: consent, CSRF, prompt, account binding, expiry, and revocation passed')