294 lines
15 KiB
Python
294 lines
15 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from integrations.ohip import audit_arr_capture as audit
|
|
from integrations.ohip import collect_arr_source as collector
|
|
from tests.test_ohip_arr_collection import DAY, HOTEL, FakeService, row
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "integrations/ohip/audit_arr_capture.py"
|
|
|
|
|
|
class ArchiveAuditTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.directory = Path(self.temp.name) / "capture"
|
|
archive = collector.Archive(self.directory)
|
|
self.service = FakeService(count=3)
|
|
reader = collector.Reader(archive, HOTEL, self.service, sleep=lambda _: None)
|
|
self.result = collector.collect(collector.Options(DAY, HOTEL, page_size=2), archive, reader)
|
|
self.pin = self.result["manifest_sha256"]
|
|
|
|
def inspect(self, expected_error=None):
|
|
result = audit.audit_capture(self.directory, self.pin)
|
|
if expected_error:
|
|
self.assertEqual(result["status"], "invalid_capture")
|
|
self.assertEqual(result["error"], expected_error)
|
|
self.assertNotIn("analysis", result)
|
|
self.assertFalse(result["finance_ready"])
|
|
return result
|
|
|
|
def repin(self, result):
|
|
# Adversarial tests deliberately mint a new test pin to reach deeper validation.
|
|
raw = collector.json_bytes(result)
|
|
(self.directory / "result.json").write_bytes(raw)
|
|
self.pin = hashlib.sha256(raw).hexdigest()
|
|
|
|
def edit_file(self, name, edit):
|
|
path = self.directory / name
|
|
doc = json.loads(path.read_bytes())
|
|
edit(doc)
|
|
raw = collector.json_bytes(doc)
|
|
path.write_bytes(raw)
|
|
manifest = json.loads((self.directory / "result.json").read_bytes())
|
|
entry = next(item for item in manifest["files"] if item["name"] == name)
|
|
entry.update(bytes=len(raw), sha256=hashlib.sha256(raw).hexdigest())
|
|
self.repin(manifest)
|
|
|
|
def test_complete_capture_audits_without_network_or_writes(self):
|
|
before = {p.name: p.read_bytes() for p in self.directory.iterdir()}
|
|
with patch.object(collector, "HTTPTransport", side_effect=AssertionError("network forbidden")):
|
|
result = self.inspect()
|
|
self.assertEqual(result["status"], "verified_capture_audited")
|
|
self.assertEqual(result["analysis"]["source_records"], 3)
|
|
self.assertEqual(result["verified_files"], len(before))
|
|
self.assertEqual(result["network_calls"], 0)
|
|
self.assertEqual({p.name: p.read_bytes() for p in self.directory.iterdir()}, before)
|
|
text = json.dumps(result)
|
|
for sensitive in ("PRIVATE_NOTE", "id0", "confirmation-id0"):
|
|
self.assertNotIn(sensitive, text)
|
|
|
|
def test_manifest_pin_not_computed_from_candidate_implicitly(self):
|
|
self.pin = "0" * 64
|
|
self.inspect("manifest_hash_mismatch")
|
|
|
|
def test_changed_raw_source_fails_hash_check(self):
|
|
path = self.directory / "request-000001.response.bin"
|
|
raw = path.read_bytes().replace(b"PRIVATE_NOTE", b"CHANGED_NOTE")
|
|
path.write_bytes(raw)
|
|
self.inspect("archive_hash_mismatch")
|
|
|
|
def test_missing_or_unlisted_file_fails(self):
|
|
extra = self.directory / "extra.txt"
|
|
extra.write_text("private")
|
|
self.inspect("archive_file_set_mismatch")
|
|
extra.unlink()
|
|
(self.directory / "request-000001.response.bin").unlink()
|
|
self.inspect("archive_file_set_mismatch")
|
|
|
|
def test_inventory_traversal_cannot_read_outside_archive(self):
|
|
doc = json.loads((self.directory / "result.json").read_bytes())
|
|
doc["files"][0]["name"] = "../private.json"
|
|
self.repin(doc)
|
|
self.inspect("unsafe_inventory_name")
|
|
|
|
def test_duplicate_inventory_rejected(self):
|
|
doc = json.loads((self.directory / "result.json").read_bytes())
|
|
doc["files"].append(copy.deepcopy(doc["files"][0]))
|
|
self.repin(doc)
|
|
self.inspect("duplicate_inventory_name")
|
|
|
|
def test_symlink_file_refused(self):
|
|
path = self.directory / "request-000001.response.bin"
|
|
outside = self.directory.parent / "outside.bin"
|
|
path.rename(outside)
|
|
path.symlink_to(outside)
|
|
self.inspect("archive_or_shape_invalid")
|
|
|
|
def test_world_readable_file_or_directory_refused(self):
|
|
path = self.directory / "result.json"
|
|
path.chmod(0o644)
|
|
self.inspect("unsafe_archive_file")
|
|
path.chmod(0o600)
|
|
self.directory.chmod(0o755)
|
|
self.inspect("unsafe_archive_directory")
|
|
|
|
def test_failed_capture_and_business_ready_claim_rejected(self):
|
|
doc = json.loads((self.directory / "result.json").read_bytes())
|
|
doc["status"] = "failed"
|
|
self.repin(doc)
|
|
self.inspect("incomplete_capture")
|
|
doc["status"] = "complete_candidate_capture"
|
|
doc["finance_ready"] = True
|
|
self.repin(doc)
|
|
self.inspect("unsupported_capture_claim")
|
|
|
|
def test_rehashed_wrong_detail_cannot_pass_protocol_replay(self):
|
|
def change(doc):
|
|
doc["data"]["reservations"]["reservation"][0]["reservationIdList"][0]["id"] = "wrong-private-id"
|
|
self.edit_file("request-000003.response.bin", change)
|
|
self.inspect("capture_protocol_replay_failed")
|
|
|
|
def test_rehashed_wrong_request_cannot_pass(self):
|
|
self.edit_file("request-000001.json", lambda doc: doc["body"].update(arrivalEndDate="2026-09-14"))
|
|
self.inspect("capture_protocol_replay_failed")
|
|
|
|
def test_rehashed_wrong_attempt_cannot_pass(self):
|
|
self.edit_file("request-000001.json", lambda doc: doc.update(attempt=2))
|
|
self.inspect("capture_protocol_replay_failed")
|
|
|
|
def test_summary_counts_not_trusted(self):
|
|
doc = json.loads((self.directory / "result.json").read_bytes())
|
|
doc["verified_details"] = 2
|
|
self.repin(doc)
|
|
self.inspect("capture_summary_mismatch")
|
|
|
|
def test_capture_context_not_trusted(self):
|
|
self.edit_file("capture.json", lambda doc: doc.update(hotel_id="wrong"))
|
|
self.inspect("capture_context_mismatch")
|
|
|
|
def test_capture_grants_not_expanded_during_replay(self):
|
|
self.edit_file("capture.json", lambda doc: doc["operations"].append("postReservation"))
|
|
self.inspect("capture_contract_mismatch")
|
|
|
|
def test_successful_retried_capture_replays_without_sleep(self):
|
|
directory = self.directory.parent / "retried"
|
|
archive = collector.Archive(directory)
|
|
service = FakeService(count=1)
|
|
count = 0
|
|
def transport(*args):
|
|
nonlocal count
|
|
count += 1
|
|
if count == 1:
|
|
raise collector.urllib.error.URLError("private-error")
|
|
if count == 2:
|
|
return 429, {"Retry-After": "0"}, b"limited"
|
|
return service(*args)
|
|
reader = collector.Reader(archive, HOTEL, transport, sleep=lambda _: None)
|
|
result = collector.collect(collector.Options(DAY, HOTEL), archive, reader)
|
|
inspected = audit.audit_capture(directory, result["manifest_sha256"])
|
|
self.assertEqual(inspected["status"], "verified_capture_audited")
|
|
self.assertEqual(inspected["network_calls"], 0)
|
|
|
|
def test_cli_produces_safe_summary_and_nonzero_invalid_input(self):
|
|
result = subprocess.run([sys.executable, str(SCRIPT), "--capture-dir", str(self.directory),
|
|
"--manifest-sha256", self.pin], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(json.loads(result.stdout)["analysis"]["source_records"], 3)
|
|
self.assertNotIn("PRIVATE_NOTE", result.stdout + result.stderr)
|
|
result = subprocess.run([sys.executable, str(SCRIPT), "--capture-dir", str(self.directory),
|
|
"--manifest-sha256", "0" * 64], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertEqual(json.loads(result.stdout)["error"], "manifest_hash_mismatch")
|
|
self.assertEqual(result.stderr, "")
|
|
|
|
|
|
class CandidateFieldAuditTests(unittest.TestCase):
|
|
def pair(self):
|
|
search = row()
|
|
detail = copy.deepcopy(search)
|
|
detail["roomStay"].update(departureDate=DAY, guestCounts={"adults": 1, "children": 0},
|
|
currentRoomInfo={"roomId": "PRIVATE_ROOM"}, roomRates=[{
|
|
"start": DAY, "end": DAY, "numberOfUnits": 1, "roomId": "PRIVATE_ROOM",
|
|
"ratePlanCode": "PRIVATE_RATE", "roomType": "PRIVATE_TYPE",
|
|
"rates": {"rate": [{"effectiveRate": {"amountBeforeTax": 0}}]}}])
|
|
search["reservationGuest"] = {"fullName": "PRIVATE_FULL_NAME"}
|
|
return search, detail
|
|
|
|
def analyze(self, pairs):
|
|
result = audit.analyze_fields([pair[0] for pair in pairs], [pair[1] for pair in pairs])
|
|
self.assertNotIn("PRIVATE_", json.dumps(result))
|
|
self.assertEqual(len(result["fields"]), 16)
|
|
self.assertTrue(all(field["report_mapping_verified"] is False for field in result["fields"].values()))
|
|
return result
|
|
|
|
def test_zero_values_zero_night_and_single_day_segment_present(self):
|
|
result = self.analyze([self.pair()])
|
|
self.assertEqual(result["fields"]["CHILDREN"]["records_with_candidate"], 1)
|
|
self.assertEqual(result["fields"]["EFFECTIVE_RATE_AMOUNT"]["records_with_candidate"], 1)
|
|
self.assertEqual(result["scenarios"]["zero_night_records"], 1)
|
|
self.assertEqual(result["scenarios"]["arrival_interval_one_candidate"], 1)
|
|
self.assertEqual(result["scenarios"]["single_day_rate_segments"], 1)
|
|
self.assertEqual(result["scenarios"]["rate_segments_total"], 1)
|
|
|
|
def test_overlapping_segments_detected_without_choosing_first(self):
|
|
search, detail = self.pair()
|
|
another = copy.deepcopy(detail["roomStay"]["roomRates"][0])
|
|
another.update(ratePlanCode="PRIVATE_SECOND_RATE", numberOfUnits=2)
|
|
detail["roomStay"]["roomRates"].append(another)
|
|
result = self.analyze([(search, detail)])
|
|
self.assertEqual(result["scenarios"]["arrival_interval_multiple_candidates"], 1)
|
|
self.assertEqual(result["fields"]["RATE_CODE"]["records_with_distinct_candidates"], 1)
|
|
self.assertEqual(result["scenarios"]["multi_room_records"], 1)
|
|
|
|
def test_missing_rate_is_visible_without_whitelist_exclusion(self):
|
|
search, detail = self.pair()
|
|
del detail["roomStay"]["roomRates"][0]["ratePlanCode"]
|
|
result = self.analyze([(search, detail)])
|
|
self.assertEqual(result["source_records"], 1)
|
|
self.assertEqual(result["fields"]["RATE_CODE"]["records_without_candidate"], 1)
|
|
self.assertTrue(result["fields"]["RATE_CODE"]["required_for_all_source_rows"])
|
|
|
|
def test_base_amount_not_used_as_effective_price(self):
|
|
search, detail = self.pair()
|
|
detail["roomStay"]["roomRates"][0]["rates"]["rate"] = [{"base": {"amountBeforeTax": 50}, "total": {"amountBeforeTax": 70}}]
|
|
result = self.analyze([(search, detail)])
|
|
self.assertEqual(result["fields"]["EFFECTIVE_RATE_AMOUNT"]["records_without_candidate"], 1)
|
|
|
|
def test_company_roles_not_given_priority(self):
|
|
search, detail = self.pair()
|
|
detail["reservationProfiles"] = {"reservationProfile": [
|
|
{"reservationProfileType": role, "profile": {"company": {"companyName": "PRIVATE_" + role}}}
|
|
for role in ["Company", "TravelAgent"]]}
|
|
result = self.analyze([(search, detail)])
|
|
self.assertEqual(result["fields"]["COMPANY_NAME"]["records_with_distinct_candidates"], 1)
|
|
self.assertEqual(result["profile_roles"], {"Company": 1, "TravelAgent": 1})
|
|
|
|
def test_unknown_role_does_not_leak_value(self):
|
|
search, detail = self.pair()
|
|
detail["reservationProfiles"] = {"reservationProfile": [{"reservationProfileType": "PRIVATE_ROLE"}]}
|
|
self.assertEqual(self.analyze([(search, detail)])["profile_roles"], {"Other": 1})
|
|
|
|
def test_multiple_gen_notes_and_traces_keep_ambiguity(self):
|
|
search, detail = self.pair()
|
|
detail["comments"] = [{"comment": {"type": "GEN", "notificationLocation": "RESERVATION", "internal": internal,
|
|
"text": {"value": "PRIVATE_NOTE_" + str(internal)}}} for internal in (False, True)]
|
|
detail["traces"] = [{"traceText": "PRIVATE_TRACE_A"}, {"traceText": "PRIVATE_TRACE_B"}]
|
|
result = self.analyze([(search, detail)])
|
|
self.assertEqual(result["scenarios"]["records_with_multiple_nonempty_gen_notes"], 1)
|
|
self.assertEqual(result["scenarios"]["records_with_internal_gen_note"], 1)
|
|
self.assertEqual(result["scenarios"]["records_with_internal_and_external_gen_notes"], 1)
|
|
self.assertEqual(result["scenarios"]["records_with_multiple_nonempty_traces"], 1)
|
|
|
|
def test_duplicates_are_reported_not_removed_and_leading_zero_preserved(self):
|
|
pairs = [self.pair() for _ in range(3)]
|
|
for pair, room in zip(pairs, ("0101", "101", " 0101 ")):
|
|
pair[1]["roomStay"]["currentRoomInfo"]["roomId"] = room
|
|
result = self.analyze(pairs)
|
|
self.assertEqual(result["source_records"], 3)
|
|
self.assertEqual(result["scenarios"]["duplicate_nonblank_room_arrival_groups"], 1)
|
|
self.assertEqual(result["scenarios"]["records_in_duplicate_nonblank_room_arrival_groups"], 2)
|
|
|
|
def test_current_and_arrival_segment_room_conflict_visible(self):
|
|
search, detail = self.pair()
|
|
detail["roomStay"]["currentRoomInfo"]["roomId"] = "PRIVATE_DIFFERENT_ROOM"
|
|
result = self.analyze([(search, detail)])
|
|
self.assertEqual(result["fields"]["DISP_ROOM_NO"]["records_with_distinct_candidates"], 1)
|
|
self.assertEqual(result["scenarios"]["current_vs_arrival_segment_room_disagreement"], 1)
|
|
|
|
def test_missing_display_name_not_silently_built_from_components(self):
|
|
search, detail = self.pair()
|
|
del search["reservationGuest"]
|
|
detail["reservationGuests"] = [{"primary": True, "profileInfo": {"profile": {"customer": {
|
|
"personName": [{"nameType": "Primary", "givenName": "PRIVATE_GIVEN", "surname": "PRIVATE_SURNAME"}]}}}}]
|
|
result = self.analyze([(search, detail)])
|
|
self.assertEqual(result["fields"]["FULL_NAME"]["records_without_candidate"], 1)
|
|
self.assertEqual(result["scenarios"]["records_without_unique_primary_name"], 0)
|
|
self.assertEqual(result["scenarios"]["records_with_name_components_but_no_display_name"], 1)
|
|
self.assertEqual(result["scenarios"]["records_with_primary_surname"], 1)
|
|
self.assertEqual(result["scenarios"]["records_with_primary_givenName"], 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|