Blame for extractor/behaviour/pyjs.py

"""Tree-sitter extractor for Python and JavaScript/TypeScript.

Tree-sitter is error tolerant, so diff hunks and snippets that are not complete programs still yield call sites.
Per file it builds: import aliases, simple constant/expression bindings, constructor types, function definitions and
call sites. Rules map normalised callees to behaviours. A small flow-insensitive label propagation (names, function
returns, one level of call-site arguments) connects sources (network content, decoders, env, credential files,
archives) to sinks (exec/eval, spawns, egress bodies). Nothing is executed.
"""
from __future__ import annotations

import ast
import re
import warnings

from .model import Behaviour, clip
from . import patterns as P
from . import shell

try:
    import tree_sitter_language_pack as _tslp
except Exception:  # pragma: no cover - fallback handled by caller
    _tslp = None

_PARSERS = {}


def available():
    return _tslp is not None


def parser(lang):
    key = {"javascript": "javascript", "typescript": "typescript", "tsx": "tsx", "python": "python"}[lang]
    if key not in _PARSERS:
        _PARSERS[key] = _tslp.get_parser(key)
    return _PARSERS[key]


HTTP_VERBS = {"get", "post", "put", "patch", "delete", "head", "options", "request", "stream", "send"}

# ---------------------------------------------------------------- rule tables (normalised callee -> kind, info)
PY_SPAWN = {f"subprocess.{f}" for f in ("run", "call", "check_call", "check_output", "Popen", "getoutput",
                                         "getstatusoutput")} | \
    {f"os.{f}" for f in ("system", "popen", "execv", "execve", "execl", "execle", "execlp", "execlpe", "execvp",
                         "execvpe", "spawnl", "spawnle", "spawnlp", "spawnv", "spawnve", "spawnvp", "posix_spawn",
                         "posix_spawnp", "startfile")} | \
    {"pty.spawn", "asyncio.create_subprocess_exec", "asyncio.create_subprocess_shell", "commands.getoutput",
     "sh.Command", "pexpect.spawn", "pexpect.run"}
PY_NET_PREFIX = ("requests.", "httpx.", "aiohttp.", "urllib3.", "treq.", "grequests.")
PY_NET = {"urllib.request.urlopen", "urllib.request.urlretrieve", "urllib.urlopen",
          "urllib.urlretrieve", "urllib2.urlopen", "http.client.HTTPConnection",
          "http.client.HTTPSConnection", "httplib.HTTPConnection", "httplib.HTTPSConnection",
          "socket.create_connection", "socket.socket.connect", "socket.socket.connect_ex", "socket.socket.sendto",
          "ftplib.FTP", "ftplib.FTP_TLS", "smtplib.SMTP", "smtplib.SMTP_SSL", "telnetlib.Telnet",
          "paramiko.SSHClient.connect", "websocket.create_connection", "websockets.connect",
          "xmlrpc.client.ServerProxy", "socket.gethostbyname", "socket.getaddrinfo", "dns.resolver.resolve",
          "dns.resolver.query", "pycurl.Curl", "webbrowser.open", "aiohttp.ClientSession", "urllib.request.build_opener"}
PY_NET_SKIP = {"requests.exceptions", "requests.adapters", "httpx.Timeout", "requests.Response", "httpx.Response",
               "requests.codes", "httpx.HTTPError", "requests.HTTPError", "aiohttp.ClientTimeout",
               "requests.RequestException", "httpx.HTTPStatusError", "requests.utils", "requests.structures",
               "requests.auth", "httpx.Limits", "urllib3.util", "urllib3.exceptions", "urllib3.disable_warnings",
               "requests.Session", "httpx.Client", "httpx.AsyncClient", "urllib3.PoolManager", "requests.packages",
               "httpx.URL", "httpx.Headers", "requests.models", "requests.compat", "requests.status_codes",
               "aiohttp.web", "aiohttp.TCPConnector", "httpx.MockTransport", "httpx.HTTPTransport", "aiohttp.ClientError",
               "httpx.Request", "requests.Request", "requests.PreparedRequest", "httpx.BasicAuth"}
PY_DECODE = {f"base64.{f}" for f in ("b64decode", "b32decode", "b16decode", "a85decode", "b85decode",
                                      "urlsafe_b64decode", "decodebytes", "decodestring", "standard_b64decode")} | \
    {"binascii.unhexlify", "binascii.a2b_base64", "binascii.a2b_hex", "bytes.fromhex", "bytearray.fromhex",
     "codecs.decode", "zlib.decompress", "gzip.decompress", "bz2.decompress", "lzma.decompress", "marshal.loads"}
PY_EXEC = {"exec", "eval", "compile", "builtins.exec", "builtins.eval", "builtins.compile", "execfile"}
PY_DYN = {"__import__", "importlib.import_module", "importlib.util.spec_from_file_location",
          "importlib.util.module_from_spec", "runpy.run_path", "runpy.run_module", "imp.load_source",
          "imp.load_module", "importlib.machinery.SourceFileLoader", "types.FunctionType", "types.CodeType"}
PY_DESER = {"pickle.loads", "pickle.load", "cPickle.loads", "cPickle.load", "marshal.loads", "marshal.load",
            "dill.loads", "dill.load", "cloudpickle.loads", "jsonpickle.decode", "yaml.unsafe_load", "shelve.open",
            "pandas.read_pickle", "torch.load", "joblib.load"}
PY_NATIVE = {"ctypes.CDLL", "ctypes.cdll.LoadLibrary", "ctypes.WinDLL", "ctypes.windll.LoadLibrary", "ctypes.PyDLL",
             "cffi.FFI.dlopen"}
PY_DELETE = {"os.remove", "os.unlink", "os.rmdir", "os.removedirs", "shutil.rmtree", "send2trash.send2trash"}
PY_COPY = {"shutil.copy", "shutil.copy2", "shutil.copyfile", "shutil.copytree", "shutil.move", "os.rename",
           "os.replace", "os.renames"}
PY_ARCHIVE_NEW = {"tarfile.open", "tarfile.TarFile", "zipfile.ZipFile", "shutil.make_archive", "gzip.open",
                  "zipfile.PyZipFile", "py7zr.SevenZipFile"}

JS_SPAWN = {f"child_process.{f}" for f in ("exec", "execSync", "spawn", "spawnSync", "execFile", "execFileSync",
                                            "fork")} | \
    {"execa", "execa.execa", "execa.command", "execa.commandSync", "execa.execaSync", "execa.execaCommand",
     "execa.execaCommandSync", "execa.$", "shelljs.exec", "Bun.spawn", "Bun.spawnSync", "Deno.run", "Deno.Command",
     "cross-spawn", "cross-spawn.sync", "zx.$", "Bun.$"}
JS_BARE_SPAWN = {"execSync", "spawnSync", "execFileSync", "execFile"}
JS_NET = {"fetch", "axios", "got", "superagent", "ky", "undici.request", "undici.fetch", "http.request", "http.get",
          "https.request", "https.get", "http2.connect", "net.connect", "net.createConnection", "tls.connect",
          "dgram.createSocket", "WebSocket", "ws", "EventSource", "navigator.sendBeacon", "dns.lookup",
          "dns.resolve", "dns.resolve4", "dns.resolveTxt", "dns.promises.lookup", "request", "needle",
          "XMLHttpRequest", "net.Socket", "jQuery.ajax", "$.ajax", "$.get", "$.post", "$.getJSON", "Deno.connect",
          "io", "socket.io-client", "socket.io-client.io", "phin", "bent", "node-fetch", "urllib.request", "ofetch",
          "$fetch", "wretch"}
JS_NET_PREFIX = ("axios.", "got.", "superagent.", "ky.", "needle.", "request.")
JS_WRITE = {f"fs.{f}": 0 for f in ("writeFile", "writeFileSync", "appendFile", "appendFileSync", "createWriteStream",
                                   "outputFile", "outputFileSync", "writeJson", "writeJsonSync", "outputJson",
                                   "outputJsonSync", "chmod", "chmodSync", "truncate", "truncateSync", "mkdtemp")}
JS_WRITE.update({f"fs.{f}": 1 for f in ("copyFile", "copyFileSync", "cp", "cpSync", "rename", "renameSync", "symlink",
                                        "symlinkSync", "link", "linkSync", "copy", "copySync", "move", "moveSync",
                                        "ensureSymlink")})
JS_WRITE.update({"Deno.writeTextFile": 0, "Deno.writeFile": 0, "Bun.write": 0})
JS_BARE_WRITE = {"writeFileSync", "appendFileSync", "writeFile", "appendFile", "createWriteStream", "copyFileSync",
                 "renameSync", "symlinkSync", "chmodSync"}
JS_DELETE = {f"fs.{f}" for f in ("unlink", "unlinkSync", "rm", "rmSync", "rmdir", "rmdirSync", "remove",
                                  "removeSync", "emptyDir", "emptyDirSync")} | \
    {"rimraf", "rimraf.sync", "rimraf.rimraf", "rimraf.rimrafSync", "del", "del.deleteAsync", "del.deleteSync",
     "Deno.remove", "trash"}
JS_BARE_DELETE = {"unlinkSync", "rmSync", "rmdirSync"}
JS_EXEC = {"eval", "Function", "vm.runInNewContext", "vm.runInThisContext", "vm.runInContext", "vm.Script",
           "vm.compileFunction", "vm.SourceTextModule", "globalThis.eval", "window.eval", "global.eval"}
JS_DECODE = {"atob", "String.fromCharCode", "zlib.inflateSync", "zlib.gunzipSync", "zlib.inflateRawSync",
             "zlib.brotliDecompressSync", "zlib.unzipSync", "zlib.inflate", "zlib.gunzip", "Buffer.from.base64"}
JS_ARCHIVE_NEW = {"archiver", "tar.c", "tar.create", "adm-zip", "jszip", "JSZip", "yazl.ZipFile", "zlib.gzipSync"}
JS_MOD_NORM = {"node-fetch": "fetch", "cross-fetch": "fetch", "isomorphic-fetch": "fetch", "fs-extra": "fs",
               "fs/promises": "fs", "graceful-fs": "fs", "child-process-promise": "child_process", "jquery": "jQuery",
               "dns/promises": "dns", "shelljs": "shelljs", "execa": "execa", "zx": "zx", "rimraf": "rimraf"}


def _t(n):
    return n.text.decode("utf-8", errors="replace") if n is not None else ""


def _line(n):
    return n.start_point[0] + 1


def _end(n):
    return n.end_point[0] + 1


class Call:
    __slots__ = ("node", "callee", "args", "kwargs", "recv", "func", "top", "line", "end", "kind_labels", "new")

    def __init__(self, node, callee, args, kwargs, recv, func, top, new=False):
        self.node, self.callee, self.args, self.kwargs, self.recv = node, callee, args, kwargs, recv
        self.func, self.top, self.new = func, top, new
        self.line, self.end = _line(node), _end(node)
        self.kind_labels = set()


class FileAnalysis:
    def __init__(self, unit, lang, context=""):
        self.u = unit
        self.lang = lang
        self.py = lang == "python"
        self.context = context
        self.text = unit.text
        self.src = unit.text.encode("utf-8", errors="replace")
        self.lines = unit.text.split("\n")
        self.alias, self.types, self.consts, self.exprs = {}, {}, {}, {}
        self.funcs = {}          # name -> (node, [params])
        self.calls = []
        self.name_labels, self.func_labels = {}, {}
        self.callgraph = {}      # func -> set(local funcs called)
        self.top_called = set()  # local funcs called at module level
        self.out = []
        self.consumed = set()    # string literal byte ranges already analysed as commands
        self.withs, self.fors, self.returns = [], [], []
        self.idents_cache = {}

    # ------------------------------------------------------------------ helpers
    def lit(self, n):
        if n is None:
            return None
        t = n.type
        if self.py:
            if t == "string":
                if any(c.type == "interpolation" for c in n.children):
                    return None
                try:
                    with warnings.catch_warnings():
                        warnings.simplefilter("ignore")
                        v = ast.literal_eval(_t(n))
                    return v.decode("latin-1") if isinstance(v, bytes) else v
                except Exception:
                    s = _t(n)
                    m = re.match(r"""^[rRbBuU]{0,2}('''|\"\"\"|'|")(.*)\1$""", s, re.S)
                    return m.group(2) if m else None
            if t == "concatenated_string":
                parts = [self.lit(c) for c in n.named_children]
                return "".join(parts) if all(p is not None for p in parts) else None
            if t == "binary_operator" and _t(n.child_by_field_name("operator")) == "+":
                a, b = self.lit(n.child_by_field_name("left")), self.lit(n.child_by_field_name("right"))
                return a + b if a is not None and b is not None else None
            if t == "identifier" and _t(n) in self.consts:
                return self.consts[_t(n)]
            if t == "parenthesized_expression" and n.named_children:
                return self.lit(n.named_children[0])
            return None
        if t == "string":
            s = _t(n)
            body = s[1:-1]
            try:
                if "\\" not in body:
                    return body
                with warnings.catch_warnings():
                    warnings.simplefilter("ignore")
                    return ast.literal_eval('"' + body.replace('"', '\\"') + '"')
            except Exception:
                return body
        if t == "template_string":
            if any(c.type == "template_substitution" for c in n.children):
                return None
            return _t(n)[1:-1]
        if t == "binary_expression" and _t(n.child_by_field_name("operator")) == "+":
            a, b = self.lit(n.child_by_field_name("left")), self.lit(n.child_by_field_name("right"))
            return a + b if a is not None and b is not None else None
        if t == "identifier" and _t(n) in self.consts:
            return self.consts[_t(n)]
        if t == "parenthesized_expression" and n.named_children:
            return self.lit(n.named_children[0])
        return None

    def show(self, n, depth=3):
        """Display an operand: literal value, else the expression with single-definition names substituted."""
        if n is None:
            return "dynamic"
        v = self.lit(n)
        if v is not None:
            return v
        s = _t(n)
        if depth > 0:
            idents = [c for c in self._idents(n) if _t(c) not in self.alias and (
                _t(c) in self.exprs and len(self.exprs[_t(c)]) == 1 or _t(c) in dict(self.fors) and _t(c) not in self.exprs)]
            for c in sorted(idents, key=lambda c: -c.start_byte):
                if _t(c) in self.exprs:
                    rhs = self.exprs[_t(c)][0]
                    if rhs.start_byte <= n.start_byte <= rhs.end_byte:
                        continue
                    rep = self.show(rhs, depth - 1)
                    rep = repr(rep) if self.lit(rhs) is not None else rep
                else:
                    rep = f"<each of {self.show(dict(self.fors)[_t(c)], depth - 1)}>"
                a, b = c.start_byte - n.start_byte, c.end_byte - n.start_byte
                bs = s.encode()
                s = (bs[:a] + rep.encode() + bs[b:]).decode(errors="replace")
            if n.type in ("call", "call_expression"):
                fn = n.child_by_field_name("function")
                if fn is not None and fn.type == "identifier" and _t(fn) in self.funcs:
                    rets = [r for r, f in self.returns if f == _t(fn) and r.type == "return_statement" and r.named_children]
                    if 0 < len(rets) <= 2:
                        s += " → returns " + " | ".join(self.show(r.named_children[0], depth - 1) for r in rets)
        return clip(s, 140)

    def _idents(self, n):
        key = (n.start_byte, n.end_byte)
        if key in self.idents_cache:
            return self.idents_cache[key]
        out, stack = [], [n]
        while stack:
            x = stack.pop()
            if x.type in ("identifier", "shorthand_property_identifier"):
                p = x.parent
                if p is not None and (p.type in ("attribute", "member_expression") and
                                      p.child_by_field_name("object") is not None and
                                      p.child_by_field_name("object") != x and
                                      (p.child_by_field_name("object").start_byte != x.start_byte)):
                    continue
                if p is not None and p.type in ("keyword_argument",) and p.child_by_field_name("name") == x:
                    continue
                out.append(x)
            stack.extend(x.children)
        self.idents_cache[key] = out
        return out

    def ev(self, line):
        return self.lines[line - 1].strip() if 0 < line <= len(self.lines) else ""

    def add(self, kind, line, operand, detail="", source="ast", context=None, end=None):
        if not self.u.in_focus(line, end or line):
            return
        ln = line
        if self.u.focus is not None and end and end > line:
            ln = next((x for x in range(line, end + 1) if x in self.u.focus), line)
        self.out.append(Behaviour(kind, self.u.path, ln, clip(str(operand), 140) or "dynamic", self.ev(ln), detail,
                                  self.context if context is None else context, source))

    # ------------------------------------------------------------------ pass 1: bindings
    def collect(self, root):
        stack = [(root, None, True)]
        while stack:
            n, func, top = stack.pop()
            t = n.type
            nfunc, ntop = func, top
            if t in ("function_definition", "function_declaration", "method_definition", "generator_function_declaration"):
                name = _t(n.child_by_field_name("name"))
                params = self._params(n)
                if name:
                    self.funcs.setdefault(name, (n, params))
                nfunc, ntop = name or "<fn>", False
            elif t in ("class_definition", "class_declaration"):
                ntop = False
            elif t in ("arrow_function", "function_expression", "function", "lambda"):
                p = n.parent
                name = _t(p.child_by_field_name("name")) if p is not None and p.type == "variable_declarator" else ""
                if name:
                    self.funcs.setdefault(name, (n, self._params(n)))
                nfunc, ntop = name or func or "<fn>", False
            elif t == "if_statement" and self.py and re.search(r"__name__\s*==\s*['\"]__main__['\"]",
                                                               _t(n.child_by_field_name("condition"))):
                ntop = False
            if self.py:
                self._py_binding(n)
            else:
                self._js_binding(n)
            if t in ("call", "call_expression", "new_expression"):
                c = self._call(n, func, top)
                if c is not None:
                    self.calls.append(c)
            if t in ("return_statement", "yield"):
                self.returns.append((n, func))
            for ch in reversed(n.children):
                stack.append((ch, nfunc, ntop))

    def _params(self, n):
        p = n.child_by_field_name("parameters")
        if p is None:
            return []
        out = []
        for c in p.named_children:
            if c.type == "identifier":
                out.append(_t(c))
            else:
                idn = c.child_by_field_name("name") or c.child_by_field_name("pattern") or \
                    next((x for x in c.named_children if x.type == "identifier"), None)
                out.append(_t(idn) if idn is not None else "")
        if out and out[0] in ("self", "cls"):
            out = out[1:]
        return out

    def _bind(self, name, rhs):
        if not name or rhs is None:
            return
        self.exprs.setdefault(name, []).append(rhs)
        v = self.lit(rhs)
        if v is not None and isinstance(v, str):
            self.consts.setdefault(name, v)
        if rhs.type in ("call", "call_expression", "new_expression", "await_expression", "await"):
            inner = rhs
            if rhs.type in ("await_expression", "await") and rhs.named_children:
                inner = rhs.named_children[0]
            fn = inner.child_by_field_name("function") or inner.child_by_field_name("constructor")
            cal = self._callee(fn) if fn is not None else None
            if cal:
                self.types[name] = cal

    def _py_binding(self, n):
        t = n.type
        if t == "import_statement":
            for c in n.named_children:
                if c.type == "dotted_name":
                    first = _t(c).split(".")[0]
                    self.alias[first] = first
                elif c.type == "aliased_import":
                    self.alias[_t(c.child_by_field_name("alias"))] = _t(c.child_by_field_name("name"))
        elif t == "import_from_statement":
            mod = _t(n.child_by_field_name("module_name"))
            seen_mod = False
            for c in n.named_children:
                if c == n.child_by_field_name("module_name") or not seen_mod and c.type in ("dotted_name", "relative_import") and _t(c) == mod:
                    seen_mod = True
                    continue
                if c.type == "dotted_name":
                    self.alias[_t(c)] = f"{mod}.{_t(c)}"
                elif c.type == "aliased_import":
                    self.alias[_t(c.child_by_field_name("alias"))] = f"{mod}.{_t(c.child_by_field_name('name'))}"
        elif t == "assignment":
            left, right = n.child_by_field_name("left"), n.child_by_field_name("right")
            if left is not None and left.type == "identifier":
                self._bind(_t(left), right)
            elif left is not None and left.type == "attribute" and _t(left).startswith("self."):
                self._bind(_t(left), right)
        elif t == "with_item":
            v = n.child_by_field_name("value")
            if v is not None and v.type == "as_pattern":
                target = v.child_by_field_name("alias")
                src = v.named_children[0] if v.named_children else None
                if target is not None and src is not None:
                    name = _t(target).strip()
                    self.withs.append((name, src))
                    self._bind(name, src)
        elif t == "for_statement":
            left, right = n.child_by_field_name("left"), n.child_by_field_name("right")
            if left is not None and right is not None:
                for idn in ([left] if left.type == "identifier" else [x for x in left.named_children if x.type == "identifier"]):
                    self.fors.append((_t(idn), right))

    def _js_binding(self, n):
        t = n.type
        if t == "import_statement":
            srcn = n.child_by_field_name("source")
            mod = self._jsmod(self.lit(srcn) or "")
            for c in n.named_children:
                if c.type != "import_clause":
                    continue
                for d in c.named_children:
                    if d.type == "identifier":
                        self.alias[_t(d)] = mod
                    elif d.type == "namespace_import":
                        idn = next((x for x in d.named_children if x.type == "identifier"), None)
                        if idn is not None:
                            self.alias[_t(idn)] = mod
                    elif d.type == "named_imports":
                        for s in d.named_children:
                            if s.type == "import_specifier":
                                nm, al = s.child_by_field_name("name"), s.child_by_field_name("alias")
                                self.alias[_t(al or nm)] = f"{mod}.{_t(nm)}" if _t(nm) != "default" else mod
        elif t in ("variable_declarator", "assignment_expression"):
            name = n.child_by_field_name("name") if t == "variable_declarator" else n.child_by_field_name("left")
            val = n.child_by_field_name("value") if t == "variable_declarator" else n.child_by_field_name("right")
            if name is None or val is None:
                return
            mod = self._require_mod(val)
            if name.type == "identifier":
                if mod is not None:
                    self.alias[_t(name)] = mod
                self._bind(_t(name), val)
            elif name.type == "member_expression" and _t(name).startswith("this."):
                self._bind(_t(name), val)
            elif name.type == "object_pattern" and mod is not None:
                for c in name.named_children:
                    if c.type == "shorthand_property_identifier_pattern":
                        self.alias[_t(c)] = f"{mod}.{_t(c)}"
                    elif c.type == "pair_pattern":
                        k, v = c.child_by_field_name("key"), c.child_by_field_name("value")
                        if v is not None and v.type == "identifier":
                            self.alias[_t(v)] = f"{mod}.{_t(k)}"
        elif t in ("for_in_statement",):
            left, right = n.child_by_field_name("left"), n.child_by_field_name("right")
            if left is not None and right is not None:
                idn = left if left.type == "identifier" else next((x for x in left.named_children if x.type == "identifier"), None)
                if idn is not None:
                    self.fors.append((_t(idn), right))

    def _jsmod(self, m):
        m = m.removeprefix("node:")
        if m in JS_MOD_NORM:
            return JS_MOD_NORM[m]
        return m

    def _require_mod(self, val):
        v = val
        if v.type in ("await_expression",) and v.named_children:
            v = v.named_children[0]
        suffix = ""
        while v.type == "member_expression":
            suffix = "." + _t(v.child_by_field_name("property")) + suffix
            v = v.child_by_field_name("object")
        if v.type == "call_expression":
            fn = v.child_by_field_name("function")
            if fn is not None and _t(fn) in ("require", "import"):
                args = v.child_by_field_name("arguments")
                a0 = args.named_children[0] if args is not None and args.named_children else None
                s = self.lit(a0)
                if s:
                    return self._jsmod(s) + suffix
        return None

    def _dotted(self, n):
        if n.type in ("identifier", "property_identifier", "this"):
            return _t(n)
        if n.type in ("attribute", "member_expression"):
            o = n.child_by_field_name("object")
            a = n.child_by_field_name("attribute") or n.child_by_field_name("property")
            if o is None or a is None:
                return None
            od = self._dotted(o)
            return f"{od}.{_t(a)}" if od is not None else None
        return None

    def _callee(self, fn):
        if fn is None:
            return None
        if fn.type in ("identifier",):
            n = _t(fn)
            return self.alias.get(n, n)
        if fn.type == "import":
            return "import"
        if fn.type in ("attribute", "member_expression"):
            d = self._dotted(fn)
            attr = _t(fn.child_by_field_name("attribute") or fn.child_by_field_name("property"))
            if d is not None:
                first, _, rest = d.partition(".")
                if first in self.alias:
                    d = self.alias[first] + ("." + rest if rest else "")
                elif first in self.types and rest:
                    d = self.types[first] + "()." + rest
                elif first in ("self", "this") and rest:
                    head, _, tail = rest.partition(".")
                    key = f"{first}.{head}"
                    if key in self.types and tail:
                        d = self.types[key] + "()." + tail
                if not self.py:
                    d = re.sub(r"^fs\.promises\.", "fs.", d)
                return d
            o = fn.child_by_field_name("object")
            if o is not None and o.type in ("call", "call_expression", "new_expression"):
                inner = o.child_by_field_name("function") or o.child_by_field_name("constructor")
                m = self._require_mod(o) if not self.py else None
                if m:
                    return f"{m}.{attr}"
                ic = self._callee(inner)
                if ic:
                    return f"{ic}().{attr}"
            if o is not None and o.type in ("parenthesized_expression", "await_expression", "await"):
                return f"?.{attr}"
            return f"?.{attr}"
        return None

    def _call(self, n, func, top):
        new = n.type == "new_expression"
        fn = n.child_by_field_name("constructor" if new else "function")
        if fn is None:
            return None
        if fn.type == "new_expression":
            return None              # `new F(x)()`: the new_expression is its own call site
        callee = self._callee(fn)
        if callee is None:
            return None
        recv = None
        if fn.type in ("attribute", "member_expression"):
            recv = fn.child_by_field_name("object")
        argn = n.child_by_field_name("arguments")
        args, kwargs = [], {}
        if argn is not None:
            if argn.type == "template_string":
                args = [argn]
            else:
                for c in argn.named_children:
                    if c.type == "keyword_argument":
                        kwargs[_t(c.child_by_field_name("name"))] = c.child_by_field_name("value")
                    elif c.type in ("comment",):
                        continue
                    else:
                        args.append(c)
        return Call(n, callee, args, kwargs, recv, func, top, new)

    # ------------------------------------------------------------------ labels
    def call_labels(self, c):
        k, labs = c.callee, set()
        last = k.rsplit(".", 1)[-1]
        if self._is_net(c):
            labs.add("network")
        if self._is_decode(c):
            labs.add("decode")
        if self._archive_new(c):
            labs.add("archive")
        if self.py and (k in ("open", "io.open", "codecs.open") and not self._write_mode(c)) or \
                last in ("read_text", "read_bytes", "readFileSync", "readFile") or k in ("fs.readFileSync", "fs.readFile"):
            labs.add("file")
        if k in self.funcs or last in self.funcs and k.startswith(("?.", "self.", "this.")):
            labs |= self.func_labels.get(k if k in self.funcs else last, set())
        return labs

    def node_labels(self, n):
        labs = set()
        s = _t(n)
        if re.search(r"\bos\.environ\b|\bgetenv\s*\(|\bprocess\.env\b|\benviron\.(items|get)\b|Deno\.env|import\.meta\.env", s):
            labs.add("env")
        if P.secret_hits(s):
            labs.add("credential")
        a, b = n.start_byte, n.end_byte
        for c in self.calls:
            if a <= c.node.start_byte and c.node.end_byte <= b:
                labs |= c.kind_labels
        for idn in self._idents(n):
            labs |= self.name_labels.get(_t(idn), set())
        return labs

    def propagate(self):
        for c in self.calls:
            c.kind_labels = self.call_labels(c)
        for _ in range(3):
            changed = False
            for name, rhss in list(self.exprs.items()):
                labs = set()
                for r in rhss:
                    labs |= self.node_labels(r)
                if labs - self.name_labels.get(name, set()):
                    self.name_labels.setdefault(name, set()).update(labs)
                    changed = True
            for name, src in self.fors:
                labs = self.node_labels(src)
                if labs - self.name_labels.get(name, set()):
                    self.name_labels.setdefault(name, set()).update(labs)
                    changed = True
            for c in self.calls:
                if "archive" in c.kind_labels and self._archive_new(c):
                    for kw in ("fileobj",):
                        v = c.kwargs.get(kw)
                        if v is not None and v.type == "identifier" and "archive" not in self.name_labels.get(_t(v), set()):
                            self.name_labels.setdefault(_t(v), set()).add("archive")
                            changed = True
            fl = {}
            for rn, func in self.returns:
                if func and func in self.funcs:
                    fl.setdefault(func, set()).update(self.node_labels(rn))
            for f, (fnode, _) in self.funcs.items():
                body = _t(fnode)
                if P.secret_hits(body) and re.search(r"read_text|read_bytes|open\(|readFile", body):
                    fl.setdefault(f, set()).add("credential")
            for f, labs in fl.items():
                if labs - self.func_labels.get(f, set()):
                    self.func_labels.setdefault(f, set()).update(labs)
                    changed = True
            for c in self.calls:
                c.kind_labels = self.call_labels(c)
            if not changed:
                break
        # call graph for import-time reachability
        for c in self.calls:
            name = c.callee if c.callee in self.funcs else None
            if name is None:
                continue
            if c.top:
                self.top_called.add(name)
            elif c.func:
                self.callgraph.setdefault(c.func, set()).add(name)

    def reachable_at_import(self):
        seen, todo = set(), list(self.top_called)
        while todo:
            f = todo.pop()
            if f in seen:
                continue
            seen.add(f)
            todo.extend(self.callgraph.get(f, ()))
        return seen

    # ------------------------------------------------------------------ classifiers
    def _is_net(self, c):
        k = c.callee
        if self.py:
            if k in PY_NET:
                return True
            if k.startswith(PY_NET_PREFIX) and "()" not in k and not any(k == s or k.startswith(s + ".") for s in PY_NET_SKIP):
                last = k.rsplit(".", 1)[-1]
                return last in HTTP_VERBS or last in ("urlopen", "request_encode_body", "request") or \
                    k.count(".") == 1 and last in HTTP_VERBS
            if re.match(r"^(requests\.Session|httpx\.(Async)?Client|aiohttp\.ClientSession|urllib3\.PoolManager)(\(\))?\.(get|post|put|patch|delete|head|options|request|stream|send|urlopen)$", k):
                return True
            if k in ("urlopen", "urlretrieve"):
                return True
            return False
        if k in JS_NET:
            return k not in ("request",) or "request" in self.alias
        if k.startswith(JS_NET_PREFIX) and "()" not in k:
            return k.rsplit(".", 1)[-1] in HTTP_VERBS
        if re.match(r"^(axios|got|ky|superagent)(\(\))?\.\w+$", k) and k.rsplit(".", 1)[-1] in HTTP_VERBS:
            return True
        return bool(re.match(r"^(net\.Socket|WebSocket|dgram\.createSocket)(\(\))?\.(connect|send)$", k))

    def _is_decode(self, c):
        k = c.callee
        if self.py:
            if k in PY_DECODE or k in ("b64decode", "unhexlify", "a2b_base64", "urlsafe_b64decode", "decompress"):
                return True
            if k.endswith(".decode") and c.args and (self.lit(c.args[0]) or "").lower() in ("base64", "hex", "rot13", "rot_13", "zlib", "bz2"):
                return True
            if k == "?.join" and c.args and re.search(r"\bchr\(", _t(c.args[0])):
                return True
            return k == "codecs.decode"
        if k in JS_DECODE:
            return True
        if k in ("Buffer.from", "buffer.Buffer.from") and len(c.args) > 1 and (self.lit(c.args[1]) or "") in ("base64", "hex", "base64url"):
            return True
        return False

    def _archive_new(self, c):
        k = c.callee
        if self.py:
            if k in ("tarfile.open", "tarfile.TarFile", "zipfile.ZipFile", "gzip.open", "zipfile.PyZipFile"):
                mode = c.kwargs.get("mode")
                mv = self.lit(mode) if mode is not None else (self.lit(c.args[1]) if len(c.args) > 1 else None)
                return bool(mv) and mv[:1] in ("w", "x", "a")
            return k in ("shutil.make_archive", "py7zr.SevenZipFile")
        return k in JS_ARCHIVE_NEW

    def _write_mode(self, c):
        mode = c.kwargs.get("mode")
        if mode is None and len(c.args) > 1:
            mode = c.args[1]
        mv = self.lit(mode) if mode is not None else None
        return bool(mv) and any(ch in mv for ch in "wax+")

    # ------------------------------------------------------------------ pass 2: rules
    def apply(self):
        imp = self.reachable_at_import() if self.py else set()
        params = {f: p for f, (_, p) in self.funcs.items()}
        sites = {}
        for c in self.calls:
            nm = c.callee if c.callee in self.funcs else (c.callee.rsplit(".", 1)[-1] if c.callee.startswith(("?.", "self.", "this.")) else None)
            if nm in self.funcs:
                sites.setdefault(nm, []).append(c)
        for c in self.calls:
            ctx = self.context
            if self.py and (c.top or c.func in imp) and self.u.path.rsplit("/", 1)[-1] != "setup.py":
                via = "" if c.top else f" via {sorted(f for f in imp if f == c.func)[0]}()"
                ctx = "; ".join(x for x in (ctx, "runs at import" + via) if x)
            self._rules(c, ctx, params, sites)
        self._hooks()

    def _arg(self, c, i, *kw):
        for k in kw:
            if k in c.kwargs:
                return c.kwargs[k]
        return c.args[i] if len(c.args) > i else None

    def _operand_via_sites(self, c, node, params, sites):
        """If node is a parameter of the enclosing function, show what call sites pass."""
        if node is None or node.type != "identifier" or not c.func or c.func not in params:
            return None
        name = _t(node)
        ps = params[c.func]
        if name not in ps:
            return None
        i = ps.index(name)
        shown = []
        for s in sites.get(c.func, []):
            a = s.kwargs.get(name) or (s.args[i] if len(s.args) > i else None)
            if a is not None:
                shown.append(self.show(a))
        return ", ".join(dict.fromkeys(shown)) if shown else None

    def _site_labels(self, c, params, sites):
        if not c.func or c.func not in params:
            return set()
        labs = set()
        body_args = c.args + list(c.kwargs.values())
        used = {(_t(i)) for a in body_args for i in self._idents(a)}
        for i, p in enumerate(params[c.func]):
            if p in used:
                for s in sites.get(c.func, []):
                    a = s.kwargs.get(p) or (s.args[i] if len(s.args) > i else None)
                    if a is not None:
                        labs |= self.node_labels(a)
        return labs

    def _rules(self, c, ctx, params, sites):
        k = c.callee
        last = k.rsplit(".", 1)[-1]
        L, E = c.line, c.end
        add = lambda kind, operand, detail="": self.add(kind, L, operand, detail, context=ctx, end=E)
        arglabs = set()
        for a in c.args + list(c.kwargs.values()):
            if a is not None:
                arglabs |= self.node_labels(a)
        arglabs |= self._site_labels(c, params, sites)
        # ---- process spawn
        is_spawn = (k in PY_SPAWN or k in ("check_output", "check_call", "Popen", "getoutput")) if self.py else \
            (k in JS_SPAWN or k in JS_BARE_SPAWN or (k in ("exec", "spawn") and k not in self.funcs and k not in self.alias and
                                                     not self.alias))
        if is_spawn:
            a0 = self._arg(c, 0, "args", "cmd", "command")
            cmd = self._command_text(a0)
            det = []
            if self.py and "shell" in c.kwargs and _t(c.kwargs["shell"]) == "True" or not self.py and last in ("exec", "execSync"):
                det.append("via shell")
            if "network" in arglabs:
                self.add("download_then_exec", L, cmd, "network content reaches a process", context=ctx, end=E)
            if "decode" in arglabs:
                self.add("decode_then_exec", L, cmd, "decoded data reaches a process", context=ctx, end=E)
            add("process_spawn", cmd, "; ".join(det))
            if a0 is not None:
                full = self._command_text(a0, literal_only=True)
                if full:
                    self._mark_consumed(a0)
                    for b in shell.analyse(full, self.u.path, base_line=_line(a0), in_focus=lambda a, b: self.u.in_focus(L, E),
                                           context=ctx, embedded=f"via {k}", emit_spawn=False):
                        b.line = L if self.u.focus is None or L in self.u.focus else b.line
                        b.evidence = self.ev(b.line)
                        self.out.append(b)
            return
        # ---- network egress
        if self._is_net(c):
            urln = self._arg(c, 0, "url", "host", "address")
            reqcall = self._request_obj(urln)
            if reqcall is not None:
                urln = self._arg(reqcall, 0, "url")
                for kk, vv in reqcall.kwargs.items():
                    c.kwargs.setdefault(kk, vv)
                if len(reqcall.args) > 1:
                    c.kwargs.setdefault("data", reqcall.args[1])
                for a in [*reqcall.args, *reqcall.kwargs.values()]:
                    arglabs |= self.node_labels(a)
            if self.py and k.endswith("socket.connect") and c.args and c.args[0].type == "tuple":
                urln = c.args[0]
            op = self.show(urln) if urln is not None else "dynamic"
            if urln is not None and self.lit(urln) is None:
                via = self._operand_via_sites(c, urln, params, sites)
                if via:
                    op = f"{op} ← {via}"
                elif not re.search(r"://|^\(?['\"]?\d+\.\d+\.\d+\.\d+|^\(?['\"]?[a-z0-9-]+(\.[a-z0-9-]+)*\.[a-z]{2,}(:\d+)?(/|['\"]|$)", op, re.I):
                    op = f"dynamic ({clip(op, 60)})"
            det = []
            if last.upper() in ("POST", "PUT", "PATCH", "DELETE"):
                det.append(last.upper())
            body_kw = [x for x in ("data", "json", "files", "content", "body") if x in c.kwargs]
            if self.py and k.endswith("Request") and (len(c.args) > 1 or "data" in c.kwargs):
                body_kw.append("data")
            if not self.py:
                txt = _t(c.node)
                m = re.search(r"method\s*:\s*['\"](\w+)['\"]", txt)
                if m:
                    det.append(m.group(1).upper())
                if re.search(r"\bbody\s*:", txt) or last in ("post", "put", "patch") and len(c.args) > 1 or last == "sendBeacon":
                    body_kw.append("body")
            if body_kw:
                det.append("sends body")
            src = sorted(arglabs & {"env", "credential", "archive", "file"})
            if src:
                det.append("data from: " + ", ".join(src))
            if k.startswith("socket.get") or k.startswith("dns"):
                det.append("DNS lookup")
            add("network_egress", op, "; ".join(det))
            return
        # ---- dynamic code / eval
        exec_like = (k in PY_EXEC) if self.py else (k in JS_EXEC or k in ("setTimeout", "setInterval") and c.args and
                                                     c.args[0].type in ("string", "template_string"))
        if self.py and k in ("eval", "exec", "compile") and (c.recv is not None):
            exec_like = False
        if exec_like and last == "compile" and self._inside_exec(c):
            return
        if exec_like:
            a0 = self._arg(c, 0, "source", "code")
            op = clip(_t(a0), 80) if a0 is not None else "dynamic"
            if "network" in arglabs:
                add("download_then_exec", op, f"network content reaches {last}")
            if "decode" in arglabs:
                add("decode_then_exec", op, f"decoded data reaches {last}")
            add("dynamic_import_or_eval", op, last + ("" if a0 is None or self.lit(a0) is None else " of literal"))
            return
        dyn = None
        if self.py:
            if k in PY_DYN:
                a0 = self._arg(c, 0, "name", "path", "location")
                dyn = (self.show(a0) if a0 is not None else "dynamic", k.rsplit(".", 1)[-1] +
                       (" (literal)" if a0 is not None and self.lit(a0) is not None else ""))
            elif k.endswith(".exec_module") or k == "?.exec_module":
                dyn = (clip(_t(c.node), 80), "loader.exec_module")
            elif k in PY_DESER or k == "yaml.load" and not re.search(r"Safe|BaseLoader", _t(c.node)):
                dyn = (self.show(self._arg(c, 0)), "deserialisation " + k)
            elif k in PY_NATIVE or k.startswith("ctypes.") and last in ("LoadLibrary", "CDLL", "WinDLL"):
                dyn = (self.show(self._arg(c, 0)), "native library load")
        else:
            a0 = self._arg(c, 0)
            if k in ("require", "import") and a0 is not None and self.lit(a0) is None:
                dyn = (clip(_t(a0), 80), f"non-literal {k}")
            elif k == "import" and a0 is not None and re.match(r"(https?:|data:)", self.lit(a0) or ""):
                dyn = (self.lit(a0), "remote/data URL import")
                if (self.lit(a0) or "").startswith("http"):
                    add("download_then_exec", self.lit(a0), "remote module import")
            elif last == "_compile" or k in ("process.binding", "WebAssembly.instantiate", "WebAssembly.compile",
                                              "module.constructor._load", "Module._load"):
                dyn = (clip(_t(c.node), 80), k)
            elif k in ("process.dlopen",):
                dyn = (self.show(a0), "native addon load")
        if dyn:
            if "network" in arglabs:
                add("download_then_exec", dyn[0], "network content reaches dynamic import")
            if "decode" in arglabs:
                add("decode_then_exec", dyn[0], "decoded data reaches dynamic import")
            add("dynamic_import_or_eval", dyn[0], dyn[1])
            return
        # ---- archives
        if self._archive_new(c):
            add("archive_or_bundle", self.show(self._arg(c, 0, "name", "base_name")), "creates archive (" + k + ")")
        if last in ("extractall", "extract", "unpack_archive") and (
                k.startswith(("tarfile", "zipfile", "shutil")) or re.search(r"tar|zip|archive|bundle|whl|7z", _t(c.recv) if c.recv is not None else "", re.I)):
            tgt = self._arg(c, 0, "path", "extract_dir") if last != "unpack_archive" else self._arg(c, 1, "extract_dir")
            t = self.show(tgt) if tgt is not None else "."
            add("archive_or_bundle", t, "extracts archive")
            add("file_write", t, f"archive extraction; scope {P.path_scope(t)}")
        # ---- file writes / deletes
        w = self._file_write(c)
        if w:
            target, det = w
            labs = sorted(arglabs & {"network", "env", "credential", "decode"})
            if labs:
                det += "; content from: " + ", ".join(labs)
            add("file_write", target, det)
            for label, rx in P.PERSIST_LOCATIONS:
                if rx.search(target):
                    add("persistence", label, "write target " + clip(target, 80))
            return
        d = self._file_delete(c)
        if d:
            add("file_delete", d, "scope " + P.path_scope(d))
            return
        if self.py and k in ("os.environ.setdefault", "os.putenv") or k.endswith("environ.update"):
            pass

    def _inside_exec(self, c):
        p = c.node.parent
        while p is not None and p.type in ("argument_list", "parenthesized_expression", "keyword_argument"):
            p = p.parent
        if p is not None and p.type == "call":
            fn = p.child_by_field_name("function")
            return fn is not None and _t(fn) in ("exec", "eval", "builtins.exec", "builtins.eval")
        return False

    def _request_obj(self, n):
        """urlopen(req) where req = Request(url, data=...) -> the Request call."""
        if n is None:
            return None
        if n.type == "identifier" and len(self.exprs.get(_t(n), [])) == 1:
            n = self.exprs[_t(n)][0]
        if n.type != "call":
            return None
        for c in self.calls:
            if c.node.start_byte == n.start_byte and c.node.end_byte == n.end_byte:
                return c if c.callee.endswith(("urllib.request.Request", "urllib2.Request")) or c.callee == "Request" else None
        return None

    def _command_text(self, n, literal_only=False):
        if n is None:
            return "dynamic"
        if _t(n) in ("sys.executable", "process.execPath", "process.argv0"):
            return "python" if self.py else "node"
        v = self.lit(n)
        if v is not None:
            return v
        if n.type in ("list", "array", "tuple"):
            parts = [("python" if self.py else "node") if _t(x) in ("sys.executable", "process.execPath") else self.lit(x)
                     for x in n.named_children]
            if literal_only and any(p is None for p in parts):
                parts = [p if p is not None else "X" for p in parts]
                return " ".join(parts) if any(parts) else None
            return " ".join(p if p is not None else "{" + _t(x) + "}" for p, x in zip(parts, n.named_children))
        if n.type == "template_string" and not literal_only:
            return clip(_t(n)[1:-1], 140)
        if n.type == "template_string":
            return re.sub(r"\$\{[^}]*\}", "X", _t(n)[1:-1])
        if n.type == "identifier" and _t(n) in self.exprs and len(self.exprs[_t(n)]) == 1:
            return self._command_text(self.exprs[_t(n)][0], literal_only)
        return None if literal_only else self.show(n)

    def _mark_consumed(self, n):
        stack = [n]
        while stack:
            x = stack.pop()
            if x.type in ("string", "template_string"):
                self.consumed.add((x.start_byte, x.end_byte))
            stack.extend(x.named_children)
        if n.type == "identifier" and _t(n) in self.exprs:
            for r in self.exprs[_t(n)]:
                self._mark_consumed(r)

    def _file_write(self, c):
        k = c.callee
        last = k.rsplit(".", 1)[-1]
        if self.py:
            if k in ("open", "io.open", "codecs.open", "builtins.open", "os.open"):
                if k == "os.open":
                    flags = _t(self._arg(c, 1, "flags")) if self._arg(c, 1, "flags") is not None else ""
                    if not re.search(r"O_(WRONLY|RDWR|CREAT|APPEND|TRUNC)", flags):
                        return None
                elif not self._write_mode(c):
                    return None
                t = self.show(self._arg(c, 0, "file"))
                mode = self.lit(c.kwargs.get("mode") or (c.args[1] if len(c.args) > 1 else None)) or ""
                return t, ("append; " if "a" in mode else "") + "scope " + P.path_scope(t)
            if last in ("write_text", "write_bytes") and c.recv is not None:
                t = self.show(c.recv)
                return t, f"{last}; scope {P.path_scope(t)}"
            if last == "open" and c.recv is not None and k not in ("open",) and not k.startswith(("tarfile", "zipfile", "gzip", "webbrowser", "os.")):
                mode = self.lit(self._arg(c, 0, "mode"))
                if mode and any(ch in mode for ch in "wax+"):
                    t = self.show(c.recv)
                    return t, ("append; " if "a" in mode else "") + "scope " + P.path_scope(t)
            if k in PY_COPY or last in ("copy", "copy2", "copyfile", "copytree", "move") and k.startswith("shutil"):
                t = self.show(self._arg(c, 1, "dst"))
                return t, f"{last}; scope {P.path_scope(t)}"
            if k in ("os.symlink", "os.link"):
                t = self.show(self._arg(c, 1, "dst"))
                return t, f"{last}; scope {P.path_scope(t)}"
            if k in ("os.chmod", "os.lchmod") or last == "chmod" and c.recv is not None and not k.startswith("os."):
                t = self.show(self._arg(c, 0, "path")) if k.startswith("os.") else self.show(c.recv)
                mode = _t(self._arg(c, 1, "mode") if k.startswith("os.") else self._arg(c, 0, "mode"))
                return t, f"chmod {clip(mode, 30)}; scope {P.path_scope(t)}"
            if last in ("symlink_to", "hardlink_to", "rename", "replace") and c.recv is not None and not k.startswith(("os.", "?.replace", "str")):
                if last in ("rename", "replace") and not re.search(r"Path|path|file|dir|target|dest", _t(c.recv)):
                    return None
                t = self.show(c.recv) if last.endswith("_to") else self.show(self._arg(c, 0))
                return t, f"{last}; scope {P.path_scope(t)}"
            return None
        if k in JS_WRITE or k in JS_BARE_WRITE:
            i = JS_WRITE.get(k, 1 if k in ("copyFileSync", "renameSync", "symlinkSync") else 0)
            t = self.show(self._arg(c, i))
            return t, f"{last}; scope {P.path_scope(t)}"
        if k in ("fs.open", "fs.openSync") and len(c.args) > 1 and (self.lit(c.args[1]) or "")[:1] in ("w", "a"):
            t = self.show(c.args[0])
            return t, f"{last}; scope {P.path_scope(t)}"
        return None

    def _file_delete(self, c):
        k = c.callee
        last = k.rsplit(".", 1)[-1]
        if self.py:
            if k in PY_DELETE or k in ("rmtree",):
                return self.show(self._arg(c, 0, "path"))
            if last in ("unlink", "rmdir") and c.recv is not None and not k.startswith("os."):
                return self.show(c.recv)
            return None
        if k in JS_DELETE or k in JS_BARE_DELETE:
            return self.show(self._arg(c, 0))
        return None

    # ------------------------------------------------------------------ hooks, env, strings, regex layer
    def _hooks(self):
        name = self.u.path.rsplit("/", 1)[-1]
        if not self.py:
            return
        for f, (fnode, _) in self.funcs.items():
            if re.match(r"pytest_(configure|sessionstart|sessionfinish|unconfigure|collection\w*|runtest\w*|"
                        r"load_initial_conftests|addoption|plugin_registered|report\w*|terminal_summary)$", f):
                self.add("install_or_startup_hook", _line(fnode), f"pytest hook {f}()", "runs inside pytest",
                         context=self.context)
            if f == "run" and re.search(r"class \w+\((\w+\.)?(install|develop|egg_info|build_py|build_ext|sdist|bdist_wheel)\)", self.text):
                pass
        for m in re.finditer(r"^class (\w+)\(([\w.]*\b(install|develop|egg_info|build_py|build_ext|sdist|bdist_wheel|"
                             r"BuildHookInterface|MetadataHookInterface|BuildHook))\)", self.text, re.M):
            ln = self.text.count("\n", 0, m.start()) + 1
            self.add("install_or_startup_hook", ln, f"class {m.group(1)}({m.group(2)})", "build/install hook class",
                     source="regex")
        for i, l in enumerate(self.lines, 1):
            if re.search(r"\bautouse\s*=\s*True", l):
                nxt = next((self.lines[j] for j in range(i, min(i + 4, len(self.lines))) if "def " in self.lines[j]), "")
                fm = re.search(r"def (\w+)", nxt)
                self.add("install_or_startup_hook", i, f"autouse fixture {fm.group(1) if fm else '?'}",
                         "runs for every test", source="regex")
            if re.search(r"\bcmdclass\s*=", l):
                self.add("install_or_startup_hook", i, "setup(cmdclass=…)", "custom install/build commands", source="regex")
        for c in self.calls:
            if c.top and c.callee in self.funcs and name != "setup.py":
                self.add("install_or_startup_hook", c.line, f"import-time call {c.callee}()",
                         "module-level call of a local function", context=self.context)

    def strings_and_lines(self, root):
        # embedded command text in string literals
        stack = [root]
        while stack:
            n = stack.pop()
            if n.type in ("concatenated_string", "string", "template_string", "binary_expression") and _end(n) > _line(n):
                v = self.lit(n)
                if isinstance(v, str) and len(v) >= 200 and re.fullmatch(r"[A-Za-z0-9+/=_\-\s]+", v) and \
                        re.search(r"\d", v) and re.search(r"[A-Z]", v) and re.search(r"[a-z]", v):
                    self.add("obfuscation_markers", _line(n), f"encoded string literal ({len(''.join(v.split()))} chars)", "long encoded literal across lines", source="ast",
                             end=_end(n))
                    continue
            if n.type in ("string", "template_string") and (n.start_byte, n.end_byte) not in self.consumed:
                v = self.lit(n)
                if v is None and n.type == "template_string":
                    v = re.sub(r"\$\{[^}]*\}", "X", _t(n)[1:-1])
                if v is None and self.py and n.type == "string":
                    v = re.sub(r"\{[^}]*\}", "X", re.sub(r"^[a-zA-Z]{0,2}('''|\"\"\"|'|\")|('''|\"\"\"|'|\")$", "", _t(n)))
                if isinstance(v, str) and shell.looks_like_shell(v) and self.u.in_focus(_line(n), _end(n)):
                    for b in shell.analyse(v, self.u.path, base_line=_line(n), in_focus=self.u.in_focus,
                                           context=self.context, embedded="command text in string literal",
                                           emit_spawn=False, line_patterns=False):
                        if b.kind in ("env_read",):
                            continue
                        b.evidence = self.ev(b.line) if b.line <= len(self.lines) else b.evidence
                        self.out.append(b)
                continue
            stack.extend(n.children)
        # line regex layer
        comment = ("#",) if self.py else ("//", "/*", "*")
        env_names, bulk = {}, []
        docs = set()
        if self.py:
            st = [root]
            while st:
                x = st.pop()
                if x.type == "string" and x.parent is not None and x.parent.type in ("module", "block") or \
                        x.type == "expression_statement" and len(x.named_children) == 1 and x.named_children[0].type == "string":
                    docs.update(range(_line(x), _end(x) + 1))
                elif x.type in ("module", "block", "function_definition", "class_definition", "decorated_definition"):
                    st.extend(x.named_children)
        for i, l in enumerate(self.lines, 1):
            s = l.strip()
            if not s or s.startswith(comment) or i in docs 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" if "warn" not in label else "",
                             source="regex")
            for rx, label in P.PRIV_GRANT:
                if rx.search(s):
                    self.add("security_setting_change", i, label, "privilege/permission", source="regex")
            for label, rx in P.PERSIST_LOCATIONS:
                if rx.search(s):
                    self.add("persistence", i, label, "location referenced", source="regex")
            for op, det in P.obfuscation(l, self.lang):
                self.add("obfuscation_markers", i, op, det, source="regex")
            if self.py:
                for m in re.finditer(r"\b(?:os\.)?environ(?:b)?\s*\[\s*['\"]([^'\"]+)['\"]\s*\](\s*=(?!=))?|"
                                     r"\b(?:os\.)?(?:environ\.(?:get|pop|setdefault)|getenv)\(\s*['\"]([^'\"]+)['\"]", s):
                    name = m.group(1) or m.group(3)
                    if m.group(2):
                        continue
                    env_names.setdefault(name, i)
                if re.search(r"\bos\.environ\.(items|keys|values|copy)\(|dict\(os\.environ\)|\*\*os\.environ|"
                             r"\bfor \w+(, *\w+)? in os\.environ\b|json\.dumps\(\s*(dict\()?os\.environ", s):
                    bulk.append(i)
                elif re.search(r"\b(os\.)?getenv\(\s*[A-Za-z_]|\benviron\.get\(\s*[A-Za-z_]|\benviron\[\s*[A-Za-z_]", s):
                    env_names.setdefault("dynamic name", i)
            else:
                for m in re.finditer(r"\bprocess\.env\.([A-Za-z_$][\w$]*)(\s*=(?!=))?|\bprocess\.env\[\s*['\"`]([^'\"`]+)['\"`]\s*\](\s*=(?!=))?|"
                                     r"\b(?:Deno\.env\.get|Bun\.env\.get)\(\s*['\"]([^'\"]+)|\bimport\.meta\.env\.([A-Za-z_]\w*)|\bBun\.env\.([A-Za-z_]\w*)", s):
                    if m.group(2) or m.group(4):
                        continue
                    name = m.group(1) or m.group(3) or m.group(5) or m.group(6) or m.group(7)
                    env_names.setdefault(name, i)
                if re.search(r"Object\.(keys|entries|values|assign)\(\s*\{?\s*(\.\.\.)?process\.env|JSON\.stringify\(\s*process\.env|"
                             r"\.\.\.process\.env\b|[(,:]\s*process\.env\s*[),}]|=\s*process\.env\s*;?$|Deno\.env\.toObject", s):
                    bulk.append(i)
                elif re.search(r"process\.env\[\s*[A-Za-z_]", s):
                    env_names.setdefault("dynamic name", i)
        by_line = {}
        for name, ln in env_names.items():
            by_line.setdefault(ln, []).append(name)
        for ln, names in sorted(by_line.items()):
            det = "secret-like name" if any(P.SECRET_ENV_NAME.search(n) for n in names) else ""
            self.add("env_read", ln, ", ".join(sorted(names)), det, source="regex")
        for ln in bulk:
            markers = sorted(set(m.group(1).upper() for m in P.SECRET_MARKER_LITERAL.finditer(self.text)))
            det = "enumerates all variables" + (f"; name filter literals: {', '.join(markers)}" if markers else "")
            self.add("env_read", ln, "* (all)", det, source="regex")
            if markers:
                self.add("secret_or_credential_path", ln, "env vars matching " + "|".join(markers),
                         "secret-like environment selection", source="regex")


def analyse(unit, lang, context=""):
    fa = FileAnalysis(unit, lang, context)
    tree = parser(lang).parse(fa.src)
    root = tree.root_node
    fa.collect(root)
    fa.propagate()
    fa.apply()
    fa.strings_and_lines(root)
    return fa.out