Bashxtract API

Extract commands used from bash command lines

Introduction

safecmd.bashxtract provides tools for parsing and extracting commands from bash command strings. It’s designed for security-conscious applications where you need to understand exactly what commands a shell script will execute before running it.

The core use case is validating shell commands from untrusted sources (like LLM-generated commands) against an allowlist. Rather than trying to regex-match bash syntax—which is notoriously tricky—this module uses shfmt, a proper bash parser, to build an AST and then extracts all executable commands from it.

It’s likely that the only function you’ll actually need from here is extract_commands. But we provide a full API of all the pieces we use to build that function, which we’ll take you through here.

The Problem

Suppose we want to find the commands in this pipeline. Let’s try shlex.split:

cmd = '''
echo | head 2 <<EOF
asdf
jkljl
EOF
'''

shlex.split(cmd)

['echo', '|', 'head', '2', '<<EOF', 'asdf', 'jkljl', 'EOF']

That gives us tokens, but it loses the distinction we need. echo and head are separate commands; | connects them. The text after <<EOF supplies input to head, not more arguments. shlex treats them all as arguments.

We’ll use shfmt to parse the bash syntax into a JSON AST (abstract syntax tree). The shfmt-py dependency supplies its binary when you install safecmd:

!shfmt --help 2>&1 | head -6

usage: shfmt [flags] [path ...]

shfmt formats shell programs. If the only argument is a dash ('-') or no
arguments are given, standard input will be used. If a given path is a
directory, all shell scripts found under that directory will be used.

The flag we need is --to-json:

!shfmt --help 2>&1 | grep to-json

  --to-json           print syntax tree to stdout as a typed JSON

shfmt returns nested dictionaries. A node’s Type tells us what it contains: CallExpr is a command invocation, BinaryCmd joins commands with a pipeline or logical operator, and Word holds an argument.

For our heredoc example, we want to extract:

1. The commands: ['echo'] and ['head', '2']
2. The operators used: {'|'} (a pipe)
3. The heredoc content attached to the command that receives it

Our goal is to walk the AST and produce commands with their arguments and a set of operators that we can validate against an allowlist.

We’ll build up to extract_commands in these stages:

1. Parsing (parse_bash): Convert bash syntax to a JSON AST using shfmt.
2. Text extraction (part_text, word_text): Recover argument text, including quotes, escapes, and expansions.
3. AST walking (visit_stmts, nested_stmts): Find commands throughout the AST, including commands nested in substitutions.
4. Operator detection (collect_ops): Identify pipes, redirects, and logical operators.
5. Validation (check_types): Reject bash constructs we don’t understand.
6. Main API (extract_commands): Combine these pieces into one interface.

Parsing

---

source

parse_bash

def parse_bash(
    cmd:str, shfmt:str='shfmt'
):

Parse cmd using shfmt

parse_bash runs shfmt --to-json and loads its output into a Python dict. A missing binary raises FileNotFoundError; a syntax error raises ValueError with shfmt’s diagnostic.

parse_bash('echo hello')

{'Type': 'File',
 'Pos': {'Offset': 0, 'Line': 1, 'Col': 1},
 'End': {'Offset': 10, 'Line': 1, 'Col': 11},
 'Stmts': [{'Pos': {'Offset': 0, 'Line': 1, 'Col': 1},
   'End': {'Offset': 10, 'Line': 1, 'Col': 11},
   'Cmd': {'Type': 'CallExpr',
    'Pos': {'Offset': 0, 'Line': 1, 'Col': 1},
    'End': {'Offset': 10, 'Line': 1, 'Col': 11},
    'Args': [{'Pos': {'Offset': 0, 'Line': 1, 'Col': 1},
      'End': {'Offset': 4, 'Line': 1, 'Col': 5},
      'Parts': [{'Type': 'Lit',
        'Pos': {'Offset': 0, 'Line': 1, 'Col': 1},
        'End': {'Offset': 4, 'Line': 1, 'Col': 5},
        'ValuePos': {'Offset': 0, 'Line': 1, 'Col': 1},
        'ValueEnd': {'Offset': 4, 'Line': 1, 'Col': 5},
        'Value': 'echo'}]},
     {'Pos': {'Offset': 5, 'Line': 1, 'Col': 6},
      'End': {'Offset': 10, 'Line': 1, 'Col': 11},
      'Parts': [{'Type': 'Lit',
        'Pos': {'Offset': 5, 'Line': 1, 'Col': 6},
        'End': {'Offset': 10, 'Line': 1, 'Col': 11},
        'ValuePos': {'Offset': 5, 'Line': 1, 'Col': 6},
        'ValueEnd': {'Offset': 10, 'Line': 1, 'Col': 11},
        'Value': 'hello'}]}]},
   'Position': {'Offset': 0, 'Line': 1, 'Col': 1}}]}

Text Extraction

---

source

part_text

def part_text(
    p, cmd
):

Extracts the text value from a single word part node in the shfmt AST.

An argument can contain several parts. part_text removes backslash-space escapes from literals and recovers single- or double-quoted text. It keeps parameter expressions such as $var, ${var}, and ${arr[0]}. For command and process substitutions, it uses the AST’s source offsets to copy the original expression.

part_text({'Type': 'SglQuoted', 'Value': 'foo bar'}, "echo 'foo bar'")

'foo bar'

---

source

word_text

def word_text(
    w, cmd
):

Converts a Word node (with Parts) into its full text repr by concatenating part_text for each part.

word_text({'Parts': [{'Type': 'Lit', 'Value': 'hello'}]}, 'echo hello')

'hello'

---

source

nested_stmts

def nested_stmts(
    parts
):

Yield all Stmts lists from nested Parts recursively

Substitutions contain their own statements. nested_stmts follows nested Parts arrays and yields each Stmts list it finds. This also reaches $(...) and <(...) inside double quotes:

AST Walking

parts = [{'Type': 'CmdSubst', 'Stmts': [{'Cmd': {...}}]}]
list(nested_stmts(parts))

[[{'Cmd': {Ellipsis}}]]

The AST stores operators as numeric codes. Rather than hard-code them, _ensure_ops parses a small sample of each operator and caches the codes on first use:

_ensure_ops()
_op_map, _redir_ops, _write_ops

({11: '&&', 12: '||', 13: '|', 14: '|&'},
 {63: '>', 64: '>>', 65: '<', 67: '<&', 68: '>&', 74: '&>', 76: '&>>'},
 {63: '>', 64: '>>', 74: '&>', 76: '&>>'})

---

source

visit_stmts

def visit_stmts(
    stmts, cmd, commands:NoneType=None
):

Visit statements, appending commands and handling redirects

visit_stmts adds each command to a list as [cmd, arg1, arg2, ...]. It follows nested statements and substitutions too. For redirects, it appends heredoc text to the latest command; a here-string contributes <<< and its content. It returns the accumulated list:

parsed = parse_bash('echo foo; cat file')
visit_stmts(parsed['Stmts'], 'echo foo; cat file')

[['echo', 'foo'], ['cat', 'file']]

---

source

collect_ops

def collect_ops(
    node, ops:NoneType=None
):

Walk AST node and collect all operators into a set

collect_ops records background execution (&), semicolons (;), and assignments (=) from node attributes. It looks up binary operators such as &&, ||, |, and |& in _op_map, and redirects such as >, >>, and < in _redir_ops.

collect_ops(parse_bash('echo a && echo b | cat > out.txt'))

{'&&', '>', '|'}

---

source

collect_redirects

def collect_redirects(
    node, cmd, redirects:NoneType=None
):

Walk AST node and collect all write redirect destinations as (op, dest) tuples

collect_redirects extracts all write redirect destinations from the AST. For each output redirect (>, >>, &>, &>>, >&), it returns a tuple of (operator, destination). This is used to validate that commands only write to allowed destinations.

>&file writes stdout and stderr to a file. >&2 duplicates a file descriptor instead, so it contributes no write destination.

cmd = 'echo a && echo b | cat > out.txt'
collect_redirects(parse_bash(cmd), cmd)

[('>', 'out.txt')]

from fastcore.test import test_eq

def _redirects(cmd): return collect_redirects(parse_bash(cmd), cmd)

test_eq(_redirects('echo hi > out.txt'), [('>', 'out.txt')])
test_eq(_redirects('echo hi >> log.txt'), [('>>', 'log.txt')])
test_eq(_redirects('echo hi &> both.txt'), [('&>', 'both.txt')])
test_eq(_redirects('echo hi >& both.txt'), [('>&', 'both.txt')])
test_eq(_redirects('cat > a > b'), [('>', 'a'), ('>', 'b')])
test_eq(_redirects('echo hi | cat'), [])  # no redirects
test_eq(_redirects('echo hi < in.txt'), [])  # input redirect, not output
test_eq(_redirects('cmd > "$HOME/file"'), [('>', '$HOME/file')])  # variable in dest

---

source

scan_flag_args

def scan_flag_args(
    commands, exec_flags:NoneType=None, dest_flags:NoneType=None, dest_pos:NoneType=None, exec_pos:NoneType=None
):

Scan commands for exec/dest flags and positional exec/dest args

Some arguments name another command or a write destination. scan_flag_args finds these using four dictionaries, each keyed by command name:

Dictionary	What to collect	Example
exec_flags	The argument after each listed flag, as a command	{'find': {'-exec', '-execdir'}}
dest_flags	The argument after each listed flag, as a destination	{'curl': {'-o', '--output'}}
dest_pos	Arguments at the listed destination positions	{'ex': {0}, 'cp': {-1}}
exec_pos	Arguments at the listed command positions	{'env': {0}, 'xargs': {0}}

Positions start at zero after the command name. Negative indices count from the end, as in Python.

The result is (extra_cmds, extra_dests). The caller recursively parses each string in extra_cmds. Each destination is a (flag_or_idx, dest) pair for the redirect validator.

exec_flags = {'find': {'-exec', '-execdir'}}
dest_flags = {'curl': {'-o', '--output'}}

# Find with -exec extracts the command arg
scan_flag_args([['find', '.', '-exec', 'ls', '{}', ';']], exec_flags=exec_flags)

(['ls'], [])

# curl with -o extracts the destination
scan_flag_args([['curl', '-o', '/tmp/file', 'http://example.com']], dest_flags=dest_flags)

([], [('-o', '/tmp/file')])

# cat -o is NOT treated as a dest flag (not in dest_flags for cat)
scan_flag_args([['cat', '-o', '/etc/passwd']], dest_flags=dest_flags)

([], [])

dest_pos = {'ex': {0}, 'cp': {-1}}

scan_flag_args([['ex', 'somefile']], dest_pos=dest_pos)

([], [(0, 'somefile')])

scan_flag_args([['cp', 'src.txt', 'dest.txt']], dest_pos=dest_pos)

([], [(-1, 'dest.txt')])

In ex somefile, destination position 0 picks somefile. In cp src dest, position -1 picks dest. An entry for ex doesn’t affect commands absent from dest_pos.

The same indexing works for exec_pos. Here, position 0 picks ls from env ls and grep from xargs grep, ready for recursive command validation:

exec_pos = {'env': {0}, 'xargs': {0}}

scan_flag_args([['env', 'ls', '-la']], exec_pos=exec_pos)

(['ls'], [])

scan_flag_args([['xargs', 'grep', 'pattern']], exec_pos=exec_pos)

(['grep'], [])

---

source

check_types

def check_types(
    node
):

Raise ValueError if AST contains unhandled node types

Raises ValueError if any node has a Type not in HANDLED_TYPES. Use this to detect unsupported bash constructs early, ensuring the rest of the parsing pipeline won’t silently skip or mishandle unknown syntax.

check_types(parse_bash('echo hello'))

try: check_types(parse_bash('[[ -f foo ]]'))
except ValueError: print('Caught unhandled construct')

Caught unhandled construct

Main API

---

source

extract_commands

def extract_commands(
    cmd, shfmt:str='shfmt', exec_flags:NoneType=None, dest_flags:NoneType=None, dest_pos:NoneType=None,
    exec_pos:NoneType=None
):

Split bash command into (commands, operators, redirects)

Call extract_commands with a bash string to get (commands, operators, redirects). commands holds token lists, including nested commands; operators is a set; redirects holds (op, dest) pairs for write destinations. The tokens look like shlex.split output, but come from shfmt’s AST so we can distinguish bash constructs.

The four optional dictionaries described above tell it which arguments need further inspection. exec_flags and exec_pos contribute strings to parse recursively. dest_flags and dest_pos contribute destinations. For example:

Source and configuration	Additional result
find . -exec ls, exec_flags={'find': {'-exec'}}	Command ['ls']
curl -o /tmp/f url, dest_flags={'curl': {'-o'}}	Redirect ('-o', '/tmp/f')
ex somefile, dest_pos={'ex': {0}}	Redirect (0, 'somefile')
cp src dest, dest_pos={'cp': {-1}}	Redirect (-1, 'dest')
env ls, exec_pos={'env': {0}}	Command ['ls']

Both position dictionaries use zero-based argument indices after the command name, with Python-style negative indices. xargs can use the same exec_pos entry as env.

A simple call such as echo foo gives [['echo', 'foo']]. A pipeline such as cat file | grep x gives two commands, ['cat', 'file'] and ['grep', 'x']. The sequences ;, &, &&, and || likewise separate commands.

Substitutions keep their source text in the outer command and add the inner commands to the result. Thus echo $(whoami) gives [['echo', '$(whoami)'], ['whoami']]; backticks work the same way. For diff <(ls a) <(ls b), we get diff with its two substitution arguments, followed by ['ls', 'a'] and ['ls', 'b']. This continues recursively through nested substitutions. Subshells work too: (cd /tmp && rm *) gives ['cd', '/tmp'] and ['rm', '*'].

Quoted strings and escaped spaces stay within one token. Heredoc (<<EOF) and here-string (<<<) content each occupies one token; the here-string also keeps the <<< token. Output redirects go into the third result: echo hi > file.txt gives [('>', 'file.txt')].

Here are tests for these cases and their combinations:

from fastcore.test import test_eq

def test_split(a, *b, ops=set(), redirs=[]): test_eq(extract_commands(a), (list(b), ops, redirs))

test_split('echo <<EOF\nasdf\njkljl\nEOF\n', ['echo', 'asdf\njkljl'])
test_split('echo $(foo)', ['echo', '$(foo)'], ['foo'])
test_split('echo $(foo) | cat -a', ['echo', '$(foo)'], ['foo'], ['cat', '-a'], ops={'|'})
test_split('echo $(cat $(ls))', ['echo', '$(cat $(ls))'], ['cat', '$(ls)'], ['ls'])
test_split('echo "hello world" foo', ['echo', 'hello world', 'foo'])
test_split('echo hello\\ world', ['echo', 'hello world'])
test_split('echo foo; echo bar', ['echo', 'foo'], ['echo', 'bar'], ops={';'})
test_split('time pytest -q', ['pytest', '-q'])
test_split('echo $HOME "${USER}"', ['echo', '$HOME', '${USER}'])
test_split('sleep 10 &', ['sleep', '10'], ops={';', '&'})
test_split('cat <<< "some text"', ['cat', '<<<', 'some text'])
test_split("echo \"it's a 'test'\"", ['echo', "it's a 'test'"])
test_split('echo "hello $(whoami) there"', ['echo', 'hello $(whoami) there'], ['whoami'])
test_split('echo "path is ${HOME}/bin"', ['echo', 'path is ${HOME}/bin'])
test_split('echo ${arr[0]}', ['echo', '${arr[0]}'])
test_split('echo "$(echo "inner")"', ['echo', '$(echo "inner")'], ['echo', 'inner'])
test_split('echo "$HOME/$(whoami)/file"', ['echo', '$HOME/$(whoami)/file'], ['whoami'])
test_split('echo `whoami`', ['echo', '`whoami`'], ['whoami'])
test_split('(cd /tmp && rm -rf *)', ['cd', '/tmp'], ['rm', '-rf', '*'], ops={'&&'})
test_split('eval "rm -rf /"', ['eval', 'rm -rf /'])
test_split('echo a && echo b || echo c', ['echo', 'a'], ['echo', 'b'], ['echo', 'c'], ops={'&&', '||'})
test_split('cat file > out', ['cat', 'file'], ops={'>'}, redirs=[('>', 'out')])
test_split('cat file >> out', ['cat', 'file'], ops={'>>'}, redirs=[('>>', 'out')])
test_split('cat < in', ['cat'], ops={'<'})
test_split('diff <(ls dir1) <(ls dir2)', ['diff', '<(ls dir1)', '<(ls dir2)'], ['ls', 'dir1'], ['ls', 'dir2'])
test_split('FOO=bar', ops={'='})
test_split('FOO=bar echo hello', ['echo', 'hello'], ops={'='})
test_split('for i in a b c; do echo $i; done', ['echo', '$i'], ops={';'})

test_split('echo &>file', ['echo'], ops={'&>'}, redirs=[('&>', 'file')])
test_split('echo &>>file', ['echo'], ops={'&>>'}, redirs=[('&>>', 'file')])
test_split('echo |& cat', ['echo'], ['cat'], ops={'|&'})

# fd duplication - not file redirects, so no redirs
test_split('echo >&2', ['echo'], ops={'>&'})
test_split('cat <&3', ['cat'], ops={'<&'})
test_split("sed -i '' 's/foo/bar/' file.txt", ['sed', '-i', '', 's/foo/bar/', 'file.txt'])

exec_flags = {'find': {'-exec', '-execdir'}, 'tar': {'--to-command', '-I'}}
dest_flags = {'curl': {'-o', '--output'}}
dest_pos = dict(ex={0}, tee={0}, cp={-1}, mv={-1})
exec_pos = {'env': {0}, 'xargs': {0}}

def test_split_flags(a, *b, ops=set(), redirs=[], exec_f=exec_flags, dest_f=dest_flags, dest_p=dest_pos, exec_p=exec_pos):
    test_eq(extract_commands(a, exec_flags=exec_f, dest_flags=dest_f, dest_pos=dest_p, exec_pos=exec_p), (list(b), ops, redirs))

test_split_flags('find . -exec ls', ['find', '.', '-exec', 'ls'], ['ls'])
test_split_flags(r'find . -exec rm -rf {} \;', ['find', '.', '-exec', 'rm', '-rf', '{}', r'\;'], ['rm'])
test_split_flags(r'find . -execdir cat {} \;', ['find', '.', '-execdir', 'cat', '{}', r'\;'], ['cat'])

test_split_flags('curl -o /tmp/out http://x', ['curl', '-o', '/tmp/out', 'http://x'], redirs=[('-o', '/tmp/out')])
test_split_flags('curl --output file.txt http://x', ['curl', '--output', 'file.txt', 'http://x'], redirs=[('--output', 'file.txt')])

test_split_flags('cat -o /etc/passwd', ['cat', '-o', '/etc/passwd'])

test_split_flags('find . -exec "ls | head"', ['find', '.', '-exec', 'ls | head'], ['ls'], ['head'], ops={'|'})

test_split_flags('ex somefile', ['ex', 'somefile'], redirs=[(0, 'somefile')])
test_split_flags('tee output.log', ['tee', 'output.log'], redirs=[(0, 'output.log')])
test_split_flags('cp src.txt dest.txt', ['cp', 'src.txt', 'dest.txt'], redirs=[(-1, 'dest.txt')])
test_split_flags('mv old.txt new.txt', ['mv', 'old.txt', 'new.txt'], redirs=[(-1, 'new.txt')])