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