326 lines
18 KiB
Python
326 lines
18 KiB
Python
"""Handwritten synthetic profile responses and adversarial v3 archive replay."""
|
|
import copy
|
|
from dataclasses import replace
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from integrations.ohip import collect_arr_named_day as named, collect_arr_source as source
|
|
from integrations.ohip import audit_arr_named_day as audit, audit_arr_day as v2audit
|
|
from integrations.ohip import profile_summary as probe, profile_reader
|
|
from integrations.ohip.rate_info import RateInfoReader
|
|
from tests.test_ohip_day_capture import DayService, DAY, HOTEL, SECRET
|
|
|
|
|
|
class NamedService(DayService):
|
|
def __init__(self, count=3):
|
|
super().__init__(count)
|
|
self.profile_calls = []
|
|
self.profile_edit = lambda document, body: document
|
|
self.profiles = {}
|
|
for i, row in enumerate(self.base.rows):
|
|
name = {"nameType": "Primary", "surname": f"Surname{i}", "givenName": "Given", "namePrefix": "Dr"}
|
|
identity = f"p{i}"
|
|
row["reservationGuests"] = [{"primary": True, "profileInfo": {
|
|
"profileIdList": [{"type": "Profile", "id": identity}],
|
|
"profile": {"customer": {"personName": [name]}}}}]
|
|
self.profiles[identity] = {"name": name["surname"], "givenName": "Given", "namePrefix": "Dr",
|
|
"fullName": f" Surname{i}, Given · 测试 "}
|
|
|
|
def __call__(self, method, path, raw):
|
|
if path != probe.PATH:
|
|
return super().__call__(method, path, raw)
|
|
self.calls.append((method, path, raw))
|
|
self.profile_calls.append((method, path, raw))
|
|
body = json.loads(raw)
|
|
identity = body["profileIds"][0]
|
|
document = {"operation_id": probe.OPERATION, "hotel_id": HOTEL, "oracle_request_id": "synthetic",
|
|
"data": {"profileSummaries": {"offset": 1, "limit": 1, "totalResults": 1, "totalPages": 1,
|
|
"hasMore": False, "profileInfo": [{"profileIdList": [{"type": "Profile", "id": identity}],
|
|
"profile": {"formerName": copy.deepcopy(self.profiles[identity])}}]}}}
|
|
return 200, {}, source.json_bytes(self.profile_edit(document, body))
|
|
|
|
|
|
class NamedDayTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.root = Path(temporary.name)
|
|
self.directory = self.root / "capture"
|
|
self.service = NamedService()
|
|
self.options = named.Options(DAY, DAY, DAY, HOTEL, page_size=2, max_profiles=3)
|
|
self.delays = []
|
|
|
|
def capture(self, transport=None, options=None):
|
|
archive = source.Archive(self.directory)
|
|
options = options or self.options
|
|
transport = transport or self.service
|
|
return named.collect(options, archive,
|
|
source.Reader(archive, HOTEL, transport, key=SECRET, sleep=self.delays.append),
|
|
RateInfoReader(archive, HOTEL, transport, key=SECRET, sleep=self.delays.append),
|
|
profile_reader.ProfileSummaryReader(archive, HOTEL, transport, max_profiles=options.max_profiles,
|
|
key=SECRET, sleep=self.delays.append))
|
|
|
|
def verify(self, result):
|
|
archive = audit.VerifiedArchive(self.directory, result["manifest_sha256"])
|
|
with patch.object(source, "HTTPTransport", side_effect=AssertionError("network forbidden")), \
|
|
patch.object(source, "load_key", side_effect=AssertionError("key forbidden")):
|
|
return audit.replay(archive)
|
|
|
|
def repin(self, filename, edit):
|
|
path = self.directory / filename
|
|
doc = json.loads(path.read_bytes()); edit(doc)
|
|
raw = source.json_bytes(doc); path.write_bytes(raw)
|
|
manifest = self.directory / "result.json"
|
|
if filename != "result.json":
|
|
doc = json.loads(manifest.read_bytes())
|
|
for item in doc["files"]:
|
|
if item["name"] == filename:
|
|
item.update(bytes=len(raw), sha256=hashlib.sha256(raw).hexdigest())
|
|
manifest.write_bytes(source.json_bytes(doc))
|
|
return {"manifest_sha256": hashlib.sha256(manifest.read_bytes()).hexdigest()}
|
|
|
|
def test_v3_exact_requests_names_dates_privacy_and_offline_replay(self):
|
|
result = self.capture()
|
|
self.assertEqual(result["version"], named.VERSION)
|
|
self.assertTrue(result["candidate_capture_complete"])
|
|
self.assertTrue(result["all_names_valid"])
|
|
self.assertEqual((result["profile_records"], result["unique_profiles"], result["profile_http_attempts"]), (3, 3, 3))
|
|
self.assertEqual((result["http_attempts"], result["valid_name_candidates"]), (13, 3))
|
|
self.assertEqual([json.loads(raw) for _, _, raw in self.service.profile_calls],
|
|
[{"profileIds": [f"p{i}"], "summaryInfo": True, "limit": 1, "offset": 0} for i in range(3)])
|
|
rows, details, assessments = self.verify(result)
|
|
self.assertEqual((len(rows), len(details)), (3, 3))
|
|
for i, item in enumerate(assessments["records"]):
|
|
self.assertEqual(item["profile"]["full_name"], self.service.profiles[f"p{i}"]["fullName"])
|
|
self.assertIsNone(item["profile"]["name_type"])
|
|
self.assertEqual(item["profile"]["component_comparisons"]["namePrefix"], "equal")
|
|
public = json.dumps(result)
|
|
self.assertNotIn("Surname", public)
|
|
self.assertNotIn("full_name", public)
|
|
for path in self.directory.iterdir():
|
|
self.assertEqual(path.stat().st_mode & 0o777, 0o600)
|
|
self.assertNotIn(SECRET.encode(), path.read_bytes())
|
|
self.assertFalse(result["finance_ready"])
|
|
self.assertFalse(result["report_equivalence_verified"])
|
|
|
|
def test_shared_profile_response_reused_but_each_primary_name_compared(self):
|
|
first, second = self.service.base.rows[:2]
|
|
second["reservationGuests"] = copy.deepcopy(first["reservationGuests"])
|
|
second["reservationGuests"][0]["profileInfo"]["profile"]["customer"]["personName"][0]["givenName"] = "Different"
|
|
result = self.capture(options=replace(self.options, max_profiles=2))
|
|
self.assertTrue(result["candidate_capture_complete"])
|
|
self.assertFalse(result["all_names_valid"])
|
|
self.assertEqual((result["unique_profiles"], result["profile_records"], result["valid_name_candidates"]), (2, 3, 2))
|
|
records = self.verify(result)[2]["records"]
|
|
self.assertEqual(records[0]["profile"]["profile_request"], records[1]["profile"]["profile_request"])
|
|
self.assertEqual(records[1]["profile"]["error"], "profile_name_components_mismatch")
|
|
self.assertNotIn("full_name", records[1]["profile"])
|
|
|
|
def test_prefix_conflict_is_not_hidden_by_matching_title(self):
|
|
self.service.profiles["p0"]["namePrefix"] = "Ms"
|
|
result = self.capture()
|
|
self.assertEqual(result["profile_issues"], {"profile_name_components_mismatch": 1})
|
|
self.assertEqual(self.verify(result)[2]["records"][0]["profile"]["component_comparisons"]["namePrefix"], "different")
|
|
|
|
def test_missing_primary_keeps_record_as_gap_without_querying_unknown_identity(self):
|
|
self.service.base.rows[0]["reservationGuests"] = []
|
|
result = self.capture()
|
|
self.assertTrue(result["candidate_capture_complete"])
|
|
self.assertEqual(result["valid_name_candidates"], 2)
|
|
self.assertEqual(result["profile_issues"], {"profile_ambiguous_primary_guest": 1})
|
|
self.assertEqual(len(self.verify(result)[0]), 3)
|
|
self.assertEqual(len(self.service.profile_calls), 2)
|
|
|
|
def test_malformed_profile_identity_aborts_before_profile_request(self):
|
|
self.service.base.rows[0]["reservationGuests"][0]["profileInfo"]["profileIdList"][0]["id"] = "../bad"
|
|
result = self.capture()
|
|
self.assertFalse(result["candidate_capture_complete"])
|
|
self.assertEqual(result["error"], "profile_ambiguous_identity")
|
|
self.assertFalse(self.service.profile_calls)
|
|
|
|
def test_blank_and_missing_and_nonprimary_full_names_remain_gaps(self):
|
|
self.service.profiles["p0"]["fullName"] = " \t "
|
|
self.service.profiles["p1"].pop("fullName")
|
|
self.service.profiles["p2"]["nameType"] = "Alternate"
|
|
result = self.capture()
|
|
self.assertTrue(result["candidate_capture_complete"])
|
|
self.assertEqual(result["valid_name_candidates"], 0)
|
|
self.assertEqual(result["profile_issues"], {"profile_full_name_unavailable": 2, "profile_nonprimary_summary_name": 1})
|
|
self.verify(result)
|
|
|
|
def test_all_empty_core_components_do_not_verify_a_name(self):
|
|
name = self.service.base.rows[0]["reservationGuests"][0]["profileInfo"]["profile"]["customer"]["personName"][0]
|
|
name.update(surname="", givenName=" ")
|
|
self.service.profiles["p0"].update(name="", givenName=" ")
|
|
result = self.capture()
|
|
self.assertEqual(result["profile_issues"], {"profile_primary_name_unverifiable": 1})
|
|
self.verify(result)
|
|
|
|
def test_transient_retry_keeps_all_bytes_and_last_success_request_reference(self):
|
|
calls = []
|
|
def transport(method, path, raw):
|
|
if path == probe.PATH:
|
|
calls.append(raw)
|
|
if len(calls) == 1:
|
|
return 429, {"Retry-After": "2"}, b"{}"
|
|
if len(calls) == 2:
|
|
raise TimeoutError("PRIVATE_FAILURE")
|
|
return self.service(method, path, raw)
|
|
result = self.capture(transport)
|
|
self.assertEqual(self.delays, [2, 2])
|
|
self.assertEqual((result["unique_profiles"], result["profile_http_attempts"]), (3, 5))
|
|
records = self.verify(result)[2]["records"]
|
|
self.assertEqual(records[0]["profile"]["profile_request"], "profile-000003.json")
|
|
self.assertEqual((self.directory / "profile-000001.response.bin").read_bytes(), b"{}")
|
|
self.assertFalse((self.directory / "profile-000002.response.bin").exists())
|
|
|
|
def test_max_profiles_prevents_extra_profile_network_call(self):
|
|
result = self.capture(options=replace(self.options, max_profiles=1))
|
|
self.assertEqual(result["error"], "profile_limit_exceeded")
|
|
self.assertFalse(result["candidate_capture_complete"])
|
|
self.assertEqual(len(self.service.profile_calls), 1)
|
|
self.assertFalse((self.directory / "profile-000002.json").exists())
|
|
|
|
def test_wrong_identity_hotel_operation_pagination_warning_and_types_are_fatal(self):
|
|
edits = [lambda d: d.update(hotel_id="OTHER"), lambda d: d.update(operation_id="getProfiles"),
|
|
lambda d: d.update(warnings=[{"text": "PRIVATE"}]),
|
|
lambda d: d["data"]["profileSummaries"].update(hasMore=True),
|
|
lambda d: d["data"]["profileSummaries"].update(offset=True),
|
|
lambda d: d["data"]["profileSummaries"].update(totalResults=2),
|
|
lambda d: d["data"]["profileSummaries"]["profileInfo"][0]["profileIdList"][0].update(id="different"),
|
|
lambda d: d["data"]["profileSummaries"]["profileInfo"][0]["profile"]["formerName"].update(fullName=None)]
|
|
for i, edit in enumerate(edits):
|
|
with self.subTest(index=i):
|
|
self.directory = self.root / f"case-{i}"
|
|
self.service = NamedService()
|
|
def change(doc, _, edit=edit):
|
|
edit(doc); return doc
|
|
self.service.profile_edit = change
|
|
result = self.capture()
|
|
self.assertFalse(result["candidate_capture_complete"])
|
|
self.assertEqual(len(self.service.profile_calls), 1)
|
|
self.assertNotIn("PRIVATE", json.dumps(result))
|
|
|
|
def test_denial_and_transient_exhaustion_stop_boundedly(self):
|
|
for status, expected in [(403, 1), (503, 3)]:
|
|
self.directory = self.root / f"http-{status}"
|
|
calls = []
|
|
def transport(method, path, raw):
|
|
if path == probe.PATH:
|
|
calls.append(raw); return status, {}, b"{}"
|
|
return self.service(method, path, raw)
|
|
result = self.capture(transport)
|
|
self.assertEqual(result["error"], "profile_http_failure")
|
|
self.assertFalse(result["candidate_capture_complete"])
|
|
self.assertEqual(len(calls), expected)
|
|
|
|
def test_retry_after_beyond_bound_stops_without_early_retry(self):
|
|
calls = []
|
|
def transport(method, path, raw):
|
|
if path == probe.PATH:
|
|
calls.append(raw); return 503, {"Retry-After": "31"}, b"{}"
|
|
return self.service(method, path, raw)
|
|
result = self.capture(transport)
|
|
self.assertEqual(result["error"], "retry_after_exceeds_budget")
|
|
self.assertEqual(len(calls), 1)
|
|
self.assertFalse(self.delays)
|
|
|
|
def test_v2_verifier_and_replay_refuse_named_archive(self):
|
|
result = self.capture()
|
|
with self.assertRaisesRegex(source.CollectionError, "incomplete_capture"):
|
|
v2audit.VerifiedArchive(self.directory, result["manifest_sha256"])
|
|
with self.assertRaisesRegex(source.CollectionError, "unsupported_day_archive_version"):
|
|
v2audit.replay(audit.VerifiedArchive(self.directory, result["manifest_sha256"]))
|
|
|
|
def test_repinned_profile_request_reference_and_count_tampering_rejected(self):
|
|
for i, (filename, edit, error) in enumerate([
|
|
("profile-000001.json", lambda d: d["body"].update(profileIds=["p2"]), "capture_protocol_replay_failed"),
|
|
("rate-assessments.json", lambda d: d["records"][0]["profile"].update(profile_request="profile-000002.json"), "rate_assessment_mismatch"),
|
|
("result.json", lambda d: d.update(unique_profiles=99), "capture_summary_mismatch")]):
|
|
with self.subTest(filename=filename):
|
|
self.directory = self.root / f"tamper-{i}"
|
|
self.capture()
|
|
pin = self.repin(filename, edit)
|
|
with self.assertRaisesRegex(source.CollectionError, error):
|
|
self.verify(pin)
|
|
|
|
def test_invalid_limits_dates_and_reader_context_make_no_requests(self):
|
|
for limit in (0, True, -1, 10001):
|
|
with self.subTest(limit=limit), self.assertRaises(source.CollectionError):
|
|
replace(self.options, max_profiles=limit).validate()
|
|
with self.assertRaisesRegex(source.CollectionError, "report_dates_must_match"):
|
|
replace(self.options, rate_date="2026-09-14").validate()
|
|
archive = source.Archive(self.directory)
|
|
with self.assertRaisesRegex(source.CollectionError, "profile_reader_context_mismatch"):
|
|
named.collect(self.options, archive, source.Reader(archive, HOTEL, self.service),
|
|
RateInfoReader(archive, HOTEL, self.service),
|
|
profile_reader.ProfileSummaryReader(archive, HOTEL, self.service, max_profiles=1))
|
|
self.assertFalse(self.service.calls)
|
|
|
|
def test_replay_rejects_boolean_integer_aliases_inside_requests_and_counters(self):
|
|
for key, value in (("summaryInfo", 1), ("limit", True), ("offset", False)):
|
|
self.directory = self.root / f"typed-{key}"
|
|
self.capture()
|
|
pin = self.repin("profile-000001.json", lambda d: d["body"].update({key: value}))
|
|
with self.subTest(key=key), self.assertRaisesRegex(source.CollectionError, "capture_protocol_replay_failed"):
|
|
self.verify(pin)
|
|
self.directory = self.root / "typed-counter"
|
|
self.service.profiles["p0"].pop("fullName")
|
|
self.capture()
|
|
pin = self.repin("result.json", lambda d: d["profile_issues"].update(profile_full_name_unavailable=True))
|
|
with self.assertRaisesRegex(source.CollectionError, "capture_summary_mismatch"):
|
|
self.verify(pin)
|
|
|
|
def test_explicit_malformed_primary_containers_are_not_missing_name_gaps(self):
|
|
edits = [lambda d: d.update(reservationGuests=None),
|
|
lambda d: d["reservationGuests"][0].update(profileInfo=123),
|
|
lambda d: d["reservationGuests"][0]["profileInfo"].update(profile=[]),
|
|
lambda d: d["reservationGuests"][0]["profileInfo"]["profile"].update(customer="bad"),
|
|
lambda d: d["reservationGuests"][0]["profileInfo"]["profile"]["customer"].update(personName=None)]
|
|
for i, edit in enumerate(edits):
|
|
with self.subTest(index=i):
|
|
self.directory = self.root / f"structure-{i}"
|
|
self.service = NamedService()
|
|
edit(self.service.base.rows[0])
|
|
result = self.capture()
|
|
self.assertFalse(result["candidate_capture_complete"])
|
|
self.assertTrue(result["error"].startswith("profile_invalid"))
|
|
self.assertFalse(self.service.profile_calls)
|
|
|
|
def test_manifest_budget_exhaustion_stops_early_with_bounded_failure_result(self):
|
|
with patch.object(source, "MAX_MANIFEST_BYTES", 12000), \
|
|
patch.object(source, "MANIFEST_SUMMARY_RESERVE", 4096):
|
|
result = self.capture()
|
|
self.assertFalse(result["candidate_capture_complete"])
|
|
self.assertEqual(result["error"], "archive_manifest_budget_exceeded")
|
|
self.assertLess(len(self.service.calls), 13)
|
|
self.assertLessEqual((self.directory / "result.json").stat().st_size, 12000)
|
|
self.assertTrue((self.directory / "rate-assessments.json").exists())
|
|
saved = json.loads((self.directory / "result.json").read_bytes())
|
|
self.assertEqual(saved["error"], "archive_manifest_budget_exceeded")
|
|
|
|
def test_reordered_inventory_rejected_even_after_repinning(self):
|
|
self.capture()
|
|
pin = self.repin("result.json", lambda d: d["files"].reverse())
|
|
with self.assertRaisesRegex(source.CollectionError, "archive_sequence_mismatch"):
|
|
self.verify(pin)
|
|
|
|
def test_altered_profile_raw_refused_without_accepting_its_own_hash(self):
|
|
result = self.capture()
|
|
path = self.directory / "profile-000001.response.bin"
|
|
original = path.read_bytes()
|
|
changed = original.replace(b"Surname0", b"Changed0")
|
|
self.assertNotEqual(changed, original)
|
|
self.assertEqual(len(changed), len(original))
|
|
path.write_bytes(changed)
|
|
with self.assertRaisesRegex(source.CollectionError, "archive_hash_mismatch"):
|
|
self.verify(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|