Skills › Scientific Agent Skills › matlab › scan_m_code.py
scan_m_code.py
433 lines · skills/matlab/scripts/scan_m_code.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded static risk triage for MATLAB source and opaque artifacts."""
4from __future__ import annotations
6import argparse
7import re
8import sys
9from collections import Counter
10from pathlib import Path
11from typing import Any
13from _common import (
14 CliError,
15 bounded_int,
16 checked_input,
17 checked_root,
18 emit_json,
19 fail_json,
20 read_text,
21 relative_id,
22)
24TOOL = "scan_m_code"
25SEVERITY = {"low": 1, "medium": 2, "high": 3, "critical": 4}
26SOURCE_SUFFIX = ".m"
27OPAQUE_SUFFIXES = {
28 ".fig": ("opaque_figure", "high", "MATLAB figure object file is opaque."),
29 ".mat": ("mat_file", "high", "MAT file can contain objects and callbacks."),
30 ".mlapp": ("opaque_app", "high", "MATLAB app archive is opaque and executable."),
31 ".mlx": ("opaque_live_script", "high", "Live script archive is opaque."),
32 ".p": ("protected_code", "high", "Protected MATLAB code is not reviewable."),
33 ".prj": ("project_file", "medium", "Project metadata can configure actions."),
34 ".slx": ("simulink_model", "high", "Simulink model archive can execute callbacks."),
35 ".mdl": ("simulink_model", "high", "Simulink model can execute callbacks."),
36}
37RULES: tuple[tuple[str, str, str, re.Pattern[str]], ...] = (
38 (
39 "dynamic_eval",
40 "critical",
41 "Dynamic evaluation can execute text as MATLAB code.",
42 re.compile(r"\b(?:eval|evalin)\s*\(", re.IGNORECASE),
43 ),
44 (
45 "workspace_injection",
46 "high",
47 "assignin mutates another workspace and can hide data flow.",
48 re.compile(r"\bassignin\s*\(", re.IGNORECASE),
49 ),
50 (
51 "text_dispatch",
52 "high",
53 "Review feval/str2func inputs; text-derived dispatch executes code.",
54 re.compile(r"\b(?:feval|str2func)\s*\(", re.IGNORECASE),
55 ),
56 (
57 "shell_execution",
58 "critical",
59 "Shell entry point can execute external commands.",
60 re.compile(r"(?:^\s*!|\b(?:system|unix|dos)\s*\()", re.IGNORECASE),
61 ),
62 (
63 "python_execution",
64 "high",
65 "Python integration can execute Python/native code.",
66 re.compile(r"\b(?:pyrun|pyrunfile|pyenv)\s*\(|\bpy\.", re.IGNORECASE),
67 ),
68 (
69 "java_execution",
70 "high",
71 "Java integration or class-path mutation crosses the runtime boundary.",
72 re.compile(
73 r"\b(?:javaObject|javaMethod|javaaddpath|javarmpath|javaclasspath)"
74 r"\s*\(|\bjava\.",
75 re.IGNORECASE,
76 ),
77 ),
78 (
79 "dotnet_execution",
80 "high",
81 ".NET assembly/type access can execute external code.",
82 re.compile(r"\bNET\.addAssembly\s*\(|\bSystem\.", re.IGNORECASE),
83 ),
84 (
85 "native_execution",
86 "critical",
87 "Native library or MEX entry point can execute process-native code.",
88 re.compile(
89 r"\b(?:mex|loadlibrary|calllib|libpointer|javaaddpath)\s*\(",
90 re.IGNORECASE,
91 ),
92 ),
93 (
94 "code_generation",
95 "high",
96 "Code generation/build invokes separate products and toolchains.",
97 re.compile(r"\b(?:codegen|mcc|compiler\.build)\s*\(", re.IGNORECASE),
98 ),
99 (
100 "mat_load",
101 "high",
102 "Loading MAT/object data can invoke installed class deserialization code.",
103 re.compile(r"(?:^\s*load(?:\s|\()|\bload\s*\()", re.IGNORECASE),
104 ),
105 (
106 "deserialization_callback",
107 "critical",
108 "Object load callback/custom serialization executes during restoration.",
109 re.compile(
110 r"\bloadobj\s*\(|\bloadObjectImpl\s*\(|"
111 r"matlab\.mixin\.CustomElementSerialization",
112 re.IGNORECASE,
113 ),
114 ),
115 (
116 "network_io",
117 "high",
118 "Network API can transmit data or retrieve executable/untrusted content.",
119 re.compile(
120 r"\b(?:webread|webwrite|websave|urlread|urlwrite|tcpclient|udpport)"
121 r"\s*\(",
122 re.IGNORECASE,
123 ),
124 ),
125 (
126 "callback",
127 "medium",
128 "Review callback provenance and lifecycle.",
129 re.compile(
130 r"\b(?:addlistener|timer)\s*\(|"
131 r"\b(?:Callback|TimerFcn|StartFcn|StopFcn)\b",
132 re.IGNORECASE,
133 ),
134 ),
135 (
136 "path_mutation",
137 "medium",
138 "Path mutation can shadow trusted functions.",
139 re.compile(
140 r"\b(?:addpath|rmpath|path|userpath|savepath|genpath)\s*\(",
141 re.IGNORECASE,
142 ),
143 ),
144 (
145 "file_mutation",
146 "medium",
147 "File mutation requires reviewed local paths and collision policy.",
148 re.compile(
149 r"\b(?:delete|movefile|copyfile|mkdir|rmdir|fopen|diary)\s*\(",
150 re.IGNORECASE,
151 ),
152 ),
153 (
154 "serialization_write",
155 "medium",
156 "Save/export can overwrite files or serialize executable objects.",
157 re.compile(r"\b(?:save|savefig|exportgraphics|writetable)\s*\(", re.IGNORECASE),
158 ),
159 (
160 "interactive_input",
161 "medium",
162 "Interactive input/dialog can hang or fail in batch mode.",
163 re.compile(
164 r"\b(?:input|uigetfile|uiputfile|questdlg|inputdlg)\s*\(",
165 re.IGNORECASE,
166 ),
167 ),
168 (
169 "broad_clear",
170 "low",
171 "Broad clearing hides workspace/function-state dependencies.",
172 re.compile(r"\bclear\s+all\b", re.IGNORECASE),
173 ),
174)
177def _without_comments(line: str, *, in_block: bool) -> tuple[str, bool]:
178 stripped = line.lstrip()
179 if in_block:
180 if stripped.startswith("%}"):
181 return "", False
182 return "", True
183 if stripped.startswith("%{"):
184 return "", True
185 single = False
186 double = False
187 index = 0
188 while index < len(line):
189 char = line[index]
190 if char == "'" and not double:
191 if single and index + 1 < len(line) and line[index + 1] == "'":
192 index += 2
193 continue
194 single = not single
195 elif char == '"' and not single:
196 if double and index + 1 < len(line) and line[index + 1] == '"':
197 index += 2
198 continue
199 double = not double
200 elif char == "%" and not single and not double:
201 return line[:index], False
202 index += 1
203 return line, False
206def scan_source(
207 path: Path, root: Path, *, max_file_bytes: int
208) -> list[dict[str, Any]]:
209 findings: list[dict[str, Any]] = []
210 text = read_text(path, max_bytes=max_file_bytes)
211 in_block = False
212 for line_number, line in enumerate(text.splitlines(), start=1):
213 code, in_block = _without_comments(line, in_block=in_block)
214 for rule, severity, message, pattern in RULES:
215 for match in pattern.finditer(code):
216 findings.append(
217 {
218 "column": match.start() + 1,
219 "file": relative_id(path, root),
220 "line": line_number,
221 "message": message,
222 "rule": rule,
223 "severity": severity,
224 }
225 )
226 if path.name.casefold() in {"startup.m", "finish.m", "pathdef.m"}:
227 findings.append(
228 {
229 "column": 1,
230 "file": relative_id(path, root),
231 "line": 1,
232 "message": "Lifecycle/path file can execute implicitly.",
233 "rule": "implicit_lifecycle_file",
234 "severity": "high",
235 }
236 )
237 return findings
240def _opaque_finding(path: Path, root: Path) -> dict[str, Any] | None:
241 suffix = path.suffix.casefold()
242 details = OPAQUE_SUFFIXES.get(suffix)
243 if suffix.startswith(".mex"):
244 details = (
245 "mex_binary",
246 "critical",
247 "MEX file is unreviewed native executable code.",
248 )
249 if details is None:
250 return None
251 rule, severity, message = details
252 return {
253 "column": None,
254 "file": relative_id(path, root),
255 "line": None,
256 "message": message,
257 "rule": rule,
258 "severity": severity,
259 }
262def collect(
263 source: Path,
264 *,
265 root: Path,
266 recursive: bool,
267 max_files: int,
268 max_file_bytes: int,
269 max_total_bytes: int,
270) -> list[Path]:
271 if source.is_file():
272 suffix = source.suffix.casefold()
273 supported = (
274 suffix == SOURCE_SUFFIX
275 or suffix in OPAQUE_SUFFIXES
276 or suffix.startswith(".mex")
277 )
278 if not supported:
279 raise CliError("input file is not .m or a recognized opaque artifact")
280 size = source.stat().st_size
281 if size > max_file_bytes or size > max_total_bytes:
282 raise CliError(
283 f"file is {size} bytes; per-file/total limits are "
284 f"{max_file_bytes}/{max_total_bytes}"
285 )
286 return [source]
287 output: list[Path] = []
288 stack = [source]
289 total_bytes = 0
290 while stack:
291 directory = stack.pop()
292 try:
293 entries = sorted(directory.iterdir(), key=lambda item: item.name.casefold())
294 except OSError as exc:
295 raise CliError(f"cannot read directory: {directory}") from exc
296 for entry in entries:
297 if entry.is_symlink():
298 raise CliError(f"symlink encountered during scan: {entry}")
299 if entry.is_dir():
300 if recursive:
301 stack.append(entry)
302 continue
303 if not entry.is_file():
304 continue
305 suffix = entry.suffix.casefold()
306 if suffix != SOURCE_SUFFIX and suffix not in OPAQUE_SUFFIXES and not (
307 suffix.startswith(".mex")
308 ):
309 continue
310 size = entry.stat().st_size
311 if size > max_file_bytes:
312 raise CliError(f"file is {size} bytes; limit is {max_file_bytes}")
313 total_bytes += size
314 if total_bytes > max_total_bytes:
315 raise CliError(
316 f"selected files total {total_bytes} bytes; "
317 f"limit is {max_total_bytes}"
318 )
319 output.append(entry)
320 if len(output) > max_files:
321 raise CliError(f"more than {max_files} relevant files found")
322 return sorted(output)
325def scan(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
326 root = checked_root(args.root)
327 source = checked_input(args.input, root=root, kind="any")
328 max_files = bounded_int(
329 args.max_files, name="max_files", minimum=1, maximum=5000
330 )
331 max_file_bytes = bounded_int(
332 args.max_file_bytes,
333 name="max_file_bytes",
334 minimum=1,
335 maximum=64 * 1024 * 1024,
336 )
337 max_total_bytes = bounded_int(
338 args.max_total_bytes,
339 name="max_total_bytes",
340 minimum=1,
341 maximum=256 * 1024 * 1024,
342 )
343 files = collect(
344 source,
345 root=root,
346 recursive=args.recursive,
347 max_files=max_files,
348 max_file_bytes=max_file_bytes,
349 max_total_bytes=max_total_bytes,
350 )
351 findings: list[dict[str, Any]] = []
352 scanned_text = 0
353 opaque = 0
354 for path in files:
355 if path.suffix.casefold() == ".m":
356 scanned_text += 1
357 findings.extend(
358 scan_source(path, root, max_file_bytes=max_file_bytes)
359 )
360 else:
361 finding = _opaque_finding(path, root)
362 if finding:
363 opaque += 1
364 findings.append(finding)
365 findings.sort(
366 key=lambda item: (
367 item["file"],
368 item["line"] if item["line"] is not None else 0,
369 item["column"] if item["column"] is not None else 0,
370 item["rule"],
371 )
372 )
373 counts = Counter(item["severity"] for item in findings)
374 threshold = 5 if args.fail_on == "none" else SEVERITY[args.fail_on]
375 failing = sum(SEVERITY[item["severity"]] >= threshold for item in findings)
376 report = {
377 "content_emitted": False,
378 "executes": False,
379 "fail_on": args.fail_on,
380 "files_considered": len(files),
381 "findings": findings,
382 "network_accessed": False,
383 "ok": failing == 0,
384 "opaque_files": opaque,
385 "static_only": True,
386 "summary": {
387 severity: counts.get(severity, 0)
388 for severity in ("critical", "high", "medium", "low")
389 },
390 "text_files_scanned": scanned_text,
391 "tool": TOOL,
392 "warning": (
393 "Static pattern triage can miss dynamic behavior and can produce "
394 "false positives; never treat a clean report as permission to execute."
395 ),
396 }
397 return report, 1 if failing else 0
400def parser() -> argparse.ArgumentParser:
401 result = argparse.ArgumentParser(
402 description=(
403 "Statically triage reviewed MATLAB text and flag opaque artifacts. "
404 "No MATLAB/Octave runtime is launched."
405 )
406 )
407 result.add_argument("input", help="local .m file or directory")
408 result.add_argument("--root", default=".", help="allowed local root")
409 result.add_argument(
410 "--recursive", action="store_true", help="scan nested directories"
411 )
412 result.add_argument(
413 "--fail-on",
414 choices=("critical", "high", "medium", "low", "none"),
415 default="high",
416 )
417 result.add_argument("--max-files", default=500)
418 result.add_argument("--max-file-bytes", default=4 * 1024 * 1024)
419 result.add_argument("--max-total-bytes", default=32 * 1024 * 1024)
420 return result
423def main() -> int:
424 try:
425 report, status = scan(parser().parse_args())
426 emit_json(report)
427 return status
428 except CliError as exc:
429 return fail_json(TOOL, exc)
432if __name__ == "__main__":
433 sys.exit(main())