188 lines
12 KiB
Python
188 lines
12 KiB
Python
"""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}))
|