Update owasp2haproxy.py

- Error Handling:
    - Added error handling for file operations, JSON parsing, and invalid rule structures.
    - Logs warnings for invalid rules instead of crashing.

- Path Handling:
    - Used pathlib.Path for better path manipulation and readability.
Made paths configurable via environment variables.

- Logging:
    - Replaced print() with Python's logging module for more flexible and structured logging.

- Input Validation:
    - Added checks for missing keys in the input JSON file.

- Rule Formatting:
    - Ensured proper formatting of HAProxy ACL rules.

- Output Directory Permissions:
    - Ensured the output directory is created with parents=True to handle nested directories.

- Code Structure:
    - Encapsulated the main logic in a main() function for better organization.
    - Added docstrings to functions for clarity.
This commit is contained in:
fab 2025-01-03 13:14:27 +01:00 committed by GitHub
parent 4655a25c4d
commit 2c1401c1cf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -1,23 +1,85 @@
import os import os
import json import json
import logging
from pathlib import Path
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler()],
)
# Constants (configurable via environment variables)
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", "waf_patterns/haproxy/")) # Output directory
INPUT_FILE = Path(os.getenv("INPUT_FILE", "owasp_rules.json")) # Input JSON file
OUTPUT_DIR = "waf_patterns/haproxy/"
def load_owasp_rules(file_path): def load_owasp_rules(file_path):
with open(file_path, "r") as f: """
return json.load(f) 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}")
raise
except json.JSONDecodeError:
logging.error(f"[!] Invalid JSON in file: {file_path}")
raise
except Exception as e:
logging.error(f"[!] Error loading OWASP rules: {e}")
raise
def generate_haproxy_conf(rules): def generate_haproxy_conf(rules):
os.makedirs(OUTPUT_DIR, exist_ok=True) """
config_file = os.path.join(OUTPUT_DIR, "waf.acl") Generate HAProxy ACL rules from OWASP rules.
"""
try:
# Ensure the output directory exists
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
logging.info(f"[+] Created or verified directory: {OUTPUT_DIR}")
# Define the output file path
config_file = OUTPUT_DIR / "waf.acl"
# Write HAProxy ACL rules to the file
with open(config_file, "w") as f:
f.write("# HAProxy WAF ACL rules\n\n")
for rule in rules:
try:
category = rule["category"]
pattern = rule["pattern"]
f.write(f"acl block_{category} hdr_sub(User-Agent) -i {pattern}\n")
f.write(f"http-request deny if block_{category}\n\n")
except KeyError as e:
logging.warning(f"[!] Skipping invalid rule (missing key: {e}): {rule}")
continue
logging.info(f"[+] HAProxy WAF rules generated at {config_file}")
except Exception as e:
logging.error(f"[!] Error generating HAProxy configuration: {e}")
raise
def main():
"""
Main function to execute the script.
"""
try:
logging.info("[*] Loading OWASP rules...")
owasp_rules = load_owasp_rules(INPUT_FILE)
logging.info(f"[*] Generating HAProxy WAF configs from {len(owasp_rules)} rules...")
generate_haproxy_conf(owasp_rules)
logging.info("[✔] HAProxy WAF configurations generated successfully.")
except Exception as e:
logging.critical(f"[!] Script failed: {e}")
exit(1)
with open(config_file, "w") as f:
f.write("# HAProxy WAF ACL rules\n")
for rule in rules:
f.write(f"acl block_{rule['category']} hdr_sub(User-Agent) -i {rule['pattern']}\n")
f.write(f"http-request deny if block_{rule['category']}\n")
print(f"[+] HAProxy WAF rules generated at {config_file}")
if __name__ == "__main__": if __name__ == "__main__":
owasp_rules = load_owasp_rules("owasp_rules.json") main()
generate_haproxy_conf(owasp_rules)