File size: 8,155 Bytes
007ba8f
 
d860535
f8b17b7
007ba8f
 
f8b17b7
 
007ba8f
 
 
 
f8b17b7
007ba8f
 
 
 
 
 
 
a79668c
 
 
 
 
 
 
 
007ba8f
a79668c
 
 
 
 
 
 
007ba8f
 
 
 
f8b17b7
007ba8f
 
 
 
 
 
 
f8b17b7
 
007ba8f
a79668c
 
 
 
 
007ba8f
 
f8b17b7
a79668c
a6031f7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d860535
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
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