Auditing and Logging

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

# /etc/fluent-bit/fluent-bit.conf
# Production configuration for security log collection
 
[SERVICE]
    # Daemon mode
    Daemon       Off
    Flush        1
    Log_Level    info
    
    # Parser configuration
    Parsers_File parsers.conf
    
    # Enable metrics endpoint for monitoring
    HTTP_Server  On
    HTTP_Listen  0.0.0.0
    HTTP_Port    2020
    
    # Buffer configuration for reliability
    storage.path              /var/log/fluent-bit-buffer/
    storage.sync              normal
    storage.checksum          off
    storage.backlog.mem_limit 50M
 
# ================================================
# INPUTS: What logs to collect
# ================================================
 
# System logs via journald
[INPUT]
    Name                systemd
    Tag                 system.*
    Systemd_Filter      _SYSTEMD_UNIT=sshd.service
    Systemd_Filter      _SYSTEMD_UNIT=auditd.service
    Systemd_Filter      _SYSTEMD_UNIT=sudo.service
    Read_From_Tail      On
    DB                  /var/log/fluent-bit-systemd.db
 
# Audit logs from auditd
[INPUT]
    Name                tail
    Tag                 audit.*
    Path                /var/log/audit/audit.log
    Parser              audit
    DB                  /var/log/fluent-bit-audit.db
    Mem_Buf_Limit       10MB
    Refresh_Interval    5
    
# Kernel messages
[INPUT]
    Name                kmsg
    Tag                 kernel
 
# Application logs (JSON format)
[INPUT]
    Name                tail
    Tag                 app.*
    Path                /var/log/app/*.json
    Parser              json
    DB                  /var/log/fluent-bit-app.db
    Mem_Buf_Limit       10MB
 
# ================================================
# FILTERS: Enrich and transform
# ================================================
 
# Add hostname and timestamp
[FILTER]
    Name                record_modifier
    Match               *
    Record              hostname ${HOSTNAME}
    Record              environment production
    Record              cluster main-cluster
 
# Parse nested fields
[FILTER]
    Name                parser
    Match               app.*
    Key_Name            log
    Parser              json
    Reserve_Data        True
 
# Add Kubernetes metadata (if running in K8s)
[FILTER]
    Name                kubernetes
    Match               kube.*
    Kube_URL            https://kubernetes.default.svc:443
    Kube_CA_File        /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    Kube_Token_File     /var/run/secrets/kubernetes.io/serviceaccount/token
    Merge_Log           On
 
# ================================================
# OUTPUTS: Where to send logs
# ================================================
 
# Primary: Elasticsearch for security logs
[OUTPUT]
    Name                es
    Match               audit.* system.*
    Host                elasticsearch.logging.svc
    Port                9200
    Index               security-logs
    Type                _doc
    Logstash_Format     On
    Logstash_Prefix     security
    Time_Key            @timestamp
    Include_Tag_Key     On
    Retry_Limit         5
    
    # TLS configuration
    tls                 On
    tls.verify          On
    tls.ca_file         /etc/fluent-bit/tls/ca.crt
    tls.crt_file        /etc/fluent-bit/tls/tls.crt
    tls.key_file        /etc/fluent-bit/tls/tls.key
    
    # Authentication
    HTTP_User           ${ES_USER}
    HTTP_Passwd         ${ES_PASSWORD}
 
# Secondary: Forward to remote syslog for compliance
[OUTPUT]
    Name                syslog
    Match               audit.*
    Host                syslog-archive.corp.internal
    Port                6514
    Mode                tls
    Syslog_Format       rfc5424
    Syslog_Hostname_key hostname
    Syslog_Message_key  message
    tls                 On
    tls.verify          On
    tls.ca_file         /etc/fluent-bit/tls/syslog-ca.crt
 
# Local backup (filesystem buffer)
[OUTPUT]
    Name                file
    Match               *
    Path                /var/log/fluent-bit-backup/
    Format              out_file

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

#!/bin/bash
# Create optimized Kafka topic for security logs
 
# Topic for security/audit logs
# - 30 day retention (720 hours)
# - 12 partitions for parallelism
# - Replication factor 3 for durability
kafka-topics.sh --create \
    --bootstrap-server kafka.internal:9092 \
    --topic security-logs \
    --partitions 12 \
    --replication-factor 3 \
    --config retention.ms=2592000000 \
    --config cleanup.policy=delete \
    --config min.insync.replicas=2 \
    --config compression.type=lz4 \
    --config segment.bytes=1073741824
 
# Topic for archival (longer retention, larger segments)
kafka-topics.sh --create \
    --bootstrap-server kafka.internal:9092 \
    --topic security-logs-archive \
    --partitions 6 \
    --replication-factor 3 \
    --config retention.ms=7776000000 \
    --config cleanup.policy=delete \
    --config segment.bytes=5368709120
 
# Verify topic configuration
kafka-topics.sh --describe \
    --bootstrap-server kafka.internal:9092 \
    --topic security-logs

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_age": "1d",
            "max_primary_shard_size": "50gb"
          },
          "set_priority": {
            "priority": 100
          }
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          },
          "allocate": {
            "require": {
              "data": "warm"
            },
            "number_of_replicas": 1
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "allocate": {
            "require": {
              "data": "cold"
            },
            "number_of_replicas": 0
          },
          "set_priority": {
            "priority": 0
          },
          "searchable_snapshot": {
            "snapshot_repository": "log-archive-s3"
          }
        }
      },
      "frozen": {
        "min_age": "90d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "log-archive-glacier"
          }
        }
      },
      "delete": {
        "min_age": "365d",
        "actions": {
          "wait_for_snapshot": {
            "policy": "compliance-snapshot-policy"
          },
          "delete": {}
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

# Loki configuration for security log storage
 
auth_enabled: false
 
server:
  http_listen_port: 3100
  grpc_listen_port: 9096
 
ingester:
  wal:
    enabled: true
    dir: /loki/wal
  lifecycler:
    ring:
      kvstore:
        store: inmemory
      replication_factor: 1
  chunk_idle_period: 30m
  max_chunk_age: 1h
  chunk_target_size: 1572864
  chunk_retain_period: 30s
 
schema_config:
  configs:
    - from: 2024-01-01
      store: tsdb
      object_store: s3
      schema: v13
      index:
        prefix: loki_index_
        period: 24h
 
storage_config:
  tsdb_shipper:
    active_index_directory: /loki/tsdb-index
    cache_location: /loki/tsdb-cache
    cache_ttl: 24h
  aws:
    s3: s3://us-east-1/loki-logs-bucket
    bucketnames: loki-logs-bucket
    region: us-east-1
    access_key_id: ${AWS_ACCESS_KEY_ID}
    secret_access_key: ${AWS_SECRET_ACCESS_KEY}
 
compactor:
  working_directory: /loki/compactor
  compaction_interval: 10m
  retention_enabled: true
  retention_delete_delay: 2h
  retention_delete_worker_count: 150
  delete_request_store: s3
 
limits_config:
  retention_period: 744h  # 31 days in hot storage
  ingestion_rate_mb: 50
  ingestion_burst_size_mb: 100
  per_stream_rate_limit: 10MB
  max_query_lookback: 0  # No limit
 
# Label cardinality limits (critical for performance)
  max_label_name_length: 1024
  max_label_value_length: 2048
  max_label_names_per_series: 30

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

# Log retention policy configuration
# This example uses a declarative format that could be implemented
# by various log management systems
 
metadata:
  policy_name: security-log-retention
  version: "2.0"
  last_reviewed: "2024-01-15"
  next_review: "2024-07-15"
  owner: security-team@company.com
  approved_by: ciso@company.com
 
# Global defaults
defaults:
  online_retention_days: 90
  archive_retention_years: 7
  storage_tier_progression:
    hot_to_warm_days: 7
    warm_to_cold_days: 30
    cold_to_archive_days: 90
  integrity_protection: required
  encryption: required
 
# Per-log-type policies (override defaults)
log_types:
  # Authentication and access logs
  authentication:
    description: "Login attempts, SSO events, MFA events"
    online_retention_days: 180
    archive_retention_years: 7
    compliance_frameworks:
      - PCI-DSS
      - SOX
      - HIPAA
    alerting_required: true
    
  # Authorization and permission events
  authorization:
    description: "Access decisions, permission changes"
    online_retention_days: 180
    archive_retention_years: 7
    
  # Privileged operations
  privileged_access:
    description: "sudo, admin actions, elevated permissions"
    online_retention_days: 365
    archive_retention_years: 7
    review_frequency: weekly
    
  # Application logs (non-security)
  application:
    description: "General application logs"
    online_retention_days: 30
    archive_retention_years: 1
    
  # Network traffic logs
  network:
    description: "Firewall logs, flow data"
    online_retention_days: 90
    archive_retention_years: 3
    
  # Personal data processing (GDPR)
  personal_data:
    description: "Logs containing PII processing"
    online_retention_days: 90
    archive_retention_years: 3
    special_handling:
      - erasure_requests_honored
      - anonymization_on_archive
      - access_restricted
 
# Legal hold configuration
legal_hold:
  enabled: true
  notification_recipients:
    - legal@company.com
    - ciso@company.com
  hold_prevents_deletion: true
  hold_duration_override: unlimited
 
# Deletion procedures
deletion:
  method: cryptographic_erasure
  verification: required
  certificate_generated: true
  audit_log_retained: true  # Log of what was deleted, kept longer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

# Linux auth.log
Jan 15 14:32:47 db-prod-01 sshd[12345]: Failed password for jsmith from 192.168.1.50 port 22 ssh2
 
# Windows Security Event (XML)
<Event>
  <EventID>4625</EventID>
  <TimeCreated SystemTime="2024-01-15T14:32:47.123Z"/>
  <Computer>DB-PROD-01</Computer>
  <EventData>
    <Data Name="TargetUserName">jsmith</Data>
    <Data Name="IpAddress">192.168.1.50</Data>
    <Data Name="LogonType">10</Data>
    <Data Name="FailureReason">%%2313</Data>
  </EventData>
</Event>
 
# AWS CloudTrail (JSON)
{
  "eventTime": "2024-01-15T14:32:47Z",
  "eventSource": "signin.amazonaws.com",
  "eventName": "ConsoleLogin",
  "userIdentity": {"userName": "jsmith"},
  "sourceIPAddress": "192.168.1.50",
  "responseElements": {"ConsoleLogin": "Failure"}
}
 
# Normalized output (Common Event Format)
{
  "timestamp": "2024-01-15T14:32:47.123456Z",
  "event_type": "authentication",
  "event_outcome": "failure",
  "source_host": "db-prod-01",
  "source_ip": "192.168.1.50",
  "user": "jsmith",
  "authentication_method": "password",
  "service": "ssh"
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

# Logstash pipeline for log normalization
 
input {
  # From Kafka topics
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["raw-logs"]
    group_id => "logstash-normalizer"
    codec => json
  }
}
 
filter {
  # ================================================
  # LINUX AUTH LOG PARSING
  # ================================================
  if [log_source] == "linux-auth" {
    grok {
      match => { 
        "message" => [
          # Failed password
          "%{SYSLOGTIMESTAMP:syslog_timestamp} %{HOSTNAME:source_host} sshd[%{NUMBER:pid}]: Failed password for %{USERNAME:user} from %{IP:source_ip} port %{NUMBER:source_port}",
          # Accepted password
          "%{SYSLOGTIMESTAMP:syslog_timestamp} %{HOSTNAME:source_host} sshd[%{NUMBER:pid}]: Accepted password for %{USERNAME:user} from %{IP:source_ip} port %{NUMBER:source_port}",
          # Invalid user
          "%{SYSLOGTIMESTAMP:syslog_timestamp} %{HOSTNAME:source_host} sshd[%{NUMBER:pid}]: Invalid user %{USERNAME:user} from %{IP:source_ip}"
        ]
      }
    }
    
    # Normalize to common schema
    mutate {
      add_field => {
        "[event][category]" => "authentication"
        "[event][type]" => "start"
        "[source][address]" => "%{source_ip}"
        "[user][name]" => "%{user}"
        "[host][name]" => "%{source_host}"
      }
    }
    
    # Determine outcome
    if "Failed" in [message] or "Invalid" in [message] {
      mutate { add_field => { "[event][outcome]" => "failure" } }
    } else {
      mutate { add_field => { "[event][outcome]" => "success" } }
    }
    
    # Parse timestamp
    date {
      match => ["syslog_timestamp", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
      target => "@timestamp"
    }
  }
  
  # ================================================
  # WINDOWS SECURITY EVENT PARSING
  # ================================================
  if [log_source] == "windows-security" {
    xml {
      source => "message"
      target => "winlog"
    }
    
    # Map Windows Event ID to event type
    translate {
      field => "[winlog][EventID]"
      destination => "[event][action]"
      dictionary => {
        "4624" => "logon_success"
        "4625" => "logon_failure"
        "4648" => "explicit_credential_logon"
        "4672" => "special_privileges_assigned"
        "4688" => "process_created"
        "4720" => "user_account_created"
      }
    }
    
    # Normalize fields
    mutate {
      rename => {
        "[winlog][Computer]" => "[host][name]"
        "[winlog][EventData][TargetUserName]" => "[user][name]"
        "[winlog][EventData][IpAddress]" => "[source][address]"
      }
      add_field => {
        "[event][category]" => "authentication"
      }
    }
    
    if [winlog][EventID] == "4625" {
      mutate { add_field => { "[event][outcome]" => "failure" } }
    } else if [winlog][EventID] == "4624" {
      mutate { add_field => { "[event][outcome]" => "success" } }
    }
  }
  
  # ================================================
  # ENRICH ALL EVENTS
  # ================================================
  
  # GeoIP enrichment for source IPs
  if [source][address] {
    geoip {
      source => "[source][address]"
      target => "[source][geo]"
    }
  }
  
  # Add processing metadata
  mutate {
    add_field => {
      "[ecs][version]" => "8.0"
      "[event][ingested]" => "%{@timestamp}"
    }
  }
  
  # Remove temporary fields
  mutate {
    remove_field => ["syslog_timestamp", "pid", "message", "winlog"]
  }
}
 
output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "security-logs-%{+YYYY.MM.dd}"
    user => "logstash"
    password => "${ES_PASSWORD}"
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

"""
Log sampling strategies for high-volume environments.
These techniques reduce volume while maintaining security visibility.
"""
 
import random
import hashlib
from typing import Dict, Any
 
class LogSampler:
    """
    Implements various sampling strategies for high-volume logs.
    """
    
    def __init__(self, default_rate: float = 1.0):
        """
        Initialize sampler with default rate (1.0 = 100% / no sampling).
        """
        self.default_rate = default_rate
        self.priority_override_rules = {}
    
    def add_priority_rule(self, field: str, value: str, rate: float):
        """
        Certain events should never be sampled (security-critical).
        Rate of 1.0 means always keep.
        """
        self.priority_override_rules[(field, value)] = rate
    
    def should_sample(self, event: Dict[str, Any]) -> bool:
        """
        Determine if an event should be included (not dropped).
        Returns True to keep, False to drop.
        """
        
        # Priority events are never dropped
        for (field, value), rate in self.priority_override_rules.items():
            if event.get(field) == value:
                return random.random() < rate
        
        # Apply default sampling rate
        return random.random() < self.default_rate
    
    def consistent_hash_sample(self, event: Dict[str, Any], 
                               key_field: str, rate: float) -> bool:
        """
        Use consistent hashing so the same key always gets same decision.
        Important for keeping all events from a single session/user.
        """
        key_value = event.get(key_field, "")
        hash_val = int(hashlib.sha256(str(key_value).encode()).hexdigest()[:8], 16)
        threshold = int(rate * 0xFFFFFFFF)
        return hash_val < threshold
 
 
# Example usage
sampler = LogSampler(default_rate=0.1)  # 10% default sampling
 
# Security events: NEVER sample
sampler.add_priority_rule("event.category", "authentication", 1.0)
sampler.add_priority_rule("event.outcome", "failure", 1.0)
sampler.add_priority_rule("event.category", "intrusion_detection", 1.0)
sampler.add_priority_rule("event.severity", "critical", 1.0)
sampler.add_priority_rule("event.severity", "high", 1.0)
 
# Debug logs: Sample aggressively
sampler.add_priority_rule("log.level", "debug", 0.01)  # 1% only
 
# Normal web traffic: Moderate sampling
sampler.add_priority_rule("event.category", "web", 0.05)  # 5%
 
def process_log(event: Dict[str, Any]):
    """Process a log event with sampling."""
    if sampler.should_sample(event):
        # Forward to storage
        send_to_storage(event)
    else:
        # Update sampling statistics
        update_sampling_metrics(event)