Blame for extractor/behaviour/config.py

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