Repository: getmoto/moto
Pull request #8163 (2024-09-26 20:57:04)

WAFv2: associate_web_acl() now allows any resource to be associated


Patch:
diff --git a/docs/docs/services/wafv2.rst b/docs/docs/services/wafv2.rst
index d1ab9b4e1..f2f7f95f4 100644
--- a/docs/docs/services/wafv2.rst
+++ b/docs/docs/services/wafv2.rst
@@ -17,10 +17,6 @@ wafv2
 |start-h3| Implemented features for this service |end-h3|
 
 - [X] associate_web_acl
-  
-        Only APIGateway Stages can be associated at the moment.
-        
-
 - [ ] check_capacity
 - [ ] create_api_key
 - [X] create_ip_set
diff --git a/moto/ec2/models/elastic_block_store.py b/moto/ec2/models/elastic_block_store.py
index cee0558d3..e0c35e18e 100644
--- a/moto/ec2/models/elastic_block_store.py
+++ b/moto/ec2/models/elastic_block_store.py
@@ -269,6 +269,7 @@ class EBSBackend:
         self.volumes: Dict[str, Volume] = {}
         self.attachments: Dict[str, VolumeAttachment] = {}
         self.snapshots: Dict[str, Snapshot] = {}
+        self.default_kms_key_id: str = ""
 
     def create_volume(
         self,
@@ -284,7 +285,9 @@ class EBSBackend:
         if kms_key_id and not encrypted:
             raise InvalidParameterDependency("KmsKeyId", "Encrypted")
         if encrypted and not kms_key_id:
-            kms_key_id = self._get_default_encryption_key()
+            if not self.default_kms_key_id:
+                self._create_default_encryption_key()
+            kms_key_id = self.default_kms_key_id
         if volume_type in IOPS_REQUIRED_VOLUME_TYPES and not iops:
             raise InvalidParameterDependency("VolumeType", "Iops")
         elif volume_type == "gp3" and not iops:
@@ -546,7 +549,7 @@ class EBSBackend:
         else:
             snapshot.create_volume_permission_groups.difference_update(groups)  # type: ignore[arg-type]
 
-    def _get_default_encryption_key(self) -> str:
+    def _create_default_encryption_key(self) -> str:
         # https://aws.amazon.com/kms/features/#AWS_Service_Integration
         # An AWS managed CMK is created automatically when you first create
         # an encrypted resource using an AWS service integrated with KMS.
@@ -564,4 +567,18 @@ class EBSBackend:
             )
             kms.create_alias(key.id, ebs_alias)
         ebs_key = kms.describe_key(ebs_alias)
+        self.default_kms_key_id = ebs_key.arn
         return ebs_key.arn
+
+    def modify_ebs_default_kms_key_id(self, kms_key_id: str) -> str:
+        # If this parameter is not specified, the standard KMS key for Amazon EBS is used.
+        if not kms_key_id:
+            return self._create_default_encryption_key()
+
+        # Determine if the value is a key ID, alias, key ARN, or alias ARN. All are valid possibilities.
+        from moto.kms import kms_backends
+
+        kms = kms_backends[self.account_id][self.region_name]  # type: ignore[attr-defined]
+        kms_key = kms.describe_key(kms_key_id)
+        self.default_kms_key_id = kms_key.arn
+        return kms_key.arn
diff --git a/moto/ec2/responses/elastic_block_store.py b/moto/ec2/responses/elastic_block_store.py
index 968b4629a..d095c1d30 100644
--- a/moto/ec2/responses/elastic_block_store.py
+++ b/moto/ec2/responses/elastic_block_store.py
@@ -219,6 +219,13 @@ class ElasticBlockStore(EC2BaseResponse):
             "ElasticBlockStore.reset_snapshot_attribute is not yet implemented"
         )
 
+    def modify_ebs_default_kms_key_id(self) -> str:
+        self.error_on_dryrun()
+        kms_key_id = self._get_param("KmsKeyId")
+        new_default_kms_arn = self.ec2_backend.modify_ebs_default_kms_key_id(kms_key_id)
+        template = self.response_template(MODIFY_EBS_DEFAULT_KMS_KEY_ID_RESPONSE)
+        return template.render(kmsKeyId=new_default_kms_arn)
+
 
 CREATE_VOLUME_RESPONSE = """<CreateVolumeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
@@ -499,3 +506,11 @@ DESCRIBE_VOLUMES_MODIFICATIONS_RESPONSE = """
       {% endfor %}
     </volumeModificationSet>
 </DescribeVolumesModificationsResponse>"""
+
+
+MODIFY_EBS_DEFAULT_KMS_KEY_ID_RESPONSE = """
+<ModifyEbsDefaultKmsKeyIdResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
+    <requestId>666d2944-9276-4d6a-be12-1f4ada972fd8</requestId>
+    <kmsKeyId>{{ kmsKeyId }}</kmsKeyId>
+</ModifyEbsDefaultKmsKeyIdResponse>
+"""
diff --git a/moto/fsx/models.py b/moto/fsx/models.py
index 554974ec5..abceebbd1 100644
--- a/moto/fsx/models.py
+++ b/moto/fsx/models.py
@@ -1,7 +1,7 @@
 """FSxBackend class with methods for supported APIs."""
 
-from datetime import datetime
 from typing import Any, Dict, List, Optional, Tuple
+from uuid import uuid4
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
@@ -36,7 +36,7 @@ class FileSystem(BaseModel):
         ontap_configuration: Optional[Dict[str, Any]],
         open_zfs_configuration: Optional[Dict[str, Any]],
     ) -> None:
-        self.file_system_id = f"fs-moto{datetime.now().strftime('%Y%m%d%H%M%S')}"
+        self.file_system_id = f"fs-{uuid4().hex[:8]}"
         self.file_system_type = file_system_type
         if self.file_system_type not in FileSystemType.list_values():
             raise ValueError(f"Invalid FileSystemType: {self.file_system_type}")
diff --git a/moto/wafv2/models.py b/moto/wafv2/models.py
index 8058cb4bf..5623e8138 100644
--- a/moto/wafv2/models.py
+++ b/moto/wafv2/models.py
@@ -54,6 +54,7 @@ class FakeWebACL(BaseModel):
         self.visibility_config = visibility_config
         self.default_action = default_action
         self.lock_token = str(mock_random.uuid4())[0:6]
+        self.associated_resources: List[str] = []
 
     def update(
         self,
@@ -168,25 +169,28 @@ class WAFV2Backend(BaseBackend):
         # TODO: self.load_balancers = OrderedDict()
 
     def associate_web_acl(self, web_acl_arn: str, resource_arn: str) -> None:
-        """
-        Only APIGateway Stages can be associated at the moment.
-        """
-        if web_acl_arn not in self.wacls:
+        web_acl = self.wacls.get(web_acl_arn)
+        if not web_acl:
             raise WAFNonexistentItemException
+        web_acl.associated_resources.append(resource_arn)
+        # Special Case - APIGateway wants to know about the WebACL it's associated to
         stage = self._find_apigw_stage(resource_arn)
         if stage:
             stage.web_acl_arn = web_acl_arn
 
     def disassociate_web_acl(self, resource_arn: str) -> None:
+        for web_acl in self.wacls.values():
+            if resource_arn in web_acl.associated_resources:
+                web_acl.associated_resources.remove(resource_arn)
+                break
         stage = self._find_apigw_stage(resource_arn)
         if stage:
             stage.web_acl_arn = None
 
     def get_web_acl_for_resource(self, resource_arn: str) -> Optional[FakeWebACL]:
-        stage = self._find_apigw_stage(resource_arn)
-        if stage and stage.web_acl_arn is not None:
-            wacl_arn = stage.web_acl_arn
-            return self.wacls.get(wacl_arn)
+        for wacl in self.wacls.values():
+            if resource_arn in wacl.associated_resources:
+                return wacl
         return None
 
     def _find_apigw_stage(self, resource_arn: str) -> Optional["Stage"]:  # type: ignore
diff --git a/other_langs/terraform/wafv2/cognito.tf b/other_langs/terraform/wafv2/cognito.tf
new file mode 100644
index 000000000..d32200e6d
--- /dev/null
+++ b/other_langs/terraform/wafv2/cognito.tf
@@ -0,0 +1,61 @@
+# Create a Cognito User Pool (this assumes you need a new one)
+resource "aws_cognito_user_pool" "example" {
+  name = "example-user-pool"
+}
+
+# Create a WAFv2 Web ACL
+resource "aws_wafv2_web_acl" "example" {
+  name        = "example-web-acl"
+  description = "Example WAFv2 Web ACL for Cognito User Pool"
+  scope       = "REGIONAL" # Since Cognito is a regional service
+  default_action {
+    allow {}
+  }
+
+  visibility_config {
+    cloudwatch_metrics_enabled = true
+    metric_name                = "exampleWebACL"
+    sampled_requests_enabled   = true
+  }
+
+  rule {
+    name     = "block-bad-ips"
+    priority = 1
+
+    action {
+      block {}
+    }
+
+    statement {
+      ip_set_reference_statement {
+        arn = aws_wafv2_ip_set.example.arn
+      }
+    }
+
+    visibility_config {
+      cloudwatch_metrics_enabled = true
+      metric_name                = "blockBadIps"
+      sampled_requests_enabled   = true
+    }
+  }
+}
+
+# Create an IP Set for the WAFv2 rule
+resource "aws_wafv2_ip_set" "example" {
+  name        = "example-ip-set"
+  description = "Example IP Set for bad actors"
+  scope       = "REGIONAL"
+
+  ip_address_version = "IPV4"
+
+  addresses = [
+    "192.0.2.0/24",  # Example IP addresses to block
+    "198.51.100.0/24"
+  ]
+}
+
+# Associate the WAFv2 Web ACL with the Cognito User Pool
+resource "aws_wafv2_web_acl_association" "example" {
+  resource_arn = aws_cognito_user_pool.example.arn
+  web_acl_arn  = aws_wafv2_web_acl.example.arn
+}
\ No newline at end of file
diff --git a/other_langs/terraform/wafv2/provider.tf b/other_langs/terraform/wafv2/provider.tf
new file mode 100644
index 000000000..ee528ec15
--- /dev/null
+++ b/other_langs/terraform/wafv2/provider.tf
@@ -0,0 +1,26 @@
+provider "aws" {
+  region                      = "us-east-1"
+  s3_use_path_style           = true
+  skip_credentials_validation = true
+  skip_metadata_api_check     = true
+  skip_requesting_account_id  = true
+
+  endpoints {
+    cognitoidp     = "http://localhost:5000"
+    iam            = "http://localhost:5000"
+    wafv2          = "http://localhost:5000"
+  }
+
+  access_key = "my-access-key"
+  secret_key = "my-secret-key"
+}
+
+terraform {
+  required_providers {
+
+    aws = {
+      source  = "hashicorp/aws"
+      version = "4.67.0"
+    }
+  }
+}
\ No newline at end of file


Test patch:
diff --git a/.github/workflows/tests_terraform_examples.yml b/.github/workflows/tests_terraform_examples.yml
index 82cf15891..672afce50 100644
--- a/.github/workflows/tests_terraform_examples.yml
+++ b/.github/workflows/tests_terraform_examples.yml
@@ -11,7 +11,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        service: ["acm", "autoscaling", "awslambda", "cloudfront", "codebuild", "ds", "elb", "iam", "rds", "route53", "sqs"]
+        service: ["acm", "autoscaling", "awslambda", "cloudfront", "codebuild", "ds", "elb", "iam", "rds", "route53", "sqs", "wafv2"]
 
     steps:
     - uses: actions/checkout@v4
diff --git a/tests/test_ec2/test_elastic_block_store.py b/tests/test_ec2/test_elastic_block_store.py
index 1fdd4d75b..a5636ed2f 100644
--- a/tests/test_ec2/test_elastic_block_store.py
+++ b/tests/test_ec2/test_elastic_block_store.py
@@ -1145,3 +1145,56 @@ def test_create_volume_with_throughput_fails():
         )
     assert ex.value.response["Error"]["Code"] == "InvalidParameterDependency"
     assert "Throughput" in ex.value.response["Error"]["Message"]
+
+
+@mock_aws
+def test_modify_ebs_default_kms_key_id():
+    new_default_alias = "alias/my-new-ebs-default"
+    kms = boto3.client("kms", region_name="us-east-1")
+    ec2 = boto3.client("ec2", region_name="us-east-1")
+
+    # Get the original default key
+    original_default_ebs_key_arn = kms.describe_key(KeyId="alias/aws/ebs")[
+        "KeyMetadata"
+    ]["Arn"]
+
+    # Create a new default key
+    new_default_key = kms.create_key(Description="test ebs key")["KeyMetadata"]
+    kms.create_alias(AliasName=new_default_alias, TargetKeyId=new_default_key["KeyId"])
+
+    # Modify the default ebs key using the KeyId
+    new_default_ebs_key_arn = ec2.modify_ebs_default_kms_key_id(
+        KmsKeyId=new_default_key["KeyId"]
+    )["KmsKeyId"]
+
+    assert new_default_ebs_key_arn != original_default_ebs_key_arn
+
+    # Creating an encrypted volume should create (and use) the new default key.
+    resource = boto3.resource("ec2", region_name="us-east-1")
+    volume = resource.create_volume(
+        AvailabilityZone="us-east-1a", Encrypted=True, Size=10
+    )
+
+    assert volume.kms_key_id == new_default_ebs_key_arn
+    assert volume.encrypted is True
+
+    # Revert to the original default key by passing empty string
+    ec2.modify_ebs_default_kms_key_id(KmsKeyId="")
+    # Creating an encrypted volume should create (and use) the original default key.
+    volume = resource.create_volume(
+        AvailabilityZone="us-east-1a", Encrypted=True, Size=10
+    )
+    assert volume.kms_key_id == original_default_ebs_key_arn
+
+    # Modify the default ebs key using a key alias
+    new_default_ebs_key_arn = ec2.modify_ebs_default_kms_key_id(
+        KmsKeyId=new_default_alias
+    )["KmsKeyId"]
+    assert new_default_ebs_key_arn != original_default_ebs_key_arn
+    ec2.modify_ebs_default_kms_key_id(KmsKeyId="")
+
+    # Modify the default ebs key using a key arn
+    new_default_ebs_key_arn = ec2.modify_ebs_default_kms_key_id(
+        KmsKeyId=new_default_key["Arn"]
+    )["KmsKeyId"]
+    assert new_default_ebs_key_arn != original_default_ebs_key_arn
diff --git a/tests/test_fsx/test_fsx.py b/tests/test_fsx/test_fsx.py
index c9ea7b716..8246d3a8b 100644
--- a/tests/test_fsx/test_fsx.py
+++ b/tests/test_fsx/test_fsx.py
@@ -1,5 +1,6 @@
 """Unit tests for fsx-supported APIs."""
 
+import re
 import time
 
 import boto3
@@ -32,7 +33,7 @@ def test_create_filesystem(client):
 
     file_system = resp["FileSystem"]
 
-    assert file_system["FileSystemId"].startswith("fs-moto")
+    assert re.match(r"^fs-[0-9a-f]{8,}$", file_system["FileSystemId"])
     assert file_system["FileSystemType"] == "LUSTRE"
 
 
@@ -90,7 +91,6 @@ def test_delete_file_system():
         SecurityGroupIds=FAKE_SECURITY_GROUP_IDS,
     )
 
-    assert fs["FileSystem"]["FileSystemId"].startswith("fs-moto")
     file_system_id = fs["FileSystem"]["FileSystemId"]
     resp = client.delete_file_system(FileSystemId=file_system_id)
     assert resp["FileSystemId"] == file_system_id
diff --git a/tests/test_wafv2/test_wafv2_integration.py b/tests/test_wafv2/test_wafv2_integration.py
index bf536b594..558982e30 100644
--- a/tests/test_wafv2/test_wafv2_integration.py
+++ b/tests/test_wafv2/test_wafv2_integration.py
@@ -20,13 +20,15 @@ def test_associate_with_unknown_resource():
         WebACLArn=wacl_arn,
         ResourceArn="arn:aws:apigateway:us-east-1::/restapis/unknown/stages/unknown",
     )
-    conn.associate_web_acl(WebACLArn=wacl_arn, ResourceArn="unknownarnwithminlength20")
+    random_arn = "arn:aws:cognito-idp:us-east-1:123456789012:userpool/us-east-1_a7e89b7ced4244f48000bd5bd191d871"
+    conn.associate_web_acl(WebACLArn=wacl_arn, ResourceArn=random_arn)
+
+    wacl = conn.get_web_acl_for_resource(ResourceArn=random_arn)["WebACL"]
+    assert wacl["ARN"] == wacl_arn
 
     # We can validate if the WebACL exists
     with pytest.raises(ClientError) as exc:
-        conn.associate_web_acl(
-            WebACLArn=f"{wacl_arn}2", ResourceArn="unknownarnwithminlength20"
-        )
+        conn.associate_web_acl(WebACLArn=f"{wacl_arn}2", ResourceArn=random_arn)
     err = exc.value.response["Error"]
     assert err["Code"] == "WAFNonexistentItemException"
     assert err["Message"] == (
@@ -79,6 +81,7 @@ def test_get_web_acl_for_resource():
 @mock_aws
 def test_disassociate_unknown_resource():
     conn = boto3.client("wafv2", region_name="us-east-1")
+    conn.create_web_acl(**CREATE_WEB_ACL_BODY("John", "REGIONAL"))
     # Nothing happens
     conn.disassociate_web_acl(ResourceArn="unknownarnwithlength20")