Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
---
category: minorAnalysis
---
* Direct `-c` command arguments to recognized POSIX shell interpreters through `os.exec*`,
`os.spawn*`, `os.posix_spawn*`, and `subprocess` APIs are now treated as command-injection
sinks.
159 changes: 125 additions & 34 deletions python/ql/lib/semmle/python/frameworks/Stdlib.qll
Original file line number Diff line number Diff line change
Expand Up @@ -1171,27 +1171,67 @@ module StdlibPrivate {
override predicate isShellInterpreted(DataFlow::Node arg) { arg = this.getCommand() }
}

/**
* Holds if `flag` makes `interpreter` execute the following argument as a command.
*
* See https://docs.python.org/3/library/subprocess.html#popen-constructor.
*/
private predicate isShellCommandFlag(DataFlow::Node interpreter, DataFlow::Node flag) {
interpreter.asExpr().(StringLiteral).getText().regexpMatch("(.*/)?(sh|bash|dash|zsh)") and
flag.asExpr().(StringLiteral).getText() = "-c"
}

/** Gets the command from separate interpreter, flag, and command arguments. */
private DataFlow::Node getShellCommandFromArguments(
DataFlow::Node interpreter, DataFlow::Node flag, DataFlow::Node command
) {
isShellCommandFlag(interpreter, flag) and
result = command
}

/** Gets the command from an argument sequence passed to `interpreter`. */
private DataFlow::Node getShellCommandFromSequence(
DataFlow::Node interpreter, DataFlow::Node arguments
) {
exists(SequenceNode sequence, DataFlow::Node flag |
arguments.asCfgNode() = sequence and
flag.asCfgNode() = sequence.getElement(1) and
isShellCommandFlag(interpreter, flag) and
result.asCfgNode() = sequence.getElement(2)
)
}

/**
* A call to any of the `os.exec*` functions
* See https://docs.python.org/3.8/library/os.html#os.execl
*/
private class OsExecCall extends SystemCommandExecution::Range, FileSystemAccess::Range,
DataFlow::CallCfgNode
{
string name;

OsExecCall() {
exists(string name |
name in ["execl", "execle", "execlp", "execlpe", "execv", "execve", "execvp", "execvpe"] and
this = os().getMember(name).getACall()
)
name in ["execl", "execle", "execlp", "execlpe", "execv", "execve", "execvp", "execvpe"] and
this = os().getMember(name).getACall()
}

private DataFlow::Node getShellCommand() {
name in ["execl", "execlp"] and
result = getShellCommandFromArguments(this.getArg(0), this.getArg(2), this.getArg(3))
or
name in ["execle", "execlpe"] and
exists(this.getArg(4)) and
result = getShellCommandFromArguments(this.getArg(0), this.getArg(2), this.getArg(3))
or
name in ["execv", "execve", "execvp", "execvpe"] and
result = getShellCommandFromSequence(this.getArg(0), this.getArg(1))
}

override DataFlow::Node getCommand() { result = this.getArg(0) }
override DataFlow::Node getCommand() { result in [this.getArg(0), this.getShellCommand()] }

override DataFlow::Node getAPathArgument() { result = this.getCommand() }
override DataFlow::Node getAPathArgument() { result = this.getArg(0) }

override predicate isShellInterpreted(DataFlow::Node arg) {
none() // this is a safe API.
}
override predicate isShellInterpreted(DataFlow::Node arg) { arg = this.getShellCommand() }
}

/**
Expand All @@ -1201,28 +1241,44 @@ module StdlibPrivate {
private class OsSpawnCall extends SystemCommandExecution::Range, FileSystemAccess::Range,
DataFlow::CallCfgNode
{
string name;

OsSpawnCall() {
exists(string name |
name in [
"spawnl", "spawnle", "spawnlp", "spawnlpe", "spawnv", "spawnve", "spawnvp", "spawnvpe"
] and
this = os().getMember(name).getACall()
)
name in [
"spawnl", "spawnle", "spawnlp", "spawnlpe", "spawnv", "spawnve", "spawnvp", "spawnvpe"
] and
this = os().getMember(name).getACall()
}

override DataFlow::Node getCommand() {
private DataFlow::Node getInterpreter() {
result = this.getArg(1)
or
// `file` keyword argument only valid for the `v` variants, but this
// over-approximation is not hurting anyone, and is easy to implement.
result = this.getArgByName("file")
}

override DataFlow::Node getAPathArgument() { result = this.getCommand() }
private DataFlow::Node getArguments() { result in [this.getArg(2), this.getArgByName("args")] }

override predicate isShellInterpreted(DataFlow::Node arg) {
none() // this is a safe API.
private DataFlow::Node getShellCommand() {
name in ["spawnl", "spawnlp"] and
result = getShellCommandFromArguments(this.getInterpreter(), this.getArg(3), this.getArg(4))
or
name in ["spawnle", "spawnlpe"] and
exists(this.getArg(5)) and
result = getShellCommandFromArguments(this.getInterpreter(), this.getArg(3), this.getArg(4))
or
name in ["spawnv", "spawnve", "spawnvp", "spawnvpe"] and
result = getShellCommandFromSequence(this.getInterpreter(), this.getArguments())
}

override DataFlow::Node getCommand() {
result in [this.getInterpreter(), this.getShellCommand()]
}

override DataFlow::Node getAPathArgument() { result = this.getInterpreter() }

override predicate isShellInterpreted(DataFlow::Node arg) { arg = this.getShellCommand() }
}

/**
Expand All @@ -1234,13 +1290,23 @@ module StdlibPrivate {
{
OsPosixSpawnCall() { this = os().getMember(["posix_spawn", "posix_spawnp"]).getACall() }

override DataFlow::Node getCommand() { result in [this.getArg(0), this.getArgByName("path")] }
private DataFlow::Node getInterpreter() {
result in [this.getArg(0), this.getArgByName("path")]
}

override DataFlow::Node getAPathArgument() { result = this.getCommand() }
private DataFlow::Node getArguments() { result in [this.getArg(1), this.getArgByName("argv")] }

override predicate isShellInterpreted(DataFlow::Node arg) {
none() // this is a safe API.
private DataFlow::Node getShellCommand() {
result = getShellCommandFromSequence(this.getInterpreter(), this.getArguments())
}

override DataFlow::Node getCommand() {
result in [this.getInterpreter(), this.getShellCommand()]
}

override DataFlow::Node getAPathArgument() { result = this.getInterpreter() }

override predicate isShellInterpreted(DataFlow::Node arg) { arg = this.getShellCommand() }
}

/** An additional taint step for calls to `os.path.join` */
Expand All @@ -1267,13 +1333,11 @@ module StdlibPrivate {
* ref: https://docs.python.org/3/library/subprocess.html#legacy-shell-invocation-functions
*/
private class SubprocessPopenCall extends SystemCommandExecution::Range, API::CallNode {
string name;

SubprocessPopenCall() {
exists(string name |
name in [
"Popen", "call", "check_call", "check_output", "run", "getoutput", "getstatusoutput"
] and
this = subprocess().getMember(name).getACall()
)
name in ["Popen", "call", "check_call", "check_output", "run", "getoutput", "getstatusoutput"] and
this = subprocess().getMember(name).getACall()
}

/** Gets the API-node for the `args` argument, if any. */
Expand All @@ -1296,20 +1360,43 @@ module StdlibPrivate {
/** Gets the API-node for the `executable` argument, if any. */
private API::Node get_executable_arg() { result = this.getParameter(2, "executable") }

/** Holds if the `executable` argument overrides the executable from `args`. */
private predicate hasExecutableOverride() {
exists(DataFlow::Node executable |
executable = this.get_executable_arg().asSink() and
not executable.asExpr() instanceof None
)
}

/** Gets the interpreter that will execute `args`, if it can be determined. */
private DataFlow::Node getInterpreter() {
this.hasExecutableOverride() and
result = this.get_executable_arg().asSink()
or
not this.hasExecutableOverride() and
result.asCfgNode() = this.get_args_arg().asSink().asCfgNode().(SequenceNode).getElement(0)
}

/** Gets a shell command passed in the `args` sequence. */
private DataFlow::Node getShellCommand() {
name in ["Popen", "call", "check_call", "check_output", "run"] and
this.get_shell_arg_value() = false and
result = getShellCommandFromSequence(this.getInterpreter(), this.get_args_arg().asSink())
}

override DataFlow::Node getCommand() {
// TODO: Track arguments ("args" and "shell")
// TODO: Handle using `args=["sh", "-c", <user-input>]`
this.hasExecutableOverride() and
result = this.get_executable_arg().asSink()
or
exists(DataFlow::Node arg_args, boolean shell |
arg_args = this.get_args_arg().asSink() and
shell = this.get_shell_arg_value()
|
// When "executable" argument is set, and "shell" argument is `False`, the
// "args" argument will only be used to set the program name and arguments to
// When the `executable` argument overrides `args[0]`, and `shell` is `False`, the
// `args` argument will only be used to set the program name and arguments to
// the program, so we should not consider any of them as command execution.
not (
exists(this.get_executable_arg()) and
this.hasExecutableOverride() and
shell = false
) and
(
Expand All @@ -1327,11 +1414,15 @@ module StdlibPrivate {
result = arg_args
)
)
or
result = this.getShellCommand()
}

override predicate isShellInterpreted(DataFlow::Node arg) {
arg = [this.get_executable_arg(), this.get_args_arg()].asSink() and
this.get_shell_arg_value() = true
or
arg = this.getShellCommand()
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -105,30 +105,61 @@ def os_members():
########################################
# actively using known shell as the executable

subprocess.Popen(["/bin/sh", "-c", "vuln"]) # $ getCommand="/bin/sh" MISSING: getCommand="vuln"
subprocess.Popen(["/bin/bash", "-c", "vuln"]) # $ getCommand="/bin/bash" MISSING: getCommand="vuln"
subprocess.Popen(["/bin/dash", "-c", "vuln"]) # $ getCommand="/bin/dash" MISSING: getCommand="vuln"
subprocess.Popen(["/bin/zsh", "-c", "vuln"]) # $ getCommand="/bin/zsh" MISSING: getCommand="vuln"

subprocess.Popen(["sh", "-c", "vuln"]) # $ getCommand="sh" MISSING: getCommand="vuln"
subprocess.Popen(["bash", "-c", "vuln"]) # $ getCommand="bash" MISSING: getCommand="vuln"
subprocess.Popen(["dash", "-c", "vuln"]) # $ getCommand="dash" MISSING: getCommand="vuln"
subprocess.Popen(["zsh", "-c", "vuln"]) # $ getCommand="zsh" MISSING: getCommand="vuln"
subprocess.Popen(["/bin/sh", "-c", "vuln"]) # $ getCommand="/bin/sh" getCommand="vuln"
subprocess.Popen(["/bin/bash", "-c", "vuln"]) # $ getCommand="/bin/bash" getCommand="vuln"
subprocess.Popen(["/bin/dash", "-c", "vuln"]) # $ getCommand="/bin/dash" getCommand="vuln"
subprocess.Popen(["/bin/zsh", "-c", "vuln"]) # $ getCommand="/bin/zsh" getCommand="vuln"

subprocess.Popen(["sh", "-c", "vuln"]) # $ getCommand="sh" getCommand="vuln"
subprocess.Popen(["bash", "-c", "vuln"]) # $ getCommand="bash" getCommand="vuln"
subprocess.Popen(["dash", "-c", "vuln"]) # $ getCommand="dash" getCommand="vuln"
subprocess.Popen(["zsh", "-c", "vuln"]) # $ getCommand="zsh" getCommand="vuln"
subprocess.run(("/usr/local/bin/sh", "-c", "vuln")) # $ getCommand="/usr/local/bin/sh" getCommand="vuln"
subprocess.run(args=["sh", "-c", "vuln"]) # $ getCommand="sh" getCommand="vuln"
subprocess.Popen(["sh", "-c", "vuln", "not-command"]) # $ getCommand="sh" getCommand="vuln"

# Check that we don't consider ANY argument a command injection sink
subprocess.Popen(["sh", "/bin/python"]) # $ getCommand="sh"
subprocess.Popen(["sh", "--command", "not-vuln"]) # $ getCommand="sh"
subprocess.Popen(["sh", "-C", "not-vuln"]) # $ getCommand="sh"
subprocess.Popen(["not-a-shell", "-c", "not-vuln"]) # $ getCommand="not-a-shell"
subprocess.Popen(["sh", "-c"]) # $ getCommand="sh"
subprocess.Popen(["sh", "-c", "not-vuln"], shell=True) # $ getCommand="sh"
subprocess.getoutput(["sh", "-c", "not-vuln"]) # $ getCommand="sh"
subprocess.getstatusoutput(("sh", "-c", "not-vuln")) # $ getCommand="sh"

subprocess.Popen(["cmd.exe", "/c", "vuln"]) # $ getCommand="cmd.exe" MISSING: getCommand="vuln"
subprocess.Popen(["cmd.exe", "/C", "vuln"]) # $ getCommand="cmd.exe" MISSING: getCommand="vuln"
subprocess.Popen(["cmd", "/c", "vuln"]) # $ getCommand="cmd" MISSING: getCommand="vuln"
subprocess.Popen(["cmd", "/C", "vuln"]) # $ getCommand="cmd" MISSING: getCommand="vuln"

subprocess.Popen(["<progname>", "-c", "vuln"], executable="/bin/bash") # $ getCommand="/bin/bash" MISSING: getCommand="vuln"
subprocess.Popen(["<progname>", "-c", "vuln"], executable="/bin/bash") # $ getCommand="/bin/bash" getCommand="vuln"
subprocess.Popen(["sh", "-c", "not-vuln"], executable="/bin/echo") # $ getCommand="/bin/echo"
subprocess.Popen(["sh", "-c", "vuln"], executable=None) # $ getCommand="sh" getCommand="vuln"

if UNKNOWN:
os.execl("/bin/sh", "<progname>", "-c", "vuln") # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" MISSING: getCommand="vuln"

os.spawnl(os.P_WAIT, "/bin/sh", "<progname>", "-c", "vuln") # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" MISSING: getCommand="vuln"
os.execl("/bin/sh", "<progname>", "-c", "vuln") # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.execle("/bin/sh", "<progname>", "-c", "vuln", env) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.execlp("sh", "<progname>", "-c", "vuln") # $ getCommand="sh" getAPathArgument="sh" getCommand="vuln"
os.execlpe("sh", "<progname>", "-c", "vuln", env) # $ getCommand="sh" getAPathArgument="sh" getCommand="vuln"
os.execv("/bin/sh", ("<progname>", "-c", "vuln")) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.execvp("sh", ["<progname>", "-c", "vuln"]) # $ getCommand="sh" getAPathArgument="sh" getCommand="vuln"

os.execl("/bin/sh", "<progname>", "--command", "not-vuln") # $ getCommand="/bin/sh" getAPathArgument="/bin/sh"
os.execle("/bin/sh", "<progname>", "-c", env) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh"
os.execlpe("sh", "<progname>", "-c", env) # $ getCommand="sh" getAPathArgument="sh"
os.execv("not-a-shell", ["<progname>", "-c", "not-vuln"]) # $ getCommand="not-a-shell" getAPathArgument="not-a-shell"

os.spawnl(os.P_WAIT, "/bin/sh", "<progname>", "-c", "vuln") # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.spawnle(os.P_WAIT, "/bin/sh", "<progname>", "-c", "vuln", env) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.spawnle(os.P_WAIT, "/bin/sh", "<progname>", "-c", env) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh"
os.spawnlpe(os.P_WAIT, "sh", "<progname>", "-c", "vuln", env) # $ getCommand="sh" getAPathArgument="sh" getCommand="vuln"
os.spawnlpe(os.P_WAIT, "sh", "<progname>", "-c", env) # $ getCommand="sh" getAPathArgument="sh"
os.spawnv(os.P_WAIT, "/bin/sh", ["<progname>", "-c", "vuln"]) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.spawnv(mode=os.P_WAIT, file="/bin/sh", args=["<progname>", "-c", "vuln"]) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.posix_spawn("/bin/sh", ["<progname>", "-c", "vuln"], env) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.posix_spawn(path="/bin/sh", argv=["<progname>", "-c", "vuln"], env=env) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh" getCommand="vuln"
os.posix_spawnp("sh", ["<progname>", "-c", "vuln"], env) # $ getCommand="sh" getAPathArgument="sh" getCommand="vuln"


########################################
Expand All @@ -137,6 +168,10 @@ def os_members():
args = ["/bin/sh", "-c", "vuln"]
subprocess.Popen(args) # $ getCommand=args

exec_args = ["<progname>", "-c", "vuln"] # $ MISSING: getCommand="vuln"
if UNKNOWN:
os.execv("/bin/sh", exec_args) # $ getCommand="/bin/sh" getAPathArgument="/bin/sh"

args = "<progname>"
use_shell = False
exe = "executable"
Expand Down
Loading