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