SONARPY-1709 S5527 Add PyOpenSSL support (#194)

GitOrigin-RevId: 8fc6a751314e59aeb12d055deb919099ff99ab67

Author: guillaume-dequenne

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

@@ -30,20 +30,28 @@
 import org.sonar.plugins.python.api.tree.Expression;
 import org.sonar.plugins.python.api.tree.HasSymbol;
 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.tree.TreeUtils;
+import org.sonar.python.types.v2.TypeCheckBuilder;
 
 @Rule(key = "S5527")
 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 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 Set<String> functionsToCheck;
 
+  private TypeCheckBuilder openSSLConnectionTypeCheckBuilder;
+  private TypeCheckBuilder openSSLContextTypeCheckBuilder;
+
   private static Set<String> functionsToCheck() {
     if (functionsToCheck == null) {
       functionsToCheck = new HashSet<>();
@@ -97,17 +105,43 @@ private static boolean opensUnsecureConnection(Symbol calleeSymbol, CallExpressi
 
   @Override
   public void initialize(Context context) {
-    context.registerSyntaxNodeConsumer(Tree.Kind.CALL_EXPR, UnverifiedHostnameCheck::checkCallExpression);
+    context.registerSyntaxNodeConsumer(Tree.Kind.CALL_EXPR, this::checkCallExpression);
+    context.registerSyntaxNodeConsumer(Tree.Kind.FILE_INPUT, ctx -> {
+      openSSLConnectionTypeCheckBuilder = ctx.typeChecker().typeCheckBuilder().isTypeWithName("OpenSSL.SSL.Connection");
+      openSSLContextTypeCheckBuilder = ctx.typeChecker().typeCheckBuilder().isInstanceOf("OpenSSL.SSL.Context");
+    });
   }
 
-  private static void checkCallExpression(SubscriptionContext ctx) {
+  private void checkCallExpression(SubscriptionContext ctx) {
     CallExpression callExpression = (CallExpression) ctx.syntaxNode();
     Symbol calleeSymbol = callExpression.calleeSymbol();
-    if (calleeSymbol == null) {
+    if (calleeSymbol != null && functionsToCheck().contains(calleeSymbol.fullyQualifiedName())) {
+      checkSuspiciousCall(callExpression, calleeSymbol, ctx);
+    }
+    checkOpenSSLConnection(ctx, callExpression);
+  }
+
+  private void checkOpenSSLConnection(SubscriptionContext ctx, CallExpression callExpression) {
+    Expression callee = callExpression.callee();
+    PythonType pythonType = callee.typeV2();
+
+    if (openSSLConnectionTypeCheckBuilder.check(pythonType) != TriBool.TRUE) {
       return;
     }
-    if (functionsToCheck().contains(calleeSymbol.fullyQualifiedName())) {
-      checkSuspiciousCall(callExpression, calleeSymbol, ctx);
+
+    RegularArgument contextArg = TreeUtils.nthArgumentOrKeyword(0, "context", callExpression.arguments());
+    if (contextArg == null) {
+      return;
+    }
+
+    Expression contextExpr = contextArg.expression();
+    PythonType contextType = contextExpr.typeV2();
+    if (openSSLContextTypeCheckBuilder.check(contextType) == TriBool.TRUE) {
+      PreciseIssue issue = ctx.addIssue(callee, MESSAGE);
+      Expressions.ifNameGetSingleAssignedNonNameValue(contextExpr)
+        .ifPresentOrElse(e -> issue.secondary(e, SECONDARY_OPENSSL),
+          () -> issue.secondary(contextExpr, SECONDARY_OPENSSL)
+        );
     }
   }
 }

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

@@ -46,3 +46,28 @@
 
 if ssl._create_unverified_context() == whatever:
   pass
+
+def pyopenssl_noncompliant():
+  import socket
+  from OpenSSL import SSL
+
+  ctx = SSL.Context(SSL.TLSv1_2_METHOD)
+  #     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^> {{This context does not perform hostname verification.}}
+  ctx.set_verify(SSL.VERIFY_PEER)
+
+  conn = SSL.Connection(ctx, socket.socket(socket.AF_INET, socket.SOCK_STREAM)) # Noncompliant {{Enable server hostname verification on this SSL/TLS connection.}}
+  #      ^^^^^^^^^^^^^^
+
+  SSL.Connection(SSL.Context(SSL.TLSv1_2_METHOD)) # Noncompliant
+# ^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^<
+
+  if foo:
+    ctx2 = SSL.Context(SSL.TLSv1_2_METHOD)
+  else:
+    ctx2 = SSL.Context(SSL.TLSv1_3_METHOD)
+  SSL.Connection(ctx2) # Noncompliant
+# ^^^^^^^^^^^^^^ ^^^^<
+
+  ctx3 = foo()
+  SSL.Connection() # OK, no context argument
+  SSL.Connection(ctx3)

python-frontend/src/main/resources/org/sonar/python/types/custom_protobuf/OpenSSL.SSL.protobuf

@@ -9,7 +9,9 @@ set_verifyOpenSSL.SSL.Context.set_verify"
 args
 Any*
 kwargs
-Any*�

+AnyP
+
+ConnectionOpenSSL.SSL.Connection"*SonarPythonAnalyzerFakeStub.CustomStubBase*�

 __annotations__OpenSSL.SSL.__annotations__W
 builtins.dict[builtins.str,Any]
 builtins.str"builtins.str

python-frontend/typeshed_serializer/checksum

@@ -1,2 +1,2 @@
-28d817d04de5e4f99762476c6c950e6f0c9c5fa91af954afa9b1d44eab3102a5
-bbcc49c75d53b7cc9fd2135583b435385795dddb5696ac8094ed88440da841de
+cb2ccbe2913935c2c4dfdb7d09b8f40c6d2a2ec9840a74573f0cc6adb6ab6e69
+3c1a63e0fbd63e97ebbbd0eb13d3c27c3f9051aaac329868be6e90ccdc31546c

python-frontend/typeshed_serializer/resources/custom/OpenSSL/SSL.pyi

@@ -6,3 +6,7 @@ VERIFY_NONE: int
 
 class Context(CustomStubBase):
     def set_verify(self, *args, **kwargs) -> None: ...
+
+
+class Connection(CustomStubBase):
+    ...