Building a Production-Ready OCSF Security Data Pipeline: Wazuh to Tenzir Integration Guide#

1
{
2
  "metadata": {
3
    "version": "1.0.0",
4
    "product": {
5
      "name": "Wazuh",
6
      "vendor_name": "Wazuh Inc.",
7
      "version": "4.8.0"
8
    },
9
    "profiles": ["host", "security_control"],
10
    "event_code": "authentication"
11
  },
12
  "authentication": {
13
    "activity_id": 1,
14
    "activity_name": "Logon",
15
    "actor": {
16
      "user": {
17
        "name": "john.doe",
18
        "uid": "S-1-5-21-123456",
19
        "type": "User",
20
        "type_id": 1
21
      }
22
    },
23
    "device": {
24
      "hostname": "WORKSTATION-01",
25
      "ip": "192.168.1.100",
26
      "os": {
27
        "name": "Windows",
28
        "version": "10.0.19044"
29
      }
30
    },
31
    "time": 1704538800000,
32
    "severity_id": 1,
33
    "status": "Success",
34
    "status_id": 1
35
  }
36
}

1
<ossec_config>
2
  <logging>
3
    <log_format>json</log_format>
4
    <jsonout_output>yes</jsonout_output>
5
  </logging>
6

7
  <global>
8
    <jsonout_output>yes</jsonout_output>
9
    <alerts_log>yes</alerts_log>
10
    <logall>yes</logall>
11
    <logall_json>yes</logall_json>
12
  </global>
13
</ossec_config>

1
<ossec_config>
2
  <syslog_output>
3
    <server>tenzir-host</server>
4
    <port>514</port>
5
    <format>json</format>
6
    <level>1</level>
7
  </syslog_output>
8
</ossec_config>

1
<group name="authentication,ocsf">
2
  <rule id="100001" level="5">
3
    <if_sid>5715</if_sid>
4
    <field name="win.eventdata.logonType">^3$</field>
5
    <description>Windows Network Logon - OCSF Class 3002</description>
6
    <options>no_full_log</options>
7
    <group>authentication_success,ocsf_3002</group>
8
  </rule>
9

10
  <rule id="100002" level="10">
11
    <if_sid>5716</if_sid>
12
    <field name="win.eventdata.status">^0xC000006D$</field>
13
    <description>Windows Logon Failed - Bad Username - OCSF Class 3002</description>
14
    <group>authentication_failed,ocsf_3002</group>
15
  </rule>
16
</group>

1
tenzir:
2
  # Data directory
3
  state-directory: /var/lib/tenzir
4

5
  # Network settings
6
  endpoint: 0.0.0.0:5158
7

8
  # Performance tuning
9
  max-partition-size: 1048576
10
  max-resident-partitions: 10
11
  max-taste-partitions: 5
12

13
  # Plugins
14
  plugins:
15
    - ocsf
16
    - wazuh
17
    - parquet

1
// Wazuh to OCSF Pipeline
2
// This pipeline transforms Wazuh alerts to OCSF format
3

4
// Stage 1: Ingest Wazuh JSON
5
read_json file=/var/ossec/logs/alerts/alerts.json
6
| where rule.groups has "ocsf_3002"  // Filter authentication events
7

8
// Stage 2: Map Wazuh fields to OCSF
9
| put ocsf = {
10
    metadata: {
11
      version: "1.0.0",
12
      product: {
13
        name: "Wazuh",
14
        vendor_name: "Wazuh Inc.",
15
        version: "4.8.0"
16
      },
17
      profiles: ["host"],
18
      logged_time: timestamp,
19
      original_time: timestamp
20
    },
21
    class_uid: 3002,
22
    class_name: "Authentication",
23
    category_uid: 3,
24
    category_name: "Identity & Access Management",
25
    severity_id: if (rule.level <= 3) {1}
26
                 else if (rule.level <= 6) {2}
27
                 else if (rule.level <= 9) {3}
28
                 else {4},
29
    activity_id: if (rule.groups has "authentication_success") {1} else {2},
30
    activity_name: if (rule.groups has "authentication_success") {"Logon"} else {"Logon Failed"},
31
    time: timestamp,
32
    actor: {
33
      user: {
34
        name: data.win.eventdata.targetUserName,
35
        domain: data.win.eventdata.targetDomainName,
36
        type: "User",
37
        type_id: 1
38
      },
39
      session: {
40
        uid: data.win.eventdata.logonGuid
41
      }
42
    },
43
    device: {
44
      hostname: agent.name,
45
      ip: agent.ip,
46
      os: {
47
        name: "Windows",
48
        type: "Windows",
49
        type_id: 100
50
      }
51
    },
52
    logon_type: data.win.eventdata.logonType,
53
    status: if (rule.groups has "authentication_success") {"Success"} else {"Failure"},
54
    status_id: if (rule.groups has "authentication_success") {1} else {2}
55
  }
56

57
// Stage 3: Apply OCSF validation
58
| ocsf::derive
59
| ocsf::apply
60
| ocsf::trim profile=standard
61

62
// Stage 4: Output to multiple destinations
63
| publish "ocsf-events"

1
// Advanced Wazuh to OCSF Pipeline with Enrichment
2

3
// Define lookup tables
4
let threat_intel = read_csv file=/etc/tenzir/threat_intel.csv
5
let asset_inventory = read_parquet file=/etc/tenzir/assets.parquet
6

7
// Main pipeline
8
subscribe "wazuh-raw"
9
| read_json
10

11
// Enrich with threat intelligence
12
| enrich threat_intel on ip = src_ip
13
| put threat_score = if (threat_intel.reputation == "malicious") {100}
14
                     else if (threat_intel.reputation == "suspicious") {50}
15
                     else {0}
16

17
// Enrich with asset information
18
| enrich asset_inventory on hostname = agent.name
19
| put asset_criticality = asset_inventory.criticality ?? "medium"
20

21
// Transform to OCSF with enrichments
22
| put ocsf = {
23
    // ... base OCSF fields ...
24
    enrichments: [
25
      {
26
        name: "threat_intelligence",
27
        type: "reputation",
28
        value: threat_intel.reputation,
29
        provider: "internal_ti"
30
      },
31
      {
32
        name: "asset_context",
33
        type: "criticality",
34
        value: asset_criticality,
35
        provider: "cmdb"
36
      }
37
    ],
38
    risk_score: threat_score,
39
    observables: [
40
      {
41
        name: "source_ip",
42
        type: "ip_address",
43
        type_id: 2,
44
        value: src_ip,
45
        reputation: {
46
          score: threat_score,
47
          provider: "internal_ti"
48
        }
49
      }
50
    ]
51
  }
52

53
// Validate and optimize
54
| ocsf::derive
55
| ocsf::apply
56
| ocsf::trim profile=detection
57

58
// Route based on severity
59
| where ocsf.severity_id >= 3
60
| publish "ocsf-high-priority"
61

62
// Archive all events
63
| write_parquet file=/data/ocsf/archive/authentication.parquet

1
// OCSF Class 3002: Authentication Activity
2
export auth_mapper = function(event) {
3
  return {
4
    class_uid: 3002,
5
    class_name: "Authentication",
6
    activity_id: switch {
7
      event.action == "login_success" => 1,  // Logon
8
      event.action == "logout" => 2,         // Logoff
9
      event.action == "login_failed" => 3,   // Authentication Failed
10
      default => 0                           // Unknown
11
    },
12
    actor: {
13
      user: {
14
        name: event.username,
15
        uid: event.user_id,
16
        type: "User",
17
        type_id: 1,
18
        credential_uid: event.session_id
19
      },
20
      process: if (event.process_name != null) {{
21
        name: event.process_name,
22
        pid: event.process_id
23
      }} else {null}
24
    },
25
    auth_protocol: event.auth_method,
26
    auth_protocol_id: switch {
27
      event.auth_method == "NTLM" => 1,
28
      event.auth_method == "Kerberos" => 2,
29
      event.auth_method == "LDAP" => 3,
30
      event.auth_method == "OAuth2" => 4,
31
      default => 99
32
    },
33
    dst_endpoint: {
34
      hostname: event.target_host,
35
      ip: event.target_ip,
36
      port: event.target_port
37
    },
38
    logon_type: event.logon_type,
39
    logon_type_id: event.logon_type_id,
40
    response_time: event.duration_ms,
41
    session: {
42
      uid: event.session_id,
43
      created_time: event.session_start,
44
      is_remote: event.is_remote
45
    },
46
    src_endpoint: {
47
      hostname: event.source_host,
48
      ip: event.source_ip,
49
      location: {
50
        city: event.geo_city,
51
        country: event.geo_country,
52
        coordinates: [event.geo_lon, event.geo_lat]
53
      }
54
    },
55
    status: event.status,
56
    status_code: event.status_code,
57
    status_detail: event.error_message,
58
    status_id: if (event.success) {1} else {2}
59
  }
60
}

1
// OCSF Class 4001: Network Activity
2
export network_mapper = function(event) {
3
  return {
4
    class_uid: 4001,
5
    class_name: "Network Activity",
6
    activity_id: switch {
7
      event.action == "allowed" => 1,    // Allowed
8
      event.action == "denied" => 2,     // Denied
9
      event.action == "dropped" => 3,    // Dropped
10
      default => 0                       // Unknown
11
    },
12
    connection_info: {
13
      direction: event.direction,
14
      direction_id: switch {
15
        event.direction == "inbound" => 1,
16
        event.direction == "outbound" => 2,
17
        event.direction == "lateral" => 3,
18
        default => 0
19
      },
20
      protocol_num: event.protocol_number,
21
      protocol_name: event.protocol,
22
      tcp_flags: event.tcp_flags
23
    },
24
    dst_endpoint: {
25
      hostname: event.dst_hostname,
26
      ip: event.dst_ip,
27
      port: event.dst_port,
28
      mac: event.dst_mac,
29
      interface_name: event.dst_interface
30
    },
31
    src_endpoint: {
32
      hostname: event.src_hostname,
33
      ip: event.src_ip,
34
      port: event.src_port,
35
      mac: event.src_mac,
36
      interface_name: event.src_interface
37
    },
38
    traffic: {
39
      bytes: event.bytes_total,
40
      bytes_in: event.bytes_received,
41
      bytes_out: event.bytes_sent,
42
      packets: event.packets_total,
43
      packets_in: event.packets_received,
44
      packets_out: event.packets_sent
45
    },
46
    duration: event.duration_ms,
47
    start_time: event.flow_start,
48
    end_time: event.flow_end
49
  }
50
}

1
// OCSF Class 1001: File Activity
2
export file_mapper = function(event) {
3
  return {
4
    class_uid: 1001,
5
    class_name: "File Activity",
6
    activity_id: switch {
7
      event.action == "create" => 1,
8
      event.action == "read" => 2,
9
      event.action == "update" => 3,
10
      event.action == "delete" => 4,
11
      event.action == "rename" => 5,
12
      event.action == "attributes_modified" => 6,
13
      event.action == "permissions_modified" => 7,
14
      default => 0
15
    },
16
    actor: {
17
      process: {
18
        name: event.process_name,
19
        pid: event.process_id,
20
        file: {
21
          path: event.process_path,
22
          hash: {
23
            algorithm: "SHA256",
24
            value: event.process_hash
25
          }
26
        }
27
      },
28
      user: {
29
        name: event.username,
30
        uid: event.user_id
31
      }
32
    },
33
    file: {
34
      path: event.file_path,
35
      name: event.file_name,
36
      parent_folder: event.file_directory,
37
      type: event.file_type,
38
      type_id: map_file_type(event.file_type),
39
      size: event.file_size,
40
      hash: if (event.file_hash != null) {{
41
        algorithm: "SHA256",
42
        value: event.file_hash
43
      }} else {null},
44
      modified_time: event.file_mtime,
45
      accessed_time: event.file_atime,
46
      created_time: event.file_ctime,
47
      is_system: event.is_system_file,
48
      security_descriptor: event.file_acl
49
    },
50
    device: {
51
      hostname: event.hostname,
52
      os: {
53
        name: event.os_name,
54
        version: event.os_version
55
      }
56
    }
57
  }
58
}

1
# docker-compose.yml for HA Tenzir deployment
2
version: '3.8'
3

4
services:
5
  tenzir-node1:
6
    image: tenzir/tenzir:latest
7
    container_name: tenzir-node1
8
    environment:
9
      - TENZIR_ENDPOINT=0.0.0.0:5158
10
      - TENZIR_NODE_ID=node1
11
      - TENZIR_CLUSTER_ENDPOINTS=tenzir-node2:5158,tenzir-node3:5158
12
    volumes:
13
      - tenzir-data1:/var/lib/tenzir
14
      - ./pipelines:/etc/tenzir/pipelines
15
    ports:
16
      - "5158:5158"
17
    networks:
18
      - tenzir-cluster
19
    deploy:
20
      resources:
21
        limits:
22
          cpus: '4'
23
          memory: 8G
24
        reservations:
25
          cpus: '2'
26
          memory: 4G
27

28
  tenzir-node2:
29
    image: tenzir/tenzir:latest
30
    container_name: tenzir-node2
31
    environment:
32
      - TENZIR_ENDPOINT=0.0.0.0:5158
33
      - TENZIR_NODE_ID=node2
34
      - TENZIR_CLUSTER_ENDPOINTS=tenzir-node1:5158,tenzir-node3:5158
35
    volumes:
36
      - tenzir-data2:/var/lib/tenzir
37
      - ./pipelines:/etc/tenzir/pipelines
38
    networks:
39
      - tenzir-cluster
40
    deploy:
41
      resources:
42
        limits:
43
          cpus: '4'
44
          memory: 8G
45

46
  tenzir-node3:
47
    image: tenzir/tenzir:latest
48
    container_name: tenzir-node3
49
    environment:
50
      - TENZIR_ENDPOINT=0.0.0.0:5158
51
      - TENZIR_NODE_ID=node3
52
      - TENZIR_CLUSTER_ENDPOINTS=tenzir-node1:5158,tenzir-node2:5158
53
    volumes:
54
      - tenzir-data3:/var/lib/tenzir
55
      - ./pipelines:/etc/tenzir/pipelines
56
    networks:
57
      - tenzir-cluster
58
    deploy:
59
      resources:
60
        limits:
61
          cpus: '4'
62
          memory: 8G
63

64
  haproxy:
65
    image: haproxy:2.9-alpine
66
    container_name: tenzir-lb
67
    volumes:
68
      - ./haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg:ro
69
    ports:
70
      - "80:80"
71
      - "443:443"
72
      - "5158:5158"
73
    networks:
74
      - tenzir-cluster
75
    depends_on:
76
      - tenzir-node1
77
      - tenzir-node2
78
      - tenzir-node3
79

80
volumes:
81
  tenzir-data1:
82
  tenzir-data2:
83
  tenzir-data3:
84

85
networks:
86
  tenzir-cluster:
87
    driver: bridge

1
// Optimized pipeline with parallel processing
2
export optimized_pipeline = (
3
  // Use parallel processing for high-volume streams
4
  read_json file=/var/ossec/logs/alerts/alerts.json
5
  | batch 1000
6
  | parallel apply=transform_to_ocsf
7

8
  // Implement smart caching
9
  | cache key=hash(rule.id, agent.id) ttl=300s
10

11
  // Compress before storage
12
  | compress algorithm=zstd level=3
13

14
  // Partition by time for efficient queries
15
  | partition by=floor(time, 1h)
16
  | write_parquet file="/data/ocsf/{partition}/events.parquet"
17
)
18

19
// Resource-aware processing
20
export adaptive_pipeline = (
21
  read_json
22
  | measure cpu_usage
23
  | if (cpu_usage > 0.8) {
24
      // Reduce processing when under load
25
      sample rate=0.5
26
      | ocsf::trim profile=minimal
27
    } else {
28
      // Full processing when resources available
29
      ocsf::derive
30
      | ocsf::apply
31
      | enrich_all
32
    }
33
  | write
34
)

1
# prometheus.yml configuration
2
global:
3
  scrape_interval: 15s
4

5
scrape_configs:
6
  - job_name: 'tenzir'
7
    static_configs:
8
      - targets:
9
        - 'tenzir-node1:9090'
10
        - 'tenzir-node2:9090'
11
        - 'tenzir-node3:9090'
12

13
  - job_name: 'wazuh'
14
    static_configs:
15
      - targets: ['wazuh-manager:55000']
16

17
# Alert rules
18
groups:
19
  - name: pipeline_health
20
    rules:
21
      - alert: PipelineBacklog
22
        expr: tenzir_pipeline_backlog_events > 10000
23
        for: 5m
24
        annotations:
25
          summary: "Pipeline backlog growing"
26

27
      - alert: TransformationErrors
28
        expr: rate(tenzir_ocsf_transform_errors[5m]) > 0.01
29
        annotations:
30
          summary: "OCSF transformation errors detected"
31

32
      - alert: HighMemoryUsage
33
        expr: tenzir_memory_usage_bytes / tenzir_memory_limit_bytes > 0.9
34
        for: 10m
35
        annotations:
36
          summary: "Tenzir memory usage critical"

1
// Direct integration with Amazon Security Lake
2
export asl_publisher = (
3
  subscribe "ocsf-events"
4
  | where ocsf.severity_id >= 2  // Only send medium+ severity
5

6
  // Add Security Lake required fields
7
  | put ocsf.metadata.product.feature = {
8
      name: "Wazuh SIEM",
9
      version: "4.8.0"
10
    }
11
  | put ocsf.metadata.labels = ["wazuh", "production", "regulated"]
12

13
  // Write to S3 in Parquet format
14
  | to_s3
15
      bucket="my-security-lake-bucket"
16
      prefix="ext/wazuh-ocsf/{year}/{month}/{day}/"
17
      format="parquet"
18
      compression="snappy"
19
      partition_by=["year", "month", "day"]
20
)

1
// Send OCSF events to Splunk
2
export splunk_forwarder = (
3
  subscribe "ocsf-events"
4
  | to_splunk
5
      url="https://splunk-hec.company.com:8088"
6
      token=env("SPLUNK_HEC_TOKEN")
7
      index="security_ocsf"
8
      sourcetype="ocsf:json"
9
)

1
// Index OCSF events in OpenSearch
2
export opensearch_indexer = (
3
  subscribe "ocsf-events"
4

5
  // Create time-based indices
6
  | put _index = strftime("ocsf-wazuh-%Y.%m.%d", ocsf.time)
7

8
  // Send to OpenSearch
9
  | to_opensearch
10
      nodes=["https://opensearch-node1:9200", "https://opensearch-node2:9200"]
11
      username=env("OPENSEARCH_USER")
12
      password=env("OPENSEARCH_PASS")
13
      ssl_verify=true
14
      bulk_size=1000
15
      flush_interval=5s
16
)

1
# Version management for OCSF schemas
2
class OCSFSchemaManager:
3
    def __init__(self):
4
        self.schema_versions = {
5
            "1.0.0": self.load_schema_v1_0_0(),
6
            "1.1.0": self.load_schema_v1_1_0()
7
        }
8

9
    def migrate_event(self, event, from_version, to_version):
10
        """Migrate events between OCSF versions"""
11
        if from_version == "1.0.0" and to_version == "1.1.0":
12
            # Add new required fields
13
            event['type_uid'] = self.calculate_type_uid(event)
14
            event['metadata']['processed_time'] = int(time.time() * 1000)
15

16
        return event
17

18
    def validate_event(self, event, version="1.1.0"):
19
        """Validate event against OCSF schema"""
20
        schema = self.schema_versions[version]
21
        return jsonschema.validate(event, schema)

1
# Tenzir performance configuration
2
performance_settings:
3
  # Buffer sizes for different event volumes
4
  low_volume:  # < 1K events/sec
5
    batch_size: 100
6
    buffer_size: 10000
7
    parallelism: 2
8

9
  medium_volume:  # 1K-10K events/sec
10
    batch_size: 1000
11
    buffer_size: 100000
12
    parallelism: 8
13

14
  high_volume:  # > 10K events/sec
15
    batch_size: 5000
16
    buffer_size: 500000
17
    parallelism: 16
18
    compression: true
19

20
  # Memory management
21
  memory_limits:
22
    heap_size: "8g"
23
    off_heap_size: "4g"
24
    direct_memory: "2g"

1
// Robust pipeline with error handling
2
export resilient_pipeline = (
3
  read_json file=/var/ossec/logs/alerts/alerts.json
4

5
  // Validate input
6
  | where timestamp != null && rule != null
7

8
  // Transform with error catching
9
  | try {
10
      transform_to_ocsf
11
      | ocsf::validate
12
    } catch {
13
      // Send failed events to error queue
14
      put error = {
15
        original_event: this,
16
        error_message: error.message,
17
        error_time: now()
18
      }
19
      | publish "ocsf-errors"
20
    }
21

22
  // Continue with valid events
23
  | where ocsf != null
24
  | publish "ocsf-valid"
25
)
26

27
// Error recovery pipeline
28
export error_recovery = (
29
  subscribe "ocsf-errors"
30
  | limit 1000  // Process in batches
31

32
  // Attempt to fix common issues
33
  | put fixed_event = fix_common_errors(error.original_event)
34

35
  // Retry transformation
36
  | try {
37
      select fixed_event
38
      | transform_to_ocsf
39
      | publish "ocsf-recovered"
40
    } catch {
41
      // Log permanently failed events
42
      write_json file="/var/log/tenzir/permanent_errors.json"
43
    }
44
)

1
-- SQL queries for OCSF data quality monitoring
2
-- Run these against your data lake
3

4
-- Check event distribution by class
5
SELECT
6
    class_name,
7
    class_uid,
8
    COUNT(*) as event_count,
9
    COUNT(*) * 100.0 / SUM(COUNT(*)) OVER() as percentage
10
FROM ocsf_events
11
WHERE date = CURRENT_DATE
12
GROUP BY class_name, class_uid
13
ORDER BY event_count DESC;
14

15
-- Monitor schema compliance
16
SELECT
17
    DATE_TRUNC('hour', time) as hour,
18
    COUNT(*) as total_events,
19
    SUM(CASE WHEN severity_id IS NULL THEN 1 ELSE 0 END) as missing_severity,
20
    SUM(CASE WHEN activity_id = 0 THEN 1 ELSE 0 END) as unknown_activity,
21
    SUM(CASE WHEN metadata.version != '1.1.0' THEN 1 ELSE 0 END) as wrong_version
22
FROM ocsf_events
23
WHERE date >= CURRENT_DATE - INTERVAL '7 days'
24
GROUP BY hour
25
ORDER BY hour DESC;
26

27
-- Identify enrichment gaps
28
SELECT
29
    class_name,
30
    COUNT(*) as total_events,
31
    SUM(CASE WHEN enrichments IS NULL THEN 1 ELSE 0 END) as not_enriched,
32
    SUM(CASE WHEN risk_score = 0 THEN 1 ELSE 0 END) as no_risk_score
33
FROM ocsf_events
34
WHERE date = CURRENT_DATE
35
GROUP BY class_name;

1
// Implement data masking for PII
2
export privacy_pipeline = (
3
  read_json
4
  | transform_to_ocsf
5

6
  // Mask sensitive fields
7
  | put ocsf.actor.user.email = mask_email(ocsf.actor.user.email)
8
  | put ocsf.src_endpoint.ip = mask_ip(ocsf.src_endpoint.ip, preserve_subnet=true)
9

10
  // Remove unnecessary PII
11
  | remove ocsf.actor.user.phone
12
  | remove ocsf.actor.user.address
13

14
  // Add compliance tags
15
  | put ocsf.metadata.labels = append(ocsf.metadata.labels, "gdpr_compliant")
16

17
  | write
18
)

1
# Tenzir RBAC configuration
2
rbac:
3
  roles:
4
    - name: ocsf_reader
5
      permissions:
6
        - read:ocsf-events
7
        - read:ocsf-archive
8

9
    - name: ocsf_admin
10
      permissions:
11
        - "*:ocsf-*"
12
        - manage:pipelines
13

14
  users:
15
    - name: wazuh_service
16
      roles: [ocsf_writer]
17

18
    - name: analyst_team
19
      roles: [ocsf_reader]
20

21
    - name: security_admin
22
      roles: [ocsf_admin]