Skip to content

security: escape untrusted model output in HTML reports, harden fetch/report paths #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
devin-ai-integration wants to merge 1 commit into
base: main
Choose a base branch
from
Open

security: escape untrusted model output in HTML reports, harden fetch/report paths #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 37 additions & 4 deletions src/code_fetcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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

Expand All @@ -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"""
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -378,14 +389,17 @@ 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", [])

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")

Expand All @@ -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 ----
Expand Down
50 changes: 29 additions & 21 deletions src/report_generator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import html
import json
import re
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
Expand All @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -199,7 +207,7 @@ def _generate_html_report(self, results: List[Dict[str, Any]],
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{metadata.title}</title>
<title>{html.escape(metadata.title)}</title>
<style>
*, *::before, *::after {{ box-sizing: border-box; }}
body {{ font-family: 'Inter', -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; margin: 0; background: #f0f2f5; color: #1a1a2e; line-height: 1.6; }}
Expand Down Expand Up @@ -233,12 +241,12 @@ def _generate_html_report(self, results: List[Dict[str, Any]],
<body>
<div class="page">
<header>
<h1>{metadata.title}</h1>
<div class="meta">{metadata.timestamp.strftime('%Y-%m-%d %H:%M')} &middot; {metadata.analyzer} &middot; v{metadata.version}</div>
<h1>{html.escape(metadata.title)}</h1>
<div class="meta">{metadata.timestamp.strftime('%Y-%m-%d %H:%M')} &middot; {html.escape(metadata.analyzer)} &middot; v{html.escape(metadata.version)}</div>
</header>
<div class="body">
<div class="kpi-row">
<div class="kpi"><div class="num">{summary['overall_status']}</div><div class="lbl">Status</div></div>
<div class="kpi"><div class="num">{html.escape(str(summary['overall_status']))}</div><div class="lbl">Status</div></div>
<div class="kpi"><div class="num">{summary['average_confidence']}%</div><div class="lbl">Confidence</div></div>
<div class="kpi"><div class="num">{summary['files_analyzed']}</div><div class="lbl">Files</div></div>
<div class="kpi"><div class="num" style="color:#e03e3e">{summary['high_severity']}</div><div class="lbl">High</div></div>
Expand All @@ -247,23 +255,23 @@ def _generate_html_report(self, results: List[Dict[str, Any]],
{verified_kpi}
</div>

<div style="background:#f8f9fb;border:1px solid #e8eaed;border-radius:8px;padding:18px 22px;margin-bottom:28px;font-size:0.93em;line-height:1.7;white-space:pre-line;">{self._build_narrative(results, metadata)}</div>
<div style="background:#f8f9fb;border:1px solid #e8eaed;border-radius:8px;padding:18px 22px;margin-bottom:28px;font-size:0.93em;line-height:1.7;white-space:pre-line;">{html.escape(self._build_narrative(results, metadata))}</div>

<h2>Findings</h2>
"""

for result in results:
status = result.get('status', 'UNKNOWN')
status = str(result.get('status', 'UNKNOWN'))
status_color = status_colors.get(status, '#6c757d')

html_content += f"""
<div class="file-card">
<div class="head">
<span class="path">{result.get('file_name', 'Unknown File')}</span>
<span class="badge" style="background:{status_color}">{status}</span>
<span class="path">{html.escape(str(result.get('file_name', 'Unknown File')))}</span>
<span class="badge" style="background:{status_color}">{html.escape(str(status))}</span>
</div>
<div class="content">
<p><strong>Confidence:</strong> {result.get('confidence', 0)}%</p>
<p><strong>Confidence:</strong> {html.escape(str(result.get('confidence', 0)))}%</p>
<p>{html.escape(result.get('summary', 'No summary available.'))}</p>
"""

Expand Down Expand Up @@ -293,15 +301,15 @@ def _generate_html_report(self, results: List[Dict[str, Any]],

html_content += f"""
<div class="footer">
PRSpec v{metadata.version} &middot; {metadata.analyzer} &middot; EIP-{metadata.eip_number} &middot; {metadata.client} &middot; {metadata.timestamp.strftime('%Y-%m-%d %H:%M')}
PRSpec v{html.escape(metadata.version)} &middot; {html.escape(metadata.analyzer)} &middot; EIP-{html.escape(str(metadata.eip_number))} &middot; {html.escape(metadata.client)} &middot; {metadata.timestamp.strftime('%Y-%m-%d %H:%M')}
</div>
</div>
</div>
</body>
</html>
"""

filename = f"prspec_eip{metadata.eip_number}_{metadata.client}_{metadata.timestamp.strftime('%Y%m%d_%H%M%S')}.html"
filename = f"prspec_eip{_slug(metadata.eip_number)}_{_slug(metadata.client)}_{metadata.timestamp.strftime('%Y%m%d_%H%M%S')}.html"
filepath = self.output_dir / filename

with open(filepath, 'w', encoding='utf-8') as f:
Expand Down Expand Up @@ -379,7 +387,7 @@ def _verdict_badge_html(self, issue: Dict[str, Any]) -> str:
f'<div class="detail"><span class="badge" style="background:{color}">'
f'{html.escape(verdict)}</span> '
f'<span style="color:#666;font-size:0.85em">'
f'{v.get("verification_score", 0)}/100 &middot; {grounded}</span></div>'
f'{html.escape(str(v.get("verification_score", 0)))}/100 &middot; {grounded}</span></div>'
)

def _generate_summary(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
Expand Down Expand Up @@ -459,7 +467,7 @@ def generate_differential_report(self, differential: Any,

def _diff_filename(self, differential: Any, ext: str) -> Path:
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
return self.output_dir / f"prspec_diff_eip{differential.eip}_{stamp}.{ext}"
return self.output_dir / f"prspec_diff_eip{_slug(differential.eip)}_{stamp}.{ext}"

def _generate_differential_json(self, differential: Any) -> str:
filepath = self._diff_filename(differential, "json")
Expand Down Expand Up @@ -527,11 +535,11 @@ def _generate_differential_html(self, differential: Any) -> str:
s = d.client_summaries[c]
kpi_cards += (
f'<div class="kpi"><div class="num">{html.escape(c)}</div>'
f'<div class="lbl">{html.escape(s["overall_status"])} '
f'&middot; {s["average_confidence"]}%</div>'
f'<div class="lbl">{s["total_issues"]} issues '
f'(H{s["high_severity"]}/M{s["medium_severity"]}/'
f'L{s["low_severity"]})</div></div>'
f'<div class="lbl">{html.escape(str(s["overall_status"]))} '
f'&middot; {html.escape(str(s["average_confidence"]))}%</div>'
f'<div class="lbl">{html.escape(str(s["total_issues"]))} issues '
f'(H{html.escape(str(s["high_severity"]))}/M{html.escape(str(s["medium_severity"]))}/'
f'L{html.escape(str(s["low_severity"]))})</div></div>'
)

# Comparison table
Expand All @@ -548,7 +556,7 @@ def _generate_differential_html(self, differential: Any) -> str:
rows_html += (
f"<tr><td class='dim'>{html.escape(row.dimension)}</td>{cells}"
f"<td><span class='badge' style='background:{color}'>"
f"{row.verdict}</span></td></tr>"
f"{html.escape(str(row.verdict))}</span></td></tr>"
)

divergences_html = ""
Expand Down
23 changes: 17 additions & 6 deletions src/spec_fetcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,22 @@
"""Fetches Ethereum EIP specs, execution specs, and consensus specs from GitHub."""

import difflib
import re
from pathlib import Path
from typing import Dict, List, Optional

import requests

# Seconds to wait for a spec download 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 SpecFetcher:
"""Fetches Ethereum specifications from GitHub and other sources"""
Expand Down Expand Up @@ -133,15 +144,15 @@ def get_eip_title(cls, eip_number: int) -> str:

def fetch_eip(self, eip_number: int, use_cache: bool = True) -> str:
"""Fetch the raw EIP markdown. Works for any EIP number."""
cache_file = self.cache_dir / f"eip-{eip_number}.md"
cache_file = self.cache_dir / _cache_name(f"eip-{eip_number}.md")

# Check cache
if use_cache and cache_file.exists():
return cache_file.read_text(encoding="utf-8")

# Fetch from GitHub
url = f"https://raw.githubusercontent.com/ethereum/EIPs/master/EIPS/eip-{eip_number}.md"
response = self.session.get(url)
response = self.session.get(url, timeout=DEFAULT_TIMEOUT)
response.raise_for_status()

content = response.text
Expand All @@ -154,13 +165,13 @@ def fetch_eip(self, eip_number: int, use_cache: bool = True) -> str:
def fetch_execution_spec(self, file_path: str, branch: str = "master",
use_cache: bool = True) -> str:
"""Fetch a Python file from ethereum/execution-specs."""
cache_file = self.cache_dir / f"exec_spec_{file_path.replace('/', '_')}"
cache_file = self.cache_dir / _cache_name("exec_spec", branch, file_path)

if use_cache and cache_file.exists():
return cache_file.read_text(encoding="utf-8")

url = f"https://raw.githubusercontent.com/ethereum/execution-specs/{branch}/{file_path}"
response = self.session.get(url)
response = self.session.get(url, timeout=DEFAULT_TIMEOUT)
response.raise_for_status()

content = response.text
Expand All @@ -171,13 +182,13 @@ def fetch_execution_spec(self, file_path: str, branch: str = "master",
def fetch_consensus_spec(self, file_path: str, branch: str = "dev",
use_cache: bool = True) -> str:
"""Fetch a file from ethereum/consensus-specs."""
cache_file = self.cache_dir / f"consensus_spec_{file_path.replace('/', '_')}"
cache_file = self.cache_dir / _cache_name("consensus_spec", branch, file_path)

if use_cache and cache_file.exists():
return cache_file.read_text(encoding="utf-8")

url = f"https://raw.githubusercontent.com/ethereum/consensus-specs/{branch}/{file_path}"
response = self.session.get(url)
response = self.session.get(url, timeout=DEFAULT_TIMEOUT)
response.raise_for_status()

content = response.text
Expand Down
Loading
Loading