"""Exercise browser authorization against the local smoke server, without tokens in output.""" import hashlib import http.client import re import sqlite3 import sys import 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] == 403 status, headers, _ = call('POST', '/oauth/authorize', fields) assert status == 303 cookie_header = headers['set-cookie'] for attribute in ('HttpOnly', 'Secure', 'SameSite=Lax', 'Max-Age=31536000', 'Path=/'): assert attribute in cookie_header cookie = 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 html assert 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_uri assert 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] == 403 assert call('POST', '/oauth/authorize', {**fields, 'csrf': '0'*64}, cookie)[0] == 403 assert call('POST', '/oauth/authorize', fields)[0] == 403 assert call('POST', '/oauth/authorize', fields, cookie)[0] == 303 assert 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] == 403 conn.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] == 303 assert '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')