Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WAFv2: Extend coverage (#5460)

Browse Source
This commit is contained in:
Bert Blommers 2022-09-10 13:30:45 +00:00 committed by GitHub
parent 64dec7e5d9
commit 872b9ddbdb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 653 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
IMPLEMENTATION_COVERAGE.md
View File

@ -6198,9 +6198,9 @@
## wafv2
<details>
<summary>4% implemented</summary>
<summary>25% implemented</summary>
- [ ] associate_web_acl
- [X] associate_web_acl
- [ ] check_capacity
- [ ] create_ip_set
- [ ] create_regex_pattern_set
@ -6212,9 +6212,9 @@
- [ ] delete_permission_policy
- [ ] delete_regex_pattern_set
- [ ] delete_rule_group
- [ ] delete_web_acl
- [X] delete_web_acl
- [ ] describe_managed_rule_group
- [ ] disassociate_web_acl
- [X] disassociate_web_acl
- [ ] generate_mobile_sdk_release_url
- [ ] get_ip_set
- [ ] get_logging_configuration
@ -6225,8 +6225,8 @@
- [ ] get_regex_pattern_set
- [ ] get_rule_group
- [ ] get_sampled_requests
- [ ] get_web_acl
- [ ] get_web_acl_for_resource
- [X] get_web_acl
- [X] get_web_acl_for_resource
- [ ] list_available_managed_rule_group_versions
- [ ] list_available_managed_rule_groups
- [ ] list_ip_sets
@ -6235,19 +6235,19 @@
- [ ] list_mobile_sdk_releases
- [ ] list_regex_pattern_sets
- [ ] list_resources_for_web_acl
- [ ] list_rule_groups
- [ ] list_tags_for_resource
- [X] list_rule_groups
- [X] list_tags_for_resource
- [X] list_web_acls
- [ ] put_logging_configuration
- [ ] put_managed_rule_set_versions
- [ ] put_permission_policy
- [ ] tag_resource
- [ ] untag_resource
- [X] tag_resource
- [X] untag_resource
- [ ] update_ip_set
- [ ] update_managed_rule_set_version_expiry_date
- [ ] update_regex_pattern_set
- [ ] update_rule_group
- [ ] update_web_acl
- [X] update_web_acl
</details>
## Unimplemented:

40
docs/docs/services/wafv2.rst
View File

@ -27,21 +27,33 @@ wafv2
|start-h3| Implemented features for this service |end-h3|
- [ ] associate_web_acl
- [X] associate_web_acl
Only APIGateway Stages can be associated at the moment.
- [ ] check_capacity
- [ ] create_ip_set
- [ ] create_regex_pattern_set
- [ ] create_rule_group
- [X] create_web_acl
The following parameters are not yet implemented: CustomResponseBodies, CaptchaConfig
- [ ] delete_firewall_manager_rule_groups
- [ ] delete_ip_set
- [ ] delete_logging_configuration
- [ ] delete_permission_policy
- [ ] delete_regex_pattern_set
- [ ] delete_rule_group
- [ ] delete_web_acl
- [X] delete_web_acl
The LockToken-parameter is not yet implemented
- [ ] describe_managed_rule_group
- [ ] disassociate_web_acl
- [X] disassociate_web_acl
- [ ] generate_mobile_sdk_release_url
- [ ] get_ip_set
- [ ] get_logging_configuration
@ -52,8 +64,8 @@ wafv2
- [ ] get_regex_pattern_set
- [ ] get_rule_group
- [ ] get_sampled_requests
- [ ] get_web_acl
- [ ] get_web_acl_for_resource
- [X] get_web_acl
- [X] get_web_acl_for_resource
- [ ] list_available_managed_rule_group_versions
- [ ] list_available_managed_rule_groups
- [ ] list_ip_sets
@ -62,17 +74,25 @@ wafv2
- [ ] list_mobile_sdk_releases
- [ ] list_regex_pattern_sets
- [ ] list_resources_for_web_acl
- [ ] list_rule_groups
- [ ] list_tags_for_resource
- [X] list_rule_groups
- [X] list_tags_for_resource
Pagination is not yet implemented
- [X] list_web_acls
- [ ] put_logging_configuration
- [ ] put_managed_rule_set_versions
- [ ] put_permission_policy
- [ ] tag_resource
- [ ] untag_resource
- [X] tag_resource
- [X] untag_resource
- [ ] update_ip_set
- [ ] update_managed_rule_set_version_expiry_date
- [ ] update_regex_pattern_set
- [ ] update_rule_group
- [ ] update_web_acl
- [X] update_web_acl
The following parameters are not yet implemented: LockToken, CustomResponseBodies, CaptchaConfig

4
moto/apigateway/models.py
View File

@ -376,7 +376,7 @@ class Resource(CloudFormationModel):
return method
def delete_method(self, method_type):
self.resource_methods.pop(method_type)
self.resource_methods.pop(method_type, None)
def add_integration(
self,
@ -1485,7 +1485,7 @@ class APIGatewayBackend(BaseBackend):
api = self.get_rest_api(restapi_id)
del api.authorizers[authorizer_id]
def get_stage(self, function_id, stage_name):
def get_stage(self, function_id, stage_name) -> Stage:
api = self.get_rest_api(function_id)
stage = api.stages.get(stage_name)
if stage is None:

12
moto/wafv2/exceptions.py
View File

@ -1,7 +1,7 @@
from moto.core.exceptions import RESTError
from moto.core.exceptions import JsonRESTError
class WAFv2ClientError(RESTError):
class WAFv2ClientError(JsonRESTError):
code = 400
@ -11,3 +11,11 @@ class WAFV2DuplicateItemException(WAFv2ClientError):
"WafV2DuplicateItem",
"AWS WAF could not perform the operation because some resource in your request is a duplicate of an existing one.",
)
class WAFNonexistentItemException(WAFv2ClientError):
def __init__(self):
super().__init__(
"WAFNonexistentItemException",
"AWS WAF couldn’t perform the operation because your resource doesn’t exist.",
)

168
moto/wafv2/models.py
View File

@ -1,39 +1,21 @@
import datetime
import re
from typing import Dict
from uuid import uuid4
from moto.core import BaseBackend, BaseModel
from moto.wafv2 import utils
from .utils import make_arn_for_wacl, pascal_to_underscores_dict
from .exceptions import WAFV2DuplicateItemException
from .utils import make_arn_for_wacl
from .exceptions import WAFV2DuplicateItemException, WAFNonexistentItemException
from moto.core.utils import iso_8601_datetime_with_milliseconds, BackendDict
import datetime
from moto.utilities.tagging_service import TaggingService
from collections import OrderedDict
US_EAST_1_REGION = "us-east-1"
GLOBAL_REGION = "global"
class VisibilityConfig(BaseModel):
"""
https://docs.aws.amazon.com/waf/latest/APIReference/API_VisibilityConfig.html
"""
def __init__(
self, metric_name, sampled_requests_enabled, cloud_watch_metrics_enabled
):
self.cloud_watch_metrics_enabled = cloud_watch_metrics_enabled
self.metric_name = metric_name
self.sampled_requests_enabled = sampled_requests_enabled
class DefaultAction(BaseModel):
"""
https://docs.aws.amazon.com/waf/latest/APIReference/API_DefaultAction.html
"""
def __init__(self, allow=None, block=None):
self.allow = allow or {}
self.block = block or {}
APIGATEWAY_REGEX = (
r"arn:aws:apigateway:[a-zA-Z0-9-]+::/restapis/[a-zA-Z0-9]+/stages/[a-zA-Z0-9]+"
)
# TODO: Add remaining properties
@ -42,19 +24,30 @@ class FakeWebACL(BaseModel):
https://docs.aws.amazon.com/waf/latest/APIReference/API_WebACL.html
"""
def __init__(self, name, arn, wacl_id, visibility_config, default_action):
self.name = name if name else utils.create_test_name("Mock-WebACL-name")
def __init__(
self, name, arn, wacl_id, visibility_config, default_action, description, rules
):
self.name = name
self.created_time = iso_8601_datetime_with_milliseconds(datetime.datetime.now())
self.id = wacl_id
self.arn = arn
self.description = "Mock WebACL named {0}".format(self.name)
self.description = description or ""
self.capacity = 3
self.visibility_config = VisibilityConfig(
**pascal_to_underscores_dict(visibility_config)
)
self.default_action = DefaultAction(
**pascal_to_underscores_dict(default_action)
)
self.rules = rules
self.visibility_config = visibility_config
self.default_action = default_action
self.lock_token = str(uuid4())[0:6]
def update(self, default_action, rules, description, visibility_config):
if default_action is not None:
self.default_action = default_action
if rules is not None:
self.rules = rules
if description is not None:
self.description = description
if visibility_config is not None:
self.visibility_config = visibility_config
self.lock_token = str(uuid4())[0:6]
def to_dict(self):
# Format for summary https://docs.aws.amazon.com/waf/latest/APIReference/API_CreateWebACL.html (response syntax section)
@ -62,8 +55,10 @@ class FakeWebACL(BaseModel):
"ARN": self.arn,
"Description": self.description,
"Id": self.id,
"LockToken": "Not Implemented",
"Name": self.name,
"Rules": self.rules,
"DefaultAction": self.default_action,
"VisibilityConfig": self.visibility_config,
}
@ -74,10 +69,52 @@ class WAFV2Backend(BaseBackend):
def __init__(self, region_name, account_id):
super().__init__(region_name, account_id)
self.wacls = OrderedDict() # self.wacls[ARN] = FakeWacl
self.wacls: Dict[str, FakeWebACL] = OrderedDict()
self.tagging_service = TaggingService()
# TODO: self.load_balancers = OrderedDict()
def create_web_acl(self, name, visibility_config, default_action, scope):
def associate_web_acl(self, web_acl_arn, resource_arn):
"""
Only APIGateway Stages can be associated at the moment.
"""
if web_acl_arn not in self.wacls:
raise WAFNonexistentItemException
stage = self._find_apigw_stage(resource_arn)
if stage:
stage["webAclArn"] = web_acl_arn
def disassociate_web_acl(self, resource_arn):
stage = self._find_apigw_stage(resource_arn)
if stage:
stage.pop("webAclArn", None)
def get_web_acl_for_resource(self, resource_arn):
stage = self._find_apigw_stage(resource_arn)
if stage and stage.get("webAclArn"):
wacl_arn = stage.get("webAclArn")
return self.wacls.get(wacl_arn)
return None
def _find_apigw_stage(self, resource_arn):
try:
if re.search(APIGATEWAY_REGEX, resource_arn):
region = resource_arn.split(":")[3]
rest_api_id = resource_arn.split("/")[-3]
stage_name = resource_arn.split("/")[-1]
from moto.apigateway import apigateway_backends
apigw = apigateway_backends[self.account_id][region]
return apigw.get_stage(rest_api_id, stage_name)
except: # noqa: E722 Do not use bare except
return None
def create_web_acl(
self, name, visibility_config, default_action, scope, description, tags, rules
):
"""
The following parameters are not yet implemented: CustomResponseBodies, CaptchaConfig
"""
wacl_id = str(uuid4())
arn = make_arn_for_wacl(
name=name,
@ -88,10 +125,29 @@ class WAFV2Backend(BaseBackend):
)
if arn in self.wacls or self._is_duplicate_name(name):
raise WAFV2DuplicateItemException()
new_wacl = FakeWebACL(name, arn, wacl_id, visibility_config, default_action)
new_wacl = FakeWebACL(
name, arn, wacl_id, visibility_config, default_action, description, rules
)
self.wacls[arn] = new_wacl
self.tag_resource(arn, tags)
return new_wacl
def delete_web_acl(self, name, _id):
"""
The LockToken-parameter is not yet implemented
"""
self.wacls = {
arn: wacl
for arn, wacl in self.wacls.items()
if wacl.name != name and wacl.id != _id
}
def get_web_acl(self, name, _id) -> FakeWebACL:
for wacl in self.wacls.values():
if wacl.name == name and wacl.id == _id:
return wacl
raise WAFNonexistentItemException
def list_web_acls(self):
return [wacl.to_dict() for wacl in self.wacls.values()]
@ -99,16 +155,30 @@ class WAFV2Backend(BaseBackend):
allWaclNames = set(wacl.name for wacl in self.wacls.values())
return name in allWaclNames
# TODO: This is how you link wacl to ALB
# @property
# def elbv2_backend(self):
# """
# EC2 backend
def list_rule_groups(self):
return []
# :return: EC2 Backend
# :rtype: moto.ec2.models.EC2Backend
# """
# return ec2_backends[self.region_name]
def list_tags_for_resource(self, arn):
"""
Pagination is not yet implemented
"""
return self.tagging_service.list_tags_for_resource(arn)["Tags"]
def tag_resource(self, arn, tags):
self.tagging_service.tag_resource(arn, tags)
def untag_resource(self, arn, tag_keys):
self.tagging_service.untag_resource_using_names(arn, tag_keys)
def update_web_acl(
self, name, _id, default_action, rules, description, visibility_config
):
"""
The following parameters are not yet implemented: LockToken, CustomResponseBodies, CaptchaConfig
"""
acl = self.get_web_acl(name, _id)
acl.update(default_action, rules, description, visibility_config)
return acl.lock_token
wafv2_backends = BackendDict(

116
moto/wafv2/responses.py
View File

@ -13,6 +13,30 @@ class WAFV2Response(BaseResponse):
def wafv2_backend(self):
return wafv2_backends[self.current_account][self.region]
@amzn_request_id
def associate_web_acl(self):
body = json.loads(self.body)
web_acl_arn = body["WebACLArn"]
resource_arn = body["ResourceArn"]
self.wafv2_backend.associate_web_acl(web_acl_arn, resource_arn)
return 200, {}, "{}"
@amzn_request_id
def disassociate_web_acl(self):
body = json.loads(self.body)
resource_arn = body["ResourceArn"]
self.wafv2_backend.disassociate_web_acl(resource_arn)
return 200, {}, "{}"
@amzn_request_id
def get_web_acl_for_resource(self):
body = json.loads(self.body)
resource_arn = body["ResourceArn"]
web_acl = self.wafv2_backend.get_web_acl_for_resource(resource_arn)
response = {"WebACL": web_acl.to_dict() if web_acl else None}
response_headers = {"Content-Type": "application/json"}
return 200, response_headers, json.dumps(response)
@amzn_request_id
def create_web_acl(self):
"""https://docs.aws.amazon.com/waf/latest/APIReference/API_CreateWebACL.html (response syntax section)"""
@ -22,12 +46,42 @@ class WAFV2Response(BaseResponse):
self.region = GLOBAL_REGION
name = self._get_param("Name")
body = json.loads(self.body)
description = body.get("Description")
tags = body.get("Tags", [])
rules = body.get("Rules", [])
web_acl = self.wafv2_backend.create_web_acl(
name, body["VisibilityConfig"], body["DefaultAction"], scope
name,
body["VisibilityConfig"],
body["DefaultAction"],
scope,
description,
tags,
rules,
)
response = {
"Summary": web_acl.to_dict(),
}
response = {"Summary": web_acl.to_dict()}
response_headers = {"Content-Type": "application/json"}
return 200, response_headers, json.dumps(response)
@amzn_request_id
def delete_web_acl(self):
scope = self._get_param("Scope")
if scope == "CLOUDFRONT":
self.region = GLOBAL_REGION
name = self._get_param("Name")
_id = self._get_param("Id")
self.wafv2_backend.delete_web_acl(name, _id)
response_headers = {"Content-Type": "application/json"}
return 200, response_headers, "{}"
@amzn_request_id
def get_web_acl(self):
scope = self._get_param("Scope")
if scope == "CLOUDFRONT":
self.region = GLOBAL_REGION
name = self._get_param("Name")
_id = self._get_param("Id")
web_acl = self.wafv2_backend.get_web_acl(name, _id)
response = {"WebACL": web_acl.to_dict(), "LockToken": web_acl.lock_token}
response_headers = {"Content-Type": "application/json"}
return 200, response_headers, json.dumps(response)
@ -43,6 +97,60 @@ class WAFV2Response(BaseResponse):
response_headers = {"Content-Type": "application/json"}
return 200, response_headers, json.dumps(response)
@amzn_request_id
def list_rule_groups(self):
scope = self._get_param("Scope")
if scope == "CLOUDFRONT":
self.region = GLOBAL_REGION
rule_groups = self.wafv2_backend.list_rule_groups()
response = {"RuleGroups": [rg.to_dict() for rg in rule_groups]}
response_headers = {"Content-Type": "application/json"}
return 200, response_headers, json.dumps(response)
@amzn_request_id
def list_tags_for_resource(self):
arn = self._get_param("ResourceARN")
self.region = arn.split(":")[3]
tags = self.wafv2_backend.list_tags_for_resource(arn)
response = {"TagInfoForResource": {"ResourceARN": arn, "TagList": tags}}
response_headers = {"Content-Type": "application/json"}
return 200, response_headers, json.dumps(response)
@amzn_request_id
def tag_resource(self):
body = json.loads(self.body)
arn = body.get("ResourceARN")
self.region = arn.split(":")[3]
tags = body.get("Tags")
self.wafv2_backend.tag_resource(arn, tags)
return 200, {}, "{}"
@amzn_request_id
def untag_resource(self):
body = json.loads(self.body)
arn = body.get("ResourceARN")
self.region = arn.split(":")[3]
tag_keys = body.get("TagKeys")
self.wafv2_backend.untag_resource(arn, tag_keys)
return 200, {}, "{}"
@amzn_request_id
def update_web_acl(self):
body = json.loads(self.body)
name = body.get("Name")
_id = body.get("Id")
scope = body.get("Scope")
if scope == "CLOUDFRONT":
self.region = GLOBAL_REGION
default_action = body.get("DefaultAction")
rules = body.get("Rules")
description = body.get("Description")
visibility_config = body.get("VisibilityConfig")
lock_token = self.wafv2_backend.update_web_acl(
name, _id, default_action, rules, description, visibility_config
)
return 200, {}, json.dumps({"NextLockToken": lock_token})
# notes about region and scope
# --scope = CLOUDFRONT is ALWAYS us-east-1 (but we use "global" instead to differentiate between REGIONAL us-east-1)

10
moto/wafv2/utils.py
View File

@ -1,6 +1,3 @@
from moto.core.utils import pascal_to_camelcase, camelcase_to_underscores
def make_arn_for_wacl(name, account_id, region_name, wacl_id, scope):
"""https://docs.aws.amazon.com/waf/latest/developerguide/how-aws-waf-works.html - explains --scope (cloudfront vs regional)"""
@ -9,10 +6,3 @@ def make_arn_for_wacl(name, account_id, region_name, wacl_id, scope):
elif scope == "CLOUDFRONT":
scope = "global"
return f"arn:aws:wafv2:{region_name}:{account_id}:{scope}/webacl/{name}/{wacl_id}"
def pascal_to_underscores_dict(original_dict):
outdict = {}
for k, v in original_dict.items():
outdict[camelcase_to_underscores(pascal_to_camelcase(k))] = v
return outdict

2
tests/terraformtests/terraform-tests.failures.txt
View File

@ -1,7 +1,6 @@
# The Tests in this file worked against an older version of Terraform
# Either they do not work anymore, or have not been verified to work yet
TestAccAPIGatewayStage
TestAccAPIGatewayV2Authorizer
TestAccAPIGatewayV2Route
TestAccAppsyncApiKey
@ -43,7 +42,6 @@ TestAccEksClusterDataSource
TestAccIAMRole
TestAccIotThing
TestAccIPRanges
TestAccKinesisStream
TestAccELBPolicy
TestAccPartition
TestAccPinpointApp

22
tests/terraformtests/terraform-tests.success.txt
View File

@ -4,7 +4,22 @@ amp:
- TestAccAMPWorkspace
- TestAccAMPRuleGroupNamespace
apigateway:
- TestAccAPIGatewayAPIKeyDataSource_basic
- TestAccAPIGatewayAPIKey_disappears
- TestAccAPIGatewayAPIKey_enabled
- TestAccAPIGatewayAPIKey_value
- TestAccAPIGatewayGatewayResponse
- TestAccAPIGatewayRestAPI_apiKeySource
- TestAccAPIGatewayRestAPI_basic
- TestAccAPIGatewayRestAPI_description
- TestAccAPIGatewayRestAPI_disappears
- TestAccAPIGatewayRestAPI_Endpoint_private
- TestAccAPIGatewayStage_basic
- TestAccAPIGatewayStage_Disappears_restAPI
- TestAccAPIGatewayStage_disappears
- TestAccAPIGatewayStage_Disappears_referencingDeployment
- TestAccAPIGatewayStage_tags
- TestAccAPIGatewayStage_accessLogSettings
apigatewayv2:
- TestAccAPIGatewayV2IntegrationResponse
- TestAccAPIGatewayV2Model
@ -244,3 +259,10 @@ sqs:
timestreamwrite:
- TestAccTimestreamWriteDatabase
- TestAccTimestreamWriteTable
wafv2:
- TestAccWAFV2WebACL_basic
- TestAccWAFV2WebACL_disappears
- TestAccWAFV2WebACL_minimal
- TestAccWAFV2WebACL_tags
- TestAccWAFV2WebACL_Update_rule
- TestAccWAFV2WebACL_RuleLabels

158
tests/test_wafv2/test_wafv2.py
View File

@ -25,7 +25,7 @@ def test_create_web_acl():
err["Message"].should.contain(
"AWS WAF could not perform the operation because some resource in your request is a duplicate of an existing one."
)
err["Code"].should.equal("400")
err["Code"].should.equal("WafV2DuplicateItem")
res = conn.create_web_acl(**CREATE_WEB_ACL_BODY("Carl", "CLOUDFRONT"))
web_acl = res["Summary"]
@ -35,8 +35,79 @@ def test_create_web_acl():
@mock_wafv2
def test_list_web_acl():
def test_create_web_acl_with_all_arguments():
client = boto3.client("wafv2", region_name="us-east-2")
web_acl_id = client.create_web_acl(
Name="test",
Scope="CLOUDFRONT",
DefaultAction={"Allow": {}},
Description="test desc",
VisibilityConfig={
"SampledRequestsEnabled": False,
"CloudWatchMetricsEnabled": False,
"MetricName": "idk",
},
Rules=[
{
"Action": {"Allow": {}},
"Name": "tf-acc-test-8205974093017792151-2",
"Priority": 10,
"Statement": {"GeoMatchStatement": {"CountryCodes": ["US", "NL"]}},
"VisibilityConfig": {
"CloudWatchMetricsEnabled": False,
"MetricName": "tf-acc-test-8205974093017792151-2",
"SampledRequestsEnabled": False,
},
},
{
"Action": {"Count": {}},
"Name": "tf-acc-test-8205974093017792151-1",
"Priority": 5,
"Statement": {
"SizeConstraintStatement": {
"ComparisonOperator": "LT",
"FieldToMatch": {"QueryString": {}},
"Size": 50,
"TextTransformations": [
{"Priority": 2, "Type": "CMD_LINE"},
{"Priority": 5, "Type": "NONE"},
],
}
},
"VisibilityConfig": {
"CloudWatchMetricsEnabled": False,
"MetricName": "tf-acc-test-8205974093017792151-1",
"SampledRequestsEnabled": False,
},
},
],
)["Summary"]["Id"]
wacl = client.get_web_acl(Name="test", Scope="CLOUDFRONT", Id=web_acl_id)["WebACL"]
wacl.should.have.key("Description").equals("test desc")
wacl.should.have.key("DefaultAction").equals({"Allow": {}})
wacl.should.have.key("VisibilityConfig").equals(
{
"SampledRequestsEnabled": False,
"CloudWatchMetricsEnabled": False,
"MetricName": "idk",
}
)
wacl.should.have.key("Rules").length_of(2)
@mock_wafv2
def test_get_web_acl():
conn = boto3.client("wafv2", region_name="us-east-1")
body = CREATE_WEB_ACL_BODY("John", "REGIONAL")
web_acl_id = conn.create_web_acl(**body)["Summary"]["Id"]
wacl = conn.get_web_acl(Name="John", Scope="REGIONAL", Id=web_acl_id)["WebACL"]
wacl.should.have.key("Name").equals("John")
wacl.should.have.key("Id").equals(web_acl_id)
@mock_wafv2
def test_list_web_acl():
conn = boto3.client("wafv2", region_name="us-east-1")
conn.create_web_acl(**CREATE_WEB_ACL_BODY("Daphne", "REGIONAL"))
conn.create_web_acl(**CREATE_WEB_ACL_BODY("Penelope", "CLOUDFRONT"))
@ -51,3 +122,86 @@ def test_list_web_acl():
web_acls = res["WebACLs"]
assert len(web_acls) == 1
assert web_acls[0]["Name"] == "Penelope"
@mock_wafv2
def test_delete_web_acl():
conn = boto3.client("wafv2", region_name="us-east-1")
wacl_id = conn.create_web_acl(**CREATE_WEB_ACL_BODY("Daphne", "REGIONAL"))[
"Summary"
]["Id"]
conn.delete_web_acl(Name="Daphne", Id=wacl_id, Scope="REGIONAL", LockToken="n/a")
res = conn.list_web_acls(**LIST_WEB_ACL_BODY("REGIONAL"))
res["WebACLs"].should.have.length_of(0)
with pytest.raises(ClientError) as exc:
conn.get_web_acl(Name="Daphne", Scope="REGIONAL", Id=wacl_id)
err = exc.value.response["Error"]
err["Code"].should.equal("WAFNonexistentItemException")
err["Message"].should.equal(
"AWS WAF couldn’t perform the operation because your resource doesn’t exist."
)
@mock_wafv2
def test_update_web_acl():
conn = boto3.client("wafv2", region_name="us-east-1")
wacl_id = conn.create_web_acl(**CREATE_WEB_ACL_BODY("Daphne", "REGIONAL"))[
"Summary"
]["Id"]
resp = conn.update_web_acl(
Name="Daphne",
Scope="REGIONAL",
Id=wacl_id,
DefaultAction={"Block": {"CustomResponse": {"ResponseCode": 412}}},
Description="updated_desc",
Rules=[
{
"Name": "rule1",
"Priority": 456,
"Statement": {},
"VisibilityConfig": {
"SampledRequestsEnabled": True,
"CloudWatchMetricsEnabled": True,
"MetricName": "updated",
},
}
],
LockToken="n/a",
VisibilityConfig={
"SampledRequestsEnabled": True,
"CloudWatchMetricsEnabled": True,
"MetricName": "updated",
},
)
resp.should.have.key("NextLockToken")
acl = conn.get_web_acl(Name="Daphne", Scope="REGIONAL", Id=wacl_id)["WebACL"]
acl.should.have.key("Description").equals("updated_desc")
acl.should.have.key("DefaultAction").equals(
{"Block": {"CustomResponse": {"ResponseCode": 412}}}
)
acl.should.have.key("Rules").equals(
[
{
"Name": "rule1",
"Priority": 456,
"Statement": {},
"VisibilityConfig": {
"SampledRequestsEnabled": True,
"CloudWatchMetricsEnabled": True,
"MetricName": "updated",
},
}
]
)
acl.should.have.key("VisibilityConfig").equals(
{
"SampledRequestsEnabled": True,
"CloudWatchMetricsEnabled": True,
"MetricName": "updated",
}
)

99
tests/test_wafv2/test_wafv2_integration.py Normal file
View File

@ -0,0 +1,99 @@
import pytest
import sure # noqa # pylint: disable=unused-import
import boto3
from botocore.exceptions import ClientError
from moto import mock_apigateway, mock_wafv2
from .test_helper_functions import CREATE_WEB_ACL_BODY
from tests.test_apigateway.test_apigateway_stage import create_method_integration
@mock_wafv2
def test_associate_with_unknown_resource():
conn = boto3.client("wafv2", region_name="us-east-1")
wacl_arn = conn.create_web_acl(**CREATE_WEB_ACL_BODY("John", "REGIONAL"))[
"Summary"
]["ARN"]
# We do not have any validation yet on the existence or format of the resource arn
conn.associate_web_acl(
WebACLArn=wacl_arn,
ResourceArn="arn:aws:apigateway:us-east-1::/restapis/unknown/stages/unknown",
)
conn.associate_web_acl(WebACLArn=wacl_arn, ResourceArn="unknownarnwithminlength20")
# We can validate if the WebACL exists
with pytest.raises(ClientError) as exc:
conn.associate_web_acl(
WebACLArn=f"{wacl_arn}2", ResourceArn="unknownarnwithminlength20"
)
err = exc.value.response["Error"]
err["Code"].should.equal("WAFNonexistentItemException")
err["Message"].should.equal(
"AWS WAF couldn’t perform the operation because your resource doesn’t exist."
)
@mock_apigateway
@mock_wafv2
def test_associate_with_apigateway_stage():
conn = boto3.client("wafv2", region_name="us-east-1")
wacl_arn = conn.create_web_acl(**CREATE_WEB_ACL_BODY("John", "REGIONAL"))[
"Summary"
]["ARN"]
apigw = boto3.client("apigateway", region_name="us-east-1")
api_id, stage_arn = create_apigateway_stage(client=apigw)
conn.associate_web_acl(WebACLArn=wacl_arn, ResourceArn=stage_arn)
stage = apigw.get_stage(restApiId=api_id, stageName="test")
stage.should.have.key("webAclArn").equals(wacl_arn)
conn.disassociate_web_acl(ResourceArn=stage_arn)
stage = apigw.get_stage(restApiId=api_id, stageName="test")
stage.shouldnt.have.key("webAclArn")
@mock_apigateway
@mock_wafv2
def test_get_web_acl_for_resource():
conn = boto3.client("wafv2", region_name="us-east-1")
wacl_arn = conn.create_web_acl(**CREATE_WEB_ACL_BODY("John", "REGIONAL"))[
"Summary"
]["ARN"]
apigw = boto3.client("apigateway", region_name="us-east-1")
_, stage_arn = create_apigateway_stage(client=apigw)
resp = conn.get_web_acl_for_resource(ResourceArn=stage_arn)
resp.shouldnt.have.key("WebACL")
conn.associate_web_acl(WebACLArn=wacl_arn, ResourceArn=stage_arn)
resp = conn.get_web_acl_for_resource(ResourceArn=stage_arn)
resp.should.have.key("WebACL")
resp["WebACL"].should.have.key("Name").equals("John")
resp["WebACL"].should.have.key("ARN").equals(wacl_arn)
@mock_wafv2
def test_disassociate_unknown_resource():
conn = boto3.client("wafv2", region_name="us-east-1")
# Nothing happens
conn.disassociate_web_acl(ResourceArn="unknownarnwithlength20")
def create_apigateway_stage(client):
stage_name = "staging"
response = client.create_rest_api(name="my_api", description="this")
api_id = response["id"]
create_method_integration(client=client, api_id=api_id)
response = client.create_deployment(restApiId=api_id, stageName=stage_name)
deployment_id = response["id"]
client.create_stage(restApiId=api_id, stageName="test", deploymentId=deployment_id)
stage_arn = f"arn:aws:apigateway:us-east-1::/restapis/{api_id}/stages/test"
return api_id, stage_arn

9
tests/test_wafv2/test_wafv2_rules.py Normal file
View File

@ -0,0 +1,9 @@
import boto3
from moto import mock_wafv2
@mock_wafv2
def test_list_rule_groups():
client = boto3.client("wafv2", region_name="us-east-2")
resp = client.list_rule_groups(Scope="CLOUDFRONT")
resp.should.have.key("RuleGroups").equals([])

83
tests/test_wafv2/test_wafv2_tags.py Normal file
View File

@ -0,0 +1,83 @@
import boto3
import sure # noqa # pylint: disable=unused-import
from moto import mock_wafv2
from .test_helper_functions import CREATE_WEB_ACL_BODY
@mock_wafv2
def test_list_tags_for_resource__none_supplied():
conn = boto3.client("wafv2", region_name="us-east-1")
arn = conn.create_web_acl(**CREATE_WEB_ACL_BODY("John", "REGIONAL"))["Summary"][
"ARN"
]
tag_info = conn.list_tags_for_resource(ResourceARN=arn)["TagInfoForResource"]
tag_info.should.have.key("TagList").equals([])
@mock_wafv2
def test_list_tags_for_resource():
conn = boto3.client("wafv2", region_name="us-east-1")
arn = conn.create_web_acl(
Name="test",
Scope="CLOUDFRONT",
DefaultAction={"Allow": {}},
Tags=[{"Key": "k1", "Value": "v1"}],
VisibilityConfig={
"SampledRequestsEnabled": False,
"CloudWatchMetricsEnabled": False,
"MetricName": "idk",
},
)["Summary"]["ARN"]
tag_info = conn.list_tags_for_resource(ResourceARN=arn)["TagInfoForResource"]
tag_info.should.have.key("TagList").equals([{"Key": "k1", "Value": "v1"}])
@mock_wafv2
def test_tag_resource():
conn = boto3.client("wafv2", region_name="us-east-1")
arn = conn.create_web_acl(
Name="test",
Scope="CLOUDFRONT",
DefaultAction={"Allow": {}},
Tags=[{"Key": "k1", "Value": "v1"}],
VisibilityConfig={
"SampledRequestsEnabled": False,
"CloudWatchMetricsEnabled": False,
"MetricName": "idk",
},
)["Summary"]["ARN"]
conn.tag_resource(
ResourceARN=arn,
Tags=[
{"Key": "k2", "Value": "v2"},
],
)
tag_info = conn.list_tags_for_resource(ResourceARN=arn)["TagInfoForResource"]
tag_info.should.have.key("TagList").equals(
[{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}]
)
@mock_wafv2
def test_untag_resource():
conn = boto3.client("wafv2", region_name="us-east-1")
arn = conn.create_web_acl(
Name="test",
Scope="CLOUDFRONT",
DefaultAction={"Allow": {}},
Tags=[{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}],
VisibilityConfig={
"SampledRequestsEnabled": False,
"CloudWatchMetricsEnabled": False,
"MetricName": "idk",
},
)["Summary"]["ARN"]
conn.untag_resource(ResourceARN=arn, TagKeys=["k1"])
tag_info = conn.list_tags_for_resource(ResourceARN=arn)["TagInfoForResource"]
tag_info.should.have.key("TagList").equals([{"Key": "k2", "Value": "v2"}])