waf regex match and pattern sets

README.md

@@ -49,3 +49,17 @@ Required parameters:
 - `Destination` - Destination bucket and prefix in `s3://bucket-name/destination-prefix` format
 - `CannedAcl` - Canned ACL for created objects in destination
 No optional parameters.
+
+### Create Regex Waf Rules
+
+This custom resource allows create/update/delete match regex rules.
+
+handler: `waf_regex/handler.lambda_handler`
+runtime:  `python3.6`
+
+Required parameters:
+
+- `RegexPatterns` - List format, regex pattern to match.
+- `Type` - The part of the web request that you want AWS WAF to search for a specified string
+- `Data` - Data such as when the value of Type is HEADER , enter the name of the header that you want AWS WAF to search, for example, User-Agent or Referer
+- `Transform` - Text transformations eliminate some of the unusual formatting that attackers use in web requests in an effort to bypass AWS WAF.

waf_regex/__init__.py

@@ -0,0 +1 @@
+# package marker

waf_regex/cr_response.py

@@ -0,0 +1,56 @@
+import logging
+from urllib.request import urlopen, Request, HTTPError, URLError
+import json
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+class CustomResourceResponse:
+    def __init__(self, request_payload):
+        self.payload = request_payload
+        self.response = {
+            "StackId": request_payload["StackId"],
+            "RequestId": request_payload["RequestId"],
+            "LogicalResourceId": request_payload["LogicalResourceId"],
+            "Status": 'SUCCESS',
+        }
+
+
+    def respond_error(self, message):
+        self.response['Status'] = 'FAILED'
+        self.response['Reason'] = message
+        self.respond()
+
+    def respond(self, data):
+        event = self.payload
+        response = self.response
+        ####
+        #### copied from https://github.com/ryansb/cfn-wrapper-python/blob/master/cfn_resource.py
+        ####
+
+        if event.get("PhysicalResourceId", False):
+            response["PhysicalResourceId"] = event["PhysicalResourceId"]
+
+        logger.debug("Received %s request with event: %s" %
+                     (event['RequestType'], json.dumps(event)))
+
+        serialized = json.dumps(response)
+        logger.info(f"Responding to {event['RequestType']} request with: {serialized}")
+        response["Data"] = data
+        req_data = serialized.encode('utf-8')
+
+        req = Request(
+            event['ResponseURL'],
+            data=req_data,
+            headers={'Content-Length': len(req_data), 'Content-Type': ''}
+        )
+        req.get_method = lambda: 'PUT'
+
+        try:
+            urlopen(req)
+            logger.debug("Request to CFN API succeeded, nothing to do here")
+        except HTTPError as e:
+            logger.error("Callback to CFN API failed with status %d" % e.code)
+            logger.error("Response: %s" % e.reason)
+        except URLError as e:
+            logger.error("Failed to reach the server - %s" % e.reason)

waf_regex/handler.py

@@ -0,0 +1,50 @@
+import sys
+import os
+import re
+import random
+
+sys.path.append(f"{os.environ['LAMBDA_TASK_ROOT']}/lib")
+sys.path.append(os.path.dirname(os.path.realpath(__file__)))
+
+import cr_response
+from logic import WafRegexLogic
+import json
+
+def lambda_handler(event, context):
+
+    print(f"Received event:{json.dumps(event)}")
+
+    lambda_response = cr_response.CustomResourceResponse(event)
+    cr_params = event['ResourceProperties']
+    match_name = event['LogicalResourceId']
+    waf_logic = WafRegexLogic(match_name , cr_params)
+    try:
+        # if create request, generate physical id, both for create/update copy files
+        if event['RequestType'] == 'Create':
+            print("Create request")
+            event['PhysicalResourceId'] = waf_logic.new_match_set()
+            data = {
+                "MatchID" : event['PhysicalResourceId']
+            }
+            lambda_response.respond(data)
+
+        elif event['RequestType'] == 'Update':
+            print("Update request")
+            waf_logic.update_match_set(event['PhysicalResourceId'])
+            data = {
+                "MatchID" : event['PhysicalResourceId']
+            }
+            lambda_response.respond(data)
+
+        elif event['RequestType'] == 'Delete':
+            print(event['PhysicalResourceId'])
+            waf_logic.remove_match_set(event['PhysicalResourceId'])
+            print("Delete request")
+            data = { }
+            lambda_response.respond(data)
+
+    except Exception as e:
+        message = str(e)
+        lambda_response.respond_error(message)
+
+    return 'OK'

waf_regex/logic.py

@@ -0,0 +1,159 @@
+import boto3
+import os
+import glob
+import logging
+import pprint
+
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+
+class WafRegexLogic:
+
+    def __init__(self, name, resource_properties):
+        self.regex_patterns = resource_properties['RegexPatterns']
+        self.match_type = resource_properties['Type']
+        self.match_data = resource_properties['Data']
+        self.transform = resource_properties['Transform']
+        self.match_name = name
+        self.pattern_name = f"{name}-pattern"
+        self.client = boto3.client('waf-regional')
+
+    def new_pattern_set(self):
+        changeToken = self.client.get_change_token()
+        response_create_pattern_set = self.client.create_regex_pattern_set(
+            Name=self.pattern_name,
+            ChangeToken=changeToken['ChangeToken']
+        )
+        for pattern in self.regex_patterns:
+            self.insert_pattern_set(response_create_pattern_set['RegexPatternSet']['RegexPatternSetId'], pattern)
+        return response_create_pattern_set['RegexPatternSet']['RegexPatternSetId']
+
+    def remove_pattern_set(self,pattern_set_id):
+        pattern_set_object = self.get_pattern_set(pattern_set_id)
+        for pattern_set_string in pattern_set_object['RegexPatternSet']['RegexPatternStrings']:
+            self.delete_pattern_set(pattern_set_id, pattern_set_string)
+        changeToken = self.client.get_change_token()
+        response = self.client.delete_regex_pattern_set(
+            RegexPatternSetId=pattern_set_id,
+            ChangeToken=changeToken['ChangeToken']
+        )
+
+    def get_pattern_set(self,pattern_set_id):
+        pattern_set_object = self.client.get_regex_pattern_set(
+            RegexPatternSetId=pattern_set_id
+        )
+        return pattern_set_object
+
+
+    def update_pattern_set(self,pattern_set_id):
+        pattern_set_object = self.get_pattern_set(pattern_set_id)
+        #delete existing and add a new one
+        for pattern_set_string in pattern_set_object['RegexPatternSet']['RegexPatternStrings']:
+            self.delete_pattern_set(pattern_set_id, pattern_set_string)
+        for pattern in self.regex_patterns:
+            self.insert_pattern_set(pattern_set_id, pattern)
+
+
+    def insert_pattern_set(self,pattern_set_id, pattern_set_string):
+        changeToken = self.client.get_change_token()
+        update_regex_patternset = self.client.update_regex_pattern_set(
+            RegexPatternSetId=pattern_set_id,
+            Updates=[
+                {
+                    'Action': 'INSERT',
+                    'RegexPatternString': pattern_set_string
+                },
+            ],
+            ChangeToken=changeToken['ChangeToken']
+        )
+
+    def delete_pattern_set(self,pattern_set_id, pattern_set_string):
+        changeToken = self.client.get_change_token()
+        update_regex_patternset = self.client.update_regex_pattern_set(
+            RegexPatternSetId=pattern_set_id,
+            Updates=[
+                {
+                    'Action': 'DELETE',
+                    'RegexPatternString': pattern_set_string
+                },
+            ],
+            ChangeToken=changeToken['ChangeToken']
+        )
+
+    #
+    # Match Sets
+    #
+
+    def new_match_set(self):
+        changeToken = self.client.get_change_token()
+        #create match set
+        response_create_match_set = self.client.create_regex_match_set(
+            Name=self.match_name,
+            ChangeToken=changeToken['ChangeToken']
+        )
+        #create pattern set
+        pattern_set_id = self.new_pattern_set()
+        self.insert_match_set(response_create_match_set['RegexMatchSet']['RegexMatchSetId'], pattern_set_id)
+        return response_create_match_set['RegexMatchSet']['RegexMatchSetId']
+
+    def insert_match_set(self, match_set_id, pattern_set_id):
+        changeToken = self.client.get_change_token()
+        update_regex_matchset = self.client.update_regex_match_set(
+            RegexMatchSetId=match_set_id,
+            Updates=[
+                {
+                    'Action': 'INSERT',
+                    'RegexMatchTuple': {
+                        'FieldToMatch': {
+                            'Type': self.match_type,
+                            'Data': self.match_data
+                        },
+                        'TextTransformation': self.transform,
+                        'RegexPatternSetId': pattern_set_id
+                    }
+                },
+            ],
+            ChangeToken=changeToken['ChangeToken']
+        )
+
+    def update_match_set(self,match_set_id):
+        match_set_object = self.get_match_set(match_set_id)
+        for match_tuple in match_set_object['RegexMatchSet']['RegexMatchTuples']:
+            self.update_pattern_set(match_tuple['RegexPatternSetId'])
+            self.delete_match_set(match_set_id,match_tuple)
+            self.insert_match_set(match_set_id,match_tuple['RegexPatternSetId'])
+
+
+    def get_match_set(self,match_set_id):
+        match_set_object = self.client.get_regex_match_set(
+            RegexMatchSetId=match_set_id
+        )
+        return match_set_object
+
+    def remove_match_set(self,match_set_id):
+        match_set_object = self.get_match_set(match_set_id)
+
+        for match_tuple in match_set_object['RegexMatchSet']['RegexMatchTuples']:
+            self.delete_match_set(match_set_id,match_tuple)
+            self.remove_pattern_set(match_tuple['RegexPatternSetId'])
+
+        changeToken = self.client.get_change_token()
+        response = self.client.delete_regex_match_set(
+            RegexMatchSetId=match_set_id,
+            ChangeToken=changeToken['ChangeToken']
+        )
+
+    def delete_match_set(self,match_set_id,match_tuple):
+        changeToken = self.client.get_change_token()
+        update_regex_matchset = self.client.update_regex_match_set(
+            RegexMatchSetId=match_set_id,
+            Updates=[
+                {
+                    'Action': 'DELETE',
+                    'RegexMatchTuple': match_tuple
+                },
+            ],
+            ChangeToken=changeToken['ChangeToken']
+        )