feat: validate aws:CalledVia condition key service values

parliament/community_auditors/calledvia_services.py

@@ -0,0 +1,58 @@
+"""
+Validate that aws:CalledVia, aws:CalledViaFirst, and aws:CalledViaLast condition
+keys only reference services documented by AWS as supporting CalledVia.
+
+See https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_policies_condition-keys.html#condition-keys-calledvia
+
+Fixes https://github.com/duo-labs/parliament/issues/84
+"""
+from parliament import Policy
+from parliament.misc import make_list
+
+
+# Services documented by AWS as supporting CalledVia
+CALLEDVIA_SUPPORTED_SERVICES = {
+    "athena.amazonaws.com",
+    "cloudformation.amazonaws.com",
+    "databrew.amazonaws.com",
+    "dynamodb.amazonaws.com",
+    "kms.amazonaws.com",
+    "macie.amazonaws.com",
+    "ram.amazonaws.com",
+    "rolesanywhere.amazonaws.com",
+    "s3.amazonaws.com",
+}
+
+CALLEDVIA_CONDITION_KEYS = [
+    "aws:calledvia",
+    "aws:calledviafirst",
+    "aws:calledvialast",
+]
+
+
+def audit(policy: Policy) -> None:
+    for stmt in policy.statements:
+        if "Condition" not in stmt.stmt:
+            continue
+
+        conditions = stmt.stmt["Condition"]
+        for condition in conditions:
+            condition_block = condition[1]
+
+            for block in condition_block:
+                key = block[0]
+                if key.lower() not in CALLEDVIA_CONDITION_KEYS:
+                    continue
+
+                values = []
+                for v in make_list(block[1]):
+                    values.append(v.value)
+
+                for value in values:
+                    if value.lower() not in CALLEDVIA_SUPPORTED_SERVICES:
+                        policy.add_finding(
+                            "INVALID_CALLEDVIA_SERVICE",
+                            detail="Service {} is not in the list of services that support aws:CalledVia. Supported services: {}".format(
+                                value, ", ".join(CALLEDVIA_SUPPORTED_SERVICES)
+                            ),
+                        )

parliament/community_auditors/config_override.yaml

@@ -42,3 +42,9 @@ SINGLE_VALUE_CONDITION_TOO_PERMISSIVE:
   description: Checking a single value conditional key against a set of values results in overly permissive policies.
   severity: MEDIUM
   group: CUSTOM
+
+INVALID_CALLEDVIA_SERVICE:
+  title: Invalid service for aws:CalledVia condition key
+  description: The service referenced in aws:CalledVia, aws:CalledViaFirst, or aws:CalledViaLast is not in the documented list of supported services.
+  severity: MEDIUM
+  group: CUSTOM

parliament/community_auditors/tests/test_calledvia_services.py

@@ -0,0 +1,174 @@
+from parliament import analyze_policy_string
+
+
+class TestCalledViaServices:
+    """Test class for CalledVia service validation auditor"""
+
+    def test_valid_calledvia_service(self):
+        policy_string = """
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Effect": "Allow",
+              "Action": "kms:Decrypt",
+              "Resource": "*",
+              "Condition": {
+                "ForAnyValue:StringEquals": {
+                  "aws:CalledVia": ["cloudformation.amazonaws.com"]
+                }
+              }
+            }
+          ]
+        }
+        """
+        policy = analyze_policy_string(
+            policy_string, include_community_auditors=True
+        )
+        assert "INVALID_CALLEDVIA_SERVICE" not in policy.finding_ids
+
+    def test_invalid_calledvia_service(self):
+        policy_string = """
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Effect": "Allow",
+              "Action": "kms:Decrypt",
+              "Resource": "*",
+              "Condition": {
+                "ForAnyValue:StringEquals": {
+                  "aws:CalledVia": ["ec2.amazonaws.com"]
+                }
+              }
+            }
+          ]
+        }
+        """
+        policy = analyze_policy_string(
+            policy_string, include_community_auditors=True
+        )
+        assert "INVALID_CALLEDVIA_SERVICE" in policy.finding_ids
+
+    def test_valid_calledviafirst(self):
+        policy_string = """
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Effect": "Allow",
+              "Action": "kms:Decrypt",
+              "Resource": "*",
+              "Condition": {
+                "StringEquals": {
+                  "aws:CalledViaFirst": "athena.amazonaws.com"
+                }
+              }
+            }
+          ]
+        }
+        """
+        policy = analyze_policy_string(
+            policy_string, include_community_auditors=True
+        )
+        assert "INVALID_CALLEDVIA_SERVICE" not in policy.finding_ids
+
+    def test_invalid_calledvialast(self):
+        policy_string = """
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Effect": "Allow",
+              "Action": "kms:Decrypt",
+              "Resource": "*",
+              "Condition": {
+                "StringEquals": {
+                  "aws:CalledViaLast": "ec2.amazonaws.com"
+                }
+              }
+            }
+          ]
+        }
+        """
+        policy = analyze_policy_string(
+            policy_string, include_community_auditors=True
+        )
+        assert "INVALID_CALLEDVIA_SERVICE" in policy.finding_ids
+
+    def test_multiple_services_mixed_valid_invalid(self):
+        policy_string = """
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Effect": "Allow",
+              "Action": "kms:Decrypt",
+              "Resource": "*",
+              "Condition": {
+                "ForAnyValue:StringEquals": {
+                  "aws:CalledVia": [
+                    "dynamodb.amazonaws.com",
+                    "lambda.amazonaws.com"
+                  ]
+                }
+              }
+            }
+          ]
+        }
+        """
+        policy = analyze_policy_string(
+            policy_string, include_community_auditors=True
+        )
+        assert "INVALID_CALLEDVIA_SERVICE" in policy.finding_ids
+
+    def test_all_supported_services(self):
+        policy_string = """
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Effect": "Allow",
+              "Action": "kms:Decrypt",
+              "Resource": "*",
+              "Condition": {
+                "ForAnyValue:StringEquals": {
+                  "aws:CalledVia": [
+                    "athena.amazonaws.com",
+                    "cloudformation.amazonaws.com",
+                    "databrew.amazonaws.com",
+                    "dynamodb.amazonaws.com",
+                    "kms.amazonaws.com",
+                    "macie.amazonaws.com",
+                    "ram.amazonaws.com",
+                    "rolesanywhere.amazonaws.com",
+                    "s3.amazonaws.com"
+                  ]
+                }
+              }
+            }
+          ]
+        }
+        """
+        policy = analyze_policy_string(
+            policy_string, include_community_auditors=True
+        )
+        assert "INVALID_CALLEDVIA_SERVICE" not in policy.finding_ids
+
+    def test_no_condition_no_finding(self):
+        policy_string = """
+        {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Effect": "Allow",
+              "Action": "s3:GetObject",
+              "Resource": "arn:aws:s3:::mybucket/*"
+            }
+          ]
+        }
+        """
+        policy = analyze_policy_string(
+            policy_string, include_community_auditors=True
+        )
+        assert "INVALID_CALLEDVIA_SERVICE" not in policy.finding_ids