from __future__ import annotations import argparse import fcntl import json import os import sys import time import tempfile from dataclasses import asdict from pathlib import Path from urllib.parse import urlsplit from urllib.request import Request, urlopen from .community_sources import ( parse_rf4db_catches, parse_rf4map_point, parse_rf4posts_spot, parse_rf4stat_fishing, parse_rf4stat_posts, ) SOURCES = { "rf4db": ("https://download.rf4db.com/ru/catches", parse_rf4db_catches), "rf4stat-fishing": ("https://rf4-stat.ru/fishing/", parse_rf4stat_fishing), "rf4stat-posts": ("https://rf4-stat.ru/posts/", parse_rf4stat_posts), } DETAIL_SOURCES = { "rf4map-point": parse_rf4map_point, "rf4posts-spot": parse_rf4posts_spot, } USER_AGENT = "RF4-Spotter/0.1 (authorized data integration)" MIN_FETCH_INTERVAL_SECONDS = 30 * 60 DEFAULT_STATE_FILE = Path(".cache/community-fetch-state.json") ALLOWED_HOSTS = frozenset({ "download.rf4db.com", "rf4db.com", "rf4-stat.ru", "rf4map.ru", "rf4-posts.com", }) MAX_RESPONSE_BYTES = 5 * 1024 * 1024 # 5 MB def _read_state(state_file: Path) -> dict: """Read state file with shared lock; return empty dict if missing/corrupt.""" try: with open(state_file, "r") as f: fcntl.flock(f, fcntl.LOCK_SH) try: return json.loads(f.read()) finally: fcntl.flock(f, fcntl.LOCK_UN) except (FileNotFoundError, json.JSONDecodeError, ValueError, OSError): return {} def _write_state(state_file: Path, state: dict) -> None: """A02: Atomic write with exclusive lock, flush before unlock. Writes to temp file first, then renames atomically. Lock is acquired before any file operations to prevent race conditions. """ state_file.parent.mkdir(parents=True, exist_ok=True) temp_file = state_file.with_suffix(".tmp") with open(temp_file, "w") as f: fcntl.flock(f, fcntl.LOCK_EX) try: f.write(json.dumps(state, sort_keys=True)) f.flush() os.fsync(f.fileno()) finally: fcntl.flock(f, fcntl.LOCK_UN) temp_file.replace(state_file) def fetch_site_key(url: str) -> str: """Return a stable cooldown key shared by all endpoints of one site.""" hostname = (urlsplit(url).hostname or "").lower() if hostname.startswith("www."): hostname = hostname[4:] if not hostname: raise ValueError("source URL must include a hostname") return hostname def enforce_fetch_interval( source: str, *, state_file: Path, now: float | None = None, ) -> None: now = time.time() if now is None else now state = _read_state(state_file) last_fetch = state.get(source) if isinstance(last_fetch, (int, float)) and now - last_fetch < MIN_FETCH_INTERVAL_SECONDS: wait = int(MIN_FETCH_INTERVAL_SECONDS - (now - last_fetch)) raise RuntimeError(f"source cooldown is active; retry in {wait} seconds") def mark_fetch(source: str, *, state_file: Path, now: float | None = None) -> None: now = time.time() if now is None else now state = _read_state(state_file) state[source] = now _write_state(state_file, state) MAX_REDIRECT_HOPS = 5 ALLOWED_PORTS = frozenset({443, 80}) def _validate_url_before_io(url: str) -> tuple[str, str]: """A03: Validate scheme, host, port before any network I/O. Returns (normalized_hostname, scheme). Raises ValueError for disallowed schemes, ports, or hosts. """ parsed = urlsplit(url) # Validate scheme if parsed.scheme not in ("http", "https"): raise ValueError(f"URL scheme {parsed.scheme!r} not allowed") if parsed.scheme != "https": raise ValueError("Only HTTPS URLs are allowed") # Validate port if parsed.port is not None and parsed.port not in ALLOWED_PORTS: raise ValueError(f"Port {parsed.port} not in allowed ports {ALLOWED_PORTS}") # Validate hostname hostname = (parsed.hostname or "").lower() if hostname.startswith("www."): hostname = hostname[4:] if not hostname: raise ValueError("URL must include a valid hostname") if hostname not in ALLOWED_HOSTS: raise ValueError(f"URL hostname {hostname} not in allowlist") return hostname, parsed.scheme def _validate_url_host(url: str) -> str: """Legacy alias for _validate_url_before_io (returns hostname only).""" hostname, _ = _validate_url_before_io(url) return hostname def fetch_html(url: str, *, timeout: float = 30, _redirects: int = 0) -> str: """A03: Fetch with pre-I/O validation and redirect hop limit.""" if _redirects > MAX_REDIRECT_HOPS: raise ValueError(f"Redirect chain exceeds {MAX_REDIRECT_HOPS} hops") # Validate BEFORE any network I/O hostname, scheme = _validate_url_before_io(url) request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"}) with urlopen(request, timeout=timeout) as response: final_url = response.url # Validate redirect target final_parsed = urlsplit(final_url) final_hostname = (final_parsed.hostname or "").lower() if final_hostname.startswith("www."): final_hostname = final_hostname[4:] if final_hostname not in ALLOWED_HOSTS: raise ValueError(f"Redirect hostname {final_hostname} not in allowlist") if final_parsed.scheme != "https": raise ValueError("Redirect to non-HTTPS not allowed") if final_parsed.port is not None and final_parsed.port not in ALLOWED_PORTS: raise ValueError(f"Redirect port {final_parsed.port} not allowed") # Recursively follow if redirect (urlopen follows automatically, but we check) if final_url != url: return fetch_html(final_url, timeout=timeout, _redirects=_redirects + 1) if response.headers.get_content_type() != "text/html": raise ValueError(f"expected text/html, got {response.headers.get_content_type()}") data = response.read(MAX_RESPONSE_BYTES + 1) if len(data) > MAX_RESPONSE_BYTES: raise ValueError("response exceeded 5MB limit") return data.decode(response.headers.get_content_charset() or "utf-8") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Fetch one authorized RF4 community source page") parser.add_argument("source", choices=(*SOURCES, *DETAIL_SOURCES)) parser.add_argument("--url", help="Override the configured public page URL") parser.add_argument("--limit", type=int, default=100, choices=range(1, 501), metavar="1..500") parser.add_argument( "--state-file", type=Path, default=Path(os.environ.get("RF4_COMMUNITY_FETCH_STATE", DEFAULT_STATE_FILE)), help="Persistent per-source cooldown state", ) args = parser.parse_args(argv) if args.source in DETAIL_SOURCES and not args.url: parser.error(f"--url is required for {args.source}") default_url, parse = SOURCES.get(args.source, (None, DETAIL_SOURCES.get(args.source))) url = args.url or default_url try: site_key = fetch_site_key(url) enforce_fetch_interval(site_key, state_file=args.state_file) # Reserve before network I/O: failed attempts count toward the limit too. mark_fetch(site_key, state_file=args.state_file) html = fetch_html(url) records = (parse(html, source_url=url) if args.source in DETAIL_SOURCES else parse(html))[:args.limit] except Exception as exc: print(f"community source failed: {exc}", file=sys.stderr) return 1 print(json.dumps([asdict(item) for item in records], ensure_ascii=False, default=str)) return 0 if __name__ == "__main__": raise SystemExit(main())