graders/static_analysis.py

"""
graders/static_analysis.py — Static security grader.
Weight: 15% of total reward.

Tools:
  bandit:  AST-based Python security scanner, zero-config, maps to CWE IDs
  semgrep: Rule-based pattern matching — catches what bandit misses

Penalty schedule:
  HIGH severity issue:   -0.30
  MEDIUM severity issue: -0.15
  LOW severity issue:    -0.05

Score = max(0.0, 1.0 - total_penalty)
No penalty stacking beyond score floor of 0.0.
"""
import subprocess
import json
import tempfile
import os
import re
from typing import Dict, Any


# ── bandit ────────────────────────────────────────────────────────────────────

def run_bandit(code: str) -> Dict[str, Any]:
    """Run bandit static analysis. Returns score + issues list."""
    with tempfile.NamedTemporaryFile(
        mode="w", suffix=".py", delete=False, encoding="utf-8"
    ) as f:
        f.write(code)
        tmp = f.name

    try:
        result = subprocess.run(
            ["bandit", "-r", tmp, "-f", "json", "-q", "--exit-zero"],
            capture_output=True, text=True, timeout=15,
        )
        try:
            data = json.loads(result.stdout or '{"results": []}')
        except json.JSONDecodeError:
            data = {"results": []}

        issues = data.get("results", [])
        penalty = 0.0
        for issue in issues:
            sev = issue.get("issue_severity", "LOW")
            if sev == "HIGH":
                penalty += 0.30
            elif sev == "MEDIUM":
                penalty += 0.15
            else:
                penalty += 0.05

        score = max(0.0, 1.0 - penalty)
        return {
            "score": round(score, 4),
            "issues": issues[:5],   # Return top 5 for feedback
            "issue_count": len(issues),
        }
    except FileNotFoundError:
        # bandit not installed — skip gracefully
        return {"score": 0.9, "issues": [], "issue_count": 0, "note": "bandit not available"}
    except subprocess.TimeoutExpired:
        return {"score": 0.7, "issues": [], "issue_count": 0, "note": "bandit timeout"}
    finally:
        try:
            os.unlink(tmp)
        except OSError:
            pass


# ── AST heuristics (zero-dependency fallback + extras bandit misses) ──────────

_DANGEROUS_PATTERNS = [
    (r'\beval\s*\(', "HIGH", "eval() usage — arbitrary code execution risk"),
    (r'\bexec\s*\(', "HIGH", "exec() usage — arbitrary code execution risk"),
    (r'hashlib\.md5\b', "HIGH", "MD5 usage — broken cryptographic algorithm (CWE-327)"),
    (r'hashlib\.sha1\b', "MEDIUM", "SHA1 usage — deprecated for security (CWE-327)"),
    (r'random\.random\b', "MEDIUM", "random.random() — not cryptographically secure (use secrets)"),
    (r'subprocess.*shell\s*=\s*True', "HIGH", "shell=True — shell injection risk (CWE-78)"),
    (r'os\.system\s*\(', "HIGH", "os.system() — shell injection risk (CWE-78)"),
    (r'pickle\.loads?\s*\(', "HIGH", "pickle — arbitrary code execution on untrusted data"),
    (r'yaml\.load\s*\([^)]*\)', "MEDIUM", "yaml.load() without Loader — use yaml.safe_load()"),
    (r'password\s*=\s*["\']', "MEDIUM", "Potential hardcoded password (CWE-259)"),
    (r'secret\s*=\s*["\']', "MEDIUM", "Potential hardcoded secret"),
    (r'f["\'].*SELECT.*\{', "HIGH", "f-string SQL construction — injection risk (CWE-89)"),
    (r'%.*SELECT.*%', "HIGH", "%-format SQL construction — injection risk (CWE-89)"),
    (r'\.format\(.*\).*SELECT|SELECT.*\.format', "HIGH", "str.format() SQL — injection risk (CWE-89)"),
]


def run_ast_heuristics(code: str) -> Dict[str, Any]:
    """Fast regex-based heuristic checks as bandit supplement."""
    issues = []
    for pattern, severity, message in _DANGEROUS_PATTERNS:
        if re.search(pattern, code, re.IGNORECASE):
            issues.append({"severity": severity, "message": message})

    penalty = 0.0
    for issue in issues:
        if issue["severity"] == "HIGH":
            penalty += 0.25
        elif issue["severity"] == "MEDIUM":
            penalty += 0.10
        else:
            penalty += 0.04

    return {
        "score": max(0.0, 1.0 - penalty),
        "issues": issues,
    }


# ── Combined grader ───────────────────────────────────────────────────────────

def grade_static(code: str) -> Dict[str, Any]:
    """
    Run bandit + AST heuristics, return combined score.
    Final score = min(bandit_score, heuristic_score) — take the more pessimistic view.
    """
    bandit_result = run_bandit(code)
    heuristic_result = run_ast_heuristics(code)

    # Combine: worst of both tools wins
    combined_score = min(bandit_result["score"], heuristic_result["score"])

    all_issues = bandit_result.get("issues", []) + heuristic_result.get("issues", [])
    issue_count = len(all_issues)

    if combined_score >= 0.9:
        feedback = "No significant static vulnerabilities detected."
    elif combined_score >= 0.7:
        feedback = f"{issue_count} minor issue(s) found. Review bandit output."
    elif combined_score >= 0.5:
        feedback = f"{issue_count} moderate issue(s). Avoid eval/exec, weak crypto, shell=True."
    else:
        feedback = f"{issue_count} HIGH severity issue(s). Critical: remove eval/exec, use parameterised queries, avoid MD5/SHA1."

    return {
        "score": round(combined_score, 4),
        "feedback": feedback,
        "issue_count": issue_count,
        "bandit_score": bandit_result["score"],
        "heuristic_score": heuristic_result["score"],
        "issues": all_issues[:5],
    }