| """Manifests, configuration, CI workflows, Dockerfiles, Makefiles, .pth files and binary/archive artefacts.""" |
| from __future__ import annotations |
|
|
| import json |
| import re |
|
|
| from .model import Behaviour, clip |
| from . import patterns as P |
| from . import shell |
|
|
| PKG_META = {"url", "homepage", "repository", "bugs", "funding", "email", "author", "license", "$schema", "description", |
| "name", "version", "main", "module", "types", "typings", "browser", "registry", "type", "web", "docs", |
| "issues", "source", "changelog", "documentation", "icon", "logo", "image"} |
| NPM_LIFECYCLE = ("preinstall", "install", "postinstall", "preuninstall", "uninstall", "postuninstall", "prepare", |
| "prepublish", "prepublishOnly", "prepack", "postpack", "dependencies") |
| CI_PATH = re.compile(r"(^|/)(\.github/workflows/[^/]+\.ya?ml|\.gitlab-ci\.ya?ml|\.circleci/config\.ya?ml|" |
| r"azure-pipelines[^/]*\.ya?ml|\.travis\.ya?ml|bitbucket-pipelines\.ya?ml|\.drone\.ya?ml|" |
| r"\.buildkite/[^/]+\.ya?ml|\.woodpecker[^/]*\.ya?ml|\.github/actions/[^/]+/action\.ya?ml|action\.ya?ml)$") |
| MAGIC = [(b"\x7fELF", "ELF executable/library"), (b"MZ", "PE (Windows) executable"), |
| (b"\xcf\xfa\xed\xfe", "Mach-O 64"), (b"\xce\xfa\xed\xfe", "Mach-O"), (b"\xca\xfe\xba\xbe", "Mach-O fat / Java class"), |
| (b"\x00asm", "WebAssembly"), (b"PK\x03\x04", "zip archive"), (b"\x1f\x8b", "gzip"), |
| (b"7z\xbc\xaf\x27\x1c", "7z archive"), (b"Rar!", "rar archive"), (b"BZh", "bzip2"), (b"\xfd7zXZ", "xz"), |
| (b"\x89PNG", "PNG image"), (b"\xff\xd8\xff", "JPEG image"), (b"GIF8", "GIF image"), (b"%PDF", "PDF"), |
| (b"SQLite format 3", "SQLite database")] |
| MEDIA = ("PNG image", "JPEG image", "GIF image", "PDF") |
| MEDIA_EXT = (".png", ".jpg", ".jpeg", ".gif", ".ico", ".webp", ".bmp", ".woff", ".woff2", ".ttf", ".otf", ".eot", |
| ".mp3", ".mp4", ".wav", ".ogg", ".pdf", ".svgz", ".icns", ".avif") |
|
|
|
|
| class Emitter: |
| def __init__(self, unit, context=""): |
| self.u, self.ctx, self.out = unit, context, [] |
| self.lines = unit.text.split("\n") |
|
|
| def ev(self, ln): |
| return self.lines[ln - 1].strip() if 0 < ln <= len(self.lines) else "" |
|
|
| def add(self, kind, ln, operand, detail="", source="manifest", ctx=None, force=False): |
| if force or self.u.in_focus(ln): |
| self.out.append(Behaviour(kind, self.u.path, ln, clip(operand, 140) or "dynamic", self.ev(ln), detail, |
| self.ctx if ctx is None else ctx, source)) |
|
|
| def sh(self, text, ln, ctx=None, embedded="", emit_spawn=True): |
| self.out.extend(shell.analyse(text, self.u.path, base_line=ln, in_focus=self.u.in_focus, |
| context=self.ctx if ctx is None else ctx, embedded=embedded, |
| emit_spawn=emit_spawn)) |
|
|
| def common(self): |
| for i, l in enumerate(self.lines, 1): |
| s = l.strip() |
| if not s or s.startswith(("#", "//", ";")) or not self.u.in_focus(i): |
| continue |
| for label in P.secret_hits(s): |
| self.add("secret_or_credential_path", i, label, source="regex") |
| for rx, label in P.TLS_OFF: |
| if rx.search(s): |
| self.add("security_setting_change", i, label, "TLS verification off", source="regex") |
| for op, det in P.obfuscation(l, "other"): |
| self.add("obfuscation_markers", i, op, det, source="regex") |
|
|
|
|
| def package_json(u, ctx=""): |
| e = Emitter(u, ctx) |
| in_scripts = in_deps = in_publish = False |
| depth_marker = None |
| for i, l in enumerate(e.lines, 1): |
| s = l.strip() |
| m = re.match(r'"(scripts|dependencies|devDependencies|optionalDependencies|peerDependencies|publishConfig|' |
| r'bundleDependencies|bundledDependencies|overrides|resolutions)"\s*:\s*\{', s) |
| if m: |
| in_scripts, in_deps, in_publish = m.group(1) == "scripts", "ependencies" in m.group(1) or m.group(1) in ("overrides", "resolutions"), m.group(1) == "publishConfig" |
| depth_marker = len(l) - len(l.lstrip()) |
| continue |
| if s.startswith("}") and depth_marker is not None and len(l) - len(l.lstrip()) <= depth_marker: |
| in_scripts = in_deps = in_publish = False |
| depth_marker = None |
| continue |
| kv = re.match(r'"([^"]+)"\s*:\s*"((?:[^"\\]|\\.)*)"', s) |
| if not kv: |
| continue |
| key, val = kv.group(1), kv.group(2).encode().decode("unicode_escape", errors="replace") |
| diffmode_script = depth_marker is None and u.focus is not None and key in NPM_LIFECYCLE |
| if in_scripts or diffmode_script: |
| if key in NPM_LIFECYCLE: |
| e.add("install_or_startup_hook", i, f"npm {key}: {val}", "npm lifecycle script") |
| e.sh(val, i, ctx=f"npm {key}", emit_spawn=False) |
| else: |
| e.sh(val, i, ctx=f"npm script {key}", emit_spawn=False) |
| elif in_deps or depth_marker is None and u.focus is not None and key not in PKG_META and \ |
| re.match(r"(@[\w.-]+/)?[\w.-]+$", key) and re.match(r"(https?|git(\+\w+)?|github|gist|file|link):", val): |
| if re.match(r"(https?://|git(\+\w+)?:|git@|github:|gist:|bitbucket:|gitlab:)", val) or \ |
| re.match(r"^[\w-]+/[\w.-]+(#.*)?$", val): |
| e.add("network_egress", i, val, f"dependency {key} fetched from URL/git at install") |
| elif val.startswith(("file:", "link:")): |
| e.add("file_write", i, val, f"dependency {key} from local path", ) |
| elif in_publish and key == "registry" or key == "registry": |
| e.add("security_setting_change", i, f"registry {val}", "registry setting") |
| e.common() |
| return e.out |
|
|
|
|
| def lifecycle_targets(text): |
| """package mode: map files run by npm lifecycle scripts -> context label.""" |
| out = {} |
| try: |
| scripts = json.loads(text).get("scripts", {}) or {} |
| except Exception: |
| return out |
| for k, v in scripts.items(): |
| if k in NPM_LIFECYCLE and isinstance(v, str): |
| for m in re.finditer(r"\b(?:node|bun|deno run|python3?|sh|bash|tsx|ts-node)\s+(?:-\S+\s+)*([\w./-]+)", v): |
| out[m.group(1).removeprefix("./")] = f"npm {k}" |
| return out |
|
|
|
|
| def npmrc_like(u, ctx=""): |
| e = Emitter(u, ctx) |
| name = u.path.rsplit("/", 1)[-1] |
| for i, l in enumerate(e.lines, 1): |
| s = l.strip() |
| if not s or s.startswith(("#", ";")): |
| continue |
| for rx, label in P.REGISTRY: |
| m = rx.search(s) |
| if m: |
| e.add("security_setting_change", i, f"{label}: {m.group(m.lastindex)}", "registry/index change") |
| break |
| if re.search(r"_auth(Token)?\s*=|password\s*[=:]|//[^ ]+:_password", s, re.I): |
| e.add("secret_or_credential_path", i, f"{name} credential entry", "credential stored in config") |
| if re.match(r"(ignore-scripts\s*=\s*false|unsafe-perm\s*=\s*true)", s): |
| e.add("security_setting_change", i, s, "package manager setting") |
| if name == ".env" or name.startswith(".env.") or name == ".netrc": |
| e.add("secret_or_credential_path", u.first_focus(), f"{name} file", "credential/env file in change set", force=True) |
| e.common() |
| return e.out |
|
|
|
|
| def requirements(u, ctx=""): |
| e = Emitter(u, ctx) |
| for i, l in enumerate(e.lines, 1): |
| s = l.split(" #", 1)[0].strip() |
| if not s or s.startswith("#"): |
| continue |
| m = re.match(r"(-i|--index-url|--extra-index-url|--trusted-host|-f|--find-links)[ =]+(\S+)", s) |
| if m: |
| e.add("security_setting_change", i, f"pip {m.group(1)} {m.group(2)}", "package index change") |
| elif re.search(r"(@\s*|^)(https?://|git\+|svn\+|hg\+|file:)", s): |
| e.add("network_egress", i, s, "dependency fetched from URL at install") |
| return e.out |
|
|
|
|
| def toml_ini(u, ctx=""): |
| e = Emitter(u, ctx) |
| name = u.path.rsplit("/", 1)[-1] |
| section = "" |
| for i, l in enumerate(e.lines, 1): |
| s = l.strip() |
| if s.startswith("["): |
| section = s |
| if re.match(r"\[tool\.hatch\.(build\.(targets\.\w+\.)?hooks|metadata\.hooks)", s): |
| e.add("install_or_startup_hook", i, s, "hatch build/metadata hook") |
| if re.match(r"\[tool\.setuptools\.cmdclass\]|\[tool\.pdm\.build\]|\[tool\.poetry\.build\]", s): |
| e.add("install_or_startup_hook", i, s, "build hook configuration") |
| continue |
| if not s or s.startswith(("#", ";")): |
| continue |
| m = re.match(r"build-backend\s*=\s*['\"]([^'\"]+)", s) |
| if m: |
| e.add("install_or_startup_hook", i, f"build-backend {m.group(1)}", "build backend runs at install") |
| m = re.match(r"(backend-path)\s*=\s*(.+)", s) |
| if m: |
| e.add("install_or_startup_hook", i, f"backend-path {m.group(2)}", "in-tree build backend") |
| for rx, label in P.REGISTRY[3:]: |
| m = rx.search(s) |
| if m: |
| e.add("security_setting_change", i, f"{label}: {m.group(m.lastindex)}", "package index change") |
| break |
| if re.search(r"tool\.poetry\.source|tool\.pdm\.source|\[\[tool\.uv\.index", section) and re.match(r"url\s*=", s): |
| e.add("security_setting_change", i, "index " + s.split("=", 1)[1].strip(), "package index change") |
| if name in ("Cargo.toml",) and re.search(r"\bbuild\s*=", s): |
| e.add("install_or_startup_hook", i, s, "cargo build script") |
| e.common() |
| return e.out |
|
|
|
|
| def _yaml_keys(path): |
| if ".gitlab-ci" in path: |
| return r"(script|before_script|after_script)" |
| if ".travis" in path: |
| return r"(script|before_install|install|before_script|after_success|after_script|before_deploy)" |
| if ".circleci" in path or ".drone" in path or ".woodpecker" in path or ".buildkite" in path: |
| return r"(command|commands|run)" |
| if "azure-pipelines" in path: |
| return r"(script|bash|pwsh|powershell|inlineScript)" |
| if "bitbucket" in path: |
| return r"(script)" |
| return r"(run)" |
|
|
|
|
| def _yaml_blocks(lines, keys=r"(run)"): |
| """Yield (key, start_line, text) for run/script-like values in YAML (block scalars, inline, lists).""" |
| i = 0 |
| n = len(lines) |
| while i < n: |
| l = lines[i] |
| m = re.match(r"^(\s*)(?:-\s+)?" + keys + r"\s*:\s*(.*)$", l) |
| if not m: |
| i += 1 |
| continue |
| indent, key, val = len(m.group(1)), m.group(2), m.group(3).strip() |
| if val in ("|", "|-", "|+", ">", ">-", ">+") or val == "": |
| body, j = [], i + 1 |
| while j < n and (not lines[j].strip() or len(lines[j]) - len(lines[j].lstrip()) > indent): |
| body.append(lines[j]) |
| j += 1 |
| if val == "": |
| items = [(k, re.sub(r"^\s*-\s+", "", b)) for k, b in enumerate(body) if re.match(r"^\s*-\s+", b)] |
| for k, txt in items: |
| yield key, i + 2 + k, txt.strip().strip("'\"") |
| else: |
| ind = min((len(b) - len(b.lstrip()) for b in body if b.strip()), default=0) |
| yield key, i + 2, "\n".join(b[ind:] for b in body) |
| i = j |
| continue |
| if val.startswith("["): |
| for part in re.findall(r"""['"]([^'"]+)['"]""", val): |
| yield key, i + 1, part |
| else: |
| yield key, i + 1, val.strip("'\"") |
| i += 1 |
|
|
|
|
| def ci_yaml(u, ctx=""): |
| e = Emitter(u, ctx or "CI workflow") |
| text = u.text |
| triggers = [] |
| for i, l in enumerate(e.lines, 1): |
| s = l.strip() |
| m = re.match(r"^(on|true)\s*:\s*(.*)$", l) |
| if m and m.group(2): |
| triggers += [(i, t) for t in re.findall(r"[\w-]+", m.group(2))] |
| m = re.match(r"^\s{1,6}(push|pull_request|pull_request_target|schedule|workflow_dispatch|workflow_run|" |
| r"issue_comment|issues|release|repository_dispatch|merge_group|workflow_call|discussion|fork|watch)\s*:", l) |
| if m: |
| triggers.append((i, m.group(1))) |
| m = re.match(r"^\s*-?\s*cron\s*:\s*['\"]?([^'\"]+)", l) |
| if m: |
| triggers.append((i, f"schedule {m.group(1).strip()}")) |
| for sm in P.CI_SECRET.finditer(l): |
| em = re.match(r"^\s*([A-Za-z_][\w-]*)\s*:\s*", l) |
| det = f"exposed as {em.group(1)}" if em and em.group(1) not in ("with", "env") else "CI secret reference" |
| e.add("secret_or_credential_path", i, f"CI secret {sm.group(1)}", det) |
| if re.search(r"\bpermissions\s*:\s*write-all|^\s*(contents|packages|id-token|actions|pull-requests)\s*:\s*write", l): |
| e.add("security_setting_change", i, "CI token permission " + s, "privilege/permission") |
| if re.search(r"\$\{\{\s*github\.event\.(issue|pull_request|comment|review|head_commit|discussion)\.[\w.]*(title|body|head\.ref|label|message|name|email)", l): |
| e.add("dynamic_import_or_eval", i, re.search(r"\$\{\{[^}]*\}\}", l).group(0), "untrusted event text interpolated into the workflow") |
| if re.search(r"ref\s*:\s*\$\{\{\s*github\.event\.pull_request\.head\.(sha|ref)", l): |
| e.add("install_or_startup_hook", i, "checks out PR head code", "pull-request code in CI") |
| m = re.match(r"^\s*-?\s*uses\s*:\s*['\"]?([^\s'\"]+)", l) |
| if m and not re.match(r"(actions|github)/", m.group(1)) and not m.group(1).startswith("./"): |
| e.add("process_spawn", i, f"action {m.group(1)}", "third-party CI action") |
| special = [t for _, t in triggers if t in ("pull_request_target", "workflow_run", "issue_comment") or t.startswith("schedule")] |
| names = list(dict.fromkeys(t for _, t in triggers)) |
| if any(t.startswith("schedule ") for t in names): |
| names = [t for t in names if t != "schedule"] |
| special = [t for t in special if t != "schedule"] |
| if names and (u.focus is None or u.focus): |
| first = min((ln for ln, _ in triggers if u.in_focus(ln)), default=u.first_focus()) |
| e.add("install_or_startup_hook", first, "CI triggers: " + ", ".join(names), |
| "runs automatically" + ("; " + ", ".join(dict.fromkeys(special)) if special else ""), force=u.focus is None or bool(u.focus)) |
| for key, ln, body in _yaml_blocks(e.lines, _yaml_keys(u.path)): |
| if "${{" in body: |
| body = re.sub(r"\$\{\{\s*([^}]*?)\s*\}\}", lambda m: "CI_EXPR_" + re.sub(r"\W", "_", m.group(1))[:40], body) |
| e.sh(body, ln, ctx=f"CI {key}") |
| e.common() |
| return e.out |
|
|
|
|
| def precommit_yaml(u, ctx=""): |
| e = Emitter(u, ctx) |
| for i, l in enumerate(e.lines, 1): |
| m = re.match(r"^\s*-?\s*repo\s*:\s*(\S+)", l) |
| if m: |
| e.add("install_or_startup_hook", i, f"pre-commit repo {m.group(1)}", "git hook framework") |
| if m.group(1).startswith("http"): |
| e.add("network_egress", i, m.group(1), "pre-commit fetches hook repo") |
| for key, ln, body in _yaml_blocks(e.lines, r"(entry)"): |
| if key in ("entry",): |
| e.sh(body, ln, ctx="pre-commit hook entry") |
| e.common() |
| return e.out |
|
|
|
|
| def generic_yaml_json_toml(u, ctx=""): |
| e = Emitter(u, ctx) |
| for i, l in enumerate(e.lines, 1): |
| s = l.strip() |
| if not u.in_focus(i) or not s: |
| continue |
| for rx, label in P.REGISTRY[1:3]: |
| m = rx.search(s) |
| if m: |
| e.add("security_setting_change", i, f"{label}: {m.group(1)}", "registry setting") |
| if re.search(r"""['"]?runOn['"]?\s*:\s*['"]folderOpen""", s): |
| e.add("install_or_startup_hook", i, "task runOn folderOpen", "runs when the folder is opened in the editor") |
| m = re.search(r"""['"]command['"]\s*:\s*['"]([^'"]+)""", s) |
| if m and "tasks.json" in u.path: |
| e.sh(m.group(1), i, ctx="editor task") |
| e.common() |
| return e.out |
|
|
|
|
| def dockerfile(u, ctx=""): |
| e = Emitter(u, ctx or "container build") |
| buf, start = "", 0 |
| for i, l in enumerate(e.lines + [""], 1): |
| if not buf: |
| start = i |
| buf += (" " if buf else "") + l.rstrip().rstrip("\\") |
| if l.rstrip().endswith("\\"): |
| continue |
| s = buf.strip() |
| buf = "" |
| m = re.match(r"(?i)(RUN|CMD|ENTRYPOINT)\s+(.*)", s) |
| if m: |
| cmd = m.group(2) |
| if cmd.startswith("["): |
| cmd = " ".join(re.findall(r'"((?:[^"\\]|\\.)*)"', cmd)) |
| e.out.extend(shell.analyse(cmd, u.path, base_line=start, in_focus=lambda a, b, s=start, i=i: u.in_focus(s, i), |
| context=f"Dockerfile {m.group(1).upper()}")) |
| m = re.match(r"(?i)(ADD)\s+(?:--\S+\s+)*(https?://\S+)", s) |
| if m and u.in_focus(start, i): |
| e.add("network_egress", start, m.group(2), "Dockerfile ADD from URL", ctx="container build", force=True) |
| m = re.match(r"(?i)ENV\s+(.*)", s) |
| if m and u.in_focus(start, i): |
| for rx, label in P.TLS_OFF: |
| if rx.search(m.group(1)): |
| e.add("security_setting_change", start, label, "TLS verification off", force=True) |
| return e.out |
|
|
|
|
| def makefile(u, ctx=""): |
| e = Emitter(u, ctx) |
| target = "" |
| for i, l in enumerate(e.lines, 1): |
| m = re.match(r"^([\w./%-]+)\s*:(?!=)", l) |
| if m: |
| target = m.group(1) |
| continue |
| if l.startswith("\t") and u.in_focus(i): |
| cmd = l.strip().lstrip("@-+") |
| e.out.extend(shell.analyse(cmd, u.path, base_line=i, in_focus=u.in_focus, context=f"make {target}")) |
| return e.out |
|
|
|
|
| def pth(u, ctx=""): |
| e = Emitter(u, "interpreter startup (.pth)") |
| for i, l in enumerate(e.lines, 1): |
| if re.match(r"\s*import[\s\t]", l): |
| e.add("install_or_startup_hook", i, ".pth import line", "executes at every interpreter start") |
| e.add("persistence", i, "python startup (sitecustomize/.pth)", ".pth executable line") |
| e.add("dynamic_import_or_eval", i, clip(l.strip(), 100), ".pth code line") |
| for rx, label in P.TLS_OFF: |
| if rx.search(l): |
| e.add("security_setting_change", i, label, "TLS verification off") |
| if re.search(r"exec|eval|b64decode|__import__", l): |
| det = "decode_then_exec" if re.search(r"b64decode|decode\(|unhexlify", l) else None |
| if det: |
| e.add("decode_then_exec", i, clip(l.strip(), 100), "decoded data executed at startup") |
| e.common() |
| return e.out |
|
|
|
|
| def binary(u, ctx=""): |
| out = [] |
| head = u.raw_head |
| kind = next((label for sig, label in MAGIC if head.startswith(sig)), "") |
| if not kind and len(head) > 262 and head[257:262] == b"ustar": |
| kind = "tar archive" |
| low = u.path.lower() |
| size = u.size |
| note = u.binary_note |
| if u.lang == "other" and note.startswith("symlink"): |
| out.append(Behaviour("file_write", u.path, 1, note, "", "symlink in package", ctx, "manifest")) |
| return out |
| is_archive = kind in ("zip archive", "gzip", "7z archive", "rar archive", "bzip2", "xz", "tar archive") or \ |
| low.endswith(tuple(e for e in (".zip", ".tgz", ".tar", ".gz", ".7z", ".rar", ".whl", ".jar", ".egg", ".xz", |
| ".bz2", ".crx", ".vsix", ".nupkg", ".zst"))) |
| if low.endswith(MEDIA_EXT) or kind in MEDIA: |
| return out |
| sz = f"{size} bytes" if size else (note or "size unknown") |
| if is_archive: |
| out.append(Behaviour("archive_or_bundle", u.path, 1, f"{kind or 'archive'} ({sz})", note, "archive file", ctx, "manifest")) |
| else: |
| out.append(Behaviour("binary_files", u.path, 1, f"{kind or 'binary data'} ({sz})", note, "", ctx, "manifest")) |
| return out |
|
|
|
|
| def dispatch(u, ctx=""): |
| name = u.path.rsplit("/", 1)[-1] |
| if u.lang == "binary" or u.binary_note.startswith("symlink"): |
| return binary(u, ctx) |
| if name == "package.json": |
| return package_json(u, ctx) |
| if u.lang == "config": |
| return npmrc_like(u, ctx) |
| if u.lang == "requirements": |
| return requirements(u, ctx) |
| if u.lang == "pth": |
| return pth(u, ctx) |
| if u.lang == "dockerfile": |
| return dockerfile(u, ctx) |
| if u.lang == "makefile": |
| return makefile(u, ctx) |
| if u.lang == "yaml" and CI_PATH.search(u.path): |
| return ci_yaml(u, ctx) |
| if u.lang == "yaml" and name in (".pre-commit-config.yaml", ".pre-commit-hooks.yaml"): |
| return precommit_yaml(u, ctx) |
| if u.lang in ("toml", "ini"): |
| return toml_ini(u, ctx) |
| if u.lang in ("yaml", "json"): |
| return generic_yaml_json_toml(u, ctx) |
| return None |
|
|
