| """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) |
| 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) |
|
|
