Security hardening: XSS prevention, input validation, rate limiting

- HTML-escape agent_name/agent_url in rendered links (blocks javascript: URIs)
- Validate numeric field types, string lengths, URL schemes
- Sanitize CSV string values to prevent formula injection
- Add global rate limit (20 submissions/hour)
- Replay file size limit (10 MB), sanitized filenames
- Safe regex fallback in search to prevent ReDoS
- 24 new security tests (68 total)

app.py

@@ -11,8 +11,12 @@ Deploy on HuggingFace Spaces:
 """
 
 import csv
+import html
 import json
 import os
+import re
+import time
+from collections import defaultdict
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -49,6 +53,29 @@ DISPLAY_COLUMNS = [
 ]
 
 
+def _safe_agent_link(name: str, url) -> str:
+    """Render agent name, optionally as a hyperlink. HTML-escaped to prevent XSS."""
+    safe_name = html.escape(str(name))
+    if pd.notna(url) and str(url).strip():
+        url_str = str(url).strip()
+        # Only allow http/https URLs — block javascript:, data:, etc.
+        if url_str.startswith(("http://", "https://")):
+            safe_url = html.escape(url_str, quote=True)
+            return f'<a href="{safe_url}" target="_blank" rel="noopener">{safe_name}</a>'
+    return safe_name
+
+
+def _safe_replay_link(url) -> str:
+    """Render replay download link. Filename is sanitized to prevent XSS."""
+    if pd.notna(url) and str(url).strip():
+        # Sanitize: only allow alphanumeric, dash, underscore, dot
+        safe_name = re.sub(r"[^a-zA-Z0-9._-]", "", str(url).strip())
+        if safe_name:
+            escaped = html.escape(safe_name, quote=True)
+            return f'<a href="/replays/{escaped}" download title="Download replay">&#11015;</a>'
+    return ""
+
+
 def load_data() -> pd.DataFrame:
     """Load leaderboard data from CSV."""
     if not DATA_PATH.exists():
@@ -58,28 +85,18 @@ def load_data() -> pd.DataFrame:
     df = df.sort_values("score", ascending=False).reset_index(drop=True)
     df.insert(0, "Rank", range(1, len(df) + 1))
 
-    # Build agent name with optional hyperlink
+    # Build agent name with optional hyperlink (XSS-safe)
     if "agent_url" in df.columns:
         df["Agent"] = df.apply(
-            lambda r: (
-                f'<a href="{r["agent_url"]}" target="_blank">{r["agent_name"]}</a>'
-                if pd.notna(r.get("agent_url")) and str(r["agent_url"]).strip()
-                else r["agent_name"]
-            ),
+            lambda r: _safe_agent_link(r.get("agent_name", ""), r.get("agent_url", "")),
             axis=1,
         )
     else:
-        df["Agent"] = df["agent_name"]
+        df["Agent"] = df["agent_name"].apply(lambda n: html.escape(str(n)))
 
-    # Build replay download link
+    # Build replay download link (XSS-safe)
     if "replay_url" in df.columns:
-        df["Replay"] = df["replay_url"].apply(
-            lambda u: (
-                f'<a href="/replays/{u}" download title="Download replay">&#11015;</a>'
-                if pd.notna(u) and str(u).strip()
-                else ""
-            )
-        )
+        df["Replay"] = df["replay_url"].apply(_safe_replay_link)
     else:
         df["Replay"] = ""
 
@@ -136,12 +153,17 @@ def filter_leaderboard(
     if opponent and opponent != "All":
         df = df[df["Opponent"] == opponent]
 
-    # Search by agent name (regex)
+    # Search by agent name (regex with fallback to literal on invalid patterns)
     if search and search.strip():
         patterns = [p.strip() for p in search.split(",") if p.strip()]
         mask = pd.Series([False] * len(df), index=df.index)
         for pattern in patterns:
-            mask |= df["Agent"].str.contains(pattern, case=False, regex=True, na=False)
+            try:
+                mask |= df["Agent"].str.contains(pattern, case=False, regex=True, na=False)
+            except re.error:
+                mask |= df["Agent"].str.contains(
+                    re.escape(pattern), case=False, regex=True, na=False
+                )
         df = df[mask]
 
     # Re-rank after filtering
@@ -173,6 +195,32 @@ if os.environ.get("HF_TOKEN") and os.environ.get("SPACE_ID"):
         pass  # Running locally without HF token — skip
 
 
+def _sanitize_csv_value(val):
+    """Strip leading characters that trigger formula execution in spreadsheets."""
+    if isinstance(val, str):
+        while val and val[0] in ("=", "+", "-", "@", "\t", "\r", "\n"):
+            val = val[1:]
+        val = val.replace("\n", " ").replace("\r", " ")
+    return val
+
+
+# ── Rate Limiting ────────────────────────────────────────────────────────────
+
+_submit_times: dict[str, list[float]] = defaultdict(list)
+MAX_SUBMITS_PER_HOUR = 20
+
+
+def _check_rate_limit(identifier: str = "global") -> tuple[bool, str]:
+    """Simple in-memory rate limiter. Returns (allowed, error_message)."""
+    now = time.time()
+    times = _submit_times[identifier]
+    _submit_times[identifier] = [t for t in times if now - t < 3600]
+    if len(_submit_times[identifier]) >= MAX_SUBMITS_PER_HOUR:
+        return False, "Rate limit exceeded (max 20 submissions per hour). Try again later."
+    _submit_times[identifier].append(now)
+    return True, ""
+
+
 def save_submission(results: dict) -> None:
     """Append results to local JSONL and CSV."""
     # JSONL for CommitScheduler → HF dataset
@@ -188,15 +236,18 @@ def save_submission(results: dict) -> None:
         "score", "avg_kills", "avg_deaths", "kd_ratio", "avg_economy",
         "avg_game_length", "timestamp", "replay_url", "agent_url",
     ]
+    safe_results = {k: _sanitize_csv_value(v) for k, v in results.items()}
     with open(csv_path, "a", newline="") as f:
         writer = csv.DictWriter(f, fieldnames=fieldnames)
         if not file_exists:
             writer.writeheader()
-        writer.writerow(results)
+        writer.writerow(safe_results)
 
 
 # ── Submission Handling ───────────────────────────────────────────────────────
 
+MAX_REPLAY_SIZE = 10 * 1024 * 1024  # 10 MB
+
 VALID_OPPONENTS = {"Beginner", "Easy", "Medium", "Normal", "Hard"}
 VALID_AGENT_TYPES = {"Scripted", "LLM", "RL"}
 REQUIRED_FIELDS = [
@@ -226,6 +277,22 @@ def validate_submission(data: dict) -> tuple[bool, str]:
             f"Must be one of: {', '.join(sorted(VALID_OPPONENTS))}"
         )
 
+    # Type checks for numeric fields
+    for field in ("ticks", "kills_cost", "deaths_cost", "assets_value"):
+        if not isinstance(data[field], (int, float)):
+            return False, f"Field '{field}' must be a number"
+
+    # String length limits
+    if len(str(data["agent_name"])) > 100:
+        return False, "agent_name must be 100 characters or fewer"
+
+    # agent_url: optional, but must be http(s) if provided
+    agent_url = str(data.get("agent_url", "")).strip()
+    if agent_url and not agent_url.startswith(("http://", "https://")):
+        return False, "agent_url must be an HTTP(S) URL"
+    if len(agent_url) > 500:
+        return False, "agent_url must be 500 characters or fewer"
+
     return True, ""
 
 
@@ -269,6 +336,10 @@ def handle_upload(json_file, replay_file) -> tuple[str, pd.DataFrame]:
     if json_file is None:
         return "Please upload a JSON file.", add_type_badges(load_data())
 
+    allowed, err = _check_rate_limit()
+    if not allowed:
+        return err, add_type_badges(load_data())
+
     try:
         with open(json_file.name) as f:
             data = json.load(f)
@@ -284,8 +355,14 @@ def handle_upload(json_file, replay_file) -> tuple[str, pd.DataFrame]:
     # Save replay if provided
     if replay_file is not None:
         import shutil
-        replay_name = Path(replay_file.name).name
-        shutil.copy2(replay_file.name, SUBMISSIONS_DIR / replay_name)
+
+        orig = Path(replay_file.name)
+        if orig.stat().st_size > MAX_REPLAY_SIZE:
+            return "Replay file too large (max 10 MB).", add_type_badges(load_data())
+        ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+        slug = re.sub(r"[^a-zA-Z0-9_-]", "", data["agent_name"].replace("/", "_").replace(" ", "_"))[:30]
+        replay_name = f"replay-{slug}-{ts}.orarep"
+        shutil.copy2(str(orig), SUBMISSIONS_DIR / replay_name)
         results_row["replay_url"] = replay_name
 
     save_submission(results_row)
@@ -299,6 +376,10 @@ def handle_upload(json_file, replay_file) -> tuple[str, pd.DataFrame]:
 
 def handle_api_submit(json_data: str) -> str:
     """API endpoint: accept JSON string submission. Used by CLI auto-upload."""
+    allowed, err = _check_rate_limit()
+    if not allowed:
+        return err
+
     try:
         data = json.loads(json_data)
     except (json.JSONDecodeError, Exception) as e:
@@ -319,6 +400,10 @@ def handle_api_submit(json_data: str) -> str:
 
 def handle_api_submit_with_replay(json_data: str, replay_file) -> str:
     """API endpoint: accept JSON + replay file. Used by CLI with --replay."""
+    allowed, err = _check_rate_limit()
+    if not allowed:
+        return err
+
     try:
         data = json.loads(json_data)
     except (json.JSONDecodeError, Exception) as e:
@@ -333,11 +418,12 @@ def handle_api_submit_with_replay(json_data: str, replay_file) -> str:
     # Save replay if provided
     if replay_file is not None:
         import shutil
-        from datetime import datetime, timezone
 
         orig = Path(replay_file) if isinstance(replay_file, str) else Path(replay_file.name)
+        if orig.exists() and orig.stat().st_size > MAX_REPLAY_SIZE:
+            return "Replay file too large (max 10 MB)"
         ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
-        slug = data["agent_name"].replace("/", "_").replace(" ", "_")[:30]
+        slug = re.sub(r"[^a-zA-Z0-9_-]", "", data["agent_name"].replace("/", "_").replace(" ", "_"))[:30]
         replay_name = f"replay-{slug}-{ts}.orarep"
         shutil.copy2(str(orig), SUBMISSIONS_DIR / replay_name)
         results_row["replay_url"] = replay_name

tests/test_app.py

@@ -12,7 +12,13 @@ sys.path.insert(0, str(Path(__file__).parent.parent))
 from app import (
     AGENT_TYPE_COLORS,
     DISPLAY_COLUMNS,
+    MAX_SUBMITS_PER_HOUR,
     VALID_OPPONENTS,
+    _check_rate_limit,
+    _safe_agent_link,
+    _safe_replay_link,
+    _sanitize_csv_value,
+    _submit_times,
     add_type_badges,
     build_app,
     filter_leaderboard,
@@ -308,3 +314,165 @@ class TestReplayColumn:
             # The default test data has no replay
             replay_val = df["Replay"].iloc[0]
             assert replay_val == "" or not str(replay_val).strip()
+
+
+class TestXssPrevention:
+    """Test that user input is HTML-escaped to prevent XSS."""
+
+    def test_javascript_url_blocked(self):
+        """javascript: URLs should NOT produce a clickable link."""
+        result = _safe_agent_link("Bot", "javascript:alert(1)")
+        assert "javascript:" not in result
+        assert "Bot" in result
+
+    def test_data_url_blocked(self):
+        result = _safe_agent_link("Bot", "data:text/html,<script>alert(1)</script>")
+        assert "data:" not in result
+
+    def test_html_in_name_escaped(self):
+        result = _safe_agent_link('<script>alert("xss")</script>', "")
+        assert "<script>" not in result
+        assert "&lt;script&gt;" in result
+
+    def test_quote_injection_in_url_escaped(self):
+        result = _safe_agent_link("Bot", 'https://ok.com" onclick="alert(1)')
+        assert 'onclick' not in result or '&quot;' in result
+
+    def test_valid_https_url_works(self):
+        result = _safe_agent_link("Bot", "https://github.com/user/repo")
+        assert '<a href="https://github.com/user/repo"' in result
+        assert 'rel="noopener"' in result
+
+    def test_replay_link_sanitized(self):
+        result = _safe_replay_link('"><script>alert(1)</script>.orarep')
+        assert "<script>" not in result
+
+    def test_replay_path_traversal_stripped(self):
+        """Path traversal characters (/) are stripped from replay filenames."""
+        result = _safe_replay_link("replay/../../../etc/passwd")
+        # The href after /replays/ should have no slashes (traversal stripped)
+        href_part = result.split('href="')[1].split('"')[0]
+        filename = href_part.replace("/replays/", "")
+        assert "/" not in filename
+
+
+class TestInputValidation:
+    """Test stricter input validation."""
+
+    def _valid_data(self):
+        return {
+            "agent_name": "TestBot",
+            "agent_type": "LLM",
+            "opponent": "Beginner",
+            "result": "loss",
+            "ticks": 27000,
+            "kills_cost": 1000,
+            "deaths_cost": 2900,
+            "assets_value": 9050,
+        }
+
+    def test_string_ticks_rejected(self):
+        data = self._valid_data()
+        data["ticks"] = "not a number"
+        valid, err = validate_submission(data)
+        assert not valid
+        assert "must be a number" in err
+
+    def test_dict_kills_rejected(self):
+        data = self._valid_data()
+        data["kills_cost"] = {"nested": True}
+        valid, err = validate_submission(data)
+        assert not valid
+
+    def test_long_agent_name_rejected(self):
+        data = self._valid_data()
+        data["agent_name"] = "A" * 101
+        valid, err = validate_submission(data)
+        assert not valid
+        assert "100 characters" in err
+
+    def test_javascript_agent_url_rejected(self):
+        data = self._valid_data()
+        data["agent_url"] = "javascript:alert(1)"
+        valid, err = validate_submission(data)
+        assert not valid
+        assert "HTTP(S)" in err
+
+    def test_valid_agent_url_accepted(self):
+        data = self._valid_data()
+        data["agent_url"] = "https://github.com/user/repo"
+        valid, _ = validate_submission(data)
+        assert valid
+
+    def test_empty_agent_url_accepted(self):
+        data = self._valid_data()
+        data["agent_url"] = ""
+        valid, _ = validate_submission(data)
+        assert valid
+
+    def test_long_agent_url_rejected(self):
+        data = self._valid_data()
+        data["agent_url"] = "https://example.com/" + "a" * 500
+        valid, err = validate_submission(data)
+        assert not valid
+        assert "500 characters" in err
+
+
+class TestCsvSanitization:
+    """Test CSV injection prevention."""
+
+    def test_formula_trigger_stripped(self):
+        assert _sanitize_csv_value("=cmd|'/c calc'!A0") == "cmd|'/c calc'!A0"
+
+    def test_plus_trigger_stripped(self):
+        assert _sanitize_csv_value("+cmd") == "cmd"
+
+    def test_at_trigger_stripped(self):
+        assert _sanitize_csv_value("@SUM(A1)") == "SUM(A1)"
+
+    def test_newlines_replaced(self):
+        assert _sanitize_csv_value("line1\nline2\rline3") == "line1 line2 line3"
+
+    def test_normal_string_unchanged(self):
+        assert _sanitize_csv_value("DeathBot-9000") == "DeathBot-9000"
+
+    def test_numbers_unchanged(self):
+        assert _sanitize_csv_value(42) == 42
+        assert _sanitize_csv_value(3.14) == 3.14
+
+
+class TestRateLimiting:
+    """Test rate limiting on submissions."""
+
+    def test_rate_limit_allows_normal_usage(self):
+        _submit_times.clear()
+        allowed, _ = _check_rate_limit("test_normal")
+        assert allowed
+
+    def test_rate_limit_blocks_after_max(self):
+        _submit_times.clear()
+        key = "test_flood"
+        for _ in range(MAX_SUBMITS_PER_HOUR):
+            allowed, _ = _check_rate_limit(key)
+            assert allowed
+        allowed, err = _check_rate_limit(key)
+        assert not allowed
+        assert "Rate limit" in err
+
+    def test_rate_limit_resets_after_expiry(self):
+        import time as _time
+        _submit_times.clear()
+        key = "test_expiry"
+        # Fill with old timestamps
+        _submit_times[key] = [_time.time() - 3601] * MAX_SUBMITS_PER_HOUR
+        allowed, _ = _check_rate_limit(key)
+        assert allowed
+
+
+class TestSearchSafety:
+    """Test that malformed regex doesn't crash the search."""
+
+    def test_invalid_regex_falls_back(self):
+        """An invalid regex pattern should not raise an exception."""
+        df = filter_leaderboard("[invalid(regex", [], "All")
+        assert isinstance(df, pd.DataFrame)