SONARPY-3112 Rule S7622: Failure to Handle Paginated Responses in boto3 Calls (#426)

GitOrigin-RevId: ae3e141a470b429ef332125a08b9e0ec4eef22ff

Author: maksim-grebeniuk-sonarsource

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

@@ -0,0 +1,60 @@
+/*
+ * SonarQube Python Plugin
+ * Copyright (C) 2011-2025 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the Sonar Source-Available License Version 1, as published by SonarSource SA.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the Sonar Source-Available License for more details.
+ *
+ * You should have received a copy of the Sonar Source-Available License
+ * along with this program; if not, see https://sonarsource.com/license/ssal/
+ */
+package org.sonar.python.checks;
+
+import java.util.Set;
+import org.sonar.check.Rule;
+import org.sonar.plugins.python.api.PythonSubscriptionCheck;
+import org.sonar.plugins.python.api.SubscriptionContext;
+import org.sonar.plugins.python.api.tree.CallExpression;
+import org.sonar.plugins.python.api.tree.Tree;
+import org.sonar.python.tree.TreeUtils;
+import org.sonar.python.types.v2.TypeCheckMap;
+
+@Rule(key = "S7622")
+public class AwsMissingPaginationCheck extends PythonSubscriptionCheck {
+
+  private static final Set<String> SENSITIVE_METHODS_FQNS = Set.of(
+    "botocore.client.BaseClient.list_objects_v2",
+    "botocore.client.BaseClient.scan"
+  );
+
+  private TypeCheckMap<Boolean> sensitiveTypesCheckMap;
+
+  @Override
+  public void initialize(Context context) {
+    context.registerSyntaxNodeConsumer(Tree.Kind.FILE_INPUT, this::initializeCheck);
+    context.registerSyntaxNodeConsumer(Tree.Kind.CALL_EXPR, this::check);
+  }
+
+  private void initializeCheck(SubscriptionContext ctx) {
+    sensitiveTypesCheckMap = new TypeCheckMap<>();
+    SENSITIVE_METHODS_FQNS.stream()
+      .map(fqn -> ctx.typeChecker().typeCheckBuilder().isTypeWithFqn(fqn))
+      .forEach(check -> sensitiveTypesCheckMap.put(check, true));
+
+  }
+
+  private void check(SubscriptionContext ctx) {
+    var callExpression = (CallExpression) ctx.syntaxNode();
+    if (sensitiveTypesCheckMap.containsForType(TreeUtils.inferSingleAssignedExpressionType(callExpression.callee()))) {
+      ctx.addIssue(callExpression, "Use a paginator to retrieve all results from this boto3 operation.");
+    }
+  }
+
+
+}

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

@@ -132,6 +132,7 @@ public Stream<Class<?>> getChecks() {
       AwsLambdaCrossCallCheck.class,
       AwsLambdaReservedEnvironmentVariableCheck.class,
       AwsLambdaReturnValueAreSerializableCheck.class,
+      AwsMissingPaginationCheck.class,
       BackslashInStringCheck.class,
       BackticksUsageCheck.class,
       BareRaiseInFinallyCheck.class,

python-checks/src/main/resources/org/sonar/l10n/py/rules/python/S7622.html

@@ -0,0 +1,52 @@
+<p>This rule raises an issue when <code>boto3</code> operations that support pagination are called without using paginators or manual pagination
+handling.</p>
+<h2>Why is this an issue?</h2>
+<p>Many AWS services use pagination to limit the number of items returned in a single API call. For example, S3’s <code>list_objects_v2()</code>
+returns a maximum of <strong>1000</strong> objects per call, and DynamoDB’s <code>scan()</code> returns up to <strong>1MB</strong> of data per call.
+When you call these operations through boto3 without proper pagination handling, you only receive the first page of results. This means your
+application silently operates on incomplete data, which can lead to incorrect logic and missed operations on resources that exist beyond the first
+page.</p>
+<h3>What is the potential impact?</h3>
+<p>Operating on incomplete data can cause missing critical resources, incorrect business logic based on partial datasets, and security vulnerabilities
+where policies or access changes are not applied to the full resource set. These issues are often silent and difficult to detect in testing
+environments with smaller datasets.</p>
+<h2>How to fix it</h2>
+<p>Use <code>boto3’s</code> built-in paginators to automatically handle pagination and retrieve all results. Paginators provide a simple interface
+that handles the complexity of checking for continuation tokens and making multiple API calls. Replace direct service calls with
+<code>paginator.paginate()</code> calls and iterate through all pages.</p>
+<h3>Code examples</h3>
+<h4>Noncompliant code example</h4>
+<pre data-diff-id="1" data-diff-type="noncompliant">
+import boto3
+s3 = boto3.client("s3")
+
+def lambda_handler(event, context):
+    keys = []
+    response = s3.list_objects_v2(Bucket="my-bucket")  # Noncompliant
+    for obj in response.get("Contents", []):
+        keys.append(obj["Key"])
+    return keys
+</pre>
+<h4>Compliant solution</h4>
+<pre data-diff-id="1" data-diff-type="compliant">
+import boto3
+s3 = boto3.client("s3")
+
+def lambda_handler(event, context):
+    keys = []
+    paginator = s3.get_paginator("list_objects_v2")
+    for page in paginator.paginate(Bucket="my-bucket"):
+        for obj in page.get("Contents", []):
+            keys.append(obj["Key"])
+    return keys
+</pre>
+<h2>Resources</h2>
+<h3>Documentation</h3>
+<ul>
+  <li> <a href="https://boto3.amazonaws.com/v1/documentation/api/latest/guide/paginators.html">Boto3 Paginators Guide</a> </li>
+  <li> <a href="https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/list_objects_v2.html">S3 list_objects_v2 API
+  Reference</a> </li>
+  <li> <a href="https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb/paginator/Query.html">DynamoDB Query Paginator
+  Reference</a> </li>
+</ul>
+

python-checks/src/main/resources/org/sonar/l10n/py/rules/python/S7622.json

@@ -0,0 +1,23 @@
+{
+  "title": "boto3 operations that support pagination should be performed using paginators or manual pagination handling",
+  "type": "CODE_SMELL",
+  "status": "ready",
+  "remediation": {
+    "func": "Constant\/Issue",
+    "constantCost": "5min"
+  },
+  "tags": [
+    "aws"
+  ],
+  "defaultSeverity": "Major",
+  "ruleSpecification": "RSPEC-7622",
+  "sqKey": "S7622",
+  "scope": "All",
+  "quickfix": "unknown",
+  "code": {
+    "impacts": {
+      "MAINTAINABILITY": "MEDIUM"
+    },
+    "attribute": "CONVENTIONAL"
+  }
+}

python-checks/src/main/resources/org/sonar/l10n/py/rules/python/Sonar_way_profile.json

@@ -292,6 +292,7 @@
     "S7519",
     "S7617",
     "S7613",
+    "S7622",
     "S7632"
   ]
 }

python-checks/src/test/java/org/sonar/python/checks/AwsMissingPaginationCheckTest.java

@@ -0,0 +1,28 @@
+/*
+ * SonarQube Python Plugin
+ * Copyright (C) 2011-2025 SonarSource SA
+ * mailto:info AT sonarsource DOT com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the Sonar Source-Available License Version 1, as published by SonarSource SA.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the Sonar Source-Available License for more details.
+ *
+ * You should have received a copy of the Sonar Source-Available License
+ * along with this program; if not, see https://sonarsource.com/license/ssal/
+ */
+package org.sonar.python.checks;
+
+import org.junit.jupiter.api.Test;
+import org.sonar.python.checks.utils.PythonCheckVerifier;
+
+class AwsMissingPaginationCheckTest {
+
+  @Test
+  void test() {
+    PythonCheckVerifier.verify("src/test/resources/checks/awsMissingPagination.py", new AwsMissingPaginationCheck());
+  }
+}

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

@@ -0,0 +1,8 @@
+import boto3
+
+s3 = boto3.client('s3')
+dynamodb = boto3.client('dynamodb')
+
+def foo():
+    response = s3.list_objects_v2(Bucket="my-bucket")  # Noncompliant
+    response = dynamodb.scan(TableName="table")  # Noncompliant

python-frontend/src/main/java/org/sonar/python/types/v2/TypeCheckMap.java

@@ -20,7 +20,6 @@
 import java.util.Map;
 import java.util.Optional;
 import javax.annotation.CheckForNull;
-import org.sonar.plugins.python.api.TriBool;
 import org.sonar.plugins.python.api.types.v2.PythonType;
 
 public class TypeCheckMap<V> {
@@ -55,8 +54,12 @@ public V getForType(PythonType type) {
   public Optional<V> getOptionalForType(PythonType type) {
     return map.entrySet()
       .stream()
-      .filter(entry -> entry.getKey().check(type) == TriBool.TRUE)
+      .filter(entry -> entry.getKey().check(type).isTrue())
       .findFirst()
       .map(Map.Entry::getValue);
   }
+
+  public boolean containsForType(PythonType type) {
+    return getOptionalForType(type).isPresent();
+  }
 }

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

@@ -1,5 +1,5 @@
 
-botocore.client�
+botocore.client�
 
 BaseClientbotocore.client.BaseClient"*SonarPythonAnalyzerFakeStub.CustomStubBase*�
 invoke!botocore.client.BaseClient.invoke"
@@ -11,7 +11,94 @@ BaseClientbotocore.client.BaseClient"*SonarPythonAnalyzerFakeStub.CustomStubBa
 InvocationType

 builtins.str"builtins.str*)
 Payload

-builtins.str"builtins.str*�

+builtins.str"builtins.str*�
+list_objects_v2*botocore.client.BaseClient.list_objects_v2"
+Any*B
+self
8
+botocore.client.BaseClient"botocore.client.BaseClient*(
+Bucket

+builtins.str"builtins.str*U
+	Delimiter
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*:
+EncodingType
&
+Union[Any,None]

+Any
+None 
*S
+MaxKeys
D
+Union[builtins.int,None]

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

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

+builtins.str"builtins.str
+None 
*Y
+
+FetchOwner
G
+Union[builtins.bool,None]

+builtins.bool"builtins.bool
+None 
*V
+
+StartAfter
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*:
+RequestPayer
&
+Union[Any,None]

+Any
+None 
*_
+ExpectedBucketOwner
D
+Union[builtins.str,None]

+builtins.str"builtins.str
+None 
*z
+OptionalObjectAttributes
Z
+Union[builtins.list[Any],None]
,
+builtins.list[Any]
+Any"builtins.list
+None 
*�
+scanbotocore.client.BaseClient.scan"
+Any*B
+self
8
+botocore.client.BaseClient"botocore.client.BaseClient*+
+	TableName

+builtins.str"builtins.str*-
+	IndexName

+builtins.str"builtins.str 
*a
+AttributesToGet
J
+builtins.list[builtins.str]
+builtins.str"builtins.str"builtins.list 
*)
+Limit

+builtins.int"builtins.int 
**
+Select

+builtins.str"builtins.str 
*
+
+ScanFilter

+Any 
*7
+ConditionalOperator

+builtins.str"builtins.str 
* 
+ExclusiveStartKey

+Any 
*:
+ReturnConsumedCapacity

+builtins.str"builtins.str 
*1
+TotalSegments

+builtins.int"builtins.int 
*+
+Segment

+builtins.int"builtins.int 
*8
+ProjectionExpression

+builtins.str"builtins.str 
*4
+FilterExpression

+builtins.str"builtins.str 
*'
+ExpressionAttributeNames

+Any 
*(
+ExpressionAttributeValues

+Any 
*4
+ConsistentRead

+builtins.bool"builtins.bool 
*�

 __annotations__botocore.client.__annotations__W
 builtins.dict[builtins.str,Any]
 builtins.str"builtins.str

python-frontend/src/test/java/org/sonar/python/types/v2/TypeCheckMapTest.java

@@ -44,5 +44,8 @@ void test() {
     Assertions.assertThat(map.getForType(strClassType)).isNull();
     Assertions.assertThat(map.getForType(fooClassType)).isEqualTo(2);
     Assertions.assertThat(map.getForType(PythonType.UNKNOWN)).isNull();
+    Assertions.assertThat(map.containsForType(intClassType)).isTrue();
+    Assertions.assertThat(map.containsForType(strClassType)).isFalse();
+    Assertions.assertThat(map.containsForType(fooClassType)).isTrue();
   }
 }