"""Real upload/scans/derivatives/download through the QA web PHP process; no visual claim.""" import base64 import hashlib import hmac import time import http.cookiejar import json import re import sys import struct import subprocess import urllib.error import urllib.parse import urllib.request import uuid import zlib 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' jar = http.cookiejar.CookieJar() opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) events = [] php = Path.home() / '.config/investmoz/php-8.5.11/php.exe' asset_version = subprocess.check_output([str(php), '-r', "echo hash_file('xxh128','public/build-qa-opportunities/manifest.json');"], cwd=root, text=True).strip() def call(method, path, data=None, status=200, binary=False, upload=None, inertia=False, 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) if inertia: headers['X-Inertia'] = 'true' headers['X-Inertia-Version'] = asset_version body = None if data is None else json.dumps(data).encode() if upload: boundary = 'QA-' + uuid.uuid4().hex pieces = [] for key, value in data.items(): pieces.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n'.encode() + (json.dumps(value).encode() if isinstance(value, dict) else str(value).encode()) + b'\r\n') name, mime, content = upload pieces.append(f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\nContent-Type: {mime}\r\n\r\n'.encode() + content + b'\r\n') body = b''.join(pieces) + f'--{boundary}--\r\n'.encode() headers['Content-Type'] = 'multipart/form-data; boundary=' + boundary req = urllib.request.Request('http://127.0.0.1:18097' + path, headers=headers, data=body, method=method) try: response = opener.open(req) except urllib.error.HTTPError as e: response = e assert response.headers.get('X-Investmoz-QA-Target') == expected raw = response.read() assert response.status == status, f'{method} {path}: expected {status}, observed {response.status}; response body withheld' events.append({'method': method, 'url': path.split('?')[0], 'status': response.status, 'qa_target_verified': True}) return (raw, response.headers) if binary else None if html or not raw else json.loads(raw) actors = json.loads((private / 'resource-http-actors-private.json').read_text(encoding='utf-8-sig')) def login(role): credentials = actors[role] result = call('POST', '/login', {k: credentials[k] for k in ['email', 'password']}) if result.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) call('GET', '/login', html=True) login('manager') installation = call('GET', '/management/opportunity-resources-install') assert installation['installed'] and installation['imageProcessingAvailable'] if '--verify-only' in sys.argv: retained = json.loads((private / 'resources-real-http-qa.json').read_text(encoding='utf-8-sig')) assert retained['target'] == target and retained['valid_pdf_only_increment'] and retained['original_pdf_strictly_parseable'] base = f"/management/opportunities/{retained['opportunity_id']}/resources" resource_props = call('GET', base, inertia=True)['props'] previous = next(json.loads(p.read_text(encoding='utf-8-sig')) for p in private.glob('resources-real-http-qa-*.json') if json.loads(p.read_text(encoding='utf-8-sig'))['document_resource_id'] == retained['previous_document_resource_id']) old_pdf, _ = call('GET', base + '/documents/' + previous['document_resource_id'] + '/download', binary=True) old_pdf_sha = previous.get('original_pdf_sha256', hashlib.sha256(b'%PDF-1.7\nSynthetic web QA only\n%%EOF').hexdigest()) assert hashlib.sha256(old_pdf).hexdigest() == old_pdf_sha old_image, _ = call('GET', base + '/images/' + previous['image_resource_id'] + '/download', binary=True) assert hashlib.sha256(old_image).hexdigest() == previous['original_image_sha256'] image_record = next(r for r in resource_props['resources']['images'] if r['id'] == previous['image_resource_id']) old_derived = {d['document_id']: d['sha256'] for d in previous['derivatives']} for derivative in image_record['derivativePreviews']: url = urllib.parse.urlsplit(derivative['url']) assert url.hostname == '127.0.0.1' and url.port == 18097 content, _ = call('GET', url.path + '?' + url.query, binary=True) assert hashlib.sha256(content).hexdigest() == old_derived[derivative['document_id']] new_pdf, _ = call('GET', base + '/documents/' + retained['document_resource_id'] + '/download', binary=True) assert hashlib.sha256(new_pdf).hexdigest() == retained['original_pdf_sha256'] sys.path.insert(0, str(private / 'pdf-proof-tooling')) import io import pypdf reader = pypdf.PdfReader(io.BytesIO(new_pdf), strict=True) assert len(reader.pages) == 1 and 'no official or legal value' in reader.pages[0].extract_text() call('POST', '/logout', status=204) proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'previous_binaries_preserved': 5, 'new_valid_pdf_resource_id': retained['document_resource_id'], 'new_valid_pdf_sha256': retained['original_pdf_sha256'], 'new_pdf_pages': 1, 'new_pdf_strict_parse': True, 'resources_created': 0, 'published': False, 'requests': events, 'interface_observed': False, 'active_writes': 0} (private / 'resource-originals-preservation-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 not in ['requests', 'new_valid_pdf_resource_id', 'new_valid_pdf_sha256']})) sys.exit(0) valid_pdf_only = '--valid-pdf-only' in sys.argv retained = json.loads((private / 'resources-real-http-qa.json').read_text(encoding='utf-8-sig')) if valid_pdf_only else None if retained: assert retained['target'] == target and not retained['published'] id = retained['opportunity_id'] else: draft = call('POST', '/management/opportunities/drafts', {'creation_key': str(uuid.uuid4()), 'reason': 'Ensaio de operações documentais web com ClamD real em QA'}, status=201) id = draft['id'] base = f'/management/opportunities/{id}/resources' def props(): return call('GET', base, inertia=True)['props'] def input(payload): return {'etag': props()['etag'], 'creation_key': str(uuid.uuid4()), 'payload': payload, 'reason': 'Ensaio real de gravação e recuperação de recursos privados QA'} pdf_proof = json.loads((private / 'document-types-fixture-qa.json').read_text(encoding='utf-8-sig')) document_bytes = (private / 'synthetic-valid-original.pdf').read_bytes() assert hashlib.sha256(document_bytes).hexdigest() == pdf_proof['fixtures'][0]['sha256'] and pdf_proof['fixtures'][0]['strict_parse'] document = call('POST', base + '/documents', input({'title': 'Synthetic QA PDF - valid one page, no legal value'}), upload=('qa-valid-one-page.pdf', 'application/pdf', document_bytes), status=201) document_record = next(r for r in props()['resources']['documents'] if r['id'] == document['id']) assert document_record['file']['malware_scan_status'] == 'clean' and document_record['file']['malware_engine'] == 'ClamAV' chunk = lambda name, content: struct.pack('>I', len(content)) + name + content + struct.pack('>I', zlib.crc32(name + content)) png = b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', struct.pack('>IIBBBBB', 1600, 1600, 8, 2, 0, 0, 0)) + chunk(b'IDAT', zlib.compress(bytes(1600 * 4801))) + chunk(b'IEND', b'') image = {'id': retained['image_resource_id']} if retained else call('POST', base + '/images', input({'title': 'Synthetic web QA image', 'caption': 'Synthetic square no active data', 'alt_text': 'Synthetic square for QA only', 'rights_status': 'owned'}), upload=('qa.png', 'image/png', png), status=201) original = next(r for r in props()['resources']['images'] if r['id'] == image['id']) assert original['file']['malware_scan_status'] == 'clean' and original['file']['malware_engine'] == 'ClamAV' if not valid_pdf_only: call('POST', base + '/images/' + image['id'] + '/prepare', {'etag': props()['etag'], 'version': original['version'], 'reason': 'Preparar três derivadas na execução PHP web QA'}) p = props() prepared = next(r for r in p['resources']['images'] if r['id'] == image['id']) assert len(prepared['derivativePreviews']) == 3 assert [r['width'] for r in prepared['derivativePreviews']] == [320, 640, 1280] derivatives = [] for derived in prepared['derivativePreviews']: url = urllib.parse.urlsplit(derived['url']) assert url.hostname == '127.0.0.1' and url.port == 18097 webp, headers = call('GET', url.path + '?' + url.query, binary=True) assert headers['Content-Type'] == 'image/webp' and webp[:4] == b'RIFF' and webp[8:12] == b'WEBP' and b'EXIF' not in webp call('GET', url.path, status=403) derivatives.append({'document_id': derived['document_id'], 'width': derived['width'], 'sha256': hashlib.sha256(webp).hexdigest(), 'clean_signed_preview': True}) for kind, resource, expected_bytes in [('documents', document, document_bytes), ('images', image, png)]: downloaded, _ = call('GET', base + '/' + kind + '/' + resource['id'] + '/download', binary=True) assert downloaded == expected_bytes call('POST', '/logout', status=204) jar.clear() call('GET', '/login', html=True) login('outsider') call('GET', base + '/documents/' + document['id'] + '/download', status=403) call('GET', url.path + '?' + url.query, status=403) call('POST', '/logout', status=204) jar.clear() call('GET', '/login', html=True) login('manager') recovered = next(r for r in props()['resources']['images'] if r['id'] == image['id']) assert len(recovered['derivativePreviews']) == 3 and recovered['document_id'] == original['document_id'] recovered_bytes, _ = call('GET', base + '/documents/' + document['id'] + '/download', binary=True) assert recovered_bytes == document_bytes call('POST', '/logout', status=204) proof = {'captured_at': datetime.now(timezone.utc).isoformat(), 'target': target, 'code_hash': installation['code_hash'], 'asset_version': asset_version, 'opportunity_id': id, 'document_resource_id': document['id'], 'original_pdf_sha256': hashlib.sha256(document_bytes).hexdigest(), 'original_pdf_strictly_parseable': True, 'pdf_fixture_evidence': 'document-types-fixture-qa.json', 'image_resource_id': image['id'], 'original_image_document_id': original['document_id'], 'original_image_sha256': hashlib.sha256(png).hexdigest(), 'web_gd_available': True, 'original_image_engine': original['file']['malware_engine'], 'derivatives': derivatives, 'requests': events, 'new_authenticated_session_recovered': True, 'api_persistence_passed': True, 'interface_observed': False, 'published': False, 'active_writes': 0} destination = private / 'resources-real-http-qa.json' proof.update({'valid_pdf_only_increment': valid_pdf_only, 'new_derivative_scans_performed': not valid_pdf_only, 'existing_originals_replaced': False, 'original_pdf_real_scan_engine': document_record['file']['malware_engine'], 'previous_document_resource_id': retained['document_resource_id'] if retained else None}) if destination.exists(): destination.rename(private / ('resources-real-http-qa-' + datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') + '.json')) destination.write_text(json.dumps(proof, indent=2), encoding='utf-8') print(json.dumps({'target': target, 'opportunity_id': id, 'web_gd_available': True, 'real_original_scanned': True, 'signed_derivatives_downloaded': len(derivatives), 'foreign_actor_denied': True, 'interface_observed': False, 'published': False, 'active_writes': 0}))