mTLS Service Mesh Audit Trail Logging

mTLS Audit Trail

mTLS Service Mesh Audit Trail Logging Certificate Management Access Log Compliance SIEM PCI-DSS HIPAA SOC2 Zero Trust Encryption Identity Verification

mTLS Configuration

# === mTLS in Service Mesh ===

# Istio — Strict mTLS
# apiVersion: security.istio.io/v1beta1
# kind: PeerAuthentication
# metadata:
#   name: default
#   namespace: istio-system
# spec:
#   mtls:
#     mode: STRICT
#
# ---
# # Authorization Policy — Only allow specific services
# apiVersion: security.istio.io/v1beta1
# kind: AuthorizationPolicy
# metadata:
#   name: order-service-policy
#   namespace: production
# spec:
#   selector:
#     matchLabels:
#       app: order-service
#   rules:
#     - from:
#         - source:
#             principals: ["cluster.local/ns/production/sa/api-gateway"]
#             principals: ["cluster.local/ns/production/sa/payment-service"]
#       to:
#         - operation:
#             methods: ["GET", "POST"]
#             paths: ["/api/orders/*"]

# Linkerd — mTLS is automatic
# linkerd install | kubectl apply -f -
# kubectl annotate namespace production linkerd.io/inject=enabled
# linkerd viz edges deployment -n production  # Verify mTLS

# Certificate Rotation
# Istio: Auto-rotate every 24h (default)
# Linkerd: Auto-rotate every 24h (default)
# Custom CA:
# istioctl install --set values.pilot.env.PILOT_CERT_PROVIDER=custom
# kubectl create secret tls cacerts -n istio-system \
#   --cert=ca-cert.pem --key=ca-key.pem

from dataclasses import dataclass

@dataclass
class mTLSConfig:
    mesh: str
    mtls_mode: str
    cert_rotation: str
    ca_type: str
    identity_format: str

configs = [
    mTLSConfig("Istio", "STRICT/PERMISSIVE", "24h auto", "Built-in / Custom CA", "spiffe://cluster/ns/sa"),
    mTLSConfig("Linkerd", "Always on", "24h auto", "Built-in / cert-manager", "identity.linkerd.cluster.local"),
    mTLSConfig("Cilium", "Required/Optional", "Configurable", "Built-in / SPIFFE", "spiffe://cluster/ns/sa"),
    mTLSConfig("Consul", "Required/Optional", "72h auto", "Built-in Vault", "spiffe://consul/ns/sa"),
]

print("=== mTLS Configurations ===")
for c in configs:
    print(f"  [{c.mesh}] Mode: {c.mtls_mode}")
    print(f"    Cert Rotation: {c.cert_rotation} | CA: {c.ca_type}")
    print(f"    Identity: {c.identity_format}")

Audit Trail Setup

# === Audit Trail Logging ===

# Istio Access Log Configuration
# apiVersion: telemetry.istio.io/v1alpha1
# kind: Telemetry
# metadata:
#   name: mesh-default
#   namespace: istio-system
# spec:
#   accessLogging:
#     - providers:
#         - name: envoy
#       filter:
#         expression: "response.code >= 400 || connection.mtls == false"

# Envoy Access Log Format (JSON)
# {
#   "timestamp": "%START_TIME%",
#   "source_identity": "%DOWNSTREAM_PEER_SUBJECT%",
#   "source_ip": "%DOWNSTREAM_REMOTE_ADDRESS%",
#   "destination_service": "%UPSTREAM_CLUSTER%",
#   "method": "%REQ(:METHOD)%",
#   "path": "%REQ(X-ENVOY-ORIGINAL-PATH?:PATH)%",
#   "response_code": "%RESPONSE_CODE%",
#   "response_flags": "%RESPONSE_FLAGS%",
#   "duration_ms": "%DURATION%",
#   "bytes_sent": "%BYTES_SENT%",
#   "bytes_received": "%BYTES_RECEIVED%",
#   "mtls": "%DOWNSTREAM_TLS_VERSION%",
#   "user_agent": "%REQ(USER-AGENT)%",
#   "request_id": "%REQ(X-REQUEST-ID)%",
#   "trace_id": "%REQ(X-B3-TRACEID)%"
# }

# Fluentd/Fluent Bit — Ship to SIEM
# [INPUT]
#     Name              tail
#     Path              /var/log/istio-proxy/*.log
#     Parser            json
#     Tag               audit.*
#
# [FILTER]
#     Name              modify
#     Match             audit.*
#     Add               cluster production
#     Add               environment prod
#
# [OUTPUT]
#     Name              es
#     Match             audit.*
#     Host              elasticsearch
#     Port              9200
#     Index             audit-trail
#     Type              _doc

@dataclass
class AuditEvent:
    event_type: str
    log_fields: str
    alert_on: str
    retention: str
    compliance: str

events = [
    AuditEvent("Auth Failure", "source identity method path 403", "3+ failures/min", "1 year", "PCI-DSS SOC2"),
    AuditEvent("mTLS Bypass", "source_ip no_cert method path", "Any occurrence", "1 year", "All"),
    AuditEvent("Privilege Escalation", "identity method admin_path 200", "Unexpected identity", "7 years", "HIPAA SOC2"),
    AuditEvent("Data Access", "identity resource_type action", "Bulk access pattern", "6 years", "HIPAA GDPR"),
    AuditEvent("Config Change", "admin identity resource change", "Any change", "1 year", "SOC2 ISO27001"),
    AuditEvent("Service Discovery", "identity scanned_services count", "> 10 services/min", "1 year", "SOC2"),
]

print("\n=== Audit Events ===")
for e in events:
    print(f"  [{e.event_type}] Alert: {e.alert_on}")
    print(f"    Fields: {e.log_fields}")
    print(f"    Retention: {e.retention} | Compliance: {e.compliance}")

SIEM Integration

# === SIEM Integration & Monitoring ===

# Elasticsearch Index Template
# PUT _index_template/audit-trail
# {
#   "index_patterns": ["audit-trail-*"],
#   "template": {
#     "settings": {
#       "number_of_shards": 3,
#       "number_of_replicas": 1,
#       "index.lifecycle.name": "audit-retention",
#       "index.lifecycle.rollover_alias": "audit-trail"
#     },
#     "mappings": {
#       "properties": {
#         "timestamp": { "type": "date" },
#         "source_identity": { "type": "keyword" },
#         "method": { "type": "keyword" },
#         "path": { "type": "keyword" },
#         "response_code": { "type": "integer" },
#         "mtls": { "type": "keyword" },
#         "duration_ms": { "type": "float" }
#       }
#     }
#   }
# }

# ILM Policy — Retention
# PUT _ilm/policy/audit-retention
# {
#   "policy": {
#     "phases": {
#       "hot": { "actions": { "rollover": { "max_size": "50gb", "max_age": "1d" } } },
#       "warm": { "min_age": "30d", "actions": { "shrink": { "number_of_shards": 1 } } },
#       "cold": { "min_age": "90d", "actions": { "freeze": {} } },
#       "delete": { "min_age": "365d", "actions": { "delete": {} } }
#     }
#   }
# }

@dataclass
class SIEMDashboard:
    panel: str
    query: str
    visualization: str
    alert: str

dashboards = [
    SIEMDashboard("Auth Failures", "response_code:403 AND source_identity:*", "Time Series + Heatmap", "> 10/min per identity"),
    SIEMDashboard("mTLS Status", "mtls:* | stats by mtls", "Pie Chart", "Any non-TLS connection"),
    SIEMDashboard("Top Callers", "source_identity:* | top 10", "Bar Chart", "Unknown identity"),
    SIEMDashboard("Latency P99", "duration_ms:* | percentile 99", "Line Chart", "> 5s"),
    SIEMDashboard("Error Rate", "response_code:[400 TO 599]", "Gauge", "> 5% error rate"),
    SIEMDashboard("Data Access", "path:/api/sensitive/* | stats", "Table", "Bulk access pattern"),
]

print("SIEM Dashboard Panels:")
for d in dashboards:
    print(f"  [{d.panel}] Viz: {d.visualization}")
    print(f"    Query: {d.query}")
    print(f"    Alert: {d.alert}")

security_checklist = {
    "mTLS Strict": "ทุก Namespace ต้อง STRICT mode",
    "AuthZ Policy": "Least privilege per service",
    "Audit Logging": "ทุก Request บันทึก Source Destination",
    "Log Immutability": "Write-once storage WORM",
    "SIEM Integration": "Real-time forwarding to SIEM",
    "Alert Response": "Runbook สำหรับทุก Alert Type",
    "Cert Rotation": "Auto-rotate ทุก 24 ชั่วโมง",
    "Compliance Audit": "Review ทุกไตรมาส",
}

print(f"\n\nSecurity Checklist:")
for k, v in security_checklist.items():
    print(f"  [{k}]: {v}")

เคล็ดลับ

• STRICT: ใช้ mTLS STRICT mode เสมอใน Production
• Identity: ใช้ Service Account Identity ไม่ใช่ IP
• Immutable: Audit Log ต้อง Immutable ลบแก้ไขไม่ได้
• SIEM: ส่ง Log ไป SIEM Real-time สำหรับ Alert
• Review: Review Audit Log ทุกสัปดาห์ Compliance ทุกไตรมาส

mTLS คืออะไร

mutual TLS Client Server Certificate ยืนยันทั้งสองฝ่าย Sidecar Proxy เข้ารหัส Man-in-the-Middle Zero Trust Service Mesh Auto

สรุป

mTLS Service Mesh Audit Trail Logging Certificate Zero Trust Access Log Compliance PCI-DSS HIPAA SOC2 SIEM ELK Immutable Identity Production Security