app/tools/ast_parser.py

import ast
import os
import re as _re
from typing import Any


def parse_python_file(file_path: str) -> dict[str, Any]:
    result: dict[str, Any] = {
        "file": file_path,
        "functions": [],
        "classes": [],
        "imports": [],
        "errors": [],
    }
    try:
        with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
            source = f.read()
        tree = ast.parse(source)
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                result["functions"].append(
                    {
                        "name": node.name,
                        "line": node.lineno,
                        "args": [arg.arg for arg in node.args.args],
                        "docstring": ast.get_docstring(node) or "",
                    }
                )
            elif isinstance(node, ast.ClassDef):
                result["classes"].append(
                    {
                        "name": node.name,
                        "line": node.lineno,
                        "docstring": ast.get_docstring(node) or "",
                    }
                )
            elif isinstance(node, ast.Import):
                for alias in node.names:
                    result["imports"].append(alias.name)
            elif isinstance(node, ast.ImportFrom):
                result["imports"].append(node.module or "")
    except SyntaxError as e:
        result["errors"].append(f"SyntaxError: {str(e)}")
    except Exception as e:
        result["errors"].append(f"Error: {str(e)}")
    return result


def parse_repository(local_path: str) -> list[dict[str, Any]]:
    results: list[dict[str, Any]] = []
    for root, dirs, files in os.walk(local_path):
        dirs[:] = [
            d
            for d in dirs
            if d not in {".git", "__pycache__", ".venv", "venv", "node_modules"}
        ]
        for file in files:
            if file.endswith(".py"):
                results.append(parse_python_file(os.path.join(root, file)))
    return results

_SECRET_PATTERNS = [
    "api_key", "apikey", "secret", "password", "passwd", "token",
    "auth_token", "access_token", "private_key",
]


def detect_security_issues(file_path: str) -> list[dict]:
    """
    AST-based security scan. Detects dangerous patterns without running the code.
    Returns a list of findings dicts with keys: rule, line, detail, severity.
    """
    findings: list[dict] = []
    try:
        with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
            source = f.read()
        tree = ast.parse(source)
    except (SyntaxError, OSError):
        return findings

    for node in ast.walk(tree):
        # eval() / exec() called with any argument
        if isinstance(node, ast.Call):
            func_name = ""
            if isinstance(node.func, ast.Name):
                func_name = node.func.id
            elif isinstance(node.func, ast.Attribute):
                func_name = node.func.attr

            if func_name == "eval":
                findings.append({
                    "rule": "dangerous-eval",
                    "line": node.lineno,
                    "detail": "eval() can execute arbitrary code",
                    "severity": "critical",
                    "file": file_path,
                })

            if func_name == "exec":
                findings.append({
                    "rule": "dangerous-exec",
                    "line": node.lineno,
                    "detail": "exec() can execute arbitrary code",
                    "severity": "critical",
                    "file": file_path,
                })

            # subprocess with shell=True
            if func_name in ("run", "call", "Popen", "check_output"):
                for kw in node.keywords:
                    if kw.arg == "shell" and isinstance(kw.value, ast.Constant) and kw.value.value is True:
                        findings.append({
                            "rule": "shell-injection",
                            "line": node.lineno,
                            "detail": f"subprocess.{func_name}(..., shell=True) is a shell injection risk",
                            "severity": "high",
                            "file": file_path,
                        })

            # pickle.loads / pickle.load
            if func_name in ("loads", "load") and isinstance(node.func, ast.Attribute):
                if isinstance(node.func.value, ast.Name) and node.func.value.id == "pickle":
                    findings.append({
                        "rule": "unsafe-pickle",
                        "line": node.lineno,
                        "detail": "pickle.loads() can execute arbitrary code on deserialization",
                        "severity": "high",
                        "file": file_path,
                    })

        # Hardcoded secrets: API_KEY = "some_literal"
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name):
                    varname = target.id.lower()
                    if any(pat in varname for pat in _SECRET_PATTERNS):
                        if isinstance(node.value, ast.Constant) and isinstance(node.value.value, str) and len(node.value.value) > 4:
                            findings.append({
                                "rule": "hardcoded-secret",
                                "line": node.lineno,
                                "detail": f"Possible hardcoded secret in variable '{target.id}'",
                                "severity": "high",
                                "file": file_path,
                            })

    return findings


def scan_repository_security(local_path: str) -> list[dict]:
    """Run AST security scan across all Python files in the repo."""
    all_findings: list[dict] = []
    for root, dirs, files in os.walk(local_path):
        dirs[:] = [
            d for d in dirs
            if d not in {".git", "__pycache__", ".venv", "venv", "node_modules"}
        ]
        for file in files:
            if file.endswith(".py"):
                all_findings.extend(detect_security_issues(os.path.join(root, file)))
    return all_findings


_REGEX_SECRET_PATTERNS = [
    (r"(?i)(api_key|apikey|api-key)\s*[=:]\s*['\"]([A-Za-z0-9_\-]{16,})['\"]", "hardcoded-api-key"),
    (r"(?i)(secret|password|passwd|token|auth_token|access_token)\s*[=:]\s*['\"]([A-Za-z0-9_\-]{8,})['\"]", "hardcoded-secret"),
    (r"(?i)-----BEGIN (RSA|EC|OPENSSH) PRIVATE KEY-----", "exposed-private-key"),
    (r"(?i)(AWS_SECRET_ACCESS_KEY|GITHUB_TOKEN|SLACK_TOKEN)\s*[=:]\s*['\"]?([A-Za-z0-9/+=]{16,})['\"]?", "cloud-credential"),
]

_SCAN_EXTENSIONS = {".py", ".js", ".ts", ".env", ".yaml", ".yml", ".json", ".sh", ".cfg", ".ini"}


def scan_secrets_regex(local_path: str) -> list[dict]:
    """
    Regex-based secret scan across all common file types (not just Python).
    Complements the AST-based scanner which only covers Python.
    """
    findings: list[dict] = []
    for root, dirs, files in os.walk(local_path):
        dirs[:] = [d for d in dirs if d not in {".git", "__pycache__", ".venv", "venv", "node_modules"}]
        for file in files:
            ext = os.path.splitext(file)[1].lower()
            if ext not in _SCAN_EXTENSIONS:
                continue
            file_path = os.path.join(root, file)
            try:
                with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
                    for lineno, line in enumerate(f, 1):
                        for pattern, rule in _REGEX_SECRET_PATTERNS:
                            if _re.search(pattern, line):
                                findings.append({
                                    "rule": rule,
                                    "line": lineno,
                                    "file": file_path,
                                    "detail": f"Possible secret matched pattern '{rule}'",
                                    "severity": "high",
                                })
            except Exception:  # nosec B110
                pass
    return findings