Wazuh Ruleset as Code (RaC): DevOps-Driven Security Detection Engineering#

1
name: Update Rulesets on SIEM
2

3
on:
4
  push:
5
    branches: [ "main" ]
6
    paths: ["**.xml"]
7
  workflow_dispatch:
8

9
jobs:
10
  DaaC:
11
    runs-on: ubuntu-latest
12

13
    steps:
14
      - name: Apply modified or new decoders and rules to SIEM
15
        uses: appleboy/ssh-action@v1.0.0
16
        with:
17
          host: ${{ secrets.HOST }}
18
          username: ${{ secrets.USERNAME }}
19
          key: ${{ secrets.SSH_KEY }}
20
          port: ${{ secrets.PORT }}
21
          script: |
22
            sudo bash -c '
23
            cd /var/ossec/etc/
24

25
            # Pull latest changes
26
            git pull origin main
27

28
            # Update permissions
29
            chown wazuh:wazuh /var/ossec/etc/decoders/* && chmod 660 /var/ossec/etc/decoders/*
30
            chown wazuh:wazuh /var/ossec/etc/rules/* && chmod 660 /var/ossec/etc/rules/*
31

32
            # Restart Wazuh Manager
33
            sudo systemctl restart wazuh-manager \
34
            && echo "✅ Ruleset apply SUCCESS! Wazuh manager restarted successfully." \
35
            || echo "❌ Ruleset apply FAILURE! Check ruleset for errors..."
36

37
            # Show status
38
            sudo systemctl status wazuh-manager -l --no-pager
39
            '

1
name: Check Rule ID Conflicts
2

3
on:
4
  pull_request:
5
    branches: [ "main" ]
6

7
jobs:
8
  check-rule-ids:
9
    runs-on: ubuntu-latest
10

11
    steps:
12
      - name: Checkout PR branch
13
        uses: actions/checkout@v3
14
        with:
15
          fetch-depth: 0
16

17
      - name: Set up Python
18
        uses: actions/setup-python@v4
19
        with:
20
          python-version: '3.10'
21

22
      - name: Fetch main branch
23
        run: git fetch origin main
24

25
      - name: Run rule ID conflict checker
26
        run: python check_rule_ids.py

1
import subprocess
2
import xml.etree.ElementTree as ET
3
from pathlib import Path
4
import sys
5
from collections import defaultdict, Counter
6

7
def run_git_command(args):
8
    """Execute git command and return output"""
9
    result = subprocess.run(args, capture_output=True, text=True, check=True)
10
    return result.stdout
11

12
def get_changed_rule_files():
13
    """Get list of changed XML rule files"""
14
    try:
15
        output = run_git_command(["git", "diff", "--name-status", "origin/main...HEAD"])
16
        changed_files = []
17

18
        for line in output.strip().splitlines():
19
            parts = line.strip().split(maxsplit=1)
20
            if len(parts) != 2:
21
                continue
22

23
            status, file_path = parts
24
            if file_path.startswith("rules/") and file_path.endswith(".xml"):
25
                changed_files.append((status, Path(file_path)))
26

27
        return changed_files
28
    except subprocess.CalledProcessError as e:
29
        print("❌ Failed to get changed files:", e)
30
        sys.exit(1)
31

32
def extract_rule_ids_from_xml(content):
33
    """Extract rule IDs from XML content"""
34
    ids = []
35
    try:
36
        # Wrap content to handle multiple root elements
37
        wrapped = f"<root>{content}</root>"
38
        root = ET.fromstring(wrapped)
39

40
        for rule in root.findall(".//rule"):
41
            rule_id = rule.get("id")
42
            if rule_id and rule_id.isdigit():
43
                ids.append(int(rule_id))
44

45
    except ET.ParseError as e:
46
        print(f"⚠️ XML Parse Error: {e}")
47

48
    return ids
49

50
def detect_duplicates(rule_ids):
51
    """Find duplicate rule IDs"""
52
    counter = Counter(rule_ids)
53
    return [rule_id for rule_id, count in counter.items() if count > 1]
54

55
def main():
56
    """Main validation logic"""
57
    changed_files = get_changed_rule_files()
58

59
    if not changed_files:
60
        print("✅ No rule files were changed in this PR.")
61
        return
62

63
    # Get existing rule IDs from main branch
64
    rule_id_to_files_main = get_rule_ids_per_file_in_main()
65

66
    print(f"🔍 Checking rule ID conflicts for: {[f.name for _, f in changed_files]}")
67

68
    for status, path in changed_files:
69
        print(f"\n🔎 Checking file: {path.name}")
70

71
        try:
72
            # Read file content
73
            dev_content = path.read_text()
74
            dev_ids = extract_rule_ids_from_xml(dev_content)
75
        except Exception as e:
76
            print(f"⚠️ Could not read {path.name}: {e}")
77
            continue
78

79
        # Check for internal duplicates
80
        duplicates = detect_duplicates(dev_ids)
81
        if duplicates:
82
            print(f"❌ Duplicate rule IDs in {path.name}: {sorted(duplicates)}")
83
            sys.exit(1)
84

85
        # Check for conflicts with existing rules
86
        if status == "A":  # New file
87
            conflicting_ids = set(dev_ids) & set(rule_id_to_files_main.keys())
88
            if conflicting_ids:
89
                print(f"❌ Conflicts detected: {sorted(conflicting_ids)}")
90
                sys.exit(1)
91

92
    print("\n✅ All rule file changes passed conflict checks.")
93

94
if __name__ == "__main__":
95
    main()

1
{
2
  "name": "Protect Main Branch",
3
  "target": "branch",
4
  "enforcement": "active",
5
  "conditions": {
6
    "ref_name": {
7
      "include": ["~DEFAULT_BRANCH"],
8
      "exclude": []
9
    }
10
  },
11
  "rules": [
12
    {
13
      "type": "pull_request",
14
      "parameters": {
15
        "required_approving_review_count": 1,
16
        "required_status_checks": ["Check Rule ID Conflicts"],
17
        "dismiss_stale_reviews_on_push": true,
18
        "require_code_owner_review": false
19
      }
20
    }
21
  ]
22
}

1
<decoder name="custom-app">
2
  <program_name>custom-app</program_name>
3
</decoder>
4

5
<decoder name="custom-app-login">
6
  <parent>custom-app</parent>
7
  <regex>User (\S+) login (\S+) from (\S+)</regex>
8
  <order>user, status, srcip</order>
9
</decoder>

1
<group name="custom_app,">
2
  <!-- Successful login -->
3
  <rule id="100100" level="3">
4
    <decoded_as>custom-app-login</decoded_as>
5
    <field name="status">success</field>
6
    <description>Custom app: Successful login for $(user) from $(srcip)</description>
7
    <group>authentication_success,</group>
8
  </rule>
9

10
  <!-- Failed login -->
11
  <rule id="100101" level="5">
12
    <decoded_as>custom-app-login</decoded_as>
13
    <field name="status">failed</field>
14
    <description>Custom app: Failed login for $(user) from $(srcip)</description>
15
    <group>authentication_failed,</group>
16
  </rule>
17

18
  <!-- Multiple failed logins -->
19
  <rule id="100102" level="10" frequency="5" timeframe="300">
20
    <if_matched_sid>100101</if_matched_sid>
21
    <same_source_ip />
22
    <description>Custom app: Multiple failed logins from $(srcip) - possible brute force</description>
23
    <group>authentication_failures,attack,</group>
24
    <mitre>
25
      <id>T1110</id>
26
    </mitre>
27
  </rule>
28
</group>

1
- name: Test Rules Syntax
2
  run: |
3
    # Validate XML syntax
4
    for file in rules/*.xml decoders/*.xml; do
5
      xmllint --noout "$file" || exit 1
6
    done
7

8
- name: Test Rule Logic
9
  run: |
10
    # Use ossec-logtest to validate rules
11
    echo "Test log entry" | /var/ossec/bin/ossec-logtest

1
- name: Notify Slack
2
  if: always()
3
  uses: 8398a7/action-slack@v3
4
  with:
5
    status: ${{ job.status }}
6
    text: 'Wazuh rules deployment ${{ job.status }}'
7
    webhook_url: ${{ secrets.SLACK_WEBHOOK }}

1
- name: Update JIRA
2
  uses: atlassian/gajira-transition@master
3
  with:
4
    issue: ${{ github.event.pull_request.title }}
5
    transition: "Deploy"

1
# Analyze rule effectiveness
2
from sklearn.metrics import precision_recall_fscore_support
3

4
def analyze_rule_performance(rule_id, alerts, false_positives):
5
    precision = len(alerts) / (len(alerts) + len(false_positives))
6
    return {
7
        'rule_id': rule_id,
8
        'precision': precision,
9
        'recommendation': 'tune' if precision < 0.8 else 'keep'
10
    }

1
# Generate rules from patterns
2
def generate_auth_rules(app_name, base_id):
3
    template = load_template('authentication_template.xml')
4
    return template.replace('TEMPLATE_APP', app_name)\
5
                  .replace('TEMPLATE_ID', str(base_id))