186 lines
12 KiB
Python
186 lines
12 KiB
Python
"""Real QA MFA enrollment/login and scenario A. Never export credentials or request bodies."""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import http.cookiejar
|
|
import json
|
|
import re
|
|
import struct
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from datetime import datetime, timedelta, 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 Session:
|
|
def __init__(self, label):
|
|
self.label = label
|
|
self.jar = http.cookiejar.CookieJar()
|
|
self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar))
|
|
self.call('GET', '/login', browser=True)
|
|
|
|
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() # Only compared in memory.
|
|
|
|
def call(self, method, path, data=None, status=200, browser=False, extra=None):
|
|
assert path.startswith('/') and not path.startswith('//')
|
|
headers = {'Accept': 'text/html' if browser else 'application/json', 'Content-Type': 'application/json'}
|
|
for cookie in self.jar:
|
|
if cookie.name == 'XSRF-TOKEN':
|
|
headers['X-XSRF-TOKEN'] = urllib.parse.unquote(cookie.value)
|
|
headers.update(extra or {})
|
|
req = urllib.request.Request(BASE + path, headers=headers, data=None if data is None else json.dumps(data).encode(), 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 changed: abort'
|
|
assert response.headers.get('X-Investmoz-QA-Key-Fingerprint') == diagnosis['qa_key_fingerprint'], 'QA web/CLI key mismatch: abort'
|
|
assert response.headers.get('X-Investmoz-QA-Config-Cache') == 'bypassed', 'QA config cache mismatch: abort'
|
|
raw = response.read()
|
|
assert response.status == status, f'{method} {path}: expected {status}, observed {response.status}; response body withheld'
|
|
events.append({'actor': self.label, 'method': method, 'url': path, 'status': response.status, 'qa_target_and_key_verified': True})
|
|
return raw if browser else (json.loads(raw) if raw else None)
|
|
|
|
def login(self, credentials, mfa=False):
|
|
result = self.call('POST', '/login', {k: credentials[k] for k in ['email', 'password']})
|
|
if mfa:
|
|
assert result['two_factor'] is True
|
|
self.call('GET', '/investimentos', status=401)
|
|
self.call('POST', '/two-factor-challenge', {'code': otp(credentials['totp_secret'])}, status=204)
|
|
return result
|
|
|
|
|
|
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}'
|
|
|
|
|
|
actor_file = private / 'investor-mfa-qa-private.json'
|
|
credentials = json.loads(actor_file.read_text(encoding='utf-8-sig'))
|
|
owner = Session('synthetic-investor-MFA') # GET verifies isolation before writes.
|
|
if 'totp_secret' not in credentials:
|
|
owner.login(credentials)
|
|
owner.call('POST', '/user/confirm-password', {'password': credentials['password']}, status=201)
|
|
owner.call('POST', '/user/two-factor-authentication')
|
|
credentials['totp_secret'] = owner.call('GET', '/user/two-factor-secret-key')['secretKey']
|
|
actor_file.write_text(json.dumps(credentials, indent=2), encoding='utf-8')
|
|
owner.call('POST', '/user/confirmed-two-factor-authentication', {'code': otp(credentials['totp_secret'])})
|
|
recovery = owner.call('GET', '/user/two-factor-recovery-codes')
|
|
assert len(recovery) == 8
|
|
del recovery
|
|
owner.call('POST', '/logout', status=204)
|
|
owner = Session('synthetic-investor-MFA-enrolled')
|
|
owner.login(credentials, mfa=True)
|
|
owner.call('GET', '/investimentos')
|
|
|
|
case = owner.call('POST', '/investimentos', {'title': 'Cenário A MFA QA — sem terra nem financiamento externo',
|
|
'client_key': '61000919-0000-4000-8000-000000000018', 'reuse_consent': True}, status=201)['id']
|
|
url = '/investimentos/' + case
|
|
before = owner.call('GET', url)
|
|
assert len(before['steps']) == 19 and len(before['phases']) == 5
|
|
prior_session = owner.identity()
|
|
owner.call('POST', '/logout', status=204)
|
|
owner.call('GET', url, status=401)
|
|
owner = Session('synthetic-investor-MFA-new-session')
|
|
owner.login(credentials, mfa=True)
|
|
assert owner.identity() != prior_session
|
|
view = owner.call('GET', url)
|
|
assert view['case']['id'] == case and view['case']['title'] == before['case']['title']
|
|
if view['facts']['land_needed'] == 'unknown':
|
|
owner.call('POST', url + '/aplicabilidade', {'version': view['case']['version'], 'land_needed': 'no', 'external_funding': 'no',
|
|
'viability': 'proceed', 'reason': 'Cenário A sintético QA com MFA; sem decisão externa.'})
|
|
draft = owner.call('GET', '/api/v1/simulation-drafts/current?context=' + case)['data']
|
|
if not draft:
|
|
financial = {'version': 1, 'data': {'budgetPeriod': 'Mensal', 'loan': {'principal': '100000', 'rate': '0', 'months': '24', 'rateType': 'nominal', 'periodMonths': '1', 'gracePeriods': '0', 'upfrontFee': '0', 'recurringFee': '0'},
|
|
'breakEven': {'fixed': '1000', 'price': '100', 'variable': '50'}, 'funding': {'total': '100000', 'secured': '100000'},
|
|
'incomes': [{'label': 'Receita sintética QA', 'amount': '100'}], 'expenses': [{'label': 'Despesa sintética QA', 'amount': '50'}]}}
|
|
draft = owner.call('POST', '/api/v1/simulation-drafts', {'context': case, 'personal_snapshot': {'fiscal_residency_country': 'PT'},
|
|
'economic_snapshot': {'purpose': 'simulate_only', 'project_title': 'A MFA QA', 'project_summary': 'Descrição sintética A MFA QA', 'currency': 'MZN',
|
|
'funds_origin_countries': ['PT', 'ZA'], 'financial_tools_snapshot': json.dumps(financial)}}, status=201, extra={'Idempotency-Key': str(uuid.uuid4())})['data']
|
|
view = owner.call('GET', url)
|
|
result = owner.call('POST', url + '/resultados', {'version': view['case']['version'], 'draft_id': draft['id'], 'draft_version': draft['version_no']})['result']
|
|
view = owner.call('GET', url)
|
|
if not view['requests']:
|
|
owner.call('POST', '/balcao-virtual/dossiers/' + case + '/pedidos', {'service_version_id': view['services'][0]['id'], 'client_key': str(uuid.uuid4())}, status=201)
|
|
view = owner.call('GET', url)
|
|
if not view['tasks']:
|
|
owner.call('POST', url + '/tarefas', {'version': view['case']['version'], 'code': 'S02', 'title': 'Mercado independente A MFA QA',
|
|
'due_at': (datetime.now(timezone.utc) + timedelta(days=7)).isoformat(), 'consent': True}, status=201)
|
|
task = owner.call('GET', url)['tasks'][0]
|
|
manager = Session('manager-QA')
|
|
manager.login({'email': 'opportunity-manager@example.invalid', 'password': 'Opportunity-QA-2026!'})
|
|
installation = manager.call('GET', '/management/balcao-install')
|
|
assert manager.call('GET', url)['results'] == []
|
|
manager.call('GET', '/api/v1/simulation-drafts/' + draft['id'], status=404)
|
|
for state in ['in_progress', 'done']:
|
|
task = next(t for t in manager.call('GET', url)['tasks'] if t['id'] == task['id'])
|
|
if task['status'] != 'done':
|
|
manager.call('POST', '/balcao-virtual/dossiers/' + case + '/tarefas/' + task['id'], {'version': task['version_no'], 'status': state,
|
|
'reason': 'Acompanhamento interno sintético A com MFA.'})
|
|
owner.call('POST', '/logout', status=204)
|
|
owner = Session('synthetic-investor-MFA-final-session')
|
|
owner.login(credentials, mfa=True)
|
|
view = owner.call('GET', url)
|
|
recovered = next(r for r in view['results'] if r['id'] == result['id'])
|
|
assert recovered['draft_id'] == draft['id'] and view['tasks'][0]['status'] == 'done' and view['requests'][0]['status'] == 'draft'
|
|
steps = {s['code']: s for s in view['steps']}
|
|
assert steps['S10']['state'] == 'dispensed' and steps['S13']['applicability'] == 'not_needed_for_preparation'
|
|
old_ab = json.loads((private / 'investor-path-http-qa.json').read_text(encoding='utf-8-sig'))['scenarios']
|
|
contexts = owner.call('GET', '/investimentos')['contexts']
|
|
assert all(c['id'] not in {entry['id'] for entry in old_ab} for c in contexts)
|
|
for entry in old_ab:
|
|
owner.call('GET', entry['url'], status=403)
|
|
owner.call('GET', '/api/v1/simulation-drafts/' + entry['draft_id'], status=404)
|
|
owner.call('POST', '/logout', status=204) # No completed test session left logged in.
|
|
|
|
negative = json.loads((private / 'mfa-negative-qa-private.json').read_text(encoding='utf-8-sig'))
|
|
message = 'Não foi possível verificar a autenticação desta conta. Inicie sessão novamente ou contacte o suporte.'
|
|
for field, value in [('code', '123456'), ('recovery_code', 'synthetic-only')]:
|
|
rejected = Session('negative-wrong-key-' + field)
|
|
assert rejected.login(negative)['two_factor'] is True
|
|
old = rejected.identity()
|
|
response = rejected.call('POST', '/two-factor-challenge', {field: value}, status=422)
|
|
assert response['message'] == message and field in response['errors']
|
|
assert rejected.identity() != old
|
|
rejected.call('GET', '/investimentos', status=401)
|
|
rejected.call('POST', '/two-factor-challenge', {field: value}, status=422) # Pending challenge was removed.
|
|
rejected = Session('negative-browser-response')
|
|
assert rejected.login(negative)['two_factor'] is True
|
|
page = rejected.call('POST', '/two-factor-challenge', {'code': '123456'}, browser=True)
|
|
match = re.search(rb'<script[^>]*data-page="[^"]+"[^>]*>(.*?)</script>', page, re.DOTALL)
|
|
assert match
|
|
props = json.loads(match.group(1).decode())['props']
|
|
assert props['errors']['email'] == message
|
|
rejected.call('GET', '/investimentos', status=401)
|
|
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'base_url': BASE, 'actor_id': credentials['actor_id'],
|
|
'code_hash': installation['code_hash'], 'auth_code_sha256': hashlib.sha256((root / 'bootstrap/app.php').read_bytes()).hexdigest(),
|
|
'qa_environment_sha256': hashlib.sha256((root / 'tools/portal-qa-environment.php').read_bytes()).hexdigest(),
|
|
'enrollment': 'Fortify confirm-password, enable, secret-key, confirm; eight recovery codes; no direct secret injection',
|
|
'stable_key_matches_cli_web': True, 'config_cache_bypassed': True, 'login_mfa_logout_new_session_passed': True,
|
|
'other_accounts_AB_and_drafts_denied': True,
|
|
'decryption_failures_controlled_without_authentication': ['secret', 'recovery_codes', 'browser_redirect_message'],
|
|
'scenario_A': {'id': case, 'url': url, 'draft_id': draft['id'], 'result_id': result['id'], 'new_mfa_session_recovered': True,
|
|
'tool_balcao_internal_followup_passed': True, 'recovered_financial_result': {k: recovered[k] for k in ['id', 'engine', 'financial_snapshot', 'draft_id', 'draft_version']}},
|
|
'requests': events, 'request_headers_and_bodies_exported': False, 'interface_observed': False, 'active_writes': 0, 'candidate_fixed': False}
|
|
path = private / 'mfa-http-qa.json'
|
|
if path.exists():
|
|
path.rename(private / ('mfa-http-qa-' + datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') + '.json'))
|
|
path.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
print(json.dumps({'target': target, 'actor_id': credentials['actor_id'], 'login_mfa_logout_new_session_passed': True, 'controlled_failure_without_authentication': True,
|
|
'scenario_A_url': BASE + url, 'interface_observed': False, 'active_writes': 0}))
|