"""Shell command analysis, used for .sh files, CI run blocks, npm scripts, Dockerfile RUN, Makefile recipes, cron entries and command text found inside string literals of other languages. Line-oriented and tokenizer-based; never executes anything. """ from __future__ import annotations import re import shlex from .model import Behaviour, clip from . import patterns as P TRIVIAL = { "echo", "printf", "cd", "export", "set", "unset", "test", "[", "[[", "true", "false", "read", "local", "exit", "return", "shift", "cat", "grep", "egrep", "fgrep", "sed", "awk", "head", "tail", "sort", "uniq", "cut", "tr", "wc", "mkdir", "ls", "pwd", "basename", "dirname", "date", "sleep", "if", "then", "else", "elif", "fi", "for", "while", "do", "done", "case", "esac", "function", "declare", "typeset", "readonly", "trap", "wait", "jobs", ":", "{", "}", "(", ")", "!", "seq", "xargs", "printenv", "env", "which", "command", "type", "hash", "touch", "realpath", "readlink", "find", "stat", "du", "df", "whoami", "id", "uname", "hostname", "tee", "in", "select", } WRAPPERS = {"sudo", "nohup", "exec", "time", "nice", "ionice", "stdbuf", "timeout", "doas", "xargs", "command", "builtin", "then", "do", "else", "!", "env"} NET_CMDS = {"curl", "wget", "nc", "ncat", "netcat", "telnet", "ssh", "scp", "sftp", "ftp", "rsync", "socat", "aria2c", "http", "https", "Invoke-WebRequest", "iwr", "Invoke-RestMethod", "irm", "certutil", "bitsadmin"} INTERPRETERS = {"sh", "bash", "zsh", "dash", "ksh", "python", "python3", "python2", "node", "perl", "ruby", "php", "pwsh", "powershell", "iex", "Invoke-Expression", "deno", "bun", "source", "."} DELETE_CMDS = {"rm", "rmdir", "unlink", "shred", "srm", "Remove-Item", "del", "rd"} WRITE_CMDS = {"cp", "mv", "install", "ln", "dd", "chmod", "chown", "rsync", "tee"} COMMON_ENV = {"HOME", "PATH", "PWD", "USER", "SHELL", "TMPDIR", "TMP", "TEMP", "LANG", "LC_ALL", "TERM", "OLDPWD", "IFS", "PS1", "HOSTNAME", "UID", "EUID", "RANDOM", "SECONDS", "LINENO", "BASH_SOURCE", "OSTYPE", "CI", "GITHUB_WORKSPACE", "GITHUB_OUTPUT", "GITHUB_ENV", "GITHUB_PATH", "GITHUB_STEP_SUMMARY", "RUNNER_TEMP", "RUNNER_OS", "DEBIAN_FRONTEND", "XDG_CONFIG_HOME", "XDG_CACHE_HOME", "XDG_DATA_HOME", "LOGNAME"} _URL = re.compile(r"""(?:https?|ftp|wss?)://[^\s'"`<>)\]}|;]+""", re.I) _HOSTPORT = re.compile(r"\b(\d{1,3}(?:\.\d{1,3}){3}|[a-z0-9-]+(?:\.[a-z0-9-]+)+)(?::\d+)?\b", re.I) _VAR = re.compile(r"\$\{?([A-Za-z_][A-Za-z0-9_]*)") _SPLIT = re.compile(r"\|\||&&|;|\||\$\(|`|\n|\(|\)|&(?!>)") def _logical_lines(text: str): """Yield (line_no, text) joining backslash continuations.""" buf, start = "", None for i, ln in enumerate(text.split("\n"), 1): if start is None: start = i if ln.rstrip().endswith("\\"): buf += ln.rstrip()[:-1] + " " continue buf += ln yield start, buf, i buf, start = "", None if buf: yield start, buf, start def _blank_quotes(s: str) -> str: """Same length as s with quoted content replaced by '_' so separators inside quotes are ignored.""" out, q, i = list(s), None, 0 while i < len(s): ch = s[i] if q: if ch == "\\" and q == '"' and i + 1 < len(s): out[i] = out[i + 1] = "_" i += 2 continue if ch == q: q = None else: out[i] = "_" elif ch in ("'", '"'): q = ch i += 1 return "".join(out) def _split_by(rx, s: str): b = _blank_quotes(s) parts, last = [], 0 for m in rx.finditer(b): parts.append(s[last:m.start()]) last = m.end() parts.append(s[last:]) return parts _CHAIN = re.compile(r"\|\||&&|;|\$\(|`|\(|\)|&(?![>&])") _PIPE = re.compile(r"(?") lex.whitespace_split = True lex.commenters = "" return list(lex) except ValueError: return seg.split() def _argv0(toks): i = 0 while i < len(toks): t = toks[i] if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*=.*", t): i += 1 continue if t in ("command", "type", "hash") and i + 1 < len(toks) and toks[i + 1] in ("-v", "-V", "-p"): return i, "command" if t in WRAPPERS or t.startswith("-") and i > 0 and toks[i - 1] in WRAPPERS: i += 1 continue return i, t.rsplit("/", 1)[-1] if "/" in t and not t.startswith((".", "$")) else t return len(toks), "" def _scope(path: str) -> str: return P.path_scope(path) def analyse(text: str, path: str, *, base_line: int = 1, in_focus=lambda a, b: True, context: str = "", embedded: str = "", emit_spawn: bool = True, src: str = "shell", line_patterns: bool = True): """Analyse shell text. `embedded` is a note such as 'in string literal'; base_line maps text line 1.""" out = [] lines_seen = 0 def add(kind, operand, line, ev, detail=""): d = "; ".join(x for x in (detail, embedded) if x) out.append(Behaviour(kind, path, line, clip(operand, 140) or "dynamic", ev, d, context, src)) for start, logical, end in _logical_lines(text): a, b = base_line + start - 1, base_line + end - 1 lines_seen += 1 stripped = logical.strip() if not stripped or stripped.startswith("#"): continue if not in_focus(a, b): continue ev = stripped lower = stripped.lower() line_out_start = len(out) # Whole-line patterns first: pipelines into interpreters. for chain in _split_by(_CHAIN, stripped): pipe_parts = [p.strip() for p in _split_by(_PIPE, chain)] if len(pipe_parts) < 2: continue heads = [_argv0(_tokens(p))[1] for p in pipe_parts] for j in range(1, len(pipe_parts)): h = heads[j] if h in INTERPRETERS: before = heads[:j] if any(x in NET_CMDS for x in before): add("download_then_exec", " | ".join(heads[:j + 1]), a, ev, "network output piped into interpreter") elif re.search(r"base64\s+(-d|--decode|-D)|xxd\s+-r|openssl\s+(enc|base64)\s+-d|\b(gunzip|zcat)\b", " | ".join(pipe_parts[:j])): add("decode_then_exec", " | ".join(heads[:j + 1]), a, ev, "decoded data piped into interpreter") if re.search(r"<\(\s*(curl|wget)\b|\$\(\s*(curl|wget)\b[^)]*\)\s*\"?\s*$|(sh|bash|python3?|node)\s+-c\s+\"?\$\(\s*(curl|wget)", stripped) and \ re.search(r"\b(bash|sh|zsh|source|\.|eval|python3?|node)\b", stripped): add("download_then_exec", "process substitution of a download", a, ev, "network output executed") if re.search(r"\beval\b.*\b(base64\s+(-d|--decode)|xxd\s+-r)", stripped): add("decode_then_exec", "eval of decoded data", a, ev) if "/dev/tcp/" in stripped or "/dev/udp/" in stripped: m = re.search(r"/dev/(?:tcp|udp)/([^/\s]+)/(\d+)", stripped) add("network_egress", m.group(1) + ":" + m.group(2) if m else "/dev/tcp", a, ev, "bash /dev/tcp socket") # Per simple command. for seg in _split_by(_SPLIT, stripped): seg = seg.strip() if not seg: continue toks = _tokens(seg) i, cmd = _argv0(toks) if not cmd or cmd.startswith("#"): continue args = toks[i + 1:] argline = " ".join(args) # assignments / exports with security relevance for m in re.finditer(r"\b(NODE_TLS_REJECT_UNAUTHORIZED|PYTHONHTTPSVERIFY|GIT_SSL_NO_VERIFY|" r"CURL_CA_BUNDLE|REQUESTS_CA_BUNDLE|NPM_CONFIG_STRICT_SSL|npm_config_strict_ssl|" r"PIP_INDEX_URL|PIP_EXTRA_INDEX_URL|PIP_TRUSTED_HOST|NPM_CONFIG_REGISTRY|" r"npm_config_registry|UV_INDEX_URL|YARN_REGISTRY)=(\S*)", seg): add("security_setting_change", f"{m.group(1)}={m.group(2)}", a, ev, "environment setting") if cmd in TRIVIAL and cmd not in ("tee", "find", "env", "printenv"): pass elif cmd in ("env", "printenv", "set", "export") and not args or cmd == "printenv" and not args: add("env_read", "* (all)", a, ev, f"{cmd} dumps environment") elif emit_spawn and cmd not in ("tee", "find", "env", "printenv"): add("process_spawn", seg, a, ev) if cmd in ("env", "printenv") and not args: add("env_read", "* (all)", a, ev, f"{cmd} dumps environment") if cmd in NET_CMDS: urls = _URL.findall(seg) if urls: for u in urls: add("network_egress", u, a, ev, _curl_detail(cmd, args)) else: host = next((m.group(0) for t in args if not t.startswith("-") for m in [re.search(r"(?:[\w.-]+@)?([\w.-]+\.[a-z]{2,}|\d+(?:\.\d+){3})(?::\S*)?", t, re.I)] if m), "") add("network_egress", host or "dynamic", a, ev, _curl_detail(cmd, args)) if cmd == "curl" and re.search(r"(^|\s)(-k|--insecure)\b", argline): add("security_setting_change", "curl --insecure", a, ev, "TLS verification off") if cmd == "wget" and "--no-check-certificate" in argline: add("security_setting_change", "wget --no-check-certificate", a, ev, "TLS verification off") elif cmd == "git" and args: sub = args[0] if sub in ("clone", "fetch", "pull", "push", "ls-remote", "submodule") and _URL.search(argline): add("network_egress", _URL.search(argline).group(0), a, ev, f"git {sub}") elif sub in ("push",): add("network_egress", "git remote " + " ".join(args[1:3]), a, ev, "git push") if sub == "config": if re.search(r"http\.sslverify\s+(false|0)", argline, re.I): add("security_setting_change", "git http.sslVerify false", a, ev, "TLS verification off") if "core.hookspath" in argline.lower(): add("persistence", "git core.hooksPath " + " ".join(args[2:3]), a, ev, "git hooks location") if re.search(r"\burl\..*\.insteadof\b", argline, re.I): add("security_setting_change", "git url.insteadOf " + clip(argline, 80), a, ev, "remote rewrite") if sub == "clean" and re.search(r"-\w*[fdx]", argline): add("file_delete", "git clean " + argline, a, ev, "untracked files") elif cmd in ("npm", "yarn", "pnpm", "bun") and args: if args[:2] == ["config", "set"] or args[:1] == ["set"] or args[:2] == ["config", "set"]: kv = args[2:] if args[0] == "config" else args[1:] k = kv[0] if kv else "" v = kv[1] if len(kv) > 1 else (k.split("=", 1)[1] if "=" in k else "") k = k.split("=", 1)[0] if re.search(r"registry|strict-ssl|ca$|cafile|ignore-scripts|always-auth|_authtoken|npmRegistryServer|enableStrictSsl", k, re.I): add("security_setting_change", f"{cmd} config {k}={v}", a, ev, "registry change" if "registry" in k.lower() else "package manager setting") for t in args if args[0] != "config" else (): if t.startswith("--registry"): add("security_setting_change", f"{cmd} {t}", a, ev, "registry change") if _URL.match(t) or re.match(r"(git\+|github:|gist:|file:)", t): add("network_egress", t, a, ev, f"{cmd} install from URL") elif cmd in ("pip", "pip3", "uv", "pipx", "poetry") or cmd in ("python", "python3") and args[:2] == ["-m", "pip"]: for j, t in enumerate(args): if t in ("-i", "--index-url", "--extra-index-url", "--trusted-host", "-f", "--find-links") or \ t.startswith(("--index-url=", "--extra-index-url=", "--trusted-host=")): val = t.split("=", 1)[1] if "=" in t else (args[j + 1] if j + 1 < len(args) else "") add("security_setting_change", f"pip {t.split('=')[0]} {val}", a, ev, "trusted host" if "trusted" in t else "package index change") elif _URL.match(t) or t.startswith("git+"): add("network_egress", t, a, ev, "pip install from URL") if args[:2] == ["config", "set"] and len(args) > 2: add("security_setting_change", "pip config " + " ".join(args[2:4]), a, ev, "package index change") if cmd in DELETE_CMDS or cmd == "find" and "-delete" in args: targets = [t for t in args if not t.startswith("-")] or ["dynamic"] for t in targets[:4] if cmd != "find" else targets[:1]: add("file_delete", t, a, ev, ("recursive; " if re.search(r"(^|\s)-\w*r", argline) else "") + "scope " + _scope(t)) if cmd in WRITE_CMDS and args: tgt = [t for t in args if not t.startswith("-")] if cmd == "dd": tgt = [t[3:] for t in args if t.startswith("of=")] if tgt: t = tgt[-1] if cmd != "chmod" else (tgt[-1] if len(tgt) > 1 else tgt[0]) det = {"chmod": "permission change " + (tgt[0] if len(tgt) > 1 else ""), "ln": "symlink", "tee": "tee"}.get(cmd, cmd) add("file_write", t, a, ev, f"{det.strip()}; scope {_scope(t)}") if cmd == "chmod" and re.search(r"\b(777|666|[ugoa]*\+s|4755|2755)\b", argline): add("security_setting_change", f"chmod {tgt[0]}", a, ev, "broad or setuid permission") if cmd in ("sed", "perl") and re.search(r"(^|\s)-\w*i", argline): tgt = [t for t in args if not t.startswith("-")] if len(tgt) > 1: add("file_write", tgt[-1], a, ev, f"{cmd} -i in-place edit; scope {_scope(tgt[-1])}") if cmd == "crontab" or cmd in ("systemctl",) and args[:1] in (["enable"], ["daemon-reload"]) or \ cmd in ("launchctl",) and args[:1] in (["load"], ["bootstrap"]) or cmd in ("schtasks", "at", "update-rc.d", "chkconfig"): add("persistence", f"{cmd} {' '.join(args[:3])}".strip(), a, ev, "scheduler/service registration") if cmd in ("reg", "reg.exe") and re.search(r"\\Run\b|CurrentVersion\\Run", argline, re.I): add("persistence", "registry Run key", a, ev, "Windows autostart") if cmd == "security" and args[:1] and re.match(r"(find|dump)-(generic|internet)-password|dump-keychain", args[0]): add("secret_or_credential_path", "macOS keychain: " + args[0], a, ev, "keychain access") if cmd in ("setenforce",) and args[:1] == ["0"] or cmd == "ufw" and args[:1] == ["disable"] or \ cmd == "iptables" and args[:1] in (["-F"], ["--flush"]): add("security_setting_change", seg, a, ev, "host security control off") if cmd == "eval": add("dynamic_import_or_eval", "eval " + clip(argline, 80), a, ev, "shell eval") if cmd in ("source", ".") and args and (_URL.search(argline) or "$(" in seg or "<(" in seg): add("dynamic_import_or_eval", f"{cmd} {args[0]}", a, ev, "sources dynamic file") if cmd in ("tar", "zip", "7z", "gzip", "unzip", "rar") and args: mode = "creates" if cmd in ("zip", "7z", "gzip", "rar") and not (cmd == "7z" and args[0] == "x") \ or cmd == "tar" and re.match(r"-?\w*c", args[0]) else "extracts" add("archive_or_bundle", f"{cmd} {clip(argline, 80)}", a, ev, f"{mode} archive") # Redirections to files. blanked = _blank_quotes(stripped) for m in re.finditer(r"(?>?|&>>?|[12]>>?)\s*([^\s;&|<>()]+)", blanked): tgt = stripped[m.start(2):m.end(2)] if tgt in ("/dev/null", "&1", "&2", "/dev/stderr", "/dev/stdout") or tgt.startswith("&"): continue if re.fullmatch(r"\$\{?GITHUB_(OUTPUT|ENV|PATH|STEP_SUMMARY)\}?", tgt.strip('"')): continue add("file_write", tgt.strip('"\''), a, ev, ("append; " if ">>" in m.group(1) else "") + "scope " + _scope(tgt)) # Persistence / secret locations referenced on this line. for label, rx in P.PERSIST_LOCATIONS if line_patterns else (): if rx.search(stripped): add("persistence", label, a, ev, "location referenced") for label in P.secret_hits(stripped) if line_patterns else (): add("secret_or_credential_path", label, a, ev) if re.match(r"@reboot\b|(\S+\s+){4}\S+\s+\S", stripped) and P.CRON_LINE.match(stripped): add("persistence", "cron entry: " + clip(stripped, 60), a, ev, "cron schedule line") # Environment variable reads (uppercase names, minus common ones). names = sorted({v for v in _VAR.findall(stripped) if v.isupper() and v not in COMMON_ENV and not v.isdigit()}) if names: add("env_read", ", ".join(names), a, ev) _ = line_out_start return out def _curl_detail(cmd, args): al = " ".join(args) bits = [] if cmd == "curl": m = re.search(r"(?:-X|--request)\s*([A-Z]+)", al) if m: bits.append(m.group(1)) if re.search(r"(^|\s)(-d|--data\S*|-F|--form|-T|--upload-file|--json)(\s|=|$)", al): bits.append("sends body" + (" (file upload)" if re.search(r"(^|\s)(-T|--upload-file)(\s|=)|(^|\s)(-d|--data\S*|-F|--form)\s+['\"]?@", al) else "")) if re.search(r"(^|\s)(-o|-O|--output)\b", al): bits.append("saves to file") elif cmd == "wget": if "--post" in al: bits.append("POST") if re.search(r"(^|\s)-O\s*-", al) or "-qO-" in al: bits.append("to stdout") elif cmd in ("nc", "ncat", "netcat", "socat"): if re.search(r"(^|\s)-\w*l", al): bits.append("listens") if re.search(r"(^|\s)-e\b|EXEC:", al): bits.append("attaches a process") elif cmd in ("scp", "rsync", "sftp"): bits.append("file transfer") return f"{cmd}" + (": " + ", ".join(bits) if bits else "") SHELLISH = re.compile(r"(^#!\s*/\S*(sh|bash|zsh)\b)|\b(curl|wget|ncat|netcat|crontab|systemctl|launchctl|powershell|" r"base64\s+-d|base64\s+--decode|chmod\s+\+x|/dev/tcp/|nohup|bash\s+-c|sh\s+-c|rm\s+-rf|" r"npm\s+config\s+set|pip\s+install|git\s+config)\b|@reboot\b|\|\s*(ba)?sh\b", re.M) def looks_like_shell(s: str) -> bool: return len(s) >= 6 and bool(SHELLISH.search(s))