[flake8-bandit] Add Rule for S508 (snmp insecure version) & `S509…

…` (snmp weak cryptography) (#1771)

ref: #1646

Co-authored-by: messense <messense@icloud.com>
Co-authored-by: Charlie Marsh <charlie.r.marsh@gmail.com>

README.md

@@ -777,6 +777,8 @@ For more, see [flake8-bandit](https://pypi.org/project/flake8-bandit/4.1.1/) on
 | S324 | HashlibInsecureHashFunction | Probable use of insecure hash functions in `hashlib`: "..." |  |
 | S501 | RequestWithNoCertValidation | Probable use of `...` call with `verify=False` disabling SSL certificate checks |  |
 | S506 | UnsafeYAMLLoad | Probable use of unsafe `yaml.load`. Allows instantiation of arbitrary objects. Consider `yaml.safe_load`. |  |
+| S508 | SnmpInsecureVersion | The use of SNMPv1 and SNMPv2 is insecure. Use SNMPv3 if able. |  |
+| S509 | SnmpWeakCryptography | You should not use SNMPv3 without encryption. `noAuthNoPriv` & `authNoPriv` is insecure. |  |
 
 ### flake8-blind-except (BLE)
 

resources/test/fixtures/flake8_bandit/S508.py

@@ -0,0 +1,6 @@
+from pysnmp.hlapi import CommunityData
+
+CommunityData("public", mpModel=0)  # S508
+CommunityData("public", mpModel=1)  # S508
+
+CommunityData("public", mpModel=2)  # OK

resources/test/fixtures/flake8_bandit/S509.py

@@ -0,0 +1,7 @@
+from pysnmp.hlapi import UsmUserData
+
+
+insecure = UsmUserData("securityName")  # S509
+auth_no_priv = UsmUserData("securityName", "authName")  # S509
+
+less_insecure = UsmUserData("securityName", "authName", "privName")  # OK

ruff.schema.json

@@ -1493,6 +1493,8 @@
         "S50",
         "S501",
         "S506",
+        "S508",
+        "S509",
         "SIM",
         "SIM1",
         "SIM10",

src/ast/helpers.rs

@@ -759,6 +759,11 @@ impl<'a> SimpleCallArgs<'a> {
         }
         None
     }
+
+    /// Get the number of positional and keyword arguments used.
+    pub fn len(&self) -> usize {
+        self.args.len() + self.kwargs.len()
+    }
 }
 
 #[cfg(test)]

src/checkers/ast.rs

@@ -1991,6 +1991,28 @@ where
                         self.diagnostics.push(diagnostic);
                     }
                 }
+                if self.settings.enabled.contains(&RuleCode::S508) {
+                    if let Some(diagnostic) = flake8_bandit::rules::snmp_insecure_version(
+                        func,
+                        args,
+                        keywords,
+                        &self.from_imports,
+                        &self.import_aliases,
+                    ) {
+                        self.diagnostics.push(diagnostic);
+                    }
+                }
+                if self.settings.enabled.contains(&RuleCode::S509) {
+                    if let Some(diagnostic) = flake8_bandit::rules::snmp_weak_cryptography(
+                        func,
+                        args,
+                        keywords,
+                        &self.from_imports,
+                        &self.import_aliases,
+                    ) {
+                        self.diagnostics.push(diagnostic);
+                    }
+                }
                 if self.settings.enabled.contains(&RuleCode::S106) {
                     self.diagnostics
                         .extend(flake8_bandit::rules::hardcoded_password_func_arg(keywords));

src/flake8_bandit/mod.rs

@@ -25,6 +25,8 @@ mod tests {
     #[test_case(RuleCode::S324, Path::new("S324.py"); "S324")]
     #[test_case(RuleCode::S501, Path::new("S501.py"); "S501")]
     #[test_case(RuleCode::S506, Path::new("S506.py"); "S506")]
+    #[test_case(RuleCode::S508, Path::new("S508.py"); "S508")]
+    #[test_case(RuleCode::S509, Path::new("S509.py"); "S509")]
     fn rules(rule_code: RuleCode, path: &Path) -> Result<()> {
         let snapshot = format!("{}_{}", rule_code.as_ref(), path.to_string_lossy());
         let diagnostics = test_path(

src/flake8_bandit/rules/mod.rs

@@ -11,6 +11,8 @@ pub use hardcoded_tmp_directory::hardcoded_tmp_directory;
 pub use hashlib_insecure_hash_functions::hashlib_insecure_hash_functions;
 pub use request_with_no_cert_validation::request_with_no_cert_validation;
 pub use request_without_timeout::request_without_timeout;
+pub use snmp_insecure_version::snmp_insecure_version;
+pub use snmp_weak_cryptography::snmp_weak_cryptography;
 pub use unsafe_yaml_load::unsafe_yaml_load;
 
 mod assert_used;
@@ -24,4 +26,6 @@ mod hardcoded_tmp_directory;
 mod hashlib_insecure_hash_functions;
 mod request_with_no_cert_validation;
 mod request_without_timeout;
+mod snmp_insecure_version;
+mod snmp_weak_cryptography;
 mod unsafe_yaml_load;

src/flake8_bandit/rules/snmp_insecure_version.rs

@@ -0,0 +1,40 @@
+use num_traits::{One, Zero};
+use rustc_hash::{FxHashMap, FxHashSet};
+use rustpython_ast::{Expr, ExprKind, Keyword};
+use rustpython_parser::ast::Constant;
+
+use crate::ast::helpers::{collect_call_paths, dealias_call_path, match_call_path, SimpleCallArgs};
+use crate::ast::types::Range;
+use crate::registry::Diagnostic;
+use crate::violations;
+
+/// S508
+pub fn snmp_insecure_version(
+    func: &Expr,
+    args: &[Expr],
+    keywords: &[Keyword],
+    from_imports: &FxHashMap<&str, FxHashSet<&str>>,
+    import_aliases: &FxHashMap<&str, &str>,
+) -> Option<Diagnostic> {
+    let call_path = dealias_call_path(collect_call_paths(func), import_aliases);
+
+    if match_call_path(&call_path, "pysnmp.hlapi", "CommunityData", from_imports) {
+        let call_args = SimpleCallArgs::new(args, keywords);
+
+        if let Some(mp_model_arg) = call_args.get_argument("mpModel", None) {
+            if let ExprKind::Constant {
+                value: Constant::Int(value),
+                ..
+            } = &mp_model_arg.node
+            {
+                if value.is_zero() || value.is_one() {
+                    return Some(Diagnostic::new(
+                        violations::SnmpInsecureVersion,
+                        Range::from_located(mp_model_arg),
+                    ));
+                }
+            }
+        }
+    }
+    None
+}

src/flake8_bandit/rules/snmp_weak_cryptography.rs

@@ -0,0 +1,30 @@
+use rustc_hash::{FxHashMap, FxHashSet};
+use rustpython_ast::{Expr, Keyword};
+
+use crate::ast::helpers::{collect_call_paths, dealias_call_path, match_call_path, SimpleCallArgs};
+use crate::ast::types::Range;
+use crate::registry::Diagnostic;
+use crate::violations;
+
+/// S509
+pub fn snmp_weak_cryptography(
+    func: &Expr,
+    args: &[Expr],
+    keywords: &[Keyword],
+    from_imports: &FxHashMap<&str, FxHashSet<&str>>,
+    import_aliases: &FxHashMap<&str, &str>,
+) -> Option<Diagnostic> {
+    let call_path = dealias_call_path(collect_call_paths(func), import_aliases);
+
+    if match_call_path(&call_path, "pysnmp.hlapi", "UsmUserData", from_imports) {
+        let call_args = SimpleCallArgs::new(args, keywords);
+
+        if call_args.len() < 3 {
+            return Some(Diagnostic::new(
+                violations::SnmpWeakCryptography,
+                Range::from_located(func),
+            ));
+        }
+    }
+    None
+}

src/flake8_bandit/snapshots/ruff__flake8_bandit__tests__S508_S508.py.snap

@@ -0,0 +1,25 @@
+---
+source: src/flake8_bandit/mod.rs
+expression: diagnostics
+---
+- kind:
+    SnmpInsecureVersion: ~
+  location:
+    row: 3
+    column: 32
+  end_location:
+    row: 3
+    column: 33
+  fix: ~
+  parent: ~
+- kind:
+    SnmpInsecureVersion: ~
+  location:
+    row: 4
+    column: 32
+  end_location:
+    row: 4
+    column: 33
+  fix: ~
+  parent: ~
+

src/flake8_bandit/snapshots/ruff__flake8_bandit__tests__S509_S509.py.snap

@@ -0,0 +1,25 @@
+---
+source: src/flake8_bandit/mod.rs
+expression: diagnostics
+---
+- kind:
+    SnmpWeakCryptography: ~
+  location:
+    row: 4
+    column: 11
+  end_location:
+    row: 4
+    column: 22
+  fix: ~
+  parent: ~
+- kind:
+    SnmpWeakCryptography: ~
+  location:
+    row: 5
+    column: 15
+  end_location:
+    row: 5
+    column: 26
+  fix: ~
+  parent: ~
+

src/registry.rs

@@ -417,6 +417,8 @@ define_rule_mapping!(
     S324 => violations::HashlibInsecureHashFunction,
     S501 => violations::RequestWithNoCertValidation,
     S506 => violations::UnsafeYAMLLoad,
+    S508 => violations::SnmpInsecureVersion,
+    S509 => violations::SnmpWeakCryptography,
     // flake8-boolean-trap
     FBT001 => violations::BooleanPositionalArgInFunctionDefinition,
     FBT002 => violations::BooleanDefaultValueInFunctionDefinition,
@@ -1101,6 +1103,8 @@ impl RuleCode {
             RuleCode::S324 => RuleOrigin::Flake8Bandit,
             RuleCode::S501 => RuleOrigin::Flake8Bandit,
             RuleCode::S506 => RuleOrigin::Flake8Bandit,
+            RuleCode::S508 => RuleOrigin::Flake8Bandit,
+            RuleCode::S509 => RuleOrigin::Flake8Bandit,
             // flake8-simplify
             RuleCode::SIM103 => RuleOrigin::Flake8Simplify,
             RuleCode::SIM101 => RuleOrigin::Flake8Simplify,

src/violations.rs

@@ -4810,6 +4810,33 @@ impl Violation for UnsafeYAMLLoad {
     }
 }
 
+define_violation!(
+    pub struct SnmpInsecureVersion;
+);
+impl Violation for SnmpInsecureVersion {
+    fn message(&self) -> String {
+        "The use of SNMPv1 and SNMPv2 is insecure. Use SNMPv3 if able.".to_string()
+    }
+
+    fn placeholder() -> Self {
+        SnmpInsecureVersion
+    }
+}
+
+define_violation!(
+    pub struct SnmpWeakCryptography;
+);
+impl Violation for SnmpWeakCryptography {
+    fn message(&self) -> String {
+        "You should not use SNMPv3 without encryption. `noAuthNoPriv` & `authNoPriv` is insecure."
+            .to_string()
+    }
+
+    fn placeholder() -> Self {
+        SnmpWeakCryptography
+    }
+}
+
 // flake8-boolean-trap
 
 define_violation!(