165 lines
9.7 KiB
Python
165 lines
9.7 KiB
Python
"""Verify the actual QA browser response protocol; never claim visual acceptance."""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import http.cookiejar
|
|
import json
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
root = Path(__file__).resolve().parent.parent
|
|
private = root / 'storage/app/private/investmoz-v2/portal-audit'
|
|
diagnosis = json.loads((private / 'mfa-key-diagnosis-qa.json').read_text(encoding='utf-8-sig'))
|
|
target = diagnosis['target']
|
|
assert re.fullmatch(r'desk_restore_\d{8}_\d{6}_test', target['database']) and target['port'] == 55441
|
|
BASE = 'http://127.0.0.1:18097'
|
|
events = []
|
|
|
|
|
|
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
return None
|
|
|
|
|
|
class BrowserProtocol:
|
|
def __init__(self, label):
|
|
self.label = label
|
|
self.jar = http.cookiejar.CookieJar()
|
|
self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar), NoRedirect())
|
|
self.version = None
|
|
self.page('/login', initial=True)
|
|
|
|
def call(self, method, path, data=None, status=200, initial=False):
|
|
headers = {'Accept': 'text/html', 'Content-Type': 'application/json'}
|
|
if not initial:
|
|
headers.update({'X-Inertia': 'true', 'X-Inertia-Version': self.version})
|
|
for cookie in self.jar:
|
|
if cookie.name == 'XSRF-TOKEN':
|
|
headers['X-XSRF-TOKEN'] = urllib.parse.unquote(cookie.value)
|
|
req = urllib.request.Request(BASE + path, data=None if data is None else json.dumps(data).encode(), headers=headers, method=method)
|
|
try:
|
|
response = self.opener.open(req)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
assert response.headers.get('X-Investmoz-QA-Target') == target['database'] + ':55441', 'QA target mismatch; aborted'
|
|
assert response.headers.get('X-Investmoz-QA-Key-Fingerprint') == diagnosis['qa_key_fingerprint'], 'QA key mismatch; aborted'
|
|
assert response.status == status, f'{method} {path}: expected {status}, observed {response.status}; body withheld'
|
|
location = response.headers.get('Location')
|
|
if location:
|
|
parsed = urllib.parse.urlsplit(urllib.parse.urljoin(BASE + '/', location))
|
|
expected_origin = urllib.parse.urlsplit(BASE)
|
|
assert (parsed.scheme, parsed.netloc) == (expected_origin.scheme, expected_origin.netloc), 'Redirect leaves the QA origin; aborted'
|
|
location = parsed.path or '/'
|
|
events.append({'session_label': self.label, 'method': method, 'path': path, 'status': response.status,
|
|
'redirect_path': location, 'trace': response.headers.get('X-Investmoz-QA-Auth-Trace')})
|
|
return response.read(), location
|
|
|
|
def page(self, path, initial=False):
|
|
raw, _ = self.call('GET', path, initial=initial)
|
|
if initial:
|
|
m = re.search(rb'<script[^>]*data-page="app"[^>]*>(.*?)</script>', raw, re.S)
|
|
assert m, 'Inertia page missing; body withheld'
|
|
page = json.loads(m.group(1))
|
|
else:
|
|
page = json.loads(raw)
|
|
self.version = page['version']
|
|
return page
|
|
|
|
def identity(self):
|
|
values = [c.value for c in self.jar if c.name == 'opportunity_resources_qa']
|
|
assert len(values) == 1
|
|
return hashlib.sha256(values[0].encode()).digest() # In-memory comparison only.
|
|
|
|
|
|
def otp(secret):
|
|
digest = hmac.new(base64.b32decode(secret), struct.pack('>Q', int(time.time()) // 30), hashlib.sha1).digest()
|
|
offset = digest[-1] & 15
|
|
return f'{(struct.unpack(">I", digest[offset:offset + 4])[0] & 0x7fffffff) % 1000000:06d}'
|
|
|
|
|
|
owner = json.loads((private / 'investor-mfa-qa-private.json').read_text(encoding='utf-8-sig'))
|
|
assert owner['actor_id'] == 18
|
|
selector = subprocess.run(['php', '-r', r"require 'tools/portal-qa-environment.php'; $target=portalQaEnvironment(); require 'vendor/autoload.php'; $app=require 'bootstrap/app.php'; $app->make(\Illuminate\Contracts\Console\Kernel::class)->bootstrap(); portalQaAssertTarget($target); echo json_encode(\Illuminate\Support\Facades\DB::table('users')->where('id',13)->first(['id','email']),JSON_THROW_ON_ERROR);"], cwd=root, capture_output=True)
|
|
assert selector.returncode == 0, 'Guarded selector read failed; output withheld'
|
|
restored = json.loads(selector.stdout)
|
|
assert restored['id'] == 13 and not restored['email'].endswith('@example.invalid')
|
|
bad = BrowserProtocol('restored-account-selection-negative')
|
|
_, redirect = bad.call('POST', '/login', {'email': restored['email'], 'password': 'synthetic-placeholder'}, status=302)
|
|
assert redirect == '/login'
|
|
page = bad.page('/login')
|
|
assert page['props']['qaAccount']['id'] == 18 and not page['props']['auth']['user']
|
|
assert 'contas sintéticas' in page['props']['errors']['email']
|
|
|
|
normal = BrowserProtocol('synthetic-investor18-first-session')
|
|
assert normal.call('GET', '/investimentos', status=302)[1] == '/login'
|
|
normal.page('/login')
|
|
assert normal.call('POST', '/login', {k: owner[k] for k in ['email', 'password']}, status=302)[1] == '/two-factor-challenge'
|
|
challenge = normal.page('/two-factor-challenge')
|
|
assert challenge['props']['qaAccount']['id'] == 18 and not challenge['props']['auth']['user']
|
|
assert normal.call('POST', '/two-factor-challenge', {'code': 'invalid'}, status=302)[1] == '/two-factor-challenge'
|
|
challenge = normal.page('/two-factor-challenge')
|
|
assert challenge['props']['errors']['code'] and not challenge['props']['auth']['user']
|
|
before = normal.identity()
|
|
slot = int(time.time()) // 30
|
|
assert normal.call('POST', '/two-factor-challenge', {'code': otp(owner['totp_secret'])}, status=302)[1] == '/investimentos'
|
|
assert normal.identity() != before
|
|
index = normal.page('/investimentos')
|
|
assert index['component'] == 'investor-path/index' and index['props']['auth']['user']['id'] == 18
|
|
ids = [case['id'] for case in index['props']['contexts']]
|
|
case = '413ba1f6-bdf6-402a-a561-c4e2d4098d1a'
|
|
assert case in ids
|
|
assert '5935737d-9e91-44d6-87fa-05444a37c117' not in ids and '3026f22a-d3d9-42ed-9660-529f29dc5b90' not in ids
|
|
assert normal.page('/investimentos')['props']['auth']['user']['id'] == 18
|
|
assert normal.page('/investimentos/' + case)['props']['case']['id'] == case
|
|
normal.call('GET', '/investimentos/5935737d-9e91-44d6-87fa-05444a37c117', status=403)
|
|
assert normal.call('POST', '/logout', {}, status=302)[1] == '/'
|
|
assert normal.call('GET', '/investimentos', status=302)[1] == '/login'
|
|
|
|
new = BrowserProtocol('synthetic-investor18-new-cookie-session')
|
|
assert new.call('GET', '/investimentos', status=302)[1] == '/login'
|
|
new.page('/login')
|
|
assert new.call('POST', '/login', {k: owner[k] for k in ['email', 'password']}, status=302)[1] == '/two-factor-challenge'
|
|
new.page('/two-factor-challenge')
|
|
while int(time.time()) // 30 == slot:
|
|
time.sleep(0.5) # Use a fresh TOTP without disabling replay controls; at most30s.
|
|
assert new.call('POST', '/two-factor-challenge', {'code': otp(owner['totp_secret'])}, status=302)[1] == '/investimentos'
|
|
assert new.page('/investimentos')['props']['auth']['user']['id'] == 18
|
|
assert new.page('/investimentos/' + case)['props']['case']['id'] == case
|
|
new.call('POST', '/logout', {}, status=302)
|
|
|
|
negative = json.loads((private / 'mfa-negative-qa-private.json').read_text(encoding='utf-8-sig'))
|
|
failed = BrowserProtocol('synthetic-wrong-key19-negative')
|
|
assert failed.call('POST', '/login', {k: negative[k] for k in ['email', 'password']}, status=302)[1] == '/two-factor-challenge'
|
|
failed.page('/two-factor-challenge')
|
|
assert failed.call('POST', '/two-factor-challenge', {'code': '123456'}, status=302)[1] == '/login'
|
|
page = failed.page('/login')
|
|
assert not page['props']['auth']['user'] and 'Não foi possível verificar' in page['props']['errors']['email']
|
|
assert failed.call('GET', '/investimentos', status=302)[1] == '/login'
|
|
|
|
traces = {row['trace']: row for row in [json.loads(line) for line in (private / 'auth-session-trace-qa.jsonl').read_text().splitlines()]}
|
|
successful_mfa = [traces[event['trace']] for event in events if event['path'] == '/two-factor-challenge' and event['method'] == 'POST'
|
|
and traces[event['trace']]['after']['authenticated_user'] == 18]
|
|
assert len(successful_mfa) == 2
|
|
assert all(row['persisted']['stored_web_user'] == 18 for row in successful_mfa)
|
|
assert all(row['loaded']['session'] != row['after']['session'] for row in successful_mfa)
|
|
proof = {'case': 'C37', 'captured_at_utc': datetime.now(timezone.utc).isoformat(), 'target': target, 'actor_id': 18,
|
|
'restored_accounts_stopped_before_mfa': True, 'qa_identity_on_login_and_challenge': True,
|
|
'restored_account13_selector_explicitly_tested': True,
|
|
'invalid_mfa_preserves_challenge_without_auth': True, 'decryption_failure_refuses_auth_and_returns_message': True,
|
|
'mfa_browser_redirect_protocol_passed': True, 'session_regenerated_and_user18_persisted': True,
|
|
'investimentos_get_and_reload_passed': True, 'logout_revokes_access': True, 'fresh_session_recovers_same_case': True,
|
|
'other_actor_case_forbidden': True, 'requests': events,
|
|
'headers_bodies_cookies_codes_exported': False, 'visual_acceptance_passed': False, 'active_writes': 0,
|
|
'source_hashes': {name: hashlib.sha256((root / name).read_bytes()).hexdigest() for name in [
|
|
'tools/portal-qa-account-selection.php', 'tools/opportunity-resources-qa.php',
|
|
'resources/js/pages/auth/login.tsx', 'resources/js/pages/auth/two-factor-challenge.tsx']}}
|
|
(private / 'c37-auth-redirect-qa.json').write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
print(json.dumps({k: v for k, v in proof.items() if k not in ['requests', 'source_hashes']}, ensure_ascii=True))
|