SONARPY-1552 Rule S5332: Raise an issue on server_bind calls. (#1640)

Author: joke1196

python-checks/src/main/java/org/sonar/python/checks/hotspots/ClearTextProtocolsCheck.java

@@ -22,7 +22,6 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -41,7 +40,6 @@
 import org.sonar.plugins.python.api.tree.CallExpression;
 import org.sonar.plugins.python.api.tree.ClassDef;
 import org.sonar.plugins.python.api.tree.Expression;
-import org.sonar.plugins.python.api.tree.FunctionDef;
 import org.sonar.plugins.python.api.tree.HasSymbol;
 import org.sonar.plugins.python.api.tree.Name;
 import org.sonar.plugins.python.api.tree.QualifiedExpression;
@@ -56,17 +54,15 @@
 public class ClearTextProtocolsCheck extends PythonSubscriptionCheck {
   private static final List<String> SENSITIVE_PROTOCOLS = Arrays.asList("http://", "ftp://", "telnet://");
   private static final Pattern LOOPBACK = Pattern.compile("localhost|127(?:\\.[0-9]+){0,2}\\.[0-9]+$|^(?:0*\\:)*?:?0*1", Pattern.CASE_INSENSITIVE);
-  private static final Map<String, String> ALTERNATIVES = new HashMap<>();
-  private static final String SENSITIVE_HTTP_SERVER_CALL = "socketserver.BaseServer.serve_forever";
+  private static final Map<String, String> ALTERNATIVES = Map.of(
+    "http", "https",
+    "ftp", "sftp, scp or ftps",
+    "telnet", "ssh");
+  private static final String SENSITIVE_HTTP_SERVER_START_FQN = "socketserver.BaseServer.serve_forever";
+  private static final String SENSITIVE_HTTP_SERVER_BIND_FQN = "socketserver.BaseServer.server_bind";
+  private static final Set<String> SENSITIVE_HTTP_SERVER_METHOD_NAMES = Set.of("serve_forever", "server_bind");
   private static final Set<String> SENSITIVE_HTTP_SERVER_CLASSES = Set.of("http.server.HTTPServer", "http.server.ThreadingHTTPServer");
 
-
-  static {
-    ALTERNATIVES.put("http", "https");
-    ALTERNATIVES.put("ftp", "sftp, scp or ftps");
-    ALTERNATIVES.put("telnet", "ssh");
-  }
-
   @Override
   public void initialize(Context context) {
     context.registerSyntaxNodeConsumer(Tree.Kind.STRING_ELEMENT, ctx -> {
@@ -89,28 +85,27 @@ public void initialize(Context context) {
 
     context.registerSyntaxNodeConsumer(Tree.Kind.QUALIFIED_EXPR, ClearTextProtocolsCheck::checkServerCallFromSuper);
 
-    context.registerSyntaxNodeConsumer(Tree.Kind.FUNCDEF, ClearTextProtocolsCheck::checkServerBindOverrides);
+    context.registerSyntaxNodeConsumer(Tree.Kind.CALL_EXPR, ClearTextProtocolsCheck::checkServerBindCalls);
+
     new ClearTextProtocolsCheckPart().initialize(context);
   }
 
   private static void checkServerCallFromSuper(SubscriptionContext ctx) {
     QualifiedExpression qualifiedExpression = (QualifiedExpression) ctx.syntaxNode();
     Optional.of(qualifiedExpression)
-      .filter(qe -> isName("serve_forever", qe.name()) && isCallToSensitiveSuperClass(qe))
+      .filter(qe -> SENSITIVE_HTTP_SERVER_METHOD_NAMES.contains(qe.name().name()))
+      .filter(ClearTextProtocolsCheck::isCallToSensitiveSuperClass)
       .map(qe -> TreeUtils.firstAncestorOfKind(qe, Tree.Kind.CALL_EXPR))
       .flatMap(TreeUtils.toOptionalInstanceOfMapper(CallExpression.class))
       .ifPresent(ce -> ctx.addIssue(ce, message("http")));
   }
 
-  private static void checkServerBindOverrides(SubscriptionContext ctx) {
-    FunctionDef funcDef = (FunctionDef) ctx.syntaxNode();
-    Optional.of(funcDef)
-      .filter(fd -> isName("server_bind", fd.name()) && isParentClassExtendingSensitiveClass(funcDef))
-      .ifPresent(fd -> ctx.addIssue(fd.defKeyword(), fd.rightPar(), message("http")));
-  }
-
-  private static boolean isName(String nameToCheck, Name name) {
-    return nameToCheck.equals(name.name());
+  private static void checkServerBindCalls(SubscriptionContext ctx) {
+    CallExpression callExpression = (CallExpression) ctx.syntaxNode();
+    Optional.ofNullable(callExpression.calleeSymbol())
+      .map(Symbol::fullyQualifiedName)
+      .filter(fqn -> SENSITIVE_HTTP_SERVER_BIND_FQN.equals(fqn) && isParentClassExtendingSensitiveClass(callExpression))
+      .ifPresent(fqn -> ctx.addIssue(callExpression, message("http")));
   }
 
   private static boolean isCallToSensitiveSuperClass(QualifiedExpression expression) {
@@ -205,7 +200,7 @@ private static Optional<String> isUnsafeLib(String qualifiedName) {
     if ("ftplib.FTP".equals(qualifiedName)) {
       return Optional.of("ftp");
     }
-    if (SENSITIVE_HTTP_SERVER_CALL.equals(qualifiedName)) {
+    if (SENSITIVE_HTTP_SERVER_START_FQN.equals(qualifiedName)) {
       return Optional.of("http");
     }
     return Optional.empty();

python-checks/src/test/resources/checks/hotspots/clearTextProtocols.py

@@ -156,32 +156,38 @@ def python_web_server_noncompliant():
 
     class MyServer(HTTPServer):
         def run(self):
+            HTTPServer.server_bind(self)  # Noncompliant
+        #   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
             super(self).serve_forever()  # Noncompliant
         #   ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-        def server_bind(self):  # Noncompliant
-#       ^^^^^^^^^^^^^^^^^^^^^
-            pass
-
     my_server = MyServer(('0.0.0.0', 8080), SimpleHTTPRequestHandler)
     my_server.serve_forever()  # Noncompliant
 #   ^^^^^^^^^^^^^^^^^^^^^^^^^
 
     class MyThreadingServer(ThreadingHTTPServer):
         def run(self):
+            super().server_bind()  # Noncompliant
+        #   ^^^^^^^^^^^^^^^^^^^^^
             super(self).serve_forever()  # Noncompliant
         #   ^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-        def server_bind(self):  # Noncompliant
-#       ^^^^^^^^^^^^^^^^^^^^^
-            pass
-
     my_threading_server = MyThreadingServer(('0.0.0.0', 8080), SimpleHTTPRequestHandler)
     my_threading_server.serve_forever()  # Noncompliant
 
 
 def python_web_server_compliant(ok_server):
 
+    from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
+    from http.server import HTTPServer as PythonServer
+
+    http_server = PythonServer(('0.0.0.0', 8080), SimpleHTTPRequestHandler)
+    http_server.server_bind()  # Compliant we do not raise here as the call to server bind was done already in the constructor
+
+    class MyThreadingServer(ThreadingHTTPServer):
+        def server_bind(self):  # Compliant
+            pass
+
     ok_server.serve_forever()  # Compliant
     ok_server.server_bind()
 
@@ -197,7 +203,7 @@ def run(self):
             super(self).serve_forever()  # Compliant
 
         def server_bind(self):
-            pass
+            HTTPServer.server_bind(self)
 
     my_server = MyServer()
     my_server.serve_forever()