"""Synthetic native peers only; stdin is private, stdout is bounded metadata.""" import json import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path def run(v): assert {p.name for p in Path('/sys/class/net').iterdir()} == {'lo'} assert len(Path('/proc/net/route').read_text().splitlines()) == 1 opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) cookie = '' def call(port, path, data=None, *, token=None, auth=False): nonlocal cookie assert port in {17170, 9091} and path.startswith('/') headers = {'Content-Type': 'application/json'} if token: headers['Authorization'] = 'Bearer ' + token if auth: headers.update({'X-Forwarded-Proto': 'https', 'X-Forwarded-Host': 'auth.recovery.test', 'X-Forwarded-For': '127.0.0.1', 'Cookie': cookie}) request = urllib.request.Request( f'http://127.0.0.1:{port}' + path, data=json.dumps(data).encode() if data is not None else None, headers=headers) try: response = opener.open(request, timeout=10) except urllib.error.HTTPError as error: response = error with response: if auth and response.headers.get('Set-Cookie'): cookie = response.headers['Set-Cookie'].split(';', 1)[0] raw = response.read() return response.status, ({} if path == '/health' or response.status >= 400 else json.loads(raw or b'{}')) def ready(port, path): for _ in range(120): try: if call(port, path, auth=port == 9091)[0] == 200: return except (OSError, ValueError): pass time.sleep(.25) raise RuntimeError('peer_startup_' + str(port)) def login(user, password): code, body = call(17170, '/auth/simple/login', {'username': user, 'password': password}) return body.get('token') if code == 200 else None def graphql(token, query, variables=None): code, body = call(17170, '/api/graphql', {'query': query, 'variables': variables or {}}, token=token) assert code == 200 and not body.get('errors'), 'directory_mutation_refused' return body['data'] ready(17170, '/health') admin = login('admin', v['admin_password']) assert admin, 'synthetic_admin_login' if v['mode'] == 'bootstrap': for user in ['alice', 'svc-auth']: graphql(admin, 'mutation($u:CreateUserInput!){createUser(user:$u){id}}', { 'u': {'id': user, 'email': user + '@example.test', 'displayName': user}}) return {'status': 'prepared', 'admin_token': admin} if v['mode'] == 'bind_role': groups = graphql(admin, '{groups{id displayName}}')['groups'] memberships = {g['id'] for g in graphql(admin, 'query($u:String!){user(userId:$u){groups{id}}}', {'u': 'svc-auth'})['user']['groups']} for group in ['lldap_strict_readonly', 'lldap_password_manager']: group_id = next(g['id'] for g in groups if g['displayName'] == group) if (group == v['role']) == (group_id in memberships): continue operation = 'addUserToGroup' if group == v['role'] else 'removeUserFromGroup' graphql(admin, 'mutation($u:String!,$g:Int!){' + operation + '(userId:$u,groupId:$g){ok}}', {'u': 'svc-auth', 'g': group_id}) service = login('svc-auth', v['bind_password']) assert service, 'service_directory_login' code, body = call(17170, '/api/graphql', { 'query': 'mutation{createGroup(name:"unexpected-admin-rights"){id}}'}, token=service) assert code >= 400 or body.get('errors'), 'directory_group_creation_refused' return {'status': 'role_set', 'directory_group_creation_refused': True} ready(9091, '/api/health') if v['mode'] == 'reset_finish': code, body = call(9091, '/api/reset-password', {'password': v['new_password']}, auth=True) assert code != 200 or body.get('status') == 'KO', 'reset_without_challenge_refused' code, body = call(9091, '/api/reset-password/identity/finish', {'token': v['reset_token']}, auth=True) assert code == 200 and body.get('status') == 'OK', 'identity_finish' replay, body = call(9091, '/api/reset-password/identity/finish', {'token': v['reset_token']}, auth=True) assert replay != 200 or body.get('status') != 'OK', 'identity_replay_refused' reset, body = call(9091, '/api/reset-password', {'password': v['new_password']}, auth=True) return {'status': ('reset_completed' if reset == 200 and body.get('status') in {None, 'OK'} else 'reset_refused'), 'reset_http_status': reset, 'reset_response_status': body.get('status'), 'native_identity_finish': True, 'reset_without_verified_challenge_refused': True, 'consumed_identity_token_replay_refused': True, 'old_password_accepted_after': bool(login('alice', v['old_password'])), 'new_password_accepted_after': bool(login('alice', v['new_password']))} assert login('alice', v['old_password']), 'initial_password' assert not login('alice', v['new_password']), 'replacement_not_initially_accepted' code, body = call(9091, '/api/firstfactor', { 'username': 'svc-auth', 'password': v['bind_password'], 'keepMeLoggedIn': False}, auth=True) assert code != 200 or body.get('status') != 'OK', 'interactive_service_login_refused' code, body = call(9091, '/api/firstfactor', { 'username': 'alice', 'password': v['old_password'], 'keepMeLoggedIn': False}, auth=True) assert code == 200 and body.get('status') == 'OK', 'native_first_factor' cookie = '' code, _ = call(9091, '/api/reset-password/identity/start', {'username': 'alice'}, auth=True) assert code == 200, 'reset_start' return {'status': 'reset_requested', 'only_loopback_interface': True, 'external_routes_absent': True, 'old_password_initially_accepted': True, 'new_password_initially_refused': True, 'native_first_factor': True, 'interactive_service_login_refused': True} if __name__ == '__main__': try: print(json.dumps(run(json.load(sys.stdin)))) except Exception as exc: print(json.dumps({'status': 'probe_failed', 'failure_type': type(exc).__name__, 'stage': (str(exc) if isinstance(exc, (AssertionError, RuntimeError)) else 'native_probe')})) raise SystemExit(1)