#!/usr/bin/env python3 """Compile and optionally run a make-agnostic TinyFish vehicle-evidence plan. Input is a carrier-approved claim packet, not a BMW-specific argument list. The runner validates vehicle fields, builds official-source queries and API routes, filters Search results to the packet's domain allowlist, and asks TinyFish Fetch to read only the selected URLs. It stores metadata and hashes, never credentials or fetched page bodies. Reality verification and the adjuster's decision remain downstream steps; this program does not authenticate a vehicle or decide a claim. """ from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import re import time from typing import Any import urllib.error import urllib.parse import urllib.request SEARCH_ENDPOINT = "https://api.search.tinyfish.ai/" FETCH_ENDPOINT = "https://api.fetch.tinyfish.ai/" RETRYABLE_HTTP = {429, 500, 503} SAFE_HEADERS = {"content-type", "date", "etag", "last-modified", "x-request-id", "request-id", "traceparent"} VIN_RE = re.compile(r"^[A-HJ-NPR-Z0-9*]{8,17}$") DOMAIN_RE = re.compile(r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,63}$") VIN_VALUES = { **{str(value): value for value in range(10)}, **dict(zip("ABCDEFGHJKLMNPRSTUVWXYZ", [1, 2, 3, 4, 5, 6, 7, 8, 1, 2, 3, 4, 5, 7, 9, 2, 3, 4, 5, 6, 7, 8, 9])), } VIN_WEIGHTS = [8, 7, 6, 5, 4, 3, 2, 10, 0, 9, 8, 7, 6, 5, 4, 3, 2] def canonical_json(value: Any) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() def sha256(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def clean_text(value: Any, field: str, *, required: bool = False, limit: int = 100) -> str: text = " ".join(str(value or "").strip().split()) if required and not text: raise ValueError(f"{field} is required") if len(text) > limit or any(ord(char) < 32 for char in text): raise ValueError(f"{field} is invalid") return text def normalize_domain(value: Any) -> str: raw = clean_text(value, "allowed domain", required=True).casefold() if "://" in raw: raw = urllib.parse.urlsplit(raw).hostname or "" raw = raw.strip("./") if not DOMAIN_RE.fullmatch(raw): raise ValueError(f"invalid allowed domain: {value!r}") return raw def complete_vin_checksum_valid(vin: str) -> bool: """Validate the North American VIN check digit without making a network call.""" if len(vin) != 17 or "*" in vin or any(char not in VIN_VALUES for char in vin): return False remainder = sum(VIN_VALUES[char] * weight for char, weight in zip(vin, VIN_WEIGHTS)) % 11 expected = "X" if remainder == 10 else str(remainder) return vin[8] == expected def validate_packet(raw: dict[str, Any]) -> dict[str, Any]: vehicle = raw.get("vehicle") or {} year = vehicle.get("model_year") if not isinstance(year, int) or not 1886 <= year <= datetime.now().year + 2: raise ValueError("vehicle.model_year must be a plausible integer") vin_input = clean_text(vehicle.get("vin_or_pattern"), "vehicle.vin_or_pattern").upper() if vin_input and not VIN_RE.fullmatch(vin_input): raise ValueError("vehicle.vin_or_pattern must contain 8–17 VIN characters; I, O and Q are invalid") complete_vin_supplied = len(vin_input) == 17 and "*" not in vin_input if complete_vin_supplied and not complete_vin_checksum_valid(vin_input): raise ValueError("complete VIN check digit is invalid") carrier_vehicle_ref = clean_text(vehicle.get("carrier_vehicle_ref"), "vehicle.carrier_vehicle_ref", limit=128) if complete_vin_supplied and not carrier_vehicle_ref: raise ValueError("vehicle.carrier_vehicle_ref is required when a complete VIN is supplied") make = clean_text(vehicle.get("make"), "vehicle.make", required=True) model = clean_text(vehicle.get("model"), "vehicle.model", required=True) trim = clean_text(vehicle.get("trim"), "vehicle.trim") source_policy = raw.get("source_policy") or {} allowed_domains = list(dict.fromkeys(normalize_domain(item) for item in source_policy.get("allowed_domains") or [])) if not {"nhtsa.gov", "vpic.nhtsa.dot.gov"}.issubset(set(allowed_domains)): raise ValueError("source_policy.allowed_domains must include nhtsa.gov and vpic.nhtsa.dot.gov") oem_domains = list(dict.fromkeys(normalize_domain(item) for item in source_policy.get("oem_domains") or [])) if any(not any(domain == allowed or domain.endswith("." + allowed) for allowed in allowed_domains) for domain in oem_domains): raise ValueError("every OEM domain must also be covered by allowed_domains") max_fetch_urls = source_policy.get("max_fetch_urls", 6) if not isinstance(max_fetch_urls, int) or not 1 <= max_fetch_urls <= 20: raise ValueError("source_policy.max_fetch_urls must be an integer from 1 to 20") allow_public_vin_pattern = source_policy.get("allow_public_vin_pattern", False) if not isinstance(allow_public_vin_pattern, bool): raise ValueError("source_policy.allow_public_vin_pattern must be boolean") public_vin_pattern = vin_input if vin_input and not complete_vin_supplied and allow_public_vin_pattern else "" return { "schema_version": raw.get("schema_version"), "case_id": clean_text(raw.get("case_id"), "case_id", required=True), "purpose": clean_text(raw.get("purpose"), "purpose", required=True, limit=500), "vehicle": { "public_vin_pattern": public_vin_pattern, "model_year": year, "make": make, "model": model, "trim": trim, }, "carrier_boundary": { "carrier_vehicle_ref": carrier_vehicle_ref or None, "complete_vin_received": complete_vin_supplied, "complete_vin_checksum_valid": True if complete_vin_supplied else None, "complete_vin_excluded_from_public_plan": True, }, "source_policy": { "allowed_domains": allowed_domains, "oem_domains": oem_domains, "max_fetch_urls": max_fetch_urls, "allow_public_vin_pattern": allow_public_vin_pattern, }, } def quote(value: str) -> str: return f'"{value}"' if " " in value else value def build_plan(packet: dict[str, Any]) -> dict[str, Any]: vehicle = packet["vehicle"] year, make, model, trim = vehicle["model_year"], vehicle["make"], vehicle["model"], vehicle["trim"] descriptor = " ".join(str(item) for item in (year, make, model, trim) if item) queries: list[dict[str, str]] = [] if vehicle["public_vin_pattern"]: queries.append({"kind": "vin_descriptor", "query": f'NHTSA vPIC "{vehicle["public_vin_pattern"]}" {year} {quote(make)}'}) queries.append({"kind": "model_year_safety", "query": f"site:nhtsa.gov {descriptor} recalls"}) for domain in packet["source_policy"]["oem_domains"]: queries.append({"kind": "manufacturer_context", "query": f"site:{domain} {descriptor}"}) routes: list[dict[str, str]] = [] if vehicle["public_vin_pattern"]: routes.append({ "kind": "nhtsa_vpic_decode", "url": "https://vpic.nhtsa.dot.gov/api/vehicles/DecodeVinValuesExtended/" + urllib.parse.quote(vehicle["public_vin_pattern"], safe="*") + "?format=json&modelyear=" + str(year), }) routes.append({ "kind": "nhtsa_model_year_recalls", "url": "https://api.nhtsa.gov/recalls/recallsByVehicle?" + urllib.parse.urlencode({"make": make, "model": model, "modelYear": year}), }) return { "schema_version": "reality.tinyfish-vehicle-evidence-plan/v1", "case_id": packet["case_id"], "purpose": packet["purpose"], "vehicle": vehicle, "carrier_boundary": packet["carrier_boundary"], "search_requests": [{"method": "GET", "endpoint": SEARCH_ENDPOINT, **item} for item in queries], "deterministic_official_routes": routes, "selection_policy": { **packet["source_policy"], "rule": "select only HTTPS results whose hostname equals or is a subdomain of an allowed domain", }, "downstream": { "carrier": "resolve the complete VIN and policy only inside the customer-owned adapter using carrier_vehicle_ref", "reality": "verify source provenance, packet attachment, support/conflict/unresolved state, and authorized file integrity", "human": "review evidence and decide the claim; no automated payment, denial, fraud label, or ownership conclusion", }, } def safe_headers(response: Any) -> dict[str, str]: return {key.casefold(): value for key, value in response.headers.items() if key.casefold() in SAFE_HEADERS} def request_json(request_factory: Any, *, timeout: int, max_attempts: int = 3) -> tuple[dict[str, Any], dict[str, Any]]: errors: list[dict[str, Any]] = [] for attempt in range(1, max_attempts + 1): started = time.perf_counter() try: with urllib.request.urlopen(request_factory(), timeout=timeout) as response: raw = response.read() return json.loads(raw.decode(errors="ignore"), strict=False), { "attempt": attempt, "http_status": response.status, "final_url": response.url, "elapsed_ms": round((time.perf_counter() - started) * 1000, 2), "response_headers": safe_headers(response), "raw_response_sha256": sha256(raw), "raw_response_bytes": len(raw), "prior_errors": errors, } except urllib.error.HTTPError as exc: retryable = exc.code in RETRYABLE_HTTP errors.append({"attempt": attempt, "http_status": exc.code, "retryable": retryable}) if not retryable or attempt == max_attempts: raise except (urllib.error.URLError, TimeoutError): errors.append({"attempt": attempt, "http_status": None, "retryable": True}) if attempt == max_attempts: raise time.sleep(min(2 ** (attempt - 1), 4)) raise RuntimeError("retry loop exited unexpectedly") def hostname_allowed(url: str, domains: list[str]) -> bool: parsed = urllib.parse.urlsplit(url) host = (parsed.hostname or "").casefold() return parsed.scheme == "https" and any(host == domain or host.endswith("." + domain) for domain in domains) def run_live(plan: dict[str, Any], credential: str) -> dict[str, Any]: searches = [] candidates: list[str] = [route["url"] for route in plan["deterministic_official_routes"]] for request_plan in plan["search_requests"]: encoded = urllib.parse.urlencode({"query": request_plan["query"], "purpose": plan["purpose"]}) payload, transport = request_json( lambda: urllib.request.Request(SEARCH_ENDPOINT + "?" + encoded, headers={"X-API-Key": credential}), timeout=60, ) results = [] for position, item in enumerate(payload.get("results") or [], start=1): url = str(item.get("url") or "") accepted = hostname_allowed(url, plan["selection_policy"]["allowed_domains"]) results.append({ "position": position, "title": item.get("title"), "site_name": item.get("site_name"), "url": url, "accepted_by_domain_policy": accepted, "snippet_sha256": sha256(str(item.get("snippet") or "").encode()), }) if accepted: candidates.append(url) searches.append({"kind": request_plan["kind"], "query": request_plan["query"], "transport": transport, "results": results}) selected = list(dict.fromkeys(url for url in candidates if hostname_allowed(url, plan["selection_policy"]["allowed_domains"])))[: plan["selection_policy"]["max_fetch_urls"]] body = canonical_json({"urls": selected, "format": "markdown", "links": True, "image_links": True}) payload, transport = request_json( lambda: urllib.request.Request(FETCH_ENDPOINT, data=body, headers={"X-API-Key": credential, "Content-Type": "application/json"}, method="POST"), timeout=180, ) fetched = [] for item in payload.get("results") or []: text = str(item.get("text") or "") links = list(item.get("links") or []) image_links = list(item.get("image_links") or []) fetched.append({ "url": item.get("url"), "final_url": item.get("final_url"), "title": item.get("title"), "text_chars": len(text), "text_sha256": sha256(text.encode()) if text else None, "links_count": len(links), "links_sha256": sha256(canonical_json(links)), "image_links_count": len(image_links), "image_links_sha256": sha256(canonical_json(image_links)), }) return { "schema_version": "reality.tinyfish-vehicle-evidence-receipt/v1", "captured_at": datetime.now(timezone.utc).isoformat(), "mode": "live_search_fetch", "plan": plan, "search": searches, "fetch": {"transport": transport, "selected_urls": selected, "results": fetched, "errors": payload.get("errors") or []}, "assurance": { "domain_allowlist_enforced": True, "credential_persisted": False, "fetched_page_bodies_persisted": False, "tinyfish_agent_or_browser_called": False, "reality_verdict_made": False, "claim_decision_made": False, }, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("packet", type=Path) parser.add_argument("--dry-run", action="store_true", help="compile the request plan without network access or credentials") parser.add_argument("--output", type=Path) parser.add_argument("--force", action="store_true") args = parser.parse_args() if args.output and args.output.exists() and not args.force: raise SystemExit(f"refusing to overwrite {args.output}; pass --force") packet = validate_packet(json.loads(args.packet.read_text())) plan = build_plan(packet) if args.dry_run: result = {"mode": "dry_run_no_network", "plan": plan, "assurance": {"network_called": False, "credential_required": False}} else: credential = os.environ.get("TINYFISH_API_KEY") if not credential: raise SystemExit("TINYFISH_API_KEY is required for live mode; use --dry-run to inspect the plan") result = run_live(plan, credential) rendered = json.dumps(result, ensure_ascii=False, indent=2) + "\n" if args.output: args.output.write_text(rendered) else: print(rendered, end="") if __name__ == "__main__": main()