core/scoring.py

"""Numeric risk scoring for individual findings.

Builds on top of the existing ``confidence`` and ``severity`` fields that
every finding already carries (set by ``make_finding()`` in ``core.models``).

Score range: 0–10 (clamped). Higher = more urgent.
"""
from __future__ import annotations

import re
from typing import Any

from core.models import CONFIDENCE_RANK, SEVERITY_RANK

# ---------------------------------------------------------------------------
# HuggingFace employee accounts — owner gets a small escalation bonus
# ---------------------------------------------------------------------------
EMPLOYEE_ACCOUNTS: frozenset[str] = frozenset({
    "clem", "lysandre", "thomwolf", "julien-c", "osanseviero",
    "sgugger", "victorsanh", "reach-vb", "pcuenq", "nielsr",
    "patrickvonplaten", "lewtun", "narsil", "merve", "sanchit-gandhi",
})

# ---------------------------------------------------------------------------
# Score tables (mirror SEVERITY_RANK / CONFIDENCE_RANK as additive points)
# ---------------------------------------------------------------------------
_SEV_PTS: dict[str, int] = {"CRITICAL": 4, "ERROR": 3, "HIGH": 3, "WARNING": 2, "MEDIUM": 2, "INFO": 1, "LOW": 1}
_CONF_PTS: dict[str, int] = {"confirmed": 3, "likely": 2, "possible": 1}

# Regex patterns that suggest user-supplied input reaches the finding
_USER_INPUT_PATTERNS: list[re.Pattern[str]] = [
    re.compile(r"request\.", re.IGNORECASE),
    re.compile(r"gr\.State", re.IGNORECASE),
    re.compile(r"gr\.Textbox", re.IGNORECASE),
    re.compile(r"input\(", re.IGNORECASE),
    re.compile(r"upload", re.IGNORECASE),
    re.compile(r"user_input", re.IGNORECASE),
    re.compile(r"form\.", re.IGNORECASE),
    re.compile(r"query_params", re.IGNORECASE),
]

# Upload-widget triggers (gradio / streamlit)
_UPLOAD_TRIGGERS: frozenset[str] = frozenset({
    "gr.File", "gr.UploadButton", "st.file_uploader",
    "files.upload", "UploadedFile",
})


def _extract_owner(repo_url: str) -> str:
    """Return the HF/GitHub owner from a repo URL, or empty string."""
    # Matches: https://huggingface.co/owner/name  or  https://github.com/owner/name
    m = re.search(r"(?:huggingface\.co|github\.com)/([^/]+)", repo_url)
    return m.group(1).lower() if m else ""


def score_finding(
    finding: dict[str, Any],
    repo_url: str,
    all_findings: list[dict[str, Any]],
) -> int:
    """Compute a 0–10 risk score for *finding*.

    Scoring breakdown (max 10):
    - Severity:    ERROR=3, WARNING=2, INFO=1
    - Confidence:  confirmed=3, likely=2, possible=1
    - User-input path heuristic (+3 if message or file suggests user input)
    - Upload widget present (+2 if severity is ERROR and scan has an upload widget finding)
    - Employee owner (+1 if HF employee owns the space)
    - MCP present (+1 if scan contains MCP-related findings)
    """
    sev = finding.get("severity", "INFO").upper()
    conf = finding.get("confidence", "possible").lower()

    score = _SEV_PTS.get(sev, 1) + _CONF_PTS.get(conf, 1)

    # User-input heuristic: message or file path suggests user-controlled data
    message = finding.get("message", "")
    file_path = finding.get("file", "")
    combined = f"{message} {file_path}"
    if any(p.search(combined) for p in _USER_INPUT_PATTERNS):
        score += 3

    # Upload widget escalation: only meaningful when combined with ERROR severity
    if sev == "ERROR":
        has_upload = any(
            any(tok in f.get("message", "") or tok in f.get("file", "") for tok in _UPLOAD_TRIGGERS)
            for f in all_findings
        )
        if has_upload:
            score += 2

    # Employee account bonus
    owner = _extract_owner(repo_url)
    if owner and owner in EMPLOYEE_ACCOUNTS:
        score += 1

    # MCP presence: escalate if the scan contains MCP agent findings
    has_mcp = any(
        "mcp" in f.get("rule", "").lower() or "mcp" in f.get("message", "").lower()
        for f in all_findings
    )
    if has_mcp:
        score += 1

    return min(score, 10)