Update owasp2nginx.py

Fix [nginx issue](https://github.com/fabriziosalmi/patterns/issues/6)

owasp2nginx.py

@@ -1,100 +1,104 @@
 import json
 import os
+import re
 import logging
 from pathlib import Path
 from collections import defaultdict
 
-# Configure logging
 logging.basicConfig(
     level=logging.INFO,
     format="%(asctime)s - %(levelname)s - %(message)s",
     handlers=[logging.StreamHandler()],
 )
 
-# Constants (configurable via environment variables)
+INPUT_FILE = Path(os.getenv("INPUT_FILE", "owasp_rules.json"))
-INPUT_FILE = Path(os.getenv("INPUT_FILE", "owasp_rules.json"))  # Input JSON file
+OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "waf_patterns/nginx"))
-OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "waf_patterns/nginx"))  # Output directory
 
-# Ensure output directory exists
 OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
 
 
 def load_owasp_rules(file_path):
-    """
-    Load OWASP rules from a JSON file.
-    """
     try:
         with open(file_path, "r") as f:
             return json.load(f)
     except FileNotFoundError:
-        logging.error(f"[!] Input file not found: {file_path}")
+        logging.error(f"Input file not found: {file_path}")
         raise
     except json.JSONDecodeError:
-        logging.error(f"[!] Invalid JSON in file: {file_path}")
+        logging.error(f"Invalid JSON in file: {file_path}")
-        raise
-    except Exception as e:
-        logging.error(f"[!] Error loading OWASP rules: {e}")
         raise
 
 
+def validate_regex(pattern):
+    try:
+        re.compile(pattern)
+        return True
+    except re.error:
+        return False
+
+
+def sanitize_pattern(pattern):
+    if "@pmFromFile" in pattern or "!@eq" in pattern or "!@within" in pattern or "@lt" in pattern:
+        logging.warning(f"Skipping unsupported pattern: {pattern}")
+        return None
+
+    if pattern.startswith("@rx "):
+        sanitized_pattern = pattern.replace("@rx ", "").strip()
+        return sanitized_pattern if validate_regex(sanitized_pattern) else None
+
+    return pattern if validate_regex(pattern) else None
+
+
 def generate_nginx_waf(rules):
-    """
+    categorized_rules = defaultdict(set)
-    Generate Nginx WAF configuration files from OWASP rules.
-    """
-    categorized_rules = defaultdict(list)
 
-    # Group rules by category
+    # Group rules by category without filtering any categories
     for rule in rules:
-        try:
         category = rule.get("category", "generic").lower()
-            pattern = rule["pattern"]
+        pattern = rule.get("pattern")
-            categorized_rules[category].append(pattern)
-        except KeyError as e:
-            logging.warning(f"[!] Skipping invalid rule (missing key: {e}): {rule}")
-            continue
 
-    # Convert to Nginx conf files
+        sanitized_pattern = sanitize_pattern(pattern)
+        if sanitized_pattern:
+            categorized_rules[category].add(sanitized_pattern)
+        else:
+            logging.warning(f"Invalid or unsupported pattern skipped: {pattern}")
+
+    # Write Nginx configuration per category
     for category, patterns in categorized_rules.items():
         output_file = OUTPUT_DIR / f"{category}.conf"
-
         try:
             with open(output_file, "w") as f:
                 f.write(f"# Nginx WAF rules for {category.upper()}\n")
                 f.write("location / {\n")
                 f.write("    set $attack_detected 0;\n\n")
                 
-                # Write rules as regex checks
                 for pattern in patterns:
-                    f.write(f'    if ($request_uri ~* "{pattern}") {{\n')
+                    escaped_pattern = pattern.replace('"', '\\"')
+                    f.write(f'    if ($request_uri ~* "{escaped_pattern}") {{\n')
                     f.write("        set $attack_detected 1;\n")
                     f.write("    }\n\n")
 
-                # Block the request if an attack is detected
                 f.write("    if ($attack_detected = 1) {\n")
                 f.write("        return 403;\n")
                 f.write("    }\n")
                 f.write("}\n")
 
-            logging.info(f"[+] Generated {output_file} ({len(patterns)} patterns)")
+            logging.info(f"Generated {output_file} ({len(patterns)} patterns)")
         except IOError as e:
-            logging.error(f"[!] Failed to write to {output_file}: {e}")
+            logging.error(f"Failed to write {output_file}: {e}")
-            raise
 
 
 def main():
-    """
-    Main function to execute the script.
-    """
     try:
-        logging.info("[*] Loading OWASP rules...")
+        logging.info("Loading OWASP rules...")
         owasp_rules = load_owasp_rules(INPUT_FILE)
 
-        logging.info(f"[*] Generating Nginx WAF configs from {len(owasp_rules)} rules...")
+        logging.info(f"Generating Nginx WAF configs from {len(owasp_rules)} rules...")
         generate_nginx_waf(owasp_rules)
 
-        logging.info("[✔] Nginx WAF configurations generated successfully.")
+        logging.info("Nginx WAF configurations generated successfully.")
     except Exception as e:
-        logging.critical(f"[!] Script failed: {e}")
+        logging.critical(f"Script failed: {e}")
         exit(1)