| """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: |
| _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"} |
|
|
| |
| 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 = {} |
| self.calls = [] |
| self.name_labels, self.func_labels = {}, {} |
| self.callgraph = {} |
| self.top_called = set() |
| self.out = [] |
| self.consumed = set() |
| self.withs, self.fors, self.returns = [], [], [] |
| self.idents_cache = {} |
|
|
| |
| 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)) |
|
|
| |
| 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 |
| 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) |
|
|
| |
| 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 |
| |
| 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 |
|
|
| |
| 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+") |
|
|
| |
| 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) |
| |
| 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 |
| |
| 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 |
| |
| 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 |
| |
| 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)}") |
| |
| 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 |
|
|
| |
| 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): |
| |
| 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) |
| |
| 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 |
|
|
