FixTheVuln

{
  "timestamp": "2026-01-15T10:30:45.123Z",
  "level": "INFO",
  "service": "auth-service",
  "environment": "production",
  "trace_id": "abc123def456",
  "span_id": "789xyz",
  "event": {
    "type": "authentication",
    "action": "login_success",
    "outcome": "success"
  },
  "actor": {
    "user_id": "user_12345",
    "username": "[email protected]",
    "ip_address": "192.168.1.100",
    "user_agent": "Mozilla/5.0...",
    "session_id": "sess_abc123"
  },
  "resource": {
    "type": "user_account",
    "id": "user_12345"
  },
  "metadata": {
    "mfa_method": "totp",
    "login_method": "password",
    "geo_location": "US-CA"
  }
}

import logging
import json
from datetime import datetime

class SecurityLogger:
    def __init__(self, service_name):
        self.service = service_name
        self.logger = logging.getLogger(service_name)

    def log_auth_event(self, action, outcome, user_id, ip_address, **kwargs):
        log_entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": "INFO" if outcome == "success" else "WARN",
            "service": self.service,
            "event": {
                "type": "authentication",
                "action": action,
                "outcome": outcome
            },
            "actor": {
                "user_id": user_id,
                "ip_address": ip_address
            },
            "metadata": kwargs
        }
        self.logger.info(json.dumps(log_entry))

# Usage
security_log = SecurityLogger("auth-service")
security_log.log_auth_event(
    action="login_attempt",
    outcome="success",
    user_id="user_123",
    ip_address="192.168.1.100",
    mfa_method="totp"
)

# Example: Brute Force Detection (Splunk SPL)
index=auth sourcetype=auth_logs action=login_failure
| stats count by src_ip, user
| where count > 5
| table src_ip, user, count

# Example: Impossible Travel (pseudo-code)
SELECT user_id,
       prev_location, current_location,
       time_diff_minutes,
       distance_km
FROM login_events
WHERE time_diff_minutes < (distance_km / 900) * 60
  AND distance_km > 500

# Example: Data Exfiltration (high volume download)
index=app_logs event_type=data_export
| stats sum(record_count) as total_records by user, _time span=1h
| where total_records > 10000

# Example: Privilege Escalation
index=auth_logs event_type=role_change
| where new_role IN ("admin", "superuser", "root")
| table _time, user, old_role, new_role, changed_by

# Filebeat configuration (ship to Elasticsearch)
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  json.keys_under_root: true
  json.add_error_key: true

output.elasticsearch:
  hosts: ["https://elasticsearch:9200"]
  username: "filebeat_internal"
  password: "${ES_PASSWORD}"
  ssl.certificate_authorities: ["/etc/pki/ca.crt"]

# Fluentd configuration

  @type tail
  path /var/log/app/*.log
  pos_file /var/log/fluentd/app.log.pos
  tag app.logs
  
    @type json
  



  @type elasticsearch
  host elasticsearch
  port 9200
  logstash_format true
  logstash_prefix app-logs

import re

def mask_sensitive_data(log_entry):
    # Mask credit card numbers
    log_entry = re.sub(
        r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
        '****-****-****-XXXX',
        log_entry
    )

    # Mask email addresses (partial)
    log_entry = re.sub(
        r'(\w{2})\w*@(\w{2})\w*\.(\w+)',
        r'\1***@\2***.\3',
        log_entry
    )

    # Mask API keys
    log_entry = re.sub(
        r'(api[_-]?key["\s:=]+)["\']?[\w-]{20,}["\']?',
        r'\1[REDACTED]',
        log_entry,
        flags=re.IGNORECASE
    )

    return log_entry