261 lines
14 KiB
Python
261 lines
14 KiB
Python
"""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())
|