fix: audit P0-P1 — T03-T07, D01-D08, U01-U02 (#1788956171115)

rf4_research/community_cli.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+import fcntl
 import json
 import os
 import sys
@@ -31,6 +32,13 @@ DETAIL_SOURCES = {
 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 fetch_site_key(url: str) -> str:
@@ -47,10 +55,12 @@ def enforce_fetch_interval(
     source: str, *, state_file: Path, now: float | None = None,
 ) -> None:
     now = time.time() if now is None else now
-    try:
-        state = json.loads(state_file.read_text(encoding="utf-8"))
-    except (FileNotFoundError, json.JSONDecodeError, OSError):
-        state = {}
+    with open(state_file, "r") as f:
+        fcntl.flock(f, fcntl.LOCK_SH)
+        try:
+            state = json.loads(f.read(encoding="utf-8"))
+        finally:
+            fcntl.flock(f, fcntl.LOCK_UN)
     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))
@@ -59,23 +69,35 @@ def enforce_fetch_interval(
 
 def mark_fetch(source: str, *, state_file: Path, now: float | None = None) -> None:
     now = time.time() if now is None else now
-    try:
-        state = json.loads(state_file.read_text(encoding="utf-8"))
-    except (FileNotFoundError, json.JSONDecodeError, OSError):
-        state = {}
-    state[source] = now
     state_file.parent.mkdir(parents=True, exist_ok=True)
-    temporary = state_file.with_suffix(".tmp")
-    temporary.write_text(json.dumps(state, sort_keys=True), encoding="utf-8")
-    temporary.replace(state_file)
+    with open(state_file, "r+") as f:
+        fcntl.flock(f, fcntl.LOCK_EX)
+        try:
+            try:
+                state = json.loads(f.read(encoding="utf-8"))
+            except (json.JSONDecodeError, ValueError):
+                state = {}
+            state[source] = now
+            f.seek(0)
+            f.truncate()
+            f.write(json.dumps(state, sort_keys=True))
+        finally:
+            fcntl.flock(f, fcntl.LOCK_UN)
 
 
 def fetch_html(url: str, *, timeout: float = 30) -> str:
     request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "text/html"})
     with urlopen(request, timeout=timeout) as response:
+        final_url = response.url
+        hostname = (urlsplit(final_url).hostname or "").lower()
+        if hostname not in ALLOWED_HOSTS:
+            raise ValueError(f"URL hostname {hostname} not in allowlist")
         if response.headers.get_content_type() != "text/html":
             raise ValueError(f"expected text/html, got {response.headers.get_content_type()}")
-        return response.read().decode(response.headers.get_content_charset() or "utf-8")
+        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: