SONARPY-3974 Fix serializer test and migrated HardcodedCredentials rule to typeV2 to properly raise on pg.DB (#1009)

GitOrigin-RevId: 1336278bad4790a29c57f4c69e363ac82002d996

Author: joke1196

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

@@ -21,10 +21,7 @@
 import java.net.URISyntaxException;
 import java.net.URL;
 import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -74,32 +71,28 @@ public class HardCodedCredentialsCheck extends PythonSubscriptionCheck {
   private static final TypeMatcher ENVIRON_MATCHER = TypeMatchers.isType("os.environ");
   private static final TypeMatcher GET_ENVIRON_MATCHER = TypeMatchers.isType("typing.Mapping.get");
 
+  private static final TypeMatcher SENSITIVE_ARG_2_MATCHER = TypeMatchers.any(
+    TypeMatchers.isType("mysql.connector.connect"),
+    TypeMatchers.isType("mysql.connector.connection.MySQLConnection"),
+    TypeMatchers.isType("pymysql.connect"),
+    TypeMatchers.isType("pymysql.connections.connect"),
+    TypeMatchers.isType("pymysql.connections.Connection"),
+    TypeMatchers.isType("psycopg2.connect"),
+    // pgdb.connect is a module whose connect function has FQN pgdb.connect.connect.
+    // isType can't resolve it through the type table, so use withFQN to match on the FQN directly.
+    TypeMatchers.withFQN("pgdb.connect.connect"));
+
+  private static final TypeMatcher SENSITIVE_ARG_5_MATCHER = TypeMatchers.any(
+    TypeMatchers.isType("pg.DB"),
+    TypeMatchers.isType("pg.connect"));
+
   @RuleProperty(
     key = "credentialWords",
     description = "Comma separated list of words identifying potential credentials",
     defaultValue = DEFAULT_CREDENTIAL_WORDS)
   public String credentialWords = DEFAULT_CREDENTIAL_WORDS;
   private List<Pattern> variablePatterns = null;
   private List<Pattern> literalPatterns = null;
-  private Map<String, Integer> sensitiveArgumentByFQN;
-
-  private Map<String, Integer> sensitiveArgumentByFQN() {
-    if (sensitiveArgumentByFQN == null) {
-      sensitiveArgumentByFQN = new HashMap<>();
-      sensitiveArgumentByFQN.put("mysql.connector.connect", 2);
-      sensitiveArgumentByFQN.put("mysql.connector.connection.MySQLConnection", 2);
-      sensitiveArgumentByFQN.put("pymysql.connect", 2);
-      sensitiveArgumentByFQN.put("pymysql.connections.connect", 2);
-      sensitiveArgumentByFQN.put("pymysql.connections.Connection", 2);
-      sensitiveArgumentByFQN.put("psycopg2.connect", 2);
-      sensitiveArgumentByFQN.put("pgdb.connect", 2);
-      sensitiveArgumentByFQN.put("pgdb.connect.connect", 2);
-      sensitiveArgumentByFQN.put("pg.DB", 5);
-      sensitiveArgumentByFQN.put("pg.connect", 5);
-      sensitiveArgumentByFQN = Collections.unmodifiableMap(sensitiveArgumentByFQN);
-    }
-    return sensitiveArgumentByFQN;
-  }
 
   private Stream<Pattern> variablePatterns() {
     if (variablePatterns == null) {
@@ -175,13 +168,15 @@ private void handleRegularArgument(RegularArgument regularArgument, Subscription
     }
   }
 
-  private void handleCallExpression(CallExpression callExpression, SubscriptionContext ctx) {
+  private static void handleCallExpression(CallExpression callExpression, SubscriptionContext ctx) {
     if (callExpression.arguments().isEmpty()) {
       return;
     }
-    Symbol calleeSymbol = callExpression.calleeSymbol();
-    if (calleeSymbol != null && sensitiveArgumentByFQN().containsKey(calleeSymbol.fullyQualifiedName())) {
-      checkSensitiveArgument(callExpression, sensitiveArgumentByFQN().get(calleeSymbol.fullyQualifiedName()), ctx);
+    Expression callee = callExpression.callee();
+    if (SENSITIVE_ARG_2_MATCHER.isTrueFor(callee, ctx)) {
+      checkSensitiveArgument(callExpression, 2, ctx);
+    } else if (SENSITIVE_ARG_5_MATCHER.isTrueFor(callee, ctx)) {
+      checkSensitiveArgument(callExpression, 5, ctx);
     }
   }
 

python-frontend/src/main/resources/org/sonar/python/types/custom_protobuf/pg.connection.protobuf

@@ -1,7 +1,35 @@
 
-pg.connection�
+pg.connection�
 
-Connectionpg.connection.Connection"*SonarPythonAnalyzerFakeStub.CustomStubBase*�

+Connectionpg.connection.Connection"*SonarPythonAnalyzerFakeStub.CustomStubBase*�
+__init__!pg.connection.Connection.__init__"
+None*>
+self
4
+pg.connection.Connection"pg.connection.Connection*R
+dbname
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*P
+host
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*(
+port

+builtins.int"builtins.int 
*O
+opt
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*P
+user
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*R
+passwd
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*,
+nowait

+builtins.bool"builtins.bool 
*�

 querypg.connection.Connection.query"
 None*>
 self
4

python-frontend/src/main/resources/org/sonar/python/types/custom_protobuf/pg.db.protobuf

@@ -1,6 +1,6 @@
 
-pg.db�
-DBpg.db.DB"*SonarPythonAnalyzerFakeStub.CustomStubBase*�
+pg.db�
+DBpg.db.DB"*SonarPythonAnalyzerFakeStub.CustomStubBase*�
 __init__pg.db.DB.__init__"
 None*
 self

@@ -26,7 +26,9 @@
 passwd
D
 Union[builtins.str,None]

 builtins.str"builtins.str
-None 
*
+None 
*,
+nowait

+builtins.bool"builtins.bool 
*
 querypg.db.DB.query"
 None*
 self


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

@@ -1,7 +1,35 @@
 
-pg�
+pg�
 
-Connectionpg.connection.Connection"*SonarPythonAnalyzerFakeStub.CustomStubBase*�

+Connectionpg.connection.Connection"*SonarPythonAnalyzerFakeStub.CustomStubBase*�
+__init__!pg.connection.Connection.__init__"
+None*>
+self
4
+pg.connection.Connection"pg.connection.Connection*R
+dbname
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*P
+host
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*(
+port

+builtins.int"builtins.int 
*O
+opt
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*P
+user
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*R
+passwd
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*,
+nowait

+builtins.bool"builtins.bool 
*�

 querypg.connection.Connection.query"
 None*>
 self
4
@@ -13,8 +41,8 @@ Connectionpg.connection.Connection"*SonarPythonAnalyzerFakeStub.CustomStubBase
 closepg.connection.Connection.close"
 None*>
 self
4
-pg.connection.Connection"pg.connection.Connection�
-DBpg.db.DB"*SonarPythonAnalyzerFakeStub.CustomStubBase*�
+pg.connection.Connection"pg.connection.Connection�
+DBpg.db.DB"*SonarPythonAnalyzerFakeStub.CustomStubBase*�
 __init__pg.db.DB.__init__"
 None*
 self

@@ -40,7 +68,9 @@ Connectionpg.connection.Connection"*SonarPythonAnalyzerFakeStub.CustomStubBase
 passwd
D
 Union[builtins.str,None]

 builtins.str"builtins.str
-None 
*
+None 
*,
+nowait

+builtins.bool"builtins.bool 
*
 querypg.db.DB.query"
 None*
 self


python-frontend/typeshed_serializer/checksums/custom.checksum

@@ -1,2 +1,2 @@
-0e133142aff9e9d8fee6f4c7205e699f7865502ae3921345895cc814acf0e877
-c137c6a178c97aca6334c983fabbed631672e730554ae8ee66b0aff1949cc90a
+774d53e2cc547be2fbc3e6fc9a9b5b4fadfe72eb7f7fd3444a4566d9110a0a7f
+5c09d20b8cbc107906625c2325a16ebe4e2795c0f866ef3a2c636a0e23488018

python-frontend/typeshed_serializer/resources/custom/pg/connection.pyi

@@ -1,5 +1,16 @@
 from SonarPythonAnalyzerFakeStub import CustomStubBase
+from typing import Optional
 
 class Connection(CustomStubBase):
+    def __init__(
+        self,
+        dbname: Optional[str] = None,
+        host: Optional[str] = None,
+        port: int = -1,
+        opt: Optional[str] = None,
+        user: Optional[str] = None,
+        passwd: Optional[str] = None,
+        nowait: bool = False,
+    ) -> None: ...
     def query(self, command: str, *args) -> None: ...
     def close(self) -> None: ...

python-frontend/typeshed_serializer/resources/custom/pg/db.pyi

@@ -10,6 +10,7 @@ class DB(CustomStubBase):
         opt: Optional[str] = None,
         user: Optional[str] = None,
         passwd: Optional[str] = None,
+        nowait: bool = False,
     ) -> None: ...
     def query(self, command: str, *args) -> None: ...
     def close(self) -> None: ...

python-frontend/typeshed_serializer/tests/test_serializers.py

@@ -74,7 +74,7 @@ def test_custom_stubs_serializer(typeshed_custom_stubs):
         custom_stubs_serializer.serialize()
         assert custom_stubs_serializer.get_build_result.call_count == 1
         # Not every files from "typeshed_custom_stubs" build are serialized, as some are builtins
-        assert symbols.save_module.call_count == 344
+        assert symbols.save_module.call_count == 347
 
 
 def test_importer_serializer():