SONARPY-1710 (#197)

GitOrigin-RevId: 885873307daec29225cd3acbce469ed8fc4786ba

Author: guillaume-dequenne

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

@@ -19,7 +19,10 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
+import java.util.function.Predicate;
 import org.sonar.check.Rule;
 import org.sonar.plugins.python.api.PythonSubscriptionCheck;
 import org.sonar.plugins.python.api.SubscriptionContext;
@@ -29,12 +32,15 @@
 import org.sonar.plugins.python.api.tree.CallExpression;
 import org.sonar.plugins.python.api.tree.Expression;
 import org.sonar.plugins.python.api.tree.HasSymbol;
+import org.sonar.plugins.python.api.tree.Name;
 import org.sonar.plugins.python.api.tree.QualifiedExpression;
 import org.sonar.plugins.python.api.tree.RegularArgument;
 import org.sonar.plugins.python.api.tree.Tree;
 import org.sonar.plugins.python.api.types.v2.PythonType;
 import org.sonar.plugins.python.api.types.v2.TriBool;
 import org.sonar.python.checks.utils.Expressions;
+import org.sonar.python.semantic.v2.SymbolV2;
+import org.sonar.python.semantic.v2.UsageV2;
 import org.sonar.python.tree.TreeUtils;
 import org.sonar.python.types.v2.TypeCheckBuilder;
 
@@ -43,14 +49,25 @@ public class UnverifiedHostnameCheck extends PythonSubscriptionCheck {
 
   private static final String MESSAGE = "Enable server hostname verification on this SSL/TLS connection.";
   private static final String SECONDARY_OPENSSL = "This context does not perform hostname verification.";
+  private static final String SECONDARY_DISABLED = "Hostname verification is disabled here.";
 
   private static final Set<String> SECURE_BY_DEFAULT = new HashSet<>(Arrays.asList("ssl.create_default_context", "ssl._create_default_https_context"));
   private static final Set<String> UNSECURE_BY_DEFAULT = new HashSet<>(Arrays.asList("ssl._create_unverified_context", "ssl._create_stdlib_context"));
 
+  private static final Set<String> UNSAFE_PROTOCOLS = Set.of(
+    "ssl.PROTOCOL_SSLv23",
+    "ssl.PROTOCOL_TLS",
+    "ssl.PROTOCOL_SSLv3",
+    "ssl.PROTOCOL_TLSv1",
+    "ssl.PROTOCOL_TLSv1_1",
+    "ssl.PROTOCOL_TLSv1_2"
+  );
+
   private static Set<String> functionsToCheck;
 
   private TypeCheckBuilder openSSLConnectionTypeCheckBuilder;
   private TypeCheckBuilder openSSLContextTypeCheckBuilder;
+  private TypeCheckBuilder sslContextTypeCheckBuilder;
 
   private static Set<String> functionsToCheck() {
     if (functionsToCheck == null) {
@@ -109,6 +126,7 @@ public void initialize(Context context) {
     context.registerSyntaxNodeConsumer(Tree.Kind.FILE_INPUT, ctx -> {
       openSSLConnectionTypeCheckBuilder = ctx.typeChecker().typeCheckBuilder().isTypeWithName("OpenSSL.SSL.Connection");
       openSSLContextTypeCheckBuilder = ctx.typeChecker().typeCheckBuilder().isInstanceOf("OpenSSL.SSL.Context");
+      sslContextTypeCheckBuilder = ctx.typeChecker().typeCheckBuilder().isTypeWithName("ssl.SSLContext");
     });
   }
 
@@ -119,6 +137,83 @@ private void checkCallExpression(SubscriptionContext ctx) {
       checkSuspiciousCall(callExpression, calleeSymbol, ctx);
     }
     checkOpenSSLConnection(ctx, callExpression);
+    checkSSLContext(ctx, callExpression);
+  }
+
+  private void checkSSLContext(SubscriptionContext ctx, CallExpression callExpression) {
+    Expression callee = callExpression.callee();
+    PythonType pythonType = callee.typeV2();
+
+    if (sslContextTypeCheckBuilder.check(pythonType) != TriBool.TRUE) {
+      return;
+    }
+
+    Optional<Symbol> protocolSymbolOpt = getProtocolSymbol(callExpression);
+    if (protocolSymbolOpt.isEmpty()) {
+      return;
+    }
+
+    Symbol protocolSymbol = protocolSymbolOpt.get();
+    String protocolFQN = protocolSymbol.fullyQualifiedName();
+    if (protocolFQN == null) {
+      return;
+    }
+
+    Tree parent = TreeUtils.firstAncestorOfKind(callExpression, Tree.Kind.ASSIGNMENT_STMT);
+    if (parent == null) {
+      // Direct usage case where context is created and used without assignment
+      if (UNSAFE_PROTOCOLS.contains(protocolFQN)) {
+        ctx.addIssue(callExpression, MESSAGE);
+      }
+      return;
+    }
+
+    Optional<SymbolV2> contextSymbol = getContextSymbol((AssignmentStatement) parent);
+    if (contextSymbol.isEmpty()) {
+      return;
+    }
+
+    if (UNSAFE_PROTOCOLS.contains(protocolFQN)) {
+      // For unsafe protocols, check_hostname should explicitly be set to true
+      Optional<AssignmentStatement> hostnameEnabledAssignment = findCheckHostnameStatement(contextSymbol.get(), Expressions::isTruthy);
+      if (hostnameEnabledAssignment.isEmpty()) {
+        ctx.addIssue(callExpression, MESSAGE);
+      }
+    } else {
+      // For safe protocols, report if check_hostname is explicitly set to false
+      Optional<AssignmentStatement> hostnameDisabledAssignment = findCheckHostnameStatement(contextSymbol.get(), Expressions::isFalsy);
+      hostnameDisabledAssignment.ifPresent(assignment -> {
+        PreciseIssue issue = ctx.addIssue(callee, MESSAGE);
+        issue.secondary(assignment, SECONDARY_DISABLED);
+      });
+    }
+  }
+
+  private static Optional<Symbol> getProtocolSymbol(CallExpression callExpression) {
+    return Optional.ofNullable(TreeUtils.nthArgumentOrKeyword(0, "protocol", callExpression.arguments()))
+      .map(RegularArgument::expression)
+      .filter(HasSymbol.class::isInstance)
+      .map(expr -> ((HasSymbol) expr).symbol());
+  }
+
+  private static Optional<SymbolV2> getContextSymbol(AssignmentStatement assignmentStatement) {
+    return Optional.ofNullable(assignmentStatement.lhsExpressions().get(0).expressions().get(0))
+      .filter(Name.class::isInstance)
+      .map(expr -> ((Name) expr).symbolV2());
+  }
+
+  private static Optional<AssignmentStatement> findCheckHostnameStatement(SymbolV2 contextSymbol, Predicate<Expression> valueCheck) {
+    return contextSymbol.usages().stream()
+      .map(UsageV2::tree)
+      .map(Tree::parent)
+      .filter(QualifiedExpression.class::isInstance)
+      .map(QualifiedExpression.class::cast)
+      .filter(qe -> "check_hostname".equals(qe.name().name()))
+      .map(t -> TreeUtils.firstAncestorOfKind(t, Tree.Kind.ASSIGNMENT_STMT))
+      .filter(Objects::nonNull)
+      .map(AssignmentStatement.class::cast)
+      .filter(a -> valueCheck.test(a.assignedValue()))
+      .findFirst();
   }
 
   private void checkOpenSSLConnection(SubscriptionContext ctx, CallExpression callExpression) {
@@ -145,3 +240,4 @@ private void checkOpenSSLConnection(SubscriptionContext ctx, CallExpression call
     }
   }
 }
+

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

@@ -47,6 +47,43 @@
 if ssl._create_unverified_context() == whatever:
   pass
 
+def ssl_context_constructor():
+  ctx1 = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)  # Noncompliant
+  ctx2 = ssl.SSLContext(ssl.PROTOCOL_TLS)  # Noncompliant
+  ctx3 = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)  # Compliant
+  ctx4 = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)  # Compliant
+
+  foo(ssl.SSLContext(ssl.PROTOCOL_TLSv1_2))  # Noncompliant
+  foo(ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT))  # Compliant
+
+  ctx5 = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
+  ctx5.check_hostname = True  # Compliant
+
+  ctx6 = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)  # Noncompliant
+  ctx6.check_hostname = False
+
+  ctx7 = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) # Noncompliant {{Enable server hostname verification on this SSL/TLS connection.}}
+    #    ^^^^^^^^^^^^^^
+  ctx7.check_hostname = False
+# ^^^^^^^^^^^^^^^^^^^^^^^^^^^< {{Hostname verification is disabled here.}}
+
+
+def edge_cases():
+  nonlocal no_symbol
+  ssl.SSLContext(no_symbol)
+
+  if cond:
+    from foo import bar
+  else:
+    bar = 42
+  ssl.SSLContext(bar)
+
+  something[1] = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2) # FN
+
+  ssl.SSLContext()
+  ssl.SSLContext(abc[42])
+
+
 def pyopenssl_noncompliant():
   import socket
   from OpenSSL import SSL