app.py

import fnmatch
import re
import tempfile
from pathlib import Path
from urllib.parse import urlparse

import gradio as gr
import pandas as pd
from huggingface_hub import HfApi, hf_hub_download, whoami


COMPROMISED_VERSIONS = {"1.82.7", "1.82.8"}
KNOWN_SAFE_PIN = "1.82.6"

QUICK_SCAN_PATTERNS = [
    "requirements.txt",
    "requirements*.txt",
    "pyproject.toml",
    "poetry.lock",
    "uv.lock",
    "Pipfile",
    "Pipfile.lock",
    "setup.py",
    "setup.cfg",
    "constraints.txt",
    "environment.yml",
    "environment.yaml",
    "conda.yml",
    "Dockerfile",
    "Dockerfile.*",
    "packages.txt",
]

MEGA_SCAN_PATTERNS = QUICK_SCAN_PATTERNS + [
    "*.py",
    "*.sh",
    "Makefile",
    "README.md",
    "README.txt",
    "scripts/*",
    "bin/*",
]

TEXT_EXTENSIONS = {
    ".py", ".sh", ".txt", ".toml", ".lock", ".cfg", ".ini", ".yml", ".yaml", ".md"
}

MAX_FILE_SIZE_BYTES = 1_000_000

SEVERITY_BADGES = {
    "red": "🔴",
    "orange": "🟠",
    "yellow": "🟡",
    "green": "🟢",
    "blue": "🔵",
    "none": "⚪",
}

CREDENTIAL_PATTERNS = [
    (r"\bHF_TOKEN\b", "HF_TOKEN referenced"),
    (r"\bHUGGING_FACE_HUB_TOKEN\b", "HUGGING_FACE_HUB_TOKEN referenced"),
    (r"\bOPENAI_API_KEY\b", "OPENAI_API_KEY referenced"),
    (r"\bANTHROPIC_API_KEY\b", "ANTHROPIC_API_KEY referenced"),
    (r"\bGROQ_API_KEY\b", "GROQ_API_KEY referenced"),
    (r"\bTOGETHER_API_KEY\b", "TOGETHER_API_KEY referenced"),
    (r"\bFIREWORKS_API_KEY\b", "FIREWORKS_API_KEY referenced"),
    (r"\bMISTRAL_API_KEY\b", "MISTRAL_API_KEY referenced"),
    (r"os\.getenv\s*\(", "Environment variable lookup detected"),
    (r"os\.environ\s*\[", "Environment variable access detected"),
    (r"login\s*\(\s*token\s*=", "login(token=...) detected"),
    (r"HfApi\s*\(\s*token\s*=", "HfApi(token=...) detected"),
    (r"InferenceClient\s*\(", "InferenceClient(...) detected"),
    (r"export\s+HF_TOKEN\s*=", "Shell export of HF_TOKEN detected"),
    (r"export\s+OPENAI_API_KEY\s*=", "Shell export of OPENAI_API_KEY detected"),
    (r"export\s+ANTHROPIC_API_KEY\s*=", "Shell export of ANTHROPIC_API_KEY detected"),
    (r"--env\s+HF_TOKEN", "Container env flag for HF_TOKEN detected"),
    (r"secrets\s*\[\s*[\"']HF_TOKEN[\"']\s*\]", "Secret lookup for HF_TOKEN detected"),
    (r"secrets\s*\[\s*[\"']OPENAI_API_KEY[\"']\s*\]", "Secret lookup for OPENAI_API_KEY detected"),
]

HARDCODED_SECRET_PATTERNS = [
    (r"HF_TOKEN\s*=\s*[\"'][A-Za-z0-9_\-]{12,}[\"']", "Possible hardcoded HF token"),
    (r"OPENAI_API_KEY\s*=\s*[\"'][A-Za-z0-9_\-]{12,}[\"']", "Possible hardcoded OpenAI key"),
    (r"ANTHROPIC_API_KEY\s*=\s*[\"'][A-Za-z0-9_\-]{12,}[\"']", "Possible hardcoded Anthropic key"),
]


def badge(severity: str) -> str:
    return SEVERITY_BADGES.get(severity, "⚪")


def normalize_package_name(name: str) -> str:
    return name.strip().lower().replace("_", "-")


def normalize_hf_name(value: str) -> str:
    return (value or "").strip().lower()


def normalize_repo_id(repo_id: str) -> str:
    raw = (repo_id or "").strip()
    if "/" not in raw:
        return normalize_hf_name(raw)
    owner, space = raw.split("/", 1)
    return f"{normalize_hf_name(owner)}/{normalize_hf_name(space)}"


def safe_read_text(path: str) -> str:
    return Path(path).read_text(encoding="utf-8", errors="replace")


def is_exact_pin(operator: str | None, version: str | None) -> bool:
    return operator == "==" and bool(version)


def parse_generic_dependency_line(line: str) -> tuple[str | None, str | None, str | None]:
    m = re.match(
        r"^([A-Za-z0-9_.-]+(?:\[[A-Za-z0-9_.,-]+\])?)\s*([=<>!~]{1,2})?\s*([^\s;,#]+)?",
        line,
    )
    if not m:
        return None, None, None
    return m.group(1), m.group(2), m.group(3)


def dedupe_findings(findings: list[dict]) -> list[dict]:
    seen = set()
    out = []
    for f in findings:
        key = (
            f["severity"], f["category"], f["source"], f["line_no"],
            f["package"], f["operator"], f["version"], f["message"], f["raw_line"]
        )
        if key not in seen:
            seen.add(key)
            out.append(f)
    return out


def make_finding(
    category: str,
    source: str,
    line_no: int,
    raw_line: str,
    severity: str,
    message: str,
    package: str = "",
    operator: str | None = "",
    version: str | None = "",
) -> dict:
    return {
        "severity": severity,
        "category": category,
        "source": source,
        "line_no": line_no,
        "package": package or "",
        "operator": operator or "",
        "version": version or "",
        "message": message,
        "raw_line": raw_line.strip(),
    }


def get_token_value(oauth_token: gr.OAuthToken | None) -> str | None:
    return oauth_token.token if oauth_token else None


def get_identity(oauth_token: gr.OAuthToken | None) -> dict:
    token = get_token_value(oauth_token)
    if not token:
        return {"signed_in": False, "username": None, "owners": [], "raw": None}

    try:
        info = whoami(token)
    except Exception:
        return {"signed_in": False, "username": None, "owners": [], "raw": None}

    username = info.get("name")
    owners = [username] if username else []
    for org in info.get("orgs", []) or []:
        org_name = org.get("name")
        if org_name:
            owners.append(org_name)

    owners = list(dict.fromkeys([o for o in owners if o]))
    if username and username in owners:
        others = sorted([o for o in owners if o != username])
        owners = [username] + others
    else:
        owners = sorted(owners)

    return {"signed_in": True, "username": username, "owners": owners, "raw": info}


def auth_status_text(oauth_token: gr.OAuthToken | None) -> str:
    identity = get_identity(oauth_token)
    if not identity["signed_in"]:
        return "Not signed in. Public URL scans still work."
    return f"Signed in as `@{identity['username']}` · namespaces: **{len(identity['owners'])}**"


def oauth_info_text(oauth_token: gr.OAuthToken | None) -> str:
    identity = get_identity(oauth_token)
    if not identity["signed_in"]:
        return "Private Spaces are not listed until you sign in."

    owners = identity["owners"]
    preview = ", ".join(f"`{x}`" for x in owners[:6]) if owners else "_none_"
    extra = f" (+{len(owners) - 6} more)" if len(owners) > 6 else ""
    return f"Visible namespaces: {preview}{extra}."


def list_spaces_for_owner(owner: str, token: str | None) -> list[str]:
    if not owner:
        return []
    normalized_owner = normalize_hf_name(owner)
    api = HfApi(token=token)
    spaces = list(api.list_spaces(author=normalized_owner, token=token))
    return sorted(space.id for space in spaces)


def refresh_oauth_session(oauth_token: gr.OAuthToken | None):
    return auth_status_text(oauth_token), oauth_info_text(oauth_token)


def load_namespaces(oauth_token: gr.OAuthToken | None):
    identity = get_identity(oauth_token)
    if not identity["signed_in"]:
        return (
            gr.Dropdown(choices=[], value=None, interactive=False),
            auth_status_text(oauth_token),
            oauth_info_text(oauth_token),
            "No OAuth session loaded. Use a public Space URL or owner/name.",
        )

    owners = identity["owners"]
    selected_owner = owners[0] if owners else None
    status = f"Loaded {len(owners)} namespace(s)."
    if selected_owner:
        status += f" Current: `{selected_owner}`."

    return (
        gr.Dropdown(choices=owners, value=selected_owner, interactive=True),
        auth_status_text(oauth_token),
        oauth_info_text(oauth_token),
        status,
    )


def load_spaces_for_owner(owner: str, oauth_token: gr.OAuthToken | None):
    if not owner:
        return (
            gr.Dropdown(choices=[], value=None, interactive=False),
            "No namespace selected.",
        )

    token = get_token_value(oauth_token)
    normalized_owner = normalize_hf_name(owner)
    try:
        spaces = list_spaces_for_owner(normalized_owner, token)
        return (
            gr.Dropdown(
                choices=spaces,
                value=(spaces[0] if spaces else None),
                interactive=True,
            ),
            f"Loaded {len(spaces)} Space(s) for `{normalized_owner}`.",
        )
    except Exception as e:
        return (
            gr.Dropdown(choices=[], value=None, interactive=False),
            f"Error while loading Spaces for `{normalized_owner}`: `{e}`",
        )


def extract_space_repo_id(space_input: str) -> str:
    raw = (space_input or "").strip()
    if not raw:
        raise ValueError("Please enter a Space URL, an owner/name shortcut, or select a Space from the dropdown.")

    if raw.startswith("http://") or raw.startswith("https://"):
        parsed = urlparse(raw)
        path = parsed.path.strip("/")
        parts = path.split("/")

        if parts[0] == "spaces":
            if len(parts) < 3:
                raise ValueError("Invalid Space URL format.")
            return normalize_repo_id(f"{parts[1]}/{parts[2]}")

        if len(parts) >= 2:
            return normalize_repo_id(f"{parts[0]}/{parts[1]}")

        raise ValueError("Could not extract repo_id from URL.")

    if "/" in raw:
        parts = raw.split("/", 1)
        if len(parts) >= 2 and parts[0] and parts[1]:
            return normalize_repo_id(f"{parts[0]}/{parts[1]}")

    raise ValueError("Please enter either a full Space URL or a shortcut like owner/space_name.")


def resolve_repo_id(space_input: str, selected_space: str | None) -> tuple[str, str]:
    if (space_input or "").strip():
        return extract_space_repo_id(space_input), "input"
    if (selected_space or "").strip():
        return normalize_repo_id(selected_space.strip()), "dropdown"
    raise ValueError("Please enter a Space URL or owner/name, or select a Space from the dropdown.")


def is_text_candidate(filename: str) -> bool:
    name = Path(filename).name
    suffix = Path(filename).suffix.lower()
    if name == "Dockerfile" or name.startswith("Dockerfile."):
        return True
    if name == "Makefile":
        return True
    return suffix in TEXT_EXTENSIONS


def matches_any_pattern(filename: str, patterns: list[str]) -> bool:
    return any(
        fnmatch.fnmatch(filename, pattern) or fnmatch.fnmatch(Path(filename).name, pattern)
        for pattern in patterns
    )


def list_candidate_files(api: HfApi, repo_id: str, mode: str) -> list[str]:
    files = api.list_repo_files(repo_id=repo_id, repo_type="space")
    patterns = QUICK_SCAN_PATTERNS if mode == "quick" else MEGA_SCAN_PATTERNS

    candidates = []
    for f in files:
        if matches_any_pattern(f, patterns):
            candidates.append(f)
        elif mode == "mega" and is_text_candidate(f):
            candidates.append(f)

    return sorted(set(candidates))


def download_repo_file(repo_id: str, filename: str, token: str | None = None) -> str:
    return hf_hub_download(repo_id=repo_id, filename=filename, repo_type="space", token=token)


def scan_credential_signals(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()

        for pattern, message in HARDCODED_SECRET_PATTERNS:
            if re.search(pattern, line, re.IGNORECASE):
                findings.append(make_finding(
                    category="credential_signal",
                    source=source,
                    line_no=i,
                    raw_line=raw_line,
                    severity="orange",
                    message=message,
                ))

        for pattern, message in CREDENTIAL_PATTERNS:
            if re.search(pattern, line, re.IGNORECASE):
                findings.append(make_finding(
                    category="credential_signal",
                    source=source,
                    line_no=i,
                    raw_line=raw_line,
                    severity="yellow",
                    message=message,
                ))

    return findings


def summarize_credential_risk(findings: list[dict]) -> dict:
    creds = [f for f in findings if f["category"] == "credential_signal"]

    messages = []
    seen = set()
    for f in creds:
        msg = f["message"]
        if msg not in seen:
            seen.add(msg)
            messages.append(msg)

    severities = {f["severity"] for f in creds}
    return {
        "count": len(creds),
        "messages": messages[:6],
        "has_signals": len(creds) > 0,
        "has_orange": "orange" in severities,
    }


def has_credential_signals(findings: list[dict]) -> bool:
    return any(f["category"] == "credential_signal" for f in findings)


def classify_version_spec(operator: str | None, version: str | None) -> tuple[str, str]:
    if operator == "==" and version in COMPROMISED_VERSIONS:
        return "red", f"Compromised LiteLLM version detected: {version}"

    if operator == "==" and version:
        if version == KNOWN_SAFE_PIN:
            return "green", f"Pinned to known safe version: {version}"
        return "green", f"Pinned LiteLLM version: {version}"

    if operator is None and version is None:
        return "orange", "LiteLLM present but unpinned"

    return "orange", f"LiteLLM present with loose constraint: {(operator or '') + (version or '')}"


def make_dep_finding(source: str, line_no: int, raw_line: str, operator: str | None, version: str | None) -> dict:
    severity, message = classify_version_spec(operator, version)
    return make_finding(
        category="dependency",
        source=source,
        line_no=line_no,
        raw_line=raw_line,
        severity=severity,
        message=message,
        package="litellm",
        operator=operator,
        version=version,
    )


def analyze_dependency_hygiene_requirements(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue

        line = line.split("#", 1)[0].strip()
        if not line:
            continue

        if line.startswith(("-r ", "--requirement", "-c ", "--constraint")):
            continue

        if line.startswith(("-e ", "git+", "http://", "https://")):
            findings.append(make_finding(
                category="dependency_hygiene",
                source=source,
                line_no=i,
                raw_line=raw_line,
                severity="yellow",
                message="Non-standard dependency source; verify pinning manually",
            ))
            continue

        package, operator, version = parse_generic_dependency_line(line)
        if not package or package.startswith("-"):
            continue
        if is_exact_pin(operator, version):
            continue

        findings.append(make_finding(
            category="dependency_hygiene",
            source=source,
            line_no=i,
            raw_line=raw_line,
            severity="yellow",
            message="Dependency is not exactly pinned",
            package=package,
            operator=operator,
            version=version,
        ))

    return findings


def analyze_dependency_hygiene_pyproject(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()
        m = re.match(r'^([A-Za-z0-9_.-]+)\s*=\s*["\']([^"\']+)["\']', line)
        if not m:
            continue

        package = m.group(1)
        spec = m.group(2).strip()

        if normalize_package_name(package) in {"python", "name", "version", "description", "readme", "license"}:
            continue

        exact = re.match(r"^==?\s*([0-9][^,\s]*)$", spec)
        if exact:
            continue

        findings.append(make_finding(
            category="dependency_hygiene",
            source=source,
            line_no=i,
            raw_line=raw_line,
            severity="yellow",
            message="Dependency in pyproject.toml is not exactly pinned",
            package=package,
            version=spec,
        ))

    return findings


def summarize_dependency_hygiene(findings: list[dict]) -> dict:
    hygiene = [f for f in findings if f["category"] == "dependency_hygiene"]

    packages = []
    seen = set()
    for f in hygiene:
        pkg = f.get("package", "").strip()
        if pkg and pkg not in seen:
            seen.add(pkg)
            packages.append(pkg)

    return {"count": len(hygiene), "packages": packages[:8], "has_unpinned": len(hygiene) > 0}


def has_unpinned_litellm(findings: list[dict]) -> bool:
    for f in findings:
        if f.get("package") != "litellm":
            continue
        if f.get("category") not in {"dependency", "dynamic_install"}:
            continue
        if f.get("severity") == "red":
            continue
        operator = f.get("operator", "")
        version = f.get("version", "")
        if not (operator == "==" and version):
            return True
    return False


def has_any_litellm_signal(findings: list[dict]) -> bool:
    return any(f.get("package") == "litellm" for f in findings)


def parse_requirements(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        line = line.split("#", 1)[0].strip()

        m = re.match(
            r"^(litellm(?:\[[^\]]+\])?)\s*([=<>!~]{1,2})?\s*([^\s;,#]+)?",
            line,
            re.IGNORECASE,
        )
        if m and line.lower().startswith("litellm"):
            findings.append(make_dep_finding(source, i, raw_line, m.group(2), m.group(3)))

    findings.extend(analyze_dependency_hygiene_requirements(content, source))
    findings.extend(scan_credential_signals(content, source))
    return dedupe_findings(findings)


def parse_pyproject(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()
        if "litellm" in line.lower():
            m = re.search(
                r'litellm(?:\[[^\]]+\])?\s*(==|>=|<=|~=|!=|>|<)?\s*([0-9][^,"\']*)?',
                line,
                re.IGNORECASE,
            )
            operator = m.group(1) if m else None
            version = m.group(2) if m else None
            findings.append(make_dep_finding(source, i, raw_line, operator, version))

    findings.extend(analyze_dependency_hygiene_pyproject(content, source))
    findings.extend(scan_credential_signals(content, source))
    return dedupe_findings(findings)


def parse_docker_or_shell(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()
        lower = line.lower()

        if "litellm" in lower:
            install_match = re.search(
                r"(pip|uv\s+pip|poetry)\s+(install|add)\s+.*?litellm(?:\[[^\]]+\])?\s*(==|>=|<=|~=|!=|>|<)?\s*([^\s\\\"']+)?",
                line,
                re.IGNORECASE,
            )
            if install_match:
                operator = install_match.group(3)
                version = install_match.group(4)
                severity, message = classify_version_spec(operator, version)
                findings.append(make_finding(
                    category="dynamic_install",
                    source=source,
                    line_no=i,
                    raw_line=raw_line,
                    severity=severity,
                    message=f"Dynamic install detected: {message}",
                    package="litellm",
                    operator=operator,
                    version=version,
                ))
            else:
                findings.append(make_finding(
                    category="reference",
                    source=source,
                    line_no=i,
                    raw_line=raw_line,
                    severity="orange",
                    message="LiteLLM referenced in shell or Docker context",
                    package="litellm",
                ))

    findings.extend(scan_credential_signals(content, source))
    return dedupe_findings(findings)


def parse_python(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()

        if re.search(r"^\s*import\s+litellm\b", line, re.IGNORECASE) or re.search(r"^\s*from\s+litellm\b", line, re.IGNORECASE):
            findings.append(make_finding(
                category="import",
                source=source,
                line_no=i,
                raw_line=raw_line,
                severity="orange",
                message="LiteLLM imported in Python file",
                package="litellm",
            ))

        if re.search(r"(pip|uv\s+pip|poetry)\s+(install|add).*litellm", line, re.IGNORECASE):
            findings.append(make_finding(
                category="dynamic_install",
                source=source,
                line_no=i,
                raw_line=raw_line,
                severity="orange",
                message="Dynamic LiteLLM install command found in Python file",
                package="litellm",
            ))

    findings.extend(scan_credential_signals(content, source))
    return dedupe_findings(findings)


def parse_packages_txt(content: str, source: str) -> list[dict]:
    lines = [ln.strip() for ln in content.splitlines() if ln.strip() and not ln.strip().startswith("#")]
    findings = []

    if lines:
        findings.append(make_finding(
            category="complex_env",
            source=source,
            line_no=1,
            raw_line=f"{len(lines)} system package(s)",
            severity="blue",
            message="System packages file present; build environment may be more complex than Python deps alone",
        ))

    findings.extend(scan_credential_signals(content, source))
    return dedupe_findings(findings)


def parse_generic_text(content: str, source: str) -> list[dict]:
    findings = []

    for i, raw_line in enumerate(content.splitlines(), start=1):
        line = raw_line.strip()
        lower = line.lower()

        if "litellm" in lower:
            if re.search(r"(pip|uv\s+pip|poetry)\s+(install|add).*litellm", line, re.IGNORECASE):
                findings.append(make_finding(
                    category="dynamic_install",
                    source=source,
                    line_no=i,
                    raw_line=raw_line,
                    severity="orange",
                    message="Dynamic LiteLLM install command found",
                    package="litellm",
                ))
            elif re.search(r"\bimport\s+litellm\b|\bfrom\s+litellm\b", line, re.IGNORECASE):
                findings.append(make_finding(
                    category="import",
                    source=source,
                    line_no=i,
                    raw_line=raw_line,
                    severity="orange",
                    message="LiteLLM import/reference found",
                    package="litellm",
                ))
            else:
                findings.append(make_finding(
                    category="reference",
                    source=source,
                    line_no=i,
                    raw_line=raw_line,
                    severity="orange",
                    message="LiteLLM reference found in text file",
                    package="litellm",
                ))

    findings.extend(scan_credential_signals(content, source))
    return dedupe_findings(findings)


def parse_file(filename: str, content: str) -> list[dict]:
    name = Path(filename).name
    suffix = Path(filename).suffix.lower()

    if name == "packages.txt":
        return parse_packages_txt(content, filename)
    if name == "pyproject.toml":
        return parse_pyproject(content, filename)
    if name == "Dockerfile" or name.startswith("Dockerfile.") or suffix == ".sh":
        return parse_docker_or_shell(content, filename)
    if name == "requirements.txt" or name.startswith("requirements") or name in {"constraints.txt", "Pipfile", "Pipfile.lock", "poetry.lock", "uv.lock"}:
        return parse_requirements(content, filename)
    if suffix == ".py" or name == "setup.py":
        return parse_python(content, filename)
    if name in {"Makefile", "README.md", "README.txt", "setup.cfg"} or is_text_candidate(filename):
        return parse_generic_text(content, filename)

    return scan_credential_signals(content, filename) if content else []


def compute_status(findings: list[dict]) -> tuple[str, str]:
    severities = {f["severity"] for f in findings}
    categories = {f["category"] for f in findings}
    hygiene = summarize_dependency_hygiene(findings)
    litellm_signal = has_any_litellm_signal(findings)
    unpinned_litellm = has_unpinned_litellm(findings)
    cred_summary = summarize_credential_risk(findings)

    if "red" in severities:
        if cred_summary["has_signals"]:
            return "🔴 High risk", "Compromised LiteLLM version detected and this repo appears to use sensitive credentials"
        return "🔴 High risk", "Compromised LiteLLM version detected"

    if unpinned_litellm:
        if cred_summary["has_signals"]:
            return "🔴 High risk", "LiteLLM is present without an exact safe pin and this repo appears to use sensitive credentials"
        if hygiene["has_unpinned"]:
            return "🔴 High risk", "LiteLLM is present without an exact safe pin, and other dependencies are not exactly pinned"
        return "🔴 High risk", "LiteLLM is present without an exact safe pin"

    if "orange" in severities and ("import" in categories or "reference" in categories):
        if cred_summary["has_signals"]:
            return "🟠 Needs review", "LiteLLM references found and this repo appears to use sensitive credentials"
        if hygiene["has_unpinned"]:
            return "🟠 Needs review", "LiteLLM references found and dependency hygiene could be improved"
        return "🟠 Needs review", "LiteLLM references/imports found"

    if "green" in severities and litellm_signal:
        if cred_summary["has_signals"] and hygiene["has_unpinned"]:
            return "🟢 LiteLLM pinned, hygiene reminder", "LiteLLM looks pinned, but this repo uses credentials and some other dependencies are not exactly pinned"
        if hygiene["has_unpinned"]:
            return "🟢 LiteLLM pinned, hygiene reminder", "LiteLLM looks pinned but some other dependencies are not exactly pinned"
        return "🟢 Looks pinned", "LiteLLM found with exact version pin"

    if cred_summary["has_signals"] and hygiene["has_unpinned"]:
        return "🟡 Hygiene reminder", "No LiteLLM found, but this repo appears to use sensitive credentials and some dependencies are not exactly pinned"

    if hygiene["has_unpinned"]:
        return "🟡 Hygiene reminder", "No LiteLLM found, but some dependencies are not exactly pinned"

    if cred_summary["has_signals"]:
        return "🔵 Credential usage detected", "No LiteLLM found, but this repo appears to use sensitive credentials"

    if "blue" in severities:
        return "🔵 Complex environment", "No LiteLLM found, but environment looks non-trivial"

    return "✅ No LiteLLM found", "No LiteLLM signal detected"


def recommendations(findings: list[dict]) -> list[str]:
    severities = {f["severity"] for f in findings}
    categories = {f["category"] for f in findings}
    hygiene = summarize_dependency_hygiene(findings)
    cred_summary = summarize_credential_risk(findings)
    recs = []

    compromised_litellm = "red" in severities
    unpinned_litellm = has_unpinned_litellm(findings)

    if compromised_litellm:
        recs.append("Rotate secrets immediately if this Space ever built with the compromised LiteLLM version.")
        if cred_summary["has_signals"]:
            recs.append("Action now: rotate every Hugging Face token and provider API key referenced or implied by this Space.")
        else:
            recs.append("Action now: rotate the Hugging Face token for this Space if it used any secret at build or runtime.")
        recs.append("Pin LiteLLM to a safe exact version and rebuild the Space.")

    elif unpinned_litellm:
        recs.append("Treat this as high risk: LiteLLM is present without an exact safe pin.")
        if cred_summary["has_signals"]:
            recs.append("Action now: rotate the Hugging Face token for this Space and rotate every provider credential referenced in the repo.")
        else:
            recs.append("Action now: rotate the Hugging Face token for this Space as a precaution if it used secrets.")
        recs.append("Replace loose or missing LiteLLM pins with an exact version.")
        recs.append("Rebuild the Space after updating dependency files.")

    elif "dynamic_install" in categories:
        recs.append("Treat this as high risk: runtime LiteLLM installs can resolve unsafe versions.")
        if cred_summary["has_signals"]:
            recs.append("Action now: rotate the Hugging Face token for this Space and rotate every provider credential referenced in the repo.")
        else:
            recs.append("Action now: consider rotating the Hugging Face token for this Space if it used secrets.")
        recs.append("Remove runtime install commands or pin them to an exact safe version.")

    elif "import" in categories:
        recs.append("LiteLLM is imported; verify where it is installed and which version resolves at build/runtime.")
        if cred_summary["has_signals"]:
            recs.append("If this Space ran during the affected period, rotate the Hugging Face token and provider credentials used by this Space.")

    elif cred_summary["has_signals"]:
        if cred_summary["has_orange"]:
            recs.append("Sensitive credential handling was detected, including a possible hardcoded secret. Review and rotate affected credentials.")
        else:
            recs.append("Sensitive credential usage was detected. Review which token(s) this Space uses and rotate them if appropriate.")

    elif "blue" in severities:
        recs.append("Environment looks complex; review Dockerfiles, shell scripts, and system packages manually.")
    else:
        recs.append("No LiteLLM signal was found in scanned files.")

    if hygiene["has_unpinned"]:
        examples = ", ".join(hygiene["packages"][:5]) if hygiene["packages"] else "some dependencies"
        recs.append(f"Dependency hygiene reminder: {hygiene['count']} dependency line(s) are not exactly pinned. Example(s): {examples}.")
        recs.append("Consider pinning dependencies exactly for reproducibility and supply-chain safety.")

    recs.append("Good practice: rotate Hugging Face tokens and provider credentials regularly even when no explicit token signal is detected in the repo.")
    recs.append("This scanner is best-effort only: runtime secrets configured outside the repo are not directly visible here.")
    return recs


def render_findings_markdown(findings: list[dict], limit: int | None = None) -> str:
    if not findings:
        return "### Findings\n\n- No findings"

    items = findings if limit is None else findings[:limit]
    blocks = ["### Findings"]

    for item in items:
        pkg = f" [{item['package']}]" if item.get("package") else ""
        blocks.append(
            f"**{badge(item['severity'])} {item['category']}**{pkg}\n"
            f"- File: `{item['source']}`\n"
            f"- Line: `{item['line_no']}`\n"
            f"- Message: {item['message']}\n"
            f"- Raw: `{item['raw_line']}`"
        )

    if limit is not None and len(findings) > limit:
        blocks.append(f"_... and {len(findings) - limit} more finding(s)._")

    return "\n\n".join(blocks)


def render_recommendations_markdown(findings: list[dict]) -> str:
    recs = recommendations(findings)
    lines = ["### Recommended actions", ""]
    for rec in recs:
        lines.append(f"- {rec}")
    return "\n".join(lines)


def render_hygiene_markdown(findings: list[dict]) -> str:
    hygiene = summarize_dependency_hygiene(findings)
    lines = ["### Dependency hygiene", "", f"- Unpinned dependency findings: **{hygiene['count']}**"]
    if hygiene["packages"]:
        lines.append(f"- Example package(s): **{', '.join(hygiene['packages'])}**")
    return "\n".join(lines)


def render_credential_risk_markdown(findings: list[dict]) -> str:
    cred = summarize_credential_risk(findings)
    lines = ["### Credential exposure risk", "", f"- Credential signal findings: **{cred['count']}**"]
    if cred["messages"]:
        lines.append(f"- Example signal(s): **{'; '.join(cred['messages'])}**")
    if not cred["has_signals"]:
        lines.append("- No credential usage signal detected in scanned files")
    else:
        lines.append("- Review the Space token and every provider credential used by this Space")
        lines.append("- Runtime secrets configured outside the repo may still exist")
    return "\n".join(lines)


def render_report(
    repo_id: str,
    mode: str,
    files_scanned: list[str],
    findings: list[dict],
    show_previews: bool,
    previews: dict[str, str],
) -> str:
    status, summary = compute_status(findings)

    sections = [
        "# Space Credential & Dependency Risk Scanner",
        "",
        "### Scan summary",
        "",
        f"- Space: `{repo_id}`",
        f"- Mode: `{mode}`",
        f"- Status: {status}",
        f"- Summary: {summary}",
        f"- Files scanned: **{len(files_scanned)}**",
        "",
        render_hygiene_markdown(findings),
        "",
        render_credential_risk_markdown(findings),
        "",
        render_findings_markdown(findings),
        "",
        render_recommendations_markdown(findings),
    ]

    if show_previews and previews:
        sections.extend(["", "### File previews", ""])
        for name, content in list(previews.items())[:20]:
            sections.extend([f"#### `{name}`", "```text", content[:1000].strip(), "```", ""])

    return "\n".join(sections)


def scan_one_repo(
    repo_id: str,
    token: str | None,
    mode: str,
    show_previews: bool,
    progress: gr.Progress | None = None,
):
    normalized_repo_id = normalize_repo_id(repo_id)
    api = HfApi(token=token)
    candidate_files = list_candidate_files(api, normalized_repo_id, mode=mode)

    findings = []
    previews = {}
    files_scanned = []

    total_files = max(len(candidate_files), 1)

    for idx, filename in enumerate(candidate_files, start=1):
        if progress:
            progress((idx - 1, total_files), desc=f"Scanning {normalized_repo_id}: {filename}")

        try:
            local_path = download_repo_file(normalized_repo_id, filename, token=token)
            path_obj = Path(local_path)

            if path_obj.exists() and path_obj.stat().st_size > MAX_FILE_SIZE_BYTES:
                continue

            content = safe_read_text(local_path)
            findings.extend(parse_file(filename, content))
            files_scanned.append(filename)

            if show_previews:
                previews[filename] = content

        except Exception as e:
            files_scanned.append(filename)
            if show_previews:
                previews[filename] = f"[Could not read file: {e}]"

    findings = dedupe_findings(findings)

    if progress:
        progress((total_files, total_files), desc=f"Finished {normalized_repo_id}")

    report = render_report(normalized_repo_id, mode, files_scanned, findings, show_previews, previews)
    return report, findings, files_scanned


def make_batch_summary(df: pd.DataFrame, owner: str, total_spaces: int, filtered_rows: int) -> str:
    if df.empty:
        return f"### Batch summary\n\n- Owner: `{owner}`\n- Spaces scanned: **0**"

    risky = int(df["high_risk"].sum()) if "high_risk" in df.columns else 0
    litellm = int(df["litellm_found"].sum()) if "litellm_found" in df.columns else 0
    dynamic = int(df["dynamic_install"].sum()) if "dynamic_install" in df.columns else 0
    unpinned = int(df["has_unpinned_deps"].sum()) if "has_unpinned_deps" in df.columns else 0
    creds = int(df["credential_signals"].sum()) if "credential_signals" in df.columns else 0

    return (
        "### Batch summary\n\n"
        f"- Owner: `{owner}`\n"
        f"- Spaces scanned so far: **{total_spaces}**\n"
        f"- Spaces matching current preview filter: **{filtered_rows}**\n"
        f"- High risk: **{risky}**\n"
        f"- LiteLLM signal found: **{litellm}**\n"
        f"- Dynamic install signal: **{dynamic}**\n"
        f"- Spaces with credential signals: **{creds}**\n"
        f"- Spaces with unpinned deps reminder: **{unpinned}**"
    )


def export_batch_csv(df: pd.DataFrame, owner: str) -> str | None:
    if df.empty:
        return None

    safe_owner = owner.replace("/", "_")
    out_dir = Path(tempfile.gettempdir())
    out_path = out_dir / f"space_risk_scan_{safe_owner}.csv"
    df.to_csv(out_path, index=False)
    return str(out_path)


def make_batch_preview_markdown(df: pd.DataFrame, limit: int = 10) -> str:
    if df.empty:
        return "### Preview\n\n- No rows to preview"

    blocks = ["### Preview"]
    preview = df.head(limit)

    for _, row in preview.iterrows():
        blocks.append(
            f"**{row['status']}**\n"
            f"- Space: `{row['space']}`\n"
            f"- Summary: {row['summary']}\n"
            f"- LiteLLM: `{row['litellm_found']}`\n"
            f"- Dynamic install: `{row['dynamic_install']}`\n"
            f"- Credential signals: `{row['credential_signals']}`\n"
            f"- Unpinned deps: `{row['unpinned_deps_count']}`"
        )

    if len(df) > limit:
        blocks.append(f"_... and {len(df) - limit} more matching Space(s). Use the CSV for the full result._")

    return "\n\n".join(blocks)


def single_scan(
    space_input: str,
    selected_space: str | None,
    mode: str,
    show_previews: bool,
    oauth_token: gr.OAuthToken | None,
    progress=gr.Progress(track_tqdm=False),
):
    token = get_token_value(oauth_token)
    repo_id, source = resolve_repo_id(space_input, selected_space)

    source_label = "input field" if source == "input" else "dropdown"
    progress(0, desc=f"Target: {repo_id} ({source_label})")

    report, _, _ = scan_one_repo(
        repo_id=repo_id,
        token=token,
        mode=mode,
        show_previews=show_previews,
        progress=progress,
    )

    return f"Target: `{repo_id}` · source: **{source_label}** · finished.", report


def choose_batch_owner(batch_owner_value: str, selected_owner_value: str | None) -> str:
    manual = normalize_hf_name(batch_owner_value)
    selected = normalize_hf_name(selected_owner_value)
    return manual or selected


def batch_scan_owner(
    owner: str,
    mode: str,
    max_spaces: int,
    show_only_flagged: bool,
    oauth_token: gr.OAuthToken | None,
    progress=gr.Progress(track_tqdm=False),
):
    token = get_token_value(oauth_token)
    normalized_owner = normalize_hf_name(owner)

    if not normalized_owner:
        yield "No owner selected.", "### Batch summary\n\n- Nothing to scan", "### Preview\n\n- No preview", None
        return

    try:
        spaces = list_spaces_for_owner(normalized_owner, token)
    except Exception as e:
        yield f"Could not list Spaces for `{normalized_owner}`: `{e}`", "### Batch summary\n\n- Listing failed", "### Preview\n\n- No preview", None
        return

    if max_spaces and max_spaces > 0:
        spaces = spaces[:max_spaces]

    if not spaces:
        yield f"No Spaces found for `{normalized_owner}`.", "### Batch summary\n\n- Nothing to scan", "### Preview\n\n- No preview", None
        return

    rows = []
    total = len(spaces)

    for idx, repo_id in enumerate(spaces, start=1):
        normalized_repo_id = normalize_repo_id(repo_id)
        progress((idx - 1, total), desc=f"[{idx}/{total}] {normalized_repo_id}")
        status_line = f"Scanning `{normalized_repo_id}` · **{idx}/{total}**"

        try:
            _, findings, files_scanned = scan_one_repo(
                repo_id=normalized_repo_id,
                token=token,
                mode=mode,
                show_previews=False,
                progress=None,
            )
            status, summary = compute_status(findings)
            hygiene = summarize_dependency_hygiene(findings)

            row = {
                "space": normalized_repo_id,
                "status": status,
                "summary": summary,
                "litellm_found": any(f["package"] == "litellm" for f in findings),
                "high_risk": any(f["severity"] == "red" for f in findings) or has_unpinned_litellm(findings),
                "dynamic_install": any(f["category"] == "dynamic_install" for f in findings),
                "imports": any(f["category"] == "import" for f in findings),
                "credential_signals": has_credential_signals(findings),
                "unpinned_deps_count": hygiene["count"],
                "has_unpinned_deps": hygiene["has_unpinned"],
                "files_scanned": len(files_scanned),
            }
            rows.append(row)
        except Exception as e:
            rows.append({
                "space": normalized_repo_id,
                "status": "⚪ Error",
                "summary": str(e),
                "litellm_found": False,
                "high_risk": False,
                "dynamic_install": False,
                "imports": False,
                "credential_signals": False,
                "unpinned_deps_count": 0,
                "has_unpinned_deps": False,
                "files_scanned": 0,
            })

        df = pd.DataFrame(rows)
        if not df.empty:
            df = df.sort_values(
                by=["high_risk", "litellm_found", "credential_signals", "has_unpinned_deps", "unpinned_deps_count", "space"],
                ascending=[False, False, False, False, False, True],
            )

        display_df = df
        if show_only_flagged and not df.empty:
            display_df = df[
                (df["high_risk"]) |
                (df["litellm_found"]) |
                (df["dynamic_install"]) |
                (df["credential_signals"]) |
                (df["has_unpinned_deps"])
            ].copy()

        yield (
            status_line,
            make_batch_summary(df, normalized_owner, total_spaces=len(df), filtered_rows=len(display_df)),
            make_batch_preview_markdown(display_df, limit=10),
            None,
        )

    progress((total, total), desc=f"Finished batch scan for {normalized_owner}")

    final_df = pd.DataFrame(rows).sort_values(
        by=["high_risk", "litellm_found", "credential_signals", "has_unpinned_deps", "unpinned_deps_count", "space"],
        ascending=[False, False, False, False, False, True],
    )
    display_df = final_df
    if show_only_flagged and not final_df.empty:
        display_df = final_df[
            (final_df["high_risk"]) |
            (final_df["litellm_found"]) |
            (final_df["dynamic_install"]) |
            (final_df["credential_signals"]) |
            (final_df["has_unpinned_deps"])
        ].copy()

    csv_path = export_batch_csv(final_df, normalized_owner)
    yield (
        f"Finished batch scan for `{normalized_owner}`.",
        make_batch_summary(final_df, normalized_owner, total_spaces=len(final_df), filtered_rows=len(display_df)),
        make_batch_preview_markdown(display_df, limit=10),
        csv_path,
    )


def batch_scan_entry(
    owner_text: str,
    selected_owner: str | None,
    mode: str,
    max_n: int,
    only_flagged: bool,
    oauth_token: gr.OAuthToken | None,
    progress=gr.Progress(track_tqdm=False),
):
    owner = choose_batch_owner(owner_text, selected_owner)
    yield from batch_scan_owner(
        owner=owner,
        mode=mode,
        max_spaces=int(max_n),
        show_only_flagged=only_flagged,
        oauth_token=oauth_token,
        progress=progress,
    )


with gr.Blocks(title="Space Credential & Dependency Risk Scanner") as demo:
    gr.Markdown(
        """
# Space Credential & Dependency Risk Scanner

Scan public or private Hugging Face Spaces for LiteLLM risk, credential exposure signals, and dependency hygiene issues.
"""
    )

    gr.Markdown("## OAuth / Namespace access")

    with gr.Row():
        login_btn = gr.LoginButton("Sign in with Hugging Face")
        refresh_oauth_btn = gr.Button("Refresh OAuth session")
        load_namespaces_btn = gr.Button("Load namespaces")

    auth_status = gr.Markdown()
    oauth_info = gr.Markdown()
    global_status = gr.Markdown("")

    with gr.Row():
        owners_dd = gr.Dropdown(label="Namespace / owner", choices=[], value=None, interactive=False)

    with gr.Tab("Single Space"):
        gr.Markdown(
            """
Choose a target in one of two ways:

- select one of your loaded Spaces from the dropdown
- or type a public Space URL / `owner/space_name`

Manual input is normalized to lowercase.
If the text field is filled, it overrides the dropdown.
"""
        )

        with gr.Row():
            spaces_dd = gr.Dropdown(
                label="Loaded Space from selected namespace",
                choices=[],
                value=None,
                interactive=False,
            )
            load_spaces_btn = gr.Button("Reload spaces")

        space_input = gr.Textbox(
            label="Test a Space via URL or owner/name",
            placeholder="https://huggingface.co/spaces/owner/name  or  owner/space_name",
        )

        space_input_help = gr.Markdown("Priority: input field first, dropdown second. Manual usernames are normalized to lowercase.")

        scan_mode = gr.Radio(
            choices=[("Quick scan", "quick"), ("Mega deep scan", "mega")],
            value="quick",
            label="Scan mode",
        )

        show_previews = gr.Checkbox(label="Show file previews in report", value=False)

        with gr.Row():
            scan_btn = gr.Button("Scan selected Space", variant="primary")
            stop_single_btn = gr.Button("Stop single scan")

        single_status = gr.Markdown("")
        single_report = gr.Markdown()

    with gr.Tab("Batch by owner"):
        gr.Markdown(
            """
Batch mode scans a namespace owner.

Typed owner names are normalized to lowercase.
Use **Show only flagged rows** to keep the preview compact.
Download the CSV for the full result.
"""
        )

        with gr.Row():
            batch_owner = gr.Textbox(
                label="Owner to batch scan",
                placeholder="Leave empty to use the selected namespace above",
            )

        batch_mode = gr.Radio(
            choices=[("Quick scan", "quick"), ("Mega deep scan", "mega")],
            value="quick",
            label="Batch scan mode",
        )

        with gr.Row():
            max_spaces = gr.Slider(minimum=1, maximum=500, value=50, step=1, label="Max spaces to scan")
            show_only_flagged = gr.Checkbox(label="Show only flagged rows", value=True)

        with gr.Row():
            batch_btn = gr.Button("Batch scan owner", variant="primary")
            stop_batch_btn = gr.Button("Stop batch scan")

        batch_status = gr.Markdown("")
        batch_summary = gr.Markdown()
        batch_preview = gr.Markdown()
        batch_csv = gr.File(label="Download full CSV")

    demo.load(fn=refresh_oauth_session, outputs=[auth_status, oauth_info])

    refresh_oauth_btn.click(
        fn=refresh_oauth_session,
        outputs=[auth_status, oauth_info],
        concurrency_limit=1,
    )

    load_namespaces_event = load_namespaces_btn.click(
        fn=load_namespaces,
        outputs=[owners_dd, auth_status, oauth_info, global_status],
        concurrency_limit=1,
    )

    load_namespaces_event.then(
        fn=load_spaces_for_owner,
        inputs=[owners_dd],
        outputs=[spaces_dd, global_status],
        concurrency_limit=1,
    )

    owners_dd.change(
        fn=load_spaces_for_owner,
        inputs=[owners_dd],
        outputs=[spaces_dd, global_status],
        concurrency_limit=1,
    )

    load_spaces_btn.click(
        fn=load_spaces_for_owner,
        inputs=[owners_dd],
        outputs=[spaces_dd, global_status],
        concurrency_limit=1,
    )

    single_event = scan_btn.click(
        fn=single_scan,
        inputs=[space_input, spaces_dd, scan_mode, show_previews],
        outputs=[single_status, single_report],
        concurrency_limit=1,
        trigger_mode="once",
        show_progress="minimal",
    )

    stop_single_btn.click(fn=None, cancels=[single_event])

    batch_event = batch_btn.click(
        fn=batch_scan_entry,
        inputs=[batch_owner, owners_dd, batch_mode, max_spaces, show_only_flagged],
        outputs=[batch_status, batch_summary, batch_preview, batch_csv],
        concurrency_limit=1,
        trigger_mode="once",
        show_progress="minimal",
    )

    stop_batch_btn.click(fn=None, cancels=[batch_event])

demo.queue(default_concurrency_limit=1, max_size=16)
demo.launch(ssr_mode=False)