Support replacer job

config/config-template-zap-long.yaml

@@ -151,6 +151,18 @@ scanners:
       # Use https://www.zaproxy.org/docs/alerts/ to match rule with its ID
       disabledRules: "2,10015,10024,10027,10054,10096,10109,10112"
 
+    #replacer:                          # replacer replaces strings in requests and responses during active scan
+    #  parameters:
+    #    deleteAllRules: True           # Boolean, if true then will delete all existing replacer rules, default false
+    #  rules:                           # list of rules
+    #  - description:                   # String, the name of the rule
+    #    url:                           # (optional) String, a regex which will be used to match URLs, if empty then it will match all
+    #    matchType:                     # String, one of req_header, req_header_str, req_body_str, resp_header, resp_header_str, resp_body_str
+    #    matchString:                   # String, will be used to identify what should be replaced
+    #    matchRegex:                    # Boolean, if set then the matchString will be treated as a regex, default false
+    #    replacementString:             # String, the new string that will replace the specified selection
+    #    tokenProcessing:               # (optional) Boolean, when enabled the replacementString may contain a single token
+
     activeScan:
       # The list of parameters: https://www.zaproxy.org/docs/desktop/addons/ajax-spider/automation/
       #maxRuleDurationInMins: max scan time for each Rule (default: unlimited)

scanners/zap/zap.py

@@ -365,6 +365,7 @@ def _setup_zap_automation(self):
         self._setup_api()
         self._setup_graphql()
         self._setup_import_urls()
+        self._setup_replacer()
         self._setup_active_scan()
         self._setup_passive_wait()
         self._setup_report()
@@ -618,6 +619,45 @@ def _setup_passive_wait(self):
         }
         self.automation_config["jobs"].append(waitfor)
 
+    def _setup_replacer(self):
+        """Adds the replacer to the job list"""
+
+        def _validate_rule_boolean_values(rule):
+            if not isinstance(rule["matchRegex"], bool):
+                raise ValueError("The matchRegex in the replacer rule must be set to a Boolean value")
+
+            if "tokenProcessing" in rule and not isinstance(rule["tokenProcessing"], bool):
+                raise ValueError("The tokenProcessing in the replacer rule must be set to a Boolean value")
+
+        if not self.my_conf("replacer"):
+            return
+
+        rules = self.my_conf("replacer.rules")
+        if rules:
+            if not isinstance(rules, list):
+                raise ValueError("replacer.rules must be a list")
+
+            for item in rules:
+                _validate_rule_boolean_values(item)
+        else:
+            raise ValueError("replacer must have a rule at least")
+
+        delete_all_rules = self.my_conf("replacer.parameters.deleteAllRules", default=True)
+        if not isinstance(delete_all_rules, bool):
+            raise ValueError("replacer.parameters.deleteAllRules must be set to a Boolean value")
+
+        # replacer schema
+        replacer = {
+            "name": "replacer",
+            "type": "replacer",
+            "parameters": {
+                "deleteAllRules": delete_all_rules,
+            },
+            "rules": rules,
+        }
+
+        self.automation_config["jobs"].append(replacer)
+
     def _setup_active_scan(self):
         """Adds an active scan job list, if there is one"""
 

tests/scanners/zap/test_setup.py

@@ -236,6 +236,87 @@ def test_setup_include_urls(test_config):
     assert "def" in find_context(test_zap.automation_config)["includePaths"]
 
 
+def test_setup_replacer_no_rule(test_config):
+    test_config.set("scanners.zap.replacer.parameters.deleteAllRules", False)
+    test_zap = ZapNone(config=test_config)
+
+    # test: not having a rule raises ValueError
+    with pytest.raises(ValueError):
+        test_zap.setup()
+
+
+def test_setup_replacer_parameter(test_config):
+    test_rule = {
+        "description": "test_rule1",  # String, the name of the rule
+        "url": ".*",  # (optional) String, a regex which will be used to match URLs, if empty then it will match all
+        "matchType": "req_body_str",  # String, one of req_header, req_header_str, req_body_str, resp_header, resp_header_str, resp_body_str
+        "matchString": "John Doe",  # String, will be used to identify what should be replaced
+        "matchRegex": False,  # Boolean, if set then the matchString will be treated as a regex, default false
+        "replacementString": "test_string",  # String, the new string that will replace the specified selection
+        "tokenProcessing": False,  # (optional) Boolean, when enabled the replacementString may contain a single token
+    }
+
+    test_config.set("scanners.zap.replacer.rules", [test_rule])
+
+    test_zap = ZapNone(config=test_config)
+    test_zap.setup()
+
+    # test: deleteAllRules is True when it is not set
+    for item in test_zap.automation_config["jobs"]:
+        if item["type"] == "replacer":
+            assert item["parameters"]["deleteAllRules"] is True
+
+    # test: deleteAllRules parameter is set to False
+    test_config.set("scanners.zap.replacer.parameters.deleteAllRules", False)
+    test_zap = ZapNone(config=test_config)
+    test_zap.setup()
+
+    for item in test_zap.automation_config["jobs"]:
+        if item["type"] == "replacer":
+            assert isinstance(item["parameters"]["deleteAllRules"], bool)
+            assert item["parameters"]["deleteAllRules"] is False
+
+    # test: when deleteAllRules parameter is not Boolean type
+    test_config.set("scanners.zap.replacer.parameters.deleteAllRules", "non-boolean")
+    test_zap = ZapNone(config=test_config)
+    with pytest.raises(ValueError):
+        test_zap.setup()
+
+
+def test_setup_replacer_rules(test_config):
+    # test rules
+    test_rule1 = {
+        "description": "test_rule1",  # String, the name of the rule
+        "url": ".*",  # (optional) String, a regex which will be used to match URLs, if empty then it will match all
+        "matchType": "req_body_str",  # String, one of req_header, req_header_str, req_body_str, resp_header, resp_header_str, resp_body_str
+        "matchString": "John Doe",  # String, will be used to identify what should be replaced
+        "matchRegex": False,  # Boolean, if set then the matchString will be treated as a regex, default false
+        "replacementString": "test_string",  # String, the new string that will replace the specified selection
+        "tokenProcessing": False,  # (optional) Boolean, when enabled the replacementString may contain a single token
+    }
+    test_rule2 = {
+        "description": "test_rule2",  # String, the name of the rule
+        "matchType": "req_header",  # String, one of req_header, req_header_str, req_body_str, resp_header, resp_header_str, resp_body_str
+        "matchString": "Cookie",  # String, will be used to identify what should be replaced
+        "matchRegex": True,  # Boolean, if set then the matchString will be treated as a regex, default false
+        "replacementString": "session=abc123",
+    }
+
+    test_config.set("scanners.zap.replacer.rules", [test_rule1, test_rule2])
+
+    test_zap = ZapNone(config=test_config)
+    test_zap.setup()
+
+    for item in test_zap.automation_config["jobs"]:
+        if item["type"] == "replacer":
+            assert item["rules"][0] is test_rule1
+            assert isinstance(item["rules"][0]["matchRegex"], bool)
+            assert isinstance(item["rules"][0]["tokenProcessing"], bool)
+
+            assert item["rules"][1] is test_rule2
+            assert isinstance(item["rules"][1]["matchRegex"], bool)
+
+
 @patch("scanners.zap.zap.validate_active_scan_policy")
 def test_setup_active_scan(mock_validate_active_scan_policy, test_config):
     test_config.set("scanners.zap.activeScan.maxRuleDurationInMins", 10)