feat(rules): implement flake8-bandit S507 (ssh_no_host_key_verification)

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

@@ -0,0 +1,24 @@
+from paramiko import client
+from paramiko.client import AutoAddPolicy, WarningPolicy
+
+ssh_client = client.SSHClient()
+
+# OK
+ssh_client.set_missing_host_key_policy(policy=foo)
+ssh_client.set_missing_host_key_policy(client.MissingHostKeyPolicy)
+ssh_client.set_missing_host_key_policy()
+ssh_client.set_missing_host_key_policy(foo)
+
+# Errors
+ssh_client.set_missing_host_key_policy(client.AutoAddPolicy)
+ssh_client.set_missing_host_key_policy(client.WarningPolicy)
+ssh_client.set_missing_host_key_policy(AutoAddPolicy)
+ssh_client.set_missing_host_key_policy(policy=client.AutoAddPolicy)
+ssh_client.set_missing_host_key_policy(policy=client.WarningPolicy)
+ssh_client.set_missing_host_key_policy(policy=WarningPolicy)
+
+# Unrelated
+set_missing_host_key_policy(client.AutoAddPolicy)
+foo.set_missing_host_key_policy(client.AutoAddPolicy)
+ssh_client = 1
+ssh_client.set_missing_host_key_policy(client.AutoAddPolicy)

crates/ruff/src/checkers/ast/analyze/expression.rs

@@ -583,6 +583,9 @@ pub(crate) fn expression(expr: &Expr, checker: &mut Checker) {
             if checker.enabled(Rule::ParamikoCall) {
                 flake8_bandit::rules::paramiko_call(checker, func);
             }
+            if checker.enabled(Rule::SSHNoHostKeyVerification) {
+                flake8_bandit::rules::ssh_no_host_key_verification(checker, call);
+            }
             if checker.enabled(Rule::LoggingConfigInsecureListen) {
                 flake8_bandit::rules::logging_config_insecure_listen(checker, call);
             }

crates/ruff/src/codes.rs

@@ -598,6 +598,7 @@ pub fn code_to_rule(linter: Linter, code: &str) -> Option<(RuleGroup, Rule)> {
         (Flake8Bandit, "324") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::HashlibInsecureHashFunction),
         (Flake8Bandit, "501") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::RequestWithNoCertValidation),
         (Flake8Bandit, "506") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::UnsafeYAMLLoad),
+        (Flake8Bandit, "507") => (RuleGroup::Preview, rules::flake8_bandit::rules::SSHNoHostKeyVerification),
         (Flake8Bandit, "508") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::SnmpInsecureVersion),
         (Flake8Bandit, "509") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::SnmpWeakCryptography),
         (Flake8Bandit, "601") => (RuleGroup::Unspecified, rules::flake8_bandit::rules::ParamikoCall),

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

@@ -32,6 +32,7 @@ mod tests {
     #[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::SSHNoHostKeyVerification, Path::new("S507.py"))]
     #[test_case(Rule::SnmpInsecureVersion, Path::new("S508.py"))]
     #[test_case(Rule::SnmpWeakCryptography, Path::new("S509.py"))]
     #[test_case(Rule::StartProcessWithAShell, Path::new("S605.py"))]

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

@@ -17,6 +17,7 @@ pub(crate) use request_without_timeout::*;
 pub(crate) use shell_injection::*;
 pub(crate) use snmp_insecure_version::*;
 pub(crate) use snmp_weak_cryptography::*;
+pub(crate) use ssh_no_host_key_verification::*;
 pub(crate) use suspicious_function_call::*;
 pub(crate) use try_except_continue::*;
 pub(crate) use try_except_pass::*;
@@ -41,6 +42,7 @@ mod request_without_timeout;
 mod shell_injection;
 mod snmp_insecure_version;
 mod snmp_weak_cryptography;
+mod ssh_no_host_key_verification;
 mod suspicious_function_call;
 mod try_except_continue;
 mod try_except_pass;

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

@@ -0,0 +1,97 @@
+use ruff_diagnostics::{Diagnostic, Violation};
+use ruff_macros::{derive_message_formats, violation};
+use ruff_python_ast::{Expr, ExprAttribute, ExprCall, Stmt, StmtAssign};
+use ruff_text_size::Ranged;
+
+use crate::checkers::ast::Checker;
+
+/// ## What it does
+/// Checks for uses of policies disabling SSH verification in Paramiko.
+///
+/// ## Why is this bad?
+/// By default, Paramiko checks the identity of remote host when establishing
+/// an SSH connection. Disabling the verification might lead to the client
+/// connecting to a malicious host, without the client knowing.
+///
+/// ## Example
+/// ```python
+/// from paramiko import client
+///
+/// ssh_client = client.SSHClient()
+/// ssh_client.set_missing_host_key_policy(client.AutoAddPolicy)
+/// ```
+///
+/// Use instead:
+/// ```python
+/// from paramiko import client
+///
+/// ssh_client = client.SSHClient()
+/// ssh_client.set_missing_host_key_policy()
+/// ```
+///
+/// ## References
+/// - [Paramiko documentation: set_missing_host_key_policy](https://docs.paramiko.org/en/latest/api/client.html#paramiko.client.SSHClient.set_missing_host_key_policy)
+#[violation]
+pub struct SSHNoHostKeyVerification;
+
+impl Violation for SSHNoHostKeyVerification {
+    #[derive_message_formats]
+    fn message(&self) -> String {
+        format!("Paramiko call with policy set to automatically trust the unknown host key")
+    }
+}
+
+/// S507
+pub(crate) fn ssh_no_host_key_verification(checker: &mut Checker, call: &ExprCall) {
+    let Expr::Attribute(ExprAttribute { attr, value, .. }) = call.func.as_ref() else {
+        return;
+    };
+
+    if attr.as_str() != "set_missing_host_key_policy" {
+        return;
+    }
+
+    let Some(policy_argument) = call.arguments.find_argument("policy", 0) else {
+        return;
+    };
+
+    if !checker
+        .semantic()
+        .resolve_call_path(policy_argument)
+        .is_some_and(|call_path| {
+            matches!(
+                call_path.as_slice(),
+                ["paramiko", "client", "AutoAddPolicy" | "WarningPolicy"]
+            )
+        })
+    {
+        return;
+    }
+
+    let Expr::Name(name) = value.as_ref() else {
+        return;
+    };
+
+    if let Some(binding_id) = checker.semantic().resolve_name(name) {
+        if let Some(Stmt::Assign(StmtAssign { value, .. })) = checker
+            .semantic()
+            .binding(binding_id)
+            .statement(checker.semantic())
+        {
+            if let Expr::Call(ExprCall { func, .. }) = value.as_ref() {
+                if checker
+                    .semantic()
+                    .resolve_call_path(func)
+                    .is_some_and(|call_path| {
+                        matches!(call_path.as_slice(), ["paramiko", "client", "SSHClient"])
+                    })
+                {
+                    checker.diagnostics.push(Diagnostic::new(
+                        SSHNoHostKeyVerification,
+                        policy_argument.range(),
+                    ));
+                }
+            }
+        }
+    };
+}

crates/ruff/src/rules/flake8_bandit/snapshots/ruff__rules__flake8_bandit__tests__S507_S507.py.snap

@@ -0,0 +1,62 @@
+---
+source: crates/ruff/src/rules/flake8_bandit/mod.rs
+---
+S507.py:13:40: S507 Paramiko call with policy set to automatically trust the unknown host key
+   |
+12 | # Errors
+13 | ssh_client.set_missing_host_key_policy(client.AutoAddPolicy)
+   |                                        ^^^^^^^^^^^^^^^^^^^^ S507
+14 | ssh_client.set_missing_host_key_policy(client.WarningPolicy)
+15 | ssh_client.set_missing_host_key_policy(AutoAddPolicy)
+   |
+
+S507.py:14:40: S507 Paramiko call with policy set to automatically trust the unknown host key
+   |
+12 | # Errors
+13 | ssh_client.set_missing_host_key_policy(client.AutoAddPolicy)
+14 | ssh_client.set_missing_host_key_policy(client.WarningPolicy)
+   |                                        ^^^^^^^^^^^^^^^^^^^^ S507
+15 | ssh_client.set_missing_host_key_policy(AutoAddPolicy)
+16 | ssh_client.set_missing_host_key_policy(policy=client.AutoAddPolicy)
+   |
+
+S507.py:15:40: S507 Paramiko call with policy set to automatically trust the unknown host key
+   |
+13 | ssh_client.set_missing_host_key_policy(client.AutoAddPolicy)
+14 | ssh_client.set_missing_host_key_policy(client.WarningPolicy)
+15 | ssh_client.set_missing_host_key_policy(AutoAddPolicy)
+   |                                        ^^^^^^^^^^^^^ S507
+16 | ssh_client.set_missing_host_key_policy(policy=client.AutoAddPolicy)
+17 | ssh_client.set_missing_host_key_policy(policy=client.WarningPolicy)
+   |
+
+S507.py:16:47: S507 Paramiko call with policy set to automatically trust the unknown host key
+   |
+14 | ssh_client.set_missing_host_key_policy(client.WarningPolicy)
+15 | ssh_client.set_missing_host_key_policy(AutoAddPolicy)
+16 | ssh_client.set_missing_host_key_policy(policy=client.AutoAddPolicy)
+   |                                               ^^^^^^^^^^^^^^^^^^^^ S507
+17 | ssh_client.set_missing_host_key_policy(policy=client.WarningPolicy)
+18 | ssh_client.set_missing_host_key_policy(policy=WarningPolicy)
+   |
+
+S507.py:17:47: S507 Paramiko call with policy set to automatically trust the unknown host key
+   |
+15 | ssh_client.set_missing_host_key_policy(AutoAddPolicy)
+16 | ssh_client.set_missing_host_key_policy(policy=client.AutoAddPolicy)
+17 | ssh_client.set_missing_host_key_policy(policy=client.WarningPolicy)
+   |                                               ^^^^^^^^^^^^^^^^^^^^ S507
+18 | ssh_client.set_missing_host_key_policy(policy=WarningPolicy)
+   |
+
+S507.py:18:47: S507 Paramiko call with policy set to automatically trust the unknown host key
+   |
+16 | ssh_client.set_missing_host_key_policy(policy=client.AutoAddPolicy)
+17 | ssh_client.set_missing_host_key_policy(policy=client.WarningPolicy)
+18 | ssh_client.set_missing_host_key_policy(policy=WarningPolicy)
+   |                                               ^^^^^^^^^^^^^ S507
+19 | 
+20 | # Unrelated
+   |
+
+