room_cal/scripts/har_audit.py

"""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 = '4e93a4878415297ab95d22665c666b6602dcfcbb'

_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': '5d3e94e76186b839be544fc77e5f2fcac0b8d7639c3da16941b89cda5ee03ed9',
    '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': 'dd7478b376ad0a9709c7201b7e43cf402df011ec9d2fe4c1dff2ac418bcd9fb3',
    'hint': 'open any guest message thread',
  },
  'CreateBulkMessagesMutation': {
    'hash': '94ac2c4bd07edace539dbf2b9665d9030b6dee479db345ba8a8bbc234b3bbfa3',
    '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': 'dml/graphql.json',
    'operation_name': 'conversationMessages',
    'hint': 'open any guest message thread',
    'headers': {
      'apollographql-client-version': 'ABJaNSUc',
    },
  },
  'postFreeTextMessage': {
    'url_fragment': 'dml/graphql.json',
    '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()