server/observation.py

"""Ranked action suggestions and observation helpers for SecretsAuditEnv."""
from __future__ import annotations

from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    pass

# Files that are high-risk for containing secrets
_HIGH_RISK_KEYWORDS = [
    ".env", "config", "secret", "deploy", "billing", "auth",
    "setup", "credential", "token", "key", "password", "passwd",
    "private", "cert", "vault",
]


def estimate_file_suspicion(path: str) -> float:
    """Heuristic leak-probability score for a file path (0.0–0.95)."""
    score = 0.3
    lower = path.lower()
    for keyword in _HIGH_RISK_KEYWORDS:
        if keyword in lower:
            score += 0.12
    return round(min(score, 0.95), 2)


def compute_ranked_actions(
    inspected_files: set[str],
    deep_inspected: set[str],
    fixed_secrets: set[str],
    visible_secrets: list[dict[str, Any]],
    known_files: list[str],
    difficulty: str,
    hidden_count: int,
) -> list[dict[str, Any]]:
    """Return top-5 ranked action suggestions based on current environment state."""
    candidates: list[dict[str, Any]] = []

    # Heuristic 1: visible, unfixed secrets — highest priority
    for secret in visible_secrets:
        if secret["id"] not in fixed_secrets:
            candidates.append({
                "action": "redact_secret",
                "args": {"secret_id": secret["id"], "path": secret["path"], "line": secret["line"]},
                "reason": (
                    f"Visible {secret['type']} at {secret['path']}:{secret['line']} "
                    f"not yet remediated"
                ),
                "priority": 0.92,
            })

    # Heuristic 2: uninspected high-risk files
    for path in known_files:
        if path not in inspected_files:
            score = estimate_file_suspicion(path)
            candidates.append({
                "action": "inspect_file",
                "args": {"path": path},
                "reason": f"Uninspected file with {score:.0%} leak probability",
                "priority": score,
            })

    # Heuristic 3: git history scan for medium/hard if not yet done
    if difficulty in ("medium", "hard") and not deep_inspected:
        candidates.append({
            "action": "inspect_git_history",
            "args": {"path": "."},
            "reason": (
                "Git history not yet scanned — likely contains "
                "removed-but-committed secrets"
            ),
            "priority": 0.75,
        })

    # Heuristic 4: hidden secrets hint — suggest encoded inspection
    if hidden_count > 0 and not deep_inspected:
        candidates.append({
            "action": "inspect_encoded",
            "args": {"path": ".", "line": 0},
            "reason": (
                f"{hidden_count} hidden secret(s) remain — "
                f"inspect encoded blobs or git history"
            ),
            "priority": 0.68,
        })

    # Heuristic 5: all visible secrets fixed → run gitleaks validation
    all_visible_fixed = all(s["id"] in fixed_secrets for s in visible_secrets)
    if all_visible_fixed and visible_secrets:
        candidates.append({
            "action": "run_gitleaks",
            "args": {},
            "reason": (
                "All visible secrets remediated — "
                "run gitleaks to check for residuals"
            ),
            "priority": 0.85,
        })

    # Sort descending by priority, deduplicate by action+path, return top 5
    candidates.sort(key=lambda x: x["priority"], reverse=True)
    seen: set[str] = set()
    unique: list[dict[str, Any]] = []
    for c in candidates:
        key = f"{c['action']}:{c['args'].get('path', '')}"
        if key not in seen:
            seen.add(key)
            unique.append(c)
        if len(unique) >= 5:
            break
    return unique


def get_top_blocker(
    visible_secrets: list[dict[str, Any]],
    fixed_secrets: set[str],
    hidden_count: int,
    difficulty: str,
) -> str:
    """Return a single-sentence description of the highest-priority next action."""
    for secret in visible_secrets:
        if secret["id"] not in fixed_secrets:
            return (
                f"Fix visible {secret['type']} on "
                f"{secret['path']}:{secret['line']} first."
            )
    if hidden_count > 0:
        if difficulty in ("medium", "hard"):
            return (
                f"{hidden_count} hidden secret(s) remain — "
                f"run inspect_git_history or inspect_encoded to surface them."
            )
        return f"{hidden_count} hidden secret(s) remain — run inspect_file on suspicious paths."
    return "All known secrets addressed — submit for final grading."