Add har_audit.py — HAR drift checker for service_api clients

Parses HAR files from Airbnb and Booking.com captures and reports:
  ✓ operation found, identifiers match client constants
  ✗ operation found but hash / header value is stale (shows what to update)
  — operation not captured (shows what browser action triggers it)

Covers all operations tracked in _airbnb.py (_HASHES, _API_KEY,
_CLIENT_VERSION) and all endpoints in _booking.py (_gql_headers,
_msg_headers). Manifest kept in sync with client constants manually.

Usage: uv run python scripts/har_audit.py path/to/*.har

scripts/har_audit.py

@@ -0,0 +1,333 @@
+"""Audit HAR files for service_api client drift.
+
+Filters captured browser traffic down to the API calls we care about, compares
+identifiers (GQL hashes, version headers) against current client constants, and
+lists any operations that weren't captured — with a hint for what action
+triggers each one.
+
+Usage:
+  uv run python scripts/har_audit.py path/to/*.har
+
+One or more HAR files may be passed. Platform is detected from request domains;
+a single HAR may contain both platforms.
+"""
+
+import json
+import re
+import sys
+from pathlib import Path
+
+# ── Airbnb manifest ────────────────────────────────────────────────────────
+# Sync with _airbnb.py: _API_KEY, _CLIENT_VERSION, _HASHES
+
+_AIRBNB_DOMAIN = 'www.airbnb.com'
+_AIRBNB_API_KEY = 'd306zoyjsyarp7ifhu67rjxn52tv0t20'
+_AIRBNB_CLIENT_VERSION = 'ef690fbd6464c4bd9f632ec18087662c4e4f05e1'
+
+_AIRBNB_OPS: dict[str, dict] = {
+  'getDLSHostCalendar': {
+    'hash': '048e22675ffa53121628d2d07b519b9b71f5dab342a1ceb75aa72d8814a00ea7',
+    'hint': 'load the multicalendar page',
+  },
+  'EditPanelPricingSettingsMutation': {
+    'hash': 'e27fed4f08dd4ab4c0267c5d6c0a7e4920730deadf48f92feba9b0d08befafa4',
+    'hint': 'change a price on any date in the multicalendar',
+  },
+  'EditPanelAvailabilityMutation': {
+    'hash': '2e056e7f98f8dc410a294c649adf719452980eb4d8ea95540fb1297af2637f67',
+    'hint': 'block or unblock a date in the multicalendar',
+  },
+  'StayHostingDetailsQuery': {
+    'hash': '1e81282ae4de44e05bc8b776456644efd4ebcd676451ce7f6c3c5757a1057e02',
+    'hint': 'open any reservation detail page',
+  },
+  'getNightlyPriceTips': {
+    'hash': '07538bf1a5afc8171f1e7443f0007d7f020142fc7ab60df065e7e1edd1a0b29f',
+    'hint': 'load the multicalendar page (fires automatically)',
+  },
+  'HostCalendarListingEventsQuery': {
+    'hash': '8bfe5f5572a25fdaff5b5790604e1de3aec5f37731fc505684022d590d06e0a8',
+    'hint': 'load the multicalendar page (fires automatically)',
+  },
+  'ViaductGetThreadAndDataQuery': {
+    'hash': 'c6aff32757966df45d73811219d4f3a57ae7cc9e047b39bb9ba396cd0c09defa',
+    'hint': 'open any guest message thread',
+  },
+  'CreateBulkMessagesMutation': {
+    'hash': 'b0298188a93a73218fecdd512afb7b47c0df3112a4819c236ecc4c0c10c04bd2',
+    'hint': 'send a message in any thread',
+  },
+}
+
+# ── Booking.com manifest ───────────────────────────────────────────────────
+# Sync with _booking.py: _gql_headers(), _msg_headers()
+
+_BOOKING_DOMAIN = 'admin.booking.com'
+
+_BOOKING_ENDPOINTS: dict[str, dict] = {
+  'inventory/fetch': {
+    'url_fragment': 'fresa/extranet/inventory/fetch',
+    'hint': 'load the monthly calendar',
+  },
+  'inventory/update': {
+    'url_fragment': 'fresa/extranet/inventory/update',
+    'hint': 'change a price on any date',
+  },
+  'get_reservations': {
+    'url_fragment': 'fresa/extranet/calendar/get_reservations',
+    'hint': 'load the reservations list',
+  },
+  'updateRoomDateMutation': {
+    'url_fragment': 'dml/graphql.json',
+    'operation_name': 'updateRoomDateMutation',
+    'hint': 'block or unblock a date on the calendar',
+    'headers': {
+      'apollographql-client-version': 'EQSXMHMY',
+      'x-envoy-expected-rq-timeout-ms': '15000',
+    },
+  },
+  'conversationMessages': {
+    'url_fragment': 'fresa/extranet/messaging/graphql',
+    'operation_name': 'conversationMessages',
+    'hint': 'open any guest message thread',
+    'headers': {
+      'apollographql-client-version': 'ABJaNSUc',
+    },
+  },
+  'postFreeTextMessage': {
+    'url_fragment': 'fresa/extranet/messaging/graphql',
+    'operation_name': 'postFreeTextMessage',
+    'hint': 'send a message to a guest',
+    'headers': {
+      'apollographql-client-version': 'ABJaNSUc',
+    },
+  },
+}
+
+
+# ── HAR helpers ────────────────────────────────────────────────────────────
+
+def _req_headers(entry: dict) -> dict[str, str]:
+  return {h['name'].lower(): h['value'] for h in entry['request']['headers']}
+
+def _req_body(entry: dict) -> dict | None:
+  text = (entry['request'].get('postData') or {}).get('text', '')
+  try:
+    return json.loads(text) if text else None
+  except (json.JSONDecodeError, ValueError):
+    return None
+
+def _short(h: str) -> str:
+  """Truncate a hash/token for display: first 16 chars + ellipsis."""
+  return h[:16] + '...' if len(h) > 20 else h
+
+
+# ── Airbnb ─────────────────────────────────────────────────────────────────
+
+def _collect_airbnb(entries: list[dict]) -> dict[str, list[dict]]:
+  """Return {operation_name: [entry, ...]} for Airbnb /api/v3 calls."""
+  found: dict[str, list[dict]] = {}
+  for e in entries:
+    url = e['request']['url']
+    m = re.search(r'/api/v3/([A-Za-z]+)/([0-9a-f]{64})', url)
+    if m:
+      op = m.group(1)
+      found.setdefault(op, []).append(e)
+  return found
+
+
+def _audit_airbnb(entries: list[dict], har_name: str) -> bool:
+  found = _collect_airbnb(entries)
+  if not found:
+    return False
+
+  print(f'\n{"═" * 70}')
+  print(f'  Airbnb  |  {har_name}')
+  print(f'{"═" * 70}')
+
+  # Operations
+  print('\n  Operations:\n')
+  col = max(len(op) for op in _AIRBNB_OPS) + 2
+
+  all_ok = True
+  for op, spec in _AIRBNB_OPS.items():
+    if op not in found:
+      print(f'    —  {op:<{col}} not captured')
+      print(f'       {"":>{col}}   → {spec["hint"]}')
+      continue
+
+    entry = found[op][0]
+    url = entry['request']['url']
+    m = re.search(r'/api/v3/[A-Za-z]+/([0-9a-f]{64})', url)
+    har_hash = m.group(1) if m else '?'
+    expected = spec['hash']
+
+    if har_hash == expected:
+      print(f'    ✓  {op:<{col}} hash matches')
+    else:
+      all_ok = False
+      print(f'    ✗  {op:<{col}} STALE hash')
+      print(f'       {"client:":<{col+5}} {expected}')
+      print(f'       {"HAR:":<{col+5}} {har_hash}   ← update _HASHES[\'{op}\']')
+
+  # Global headers (from any /api/v3 call)
+  sample_entry = next(iter(found.values()))[0]
+  hdrs = _req_headers(sample_entry)
+
+  print('\n  Global headers:\n')
+  hcol = max(len('X-Airbnb-API-Key'), len('X-Client-Version')) + 2
+
+  api_key = hdrs.get('x-airbnb-api-key', '')
+  if api_key == _AIRBNB_API_KEY:
+    print(f'    ✓  {"X-Airbnb-API-Key":<{hcol}} matches')
+  else:
+    all_ok = False
+    print(f'    ✗  {"X-Airbnb-API-Key":<{hcol}} MISMATCH')
+    print(f'       client: {_AIRBNB_API_KEY}')
+    print(f'       HAR:    {api_key}   ← update _API_KEY')
+
+  cv = hdrs.get('x-client-version', '')
+  if cv == _AIRBNB_CLIENT_VERSION:
+    print(f'    ✓  {"X-Client-Version":<{hcol}} matches')
+  else:
+    all_ok = False
+    print(f'    ✗  {"X-Client-Version":<{hcol}} stale')
+    print(f'       client: {_AIRBNB_CLIENT_VERSION}')
+    print(f'       HAR:    {cv}   ← update _CLIENT_VERSION')
+
+  missing = [op for op in _AIRBNB_OPS if op not in found]
+  if not missing:
+    print('\n  All operations captured.')
+  else:
+    print(f'\n  {len(missing)} operation(s) not captured — to get full coverage:')
+    printed_hints: set[str] = set()
+    for op in missing:
+      hint = _AIRBNB_OPS[op]['hint']
+      if hint not in printed_hints:
+        print(f'    • {hint}')
+        printed_hints.add(hint)
+
+  return True
+
+
+# ── Booking.com ────────────────────────────────────────────────────────────
+
+def _collect_booking(entries: list[dict]) -> dict[str, list[dict]]:
+  """Return {endpoint_key: [entry, ...]} for Booking.com API calls."""
+  found: dict[str, list[dict]] = {}
+  for e in entries:
+    url = e['request']['url']
+    if _BOOKING_DOMAIN not in url:
+      continue
+    body = _req_body(e)
+    body_op = (body or {}).get('operationName')
+    for key, spec in _BOOKING_ENDPOINTS.items():
+      if spec['url_fragment'] not in url:
+        continue
+      op_name = spec.get('operation_name')
+      if op_name and body_op != op_name:
+        continue
+      found.setdefault(key, []).append(e)
+  return found
+
+
+def _audit_booking(entries: list[dict], har_name: str) -> bool:
+  found = _collect_booking(entries)
+  if not found:
+    return False
+
+  print(f'\n{"═" * 70}')
+  print(f'  Booking.com  |  {har_name}')
+  print(f'{"═" * 70}')
+
+  print('\n  Endpoints:\n')
+  col = max(len(k) for k in _BOOKING_ENDPOINTS) + 2
+
+  all_ok = True
+  for key, spec in _BOOKING_ENDPOINTS.items():
+    if key not in found:
+      print(f'    —  {key:<{col}} not captured')
+      print(f'       {"":>{col}}   → {spec["hint"]}')
+      continue
+
+    entry = found[key][0]
+    hdrs = _req_headers(entry)
+    expected_headers = spec.get('headers', {})
+
+    mismatches: list[tuple[str, str, str]] = []
+    for hname, hexpected in expected_headers.items():
+      hactual = hdrs.get(hname, '')
+      if hactual != hexpected:
+        mismatches.append((hname, hexpected, hactual))
+
+    if not mismatches:
+      print(f'    ✓  {key:<{col}} found')
+    else:
+      all_ok = False
+      print(f'    ✗  {key:<{col}} found with stale headers')
+      for hname, hexpected, hactual in mismatches:
+        print(f'       {hname}')
+        print(f'         client: {hexpected}')
+        src = '_gql_headers()' if 'availability' in key.lower() or 'room' in key.lower() else '_msg_headers()'
+        print(f'         HAR:    {hactual}   ← update {src}')
+
+  missing = [k for k in _BOOKING_ENDPOINTS if k not in found]
+  if not missing:
+    print('\n  All endpoints captured.')
+  else:
+    print(f'\n  {len(missing)} endpoint(s) not captured — to get full coverage:')
+    printed_hints: set[str] = set()
+    for key in missing:
+      hint = _BOOKING_ENDPOINTS[key]['hint']
+      if hint not in printed_hints:
+        print(f'    • {hint}')
+        printed_hints.add(hint)
+
+  return True
+
+
+# ── Entry point ────────────────────────────────────────────────────────────
+
+def audit_har(path: Path) -> None:
+  try:
+    har = json.loads(path.read_text(encoding='utf-8'))
+  except Exception as e:
+    print(f'Error reading {path}: {e}', file=sys.stderr)
+    return
+
+  entries = har.get('log', {}).get('entries', [])
+  if not entries:
+    print(f'No entries in {path.name}', file=sys.stderr)
+    return
+
+  # Detect which platforms appear in this HAR
+  domains = {
+    e['request']['url'].split('/')[2]
+    for e in entries
+    if e['request']['url'].startswith('http')
+  }
+
+  name = path.name
+  found_any = False
+  if _AIRBNB_DOMAIN in domains:
+    found_any |= _audit_airbnb(entries, name)
+  if _BOOKING_DOMAIN in domains:
+    found_any |= _audit_booking(entries, name)
+
+  if not found_any:
+    known = f'{_AIRBNB_DOMAIN}, {_BOOKING_DOMAIN}'
+    print(f'\n{path.name}: no recognised API calls (expected domains: {known})')
+
+
+def main() -> None:
+  paths = sys.argv[1:]
+  if not paths:
+    print(__doc__)
+    sys.exit(1)
+  for p in paths:
+    audit_har(Path(p))
+  print()
+
+
+if __name__ == '__main__':
+  main()