import os import json import re import time import gzip import urllib.parse import mimetypes import http.server import threading from typing import Any, Optional from backend.config import settings from backend.logging_config import _log from backend.scraper.enrichment import get_laptops_data, get_similar_laptops, get_price_history, get_stock_history, get_last_updated_seconds_ago, get_activity, get_scrape_stats, get_featured_price_history_subject from backend.scraper.catalog_state import catalog_state from backend.db.connection import get_db_connection # Validation gate in front of an already-parameterized IN (?,?,...) clause, not the # actual SQL-safety mechanism — underscore included as cheap insurance even though no # real product_code in this codebase has been observed with one. SKU_RE = re.compile(r'^[A-Za-z0-9_-]+$') # Body-size ceiling shared by every POST/DELETE handler that reads a JSON body. MAX_BODY_BYTES = 1_000_000 # Per-endpoint cooldown for /api/push/test — the endpoint is unauthenticated by # design (no user accounts), so this is the only thing standing between it and # an on-demand spam/SSRF-trigger primitive (fable audit B-SEC-7). Module-level # dict is fine: single process, worst case is a slightly generous window across # a restart. PUSH_TEST_COOLDOWN_SECONDS = 60 _push_test_last_sent: dict[str, float] = {} _push_test_lock = threading.Lock() class LaptopTrackerHandler(http.server.SimpleHTTPRequestHandler): def end_headers(self): # GET is world-readable (public catalog data) regardless of Origin. Every # other method (POST/DELETE, plus the OPTIONS preflight browsers send ahead # of them) only gets CORS clearance for an allowlisted frontend origin — # otherwise a wildcard here would let any website's JS drive admin/push # mutation endpoints using a visitor's browser (fable audit B-SEC-2). if self.command == "GET": self.send_header('Access-Control-Allow-Origin', '*') else: origin = self.headers.get("Origin") if origin and origin in settings.FRONTEND_ORIGINS: self.send_header('Access-Control-Allow-Origin', origin) self.send_header('Vary', 'Origin') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS, DELETE') self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization, X-API-Key') super().end_headers() def do_OPTIONS(self): self.send_response(200) self.end_headers() def send_json(self, data: Any, status: int = 200) -> None: """Sends a JSON response (Arch 5 helper), gzip-compressed when the client advertises support. /api/laptops alone is ~927KB uncompressed and every poll-cycle fetch re-pays that cost; gzip gets a ~10x reduction on JSON this repetitive (fable audit B-PERF-1).""" body = json.dumps(data).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Vary", "Accept-Encoding") if "gzip" in self.headers.get("Accept-Encoding", "") and len(body) > 512: body = gzip.compress(body) self.send_header("Content-Encoding", "gzip") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def send_error_json(self, status_code: int, message: str): self.send_json({"error": message}, status_code) def _read_body(self, max_bytes: int = MAX_BODY_BYTES) -> Optional[bytes]: """Reads the request body, rejecting oversized payloads before touching self.rfile.read(). Returns None (and has already sent a response) if there is no body or it exceeds max_bytes — callers should return immediately.""" try: content_length = int(self.headers.get("Content-Length", 0)) except ValueError: self.send_error_json(400, "Invalid Content-Length header") return None if content_length <= 0: self.send_error_json(400, "Missing request body") return None if content_length > max_bytes: self.send_error_json(413, "Request body too large") return None return self.rfile.read(content_length) def is_authorized(self) -> bool: """Verifies if the client has administrative privileges.""" if not settings.ADMIN_API_KEY: # Fail closed: an unset key must never mean "anyone is admin". A # misconfigured deploy (missing/typo'd env var) would otherwise expose # /api/scrape, /api/clean_ghosts, /api/enrich_specs, /api/enrich_psref # unauthenticated to the internet. _log("WARNING: ADMIN_API_KEY is not set — admin endpoints are locked out until it is configured.") return False auth_header = self.headers.get("Authorization") if auth_header: if auth_header.startswith("Bearer "): token = auth_header.split(" ", 1)[1] if token == settings.ADMIN_API_KEY: return True api_key_header = self.headers.get("X-API-Key") if api_key_header == settings.ADMIN_API_KEY: return True return False def do_GET(self): parsed_url = urllib.parse.urlparse(self.path) path = parsed_url.path query_params = urllib.parse.parse_qs(parsed_url.query) # Route GET API endpoints if path == "/api/health": self.send_json({"status": "ok"}) return elif path == "/api/laptops": codes = query_params.get("code") product_code = codes[0] if codes else None similar_to_codes = query_params.get("similar_to") similar_to = similar_to_codes[0] if similar_to_codes else None if product_code and not SKU_RE.match(product_code): self.send_error_json(400, f"Invalid code: {product_code}") return if similar_to and not SKU_RE.match(similar_to): self.send_error_json(400, f"Invalid code: {similar_to}") return try: if similar_to: # Server-side D-00014 match (screen-size + price proximity), # not the full catalog — see enrichment.py::get_similar_laptops. similar = get_similar_laptops(similar_to) if similar is None: self.send_error_json(404, f"Product not found: {similar_to}") return self.send_json(similar) return data = get_laptops_data(product_code=product_code) if product_code: # Single-product form: {} not [item] or [] — a distinct shape # from the full-catalog list so callers can't confuse "not # found" with "empty catalog". if not data: self.send_error_json(404, f"Product not found: {product_code}") return self.send_json(data[0]) else: self.send_json(data) except Exception as e: self.send_error_json(500, f"Database error: {str(e)}") return elif path == "/api/filters": try: db_dir = os.path.dirname(os.path.abspath(settings.DB_FILE)) facet_groups_path = os.path.join(db_dir, "facet_groups.json") if os.path.exists(facet_groups_path): with open(facet_groups_path, "r", encoding="utf-8") as fg_file: groups = json.load(fg_file) if groups and isinstance(groups, list) and len(groups) > 0: self.send_json(groups[0].get("facets", [])) return self.send_json([]) except Exception as e: self.send_error_json(500, f"Error loading filters: {str(e)}") return elif path == "/api/price_history": codes = query_params.get("code") if not codes: self.send_error_json(400, "Missing required query parameter: code") return product_code = codes[0] try: data = get_price_history(product_code) self.send_json(data) except Exception as e: self.send_error_json(500, f"Database error: {str(e)}") return elif path == "/api/stock_history": codes = query_params.get("code") if not codes: self.send_error_json(400, "Missing required query parameter: code") return product_code = codes[0] try: data = get_stock_history(product_code) self.send_json(data) except Exception as e: self.send_error_json(500, f"Database error: {str(e)}") return elif path == "/api/price_history/featured": # Landing page (v2): the product with the richest price-history record, # current or historical — not tied to any single SKU, so no `code` param. try: self.send_json(get_featured_price_history_subject()) except Exception as e: self.send_error_json(500, f"Database error: {str(e)}") return elif path == "/api/vapid_public_key": self.send_json({"publicKey": settings.VAPID_PUBLIC_KEY or ""}) return elif path == "/api/config": self.send_json({ "vapidPublicKey": settings.VAPID_PUBLIC_KEY or "", "discordInvite": settings.DISCORD_INVITE_URL or "", "telegramChannel": settings.TELEGRAM_CHANNEL_URL or "" }) return elif path == "/api/stats": # Phase 9 (D-00085/D-00112): last scrape cycle's listed-vs-purchasable # snapshot. `null` before the first post-migration-12 scrape cycle — # the frontend hides the stat rather than fabricating one. try: self.send_json(get_scrape_stats()) except Exception as e: self.send_error_json(500, f"Database error: {str(e)}") return elif path == "/api/last_updated": try: latest_stock_id, latest_price_id = catalog_state.snapshot() self.send_json({ "secondsAgo": get_last_updated_seconds_ago(), "latestStockId": latest_stock_id, "latestPriceId": latest_price_id, }) except Exception as e: self.send_error_json(500, f"Database error: {str(e)}") return elif path == "/api/activity": limit_raw = query_params.get("limit", ["200"])[0] try: limit = int(limit_raw) except ValueError: self.send_error_json(400, "limit must be an integer") return if limit < 1 or limit > 500: self.send_error_json(400, "limit must be between 1 and 500") return skus: Optional[list[str]] = None skus_raw = query_params.get("skus", [None])[0] if skus_raw: skus = [s for s in skus_raw.split(",") if s] for sku in skus: if not SKU_RE.match(sku): self.send_error_json(400, f"Invalid sku: {sku}") return try: data = get_activity(limit=limit, skus=skus) self.send_json(data) except Exception as e: self.send_error_json(500, f"Database error: {str(e)}") return # Any /api/* path that fell through the routing above is an unknown/mistyped # endpoint, not a client-side route — 404 it here rather than letting it hit # the static/SPA fallback below, which would otherwise serve index.html with # a 200 (a frontend caller gets an HTML parse error far from the real cause). if path.startswith("/api/"): self.send_error_json(404, f"Endpoint not found: {path}") return # Serve static assets from build directory (P0 Sec Fix 1 & P1 FE Fix 2) rel_path = path.lstrip('/') norm_path = os.path.normpath(rel_path) if norm_path.startswith("..") or os.path.isabs(norm_path): self.send_error(403, "Access Forbidden") return static_dir_str = str(settings.STATIC_DIR) if not norm_path or norm_path == "." or norm_path == "index.html": static_file_path = os.path.join(static_dir_str, "index.html") else: static_file_path = os.path.join(static_dir_str, norm_path) if os.path.exists(static_file_path) and os.path.isfile(static_file_path): abs_static = os.path.abspath(static_file_path) abs_dir = os.path.abspath(static_dir_str) if not abs_static.startswith(abs_dir): self.send_error(403, "Access Forbidden") return mime, _ = mimetypes.guess_type(static_file_path) if not mime: mime = "application/octet-stream" self.serve_file(static_file_path, mime) else: # SPA Fallback for client-side routing index_path = os.path.join(static_dir_str, "index.html") if os.path.exists(index_path): self.serve_file(index_path, "text/html") else: super().do_GET() def do_POST(self): parsed_url = urllib.parse.urlparse(self.path) path = parsed_url.path # Protect administrative endpoints if path in ("/api/scrape", "/api/clean_ghosts", "/api/enrich_specs", "/api/enrich_psref"): if not self.is_authorized(): self.send_error_json(401, "Unauthorized") return # Scrape API launches the scan in a background thread and returns # immediately — a full multi-page Lenovo scrape (INTER_PAGE_DELAY=1.5s per # page + retries) can run for minutes, and running it synchronously in the # request thread is timeout bait for any reverse proxy in front, with the # client learning nothing until the end (fable audit B-ARC-2). Same # backgrounding pattern already used by /api/enrich_specs below. # trigger_manual_scrape() itself stays synchronous — the Discord bot calls # it directly (off its own thread via asyncio.to_thread) and depends on # the returned scan stats for its followup message. if path == "/api/scrape": # Late imports to avoid circular dependency from backend.main import scraper_lock, trigger_manual_scrape if scraper_lock.locked(): self.send_error_json(409, "A scrape or database update is already in progress.") return def _run(): res = trigger_manual_scrape() if not res.get("success"): _log(f"[API] Background scrape failed: {res.get('error')}") threading.Thread(target=_run, daemon=True).start() self.send_json( {"success": True, "message": "Scrape started in background. Reload data in a few minutes."}, status=202, ) return # Clean ghosts API triggers a cleanup under lock elif path == "/api/clean_ghosts": # Late imports to avoid circular dependency from backend.main import trigger_ghost_cleanup res = trigger_ghost_cleanup() if res.get("success"): self.send_json(res) else: self.send_error_json(409 if "progress" in res.get("error", "") else 500, res.get("error")) return # Enrich specs API launches background thread under lock elif path == "/api/enrich_specs": try: # Late imports to avoid circular dependencies from backend.main import scraper_lock from backend.scraper.enrichment import enrich_specs _log("In-process spec enrichment triggered via API...") def _run(): with scraper_lock: enrich_specs() t = threading.Thread(target=_run, daemon=True) t.start() payload = {"success": True, "message": "Spec enrichment started in background. Reload data in ~30 seconds."} self.send_json(payload) except Exception as e: self.send_error_json(500, f"Failed to start spec enrichment: {str(e)}") return # PSREF backfill API launches background thread under lock — one-off, # cached-forever enrichment (see backend/scraper/psref.py), same pattern # as /api/enrich_specs above. elif path == "/api/enrich_psref": try: from backend.main import scraper_lock from backend.scraper.psref import backfill_psref_specs _log("In-process PSREF spec enrichment triggered via API...") def _run(): with scraper_lock: backfill_psref_specs() t = threading.Thread(target=_run, daemon=True) t.start() payload = {"success": True, "message": "PSREF enrichment started in background."} self.send_json(payload) except Exception as e: self.send_error_json(500, f"Failed to start PSREF enrichment: {str(e)}") return elif path == "/api/push/subscribe": try: body = self._read_body() if body is None: return data = json.loads(body.decode("utf-8")) endpoint = data.get("endpoint") keys = data.get("keys", {}) p256dh = keys.get("p256dh") auth = keys.get("auth") user_agent = data.get("user_agent") if not endpoint or not p256dh or not auth: self.send_error_json(400, "Missing required subscription fields (endpoint, p256dh, auth)") return # Real browser push endpoints are always https:// capability URLs. # Rejecting anything else closes off registering an attacker-chosen # host (e.g. an internal address) that later gets POSTed to by # send_push/`/api/push/test` (fable audit B-SEC-5/B-SEC-7). if not endpoint.startswith("https://"): self.send_error_json(400, "endpoint must be an https:// URL") return conn = get_db_connection() try: cursor = conn.cursor() cursor.execute("SELECT id FROM push_subscriptions WHERE endpoint = ?", (endpoint,)) row = cursor.fetchone() now_str = time.strftime("%Y-%m-%d %H:%M:%S") if row: sub_id = row[0] cursor.execute(""" UPDATE push_subscriptions SET p256dh = ?, auth = ?, user_agent = ?, last_used = ? WHERE id = ? """, (p256dh, auth, user_agent, now_str, sub_id)) else: cursor.execute(""" INSERT INTO push_subscriptions ( endpoint, p256dh, auth, user_agent, created_at, last_used ) VALUES (?, ?, ?, ?, ?, ?) """, (endpoint, p256dh, auth, user_agent, now_str, now_str)) sub_id = cursor.lastrowid conn.commit() self.send_json({"success": True, "subscription_id": sub_id}) finally: conn.close() except Exception as e: self.send_error_json(500, f"Error subscribing: {str(e)}") return elif path == "/api/push/update_watchlist": try: body = self._read_body() if body is None: return data = json.loads(body.decode("utf-8")) endpoint = data.get("endpoint") sub_id = data.get("subscription_id") watchlist = data.get("watchlist", []) if not endpoint and not sub_id: self.send_error_json(400, "Missing identifier (endpoint or subscription_id)") return # /api/activity already gates its `skus` param on SKU_RE; this endpoint # didn't (fable audit B-SEC-5) — a non-list watchlist (e.g. a JSON string) # would insert one row per character, and unvalidated codes are just # garbage rows (queries are parameterized, so not an injection risk, but # still worth rejecting at the door). if not isinstance(watchlist, list) or not all( isinstance(code, str) and SKU_RE.match(code) for code in watchlist ): self.send_error_json(400, "watchlist must be a list of valid product codes") return conn = get_db_connection() try: cursor = conn.cursor() if sub_id is None: cursor.execute("SELECT id FROM push_subscriptions WHERE endpoint = ?", (endpoint,)) row = cursor.fetchone() if not row: self.send_error_json(404, "Subscription not found") return sub_id = row[0] now_str = time.strftime("%Y-%m-%d %H:%M:%S") # Sync: full replace cursor.execute("DELETE FROM subscription_watchlist WHERE subscription_id = ?", (sub_id,)) for code in watchlist: cursor.execute(""" INSERT OR IGNORE INTO subscription_watchlist (subscription_id, product_code, created_at) VALUES (?, ?, ?) """, (sub_id, code, now_str)) conn.commit() self.send_json({"success": True}) finally: conn.close() except Exception as e: self.send_error_json(500, f"Error updating watchlist: {str(e)}") return elif path == "/api/push/update_prefs": try: body = self._read_body() if body is None: return data = json.loads(body.decode("utf-8")) endpoint = data.get("endpoint") sub_id = data.get("subscription_id") notify_price_drops = 1 if data.get("notify_price_drops", True) else 0 notify_new_listings = 1 if data.get("notify_new_listings", True) else 0 notify_back_in_stock = 1 if data.get("notify_back_in_stock", True) else 0 notify_watchlist_only = 1 if data.get("notify_watchlist_only", False) else 0 min_drop_percent = float(data.get("min_drop_percent", 0.0)) if not endpoint and not sub_id: self.send_error_json(400, "Missing identifier (endpoint or subscription_id)") return conn = get_db_connection() try: cursor = conn.cursor() if sub_id is not None: cursor.execute(""" UPDATE push_subscriptions SET notify_price_drops = ?, notify_new_listings = ?, notify_back_in_stock = ?, notify_watchlist_only = ?, min_drop_percent = ? WHERE id = ? """, (notify_price_drops, notify_new_listings, notify_back_in_stock, notify_watchlist_only, min_drop_percent, sub_id)) else: cursor.execute(""" UPDATE push_subscriptions SET notify_price_drops = ?, notify_new_listings = ?, notify_back_in_stock = ?, notify_watchlist_only = ?, min_drop_percent = ? WHERE endpoint = ? """, (notify_price_drops, notify_new_listings, notify_back_in_stock, notify_watchlist_only, min_drop_percent, endpoint)) conn.commit() self.send_json({"success": True}) finally: conn.close() except Exception as e: self.send_error_json(500, f"Error updating preferences: {str(e)}") return elif path == "/api/push/test": try: body = self._read_body() if body is None: return data = json.loads(body.decode("utf-8")) endpoint = data.get("endpoint") if not endpoint: self.send_error_json(400, "Missing endpoint") return with _push_test_lock: last_sent = _push_test_last_sent.get(endpoint) now = time.time() if last_sent is not None and now - last_sent < PUSH_TEST_COOLDOWN_SECONDS: self.send_error_json(429, "Please wait before requesting another test push.") return _push_test_last_sent[endpoint] = now conn = get_db_connection() try: cursor = conn.cursor() cursor.execute("SELECT id, p256dh, auth FROM push_subscriptions WHERE endpoint = ?", (endpoint,)) row = cursor.fetchone() if not row: self.send_error_json(404, "Subscription not found") return sub_id, p256dh, auth = row[0], row[1], row[2] from backend.notifier.push import send_push payload = { "event_type": "test", "title": "Test Push Notification", "body": "Your TrackFurb push notification subscription is working!", "product_code": "TEST", "price": 0, "save_percent": 0 } success = send_push(sub_id, endpoint, p256dh, auth, payload) self.send_json({"success": success}) finally: conn.close() except Exception as e: self.send_error_json(500, f"Error testing push: {str(e)}") return self.send_error_json(404, "Endpoint not found") def do_DELETE(self): parsed_url = urllib.parse.urlparse(self.path) path = parsed_url.path if path == "/api/push/unsubscribe": try: body = self._read_body() if body is None: return data = json.loads(body.decode("utf-8")) endpoint = data.get("endpoint") if not endpoint: self.send_error_json(400, "Missing endpoint") return conn = get_db_connection() try: cursor = conn.cursor() cursor.execute("DELETE FROM push_subscriptions WHERE endpoint = ?", (endpoint,)) conn.commit() self.send_json({"success": True, "message": "Successfully unsubscribed"}) finally: conn.close() except Exception as e: self.send_error_json(500, f"Error unsubscribing: {str(e)}") return self.send_error_json(404, "Endpoint not found") def serve_file(self, file_path: str, content_type: str): try: file_size = os.path.getsize(file_path) self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(file_size)) # Add cache control for hashed assets under assets/ if "assets/" in file_path: self.send_header("Cache-Control", "public, max-age=31536000, immutable") self.end_headers() with open(file_path, "rb") as f: import shutil shutil.copyfileobj(f, self.wfile) except Exception as e: _log(f"Error serving file {os.path.basename(file_path)}: {str(e)}")