fix: add patch script validation and execution for Core Rule Set (CRS) nightly jobs

src/common/core/modsecurity/jobs/coreruleset-nightly.py

@@ -6,6 +6,7 @@ from os.path import join
 from pathlib import Path
 from re import MULTILINE, search
 from shutil import rmtree
+from subprocess import CalledProcessError, run
 from sys import exit as sys_exit, path as sys_path
 from tarfile import open as tar_open
 
@@ -22,8 +23,13 @@ LOGGER = setup_logger("MODSECURITY.coreruleset-nightly", getenv("LOG_LEVEL", "IN
 status = 0
 
 CRS_NIGHTLY_PATH = Path(sep, "var", "cache", "bunkerweb", "modsecurity", "crs", "nightly")
+PATCH_SCRIPT = Path(sep, "usr", "share", "bunkerweb", "core", "modsecurity", "misc", "patch.sh")
 
 try:
+    if not PATCH_SCRIPT.is_file():
+        LOGGER.error(f"Patch script not found: {PATCH_SCRIPT}")
+        sys_exit(1)
+
     # * Check if we're using the nightly version of the Core Rule Set (CRS)
     use_nightly_crs = False
 
@@ -108,6 +114,16 @@ try:
     example_conf = CRS_NIGHTLY_PATH.joinpath("crs-nightly", "crs-setup.conf.example")
     example_conf.rename(CRS_NIGHTLY_PATH.joinpath("crs-setup-nightly.conf"))
 
+    # * Patch the rules so we can extract the rule IDs when matching
+    try:
+        LOGGER.info("Patching Core Rule Set (CRS) nightly rules...")
+        result = run([PATCH_SCRIPT.as_posix(), CRS_NIGHTLY_PATH.as_posix()], check=True)
+    except CalledProcessError as e:
+        LOGGER.error(f"Failed to patch Core Rule Set (CRS) nightly rules: {e}")
+        sys_exit(1)
+
+    LOGGER.info("Successfully patched Core Rule Set (CRS) nightly rules.")
+
     cached, err = JOB.cache_dir(CRS_NIGHTLY_PATH)
     if not cached:
         LOGGER.error(f"Error while saving Core Rule Set (CRS) nightly data to db cache: {err}")

src/common/core/modsecurity/jobs/download-crs-plugins.py

@@ -5,6 +5,7 @@ from os import getenv, sep
 from os.path import join
 from pathlib import Path
 from re import MULTILINE, compile as re_compile
+from subprocess import CalledProcessError, run
 from sys import exit as sys_exit, path as sys_path
 from typing import Dict, Set
 from uuid import uuid4
@@ -37,10 +38,15 @@ PLUGIN_VERSION_RX = re_compile(r"^# Plugin version: (?P<version>.+)$", MULTILINE
 CRS_PLUGINS_DIR = Path(sep, "var", "cache", "bunkerweb", "modsecurity", "crs", "plugins")
 NEW_PLUGINS_DIR = Path(sep, "var", "tmp", "bunkerweb", "crs-new-plugins")
 TMP_DIR = Path(sep, "var", "tmp", "bunkerweb", "crs-plugins")
+PATCH_SCRIPT = Path(sep, "usr", "share", "bunkerweb", "core", "modsecurity", "misc", "patch.sh")
 LOGGER = setup_logger("modsecurity.download-crs-plugins", getenv("LOG_LEVEL", "INFO"))
 status = 0
 
 try:
+    if not PATCH_SCRIPT.is_file():
+        LOGGER.error(f"Patch script not found: {PATCH_SCRIPT}")
+        sys_exit(1)
+
     # * Check if we're using the 4 or nightly version of the Core Rule Set (CRS)
     use_right_crs_version = False
     use_modsecurity_crs_plugins = False
@@ -152,6 +158,9 @@ try:
                     LOGGER.error(f"Exception while decompressing plugin(s) from {crs_plugin_url} :\n{e}")
                     continue
 
+            plugin_name = ""
+            plugin_id = ""
+
             # Check if the plugins are valid, if they are already installed and if they need to be updated
             for plugin_config in temp_dir.rglob("**/*-config.conf"):
                 try:
@@ -203,6 +212,16 @@ try:
                     status = 2
                     continue
 
+            # * Patch the rules so we can extract the rule IDs when matching
+            try:
+                LOGGER.info(f"Patching Core Rule Set (CRS) plugin {plugin_name}...")
+                result = run([PATCH_SCRIPT.as_posix(), NEW_PLUGINS_DIR.joinpath(plugin_id).as_posix()], check=True)
+            except CalledProcessError as e:
+                LOGGER.error(f"Failed to patch Core Rule Set (CRS) plugin {plugin_name}: {e}")
+                sys_exit(1)
+
+            LOGGER.info(f"Successfully patched Core Rule Set (CRS) plugin {plugin_name}.")
+
             downloaded_plugins[crs_plugin_url] = installed_plugins.copy()
 
         service_plugins[service].update(installed_plugins)

src/common/core/modsecurity/misc/patch.sh

@@ -1,15 +1,32 @@
 #!/bin/bash
 
-rules_dir="./files/coreruleset-v$1/rules"
+# Validate and set the rules directory
+if [[ -z "$1" ]]; then
+  echo "Error: No argument provided. Specify a version number or directory path."
+  exit 1
+fi
 
+if [[ "$1" =~ ^[0-9]+$ ]]; then
+  rules_dir="./files/coreruleset-v$1/rules"
+else
+  rules_dir="$1"
+fi
+
+# Ensure the directory exists
+if [[ ! -d "$rules_dir" ]]; then
+  echo "Error: Rules directory '$rules_dir' does not exist."
+  exit 1
+fi
+
+# Define score types and variables
 score_types=("critical" "error" "warning" "notice")
 score_variables=("inbound" "outbound")
 
+# Process rule files
 find "$rules_dir" -type f -name "*.conf" | while read -r file; do
   for score_type in "${score_types[@]}"; do
     for score_variable in "${score_variables[@]}"; do
-      search_pattern="setvar:'tx.${score_variable}_anomaly_score_pl[0-9]=+%{tx.${score_type}_anomaly_score}'"
-      sed -i "/$search_pattern/s/\($search_pattern\)/\1,setvar:'tx.bunkerweb_rules=%{tx.bunkerweb_rules} %{rule.id}'/" "$file"
+      sed -i "/setvar:'tx.${score_variable}_anomaly_score_pl[0-9]=+%{tx.${score_type}_anomaly_score}'/s/\(setvar:'tx.${score_variable}_anomaly_score_pl[0-9]=+%{tx.${score_type}_anomaly_score}'\)/\1,setvar:'tx.bunkerweb_rules=%{tx.bunkerweb_rules} %{rule.id}'/" "$file"
     done
   done
   sed -i "/setvar:'tx.anomaly_score_pl[0-9]=+%{tx.critical_anomaly_score}'/s/\(setvar:'tx.anomaly_score_pl[0-9]=+%{tx.critical_anomaly_score}'\)/\1,setvar:'tx.bunkerweb_rules=%{tx.bunkerweb_rules} %{rule.id}'/" "$file"