"""Independently verify field evidence against a pinned, offline ARR capture. This module never imports the extractor. Only the declarative path contract and scalar encoding are shared; business mapping remains explicitly unproved. """ from __future__ import annotations import argparse from decimal import Decimal import hashlib import json from pathlib import Path import re from . import audit_arr_day as audit from . import audit_arr_named_day as named_audit from . import collect_arr_source as capture from . import rate_info from .source_facts_contract import (BLOCKERS, CONTEXT_PATHS, CONTRACT_SHA256, FIELDS, FLAGS, MAX_FACTS_BYTES, VERSION, atom, canonical, kind, definitions) def _read(archive, name): raw = archive.read(name) capture.strict_json(raw) return json.loads(raw, parse_float=Decimal) def _provenance(archive, file, pointer, request_file=None): result = {"file": file, "sha256": archive.inventory[file]["sha256"], "pointer": pointer} if request_file: metadata = archive.document(request_file) result["request"] = {key: metadata[key] for key in ("operation_id", "method", "path", "body", "attempt")} result["request"].update(file=request_file, sha256=archive.inventory[request_file]["sha256"]) return result def _profile_id(identifiers): capture.require(type(identifiers) is list and all(type(item) is dict for item in identifiers), "facts_invalid_profile_identifiers") values = [item.get("id") for item in identifiers if item.get("type") == "Profile"] capture.require(len(values) == 1 and type(values[0]) is str and re.fullmatch(r"[A-Za-z0-9_-]{1,128}", values[0]) is not None, "facts_ambiguous_profile_identity") return values[0] def _eligible_primary_profile(detail): """Derive query eligibility from raw detail, never the stored assessment.""" if "reservationGuests" not in detail: return None guests = detail["reservationGuests"] capture.require(type(guests) is list and all(type(g) is dict for g in guests), "facts_invalid_guests") primary = [g for g in guests if g.get("primary") is True] if len(primary) != 1 or "profileInfo" not in primary[0]: return None info = primary[0]["profileInfo"] capture.require(type(info) is dict, "facts_invalid_primary_info") identity = _profile_id(info.get("profileIdList")) node = info for key in ("profile", "customer"): if key not in node: return None node = node[key] capture.require(type(node) is dict, "facts_invalid_primary_name_container") if "personName" not in node: return None names = node["personName"] capture.require(type(names) is list and all(type(n) is dict for n in names), "facts_invalid_primary_names") return identity if sum(name.get("nameType") == "Primary" for name in names) == 1 else None def _scan(archive, *, named=False): """Derive the entire record set and successful rate joins from raw requests.""" first_pass, details, rates, profiles = [], {}, {}, {} first_pass_open = True for file in archive.inventory: pattern = r"(?:request|rate|profile)-[0-9]{6}\.json" if named else r"(?:request|rate)-[0-9]{6}\.json" if not re.fullmatch(pattern, file): continue label = file[:-5] metadata = archive.document(label + ".meta.json") if metadata.get("http_status") != 200: continue request = archive.document(file) response_file = label + ".response.bin" payload = _read(archive, response_file)["data"] operation = request["operation_id"] if operation == capture.SEARCH: if first_pass_open: page = payload["reservations"] for index, value in enumerate(page["reservationInfo"]): pointer = f"/data/reservations/reservationInfo/{index}" first_pass.append((_provenance(archive, response_file, pointer, file), value)) first_pass_open = page.get("hasMore", False) elif operation == capture.DETAIL: for index, value in enumerate(payload["reservations"]["reservation"]): identity = capture.reservation_id(value) capture.require(identity not in details, "facts_duplicate_detail") pointer = f"/data/reservations/reservation/{index}" details[identity] = (_provenance(archive, response_file, pointer, file), value) elif operation == rate_info.POST: body = request["body"] capture.require(body["type"] == "Reservation" and body["detailDate"] == archive.options.rate_date and body["summaryInfo"] is False, "facts_rate_request_mismatch") identity = body["id"] capture.require(identity not in rates, "facts_duplicate_rate") rates[identity] = (_provenance(archive, response_file, "/data", file), payload) elif named and operation == "searchProfiles": body = request["body"] capture.require(type(body) is dict and set(body) == {"profileIds", "summaryInfo", "limit", "offset"} and type(body["profileIds"]) is list and len(body["profileIds"]) == 1 and body["summaryInfo"] is True and type(body["limit"]) is int and body["limit"] == 1 and type(body["offset"]) is int and body["offset"] == 0, "facts_profile_request_mismatch") identity = body["profileIds"][0] values = payload["profileSummaries"]["profileInfo"] capture.require(type(values) is list and len(values) == 1 and type(values[0]) is dict and _profile_id(values[0].get("profileIdList")) == identity, "facts_profile_response_mismatch") capture.require(identity not in profiles, "facts_duplicate_profile") profiles[identity] = (_provenance(archive, response_file, "/data/profileSummaries/profileInfo/0", file), values[0]) identities = [capture.reservation_id(value) for _, value in first_pass] capture.require(not first_pass_open and len(identities) == len(set(identities)) and set(identities) == set(details) == set(rates), "facts_source_set_mismatch") if named: expected_profiles = {_eligible_primary_profile(value) for _, value in details.values()} expected_profiles.discard(None) capture.require(set(profiles) == expected_profiles, "facts_profile_set_mismatch") return first_pass, details, rates, profiles def _observations(root, base, path): # Iterative depth-first traversal, independent of the extractor's recursive # walker. Completed branches retain their positions amongst live branches. pending = [(root, base, tuple(path.split("/")))] result = [] while pending: node, pointer, remaining = pending.pop() if not remaining: result.append({"pointer": pointer, **atom(node)}) elif node is None: result.append({"pointer": pointer, "state": "null", "kind": "null"}) elif remaining[0] == "*": if type(node) is not list: result.append({"pointer": pointer, "state": "invalid_container", "kind": kind(node)}) elif not node: result.append({"pointer": pointer, "state": "empty_collection", "kind": "array"}) else: pending.extend((node[i], f"{pointer}/{i}", remaining[1:]) for i in reversed(range(len(node)))) elif type(node) is not dict: result.append({"pointer": pointer, "state": "invalid_container", "kind": kind(node)}) else: key = remaining[0] next_pointer = pointer + "/" + key.replace("~", "~0").replace("/", "~1") if key in node: pending.append((node[key], next_pointer, remaining[1:])) else: result.append({"pointer": next_pointer, "state": "missing"}) return result def _expected_fields(declarations, sources): result = {} for field, definitions in declarations.items(): variants = [] for source_name, variant, path in definitions: provenance, raw = sources[source_name] variants.append({"source": source_name, "variant": variant, "path": path, "observations": (_observations(raw, provenance["pointer"], path) if provenance is not None else [{"state": "source_not_acquired"}])}) result[field] = {"mapping_state": "unresolved", "variants": variants} return result def _same(actual, expected, error): # Python equality alone would incorrectly treat true == 1 == 1.0. try: matches = canonical(actual) == canonical(expected) except (TypeError, ValueError, OverflowError, RecursionError): matches = False capture.require(matches, error) def verify_source_facts(archive: audit.VerifiedArchive, raw: bytes) -> dict: capture.require(type(raw) is bytes and len(raw) <= MAX_FACTS_BYTES, "facts_byte_budget_exceeded") document = capture.strict_json(raw) named = type(archive) is named_audit.VerifiedArchive protocol = named_audit if named else audit version, contract_sha256, fields, context_paths, blockers = definitions(named) archive = protocol.VerifiedArchive(archive.directory, archive.pin) # Reuse archive/protocol integrity checks, never extractor-derived fields. protocol.replay(archive) searches, details, rates, profiles = _scan(archive, named=named) assessments = archive.document("rate-assessments.json")["records"] expected_header = {"version": version, "contract_sha256": contract_sha256, "status": "unaccepted_field_evidence", "capture_manifest_sha256": archive.pin, "hotel_id": archive.options.hotel_id, "from_date": archive.options.from_date, "to_date": archive.options.to_date, "rate_date": archive.options.rate_date, "record_order": "capture_order_not_report_order", **FLAGS, "business_blockers": list(blockers)} _same({key: value for key, value in document.items() if key != "records"}, expected_header, "facts_header_mismatch") records = document.get("records") capture.require(type(records) is list and len(records) == len(searches) == len(assessments), "facts_record_count_mismatch") for index, search in enumerate(searches): identity = capture.reservation_id(search[1]) assessment = assessments[index] capture.require(assessment["reservation_id"] == identity and assessment["rate_request"] == rates[identity][0]["request"]["file"], "facts_assessment_binding_mismatch") sources = {"search": search, "detail": details[identity], "rate": rates[identity], "assessment": (_provenance(archive, "rate-assessments.json", f"/records/{index}"), assessment)} if named: primary_id = _eligible_primary_profile(details[identity][1]) profile_assessment = assessment["profile"] if primary_id is None: capture.require("profile_request" not in profile_assessment and "profile_id" not in profile_assessment, "facts_unexpected_profile_query") sources["profile"] = (None, None) else: profile = profiles[primary_id] capture.require(profile_assessment.get("profile_id") == primary_id and profile_assessment.get("profile_request") == profile[0]["request"]["file"], "facts_profile_assessment_binding_mismatch") sources["profile"] = profile expected = {"capture_sequence": index + 1, "reservation_id": identity, "sources": {name: pair[0] for name, pair in sources.items()}, "fields": _expected_fields(fields, sources), "context": _expected_fields(context_paths, sources)} _same(records[index], expected, "facts_record_mismatch") return {"status": "field_evidence_verified", "facts_verified": True, "version": version, "evidence_sha256": hashlib.sha256(raw).hexdigest(), "capture_manifest_sha256": archive.pin, "contract_sha256": contract_sha256, "records": len(records), "network_calls": 0, **FLAGS} def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--capture-dir", required=True, type=Path) parser.add_argument("--capture-sha256", required=True) parser.add_argument("--facts", required=True, type=Path) parser.add_argument("--facts-sha256", required=True) parser.add_argument("--capture-version", choices=("v2", "v3"), default="v2") args = parser.parse_args(argv) try: capture.require(bool(re.fullmatch(r"[0-9a-f]{64}", args.facts_sha256)), "invalid_facts_pin") raw = audit.audit.protected_read(args.facts, MAX_FACTS_BYTES) capture.require(hashlib.sha256(raw).hexdigest() == args.facts_sha256, "facts_hash_mismatch") protocol = named_audit if args.capture_version == "v3" else audit result = verify_source_facts(protocol.VerifiedArchive(args.capture_dir, args.capture_sha256), raw) except capture.CollectionError as error: result = {"status": "failed", "facts_verified": False, "error": str(error), **FLAGS} except Exception: result = {"status": "failed", "facts_verified": False, "error": "facts_validation_failed", **FLAGS} print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 if result.get("facts_verified") is True else 1 if __name__ == "__main__": raise SystemExit(main())