File size: 10,318 Bytes
21a9138 | 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 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 | """Orchestration: units -> per-language extractors -> file-level hooks -> dedupe -> report."""
from __future__ import annotations
import re
import time
from pathlib import Path
from .model import SCHEMA, KINDS, Behaviour, clip, dedupe, render_text
from . import config, inputs, patterns as P, pyjs, shell
TEST_PATH = re.compile(r"(^|/)(tests?|__tests__|spec|testing)/|(^|/)test_[^/]+\.py$|_test\.py$|\.(test|spec)\.[jt]sx?$")
def file_context(path: str) -> str:
name = path.rsplit("/", 1)[-1]
tags = []
if name == "conftest.py":
tags.append("pytest conftest (auto-loaded)")
elif name == "setup.py":
tags.append("setup.py (runs at install/build)")
elif name in ("sitecustomize.py", "usercustomize.py"):
tags.append("interpreter startup")
elif name == "hatch_build.py":
tags.append("hatch build hook")
elif "/migrations/" in "/" + path and name.endswith(".py"):
tags.append("Django migration (runs on migrate)")
elif re.search(r"(^|/)(\.husky|\.git/hooks|\.githooks)/", path):
tags.append("git hook script")
if TEST_PATH.search(path) and not tags:
tags.append("test code")
return "; ".join(tags)
def _file_hooks(u, ctx):
out = []
name = u.path.rsplit("/", 1)[-1]
if u.focus is not None and not u.focus:
return out
ln = u.first_focus()
ev = u.text.split("\n")[ln - 1].strip() if u.text else ""
def add(op, det, kind="install_or_startup_hook"):
out.append(Behaviour(kind, u.path, ln, op, ev, det, ctx, "manifest"))
if name == "conftest.py":
add("conftest.py", "auto-loaded by pytest" + ("; new file" if u.status == "added" else ""))
elif name == "setup.py":
add("setup.py", "executed by pip/setuptools at install/build" + ("; new file" if u.status == "added" else ""))
elif name in ("sitecustomize.py", "usercustomize.py"):
add(name, "imported at every interpreter start")
add("python startup (sitecustomize/.pth)", "startup file in change set", kind="persistence")
elif name == "hatch_build.py":
add("hatch_build.py", "hatch build/metadata hook module")
elif re.search(r"(^|/)(\.husky|\.git/hooks|\.githooks)/[^/]+$", u.path) and not name.startswith("_"):
add(f"git hook {name}", "runs on git operations")
elif u.path.endswith("/__init__.py") or u.path == "__init__.py":
pass
if "/migrations/" in "/" + u.path and re.search(r"\bRunPython\b|\bRunSQL\b", u.text):
m = re.search(r"\b(RunPython|RunSQL)\b", u.text)
lnn = u.text.count("\n", 0, m.start()) + 1
if u.in_focus(lnn):
out.append(Behaviour("install_or_startup_hook", u.path, lnn, f"migrations.{m.group(1)}",
u.text.split("\n")[lnn - 1].strip(), "code runs on migrate", ctx, "regex"))
return out
def _sfc_scripts(u):
"""Keep only <script> block lines of .vue/.svelte files (others blanked, line numbers preserved)."""
lines = u.text.split("\n")
keep, inside, lang = [], False, "javascript"
for l in lines:
if not inside and re.search(r"<script\b", l):
inside = True
if re.search(r"lang=['\"]ts", l):
lang = "typescript"
keep.append(re.sub(r".*<script[^>]*>", "", l))
if "</script>" in l:
inside = False
keep[-1] = keep[-1].split("</script>")[0]
continue
if inside and "</script>" in l:
keep.append(l.split("</script>")[0])
inside = False
continue
keep.append(l if inside else "")
return "\n".join(keep), lang
def _regex_code_fallback(u, ctx):
"""Used for languages without an extractor (and when tree-sitter is unavailable): line patterns only."""
e = config.Emitter(u, ctx)
for i, l in enumerate(e.lines, 1):
s = l.strip()
if not s or not u.in_focus(i) or s.startswith(("#", "//", "*", "/*", "--")):
continue
for label in P.secret_hits(s):
e.add("secret_or_credential_path", i, label, source="regex")
for rx, label in P.TLS_OFF:
if rx.search(s):
e.add("security_setting_change", i, label, "TLS verification off", source="regex")
for op, det in P.obfuscation(l, "other"):
e.add("obfuscation_markers", i, op, det, source="regex")
for m in re.finditer(r"""['"`]((?:https?|wss?|ftp)://[^'"`\s]+)['"`]""", s):
if re.search(r"\b(fetch|get|post|request|urlopen|curl|http\.|Http|open|connect|download|send)\w*\s*\(", s, re.I):
e.add("network_egress", i, m.group(1), "URL in request-like call (regex)", source="regex")
if re.search(r"(\b(os\.)?system|\bpopen|\bspawn\w*|\bPopen|\bsubprocess\.\w+|\bexecSync|\bexecFile\w*|child_process\.\w+|"
r"Process\.Start|Runtime\.getRuntime\(\)\.exec|exec\.Command|shell_exec|passthru|proc_open)\s*\(", s):
e.add("process_spawn", i, clip(s, 100), "process call (regex)", source="regex")
if re.search(r"\b(eval|exec|Function|loadstring|instance_eval|class_eval)\s*\(", s) and "process" not in s:
e.add("dynamic_import_or_eval", i, clip(s, 100), "eval-like call (regex)", source="regex")
for b in shell.analyse("\n".join(l if shell.looks_like_shell(l) else "" for l in e.lines), u.path,
in_focus=u.in_focus, context=ctx, embedded="command text (regex)", emit_spawn=False):
if b.kind != "env_read":
e.out.append(b)
return e.out
def analyse_unit(u, ctx_extra=""):
ctx = "; ".join(x for x in (file_context(u.path), ctx_extra) if x)
out = []
if u.binary_note or u.lang == "binary":
return config.binary(u, ctx)
if u.focus is not None and not u.focus:
return out
out.extend(_file_hooks(u, ctx))
r = config.dispatch(u, ctx)
if r is not None:
return out + r
lang = u.lang
if lang == "sfc":
text, lang = _sfc_scripts(u)
u = inputs.Unit(u.path, lang, text=text, focus=u.focus, status=u.status, size=u.size)
if lang in ("python", "javascript", "typescript", "tsx"):
if pyjs.available():
out.extend(pyjs.analyse(u, lang, ctx))
else:
out.extend(_regex_code_fallback(u, ctx))
elif lang == "shell":
out.extend(shell.analyse(u.text, u.path, in_focus=u.in_focus, context=ctx))
elif lang == "doc":
pass
else:
out.extend(_regex_code_fallback(u, ctx))
if lang in ("javascript", "typescript") and (u.path.endswith((".min.js", ".bundle.js")) or u.size > 300_000) and u.focus is None:
out.append(Behaviour("archive_or_bundle", u.path, 1, f"bundled/minified JS ({u.size} bytes)", "", "bundle", ctx, "manifest"))
return out
def _task_literals(task):
if not task:
return {}
urls = sorted(set(re.findall(r"(?:https?|wss?)://[^\s'\"<>)]+", task)))[:20]
hosts = sorted(set(re.findall(r"\b(?:[a-z0-9-]+\.)+(?:com|org|net|io|dev|invalid|test|example|local|internal|ai)\b", task, re.I)) - {u.split('/')[2] for u in urls if '//' in u})[:20]
paths = sorted(set(re.findall(r"(?:~|\.{0,2})/[\w./-]{3,}", task)))[:20]
envs = sorted(set(re.findall(r"\b[A-Z][A-Z0-9]*_[A-Z0-9_]{2,}\b", task)))[:20]
return {k: v for k, v in (("urls", urls), ("hosts", hosts), ("paths", paths), ("env_names", envs)) if v}
def build_report(units, mode, target, task=None, contexts=None, extra=None, notes=None, t0=None):
t0 = t0 or time.perf_counter()
contexts = contexts or {}
found = []
for u in units:
try:
found.extend(analyse_unit(u, contexts.get(u.path, "")))
except RecursionError:
(notes := notes if notes is not None else []).append(f"{u.path}: recursion limit, skipped")
have = {(b.kind, b.path, b.line) for b in found}
found.extend(b for b in (extra or []) if (b.kind, b.path, b.line) not in have) # optional extra behaviours: additions only
written = {(b.path, b.line) for b in found if b.kind == "persistence" and b.detail.startswith("write target")}
found = [b for b in found if not (b.kind == "persistence" and b.detail.startswith("location referenced")
and (b.path, b.line) in written)]
bs = dedupe(found)
files = len(units)
lines = sum((len(u.focus) if u.focus is not None else u.text.count("\n") + (1 if u.text else 0)) for u in units)
rep = {"schema": SCHEMA, "mode": mode, "target": target, "descriptive_only": True,
"stats": {"files": files, "lines": lines, "seconds": round(time.perf_counter() - t0, 4),
"extractor": "tree-sitter" if pyjs.available() else "regex-fallback"},
"counts": {k: n for k in KINDS if (n := sum(1 for b in bs if b.kind == k))},
"behaviours": [b.to_json() for b in bs]}
if task:
rep["task"] = clip(task, 2000)
tl = _task_literals(task)
if tl:
rep["task_literals"] = tl
if not pyjs.available():
notes = (notes or []) + ["tree-sitter unavailable: regex fallback, reduced coverage (install tree-sitter and tree-sitter-language-pack)"]
if notes:
rep["notes"] = notes
rep["text"] = render_text(rep)
return rep
def analyse_package(root, task=None, include_node_modules=False):
t0 = time.perf_counter()
root = Path(root)
units = inputs.read_package(root, include_node_modules)
contexts = {}
for u in units:
if u.path.rsplit("/", 1)[-1] == "package.json":
base = u.path.rsplit("/", 1)[0] + "/" if "/" in u.path else ""
for f, c in config.lifecycle_targets(u.text).items():
contexts[base + f] = c
extra, notes = [], []
return build_report(units, "package", str(root), task, contexts, extra, notes, t0)
def analyse_diff(diff_text, task=None, target="diff"):
t0 = time.perf_counter()
return build_report(inputs.parse_diff(diff_text), "diff", target, task, t0=t0)
def analyse_snippet(text, lang=None, task=None, target="snippet"):
t0 = time.perf_counter()
u = inputs.snippet_unit(text, lang, name=target)
return build_report([u], "snippet", f"{target} ({u.lang})", task, contexts={u.path: "code-mode snippet (runs immediately)"}, t0=t0)
|