Blame for extractor/behaviour/shell.py

"""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)
        # 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"(?<![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))
        # 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))