"""Static security scanner for Mixlar plugin packages.

Parses every ``.py`` in a package with the :mod:`ast` module (no code is
executed) and reports which sensitive **capabilities** it uses and any
**high-risk** patterns. Two jobs:

  1. *Risk report* — surface dangerous constructs (shell-out, ``eval``/``exec``,
     ``ctypes``, obfuscated ``exec(base64…)``, etc.) so a reviewer isn't blind
     and CI can hard-block the worst of it before signing.
  2. *Capability inference* — the same detection feeds the permissions model
     (:mod:`mixlar.permissions`): it computes the capability set a plugin's
     code actually touches, which we compare against what the manifest
     ``permissions`` declares.

This is a heuristic AST scan, not a proof of safety — it complements review and
signing, it does not replace them. Determined obfuscation can evade any static
check; the value is catching the obvious and forcing undeclared capability use
into the open.
"""
from __future__ import annotations

import ast
import os
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set, Tuple

# Risk levels, most severe first.
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
_ORDER = {HIGH: 0, MEDIUM: 1, LOW: 2}

# Capability categories (mirror mixlar.permissions.CAPABILITIES).
CAP_NETWORK = "network"
CAP_SUBPROCESS = "subprocess"
CAP_FS = "filesystem"
CAP_NATIVE = "native"
CAP_INPUT = "input"
CAP_GUI = "gui"

# Modules whose mere import implies a capability.
_MODULE_CAP = {
    "subprocess": (CAP_SUBPROCESS, HIGH),
    "ctypes": (CAP_NATIVE, HIGH),
    "win32api": (CAP_NATIVE, HIGH), "win32gui": (CAP_NATIVE, HIGH),
    "win32process": (CAP_NATIVE, HIGH), "win32com": (CAP_NATIVE, HIGH),
    "pywintypes": (CAP_NATIVE, MEDIUM), "comtypes": (CAP_NATIVE, HIGH),
    "pycaw": (CAP_NATIVE, MEDIUM), "winreg": (CAP_NATIVE, HIGH),
    "_winreg": (CAP_NATIVE, HIGH),
    "requests": (CAP_NETWORK, MEDIUM), "urllib": (CAP_NETWORK, MEDIUM),
    "urllib2": (CAP_NETWORK, MEDIUM), "http": (CAP_NETWORK, MEDIUM),
    "socket": (CAP_NETWORK, MEDIUM), "websocket": (CAP_NETWORK, MEDIUM),
    "websockets": (CAP_NETWORK, MEDIUM), "aiohttp": (CAP_NETWORK, MEDIUM),
    "ftplib": (CAP_NETWORK, MEDIUM), "smtplib": (CAP_NETWORK, MEDIUM),
    "telnetlib": (CAP_NETWORK, HIGH), "paramiko": (CAP_NETWORK, HIGH),
    # Third-party HTTP/network clients (they all do outbound I/O under the hood).
    "httpx": (CAP_NETWORK, MEDIUM), "urllib3": (CAP_NETWORK, MEDIUM),
    "curl_cffi": (CAP_NETWORK, MEDIUM), "grpc": (CAP_NETWORK, MEDIUM),
    "paho": (CAP_NETWORK, MEDIUM), "yfinance": (CAP_NETWORK, MEDIUM),
    "openrgb": (CAP_NETWORK, MEDIUM), "pymodbus": (CAP_NETWORK, MEDIUM),
    "zeroconf": (CAP_NETWORK, MEDIUM),
    "pynput": (CAP_INPUT, HIGH), "keyboard": (CAP_INPUT, HIGH),
    "mouse": (CAP_INPUT, HIGH), "pyautogui": (CAP_INPUT, HIGH),
    "marshal": (None, HIGH), "pickle": (None, MEDIUM),
    "cPickle": (None, MEDIUM),
}

# Fully-qualified call names → (capability|None, level, human message).
_CALL_SIGNALS = {
    "os.system": (CAP_SUBPROCESS, HIGH, "runs a shell command (os.system)"),
    "os.popen": (CAP_SUBPROCESS, HIGH, "runs a shell command (os.popen)"),
    "os.startfile": (CAP_SUBPROCESS, MEDIUM, "launches a file/program (os.startfile)"),
    "os.execv": (CAP_SUBPROCESS, HIGH, "replaces the process image (os.exec*)"),
    "os.spawnv": (CAP_SUBPROCESS, HIGH, "spawns a process (os.spawn*)"),
    "os.remove": (CAP_FS, MEDIUM, "deletes a file (os.remove)"),
    "os.unlink": (CAP_FS, MEDIUM, "deletes a file (os.unlink)"),
    "os.rmdir": (CAP_FS, MEDIUM, "removes a directory (os.rmdir)"),
    "shutil.rmtree": (CAP_FS, HIGH, "recursively deletes a directory (shutil.rmtree)"),
    "subprocess.run": (CAP_SUBPROCESS, HIGH, "spawns a process (subprocess.run)"),
    "subprocess.call": (CAP_SUBPROCESS, HIGH, "spawns a process (subprocess.call)"),
    "subprocess.Popen": (CAP_SUBPROCESS, HIGH, "spawns a process (subprocess.Popen)"),
    "subprocess.check_output": (CAP_SUBPROCESS, HIGH, "spawns a process (subprocess.check_output)"),
    "importlib.import_module": (None, MEDIUM, "dynamic import (importlib.import_module)"),
}

# Bare builtins that are always worth flagging.
_DANGER_BUILTINS = {
    "eval": (HIGH, "dynamic code evaluation (eval)"),
    "exec": (HIGH, "dynamic code execution (exec)"),
    "compile": (MEDIUM, "compiles code at runtime (compile)"),
    "__import__": (MEDIUM, "dynamic import (__import__)"),
}


@dataclass
class Finding:
    level: str
    category: str          # a capability, "dynamic-code", or "obfuscation"
    message: str
    file: str
    line: int


@dataclass
class ScanReport:
    findings: List[Finding] = field(default_factory=list)
    capabilities: Set[str] = field(default_factory=set)
    files_scanned: int = 0
    parse_errors: List[Tuple[str, str]] = field(default_factory=list)

    @property
    def high(self) -> List[Finding]:
        return [f for f in self.findings if f.level == HIGH]

    @property
    def risk(self) -> str:
        if any(f.level == HIGH for f in self.findings):
            return HIGH
        if any(f.level == MEDIUM for f in self.findings):
            return MEDIUM
        return LOW if self.findings else "none"

    def sorted_findings(self) -> List[Finding]:
        return sorted(self.findings, key=lambda f: (_ORDER[f.level], f.file, f.line))

    def to_dict(self) -> dict:
        return {
            "risk": self.risk,
            "capabilities": sorted(self.capabilities),
            "files_scanned": self.files_scanned,
            "findings": [
                {"level": f.level, "category": f.category, "message": f.message,
                 "file": f.file, "line": f.line}
                for f in self.sorted_findings()
            ],
            "parse_errors": [{"file": f, "error": e} for f, e in self.parse_errors],
        }


def _dotted(node: ast.AST) -> Optional[str]:
    """Return the dotted name of an attribute/name node, e.g. ``os.system``."""
    parts = []
    while isinstance(node, ast.Attribute):
        parts.append(node.attr)
        node = node.value
    if isinstance(node, ast.Name):
        parts.append(node.id)
        return ".".join(reversed(parts))
    return None


class _Visitor(ast.NodeVisitor):
    def __init__(self, rel: str, report: ScanReport):
        self.rel = rel
        self.report = report
        self._imported_b64 = False

    def _add(self, level, category, message, line):
        self.report.findings.append(Finding(level, category, message, self.rel, line))

    def _cap(self, cap):
        if cap:
            self.report.capabilities.add(cap)

    def visit_Import(self, node: ast.Import):
        for a in node.names:
            top = a.name.split(".")[0]
            if top in ("base64", "codecs", "binascii"):
                self._imported_b64 = True
            if top in _MODULE_CAP:
                cap, level = _MODULE_CAP[top]
                self._cap(cap)
                self._add(level, cap or "dynamic-code",
                          f"imports '{a.name}'", node.lineno)
        self.generic_visit(node)

    def visit_ImportFrom(self, node: ast.ImportFrom):
        top = (node.module or "").split(".")[0]
        if top in ("base64", "codecs", "binascii"):
            self._imported_b64 = True
        if top in _MODULE_CAP:
            cap, level = _MODULE_CAP[top]
            self._cap(cap)
            self._add(level, cap or "dynamic-code",
                      f"imports from '{node.module}'", node.lineno)
        self.generic_visit(node)

    def visit_Call(self, node: ast.Call):
        dotted = _dotted(node.func)
        # builtin danger calls
        if isinstance(node.func, ast.Name) and node.func.id in _DANGER_BUILTINS:
            level, msg = _DANGER_BUILTINS[node.func.id]
            cat = "dynamic-code"
            self._add(level, cat, msg, node.lineno)
            # obfuscation heuristic: exec/eval fed by base64/decode
            if node.func.id in ("exec", "eval") and self._looks_decoded(node):
                self._add(HIGH, "obfuscation",
                          "executes decoded/obfuscated data — classic malware pattern",
                          node.lineno)
        elif dotted:
            if dotted in _CALL_SIGNALS:
                cap, level, msg = _CALL_SIGNALS[dotted]
                self._cap(cap)
                self._add(level, cap or "dynamic-code", msg, node.lineno)
            # open(..., 'w'/'a'/'x') → filesystem write
            elif dotted == "open" or dotted.endswith(".open"):
                self._flag_open_write(node)
        self.generic_visit(node)

    def _flag_open_write(self, node: ast.Call):
        mode = None
        if len(node.args) >= 2 and isinstance(node.args[1], ast.Constant):
            mode = node.args[1].value
        for kw in node.keywords:
            if kw.arg == "mode" and isinstance(kw.value, ast.Constant):
                mode = kw.value.value
        if isinstance(mode, str) and any(c in mode for c in ("w", "a", "x", "+")):
            self._cap(CAP_FS)
            self._add(MEDIUM, CAP_FS, f"opens a file for writing (mode '{mode}')",
                      node.lineno)

    def _looks_decoded(self, node: ast.Call) -> bool:
        if not node.args:
            return False
        arg = node.args[0]
        # exec(base64.b64decode(...)) / exec(codecs.decode(...)) / exec(bytes.fromhex(..))
        inner = arg
        for _ in range(3):
            if isinstance(inner, ast.Call):
                name = _dotted(inner.func) or ""
                if any(k in name for k in ("b64decode", "decode", "fromhex",
                                           "decompress", "loads")):
                    return True
                inner = inner.args[0] if inner.args else inner
            else:
                break
        return self._imported_b64 and isinstance(arg, (ast.Name, ast.Call))


def scan_source(source: str, rel: str = "<string>") -> ScanReport:
    """Scan a single source string; returns a one-file report."""
    report = ScanReport()
    report.files_scanned = 1
    try:
        tree = ast.parse(source, filename=rel)
    except SyntaxError as e:
        report.parse_errors.append((rel, str(e)))
        return report
    _Visitor(rel, report).visit(tree)
    return report


def scan_package(pkg_dir: str) -> ScanReport:
    """Scan every ``.py`` under a plugin package folder."""
    report = ScanReport()
    for root, dirs, files in os.walk(pkg_dir):
        dirs[:] = [d for d in dirs if d not in ("__pycache__", ".git")]
        for name in sorted(files):
            if not name.endswith(".py"):
                continue
            path = os.path.join(root, name)
            rel = os.path.relpath(path, pkg_dir).replace("\\", "/")
            try:
                with open(path, "r", encoding="utf-8", errors="replace") as fh:
                    src = fh.read()
            except OSError as e:
                report.parse_errors.append((rel, str(e)))
                continue
            one = scan_source(src, rel)
            report.findings.extend(one.findings)
            report.capabilities |= one.capabilities
            report.parse_errors.extend(one.parse_errors)
            report.files_scanned += 1
    return report


__all__ = [
    "scan_package", "scan_source", "ScanReport", "Finding",
    "HIGH", "MEDIUM", "LOW",
    "CAP_NETWORK", "CAP_SUBPROCESS", "CAP_FS", "CAP_NATIVE", "CAP_INPUT", "CAP_GUI",
]
