81 lines
4.8 KiB
Python
81 lines
4.8 KiB
Python
"""Authenticated requests to the guarded QA server; no UI acceptance or active writes."""
|
|
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, timezone
|
|
from pathlib import Path
|
|
|
|
root = Path(__file__).resolve().parent.parent
|
|
private = root / 'storage/app/private/investmoz-v2/portal-audit'
|
|
backup = json.loads((private.parent / 'opportunity-resources-backup-proof.json').read_text(encoding='utf-8-sig'))
|
|
target = backup['restored_target']
|
|
assert backup['restore_verified'] and re.fullmatch(r'desk_restore_\d{8}_\d{6}_test', target['database']) and int(target['port']) == 55441
|
|
expected = target['database'] + ':55441'
|
|
events = []
|
|
jar = http.cookiejar.CookieJar()
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
|
|
|
|
|
def call(method, path, data=None, status=200, html=False):
|
|
assert path.startswith('/') and not path.startswith('//')
|
|
headers = {'Accept': 'text/html' if html else 'application/json', 'Content-Type': 'application/json'}
|
|
for cookie in jar:
|
|
if cookie.name == 'XSRF-TOKEN':
|
|
headers['X-XSRF-TOKEN'] = urllib.parse.unquote(cookie.value)
|
|
request = urllib.request.Request('http://127.0.0.1:18097' + path, headers=headers, data=None if data is None else json.dumps(data).encode(), method=method)
|
|
try:
|
|
response = opener.open(request)
|
|
except urllib.error.HTTPError as e:
|
|
response = e
|
|
assert response.headers.get('X-Investmoz-QA-Target') == expected
|
|
content = response.read()
|
|
assert response.status == status, f'{method} {path}: expected {status}, observed {response.status}; response body withheld'
|
|
events.append({'method': method, 'url': path, 'status': response.status, 'qa_target_verified': True})
|
|
return None if html or not content else json.loads(content)
|
|
|
|
|
|
call('GET', '/login', html=True)
|
|
credentials = json.loads((private / 'i0-qa-actor-private.json').read_text(encoding='utf-8-sig'))
|
|
login = call('POST', '/login', {k: credentials[k] for k in ['email', 'password']})
|
|
if login.get('two_factor'):
|
|
secret = base64.b32decode(credentials['totp_secret'])
|
|
digest = hmac.new(secret, struct.pack('>Q', int(time.time()) // 30), hashlib.sha1).digest()
|
|
offset = digest[-1] & 15
|
|
code = f'{(struct.unpack(">I", digest[offset:offset + 4])[0] & 0x7fffffff) % 1000000:06d}'
|
|
call('POST', '/two-factor-challenge', {'code': code}, status=204)
|
|
plans = []
|
|
results = []
|
|
for level in [1, 2, 3]:
|
|
plan = call('GET', f'/management/territory-geometry-import?level={level}')
|
|
assert [plan['expected'], plan['resolved'], plan['pending']] == {1: [11, 9, 2], 2: [159, 120, 39], 3: [411, 232, 179]}[level]
|
|
plans.append(plan)
|
|
data = {'level': level, 'hash': plan['hash'], 'idempotency_key': str(uuid.uuid4()), 'reason': 'Reutilização autenticada do exemplar oficial existente em QA', 'confirm': True}
|
|
result = call('POST', '/management/territory-geometry-import', data)
|
|
assert result['inserted'] == 0 and result['reused'] == plan['resolved'] and not result['published']
|
|
assert call('POST', '/management/territory-geometry-import', data) == result
|
|
reverse = {'idempotency_key': str(uuid.uuid4()), 'reason': 'Reversão do lote sem remover geometrias anteriormente existentes', 'confirm': True}
|
|
r = call('POST', '/management/territory-geometry-import/' + result['batch_id'] + '/reverse', reverse)
|
|
assert r['reversed'] == 0 and r['reused_preserved'] == plan['resolved']
|
|
results.append({'apply': result, 'reverse': r})
|
|
call('GET', '/management/territory-geometry-import?level=1', html=True)
|
|
call('POST', '/logout', status=204)
|
|
jar.clear()
|
|
call('GET', '/login', html=True)
|
|
call('POST', '/login', {'email': 'investor-path-outsider-qa@example.invalid', 'password': 'Outsider-Path-QA-2026!'})
|
|
call('GET', '/management/territory-geometry-import?level=1', status=403)
|
|
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'mfa_login_completed': bool(login.get('two_factor')), 'plans': plans, 'results': results, 'requests': events, 'api_persistence_passed': True, 'interface_observed': False, 'active_writes': 0, 'published': False}
|
|
destination = private / 'geometry-auth-http-qa.json'
|
|
if destination.exists():
|
|
destination.rename(private / ('geometry-auth-http-qa-' + datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') + '.json'))
|
|
destination.write_text(json.dumps(proof, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
print(json.dumps({'target': target, 'mfa_login_completed': proof['mfa_login_completed'], 'levels': [{'level': p['level'], 'expected': p['expected'], 'resolved': p['resolved'], 'pending': p['pending']} for p in plans], 'inserted': 0, 'reused': 361, 'reversed': 0, 'published': False, 'interface_observed': False, 'active_writes': 0}))
|