From e75240f58b0b6a9626423de754a712039b18d3a1 Mon Sep 17 00:00:00 2001 From: Fosurero Date: Fri, 24 Jul 2026 22:03:16 +0000 Subject: [PATCH] security: escape untrusted model output in reports, harden fetch paths --- src/code_fetcher.py | 41 +++++++++++-- src/report_generator.py | 50 +++++++++------- src/spec_fetcher.py | 23 ++++++-- tests/test_report_security.py | 108 ++++++++++++++++++++++++++++++++++ 4 files changed, 191 insertions(+), 31 deletions(-) create mode 100644 tests/test_report_security.py diff --git a/src/code_fetcher.py b/src/code_fetcher.py index c0e10dd..3490ecc 100644 --- a/src/code_fetcher.py +++ b/src/code_fetcher.py @@ -1,8 +1,10 @@ """Fetches implementation files from Ethereum client repos (geth, Nethermind, Besu).""" +import re import tempfile from pathlib import Path from typing import Any, Dict, List, Optional +from urllib.parse import urlparse import requests @@ -12,6 +14,16 @@ except ImportError: GIT_AVAILABLE = False +# Seconds to wait for a GitHub request before giving up. +DEFAULT_TIMEOUT = 30 + +_UNSAFE_CACHE_CHARS = re.compile(r"[^A-Za-z0-9._-]") + + +def _cache_name(*parts: str) -> str: + """Build a flat, traversal-safe file name from arbitrary path parts.""" + return "_".join(_UNSAFE_CACHE_CHARS.sub("_", str(p)) for p in parts) + class CodeFetcher: """Fetches code from Ethereum client implementations""" @@ -291,15 +303,14 @@ def supported_eips_for_client(cls, client: str) -> List[int]: def fetch_file(self, owner: str, repo: str, path: str, branch: str = "master", use_cache: bool = True) -> str: """Fetch a single file from a GitHub repo via raw URL.""" - cache_key = f"{owner}_{repo}_{path.replace('/', '_')}_{branch}" - cache_file = self.cache_dir / cache_key + cache_file = self.cache_dir / _cache_name(owner, repo, path, branch) if use_cache and cache_file.exists(): return cache_file.read_text(encoding="utf-8") # Use raw GitHub URL url = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{path}" - response = self.session.get(url) + response = self.session.get(url, timeout=DEFAULT_TIMEOUT) response.raise_for_status() content = response.text @@ -378,7 +389,7 @@ def search_repository(self, owner: str, repo: str, query: str, url = "https://api.github.com/search/code" params = {"q": search_query, "per_page": 10} - response = self.session.get(url, params=params) + response = self.session.get(url, params=params, timeout=DEFAULT_TIMEOUT) response.raise_for_status() return response.json().get("items", []) @@ -386,6 +397,9 @@ def search_repository(self, owner: str, repo: str, query: str, def clone_repository(self, url: str, target_dir: Optional[str] = None, branch: str = "master", shallow: bool = True) -> str: """Clone a repo locally for deeper analysis. Requires gitpython.""" + self._validate_clone_url(url) + self._validate_branch(branch) + if not GIT_AVAILABLE: raise RuntimeError("GitPython not installed. Install with: pip install gitpython") @@ -400,6 +414,25 @@ def clone_repository(self, url: str, target_dir: Optional[str] = None, return target_dir + @staticmethod + def _validate_clone_url(url: str) -> None: + """Reject clone URLs that git would treat as an option or a transport + capable of running commands (``ext::``, ``--upload-pack=``, ...).""" + if not url or url.startswith("-"): + raise ValueError(f"Invalid repository URL: {url!r}") + scheme = urlparse(url).scheme.lower() + if scheme not in ("https", "http", "ssh", "git"): + raise ValueError( + f"Unsupported repository URL scheme: {scheme or 'none'!r}. " + "Use an https, ssh, or git URL." + ) + + @staticmethod + def _validate_branch(branch: str) -> None: + """Allow only plain ref names, so a branch cannot become a git flag.""" + if not branch or not re.fullmatch(r"[A-Za-z0-9._/-]+", branch) or branch.startswith("-"): + raise ValueError(f"Invalid branch name: {branch!r}") + # get_file_functions() removed — use CodeParser for function extraction. # ---- Cache management ---- diff --git a/src/report_generator.py b/src/report_generator.py index 539589a..7a19193 100644 --- a/src/report_generator.py +++ b/src/report_generator.py @@ -2,6 +2,7 @@ import html import json +import re from dataclasses import dataclass from datetime import datetime from pathlib import Path @@ -19,6 +20,13 @@ from src import __version__ +_UNSAFE_FILENAME_CHARS = re.compile(r"[^A-Za-z0-9._-]") + + +def _slug(value: Any) -> str: + """Filename-safe rendering of a report identifier (EIP, client name, ...).""" + return _UNSAFE_FILENAME_CHARS.sub("_", str(value)) + @dataclass class ReportMetadata: @@ -75,7 +83,7 @@ def _generate_json_report(self, results: List[Dict[str, Any]], "results": results } - filename = f"prspec_eip{metadata.eip_number}_{metadata.client}_{metadata.timestamp.strftime('%Y%m%d_%H%M%S')}.json" + filename = f"prspec_eip{_slug(metadata.eip_number)}_{_slug(metadata.client)}_{metadata.timestamp.strftime('%Y%m%d_%H%M%S')}.json" filepath = self.output_dir / filename with open(filepath, 'w', encoding='utf-8') as f: @@ -164,7 +172,7 @@ def _generate_markdown_report(self, results: List[Dict[str, Any]], *Generated by PRSpec v{metadata.version}* """ - filename = f"prspec_eip{metadata.eip_number}_{metadata.client}_{metadata.timestamp.strftime('%Y%m%d_%H%M%S')}.md" + filename = f"prspec_eip{_slug(metadata.eip_number)}_{_slug(metadata.client)}_{metadata.timestamp.strftime('%Y%m%d_%H%M%S')}.md" filepath = self.output_dir / filename with open(filepath, 'w', encoding='utf-8') as f: @@ -199,7 +207,7 @@ def _generate_html_report(self, results: List[Dict[str, Any]], - {metadata.title} + {html.escape(metadata.title)}