c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
380 lines
13 KiB
Python
380 lines
13 KiB
Python
"""
|
|
Slither + Mythril Contract Scanner
|
|
==================================
|
|
|
|
Integration with Slither (static analysis) and Mythril (symbolic execution)
|
|
for comprehensive smart contract vulnerability detection.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
from dataclasses import asdict, dataclass
|
|
from typing import Any
|
|
|
|
from app.telegram_bot.requirements import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ─── VULNERABILITY CATEGORIES ─────────────────────────────────────
|
|
|
|
VULNERABILITY_TYPES = {
|
|
"reentrancy": "Reentrancy Attack",
|
|
"overflow": "Integer Overflow/Underflow",
|
|
"unchecked_call": "Unchecked Call Return Value",
|
|
"tx_origin": "Use of tx.origin",
|
|
"delegatecall": "Dangerous delegatecall",
|
|
"selfdestruct": "Unprotected selfdestruct",
|
|
"shadowing": "Variable Shadowing",
|
|
"timestamp": "Timestamp Dependency",
|
|
"call_depth": "Call Depth Limit Exceeded",
|
|
"low_level_call": "Low-level call",
|
|
}
|
|
|
|
# Severity levels
|
|
SEVERITY = {
|
|
"High": 3,
|
|
"Medium": 2,
|
|
"Low": 1,
|
|
"Informational": 0,
|
|
}
|
|
|
|
|
|
# ─── SCAN RESULT MODELS ──────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class Finding:
|
|
"""Individual vulnerability finding."""
|
|
|
|
title: str
|
|
description: str
|
|
severity: str
|
|
contract: str
|
|
function: str
|
|
source_map: dict[str, Any]
|
|
swc_id: str = ""
|
|
reference: str = ""
|
|
|
|
|
|
@dataclass
|
|
class ScanResult:
|
|
"""Complete scan result."""
|
|
|
|
contract_address: str
|
|
chain: str
|
|
scanned: bool
|
|
timestamp: str
|
|
vulnerabilities: list[Finding]
|
|
summary: dict[str, int]
|
|
metrics: dict[str, Any]
|
|
|
|
|
|
# ─── SLITHER INTEGRATION ──────────────────────────────────────────
|
|
|
|
|
|
class SlitherScanner:
|
|
"""Slither static analysis scanner."""
|
|
|
|
def __init__(self, solc_version: str = "0.8.0"):
|
|
self.solc_version = solc_version
|
|
self._available = self._check_available()
|
|
|
|
def _check_available(self) -> bool:
|
|
"""Check if Slither is installed."""
|
|
try:
|
|
result = subprocess.run(["slither", "--version"], capture_output=True, timeout=5)
|
|
return result.returncode == 0
|
|
except (subprocess.TimeoutExpired, FileNotFoundError):
|
|
return False
|
|
|
|
def scan_file(self, file_path: str, **kwargs) -> dict[str, Any]:
|
|
"""
|
|
Scan a Solidity file with Slither.
|
|
|
|
Args:
|
|
file_path: Path to .sol file
|
|
**kwargs: Additional slither arguments
|
|
|
|
Returns:
|
|
Slither JSON output
|
|
"""
|
|
if not self._available:
|
|
return {"success": False, "error": "Slither not installed"}
|
|
|
|
cmd = ["slither", file_path, "--json", "-"]
|
|
if kwargs.get("solc"):
|
|
cmd.extend(["--solc", kwargs["solc"]])
|
|
if kwargs.get("solc_args"):
|
|
cmd.extend(["--solc-args", kwargs["solc_args"]])
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
|
|
return json.loads(result.stdout) if result.stdout else {}
|
|
except (subprocess.TimeoutExpired, json.JSONDecodeError) as e:
|
|
return {"success": False, "error": str(e)}
|
|
|
|
def scan_contract(self, contract_address: str, chain: str = "base") -> ScanResult:
|
|
"""
|
|
Scan an on-chain contract using Slither.
|
|
|
|
Args:
|
|
contract_address: Contract address to scan
|
|
chain: Blockchain network
|
|
|
|
Returns:
|
|
ScanResult with vulnerabilities
|
|
"""
|
|
if not self._available:
|
|
return self._stub_result(contract_address, chain)
|
|
|
|
# Get source code from blockchain explorer API
|
|
source_code = self._fetch_source_code(contract_address, chain)
|
|
if not source_code:
|
|
return self._stub_result(contract_address, chain, error="Could not fetch source")
|
|
|
|
# Write to temp file and scan
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".sol", delete=False) as f:
|
|
f.write(source_code)
|
|
temp_path = f.name
|
|
|
|
try:
|
|
result = self.scan_file(temp_path)
|
|
|
|
# Parse findings
|
|
findings = []
|
|
if "results" in result:
|
|
for detector in result["results"].get("detectors", []):
|
|
finding = Finding(
|
|
title=detector["check"],
|
|
description=detector["description"],
|
|
severity=detector["impact"],
|
|
contract=contract_address,
|
|
function=detector.get("function", "unknown"),
|
|
source_map=detector.get("source_mapping", {}),
|
|
swc_id=detector.get("swc_id", ""),
|
|
)
|
|
findings.append(finding)
|
|
|
|
return ScanResult(
|
|
contract_address=contract_address,
|
|
chain=chain,
|
|
scanned=True,
|
|
timestamp=json.dumps("2026-05-12T00:00:00Z"), # Use real timestamp in prod
|
|
vulnerabilities=findings,
|
|
summary=self._count_by_severity(findings),
|
|
metrics={"lines_of_code": len(source_code.splitlines())},
|
|
)
|
|
|
|
finally:
|
|
os.unlink(temp_path)
|
|
|
|
def _fetch_source_code(self, address: str, chain: str) -> str | None:
|
|
"""Fetch contract source from block explorer."""
|
|
# Base explorer
|
|
if chain == "base":
|
|
url = f"https://api.basescan.org/api?module=contract&action=getsourcecode&address={address}&apikey={os.getenv('BASESCAN_API_KEY', '')}"
|
|
elif chain == "ethereum":
|
|
url = f"https://api.etherscan.io/api?module=contract&action=getsourcecode&address={address}&apikey={os.getenv('ETHERSCAN_API_KEY', '')}"
|
|
else:
|
|
url = None
|
|
|
|
if not url:
|
|
return None
|
|
|
|
try:
|
|
resp = httpx.get(url, timeout=10)
|
|
data = resp.json()
|
|
if data.get("status") == "1":
|
|
source = data["result"][0].get("SourceCode", "")
|
|
# Handle API response format
|
|
if source.startswith("{{") and source.endswith("}}"):
|
|
source = json.loads(source[1:-1])["SourceCode"]
|
|
return source
|
|
except Exception as e:
|
|
logger.error(f"Error fetching source: {e}")
|
|
|
|
return None
|
|
|
|
def _stub_result(self, address: str, chain: str, error: str = "") -> ScanResult:
|
|
"""Return stub result when Slither unavailable."""
|
|
return ScanResult(
|
|
contract_address=address,
|
|
chain=chain,
|
|
scanned=False,
|
|
timestamp="stub",
|
|
vulnerabilities=[],
|
|
summary={"High": 0, "Medium": 0, "Low": 0, "Informational": 0},
|
|
metrics={"error": error or "Slither not available", "lines_of_code": 0},
|
|
)
|
|
|
|
def _count_by_severity(self, findings: list[Finding]) -> dict[str, int]:
|
|
"""Count findings by severity."""
|
|
counts = {"High": 0, "Medium": 0, "Low": 0, "Informational": 0}
|
|
for f in findings:
|
|
counts[f.severity] = counts.get(f.severity, 0) + 1
|
|
return counts
|
|
|
|
|
|
# ─── MYTHRIL INTEGRATION ──────────────────────────────────────────
|
|
|
|
|
|
class MythrilScanner:
|
|
"""Mythril symbolic execution scanner."""
|
|
|
|
def __init__(self):
|
|
self._available = self._check_available()
|
|
|
|
def _check_available(self) -> bool:
|
|
"""Check if Mythril is installed."""
|
|
try:
|
|
result = subprocess.run(["myth", "--version"], capture_output=True, timeout=5)
|
|
return result.returncode == 0
|
|
except (subprocess.TimeoutExpired, FileNotFoundError):
|
|
return False
|
|
|
|
def scan_bytecode(self, bytecode: str, **kwargs) -> dict[str, Any]:
|
|
"""
|
|
Scan contract bytecode with Mythril.
|
|
|
|
Args:
|
|
bytecode: Contract bytecode (0x prefixed)
|
|
**kwargs: Additional myth arguments
|
|
|
|
Returns:
|
|
Mythril JSON output
|
|
"""
|
|
if not self._available:
|
|
return {"success": False, "error": "Mythril not installed", "issues": []}
|
|
|
|
cmd = ["myth", "-x", "-a", bytecode]
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=300, # 5 minutes for symbolic execution
|
|
)
|
|
|
|
# Parse output (Mythril outputs in YAML/JSON mixed format)
|
|
# For now, return simple structure
|
|
issues = []
|
|
for line in result.stdout.split("\n"):
|
|
if "Vulnerability" in line or "Error" in line:
|
|
issues.append({"type": "Detected", "description": line.strip()})
|
|
|
|
return {"success": True, "issues": issues}
|
|
|
|
except (subprocess.TimeoutExpired, Exception) as e:
|
|
return {"success": False, "error": str(e), "issues": []}
|
|
|
|
def scan_contract(self, contract_address: str, chain: str = "base") -> dict[str, Any]:
|
|
"""Scan contract using Mythril."""
|
|
if not self._available:
|
|
return {"error": "Mythril not installed", "contract": contract_address}
|
|
|
|
# Fetch bytecode
|
|
bytecode = self._fetch_bytecode(contract_address, chain)
|
|
if not bytecode:
|
|
return {"error": "Could not fetch bytecode", "contract": contract_address}
|
|
|
|
return self.scan_bytecode(bytecode)
|
|
|
|
def _fetch_bytecode(self, address: str, chain: str) -> str | None:
|
|
"""Get contract bytecode from blockchain."""
|
|
# This would use web3 to get bytecode directly
|
|
# For now, return placeholder
|
|
return "0x608060405234801561001057600080fd5b50600080546001600160a01b03191633179055610132806100326000396000f3fe608060405234801561001057600080fd5b50600080546001600160a01b031916331790556100c8806100326000396000f3fe608060405260043610601f5760003560e01c80633b3b5fde146026575b600080fd5b60366038565b005b600054604080513381526020602083015282517faf1358d3b491c7503447674258b676119e830f6325bad0377071868593a1348099fb5b1560355760336053565b565b6001548156fea164736f6c6343000819000a"
|
|
|
|
def scan_file(self, file_path: str, **kwargs) -> dict[str, Any]:
|
|
"""Scan Solidity file with Mythril."""
|
|
if not self._available:
|
|
return {"success": False, "error": "Mythril not installed"}
|
|
|
|
cmd = ["myth", "-x", "-f", file_path]
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
|
|
return {"success": True, "output": result.stdout}
|
|
except subprocess.TimeoutExpired:
|
|
return {"success": False, "error": "Scan timed out"}
|
|
|
|
|
|
# ─── COMBINED SCANNER ─────────────────────────────────────────────
|
|
|
|
|
|
class ContractScanner:
|
|
"""Combined Slither + Mythril scanner."""
|
|
|
|
def __init__(self):
|
|
self.slither = SlitherScanner()
|
|
self.mythril = MythrilScanner()
|
|
|
|
def scan(self, contract_address: str, chain: str = "base", deep: bool = False) -> dict[str, Any]:
|
|
"""
|
|
Scan contract using available tools.
|
|
|
|
Args:
|
|
contract_address: Contract to scan
|
|
chain: Blockchain network
|
|
deep: Run both Slither and Mythril
|
|
|
|
Returns:
|
|
Combined scan result
|
|
"""
|
|
result = {
|
|
"contract": contract_address,
|
|
"chain": chain,
|
|
"timestamp": "2026-05-12T00:00:00Z", # Real timestamp in prod
|
|
"tools": [],
|
|
}
|
|
|
|
# Always run Slither
|
|
slither_result = self.slither.scan_contract(contract_address, chain)
|
|
if slither_result.scanned:
|
|
result["tools"].append("slither")
|
|
result["vulnerabilities"] = [asdict(f) for f in slither_result.vulnerabilities]
|
|
result["summary"] = slither_result.summary
|
|
else:
|
|
result["vulnerabilities"] = []
|
|
result["summary"] = {}
|
|
|
|
# Optionally run Mythril for deep analysis
|
|
if deep and self.mythril._available:
|
|
myth_result = self.mythril.scan_contract(contract_address, chain)
|
|
if myth_result.get("success"):
|
|
result["tools"].append("mythril")
|
|
if "issues" in myth_result:
|
|
for issue in myth_result["issues"]:
|
|
result["vulnerabilities"].append(
|
|
{
|
|
"title": issue.get("type", "Detected"),
|
|
"description": issue.get("description", ""),
|
|
"severity": "High",
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
# ─── SINGLETON INSTANCE ───────────────────────────────────────────
|
|
|
|
_scanner = ContractScanner()
|
|
|
|
|
|
def deep_scan_contract(contract_address: str, chain: str = "base") -> dict[str, Any]:
|
|
"""Deep scan single contract (Slither + Mythril)."""
|
|
return _scanner.scan(contract_address, chain, deep=True)
|
|
|
|
|
|
def batch_deep_scan(contracts: list[str], chain: str = "base") -> dict[str, Any]:
|
|
"""Batch scan multiple contracts."""
|
|
results = []
|
|
for addr in contracts:
|
|
result = _scanner.scan(addr, chain, deep=False)
|
|
results.append(result)
|
|
return {"contracts": results, "total": len(results)}
|