#!/usr/bin/env python3 """Replay selected product observations from a ZIP. No network or historical authentication. Usage: python3 replay.py selected-records.zip Exit0: complete, internally consistent replay (quarantines can remain). Exit2: malformed, incomplete or inconsistent selected records. The archive must contain exactly manifest.json, requests.jsonl, cells.jsonl and completion.json at its root. Raw captures and private path references are excluded. """ import argparse from collections import Counter from datetime import datetime from decimal import Decimal, InvalidOperation import hashlib import json import math from pathlib import Path import re import stat import sys from urllib.parse import urlsplit, parse_qs import zipfile FILES = {'manifest.json', 'requests.jsonl', 'cells.jsonl', 'completion.json'} ARCHIVE_LIMIT = 16 * 1024 * 1024 SAFE_HEADERS = {'content-type', 'content-encoding', 'content-length', 'etag', 'last-modified', 'cache-control', 'vary', 'date', 'age'} REDIRECTS = (301, 302, 303, 307, 308) class Invalid(ValueError): pass def need(condition, message): if not condition: raise Invalid(message) def integer(value, label, high=30 * 1024 * 1024): need(type(value) is int and 0 <= value <= high, label + ' must be a bounded nonnegative integer') return value def text(value, label, limit=2048): need(isinstance(value, str) and len(value) <= limit and all(ord(c) >= 32 and ord(c) != 127 for c in value), label + ' is invalid') return value def string_list(value, label): need(isinstance(value, list) and len(value) <= 100, label + ' must be a bounded array') for item in value: text(item, label) need(len(set(value)) == len(value), label + ' repeats a recorded value') return value def single(values): return values[0] if len(values) == 1 else None def currency(value): return isinstance(value, str) and re.fullmatch('[A-Z]{3}', value) is not None def money(value): if not isinstance(value, str): return None match = re.fullmatch(r'[^0-9.,-]*([0-9]+(?:[.,][0-9]{1,2})?)[^0-9.,-]*', value.strip()) return format(Decimal(match[1].replace(',', '.')), 'f') if match else None def amount(value): if type(value) not in (str, int) or len(str(value)) > 100: return None try: d = Decimal(value) return format(d, 'f') if d.is_finite() and 0 <= d <= Decimal('1e20') else None except InvalidOperation: return None def timestamp(value): parsed = datetime.fromisoformat(text(value, 'Timestamp', 80)) need(parsed.tzinfo is not None, 'Timestamp needs timezone') return parsed def same(actual, expected, label): # JSON true is not a substitute for the integer1 and vice versa. need(json.dumps(actual, sort_keys=True) == json.dumps(expected, sort_keys=True), label + ' disagrees with recomputed records') def strict_object(pairs): value = {} for key, item in pairs: need(key not in value, 'Duplicate JSON field') value[key] = item return value def parse_json(raw): def bad_constant(_): raise Invalid('Nonfinite JSON number') return json.loads(raw, object_pairs_hook=strict_object, parse_constant=bad_constant) def read_archive(path): need(Path(path).is_file() and Path(path).stat().st_size <= ARCHIVE_LIMIT, 'Archive missing or too large') with zipfile.ZipFile(path) as archive: entries = archive.infolist() need(len(entries) == 4 and {i.filename for i in entries} == FILES, 'Archive must contain exactly four named root files') need(sum(i.file_size for i in entries) <= ARCHIVE_LIMIT, 'Expanded archive is too large') data = {} for info in entries: need(not info.flag_bits & 1 and not info.is_dir() and not stat.S_ISLNK(info.external_attr >> 16), 'Encrypted, linked or directory entry is unsupported') with archive.open(info) as stream: raw = stream.read(ARCHIVE_LIMIT + 1) need(len(raw) == info.file_size and len(raw) <= ARCHIVE_LIMIT, 'Archive entry size is invalid') decoded = raw.decode('utf-8') data[info.filename] = ([parse_json(line) for line in decoded.splitlines() if line.strip()] if info.filename.endswith('.jsonl') else parse_json(decoded)) return data def plan_for(study): need(isinstance(study, dict) and study['schema_version'] == 1, 'Study schema invalid') need(re.fullmatch(r'[a-z0-9]+(?:[.-][a-z0-9]+)*\.[a-z]{2,}', text(study['allowed_host'], 'Host', 253)), 'Host invalid') product = study['product'] need(isinstance(product, dict) and set(product) == {'handle', 'product_id', 'variant_id', 'sku'}, 'Product fields invalid') need(re.fullmatch(r'[a-z0-9]+(?:-[a-z0-9]+)*', text(product['handle'], 'Handle', 200)), 'Handle invalid') for key in ('product_id', 'variant_id'): need(re.fullmatch(r'[1-9][0-9]{0,29}', text(product[key], key, 30)), 'Identifier invalid') text(product['sku'], 'SKU', 100) for key in ('contexts', 'entries'): rows = study[key] need(isinstance(rows, list) and 0 < len(rows) <= 8, 'Context or entry list invalid') ids = [] for row in rows: need(isinstance(row, dict), 'Context or entry row invalid') need(re.fullmatch(r'[a-z][a-z0-9_-]{0,31}', text(row['id'], 'ID', 32)), 'ID invalid') ids.append(row['id']) if key == 'contexts': need(set(row) == {'id', 'country_requested'}, 'Context fields invalid') need((row['id'] == 'direct' and row['country_requested'] is None) or (row['id'] != 'direct' and re.fullmatch('[A-Z]{2}', text(row['country_requested'], 'Country', 2))), 'Country invalid') else: need(set(row) == {'id', 'path_prefix'} and re.fullmatch(r'/(?:[a-z]{2}(?:-[a-z]{2})?/)?', text(row['path_prefix'], 'Prefix', 10)), 'Entry prefix invalid') need(len(ids) == len(set(ids)), 'Duplicate context or entry IDs') need(len({e['path_prefix'] for e in study['entries']}) == len(study['entries']), 'Duplicate entry paths') need(1 <= integer(study['repetitions'], 'Repetitions', 10), 'Repetitions invalid') need(1 <= integer(study['max_requests'], 'Request cap', 100), 'Request cap invalid') need(1 <= integer(study['max_body_bytes'], 'Body cap'), 'Body cap invalid') integer(study['max_redirects'], 'Redirect cap', 3) need(study['encoding'] in ('gzip', 'identity'), 'Encoding invalid') plan = [] for repetition in range(study['repetitions']): contexts = study['contexts'] if repetition % 2 == 0 else list(reversed(study['contexts'])) for ci, context in enumerate(contexts): entries = study['entries'] if (ci + repetition) % 2 == 0 else list(reversed(study['entries'])) for entry in entries: plan.append({'cell_id': 'r%d-%s-%s' % (repetition + 1, context['id'], entry['id']), 'repetition': repetition + 1, 'context': context, 'entry': entry}) return plan def product_url(study, prefix, kind): return 'https://' + study['allowed_host'] + prefix + 'products/' + study['product']['handle'] + ('.js' if kind == 'json' else '') def permitted(study, url, kind): return url in [product_url(study, e['path_prefix'], kind) for e in study['entries']] def offer_matches(offer, product): if offer.get('sku') is not None and str(offer['sku']) != product['sku']: return False ids = [] for key in ('url', '@id'): if isinstance(offer.get(key), str): ids.extend(parse_qs(urlsplit(offer[key]).query).get('variant', [])) if ids: return all(value == product['variant_id'] for value in ids) return str(offer.get('sku')) == product['sku'] def recompute_html(observed, product): if 'source_price_texts' not in observed: need(observed.get('state') == 'unverified' and isinstance(observed.get('reason'), str), 'Missing HTML extraction') return dict(observed) fields = ('source_price_texts', 'dom_variant_ids', 'og_price_amounts', 'og_price_currencies', 'shopify_currencies', 'shopify_countries', 'shopify_routes_roots', 'canonicals') for field in fields: string_list(observed[field], field) price_text = single(observed['source_price_texts']) price = money(price_text) og_amount = money(single(observed['og_price_amounts'])) og_currency = single(observed['og_price_currencies']) currency_known = currency(og_currency) and single(observed['shopify_currencies']) == og_currency variant_ok = single(observed['dom_variant_ids']) == product['variant_id'] amount_ok = price is not None and og_amount is not None and Decimal(price) == Decimal(og_amount) offers = observed['matching_jsonld_offers'] need(isinstance(offers, list) and len(offers) <= 30 and all(isinstance(o, dict) for o in offers), 'Selected offers invalid') need(all(offer_matches(o, product) for o in offers), 'Selected Offer lacks matching variant/SKU evidence') offer = offers[0] if len(offers) == 1 else None offer_price = amount(offer['price']) if offer else None offer_currency = offer['priceCurrency'] if offer else None offer_ok = offer_price is not None and currency(offer_currency) mismatch = bool(currency_known and offer_ok and offer_currency != og_currency) amount_mismatch = bool(price is not None and offer_price is not None and Decimal(price) != Decimal(offer_price)) reasons = [] for ok, reason in [(variant_ok, 'selected_dom_variant_missing_or_ambiguous'), (currency_known, 'storefront_currency_missing_or_conflicting'), (amount_ok, 'source_price_missing_or_og_amount_conflicting'), (offer_ok, 'selected_jsonld_offer_missing_ambiguous_or_invalid'), (not mismatch, 'jsonld_storefront_currency_disagreement'), (not amount_mismatch, 'jsonld_source_amount_disagreement')]: if not ok: reasons.append(reason) expected = {'state': 'extracted' if variant_ok and currency_known and amount_ok and offer_ok else 'unverified', 'source_price_text': price_text, 'source_price_decimal': price, 'selected_variant_matches': variant_ok, 'storefront_currency': og_currency if currency_known else None, 'shopify_routes_root': single(observed['shopify_routes_roots']), 'naive_jsonld_record': {'variant_id': product['variant_id'], 'price': offer_price, 'currency': offer_currency} if offer_ok else None, 'jsonld_currency_disagreement': mismatch, 'jsonld_amount_disagreement': amount_mismatch, 'accepted_for_price_comparison': not reasons, 'quarantine_reasons': reasons} for field, value in expected.items(): same(observed[field], value, 'HTML ' + field) return dict(observed, **expected) def recompute_json(observed, product): if 'product_id' not in observed: need(observed.get('state') == 'unverified' and isinstance(observed.get('reason'), str), 'Missing JSON extraction') return dict(observed) price = observed['price_units'] valid = (observed['product_id'] == product['product_id'] and observed['variant_id'] == product['variant_id'] and observed['sku'] == product['sku'] and type(price) is int and price >= 0) expected = {'state': 'extracted' if valid else 'unverified', 'selected_identity_matches': valid, 'accepted_as_standalone_currency_labeled_price': False} need('currency_in_payload' in observed and 'variant_currency_in_payload' in observed, 'Missing explicit JSON currency fields') explicit = [observed[k] for k in ('currency_in_payload', 'variant_currency_in_payload') if observed[k] is not None] need(all(currency(v) for v in explicit) and len(set(explicit)) <= 1, 'Explicit JSON currency fields invalid or conflicting') for field, value in expected.items(): same(observed[field], value, 'JSON ' + field) return dict(observed, **expected) def replay(data): manifest, records, cells, completion = (data[n] for n in ('manifest.json', 'requests.jsonl', 'cells.jsonl', 'completion.json')) need(isinstance(manifest, dict) and isinstance(completion, dict), 'Manifest/completion invalid') same(manifest['schema_version'], 1, 'Manifest schema') study = manifest['study'] plan = plan_for(study) same(manifest['cell_plan'], plan, 'Declared cell plan') same(manifest['hard_request_cap'], study['max_requests'], 'Request cap') same(manifest['hard_body_byte_cap'], study['max_body_bytes'], 'Body cap') same(manifest['headers'], {'accept_language': 'en-GB', 'accept_encoding': study['encoding'], 'cookies': 'absent'}, 'Fixed headers') need(isinstance(records, list) and len(records) <= study['max_requests'], 'Request count invalid') need(isinstance(cells, list) and len(cells) == len(plan), 'Planned cell coverage incomplete') ids = [c['cell_id'] for c in cells] same(ids, [p['cell_id'] for p in plan], 'Cell order/uniqueness/coverage') by_cell = {p['cell_id']: p for p in plan} last_finished = timestamp(manifest['created_at']) for index, record in enumerate(records, 1): need(isinstance(record, dict), 'Request row invalid') need(not set(record) & {'private_body', 'private_headers', 'raw_body', 'raw_headers'}, 'Private evidence references must be removed') same(record['request_id'], '%03d' % index, 'Request order/uniqueness') planned = by_cell.get(record['cell_id']) need(planned is not None, 'Request references undeclared cell') same(record['context'], planned['context']['id'], 'Request context') same(record['country_requested'], planned['context']['country_requested'], 'Request country') need(record['kind'] in ('html', 'json') and record['phase'] in ('html', 'requested_json', 'effective_json'), 'Request kind/phase invalid') need(permitted(study, record['url'], record['kind']), 'Request URL is outside declared scope') integer(record['hop'], 'Request hop', study['max_redirects']) integer(record['saved_body_bytes'], 'Saved body bytes', study['max_body_bytes']) for key, high in [('status', 599), ('connect_status', 599), ('size_download', study['max_body_bytes'])]: if record[key] is not None: integer(record[key], key, high) if record['size_download'] is not None: same(record['size_download'], record['saved_body_bytes'], 'Encoded downloaded/saved bytes') need(type(record['exitcode']) is int and -255 <= record['exitcode'] <= 255, 'Exit code invalid') elapsed = record['elapsed_seconds'] need(elapsed is None or type(elapsed) in (int, float) and math.isfinite(elapsed) and elapsed >= 0, 'Elapsed metric invalid') started, finished = timestamp(record['timestamp']), timestamp(record['finished_at']) need(started >= last_finished and finished >= started, 'Request timestamps/order invalid') last_finished = finished need(re.fullmatch('[a-f0-9]{64}', text(record['sha256'], 'Body hash', 64)), 'Body hash invalid') headers = record['headers'] need(isinstance(headers, dict) and set(headers) <= SAFE_HEADERS, 'Response headers exceed public allowlist') for value in headers.values(): text(value, 'Header') if record['redirect_url'] is not None: need(permitted(study, record['redirect_url'], record['kind']), 'Redirect outside declared scope') need(record['redirect_target_state'] in ('allowed', 'absent', 'blocked_or_ambiguous'), 'Redirect state invalid') same(record['redirect_url'] is not None, record['redirect_target_state'] == 'allowed', 'Redirect state') observed = record['inspection'] need(isinstance(observed, dict), 'Request inspection invalid') if record['exitcode'] == 0 and record['status'] == 200: if 'decoded_sha256' in observed: need(re.fullmatch('[a-f0-9]{64}', text(observed['decoded_sha256'], 'Decoded hash', 64)), 'Decoded hash invalid') integer(observed['decoded_bytes'], 'Decoded bytes', 5 * 1024 * 1024) allowed = ('text/html',) if record['kind'] == 'html' else ('application/json', 'text/javascript', 'application/javascript') need(headers.get('content-type', '').split(';')[0].strip().lower() in allowed, 'Inspected content type invalid') record['inspection'] = recompute_html(observed, study['product']) if record['kind'] == 'html' else recompute_json(observed, study['product']) else: same(observed, {'state': 'unverified', 'reason': 'transport_or_http_status'}, 'Failed HTTP inspection') saved = sum(r['saved_body_bytes'] for r in records) need(saved <= study['max_body_bytes'], 'Total body cap exceeded') same(completion['attempted_requests'], len(records), 'Completion request count') same(completion['saved_body_bytes'], saved, 'Completion body bytes') same(completion['cells_recorded'], len(cells), 'Completion cell count') same(completion['budget_reached'], len(records) >= study['max_requests'] or saved >= study['max_body_bytes'], 'Completion budget state') need(timestamp(completion['finished_at']) >= last_finished, 'Completion timestamp invalid') cursor, consumed_bytes = 0, 0 decisions = [] def fetch(stored, start_url, kind, phase, planned): nonlocal cursor, consumed_bytes need(isinstance(stored, dict) and isinstance(stored.get('request_ids'), list), 'Missing fetch/request references') ids = stored['request_ids'] need(len(ids) <= study['max_redirects'] + 1, 'Too many fetch hops') chain, seen, url = [], set(), start_url expected = None for hop in range(study['max_redirects'] + 1): if url in seen: expected = {'state': 'unverified', 'reason': 'redirect_loop', 'request_ids': chain}; break seen.add(url) if cursor >= study['max_requests'] or consumed_bytes >= study['max_body_bytes']: expected = {'state': 'unverified', 'reason': 'request_or_body_budget_exhausted', 'request_ids': chain}; break need(hop < len(ids) and cursor < len(records), 'Missing attempted request or fetch reference') row = records[cursor] same(ids[hop], row['request_id'], 'Fetch reference/order') for key, value in [('cell_id', planned['cell_id']), ('phase', phase), ('kind', kind), ('url', url), ('hop', hop)]: same(row[key], value, 'Fetch request ' + key) chain.append(row['request_id']); cursor += 1; consumed_bytes += row['saved_body_bytes'] if row['exitcode'] != 0: expected = {'state': 'unverified', 'reason': 'transport_failure', 'request_ids': chain}; break if row['status'] in REDIRECTS: if row['redirect_url'] is None: expected = {'state': 'unverified', 'reason': 'redirect_target_not_allowed', 'request_ids': chain}; break if hop == study['max_redirects']: expected = {'state': 'unverified', 'reason': 'redirect_limit', 'request_ids': chain}; break url = row['redirect_url']; continue expected = {'state': row['inspection']['state'], 'request_ids': chain, 'final_url': url, 'status': row['status'], 'inspection': row['inspection']}; break need(expected is not None, 'Incomplete fetch chain') same(stored, expected, phase + ' fetch and inspection copy') return expected for cell, planned in zip(cells, plan): prefix = planned['entry']['path_prefix'] html_url, json_url = product_url(study, prefix, 'html'), product_url(study, prefix, 'json') for key, value in [('repetition', planned['repetition']), ('context', planned['context']['id']), ('country_requested', planned['context']['country_requested']), ('requested_entry', planned['entry']['id']), ('requested_html_url', html_url), ('requested_json_url', json_url)]: same(cell[key], value, 'Cell ' + key) html = fetch(cell['html'], html_url, 'html', 'html', planned) requested = fetch(cell['requested_json'], json_url, 'json', 'requested_json', planned) observed = html.get('inspection', {}) effective_prefix = observed.get('shopify_routes_root') effective_url = product_url(study, effective_prefix, 'json') if isinstance(effective_prefix, str) else None effective_html = product_url(study, effective_prefix, 'html') if isinstance(effective_prefix, str) else None aligned = bool(effective_url and permitted(study, effective_url, 'json') and html.get('final_url') == effective_html) if not aligned: effective = {'state': 'unverified', 'reason': 'effective_routes_root_missing_or_not_aligned', 'request_ids': []} same(cell['effective_json'], effective, 'Unaligned effective JSON') elif effective_url == json_url: effective = dict(requested, reused_request_observation=True) same(cell['effective_json'], effective, 'Reused effective JSON') else: effective = fetch(cell['effective_json'], effective_url, 'json', 'effective_json', planned) facts = effective.get('inspection', {}) api_aligned = bool(aligned and effective.get('final_url') == effective_url and facts.get('state') == 'extracted') if api_aligned and currency(observed.get('storefront_currency')): for field in ('currency_in_payload', 'variant_currency_in_payload'): need(facts[field] is None or facts[field] == observed['storefront_currency'], 'Aligned JSON explicit currency contradicts storefront currency') price = observed.get('source_price_decimal') minor_equal = bool(api_aligned and price is not None and observed.get('storefront_currency') in ('EUR', 'GBP', 'USD') and Decimal(facts['price_units']) == Decimal(price) * 100) accepted = bool(observed.get('accepted_for_price_comparison') and api_aligned and minor_equal) reasons = observed.get('quarantine_reasons', ['html_unverified']) + ([] if api_aligned else ['effective_json_context_unverified']) + ([] if minor_equal else ['effective_json_amount_comparison_unverified_or_different']) expected = {'effective_html_url': html.get('final_url'), 'effective_json_url': effective_url if aligned else None, 'requested_json_matches_effective_path': aligned and effective_url == json_url, 'effective_json_context_aligned': api_aligned, 'effective_json_units_equal_source_price_x100': minor_equal, 'accepted_for_price_comparison': accepted, 'quarantine_reasons': reasons} for key, value in expected.items(): same(cell[key], value, 'Cell ' + key) decisions.append({'cell_id': planned['cell_id'], 'context': planned['context']['id'], 'html_facts_extracted': observed.get('state') == 'extracted', 'source_price_text': observed.get('source_price_text'), 'storefront_currency': observed.get('storefront_currency'), 'naive_jsonld_record': observed.get('naive_jsonld_record'), 'jsonld_currency_disagreement': observed.get('jsonld_currency_disagreement', False), **expected}) same(cursor, len(records), 'Full request reference coverage') def counts(rows): return {'cells_recorded': len(rows), 'html_facts_extracted': sum(r['html_facts_extracted'] for r in rows), 'jsonld_currency_disagreements': sum(r['jsonld_currency_disagreement'] for r in rows), 'accepted_for_price_comparison': sum(r['accepted_for_price_comparison'] for r in rows), 'quarantined_or_incomplete': sum(not r['accepted_for_price_comparison'] for r in rows)} return {'schema_version': 1, 'replay_status': 'consistent_selected_records', 'scope': 'Recomputed selected observations; does not authenticate historical requests or omitted raw bodies.', 'completion_record_present': True, 'cells_planned': len(plan), **counts(decisions), 'attempted_requests': len(records), 'http_statuses': dict(Counter(str(r['status']) for r in records)), 'transport_failures': sum(r['exitcode'] != 0 for r in records), 'non_200_responses_including_redirects': sum(r['status'] != 200 for r in records), 'missing_download_measurements': sum(r['size_download'] is None for r in records), 'measured_download_bytes': sum(r['size_download'] or 0 for r in records), 'saved_body_bytes': saved, 'contexts': [dict(context=c['id'], **counts([d for d in decisions if d['context'] == c['id']])) for c in study['contexts']], 'records': decisions, 'discrepancies': [], 'limitations': ['No network, raw-body parsing, geographic proof or historical authentication occurs during replay.', 'Recorded hashes and selected observations can be checked for consistency but not independently recovered from this archive.', 'Complete, consistently rewritten or fabricated records cannot be detected without external provenance.', 'Acceptance is a within-cell currency/variant/path consistency screen, not approval for cross-market repricing or checkout claims.', 'Missing JSON currency stays unknown; only aligned EUR/GBP/USD source amounts use the restricted units-times100 check.']} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('archive', type=Path) args = parser.parse_args() try: result = replay(read_archive(args.archive)) result['archive_sha256'] = hashlib.sha256(args.archive.read_bytes()).hexdigest() print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 except (Invalid, ValueError, TypeError, KeyError, AttributeError, OSError, EOFError, zipfile.BadZipFile, RuntimeError, RecursionError, OverflowError) as error: reason = str(error) if isinstance(error, Invalid) else 'Malformed, missing or unreadable selected-record fields' print(json.dumps({'replay_status': 'invalid', 'discrepancies': [reason], 'scope': 'Selected-record consistency only; no historical authentication.'}), file=sys.stderr) return 2 if __name__ == '__main__': sys.exit(main())