| """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"(?<!\|)\|(?!\|)") |
|
|
|
|
| def _tokens(seg: str): |
| try: |
| lex = shlex.shlex(seg, posix=True, punctuation_chars="<>") |
| 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) |
| |
| 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") |
| |
| 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) |
| |
| 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") |
| |
| blanked = _blank_quotes(stripped) |
| for m in re.finditer(r"(?<![0-9&<=])(>>?|&>>?|[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)) |
| |
| 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") |
| |
| 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)) |
|
|
