Files
Investidor-V2/tools/portal-profiles-http-qa.py
nala 01405cbf8a
tests / ci (push) Canceled after 0s
oportunidades 5
2026-10-10 14:15:28 +02:00

365 lines
24 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())
parts.append(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="synthetic-qa-proof.txt"\r\nContent-Type: text/plain\r\n\r\n'.encode() + multipart + 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()
if '--verify-current-only' in sys.argv:
draft = receipts['draft']['id']
promoter = Session('promoter').login()
nav = promoter.call('GET', '/dashboard', inertia=True)['props']['profileAccess']
assert nav['mine'] == '/minhas-oportunidades' and nav['create'] == '/minhas-oportunidades/create'
promoter.call('GET', nav['mine'], inertia=True)
promoter.call('GET', nav['create'], inertia=True)
promoter.call('GET', '/management/territory-options?type=province', status=403)
provinces = promoter.call('GET', '/minhas-oportunidades/territory-options?type=province')['data']
assert len(provinces) == 11
province = provinces[0]
geometry = promoter.call('GET', '/minhas-oportunidades/territory-geometries?territory_id=' + province['id'])
assert geometry['selected_geometry_available']
before = promoter.call('GET', f'/minhas-oportunidades/{draft}/edit', inertia=True)['props']
promoter.call('PUT', f'/minhas-oportunidades/{draft}/draft', {'etag': before['etag'], 'geography_scope': 'specific', 'province_id': province['id'], 'reason': REASON})
recovered = Session('promoter').login().call('GET', f'/minhas-oportunidades/{draft}/edit', inertia=True)['props']
assert recovered['territoryPath'][0]['id'] == province['id'] and recovered['territoryPath'][0]['name'] == province['name']
colleague = Session('colleague').login()
colleague.call('GET', f'/minhas-oportunidades/{draft}/edit', inertia=True)
colleague.call('GET', f'/minhas-oportunidades/{draft}/resources', inertia=True)
Session('outsider').login().call('GET', f'/minhas-oportunidades/{draft}/view', inertia=True, status=403)
public = Session('public').login()
public.call('GET', '/minhas-oportunidades/territory-options?type=province', status=403)
Session('validator').login().call('GET', '/minhas-oportunidades', inertia=True, status=403)
Session('publisher').login().call('GET', f'/minhas-oportunidades/{draft}/view', inertia=True)
auditor = Session('auditor').login()
auditor.call('GET', '/perfis/auditoria?action=superuser_replaced', inertia=True)
auditor.call('GET', '/perfis/auditoria?action=superuser_replaced&format=json')
report = auditor.call('GET', '/qa-auditoria/pedidos-abertos', binary=True).decode('utf-8')
projection = json.loads((PRIVATE / 'pedidos-abertos-projection.json').read_text(encoding='utf-8'))
assert report.count('<article data-group=') == projection['open_requirements']
assert 'AUD-PROFILE-32' in report and '3,386 requisitos/linhas' in report
for name in ['relatorio-executivo', 'backlog', 'perfis-v1-reconciliacao', 'roteiro-manual-perfis-v1']:
auditor.call('GET', '/qa-auditoria/' + name, binary=True)
public.call('GET', '/qa-auditoria/backlog', status=403)
multiple = Session('multiple').login()
multiple.call('PUT', f'/minhas-oportunidades/{draft}/draft', {'etag': recovered['etag'], 'title': 'forbidden', 'reason': REASON}, status=403)
holder = Session('replacement').login()
admin = holder.call('GET', '/perfis/gestao', inertia=True)['props']
assert {admin['pair']['first_user_id'], admin['pair']['second_user_id']} == {actors['super_first']['id'], actors['replacement']['id']}
assert admin['actorId'] == actors['replacement']['id'] and admin['types'] and admin['securityRequests']
holder.call('DELETE', '/user/two-factor-authentication', {}, status=403)
Session('super_second').login().call('GET', '/perfis/gestao', inertia=True, status=403)
holder.call('POST', '/logout', {}, status=204)
holder.call('GET', '/perfis/gestao', status=401)
Session('replacement').login().call('GET', '/perfis/gestao', inertia=True)
manifest = ROOT / 'public/build-qa-opportunities/manifest.json'
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'base_url': BASE,
'opportunity_id': draft, 'normal_login_mfa_logout_new_session': True, 'dashboard_navigation_followed': True,
'organization_colleague_resource_access': True, 'other_organization_denied': True, 'auditor_multiple_role_write_denied': True,
'province_count': len(provinces), 'selected_territory_id': province['id'], 'selected_territory_name': province['name'],
'geometry_available': True, 'geometry_api_area_count': len(geometry['areas']), 'selected_field_recovered_new_session': True,
'revoked_validator_and_replaced_superuser_denied': True, 'administration_forms_props_served': True,
'same_matrix_projection_open_card_count': projection['open_requirements'], 'executive_backlog_manual_authenticated_qa': True,
'qa_manifest_sha256': hashlib.sha256(manifest.read_bytes()).hexdigest(), 'browser_visual_observed': False,
'active_writes': 0, 'requests': events}
(PRIVATE / 'profiles-v1-current-version-qa.json').write_text(json.dumps(proof, indent=2))
print(json.dumps({k:v for k,v in proof.items() if k != 'requests'}))
sys.exit(0)
if '--verify-recovery-only' in sys.argv:
holder = Session('replacement').login()
holder.call('POST', '/user/confirm-password', {'password': actors['replacement']['password']}, status=201)
codes = holder.call('GET', '/user/two-factor-recovery-codes')
assert len(codes) == 8
recovery = codes[0]
holder.call('POST', '/logout', {}, status=204)
restored = Session('replacement')
restored.call('GET', '/login', binary=True)
result = restored.call('POST', '/login', {'email': actors['replacement']['email'], 'password': actors['replacement']['password']})
assert result['two_factor']
restored.call('POST', '/two-factor-challenge', {'recovery_code': recovery}, status=204)
administration = restored.call('GET', '/perfis/gestao', inertia=True)['props']
assert actors['replacement']['id'] in [administration['pair']['first_user_id'], administration['pair']['second_user_id']]
assert administration['securityAlerts']
restored.call('POST', '/logout', {}, status=204)
reused = Session('replacement')
reused.call('GET', '/login', binary=True)
reused.call('POST', '/login', {'email': actors['replacement']['email'], 'password': actors['replacement']['password']})
reused.call('POST', '/two-factor-challenge', {'recovery_code': recovery}, status=422)
reused.call('GET', '/perfis/gestao', status=401)
auditor = Session('auditor').login()
audit = auditor.call('GET', '/perfis/auditoria?action=superuser_replaced', inertia=True)['props']
assert audit['records']['total'] == 1
metadata = auditor.call('GET', '/perfis/auditoria?action=superuser_replaced&format=json')
assert len(metadata['records']) == 1 and 'after_data' not in metadata['records'][0]
Session('public').login().call('GET', '/perfis/auditoria', status=403)
protected = Session('replacement').login()
protected.call('DELETE', '/user/two-factor-authentication', {}, status=403)
protected.call('DELETE', '/settings/profile', {}, status=403)
protected.call('PATCH', '/settings/profile', {'email': 'changed-qa@example.invalid'}, status=403)
Session('replacement').login().call('POST', '/perfis/atribuicoes', {'user_id': actors['public']['id'], 'role': 'auditor', 'opportunity_types': ['investment_project'], 'responsibilities': ['validation_coordinator'], 'reason': REASON}, status=422)
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'holder_id': actors['replacement']['id'], 'fortify_recovery_code_restored_mfa_session': True,
'consumed_code_reuse_rejected_without_authentication': True, 'two_holders_preserved': True, 'visible_security_alert_props': True,
'auditor_filtered_read_and_minimized_export': True, 'public_audit_access_denied': True, 'incompatible_responsibility_denied': True, 'mandatory_mfa_and_holder_identity_protected': True,
'secrets_disclosed': False, 'active_writes': 0, 'browser_visual_observed': False, 'requests': events}
(PRIVATE / 'profiles-v1-recovery-audit-qa.json').write_text(json.dumps(proof, indent=2))
print(json.dumps({k:v for k,v in proof.items() if k != 'requests'}))
sys.exit(0)
if '--enroll' in sys.argv:
for role in actors:
Session(role).enroll()
proof = {'target': target, 'actor_ids': {r: a['id'] for r, a in actors.items()}, 'normal_fortify_mfa_enrolled': all(a.get('enrolled') for a in actors.values()), 'secrets_disclosed': False, 'active_writes': 0}
(PRIVATE / 'profiles-v1-mfa-enrollment-qa.json').write_text(json.dumps(proof, indent=2))
print(json.dumps(proof))
sys.exit(0)
if not receipts.get('initialized'):
Session('bootstrap').login().call('POST', '/qa-auditoria/perfis/inicializar', {})
receipts['initialized'] = True
save()
for role, organization in [('promoter', organizations['organization']), ('colleague', organizations['organization']), ('outsider', organizations['other_organization']), ('multiple', organizations['organization'])]:
info = promoter_request(role, organization)
if not info['approved']:
s = Session('super_first').login()
proof = s.call('GET', f"/perfis/pedidos/{info['id']}/documentos/{info['document_id']}", binary=True)
assert hashlib.sha256(proof).hexdigest() == info['document_sha256']
s.call('POST', f"/perfis/pedidos/{info['id']}/decisao", {'version': 1, 'decision': 'approved', 'reason': REASON,
'verified_identity': True, 'verified_representation': True, 'verified_documents': True, 'approved_types': ['investment_project']})
info['approved'] = True
save()
for role in ['validator', 'publisher', 'auditor']:
grant(role, role)
grant('multiple', 'validator')
grant('multiple', 'auditor')
public = Session('public').login()
dashboard = public.call('GET', '/dashboard', inertia=True)['props']
assert dashboard['profileAccess']['mine'] is None and dashboard['profileAccess']['request'] == '/perfil/promotor'
public.call('POST', '/minhas-oportunidades/drafts', {'creation_key': str(uuid.uuid4()), 'opportunity_category': 'investment_project', 'reason': REASON}, status=403)
public.call('GET', '/perfil/promotor/' + receipts['requests']['promoter']['id'], status=403)
promoter = Session('promoter').login()
navigation = promoter.call('GET', '/dashboard', inertia=True)['props']['profileAccess']
assert navigation['mine'] == '/minhas-oportunidades' and navigation['create'] == '/minhas-oportunidades/create'
promoter.call('GET', navigation['mine'], inertia=True)
promoter.call('GET', navigation['create'], inertia=True)
if 'draft' not in receipts:
payload = {'creation_key': str(uuid.uuid4()), 'organization_id': organizations['organization'], 'opportunity_category': 'investment_project', 'title': 'Synthetic QA v1 organization opportunity', 'reason': REASON}
draft = promoter.call('POST', '/minhas-oportunidades/drafts', payload, status=201)
assert promoter.call('POST', '/minhas-oportunidades/drafts', payload, status=201)['id'] == draft['id']
receipts['draft'] = draft
save()
id = receipts['draft']['id']
recovered = Session('promoter').login().call('GET', f'/minhas-oportunidades/{id}/edit', inertia=True)['props']
assert recovered['record']['id'] == id
colleague = Session('colleague').login()
page = colleague.call('GET', f'/minhas-oportunidades/{id}/edit', inertia=True)['props']
updated = colleague.call('PUT', f'/minhas-oportunidades/{id}/draft', {'etag': page['etag'], 'title': 'Synthetic QA edited by authorized organization colleague', 'reason': REASON})
colleague.call('PUT', f'/minhas-oportunidades/{id}/draft', {'etag': updated['etag'], 'organization_id': organizations['other_organization'], 'reason': REASON}, status=403)
colleague.call('PUT', f'/minhas-oportunidades/{id}/draft', {'etag': updated['etag'], 'opportunity_category': 'financing', 'reason': REASON}, status=403)
archive = colleague.call('POST', f'/minhas-oportunidades/{id}/workflow', {'etag': updated['etag'], 'action': 'request_archive', 'reason': REASON})
assert archive['status'] == 'draft'
colleague.call('POST', f'/minhas-oportunidades/{id}/workflow', {'etag': archive['etag'], 'action': 'submit', 'reason': REASON}, status=422)
outsider = Session('outsider').login()
outsider.call('GET', f'/minhas-oportunidades/{id}/view', status=403)
outsider.call('GET', f'/minhas-oportunidades/{id}/resources', status=403)
outsider.call('POST', '/minhas-oportunidades/drafts', {'creation_key': str(uuid.uuid4()), 'organization_id': organizations['organization'], 'opportunity_category': 'investment_project', 'reason': REASON}, status=403)
multiple = Session('multiple').login()
multiple.call('GET', f'/minhas-oportunidades/{id}/view', inertia=True)
multiple.call('PUT', f'/minhas-oportunidades/{id}/draft', {'etag': archive['etag'], 'title': 'Forbidden auditor union write', 'reason': REASON}, status=403)
validator = Session('validator').login()
validator.call('POST', f'/minhas-oportunidades/{id}/workflow', {'etag': archive['etag'], 'action': 'approve', 'reason': REASON}, status=409)
publisher = Session('publisher').login()
publisher.call('POST', f'/minhas-oportunidades/{id}/workflow', {'etag': archive['etag'], 'action': 'publish', 'reason': REASON}, status=409)
auditor = Session('auditor').login()
auditor.call('GET', f'/minhas-oportunidades/{id}/view', inertia=True)
auditor.call('POST', '/perfis/atribuicoes', {'user_id': actors['public']['id'], 'role': 'validator', 'opportunity_types': ['investment_project'], 'reason': REASON}, status=403)
superuser = Session('super_first').login()
superuser.call('POST', '/perfis/atribuicoes', {'user_id': actors['replacement']['id'], 'role': 'superuser', 'opportunity_types': ['investment_project'], 'reason': REASON}, status=422)
superuser.call('POST', '/minhas-oportunidades/drafts', {'creation_key': str(uuid.uuid4()), 'opportunity_category': 'investment_project', 'reason': REASON}, status=403)
if not receipts.get('revoked_validator'):
superuser.call('POST', '/perfis/atribuicoes/' + receipts['grants']['validator:validator'] + '/revogar', {'reason': REASON})
receipts['revoked_validator'] = True
save()
validator.call('GET', '/minhas-oportunidades', status=401)
new_validator = Session('validator').login()
new_validator.call('GET', '/minhas-oportunidades', status=403)
if 'replacement_approval' not in receipts:
replacement = superuser.call('POST', '/perfis/substituicoes', {'old_user_id': actors['super_second']['id'], 'new_user_id': actors['replacement']['id'], 'reason': REASON}, status=201)
receipts['replacement_approval'] = replacement['id']
save()
approval = receipts['replacement_approval']
if not receipts.get('replacement_applied'):
superuser.call('POST', f'/perfis/seguranca/{approval}/aprovar', {}, status=403)
superuser.call('POST', f'/perfis/substituicoes/{approval}/aplicar', {}, status=409)
second = Session('super_second').login()
second.call('POST', f'/perfis/seguranca/{approval}/aprovar', {})
superuser.call('POST', f'/perfis/substituicoes/{approval}/aplicar', {})
second.call('GET', '/perfis/gestao', status=401)
receipts['replacement_applied'] = True
save()
Session('replacement').login().call('GET', '/perfis/gestao', inertia=True)
Session('super_second').login().call('GET', '/perfis/gestao', status=403)
superuser.call('POST', f'/perfis/substituicoes/{approval}/aplicar', {}, status=409)
preservation = json.loads(subprocess.check_output([PHP, 'tools/portal-profiles-qa.php', 'verify'], cwd=ROOT, text=True))
proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'base_url': BASE, 'actor_ids': {r: a['id'] for r,a in actors.items()}, 'opportunity_id': id,
'normal_fortify_mfa': True, 'dashboard_navigation_props_discovered': True, 'promoter_requests_replayed_and_recovered_new_session': True,
'documents_scanned_real_and_recovered_by_hash': True, 'organization_colleague_edit_recovered': True, 'negative_organization_type_role_state': True,
'revocation_old_session_401_new_session_no_role': True, 'atomic_superuser_replacement_independent_two_holders': True,
'original_records_preserved': preservation['previous_rows_preserved_by_id_and_original_columns'], 'active_writes': 0,
'browser_visual_observed': False, 'normative_submission_approval_publication_pending': True, 'requests': events}
(PRIVATE / 'profiles-v1-http-qa.json').write_text(json.dumps(proof, indent=2), encoding='utf-8')
print(json.dumps({k:v for k,v in proof.items() if k != 'requests'}))