diff --git a/src/analyzer.py b/src/analyzer.py index 2ae2b4d..a456322 100644 --- a/src/analyzer.py +++ b/src/analyzer.py @@ -148,6 +148,27 @@ def _build_refutation_prompt(self, finding: Dict[str, Any], spec_text: str, {claim} """ + def _result_from_payload(self, payload: Dict[str, Any], + raw_response: Optional[str] = None) -> AnalysisResult: + """Build an :class:`AnalysisResult` from a parsed LLM JSON payload.""" + return AnalysisResult( + status=payload.get("status", "UNCERTAIN"), + confidence=payload.get("confidence", 0), + issues=payload.get("issues", []), + summary=payload.get("summary", ""), + raw_response=raw_response, + ) + + @staticmethod + def _error_result(provider: str, error: Exception) -> AnalysisResult: + """Build the ERROR result returned when a backend call fails.""" + return AnalysisResult( + status="ERROR", + confidence=0, + issues=[], + summary=f"{provider} analysis failed: {str(error)}", + ) + def _parse_json_response(self, response_text: str) -> Dict[str, Any]: """Parse JSON from LLM response, handling markdown code blocks and truncated output from the model.""" @@ -251,21 +272,10 @@ def analyze_compliance(self, spec_text: str, code_text: str, result = self._parse_json_response(response.text) - return AnalysisResult( - status=result.get("status", "UNCERTAIN"), - confidence=result.get("confidence", 0), - issues=result.get("issues", []), - summary=result.get("summary", ""), - raw_response=response.text - ) + return self._result_from_payload(result, response.text) except Exception as e: - return AnalysisResult( - status="ERROR", - confidence=0, - issues=[], - summary=f"Gemini analysis failed: {str(e)}" - ) + return self._error_result("Gemini", e) def get_model_info(self) -> Dict[str, Any]: """Get information about the current model""" @@ -318,21 +328,10 @@ def analyze_compliance(self, spec_text: str, code_text: str, response_text = response.choices[0].message.content result = self._parse_json_response(response_text) - return AnalysisResult( - status=result.get("status", "UNCERTAIN"), - confidence=result.get("confidence", 0), - issues=result.get("issues", []), - summary=result.get("summary", ""), - raw_response=response_text - ) + return self._result_from_payload(result, response_text) except Exception as e: - return AnalysisResult( - status="ERROR", - confidence=0, - issues=[], - summary=f"OpenAI analysis failed: {str(e)}" - ) + return self._error_result("OpenAI", e) def get_model_info(self) -> Dict[str, Any]: """Get information about the current model""" @@ -453,21 +452,10 @@ def analyze_compliance(self, spec_text: str, code_text: str, ) result = self._parse_json_response(response_text) - return AnalysisResult( - status=result.get("status", "UNCERTAIN"), - confidence=result.get("confidence", 0), - issues=result.get("issues", []), - summary=result.get("summary", ""), - raw_response=response_text - ) + return self._result_from_payload(result, response_text) except Exception as e: - return AnalysisResult( - status="ERROR", - confidence=0, - issues=[], - summary=f"Azure AI analysis failed: {str(e)}" - ) + return self._error_result("Azure AI", e) def get_model_info(self) -> Dict[str, Any]: """Get information about the current model""" @@ -479,32 +467,25 @@ def get_model_info(self) -> Dict[str, Any]: } +# Analyzer class and its required constructor arguments, per provider. +_PROVIDERS = { + "gemini": (GeminiAnalyzer, ["api_key"]), + "openai": (OpenAIAnalyzer, ["api_key"]), + "azure": (AzureAIAnalyzer, ["api_key", "endpoint", "model"]), +} + + def get_analyzer(provider: str = "gemini", **kwargs) -> BaseAnalyzer: """Factory: return a GeminiAnalyzer, OpenAIAnalyzer, or AzureAIAnalyzer.""" provider = provider.lower() - if provider == "gemini": - required = ["api_key"] - for key in required: - if key not in kwargs: - raise ValueError(f"Missing required parameter: {key}") - return GeminiAnalyzer(**kwargs) - - elif provider == "openai": - required = ["api_key"] - for key in required: - if key not in kwargs: - raise ValueError(f"Missing required parameter: {key}") - return OpenAIAnalyzer(**kwargs) - - elif provider == "azure": - required = ["api_key", "endpoint", "model"] - for key in required: - if key not in kwargs: - raise ValueError(f"Missing required parameter: {key}") - return AzureAIAnalyzer(**kwargs) - - else: + if provider not in _PROVIDERS: raise ValueError( f"Unknown provider: {provider}. Use 'gemini', 'openai', or 'azure'." ) + + analyzer_cls, required = _PROVIDERS[provider] + for key in required: + if key not in kwargs: + raise ValueError(f"Missing required parameter: {key}") + return analyzer_cls(**kwargs) diff --git a/src/cli.py b/src/cli.py index 9bf9dbe..3e517fc 100644 --- a/src/cli.py +++ b/src/cli.py @@ -42,6 +42,35 @@ def cli(): pass +def _print_info_panel(title: str, rows): + """Render the banner plus a two-column key/value configuration panel.""" + console.print(BANNER) + info_table = Table(show_header=False, box=None, padding=(0, 2)) + info_table.add_column(style="bold white") + info_table.add_column(style="cyan") + for label, value in rows: + info_table.add_row(label, value) + console.print(Panel(info_table, title=f"[bold]{title}[/bold]", border_style="blue")) + + +def _abort_on_error(error: Exception, verbose: bool): + """Print a command failure (with optional traceback) and abort.""" + if RICH_AVAILABLE: + console.print(f"[red]Error:[/red] {str(error)}") + else: + click.echo(f"Error: {str(error)}", err=True) + + if verbose: + import traceback + trace = traceback.format_exc() + if RICH_AVAILABLE: + console.print(f"[dim]{trace}[/dim]") + else: + click.echo(trace, err=True) + + raise click.Abort() + + def _analyze_one_file(analyzer, spec_text, file_path, code_content, context): """Analyze a single file — designed to run inside a thread pool.""" result = analyzer.analyze_compliance(spec_text, code_content, context) @@ -180,16 +209,13 @@ def analyze(eip: int, client: str, provider: Optional[str], output: str, # Banner + config summary if RICH_AVAILABLE: - console.print(BANNER) - info_table = Table(show_header=False, box=None, padding=(0, 2)) - info_table.add_column(style="bold white") - info_table.add_column(style="cyan") - info_table.add_row("EIP", str(eip)) - info_table.add_row("Client", client) - info_table.add_row("Provider", llm_provider) - info_table.add_row("Output", output) - info_table.add_row("Verify", f"on · {verify_rounds} rounds" if verify else "off") - console.print(Panel(info_table, title="[bold]Configuration[/bold]", border_style="blue")) + _print_info_panel("Configuration", [ + ("EIP", str(eip)), + ("Client", client), + ("Provider", llm_provider), + ("Output", output), + ("Verify", f"on · {verify_rounds} rounds" if verify else "off"), + ]) else: click.echo("\n PRSpec - Ethereum Specification Compliance Checker\n") click.echo(f" EIP: {eip} | Client: {client} | Provider: {llm_provider}") @@ -250,17 +276,7 @@ def on_file_done(fname): click.echo(f"\nReport saved to: {report_path}") except Exception as e: - if RICH_AVAILABLE: - console.print(f"[red]Error:[/red] {str(e)}") - if verbose: - import traceback - console.print(f"[dim]{traceback.format_exc()}[/dim]") - else: - click.echo(f"Error: {str(e)}", err=True) - if verbose: - import traceback - click.echo(traceback.format_exc(), err=True) - raise click.Abort() + _abort_on_error(e, verbose) @cli.command() @@ -287,7 +303,7 @@ def diff(eip: int, clients: Optional[str], provider: Optional[str], output: str, prspec diff --eip 1559 prspec diff --eip 4844 --clients go-ethereum,nethermind,besu --output html """ - from .differential import ClientAnalysis, DifferentialEngine + from .differential import analyze_clients try: cfg = Config(config) @@ -315,47 +331,29 @@ def diff(eip: int, clients: Optional[str], provider: Optional[str], output: str, f"Usable: {usable or 'none'}." ) + rows = [ + ("EIP", str(eip)), + ("Clients", ", ".join(usable)), + ("Provider", llm_provider), + ("Output", output), + ] if RICH_AVAILABLE: - console.print(BANNER) - info_table = Table(show_header=False, box=None, padding=(0, 2)) - info_table.add_column(style="bold white") - info_table.add_column(style="cyan") - info_table.add_row("EIP", str(eip)) - info_table.add_row("Clients", ", ".join(usable)) - info_table.add_row("Provider", llm_provider) - info_table.add_row("Output", output) if skipped: - info_table.add_row("Skipped", ", ".join(skipped)) - console.print(Panel(info_table, title="[bold]Differential[/bold]", border_style="blue")) + rows.append(("Skipped", ", ".join(skipped))) + _print_info_panel("Differential", rows) else: click.echo(f"\n PRSpec differential — EIP-{eip} across {', '.join(usable)}\n") - # Analyze each client through the standard pipeline. - per_client = {} - last_analyzer = None - for client in usable: + def on_client_start(client): if RICH_AVAILABLE: console.print(f"[dim]Analyzing {client}...[/dim]") - results, analyzer = _run_analysis( - eip, client, cfg, llm_provider, - verify=verify, verify_rounds=verify_rounds, - ) - per_client[client] = ClientAnalysis( - client=client, - language=CodeFetcher.client_language(client), - results=results, - ) - last_analyzer = analyzer - - # Build the differential. - engine = DifferentialEngine(focus_areas=cfg.get_eip_focus_areas(eip)) - eip_title = SpecFetcher.get_eip_title(eip) - differential = engine.build(per_client, eip, eip_title, confirmed_only=verify) - if llm_synthesis and last_analyzer is not None: - differential.llm_synthesis = engine.synthesize( - last_analyzer, differential, per_client - ) + differential = analyze_clients( + eip, usable, cfg, provider=llm_provider, + use_llm_synthesis=llm_synthesis, + verify=verify, verify_rounds=verify_rounds, + on_client_start=on_client_start, + ) # Report. report_gen = ReportGenerator(cfg.output_config.get("directory", "output")) @@ -371,17 +369,7 @@ def diff(eip: int, clients: Optional[str], provider: Optional[str], output: str, except click.ClickException: raise except Exception as e: - if RICH_AVAILABLE: - console.print(f"[red]Error:[/red] {str(e)}") - if verbose: - import traceback - console.print(f"[dim]{traceback.format_exc()}[/dim]") - else: - click.echo(f"Error: {str(e)}", err=True) - if verbose: - import traceback - click.echo(traceback.format_exc(), err=True) - raise click.Abort() + _abort_on_error(e, verbose) @cli.command() diff --git a/src/code_fetcher.py b/src/code_fetcher.py index c0e10dd..4506000 100644 --- a/src/code_fetcher.py +++ b/src/code_fetcher.py @@ -1,11 +1,12 @@ """Fetches implementation files from Ethereum client repos (geth, Nethermind, Besu).""" import tempfile -from pathlib import Path from typing import Any, Dict, List, Optional import requests +from .github_fetcher import CachedGitHubFetcher + try: from git import Repo GIT_AVAILABLE = True @@ -13,9 +14,11 @@ GIT_AVAILABLE = False -class CodeFetcher: +class CodeFetcher(CachedGitHubFetcher): """Fetches code from Ethereum client implementations""" + DEFAULT_CACHE_DIRNAME = ".code_cache" + # Client repos and per-EIP file paths. # "branch" defaults to "master" when absent. CLIENTS: Dict[str, Dict[str, Any]] = { @@ -253,16 +256,9 @@ class CodeFetcher: def __init__(self, github_token: Optional[str] = None, cache_dir: Optional[str] = None): """Set up HTTP session and local cache directory.""" - self.github_token = github_token - self.cache_dir = Path(cache_dir) if cache_dir else Path.cwd() / ".code_cache" - self.session = requests.Session() - - if github_token: - self.session.headers["Authorization"] = f"token {github_token}" + super().__init__(github_token=github_token, cache_dir=cache_dir) self.session.headers["Accept"] = "application/vnd.github.v3+json" - self.cache_dir.mkdir(parents=True, exist_ok=True) - # ---- Helpers ---- @classmethod @@ -291,21 +287,11 @@ 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 - - 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.raise_for_status() - - content = response.text - cache_file.write_text(content, encoding="utf-8") - - return content + return self.fetch_raw_file( + owner, repo, path, branch, + cache_key=f"{owner}_{repo}_{path.replace('/', '_')}_{branch}", + use_cache=use_cache, + ) def fetch_geth_file(self, path: str, branch: str = "master", use_cache: bool = True) -> str: @@ -401,18 +387,3 @@ def clone_repository(self, url: str, target_dir: Optional[str] = None, return target_dir # get_file_functions() removed — use CodeParser for function extraction. - - # ---- Cache management ---- - - def clear_cache(self): - """Clear the code cache""" - import shutil - if self.cache_dir.exists(): - shutil.rmtree(self.cache_dir) - self.cache_dir.mkdir(parents=True, exist_ok=True) - - def list_cached_files(self) -> List[str]: - """List all cached code files""" - if not self.cache_dir.exists(): - return [] - return [f.name for f in self.cache_dir.iterdir() if f.is_file()] diff --git a/src/differential.py b/src/differential.py index f6267ad..6a21838 100644 --- a/src/differential.py +++ b/src/differential.py @@ -13,66 +13,17 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from typing import Any, Callable, Dict, List, Optional -from .verifier import confirmed_issues - -# --------------------------------------------------------------------------- -# Result summarisation (kept standalone so the engine has no dependency on the -# report generator). -# --------------------------------------------------------------------------- +from .summary import summarize_results as _summarize def summarize_results(results: List[Dict[str, Any]], confirmed_only: bool = False) -> Dict[str, Any]: - """Aggregate per-file analysis dicts into client-level summary stats. - - When *confirmed_only* is set and the results carry verification verdicts, - only CONFIRMED findings are counted, so cross-client stats reflect what - survived adversarial verification rather than raw candidates. - """ - total_issues = 0 - high = med = low = 0 - confidences: List[int] = [] - statuses: List[str] = [] - type_counts: Dict[str, int] = {} - - for result in results: - issues = confirmed_issues(result) if confirmed_only else (result.get("issues", []) or []) - total_issues += len(issues) - for issue in issues: - severity = str(issue.get("severity", "")).upper() - if severity == "HIGH": - high += 1 - elif severity == "MEDIUM": - med += 1 - elif severity == "LOW": - low += 1 - itype = str(issue.get("type", "")).upper() - if itype: - type_counts[itype] = type_counts.get(itype, 0) + 1 - confidences.append(int(result.get("confidence", 0) or 0)) - statuses.append(str(result.get("status", "UNKNOWN"))) - - if "MISSING" in statuses or high > 0: - overall = "ISSUES FOUND" - elif "PARTIAL_MATCH" in statuses or med > 0: - overall = "PARTIAL" - elif statuses and all(s == "FULL_MATCH" for s in statuses): - overall = "COMPLIANT" - else: - overall = "UNCERTAIN" - - return { - "overall_status": overall, - "average_confidence": round(sum(confidences) / len(confidences)) if confidences else 0, - "files_analyzed": len(results), - "total_issues": total_issues, - "high_severity": high, - "medium_severity": med, - "low_severity": low, - "issue_types": type_counts, - } + """Aggregate per-file analysis dicts into client-level summary stats.""" + return _summarize( + results, confirmed_only=confirmed_only, count_issue_types=True + ) # --------------------------------------------------------------------------- @@ -414,9 +365,14 @@ def analyze_clients(eip: int, clients: List[str], config: Any, provider: Optional[str] = None, use_llm_synthesis: bool = False, verify: bool = False, - verify_rounds: int = 2) -> DifferentialResult: + verify_rounds: int = 2, + on_client_start: Optional[Callable[[str], None]] = None + ) -> DifferentialResult: """Run analysis for each client and build a differential. + *on_client_start* is invoked with each client name before it is analyzed, + so callers can report progress. + Thin wrapper over the standard analysis pipeline for programmatic use:: from src.config import Config @@ -435,6 +391,8 @@ def analyze_clients(eip: int, clients: List[str], config: Any, last_analyzer = None for client in clients: + if on_client_start: + on_client_start(client) results, analyzer = _run_analysis( eip, client, config, llm_provider, verify=verify, verify_rounds=verify_rounds, diff --git a/src/github_fetcher.py b/src/github_fetcher.py new file mode 100644 index 0000000..80ee159 --- /dev/null +++ b/src/github_fetcher.py @@ -0,0 +1,80 @@ +"""Shared GitHub raw-file fetching with an on-disk cache. + +Both :class:`~src.spec_fetcher.SpecFetcher` and +:class:`~src.code_fetcher.CodeFetcher` pull plain files from +``raw.githubusercontent.com`` and cache them locally, so the session setup, +cache lookup/write, and cache management live here once. +""" + +import shutil +from pathlib import Path +from typing import List, Optional + +import requests + +RAW_BASE_URL = "https://raw.githubusercontent.com" + + +def raw_url(owner: str, repo: str, branch: str, path: str) -> str: + """Build a raw.githubusercontent.com URL for a file in a repo.""" + return f"{RAW_BASE_URL}/{owner}/{repo}/{branch}/{path}" + + +class CachedGitHubFetcher: + """Authenticated ``requests`` session plus a file-backed download cache.""" + + #: Directory name used under the CWD when no *cache_dir* is supplied. + DEFAULT_CACHE_DIRNAME = ".github_cache" + + def __init__(self, github_token: Optional[str] = None, + cache_dir: Optional[str] = None): + """Set up HTTP session and local cache directory.""" + self.github_token = github_token + self.cache_dir = ( + Path(cache_dir) if cache_dir + else Path.cwd() / self.DEFAULT_CACHE_DIRNAME + ) + self.session = requests.Session() + + if github_token: + self.session.headers["Authorization"] = f"token {github_token}" + + self.cache_dir.mkdir(parents=True, exist_ok=True) + + def fetch_cached(self, url: str, cache_key: str, + use_cache: bool = True) -> str: + """Return the text at *url*, reading from / writing to the cache.""" + cache_file = self.cache_dir / cache_key + + if use_cache and cache_file.exists(): + return cache_file.read_text(encoding="utf-8") + + response = self.session.get(url) + response.raise_for_status() + + content = response.text + cache_file.write_text(content, encoding="utf-8") + + return content + + def fetch_raw_file(self, owner: str, repo: str, path: str, + branch: str, cache_key: str, + use_cache: bool = True) -> str: + """Fetch a file from a GitHub repo via its raw URL.""" + return self.fetch_cached( + raw_url(owner, repo, branch, path), cache_key, use_cache + ) + + # ---- Cache management ---- + + def clear_cache(self): + """Remove every cached file.""" + if self.cache_dir.exists(): + shutil.rmtree(self.cache_dir) + self.cache_dir.mkdir(parents=True, exist_ok=True) + + def list_cached_files(self) -> List[str]: + """List the names of all cached files.""" + if not self.cache_dir.exists(): + return [] + return [f.name for f in self.cache_dir.iterdir() if f.is_file()] diff --git a/src/parser.py b/src/parser.py index f19685f..e76c0d4 100644 --- a/src/parser.py +++ b/src/parser.py @@ -5,6 +5,42 @@ from typing import Any, Dict, List, Optional +def _brace_delta(line: str) -> int: + """Net brace nesting change contributed by *line*.""" + return line.count('{') - line.count('}') + + +def _close_block(lines: List[str], start: int, depth: int, limit: int) -> int: + """Advance from *start* until the open braces (*depth*) are closed.""" + end = start + while depth > 0 and end < limit: + end += 1 + depth += _brace_delta(lines[end]) + return end + + +def _find_block_end(lines: List[str], start: int, limit: Optional[int] = None, + lookahead: int = 0) -> int: + """Index of the line closing the brace block opened at *start*. + + Returns *start* when no block opens there. *lookahead* allows the opening + brace to sit on one of the next few lines (K&R vs. Allman brace styles), + and *limit* caps the scan (defaults to the last line). + """ + if limit is None: + limit = len(lines) - 1 + + depth = _brace_delta(lines[start]) + end = start + if depth == 0 and lookahead: + for j in range(start + 1, min(start + 1 + lookahead, len(lines))): + if '{' in lines[j]: + depth = _brace_delta(lines[j]) + end = j + break + return _close_block(lines, end, depth, limit) + + @dataclass class CodeBlock: """Represents a parsed code block (function, class, etc.)""" @@ -106,14 +142,7 @@ def _parse_go(self, content: str) -> List[CodeBlock]: receiver = match.group(2) if match.group(2) else None start_line = i + 1 - # Find matching closing brace - brace_count = line.count('{') - line.count('}') - end_line = i - - while brace_count > 0 and end_line < len(lines) - 1: - end_line += 1 - brace_count += lines[end_line].count('{') - brace_count -= lines[end_line].count('}') + end_line = _find_block_end(lines, i) content_block = '\n'.join(lines[i:end_line + 1]) @@ -240,18 +269,7 @@ def _extract_methods(self, lines: List[str], start: int, end: int, if fm and '{' not in stripped[:stripped.find(fm.group(0))]: name = fm.group(1) m_start = i + 1 - brace_count = stripped.count('{') - stripped.count('}') - m_end = i - if brace_count == 0: - for j in range(i + 1, min(i + 4, len(lines))): - if '{' in lines[j]: - brace_count = lines[j].count('{') - lines[j].count('}') - m_end = j - break - while brace_count > 0 and m_end < end: - m_end += 1 - brace_count += lines[m_end].count('{') - brace_count -= lines[m_end].count('}') + m_end = _find_block_end(lines, i, limit=end, lookahead=3) if m_end > i: methods.append(CodeBlock( name=name, type="method", @@ -280,19 +298,8 @@ def _parse_brace_language(self, content: str, language: str, if cm: name = cm.group(1) start_line = i + 1 - brace_count = stripped.count('{') - stripped.count('}') - end_line = i - # If opening brace not on this line, scan forward - if brace_count == 0: - for j in range(i + 1, min(i + 4, len(lines))): - if '{' in lines[j]: - brace_count = lines[j].count('{') - lines[j].count('}') - end_line = j - break - while brace_count > 0 and end_line < len(lines) - 1: - end_line += 1 - brace_count += lines[end_line].count('{') - brace_count -= lines[end_line].count('}') + # The opening brace may sit on one of the next few lines. + end_line = _find_block_end(lines, i, lookahead=3) class_body = '\n'.join(lines[i:end_line + 1]) blocks.append(CodeBlock( name=name, type="class", @@ -313,18 +320,7 @@ def _parse_brace_language(self, content: str, language: str, if fm and '{' not in stripped[:stripped.find(fm.group(0))]: name = fm.group(1) start_line = i + 1 - brace_count = stripped.count('{') - stripped.count('}') - end_line = i - if brace_count == 0: - for j in range(i + 1, min(i + 4, len(lines))): - if '{' in lines[j]: - brace_count = lines[j].count('{') - lines[j].count('}') - end_line = j - break - while brace_count > 0 and end_line < len(lines) - 1: - end_line += 1 - brace_count += lines[end_line].count('{') - brace_count -= lines[end_line].count('}') + end_line = _find_block_end(lines, i, lookahead=3) if end_line > i: blocks.append(CodeBlock( name=name, type="method", @@ -394,12 +390,7 @@ def _parse_rust(self, content: str) -> List[CodeBlock]: trait_for = impl_match.group(2) name = f"impl {trait_for} for {type_name}" if trait_for else f"impl {type_name}" start_line = i + 1 - brace_count = line.count('{') - line.count('}') - end_line = i - while brace_count > 0 and end_line < len(lines) - 1: - end_line += 1 - brace_count += lines[end_line].count('{') - brace_count -= lines[end_line].count('}') + end_line = _find_block_end(lines, i) blocks.append(CodeBlock( name=name, type="impl", content='\n'.join(lines[i:end_line + 1]), @@ -414,7 +405,7 @@ def _parse_rust(self, content: str) -> List[CodeBlock]: if struct_match: name = struct_match.group(1) start_line = i + 1 - brace_count = line.count('{') - line.count('}') + brace_count = _brace_delta(line) if brace_count == 0 and '{' not in line: # Tuple struct or unit struct — single line blocks.append(CodeBlock( @@ -424,11 +415,7 @@ def _parse_rust(self, content: str) -> List[CodeBlock]: )) i += 1 continue - end_line = i - while brace_count > 0 and end_line < len(lines) - 1: - end_line += 1 - brace_count += lines[end_line].count('{') - brace_count -= lines[end_line].count('}') + end_line = _close_block(lines, i, brace_count, len(lines) - 1) blocks.append(CodeBlock( name=name, type="struct", content='\n'.join(lines[i:end_line + 1]), @@ -443,20 +430,16 @@ def _parse_rust(self, content: str) -> List[CodeBlock]: if fn_match: name = fn_match.group(1) start_line = i + 1 - brace_count = line.count('{') - line.count('}') + brace_count = _brace_delta(line) end_line = i # fn signature may span multiple lines before the opening brace if brace_count == 0: while end_line < len(lines) - 1: end_line += 1 - brace_count += lines[end_line].count('{') - brace_count -= lines[end_line].count('}') + brace_count += _brace_delta(lines[end_line]) if brace_count > 0: break - while brace_count > 0 and end_line < len(lines) - 1: - end_line += 1 - brace_count += lines[end_line].count('{') - brace_count -= lines[end_line].count('}') + end_line = _close_block(lines, end_line, brace_count, len(lines) - 1) if end_line > i: blocks.append(CodeBlock( name=name, type="function", diff --git a/src/report_generator.py b/src/report_generator.py index 539589a..53256b5 100644 --- a/src/report_generator.py +++ b/src/report_generator.py @@ -19,6 +19,8 @@ from src import __version__ +from .summary import summarize_results + @dataclass class ReportMetadata: @@ -384,58 +386,7 @@ def _verdict_badge_html(self, issue: Dict[str, Any]) -> str: def _generate_summary(self, results: List[Dict[str, Any]]) -> Dict[str, Any]: """Generate summary statistics from results""" - total_issues = 0 - high_severity = 0 - medium_severity = 0 - low_severity = 0 - confidences = [] - statuses = [] - verification = {"verified": False, "confirmed": 0, - "disputed": 0, "refuted": 0} - - for result in results: - issues = result.get('issues', []) - total_issues += len(issues) - - for issue in issues: - severity = issue.get('severity', '').upper() - if severity == 'HIGH': - high_severity += 1 - elif severity == 'MEDIUM': - medium_severity += 1 - elif severity == 'LOW': - low_severity += 1 - - verdict = issue.get('verification', {}).get('verdict') - if verdict: - verification["verified"] = True - key = verdict.lower() - if key in verification: - verification[key] += 1 - - confidences.append(result.get('confidence', 0)) - statuses.append(result.get('status', 'UNKNOWN')) - - # Determine overall status - if 'MISSING' in statuses or high_severity > 0: - overall_status = "ISSUES FOUND" - elif 'PARTIAL_MATCH' in statuses or medium_severity > 0: - overall_status = "PARTIAL" - elif all(s == 'FULL_MATCH' for s in statuses): - overall_status = "COMPLIANT" - else: - overall_status = "UNCERTAIN" - - return { - "overall_status": overall_status, - "average_confidence": round(sum(confidences) / len(confidences)) if confidences else 0, - "files_analyzed": len(results), - "total_issues": total_issues, - "high_severity": high_severity, - "medium_severity": medium_severity, - "low_severity": low_severity, - "verification": verification, - } + return summarize_results(results, count_verification=True) # ------------------------------------------------------------------ # Cross-client differential reports diff --git a/src/spec_fetcher.py b/src/spec_fetcher.py index 84f6fab..67ca90a 100644 --- a/src/spec_fetcher.py +++ b/src/spec_fetcher.py @@ -1,15 +1,18 @@ """Fetches Ethereum EIP specs, execution specs, and consensus specs from GitHub.""" import difflib -from pathlib import Path from typing import Dict, List, Optional import requests +from .github_fetcher import CachedGitHubFetcher -class SpecFetcher: + +class SpecFetcher(CachedGitHubFetcher): """Fetches Ethereum specifications from GitHub and other sources""" + DEFAULT_CACHE_DIRNAME = ".spec_cache" + # Supported EIPs: title, fork, and where to find their specs. EIP_REGISTRY = { 1559: { @@ -102,18 +105,6 @@ class SpecFetcher: }, } - def __init__(self, github_token: Optional[str] = None, cache_dir: Optional[str] = None): - """Set up HTTP session and local cache directory.""" - self.github_token = github_token - self.cache_dir = Path(cache_dir) if cache_dir else Path.cwd() / ".spec_cache" - self.session = requests.Session() - - if github_token: - self.session.headers["Authorization"] = f"token {github_token}" - - # Create cache directory - self.cache_dir.mkdir(parents=True, exist_ok=True) - # ---- Supported EIP helpers ---- @classmethod @@ -133,57 +124,28 @@ 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" - - # 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.raise_for_status() - - content = response.text - - # Cache the result - cache_file.write_text(content, encoding="utf-8") - - return content + return self.fetch_raw_file( + "ethereum", "EIPs", f"EIPS/eip-{eip_number}.md", "master", + cache_key=f"eip-{eip_number}.md", use_cache=use_cache, + ) 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('/', '_')}" - - 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.raise_for_status() - - content = response.text - cache_file.write_text(content, encoding="utf-8") - - return content + return self.fetch_raw_file( + "ethereum", "execution-specs", file_path, branch, + cache_key=f"exec_spec_{file_path.replace('/', '_')}", + use_cache=use_cache, + ) 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('/', '_')}" - - 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.raise_for_status() - - content = response.text - cache_file.write_text(content, encoding="utf-8") - - return content + return self.fetch_raw_file( + "ethereum", "consensus-specs", file_path, branch, + cache_key=f"consensus_spec_{file_path.replace('/', '_')}", + use_cache=use_cache, + ) def fetch_execution_spec_diff(self, eip_number: int, branch: str = "master", use_cache: bool = True) -> Optional[str]: @@ -347,15 +309,6 @@ def get_eip1559_base_fee_spec(self) -> str: # ---- Cache management ---- - def clear_cache(self): - """Clear the specification cache""" - import shutil - if self.cache_dir.exists(): - shutil.rmtree(self.cache_dir) - self.cache_dir.mkdir(parents=True, exist_ok=True) - def list_cached_specs(self) -> List[str]: """List all cached specification files""" - if not self.cache_dir.exists(): - return [] - return [f.name for f in self.cache_dir.iterdir() if f.is_file()] + return self.list_cached_files() diff --git a/src/summary.py b/src/summary.py new file mode 100644 index 0000000..8b5ef4d --- /dev/null +++ b/src/summary.py @@ -0,0 +1,81 @@ +"""Aggregation of per-file analysis results into summary statistics. + +Shared by the report generator (single-client reports) and the differential +engine (cross-client comparison), which need the same severity/status rollup. +""" + +from typing import Any, Dict, List + +from .verifier import confirmed_issues + + +def summarize_results(results: List[Dict[str, Any]], + confirmed_only: bool = False, + count_verification: bool = False, + count_issue_types: bool = False) -> Dict[str, Any]: + """Aggregate per-file analysis dicts into summary stats. + + When *confirmed_only* is set and the results carry verification verdicts, + only CONFIRMED findings are counted, so stats reflect what survived + adversarial verification rather than raw candidates. *count_verification* + adds a ``verification`` verdict tally and *count_issue_types* an + ``issue_types`` breakdown. + """ + total_issues = 0 + high = med = low = 0 + confidences: List[int] = [] + statuses: List[str] = [] + type_counts: Dict[str, int] = {} + verification = {"verified": False, "confirmed": 0, "disputed": 0, "refuted": 0} + + for result in results: + issues = confirmed_issues(result) if confirmed_only else (result.get("issues", []) or []) + total_issues += len(issues) + for issue in issues: + severity = str(issue.get("severity", "")).upper() + if severity == "HIGH": + high += 1 + elif severity == "MEDIUM": + med += 1 + elif severity == "LOW": + low += 1 + + if count_issue_types: + itype = str(issue.get("type", "")).upper() + if itype: + type_counts[itype] = type_counts.get(itype, 0) + 1 + + if count_verification: + verdict = issue.get("verification", {}).get("verdict") + if verdict: + verification["verified"] = True + key = verdict.lower() + if key in verification: + verification[key] += 1 + + confidences.append(int(result.get("confidence", 0) or 0)) + statuses.append(str(result.get("status", "UNKNOWN"))) + + if "MISSING" in statuses or high > 0: + overall = "ISSUES FOUND" + elif "PARTIAL_MATCH" in statuses or med > 0: + overall = "PARTIAL" + elif statuses and all(s == "FULL_MATCH" for s in statuses): + overall = "COMPLIANT" + else: + overall = "UNCERTAIN" + + summary: Dict[str, Any] = { + "overall_status": overall, + "average_confidence": round(sum(confidences) / len(confidences)) if confidences else 0, + "files_analyzed": len(results), + "total_issues": total_issues, + "high_severity": high, + "medium_severity": med, + "low_severity": low, + } + if count_issue_types: + summary["issue_types"] = type_counts + if count_verification: + summary["verification"] = verification + return summary diff --git a/tests/test_shared_utils.py b/tests/test_shared_utils.py new file mode 100644 index 0000000..4a50985 --- /dev/null +++ b/tests/test_shared_utils.py @@ -0,0 +1,155 @@ +"""Tests for the shared utilities extracted from duplicated code paths.""" + +import shutil +import tempfile +import unittest +from unittest.mock import Mock, patch + +from src.analyzer import BaseAnalyzer, get_analyzer +from src.github_fetcher import CachedGitHubFetcher, raw_url +from src.parser import _brace_delta, _close_block, _find_block_end +from src.summary import summarize_results + + +class TestCachedGitHubFetcher(unittest.TestCase): + """The download cache shared by SpecFetcher and CodeFetcher.""" + + def setUp(self): + self.cache_dir = tempfile.mkdtemp(prefix="prspec_fetcher_test_") + self.fetcher = CachedGitHubFetcher(cache_dir=self.cache_dir) + + def tearDown(self): + shutil.rmtree(self.cache_dir, ignore_errors=True) + + def test_raw_url(self): + self.assertEqual( + raw_url("ethereum", "EIPs", "master", "EIPS/eip-1559.md"), + "https://raw.githubusercontent.com/ethereum/EIPs/master/EIPS/eip-1559.md", + ) + + def test_token_sets_authorization_header(self): + fetcher = CachedGitHubFetcher(github_token="abc", cache_dir=self.cache_dir) + self.assertEqual(fetcher.session.headers["Authorization"], "token abc") + + @patch("requests.Session.get") + def test_second_fetch_is_served_from_cache(self, mock_get): + mock_get.return_value = Mock(text="contents", raise_for_status=Mock()) + + first = self.fetcher.fetch_cached("https://example.com/f", "key") + second = self.fetcher.fetch_cached("https://example.com/f", "key") + + self.assertEqual(first, "contents") + self.assertEqual(second, "contents") + self.assertEqual(mock_get.call_count, 1) + self.assertIn("key", self.fetcher.list_cached_files()) + + @patch("requests.Session.get") + def test_clear_cache_empties_the_directory(self, mock_get): + mock_get.return_value = Mock(text="contents", raise_for_status=Mock()) + self.fetcher.fetch_cached("https://example.com/f", "key") + + self.fetcher.clear_cache() + + self.assertEqual(self.fetcher.list_cached_files(), []) + + +class TestBraceHelpers(unittest.TestCase): + """Brace-block scanning shared by the Go/C#/Java/Rust parsers.""" + + def test_brace_delta(self): + self.assertEqual(_brace_delta("if (x) {"), 1) + self.assertEqual(_brace_delta("}"), -1) + self.assertEqual(_brace_delta("x = y;"), 0) + + def test_find_block_end_same_line_open(self): + lines = ["func f() {", " body", "}", "after"] + self.assertEqual(_find_block_end(lines, 0), 2) + + def test_find_block_end_with_lookahead(self): + lines = ["void F()", "{", " body", "}", "after"] + self.assertEqual(_find_block_end(lines, 0, lookahead=3), 3) + # Without lookahead the brace is never seen, so no block is found. + self.assertEqual(_find_block_end(lines, 0), 0) + + def test_find_block_end_respects_limit(self): + lines = ["func f() {", " body", "}"] + self.assertEqual(_find_block_end(lines, 0, limit=1), 1) + + def test_close_block_handles_nesting(self): + lines = ["outer {", " inner {", " }", "}", "after"] + self.assertEqual(_close_block(lines, 0, _brace_delta(lines[0]), len(lines) - 1), 3) + + +class TestSummarizeResults(unittest.TestCase): + """Aggregation shared by the report generator and differential engine.""" + + RESULTS = [ + { + "status": "PARTIAL_MATCH", + "confidence": 80, + "issues": [ + {"severity": "HIGH", "type": "MISSING_CHECK", + "verification": {"verdict": "CONFIRMED"}}, + {"severity": "LOW", "type": "EDGE_CASE", + "verification": {"verdict": "REFUTED"}}, + ], + }, + {"status": "FULL_MATCH", "confidence": 90, "issues": []}, + ] + + def test_core_counts(self): + s = summarize_results(self.RESULTS) + self.assertEqual(s["overall_status"], "ISSUES FOUND") + self.assertEqual(s["average_confidence"], 85) + self.assertEqual(s["files_analyzed"], 2) + self.assertEqual(s["total_issues"], 2) + self.assertEqual(s["high_severity"], 1) + self.assertEqual(s["low_severity"], 1) + self.assertNotIn("issue_types", s) + self.assertNotIn("verification", s) + + def test_optional_breakdowns(self): + s = summarize_results(self.RESULTS, count_issue_types=True, + count_verification=True) + self.assertEqual(s["issue_types"], {"MISSING_CHECK": 1, "EDGE_CASE": 1}) + self.assertEqual(s["verification"], + {"verified": True, "confirmed": 1, + "disputed": 0, "refuted": 1}) + + def test_confirmed_only_drops_refuted_findings(self): + s = summarize_results(self.RESULTS, confirmed_only=True) + self.assertEqual(s["total_issues"], 1) + self.assertEqual(s["low_severity"], 0) + + +class _StubAnalyzer(BaseAnalyzer): + def analyze_compliance(self, spec_text, code_text, context): + raise NotImplementedError + + +class TestAnalyzerResultHelpers(unittest.TestCase): + """Result construction shared by the Gemini/OpenAI/Azure backends.""" + + def test_result_from_payload_defaults(self): + result = _StubAnalyzer()._result_from_payload({}, raw_response="{}") + self.assertEqual(result.status, "UNCERTAIN") + self.assertEqual(result.confidence, 0) + self.assertEqual(result.issues, []) + self.assertEqual(result.raw_response, "{}") + + def test_error_result(self): + result = _StubAnalyzer()._error_result("Gemini", ValueError("boom")) + self.assertEqual(result.status, "ERROR") + self.assertEqual(result.summary, "Gemini analysis failed: boom") + + def test_get_analyzer_rejects_unknown_provider(self): + with self.assertRaises(ValueError): + get_analyzer("bogus", api_key="k") + + def test_get_analyzer_requires_provider_arguments(self): + with self.assertRaises(ValueError): + get_analyzer("azure", api_key="k") + + +if __name__ == "__main__": + unittest.main()