Implement S609, linux_commands_wildcard_injection

crates/ruff/resources/test/fixtures/flake8_bandit/S609.py

@@ -0,0 +1,8 @@
+import os
+import subprocess
+
+os.popen("chmod +w foo*")
+subprocess.Popen("/bin/chown root: *", shell=True)
+subprocess.Popen(["/usr/local/bin/rsync", "*", "some_where:"], shell=True)
+subprocess.Popen("/usr/local/bin/rsync * no_injection_here:")
+os.system("tar cf foo.tar bar/*")

crates/ruff/src/checkers/ast/mod.rs

@@ -2735,6 +2735,7 @@ where
                     Rule::StartProcessWithAShell,
                     Rule::StartProcessWithNoShell,
                     Rule::StartProcessWithPartialPath,
+                    Rule::UnixCommandWildcardInjection,
                 ]) {
                     flake8_bandit::rules::shell_injection(self, func, args, keywords);
                 }

crates/ruff/src/codes.rs

@@ -529,6 +529,7 @@ pub fn code_to_rule(linter: Linter, code: &str) -> Option<(RuleGroup, Rule)> {
         (Flake8Bandit, "606") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::StartProcessWithNoShell),
         (Flake8Bandit, "607") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::StartProcessWithPartialPath),
         (Flake8Bandit, "608") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::HardcodedSQLExpression),
+        (Flake8Bandit, "609") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::UnixCommandWildcardInjection),
         (Flake8Bandit, "612") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::LoggingConfigInsecureListen),
         (Flake8Bandit, "701") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::Jinja2AutoescapeFalse),
 

crates/ruff/src/rules/flake8_bandit/mod.rs

@@ -28,6 +28,7 @@ mod tests {
     #[test_case(Rule::HashlibInsecureHashFunction, Path::new("S324.py"))]
     #[test_case(Rule::Jinja2AutoescapeFalse, Path::new("S701.py"))]
     #[test_case(Rule::LoggingConfigInsecureListen, Path::new("S612.py"))]
+    #[test_case(Rule::ParamikoCall, Path::new("S601.py"))]
     #[test_case(Rule::RequestWithNoCertValidation, Path::new("S501.py"))]
     #[test_case(Rule::RequestWithoutTimeout, Path::new("S113.py"))]
     #[test_case(Rule::SnmpInsecureVersion, Path::new("S508.py"))]
@@ -41,8 +42,8 @@ mod tests {
     #[test_case(Rule::SuspiciousTelnetUsage, Path::new("S312.py"))]
     #[test_case(Rule::TryExceptContinue, Path::new("S112.py"))]
     #[test_case(Rule::TryExceptPass, Path::new("S110.py"))]
+    #[test_case(Rule::UnixCommandWildcardInjection, Path::new("S609.py"))]
     #[test_case(Rule::UnsafeYAMLLoad, Path::new("S506.py"))]
-    #[test_case(Rule::ParamikoCall, Path::new("S601.py"))]
     fn rules(rule_code: Rule, path: &Path) -> Result<()> {
         let snapshot = format!("{}_{}", rule_code.noqa_code(), path.to_string_lossy());
         let diagnostics = test_path(

crates/ruff/src/rules/flake8_bandit/rules/mod.rs

@@ -28,7 +28,7 @@ pub(crate) use request_without_timeout::{request_without_timeout, RequestWithout
 pub(crate) use shell_injection::{
     shell_injection, CallWithShellEqualsTrue, StartProcessWithAShell, StartProcessWithNoShell,
     StartProcessWithPartialPath, SubprocessPopenWithShellEqualsTrue,
-    SubprocessWithoutShellEqualsTrue,
+    SubprocessWithoutShellEqualsTrue, UnixCommandWildcardInjection,
 };
 pub(crate) use snmp_insecure_version::{snmp_insecure_version, SnmpInsecureVersion};
 pub(crate) use snmp_weak_cryptography::{snmp_weak_cryptography, SnmpWeakCryptography};

crates/ruff/src/rules/flake8_bandit/rules/shell_injection.rs

@@ -89,106 +89,29 @@ impl Violation for StartProcessWithPartialPath {
     }
 }
 
-#[derive(Copy, Clone, Debug)]
-enum CallKind {
-    Subprocess,
-    Shell,
-    NoShell,
-}
-
-/// Return the [`CallKind`] of the given function call.
-fn get_call_kind(func: &Expr, model: &SemanticModel) -> Option<CallKind> {
-    model
-        .resolve_call_path(func)
-        .and_then(|call_path| match call_path.as_slice() {
-            &[module, submodule] => match module {
-                "os" => match submodule {
-                    "execl" | "execle" | "execlp" | "execlpe" | "execv" | "execve" | "execvp"
-                    | "execvpe" | "spawnl" | "spawnle" | "spawnlp" | "spawnlpe" | "spawnv"
-                    | "spawnve" | "spawnvp" | "spawnvpe" | "startfile" => Some(CallKind::NoShell),
-                    "system" | "popen" | "popen2" | "popen3" | "popen4" => Some(CallKind::Shell),
-                    _ => None,
-                },
-                "subprocess" => match submodule {
-                    "Popen" | "call" | "check_call" | "check_output" | "run" => {
-                        Some(CallKind::Subprocess)
-                    }
-                    _ => None,
-                },
-                "popen2" => match submodule {
-                    "popen2" | "popen3" | "popen4" | "Popen3" | "Popen4" => Some(CallKind::Shell),
-                    _ => None,
-                },
-                "commands" => match submodule {
-                    "getoutput" | "getstatusoutput" => Some(CallKind::Shell),
-                    _ => None,
-                },
-                _ => None,
-            },
-            _ => None,
-        })
-}
-
-#[derive(Copy, Clone, Debug)]
-struct ShellKeyword<'a> {
-    /// Whether the `shell` keyword argument is set and evaluates to `True`.
-    truthiness: Truthiness,
-    /// The `shell` keyword argument.
-    keyword: &'a Keyword,
-}
-
-/// Return the `shell` keyword argument to the given function call, if any.
-fn find_shell_keyword<'a>(
-    model: &SemanticModel,
-    keywords: &'a [Keyword],
-) -> Option<ShellKeyword<'a>> {
-    keywords
-        .iter()
-        .find(|keyword| keyword.arg.as_ref().map_or(false, |arg| arg == "shell"))
-        .map(|keyword| ShellKeyword {
-            truthiness: Truthiness::from_expr(&keyword.value, |id| model.is_builtin(id)),
-            keyword,
-        })
-}
-
-/// Return `true` if the value provided to the `shell` call seems safe. This is based on Bandit's
-/// definition: string literals are considered okay, but dynamically-computed values are not.
-fn shell_call_seems_safe(arg: &Expr) -> bool {
-    matches!(
-        arg,
-        Expr::Constant(ast::ExprConstant {
-            value: Constant::Str(_),
-            ..
-        })
-    )
-}
+#[violation]
+pub struct UnixCommandWildcardInjection;
 
-/// Return the [`Expr`] as a string literal, if it's a string or a list of strings.
-fn try_string_literal(expr: &Expr) -> Option<&str> {
-    match expr {
-        Expr::List(ast::ExprList { elts, .. }) => {
-            if elts.is_empty() {
-                None
-            } else {
-                string_literal(&elts[0])
-            }
-        }
-        _ => string_literal(expr),
+impl Violation for UnixCommandWildcardInjection {
+    #[derive_message_formats]
+    fn message(&self) -> String {
+        format!("Possible wildcard injection in call due to `*` usage")
     }
 }
 
-/// S602, S603, S604, S605, S606, S607
+/// S602, S603, S604, S605, S606, S607, S609
 pub(crate) fn shell_injection(
     checker: &mut Checker,
     func: &Expr,
     args: &[Expr],
     keywords: &[Keyword],
 ) {
     let call_kind = get_call_kind(func, checker.semantic_model());
+    let shell_keyword = find_shell_keyword(checker.semantic_model(), keywords);
 
     if matches!(call_kind, Some(CallKind::Subprocess)) {
         if let Some(arg) = args.first() {
-            match find_shell_keyword(checker.semantic_model(), keywords) {
+            match shell_keyword {
                 // S602
                 Some(ShellKeyword {
                     truthiness: Truthiness::Truthy,
@@ -229,7 +152,7 @@ pub(crate) fn shell_injection(
     } else if let Some(ShellKeyword {
         truthiness: Truthiness::Truthy,
         keyword,
-    }) = find_shell_keyword(checker.semantic_model(), keywords)
+    }) = shell_keyword
     {
         // S604
         if checker.enabled(Rule::CallWithShellEqualsTrue) {
@@ -240,9 +163,9 @@ pub(crate) fn shell_injection(
     }
 
     // S605
-    if matches!(call_kind, Some(CallKind::Shell)) {
-        if let Some(arg) = args.first() {
-            if checker.enabled(Rule::StartProcessWithAShell) {
+    if checker.enabled(Rule::StartProcessWithAShell) {
+        if matches!(call_kind, Some(CallKind::Shell)) {
+            if let Some(arg) = args.first() {
                 checker.diagnostics.push(Diagnostic::new(
                     StartProcessWithAShell {
                         seems_safe: shell_call_seems_safe(arg),
@@ -254,26 +177,170 @@ pub(crate) fn shell_injection(
     }
 
     // S606
-    if matches!(call_kind, Some(CallKind::NoShell)) {
-        if checker.enabled(Rule::StartProcessWithNoShell) {
+    if checker.enabled(Rule::StartProcessWithNoShell) {
+        if matches!(call_kind, Some(CallKind::NoShell)) {
             checker
                 .diagnostics
                 .push(Diagnostic::new(StartProcessWithNoShell, func.range()));
         }
     }
 
     // S607
-    if call_kind.is_some() {
-        if let Some(arg) = args.first() {
-            if checker.enabled(Rule::StartProcessWithPartialPath) {
-                if let Some(value) = try_string_literal(arg) {
-                    if FULL_PATH_REGEX.find(value).is_none() {
-                        checker
-                            .diagnostics
-                            .push(Diagnostic::new(StartProcessWithPartialPath, arg.range()));
-                    }
+    if checker.enabled(Rule::StartProcessWithPartialPath) {
+        if call_kind.is_some() {
+            if let Some(arg) = args.first() {
+                if is_partial_path(arg) {
+                    checker
+                        .diagnostics
+                        .push(Diagnostic::new(StartProcessWithPartialPath, arg.range()));
+                }
+            }
+        }
+    }
+
+    // S609
+    if checker.enabled(Rule::UnixCommandWildcardInjection) {
+        if matches!(call_kind, Some(CallKind::Shell))
+            || matches!(
+                (call_kind, shell_keyword),
+                (
+                    Some(CallKind::Subprocess),
+                    Some(ShellKeyword {
+                        truthiness: Truthiness::Truthy,
+                        keyword: _,
+                    })
+                )
+            )
+        {
+            if let Some(arg) = args.first() {
+                if is_wildcard_command(arg) {
+                    checker
+                        .diagnostics
+                        .push(Diagnostic::new(UnixCommandWildcardInjection, func.range()));
                 }
             }
         }
     }
 }
+
+#[derive(Copy, Clone, Debug)]
+enum CallKind {
+    Subprocess,
+    Shell,
+    NoShell,
+}
+
+/// Return the [`CallKind`] of the given function call.
+fn get_call_kind(func: &Expr, model: &SemanticModel) -> Option<CallKind> {
+    model
+        .resolve_call_path(func)
+        .and_then(|call_path| match call_path.as_slice() {
+            &[module, submodule] => match module {
+                "os" => match submodule {
+                    "execl" | "execle" | "execlp" | "execlpe" | "execv" | "execve" | "execvp"
+                    | "execvpe" | "spawnl" | "spawnle" | "spawnlp" | "spawnlpe" | "spawnv"
+                    | "spawnve" | "spawnvp" | "spawnvpe" | "startfile" => Some(CallKind::NoShell),
+                    "system" | "popen" | "popen2" | "popen3" | "popen4" => Some(CallKind::Shell),
+                    _ => None,
+                },
+                "subprocess" => match submodule {
+                    "Popen" | "call" | "check_call" | "check_output" | "run" => {
+                        Some(CallKind::Subprocess)
+                    }
+                    _ => None,
+                },
+                "popen2" => match submodule {
+                    "popen2" | "popen3" | "popen4" | "Popen3" | "Popen4" => Some(CallKind::Shell),
+                    _ => None,
+                },
+                "commands" => match submodule {
+                    "getoutput" | "getstatusoutput" => Some(CallKind::Shell),
+                    _ => None,
+                },
+                _ => None,
+            },
+            _ => None,
+        })
+}
+
+#[derive(Copy, Clone, Debug)]
+struct ShellKeyword<'a> {
+    /// Whether the `shell` keyword argument is set and evaluates to `True`.
+    truthiness: Truthiness,
+    /// The `shell` keyword argument.
+    keyword: &'a Keyword,
+}
+
+/// Return the `shell` keyword argument to the given function call, if any.
+fn find_shell_keyword<'a>(
+    model: &SemanticModel,
+    keywords: &'a [Keyword],
+) -> Option<ShellKeyword<'a>> {
+    keywords
+        .iter()
+        .find(|keyword| keyword.arg.as_ref().map_or(false, |arg| arg == "shell"))
+        .map(|keyword| ShellKeyword {
+            truthiness: Truthiness::from_expr(&keyword.value, |id| model.is_builtin(id)),
+            keyword,
+        })
+}
+
+/// Return `true` if the value provided to the `shell` call seems safe. This is based on Bandit's
+/// definition: string literals are considered okay, but dynamically-computed values are not.
+fn shell_call_seems_safe(arg: &Expr) -> bool {
+    matches!(
+        arg,
+        Expr::Constant(ast::ExprConstant {
+            value: Constant::Str(_),
+            ..
+        })
+    )
+}
+
+/// Return `true` if the [`Expr`] is a string literal or list of string literals that starts with a
+/// partial path.
+fn is_partial_path(expr: &Expr) -> bool {
+    let string_literal = match expr {
+        Expr::List(ast::ExprList { elts, .. }) => elts.first().and_then(string_literal),
+        _ => string_literal(expr),
+    };
+    string_literal.map_or(false, |text| !FULL_PATH_REGEX.is_match(text))
+}
+
+/// Return `true` if the [`Expr`] is a wildcard command.
+///
+/// ## Examples
+/// ```python
+/// import subprocess
+///
+/// subprocess.Popen("/bin/chown root: *", shell=True)
+/// subprocess.Popen(["/usr/local/bin/rsync", "*", "some_where:"], shell=True)
+/// ```
+fn is_wildcard_command(expr: &Expr) -> bool {
+    if let Expr::List(ast::ExprList { elts, .. }) = expr {
+        let mut has_star = false;
+        let mut has_command = false;
+        for elt in elts.iter() {
+            if let Some(text) = string_literal(elt) {
+                has_star |= text.contains('*');
+                has_command |= text.contains("chown")
+                    || text.contains("chmod")
+                    || text.contains("tar")
+                    || text.contains("rsync");
+            }
+            if has_star && has_command {
+                break;
+            }
+        }
+        has_star && has_command
+    } else {
+        let string_literal = string_literal(expr);
+        string_literal.map_or(false, |text| {
+            text.contains('*')
+                && (text.contains("chown")
+                    || text.contains("chmod")
+                    || text.contains("tar")
+                    || text.contains("rsync"))
+        })
+    }
+}