158 lines
8.3 KiB
Python
158 lines
8.3 KiB
Python
"""Profile acceptance through the guarded QA server. No credential/OTP/header output."""
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import http.cookiejar
|
|
import json
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
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'
|
|
ACTOR_FILE = PRIVATE / 'profiles-v1-actors-private.json'
|
|
actors = json.loads(ACTOR_FILE.read_text(encoding='utf-8-sig'))
|
|
organizations = json.loads((PRIVATE / 'profiles-v1-organizations-qa.json').read_text())
|
|
baseline = json.loads((PRIVATE / 'profiles-v1-before-qa.json').read_text())
|
|
target = baseline['target']
|
|
assert target['database'].endswith('_test') and int(target['port']) == 55441
|
|
BASE = 'http://127.0.0.1:18097'
|
|
EXPECTED = f"{target['database']}:55441"
|
|
PHP = 'C:/Users/macbook/.config/investmoz/php-8.5.11/php.exe'
|
|
version = subprocess.check_output([PHP, '-r', "echo hash_file('xxh128', 'public/build-qa-opportunities/manifest.json');"], cwd=ROOT, text=True).strip()
|
|
events = []
|
|
receipt_path = PRIVATE / 'profiles-v1-http-receipts-qa.json'
|
|
receipts = json.loads(receipt_path.read_text()) if receipt_path.exists() else {}
|
|
REASON = 'Ensaio sintético QA de perfis, representação e segregação de funções'
|
|
|
|
|
|
def save():
|
|
receipt_path.write_text(json.dumps(receipts, indent=2), encoding='utf-8')
|
|
|
|
|
|
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}'
|
|
|
|
|
|
class QAOnlyRedirect(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
|
assert newurl.startswith(BASE + '/'), 'Non-QA redirect refused'
|
|
return super().redirect_request(req, fp, code, msg, headers, newurl)
|
|
|
|
|
|
class Session:
|
|
def __init__(self, role):
|
|
self.role = role
|
|
self.jar = http.cookiejar.CookieJar()
|
|
self.opener = urllib.request.build_opener(QAOnlyRedirect(), urllib.request.HTTPCookieProcessor(self.jar))
|
|
self.checked = False
|
|
|
|
def call(self, method, path, data=None, status=200, inertia=False, binary=False, multipart=None):
|
|
assert path.startswith('/') and not path.startswith('//')
|
|
assert method == 'GET' or self.checked, 'Read-only QA probe required before writes'
|
|
headers = {'Accept': 'application/json'}
|
|
for cookie in self.jar:
|
|
if cookie.name == 'XSRF-TOKEN':
|
|
headers['X-XSRF-TOKEN'] = urllib.parse.unquote(cookie.value)
|
|
if inertia:
|
|
headers.update({'X-Inertia': 'true', 'X-Inertia-Version': version})
|
|
body = None
|
|
if multipart:
|
|
boundary = 'qa' + uuid.uuid4().hex
|
|
parts = []
|
|
for key, value in data.items():
|
|
parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n'.encode())
|
|
filename, mime, file_bytes = multipart if isinstance(multipart, tuple) else ('synthetic-qa-proof.txt', 'text/plain', multipart)
|
|
assert '\r' not in filename and '\n' not in filename and '"' not in filename
|
|
parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{filename}"\r\nContent-Type: {mime}\r\n\r\n'.encode() + file_bytes + b'\r\n')
|
|
parts.append(f'--{boundary}--\r\n'.encode())
|
|
body = b''.join(parts)
|
|
headers['Content-Type'] = 'multipart/form-data; boundary=' + boundary
|
|
elif data is not None:
|
|
body = json.dumps(data).encode()
|
|
headers['Content-Type'] = 'application/json'
|
|
request = urllib.request.Request(BASE + path, data=body, headers=headers, method=method)
|
|
try:
|
|
response = self.opener.open(request)
|
|
except urllib.error.HTTPError as error:
|
|
response = error
|
|
assert response.headers.get('X-Investmoz-QA-Target') == EXPECTED, 'Actual QA target mismatch'
|
|
self.checked = True
|
|
raw = response.read()
|
|
events.append({'actor_id': actors[self.role]['id'], 'method': method, 'path': path.split('?')[0], 'status': response.status})
|
|
if response.status != status:
|
|
diagnostic = {'actor_id': actors[self.role]['id'], 'method': method, 'path': path, 'expected': status, 'observed': response.status}
|
|
try:
|
|
diagnostic['error_fields'] = list(json.loads(raw).get('errors', {}))
|
|
except Exception:
|
|
pass
|
|
(PRIVATE / 'profiles-v1-last-failure-qa.json').write_text(json.dumps(diagnostic), encoding='utf-8')
|
|
raise AssertionError(f'{method} {path}: expected {status}, observed {response.status}; response body withheld')
|
|
if binary:
|
|
return raw
|
|
return json.loads(raw) if raw else None
|
|
|
|
def login(self):
|
|
self.call('GET', '/login', binary=True)
|
|
account = actors[self.role]
|
|
result = self.call('POST', '/login', {'email': account['email'], 'password': account['password']})
|
|
if result.get('two_factor'):
|
|
self.call('POST', '/two-factor-challenge', {'code': otp(account['totp_secret'])}, status=204)
|
|
return self
|
|
|
|
def enroll(self):
|
|
self.login()
|
|
if not actors[self.role].get('enrolled'):
|
|
self.call('POST', '/user/confirm-password', {'password': actors[self.role]['password']}, status=201)
|
|
if not actors[self.role].get('totp_secret'):
|
|
self.call('POST', '/user/two-factor-authentication')
|
|
actors[self.role]['totp_secret'] = self.call('GET', '/user/two-factor-secret-key')['secretKey']
|
|
ACTOR_FILE.write_text(json.dumps(actors, indent=2), encoding='utf-8')
|
|
self.call('POST', '/user/confirmed-two-factor-authentication', {'code': otp(actors[self.role]['totp_secret'])})
|
|
actors[self.role]['enrolled'] = True
|
|
ACTOR_FILE.write_text(json.dumps(actors, indent=2), encoding='utf-8')
|
|
self.call('POST', '/logout', {}, status=204)
|
|
|
|
|
|
def promoter_request(role, organization):
|
|
if role in receipts.get('requests', {}):
|
|
return receipts['requests'][role]
|
|
s = Session(role).login()
|
|
case = s.call('POST', '/balcao-virtual/dossiers', {'title': 'Synthetic QA profile proof ' + role, 'client_key': str(uuid.uuid4()), 'reuse_consent': True}, status=201)
|
|
raw = ('Synthetic QA-only proof. Not a real identity or legal document. ' + role + ' ' + str(uuid.uuid4())).encode()
|
|
s.call('POST', f"/balcao-virtual/dossiers/{case['id']}/documentos", {'version': case['version'], 'category': 'profile-proof-qa', 'source_reference': 'Synthetic QA, not official, not for publication', 'consent': '1'}, multipart=raw)
|
|
details = s.call('GET', f"/balcao-virtual/dossiers/{case['id']}")
|
|
document = details['documents'][-1]
|
|
assert document['malware_scan_status'] == 'clean' and document['sha256'].strip() == hashlib.sha256(raw).hexdigest(), 'Real scan not completed'
|
|
payload = {'creation_key': str(uuid.uuid4()), 'representation_type': 'company' if organization else 'individual', 'organization_id': organization,
|
|
'opportunity_types': ['investment_project'], 'document_ids': [document['document_id']], 'reason': REASON}
|
|
request = s.call('POST', '/perfil/promotor', payload, status=201)
|
|
replay = s.call('POST', '/perfil/promotor', payload, status=201)
|
|
assert replay['id'] == request['id']
|
|
new = Session(role).login()
|
|
recovered = new.call('GET', '/perfil/promotor/' + request['id'])
|
|
assert recovered['status'] == 'pending'
|
|
receipts.setdefault('requests', {})[role] = {'id': request['id'], 'document_id': document['document_id'], 'document_sha256': hashlib.sha256(raw).hexdigest(), 'case_id': case['id'], 'approved': False}
|
|
save()
|
|
return receipts['requests'][role]
|
|
|
|
|
|
def grant(role, granted_role):
|
|
key = role + ':' + granted_role
|
|
if key not in receipts.get('grants', {}):
|
|
s = Session('super_first').login()
|
|
result = s.call('POST', '/perfis/atribuicoes', {'user_id': actors[role]['id'], 'role': granted_role, 'opportunity_types': ['investment_project'], 'reason': REASON, 'responsibilities': []}, status=201)
|
|
receipts.setdefault('grants', {})[key] = result['id']
|
|
save()
|
|
|
|
|