SONARPY-1157 S4423 and S6308 should not raise on non-call expression matching the fqn (#1223)

Author: nils-werner-sonarsource

python-checks/src/main/java/org/sonar/python/checks/cdk/CdkPredicate.java

@@ -23,15 +23,11 @@
 import java.util.Locale;
 import java.util.Optional;
 import java.util.function.Predicate;
-import org.sonar.plugins.python.api.SubscriptionContext;
-import org.sonar.plugins.python.api.tree.CallExpression;
 import org.sonar.plugins.python.api.tree.Expression;
 import org.sonar.plugins.python.api.tree.Token;
 import org.sonar.plugins.python.api.tree.Tree;
 import org.sonar.python.tree.TreeUtils;
 
-import static org.sonar.python.checks.cdk.CdkUtils.getArgument;
-
 public class CdkPredicate {
 
   private CdkPredicate() {
@@ -84,24 +80,4 @@ public static Predicate<Expression> startsWith(String expected) {
     return expression -> CdkUtils.getString(expression).filter(str -> str.toLowerCase(Locale.ROOT).startsWith(expected)).isPresent();
   }
 
-  // TODO refactor this overloaded predicate method
-  // FIXME we should not raise on a FQN which is not a method call when we want to check for a sensitive method call
-  public static Predicate<Expression> isSensitiveMethod(SubscriptionContext ctx, String methodFqn, String argName, Predicate<Expression> sensitiveValuePredicate) {
-    return expression -> {
-      if (!isFqn(methodFqn).test(expression)) {
-        return false;
-      }
-      if (!expression.is(Tree.Kind.CALL_EXPR)) {
-        return true;
-      }
-
-      Optional<CdkUtils.ExpressionTrace> argTrace = getArgument(ctx, (CallExpression) expression, argName);
-      if (argTrace.isEmpty()) {
-        return true;
-      }
-
-      return argTrace.filter(trace -> trace.hasExpression(sensitiveValuePredicate)).isPresent();
-    };
-  }
-
 }

python-checks/src/main/java/org/sonar/python/checks/cdk/CdkUtils.java

@@ -58,6 +58,13 @@ public static Optional<String> getString(Expression expression) {
     }
   }
 
+  public static Optional<CallExpression> getCall(Expression expression, String fqn) {
+    if (expression.is(Tree.Kind.CALL_EXPR) && CdkPredicate.isFqn(fqn).test(expression)) {
+      return Optional.of((CallExpression) expression);
+    }
+    return Optional.empty();
+  }
+
   /**
    * Resolve a particular argument of a call or get an empty optional if the argument is not set.
    */

python-checks/src/main/java/org/sonar/python/checks/cdk/DisabledESDomainEncryptionCheck.java

@@ -35,8 +35,8 @@
 import org.sonar.plugins.python.api.tree.Tree;
 
 import static org.sonar.python.checks.cdk.CdkPredicate.isFalse;
-import static org.sonar.python.checks.cdk.CdkPredicate.isSensitiveMethod;
 import static org.sonar.python.checks.cdk.CdkUtils.getArgument;
+import static org.sonar.python.checks.cdk.CdkUtils.getCall;
 
 @Rule(key = "S6308")
 public class DisabledESDomainEncryptionCheck extends AbstractCdkResourceCheck {
@@ -78,8 +78,14 @@ private static String unencryptedMessage(String engine) {
     return String.format(UNENCRYPTED_MESSAGE, engine);
   }
 
+  /**
+   * @return Predicate which tests if the expression is the expected object initialization
+   * and if the expected argument is set to false or missing
+   */
   private static Predicate<Expression> isSensitiveOptionObj(SubscriptionContext ctx, String argFqnMethod) {
-    return isSensitiveMethod(ctx, argFqnMethod, ENABLED, isFalse());
+    return expression -> getCall(expression, argFqnMethod)
+      .map(call -> getArgument(ctx, call, ENABLED)).stream()
+      .anyMatch(enabled -> enabled.isEmpty() || enabled.filter(flow -> flow.hasExpression(isFalse())).isPresent());
   }
 
   // TODO : refactor below to use new available method 'isDictionaryWithKeyValue' in AbstractCdkResourceCheck

python-checks/src/main/java/org/sonar/python/checks/cdk/WeakSSLProtocolCheckPart.java

@@ -29,8 +29,9 @@
 import org.sonar.plugins.python.api.tree.Tree;
 
 import static org.sonar.python.checks.cdk.CdkPredicate.isFqn;
-import static org.sonar.python.checks.cdk.CdkPredicate.isSensitiveMethod;
 import static org.sonar.python.checks.cdk.CdkPredicate.isString;
+import static org.sonar.python.checks.cdk.CdkUtils.getArgument;
+import static org.sonar.python.checks.cdk.CdkUtils.getCall;
 
 public class WeakSSLProtocolCheckPart extends AbstractCdkResourceCheck {
   private static final String ENFORCE_MESSAGE = "Change this code to enforce TLS 1.2 or above.";
@@ -75,12 +76,22 @@ private static BiConsumer<SubscriptionContext, CallExpression> checkDomain(Predi
 
   private static BiConsumer<SubscriptionContext, CallExpression> checkCfnDomain(String domainOptionName) {
     return (ctx, callExpression) -> CdkUtils.getArgument(ctx, callExpression, "domain_endpoint_options").ifPresentOrElse(
-      argTrace -> argTrace.addIssueIf(isSensitiveMethod(ctx, domainOptionName, TLS_SECURITY_POLICY, isString(SENSITIVE_TLS_SECURITY_POLICY))
+      argTrace -> argTrace.addIssueIf(isSensitiveOptionObj(ctx, domainOptionName)
         .or(isSensitiveDictionaryTls(ctx)), ENFORCE_MESSAGE),
       () -> ctx.addIssue(callExpression.callee(), OMITTING_MESSAGE)
     );
   }
 
+  /**
+   * @return Predicate which tests if the expression is the expected object initialization
+   * and if the expected argument is set to a sensitive policy or missing
+   */
+  private static Predicate<Expression> isSensitiveOptionObj(SubscriptionContext ctx, String fqn) {
+    return expression -> getCall(expression, fqn)
+      .map(call -> getArgument(ctx, call, TLS_SECURITY_POLICY)).stream()
+      .anyMatch(policy -> policy.isEmpty() || policy.filter(flow -> flow.hasExpression(isString(SENSITIVE_TLS_SECURITY_POLICY))).isPresent());
+  }
+
   private static Predicate<Expression> isSensitiveDictionaryTls(SubscriptionContext ctx) {
     return expression -> Optional.of(expression)
       .filter(expr -> expr.is(Tree.Kind.DICTIONARY_LITERAL)).map(DictionaryLiteral.class::cast)

python-checks/src/test/resources/checks/cdk/disabledESDomainEncryptionCheck.py

@@ -17,8 +17,8 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
 
         # Sensitive test cases
         aws_os.Domain() # NonCompliant{{Omitting encryption_at_rest causes encryption of data at rest to be disabled for this OpenSearch domain. Make sure it is safe here.}}
-        aws_os.Domain(encryption_at_rest=aws_os.EncryptionAtRestOptions) # NonCompliant{{Make sure that using unencrypted OpenSearch domains is safe here.}}
-        aws_os.Domain(encryption_at_rest=aws_os.EncryptionAtRestOptions()) # NonCompliant
+        aws_os.Domain(encryption_at_rest=aws_os.EncryptionAtRestOptions)
+        aws_os.Domain(encryption_at_rest=aws_os.EncryptionAtRestOptions()) # NonCompliant {{Make sure that using unencrypted OpenSearch domains is safe here.}}
         aws_os.Domain(encryption_at_rest=aws_os.EncryptionAtRestOptions(enabled=False)) # NonCompliant{{Make sure that using unencrypted OpenSearch domains is safe here.}}
         aws_os.Domain(encryption_at_rest=encryptionRestOptionMethodBad) # NonCompliant
         aws_os.Domain(encryption_at_rest={"enabled": False}) # NonCompliant
@@ -29,7 +29,7 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
         aws_os.Domain(encryption_at_rest=aws_os.EncryptionAtRestOptions(enabled=not_encrypted)) # NonCompliant
         aws_os.Domain(encryption_at_rest={"enabled": not_encrypted}) # NonCompliant
         aws_es.Domain() # NonCompliant{{Omitting encryption_at_rest causes encryption of data at rest to be disabled for this Elasticsearch domain. Make sure it is safe here.}}
-        aws_es.Domain(encryption_at_rest=aws_es.EncryptionAtRestOptions) # NonCompliant{{Make sure that using unencrypted Elasticsearch domains is safe here.}}
+        aws_es.Domain(encryption_at_rest=aws_es.EncryptionAtRestOptions)
 
         # Compliant test cases
         aws_os.Domain(encryption_at_rest=aws_os.EncryptionAtRestOptions(enabled=True))
@@ -57,7 +57,7 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
         # Sensitive test cases
         aws_os.CfnDomain() # NonCompliant{{Omitting encryption_at_rest_options causes encryption of data at rest to be disabled for this OpenSearch domain. Make sure it is safe here.}}
         aws_os.CfnDomain(encryption_at_rest_options=aws_os.CfnDomain.EncryptionAtRestOptionsProperty()) # NonCompliant{{Make sure that using unencrypted OpenSearch domains is safe here.}}
-        aws_os.CfnDomain(encryption_at_rest_options=aws_os.CfnDomain.EncryptionAtRestOptionsProperty) # NonCompliant
+        aws_os.CfnDomain(encryption_at_rest_options=aws_os.CfnDomain.EncryptionAtRestOptionsProperty)
         aws_os.CfnDomain(encryption_at_rest_options=aws_os.CfnDomain.EncryptionAtRestOptionsProperty(enabled=False)) # NonCompliant{{Make sure that using unencrypted OpenSearch domains is safe here.}}
         aws_os.CfnDomain(encryption_at_rest_options=encryptionRestOptionPropertyMethodBad) # NonCompliant
         aws_os.CfnDomain(encryption_at_rest_options={"enabled": False}) # NonCompliant

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

@@ -47,7 +47,7 @@ def __init__(self, **kwargs) -> None:
 
         # Sensitive test case
         opensearch.CfnDomain() # Noncompliant
-        opensearch.CfnDomain(domain_endpoint_options=opensearch.CfnDomain.DomainEndpointOptionsProperty) # Noncompliant
+        opensearch.CfnDomain(domain_endpoint_options=opensearch.CfnDomain.DomainEndpointOptionsProperty)
         opensearch.CfnDomain(domain_endpoint_options=opensearch.CfnDomain.DomainEndpointOptionsProperty()) # Noncompliant
         opensearch.CfnDomain(domain_endpoint_options=opensearch.CfnDomain.DomainEndpointOptionsProperty(tls_security_policy="Policy-Min-TLS-1-0-2019-07")) # Noncompliant
         opensearch.CfnDomain(domain_endpoint_options=opensearch.CfnDomain.DomainEndpointOptionsProperty(tls_security_policy=str_tls_10)) # Noncompliant