"""Three read-only simulated hotel endpoints, restricted to authenticated loopback.""" from collections import Counter from contextlib import contextmanager from copy import deepcopy from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import math import re import secrets import threading import urllib.error import urllib.parse import urllib.request from arr_web.local_api_fixture import HOTEL, VERSION, number from integrations.ohip import collect_arr_source as source, rate_info class LocalAPI: def __init__(self, fixture): self.fixture = deepcopy(fixture) self.key = secrets.token_urlsafe(32) self.marker = {"version": VERSION, "source_sha256": fixture["source_sha256"]} self.records = {source.reservation_id(r["search"]): r for r in self.fixture["records"]} self.ordered = sorted(self.records.values(), key=lambda r: ( next(i["id"] for i in r["search"]["reservationIdList"] if i["type"] == "Confirmation"), source.reservation_id(r["search"]))) self.counts = Counter() self.lock = threading.Lock() # Tests may inject a bounded reply; no HTTP endpoint exposes this hook. self.test_reply = None def respond(self, method, target, body): parsed = urllib.parse.urlsplit(target) operation = "rejected" data = None amount = None if method == "POST" and parsed.path == "/api/v1/reservations/searches" and not parsed.query: operation = source.SEARCH value = source.strict_json(body) day = self.fixture["report_date"] source.require(set(value) == {"arrivalStartDate", "arrivalEndDate", "offset", "limit", "orderBy", "sortOrder"} and value["arrivalStartDate"] == value["arrivalEndDate"] == day and value["orderBy"] == ["ConfirmationNo"] and value["sortOrder"] == ["Asc"], "local_search_contract") offset, limit = value["offset"], value["limit"] source.require(type(offset) is int and 0 <= offset <= len(self.ordered) and type(limit) is int and 1 <= limit <= 100, "local_search_bounds") rows = [r["search"] for r in self.ordered[offset:offset + limit]] data = {"reservations": {"reservationInfo": rows, "count": len(rows), "limit": limit, "offset": offset + limit, "totalResults": len(self.ordered), "totalPages": math.ceil(len(self.ordered) / limit), "hasMore": offset + limit < len(self.ordered)}} elif method == "GET" and re.fullmatch(r"/api/v1/reservations/local[0-9]{6}", parsed.path): operation = source.DETAIL source.require(urllib.parse.parse_qsl(parsed.query) == [("fetchInstructions", v) for v in source.FETCH] and not body, "local_detail_contract") record = self.records.get(parsed.path.rsplit("/", 1)[1]) source.require(record is not None, "local_unknown_reservation") data = {"reservations": {"reservation": [record["detail"]]}} elif method == "POST" and parsed.path == rate_info.PATH + "/searches" and not parsed.query: operation = rate_info.POST value = source.strict_json(body) source.require(set(value) == {"id", "type", "summaryInfo", "detailDate"} and value["type"] == "Reservation" and value["summaryInfo"] is False and value["detailDate"] == self.fixture["report_date"], "local_rate_contract") record = self.records.get(value["id"]) source.require(record is not None, "local_unknown_reservation") amount = number(record["rate"]["amount"]) data = {"detail": {"totalRateAmount": "LOCAL_DECIMAL_VALUE", "rateSuppressed": False, "revenue": {"currencyCode": record["rate"]["currency"]}}} else: return 404, b'{"error":"local_route_not_available"}' with self.lock: self.counts[operation] += 1 count = self.counts[operation] envelope = {"operation_id": operation, "hotel_id": HOTEL, "__localSimulation": self.marker, "oracle_request_id": "LOCAL-SIMULATION-NOT-AN-ORACLE-REQUEST", "data": deepcopy(data)} if self.test_reply is not None: reply = self.test_reply(operation, count, envelope) if reply is not None: return reply payload = json.dumps(envelope, ensure_ascii=False, allow_nan=False).encode() if amount is not None: # The rate envelope has no free guest text. Preserve the original decimal # lexeme in a JSON number instead of routing currency through binary float. payload = payload.replace(b'"LOCAL_DECIMAL_VALUE"', amount.encode("ascii"), 1) return 200, payload class LocalTransport: def __init__(self, port, key, source_sha256): source.require(type(port) is int and 1024 <= port <= 65535, "local_api_port") self.authority, self.key, self.source_sha256 = f"127.0.0.1:{port}", key, source_sha256 self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), source.NoRedirect()) def __call__(self, method, path, body): source.require(method in {"GET", "POST"} and path.startswith("/api/v1/reservations/") and not path.startswith("//") and "#" not in path and "\r" not in path and "\n" not in path, "local_api_path") request = urllib.request.Request("http://" + self.authority + path, data=body, method=method, headers={"X-API-Key": self.key, "Content-Type": "application/json"}) try: response = self.opener.open(request, timeout=10) except urllib.error.HTTPError as error: response = error with response: source.require(response.headers.get("X-ARR-Local-Source") == self.source_sha256, "local_api_response_origin_mismatch") raw = response.read(source.MAX_RESPONSE_BYTES + 1) return response.code, dict(response.headers), raw @contextmanager def serve_fixture(fixture): api = LocalAPI(fixture) class Handler(BaseHTTPRequestHandler): def log_message(self, *_args): pass # Paths/bodies may contain business values. Never log them. def handle_request(self): self.connection.settimeout(10) status, raw = 403, b'{"error":"local_access_rejected"}' authority = f"127.0.0.1:{self.server.server_port}" if (self.headers.get("Host") == authority and not self.headers.get("Origin") and secrets.compare_digest(self.headers.get("X-API-Key", ""), api.key)): try: length = int(self.headers.get("Content-Length", "0")) source.require(0 <= length <= 65536 and not self.headers.get("Transfer-Encoding"), "local_body_limit") status, raw = api.respond(self.command, self.path, self.rfile.read(length)) except Exception: status, raw = 400, b'{"error":"local_request_rejected"}' self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.send_header("Cache-Control", "no-store") self.send_header("X-ARR-Local-Source", fixture["source_sha256"]) self.end_headers() self.wfile.write(raw) do_GET = do_POST = do_PUT = do_DELETE = handle_request server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": .05}, daemon=True) thread.start() try: api.transport = LocalTransport(server.server_port, api.key, fixture["source_sha256"]) yield api finally: server.shutdown() server.server_close() thread.join()