Files
ARR-2.0-0918/integrations/ohip/validate_source_facts.py

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())