SONARPY-3244: S7608 Should not raise on upload_file and download_file (#438)

GitOrigin-RevId: a0b1281af38b1e0a38b3e4513122c7562e76c2b3

Author: joke1196

python-checks/src/main/java/org/sonar/python/checks/AwsExpectedBucketOwnerCheck.java

@@ -30,6 +30,7 @@
 public class AwsExpectedBucketOwnerCheck extends PythonSubscriptionCheck {
 
   private static final String MESSAGE = "Add the 'ExpectedBucketOwner' parameter to verify S3 bucket ownership.";
+  private static final String MESSAGE_EXTRA_ARGS = "Add the 'ExpectedBucketOwner' to the 'ExtraArgs' parameter to verify S3 bucket ownership.";
   private static final List<String> S3_CLIENT_FQN_PREFIX = List.of("botocore.client.BaseClient", "aiobotocore.client.AioBaseClient");
   private static final List<String> S3_METHODS_REQUIRING_EXPECTED_BUCKET_OWNER = List.of(
     "copy_object",
@@ -56,8 +57,6 @@ public class AwsExpectedBucketOwnerCheck extends PythonSubscriptionCheck {
     "delete_object_tagging",
     "delete_objects",
     "delete_public_access_block",
-    "download_file",
-    "download_fileobj",
     "get_bucket_accelerate_configuration",
     "get_bucket_acl",
     "get_bucket_analytics_configuration",
@@ -134,16 +133,25 @@ public class AwsExpectedBucketOwnerCheck extends PythonSubscriptionCheck {
     "select_object_content",
     "update_bucket_metadata_inventory_table_configuration",
     "update_bucket_metadata_journal_table_configuration",
-    "upload_file",
-    "upload_fileobj",
     "upload_part",
     "upload_part_copy");
 
+  private static final List<String> UPLOAD_DOWNLOAD_FILE_FQN = List.of(
+    "upload_file",
+    "upload_fileobj",
+    "download_file",
+    "download_fileobj");
+
   private static final List<String> FQN_TO_CHECK = S3_CLIENT_FQN_PREFIX.stream()
     .flatMap(prefix -> S3_METHODS_REQUIRING_EXPECTED_BUCKET_OWNER.stream().map(method -> prefix + "." + method))
     .toList();
 
+  private static final List<String> UPLOAD_DOWNLOAD_FQN_TO_CHECK = S3_CLIENT_FQN_PREFIX.stream()
+    .flatMap(prefix -> UPLOAD_DOWNLOAD_FILE_FQN.stream().map(method -> prefix + "." + method))
+    .toList();
+
   private TypeCheckMap<Object> isS3MethodCall;
+  private TypeCheckMap<Object> isUploadDownloadFileMethodCall;
 
   @Override
   public void initialize(Context context) {
@@ -154,17 +162,28 @@ public void initialize(Context context) {
   private void setupTypeChecker(SubscriptionContext ctx) {
     Object object = new Object();
     isS3MethodCall = new TypeCheckMap<>();
+    isUploadDownloadFileMethodCall = new TypeCheckMap<>();
 
     for (var fqn : FQN_TO_CHECK) {
       var typeChecker = ctx.typeChecker().typeCheckBuilder().isTypeWithFqn(fqn);
       isS3MethodCall.put(typeChecker, object);
     }
+    for (var fqn : UPLOAD_DOWNLOAD_FQN_TO_CHECK) {
+      var typeChecker = ctx.typeChecker().typeCheckBuilder().isTypeWithFqn(fqn);
+      isUploadDownloadFileMethodCall.put(typeChecker, object);
+    }
   }
 
   private void checkCall(SubscriptionContext ctx) {
     CallExpression callExpression = (CallExpression) ctx.syntaxNode();
 
     PythonType type = TreeUtils.inferSingleAssignedExpressionType(callExpression.callee());
+
+    if(isUploadDownloadFileMethodCallWithoutExtraArgs(type, callExpression)){
+      ctx.addIssue(callExpression.callee(), MESSAGE_EXTRA_ARGS);
+      return;
+    }
+
     if (isS3MethodCall.getOptionalForType(type).isEmpty()) {
       return;
     }
@@ -179,10 +198,21 @@ private void checkCall(SubscriptionContext ctx) {
 
     ctx.addIssue(callExpression.callee(), MESSAGE);
   }
+
+  private boolean isUploadDownloadFileMethodCallWithoutExtraArgs(PythonType type, CallExpression callExpression){
+    return isUploadDownloadFileMethodCall.getOptionalForType(type)
+      .filter(obj -> !hasExtraArgsParameter(callExpression)).isPresent();
+  }
+
+  private static boolean hasExtraArgsParameter(CallExpression callExpression){
+    return TreeUtils.argumentByKeyword("ExtraArgs", callExpression.arguments()) != null;
+  }
+
   private static boolean hasArgsOrKwargsParams(CallExpression callExpression){
     return callExpression.arguments().stream().anyMatch(arg -> arg.is(Tree.Kind.UNPACKING_EXPR));
   }
 
+
   private static boolean hasExpectedBucketOwnerParameter(CallExpression callExpression) {
     return TreeUtils.argumentByKeyword("ExpectedBucketOwner", callExpression.arguments()) != null;
   }

python-checks/src/test/resources/checks/awsExpectedBucketOwner.py

@@ -78,6 +78,25 @@ def compliant_when_wrapped(*args, **kwargs):
     s3_client.get_object(**kwargs)
     s3_client.get_object(*args, **kwargs)
 
+def upload_download_file():
+    bucket_name = "my-production-bucket"
+    s3_client.upload_file("name", bucket_name, "1") # Noncompliant {{Add the 'ExpectedBucketOwner' to the 'ExtraArgs' parameter to verify S3 bucket ownership.}}
+    s3_client.upload_fileobj("name", bucket_name, "1") # Noncompliant
+    s3_client.download_file("name", bucket_name, "1") # Noncompliant
+    s3_client.download_fileobj("name", bucket_name, "1") # Noncompliant
+
+    # ExpectedBucketOwner is not an argument download_fileobj
+    s3_client.download_fileobj("name", bucket_name, "1", ExpectedBucketOwner="") # Noncompliant 
+
+    s3_client.upload_file("name", bucket_name, "1", ExtraArgs={}) # Compliant
+    s3_client.upload_fileobj("name", bucket_name, "1", ExtraArgs=unknown()) # Compliant
+    dict_config = {}
+    s3_client.download_file("name", bucket_name, "1", ExtraArgs=dict_config) # Compliant
+    s3_client.download_fileobj("name", bucket_name, "1", ExtraArgs={"ExpectedBucketOwner":None}) # Compliant
+
+    # We should still raise on the other FQN
+    s3_client.get_object(Bucket=bucket_name, ExtraArgs={}) # Noncompliant 
+
 async def async_function():
     bucket_name = "my-production-bucket"
     expected_owner = "123456789012"