Model shell -c commands

Author: theinfosecguy

python/ql/lib/change-notes/2026-08-26-shell-interpreter-command-arguments.md

@@ -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.

python/ql/lib/semmle/python/frameworks/Stdlib.qll

@@ -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() }
   }
 
   /**
@@ -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() }
   }
 
   /**
@@ -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` */
@@ -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. */
@@ -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
         (
@@ -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()
     }
   }
 

python/ql/test/library-tests/frameworks/stdlib/SystemCommandExecution.py

@@ -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"
 
 
 ########################################
@@ -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"