{
  "timestamp": "2025-03-15T14:30:00.000Z",
  "level": "WARN",
  "service": "auth-service",
  "version": "2.1.0",
  "environment": "production",
  "traceId": "abc-123-def",
  "spanId": "span-456",
  "requestId": "req-456",
  "event": "login_failed",
  "userId": "user-789",
  "sourceIp": "203.0.113.50",
  "userAgent": "Mozilla/5.0...",
  "geoLocation": {
    "country": "JP",
    "region": "Tokyo"
  },
  "details": {
    "reason": "invalid_password",
    "attemptCount": 5,
    "accountLocked": false,
    "mfaEnabled": true
  },
  "metadata": {
    "hostname": "auth-pod-abc123",
    "containerId": "docker-xyz",
    "kubernetes": {
      "namespace": "production",
      "pod": "auth-service-7b9f4d-abc12"
    }
  }
}

# コード例1: 構造化ログライブラリの実装
import json
import logging
import sys
import traceback
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from contextvars import ContextVar
import uuid
 
# リクエストコンテキストをスレッドセーフに保持
_request_context: ContextVar[Dict[str, Any]] = ContextVar(
    'request_context', default={}
)
 
class StructuredLogFormatter(logging.Formatter):
    """構造化ログフォーマッタ
 
    ログをJSON形式で出力し、トレースID、リクエストID等の
    コンテキスト情報を自動的に付与する。
 
    セキュリティログでは以下の情報が特に重要:
    - timestamp: UTC形式のISO 8601タイムスタンプ
    - sourceIp: リクエスト元のIPアドレス
    - userId: 操作を行ったユーザーのID
    - event: セキュリティイベントの種類
    """
 
    # ログに含めてはいけない機密フィールド
    SENSITIVE_FIELDS = {
        'password', 'secret', 'token', 'api_key',
        'credit_card', 'ssn', 'cvv', 'pin',
        'authorization', 'cookie', 'session_id',
    }
 
    def __init__(self, service_name: str, environment: str):
        super().__init__()
        self.service_name = service_name
        self.environment = environment
 
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "service": self.service_name,
            "environment": self.environment,
            "logger": record.name,
            "message": record.getMessage(),
        }
 
        # リクエストコンテキストの追加
        ctx = _request_context.get()
        if ctx:
            log_entry.update({
                "traceId": ctx.get("trace_id"),
                "requestId": ctx.get("request_id"),
                "userId": ctx.get("user_id"),
                "sourceIp": ctx.get("source_ip"),
            })
 
        # 追加フィールドの処理（機密データのマスク）
        if hasattr(record, 'extra_fields'):
            sanitized = self._sanitize(record.extra_fields)
            log_entry["details"] = sanitized
 
        # 例外情報の追加
        if record.exc_info:
            log_entry["exception"] = {
                "type": record.exc_info[0].__name__,
                "message": str(record.exc_info[1]),
                "stacktrace": traceback.format_exception(*record.exc_info),
            }
 
        return json.dumps(log_entry, ensure_ascii=False, default=str)
 
    def _sanitize(self, data: Any) -> Any:
        """機密データをマスクする"""
        if isinstance(data, dict):
            return {
                k: "***REDACTED***" if k.lower() in self.SENSITIVE_FIELDS
                else self._sanitize(v)
                for k, v in data.items()
            }
        elif isinstance(data, list):
            return [self._sanitize(item) for item in data]
        return data
 
 
class SecurityLogger:
    """セキュリティイベント専用ロガー"""
 
    def __init__(self, service_name: str, environment: str = "production"):
        self.logger = logging.getLogger(f"security.{service_name}")
        self.logger.setLevel(logging.INFO)
 
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(
            StructuredLogFormatter(service_name, environment)
        )
        self.logger.addHandler(handler)
 
    def log_auth_event(self, event: str, user_id: str,
                       success: bool, **kwargs):
        """認証イベントをログに記録"""
        record = self.logger.makeRecord(
            self.logger.name, logging.INFO, "", 0,
            f"Authentication event: {event}", (), None
        )
        record.extra_fields = {
            "event_type": "authentication",
            "event": event,
            "user_id": user_id,
            "success": success,
            **kwargs,
        }
        self.logger.handle(record)
 
    def log_access_event(self, resource: str, action: str,
                         allowed: bool, **kwargs):
        """アクセス制御イベントをログに記録"""
        level = logging.INFO if allowed else logging.WARNING
        record = self.logger.makeRecord(
            self.logger.name, level, "", 0,
            f"Access event: {action} on {resource}", (), None
        )
        record.extra_fields = {
            "event_type": "access_control",
            "resource": resource,
            "action": action,
            "allowed": allowed,
            **kwargs,
        }
        self.logger.handle(record)
 
    def log_data_event(self, operation: str, data_type: str,
                       record_count: int, **kwargs):
        """データ操作イベントをログに記録"""
        record = self.logger.makeRecord(
            self.logger.name, logging.INFO, "", 0,
            f"Data event: {operation} on {data_type}", (), None
        )
        record.extra_fields = {
            "event_type": "data_operation",
            "operation": operation,
            "data_type": data_type,
            "record_count": record_count,
            **kwargs,
        }
        self.logger.handle(record)
 
 
# 使用例
sec_logger = SecurityLogger("auth-service")
sec_logger.log_auth_event(
    event="login_failed",
    user_id="user-789",
    success=False,
    reason="invalid_password",
    attempt_count=5,
    source_ip="203.0.113.50",
)

# /etc/fluent-bit/fluent-bit.conf
 
[SERVICE]
    Flush         5
    Daemon        Off
    Log_Level     info
    Parsers_File  parsers.conf
    # メトリクスの公開（Prometheus 形式）
    HTTP_Server   On
    HTTP_Listen   0.0.0.0
    HTTP_Port     2020
    # バッファ管理
    storage.path              /var/log/fluent-bit/buffer/
    storage.sync              normal
    storage.checksum          off
    storage.backlog.mem_limit 50M
 
# アプリケーションログ
[INPUT]
    Name              tail
    Path              /var/log/app/*.log
    Parser            json
    Tag               app.*
    Refresh_Interval  5
    Mem_Buf_Limit     50MB
    # バッファをファイルに永続化（データ損失防止）
    storage.type      filesystem
    Skip_Long_Lines   On
    DB                /var/log/fluent-bit/app.db
 
# OS syslog
[INPUT]
    Name              systemd
    Tag               system.*
    Systemd_Filter    _SYSTEMD_UNIT=sshd.service
    Read_From_Tail    On
 
# Kubernetes メタデータの付与
[FILTER]
    Name              kubernetes
    Match             app.*
    Kube_URL          https://kubernetes.default.svc
    Kube_CA_File      /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    Kube_Token_File   /var/run/secrets/kubernetes.io/serviceaccount/token
    Merge_Log         On
 
# メタデータの追加
[FILTER]
    Name              record_modifier
    Match             *
    Record hostname   ${HOSTNAME}
    Record env        production
    Record cluster    prod-ap-northeast-1
 
# 機密データのマスキング
[FILTER]
    Name              lua
    Match             app.*
    script            /etc/fluent-bit/scripts/mask_sensitive.lua
    call              mask_fields
 
# OpenSearch に送信
[OUTPUT]
    Name              opensearch
    Match             app.*
    Host              opensearch.internal.example.com
    Port              443
    TLS               On
    Index             app-logs
    Type              _doc
    Suppress_Type_Name On
    Logstash_Format   On
    Logstash_Prefix   app-logs
    # リトライ設定
    Retry_Limit       5
    # バッファリング
    Buffer_Size       512KB
    # 認証
    HTTP_User         fluent-bit
    HTTP_Passwd       ${OPENSEARCH_PASSWORD}
 
# S3 にバックアップ
[OUTPUT]
    Name              s3
    Match             *
    bucket            security-logs-archive
    region            ap-northeast-1
    total_file_size   100M
    upload_timeout    10m
    s3_key_format     /logs/%Y/%m/%d/$TAG/%H-%M-%S
    # 圧縮
    compression       gzip
    # サーバサイド暗号化
    use_put_object    On

-- /etc/fluent-bit/scripts/mask_sensitive.lua
-- 機密データをログ出力前にマスクする
 
local sensitive_patterns = {
    -- クレジットカード番号（16桁）
    {pattern = "%d%d%d%d[- ]?%d%d%d%d[- ]?%d%d%d%d[- ]?%d%d%d%d",
     replacement = "****-****-****-XXXX"},
    -- メールアドレス
    {pattern = "[%w%.%-]+@[%w%.%-]+%.%w+",
     replacement = "***@***.***"},
    -- AWS アクセスキー
    {pattern = "AKIA[0-9A-Z]%d%d%d%d%d%d%d%d%d%d%d%d%d%d%d%d",
     replacement = "AKIA****************"},
}
 
function mask_fields(tag, timestamp, record)
    local modified = false
    for key, value in pairs(record) do
        if type(value) == "string" then
            for _, pat in ipairs(sensitive_patterns) do
                local new_value = string.gsub(value, pat.pattern,
                                              pat.replacement)
                if new_value ~= value then
                    record[key] = new_value
                    modified = true
                end
            end
        end
    end
    if modified then
        return 1, timestamp, record
    end
    return 0, timestamp, record
end

# /etc/vector/vector.toml
 
[api]
enabled = true
address = "127.0.0.1:8686"
 
# ソース: アプリケーションログ
[sources.app_logs]
type = "file"
include = ["/var/log/app/*.log"]
read_from = "beginning"
 
# ソース: Kubernetes ログ
[sources.k8s_logs]
type = "kubernetes_logs"
auto_partial_merge = true
 
# トランスフォーム: JSON パース
[transforms.parse_json]
type = "remap"
inputs = ["app_logs"]
source = '''
. = parse_json!(string!(.message))
.timestamp = to_timestamp!(.timestamp)
.environment = "production"
.cluster = "prod-ap-northeast-1"
'''
 
# トランスフォーム: 機密データのリダクション
[transforms.redact_sensitive]
type = "remap"
inputs = ["parse_json"]
source = '''
if exists(.details.password) {
    .details.password = "***REDACTED***"
}
if exists(.details.token) {
    .details.token = "***REDACTED***"
}
# クレジットカード番号のマスク
if exists(.details.card_number) {
    card = string!(.details.card_number)
    .details.card_number = "****-****-****-" + slice!(card, -4)
}
'''
 
# トランスフォーム: セキュリティイベントのフィルタリング
[transforms.security_filter]
type = "filter"
inputs = ["redact_sensitive"]
condition = '''
.level == "ERROR" || .level == "WARN" ||
.event_type == "authentication" ||
.event_type == "access_control"
'''
 
# シンク: OpenSearch
[sinks.opensearch]
type = "elasticsearch"
inputs = ["redact_sensitive"]
endpoints = ["https://opensearch.internal.example.com:443"]
bulk.index = "app-logs-%Y-%m-%d"
auth.strategy = "basic"
auth.user = "vector"
auth.password = "${OPENSEARCH_PASSWORD}"
 
# シンク: セキュリティイベントを専用インデックスに
[sinks.security_opensearch]
type = "elasticsearch"
inputs = ["security_filter"]
endpoints = ["https://opensearch.internal.example.com:443"]
bulk.index = "security-events-%Y-%m-%d"
 
# シンク: S3 アーカイブ
[sinks.s3_archive]
type = "aws_s3"
inputs = ["redact_sensitive"]
bucket = "security-logs-archive"
region = "ap-northeast-1"
key_prefix = "logs/%Y/%m/%d/"
compression = "gzip"
encoding.codec = "json"

# sigma/rules/credential_access/brute_force_ssh.yml
title: SSH Brute Force Attack
id: a1234567-b890-1234-cdef-567890abcdef
status: stable
description: 同一 IP から短時間に多数の SSH ログイン失敗を検知
author: Security Team
date: 2025/01/15
modified: 2025/03/15
tags:
  - attack.credential_access
  - attack.t1110.001
  - cve.none
references:
  - https://attack.mitre.org/techniques/T1110/001/
logsource:
  category: authentication
  product: linux
detection:
  selection:
    eventid: 'sshd'
    action: 'Failed'
  filter:
    source_ip|cidr:
      - '10.0.0.0/8'
      - '172.16.0.0/12'
  timeframe: 5m
  condition: selection and not filter | count(source_ip) > 10
level: high
falsepositives:
  - 正当なパスワードリセット作業
  - 自動化スクリプトの設定ミス

# sigma/rules/persistence/new_iam_user.yml
title: AWS IAM User Created
id: b2345678-c901-2345-defg-678901bcdefg
status: stable
description: 新しい IAM ユーザの作成を検知（不正アクセスの持続化の兆候）
author: Security Team
date: 2025/02/01
tags:
  - attack.persistence
  - attack.t1136.003
logsource:
  product: aws
  service: cloudtrail
detection:
  selection:
    eventName: 'CreateUser'
    eventSource: 'iam.amazonaws.com'
  filter_automation:
    userIdentity.arn|contains:
      - 'terraform'
      - 'cloudformation'
  condition: selection and not filter_automation
level: medium
falsepositives:
  - IaC による正当なユーザ作成
  - オンボーディング作業

# sigma/rules/exfiltration/large_s3_download.yml
title: Large S3 Data Download
id: c3456789-d012-3456-efgh-789012cdefgh
status: experimental
description: S3 バケットからの大量データダウンロードを検知
author: Security Team
date: 2025/03/01
tags:
  - attack.exfiltration
  - attack.t1530
logsource:
  product: aws
  service: cloudtrail
detection:
  selection:
    eventName: 'GetObject'
    eventSource: 's3.amazonaws.com'
  timeframe: 1h
  condition: selection | count(requestParameters.bucketName) > 1000
level: high
falsepositives:
  - バックアップジョブ
  - データ移行作業
  - 正当な大量ダウンロード処理

# Sigma CLI で各 SIEM のクエリ形式に変換
pip install sigma-cli
 
# 利用可能なバックエンドの確認
sigma list backends
 
# Splunk SPL に変換
sigma convert -t splunk -p sysmon sigma/rules/
 
# Elastic EQL に変換
sigma convert -t elasticsearch sigma/rules/
 
# OpenSearch に変換
sigma convert -t opensearch sigma/rules/
 
# Microsoft Sentinel KQL に変換
sigma convert -t microsoft365defender sigma/rules/
 
# 変換例 (Splunk SPL):
# source=sshd action="Failed"
# NOT (source_ip="10.0.0.0/8" OR source_ip="172.16.0.0/12")
# | stats count by source_ip
# | where count > 10
 
# バッチ変換とテスト
sigma convert -t splunk sigma/rules/ -o splunk_rules/
sigma check sigma/rules/  # ルールの構文チェック

# コード例2: SIEM 相関ルールの実装（概念実装）
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from collections import defaultdict
import heapq
 
@dataclass
class LogEvent:
    """ログイベントの共通構造"""
    timestamp: datetime
    source: str
    event_type: str
    severity: str
    source_ip: str
    user_id: Optional[str] = None
    details: Dict = field(default_factory=dict)
 
@dataclass
class Alert:
    """相関分析結果のアラート"""
    rule_id: str
    title: str
    severity: str
    description: str
    events: List[LogEvent]
    created_at: datetime = field(
        default_factory=lambda: datetime.utcnow()
    )
    mitre_tactic: Optional[str] = None
    mitre_technique: Optional[str] = None
 
class CorrelationEngine:
    """SIEM 相関分析エンジン
 
    複数のログイベントを時系列で分析し、
    単体のイベントでは検知できない攻撃パターンを検出する。
 
    内部動作:
    1. イベントを時系列でバッファリング
    2. 各相関ルールとマッチングを実行
    3. 閾値を超えた場合にアラートを生成
    4. 重複排除と優先度計算を行う
    """
 
    def __init__(self, window_minutes: int = 60):
        self.window = timedelta(minutes=window_minutes)
        self.event_buffer: List[LogEvent] = []
        self.rules: List[callable] = []
        self.alerts: List[Alert] = []
        # ルールごとのイベントカウンタ
        self._counters: Dict[str, Dict] = defaultdict(
            lambda: defaultdict(int)
        )
 
    def add_event(self, event: LogEvent) -> List[Alert]:
        """イベントを追加し、相関ルールをチェック"""
        self.event_buffer.append(event)
        self._cleanup_old_events()
 
        new_alerts = []
        for rule in self.rules:
            alert = rule(event, self.event_buffer)
            if alert:
                # 重複アラートの排除
                if not self._is_duplicate(alert):
                    new_alerts.append(alert)
                    self.alerts.append(alert)
 
        return new_alerts
 
    def _cleanup_old_events(self):
        """ウィンドウ外の古いイベントを削除"""
        cutoff = datetime.utcnow() - self.window
        self.event_buffer = [
            e for e in self.event_buffer
            if e.timestamp > cutoff
        ]
 
    def _is_duplicate(self, alert: Alert) -> bool:
        """過去1時間以内の同一ルールIDのアラートを重複とみなす"""
        cutoff = datetime.utcnow() - timedelta(hours=1)
        return any(
            a.rule_id == alert.rule_id and a.created_at > cutoff
            for a in self.alerts
        )
 
    def register_rule(self, rule_func: callable):
        """相関ルールを登録"""
        self.rules.append(rule_func)
 
 
def brute_force_rule(event: LogEvent,
                     buffer: List[LogEvent]) -> Optional[Alert]:
    """ブルートフォース検知ルール
 
    5分以内に同一IPから10回以上のログイン失敗を検知
    """
    if event.event_type != "login_failed":
        return None
 
    window = timedelta(minutes=5)
    cutoff = event.timestamp - window
 
    failed_from_same_ip = [
        e for e in buffer
        if e.event_type == "login_failed"
        and e.source_ip == event.source_ip
        and e.timestamp > cutoff
    ]
 
    if len(failed_from_same_ip) >= 10:
        return Alert(
            rule_id="BRUTE_FORCE_001",
            title=f"ブルートフォース攻撃検知: {event.source_ip}",
            severity="HIGH",
            description=(
                f"IP {event.source_ip} から5分間に"
                f"{len(failed_from_same_ip)}回のログイン失敗"
            ),
            events=failed_from_same_ip,
            mitre_tactic="Credential Access",
            mitre_technique="T1110.001",
        )
    return None
 
 
def impossible_travel_rule(event: LogEvent,
                           buffer: List[LogEvent]) -> Optional[Alert]:
    """不可能な移動検知ルール
 
    短時間内に地理的に離れた場所からのログインを検知
    """
    if event.event_type != "login_success":
        return None
 
    window = timedelta(hours=1)
    cutoff = event.timestamp - window
 
    recent_logins = [
        e for e in buffer
        if e.event_type == "login_success"
        and e.user_id == event.user_id
        and e.timestamp > cutoff
        and e.source_ip != event.source_ip
    ]
 
    for prev_login in recent_logins:
        # 地理的距離を計算（実装略）
        distance = calculate_geo_distance(
            prev_login.details.get("geo"),
            event.details.get("geo"),
        )
        time_diff = (event.timestamp - prev_login.timestamp).total_seconds()
 
        # 時速1000km以上の移動は不可能と判定
        if distance > 0 and time_diff > 0:
            speed_kmh = (distance / time_diff) * 3600
            if speed_kmh > 1000:
                return Alert(
                    rule_id="IMPOSSIBLE_TRAVEL_001",
                    title=f"不可能な移動検知: {event.user_id}",
                    severity="HIGH",
                    description=(
                        f"ユーザ {event.user_id} が"
                        f"{time_diff/60:.0f}分間で"
                        f"{distance:.0f}km 移動"
                    ),
                    events=[prev_login, event],
                    mitre_tactic="Initial Access",
                    mitre_technique="T1078",
                )
    return None
 
 
# エンジンの使用例
engine = CorrelationEngine(window_minutes=60)
engine.register_rule(brute_force_rule)
engine.register_rule(impossible_travel_rule)

# コード例3: CloudWatch による異常検知の自動構築
import boto3
import json
 
logs = boto3.client('logs')
cloudwatch = boto3.client('cloudwatch')
sns = boto3.client('sns')
 
class SecurityAlarmBuilder:
    """CloudWatch セキュリティアラームの自動構築
 
    以下のベストプラクティスに従って構築する:
    - 重要なセキュリティイベントのみアラート化
    - 閾値は環境の特性に合わせて調整
    - 通知先は重大度に応じて分離
    """
 
    def __init__(self, log_group: str, sns_topic_arn: str):
        self.log_group = log_group
        self.sns_topic_arn = sns_topic_arn
 
    def create_root_account_alarm(self):
        """Root アカウント使用の検知"""
        logs.put_metric_filter(
            logGroupName=self.log_group,
            filterName='RootAccountUsage',
            filterPattern=(
                '{ $.userIdentity.type = "Root" '
                '&& $.userIdentity.invokedBy NOT EXISTS '
                '&& $.eventType != "AwsServiceEvent" }'
            ),
            metricTransformations=[{
                'metricName': 'RootAccountUsageCount',
                'metricNamespace': 'SecurityMetrics',
                'metricValue': '1',
            }],
        )
 
        cloudwatch.put_metric_alarm(
            AlarmName='RootAccountUsage',
            MetricName='RootAccountUsageCount',
            Namespace='SecurityMetrics',
            Statistic='Sum',
            Period=300,
            EvaluationPeriods=1,
            Threshold=1,
            ComparisonOperator='GreaterThanOrEqualToThreshold',
            AlarmActions=[self.sns_topic_arn],
            TreatMissingData='notBreaching',
        )
 
    def create_unauthorized_api_alarm(self):
        """未認可 API 呼び出しの検知"""
        logs.put_metric_filter(
            logGroupName=self.log_group,
            filterName='UnauthorizedAPICalls',
            filterPattern=(
                '{ ($.errorCode = "*UnauthorizedAccess*") '
                '|| ($.errorCode = "AccessDenied*") }'
            ),
            metricTransformations=[{
                'metricName': 'UnauthorizedAPICallCount',
                'metricNamespace': 'SecurityMetrics',
                'metricValue': '1',
            }],
        )
 
        cloudwatch.put_metric_alarm(
            AlarmName='UnauthorizedAPICalls',
            MetricName='UnauthorizedAPICallCount',
            Namespace='SecurityMetrics',
            Statistic='Sum',
            Period=300,
            EvaluationPeriods=1,
            Threshold=5,
            ComparisonOperator='GreaterThanOrEqualToThreshold',
            AlarmActions=[self.sns_topic_arn],
            TreatMissingData='notBreaching',
        )
 
    def create_console_login_failure_alarm(self):
        """コンソールログイン失敗の検知"""
        logs.put_metric_filter(
            logGroupName=self.log_group,
            filterName='ConsoleLoginFailures',
            filterPattern=(
                '{ ($.eventName = "ConsoleLogin") '
                '&& ($.errorMessage = "Failed authentication") }'
            ),
            metricTransformations=[{
                'metricName': 'ConsoleLoginFailureCount',
                'metricNamespace': 'SecurityMetrics',
                'metricValue': '1',
            }],
        )
 
        cloudwatch.put_metric_alarm(
            AlarmName='ConsoleLoginFailures',
            MetricName='ConsoleLoginFailureCount',
            Namespace='SecurityMetrics',
            Statistic='Sum',
            Period=300,
            EvaluationPeriods=1,
            Threshold=3,
            ComparisonOperator='GreaterThanOrEqualToThreshold',
            AlarmActions=[self.sns_topic_arn],
            TreatMissingData='notBreaching',
        )
 
    def create_security_group_change_alarm(self):
        """セキュリティグループの変更検知"""
        logs.put_metric_filter(
            logGroupName=self.log_group,
            filterName='SecurityGroupChanges',
            filterPattern=(
                '{ ($.eventName = "AuthorizeSecurityGroupIngress") '
                '|| ($.eventName = "AuthorizeSecurityGroupEgress") '
                '|| ($.eventName = "RevokeSecurityGroupIngress") '
                '|| ($.eventName = "RevokeSecurityGroupEgress") '
                '|| ($.eventName = "CreateSecurityGroup") '
                '|| ($.eventName = "DeleteSecurityGroup") }'
            ),
            metricTransformations=[{
                'metricName': 'SecurityGroupChangeCount',
                'metricNamespace': 'SecurityMetrics',
                'metricValue': '1',
            }],
        )
 
        cloudwatch.put_metric_alarm(
            AlarmName='SecurityGroupChanges',
            MetricName='SecurityGroupChangeCount',
            Namespace='SecurityMetrics',
            Statistic='Sum',
            Period=300,
            EvaluationPeriods=1,
            Threshold=1,
            ComparisonOperator='GreaterThanOrEqualToThreshold',
            AlarmActions=[self.sns_topic_arn],
            TreatMissingData='notBreaching',
        )
 
    def create_all_alarms(self):
        """全セキュリティアラームを一括作成"""
        self.create_root_account_alarm()
        self.create_unauthorized_api_alarm()
        self.create_console_login_failure_alarm()
        self.create_security_group_change_alarm()
        print("All security alarms created successfully")
 
 
# 使用例
builder = SecurityAlarmBuilder(
    log_group='/aws/cloudtrail',
    sns_topic_arn='arn:aws:sns:ap-northeast-1:123456:security-alerts',
)
builder.create_all_alarms()

# コード例4: 統計的異常検知の実装
import math
from collections import defaultdict
from datetime import datetime, timedelta
from typing import List, Tuple, Optional
from dataclasses import dataclass
 
@dataclass
class AnomalyResult:
    """異常検知結果"""
    metric_name: str
    current_value: float
    baseline_mean: float
    baseline_stddev: float
    z_score: float
    is_anomaly: bool
    severity: str  # low, medium, high, critical
    timestamp: datetime
 
class StatisticalAnomalyDetector:
    """統計的ベースライン異常検知器
 
    指数移動平均（EMA）と標準偏差を用いて
    メトリクスのベースラインを動的に計算し、
    異常値を検出する。
 
    動作原理:
    1. 過去のデータポイントからEMAを計算
    2. 標準偏差を計算
    3. 新しいデータポイントのZスコアを算出
    4. Zスコアが閾値を超えた場合に異常と判定
 
    Zスコア = (観測値 - 平均) / 標準偏差
    - |Z| > 2: 低リスク異常
    - |Z| > 3: 中リスク異常
    - |Z| > 4: 高リスク異常
    """
 
    def __init__(self, window_hours: int = 168,  # 1週間
                 min_data_points: int = 100):
        self.window_hours = window_hours
        self.min_data_points = min_data_points
        # メトリクス名 → (timestamp, value) のリスト
        self.data_points: dict = defaultdict(list)
 
    def add_data_point(self, metric_name: str, value: float,
                       timestamp: Optional[datetime] = None):
        """データポイントを追加"""
        ts = timestamp or datetime.utcnow()
        self.data_points[metric_name].append((ts, value))
        self._cleanup(metric_name)
 
    def check_anomaly(self, metric_name: str,
                      value: float) -> Optional[AnomalyResult]:
        """異常値チェック"""
        points = self.data_points.get(metric_name, [])
 
        if len(points) < self.min_data_points:
            return None  # データ不足
 
        values = [p[1] for p in points]
        mean = sum(values) / len(values)
        variance = sum((v - mean) ** 2 for v in values) / len(values)
        stddev = math.sqrt(variance) if variance > 0 else 0.001
 
        z_score = (value - mean) / stddev
 
        severity = self._classify_severity(abs(z_score))
        is_anomaly = abs(z_score) > 2.0
 
        return AnomalyResult(
            metric_name=metric_name,
            current_value=value,
            baseline_mean=mean,
            baseline_stddev=stddev,
            z_score=z_score,
            is_anomaly=is_anomaly,
            severity=severity,
            timestamp=datetime.utcnow(),
        )
 
    def _classify_severity(self, abs_z: float) -> str:
        if abs_z > 4.0:
            return "critical"
        elif abs_z > 3.0:
            return "high"
        elif abs_z > 2.0:
            return "medium"
        else:
            return "low"
 
    def _cleanup(self, metric_name: str):
        """ウィンドウ外の古いデータを削除"""
        cutoff = datetime.utcnow() - timedelta(hours=self.window_hours)
        self.data_points[metric_name] = [
            p for p in self.data_points[metric_name]
            if p[0] > cutoff
        ]
 
 
# 使用例
detector = StatisticalAnomalyDetector()
 
# ベースラインの構築（過去データのシミュレーション）
import random
for i in range(200):
    # 通常のAPI呼び出し数: 平均100, 標準偏差20
    normal_value = random.gauss(100, 20)
    detector.add_data_point(
        "api_calls_per_minute",
        max(0, normal_value),
        datetime.utcnow() - timedelta(hours=200-i),
    )
 
# 異常値のチェック
result = detector.check_anomaly("api_calls_per_minute", 250)
if result and result.is_anomaly:
    print(f"ANOMALY DETECTED: {result.metric_name}")
    print(f"  Current: {result.current_value}")
    print(f"  Baseline: {result.baseline_mean:.1f} +/- "
          f"{result.baseline_stddev:.1f}")
    print(f"  Z-score: {result.z_score:.2f}")
    print(f"  Severity: {result.severity}")

# コード例5: OpenSearch ダッシュボードの自動構築スクリプト
import json
import requests
from typing import Dict, List
 
class SecurityDashboardBuilder:
    """OpenSearch Dashboards のセキュリティダッシュボード自動構築"""
 
    def __init__(self, opensearch_url: str, auth: tuple):
        self.url = opensearch_url
        self.auth = auth
        self.headers = {"Content-Type": "application/json"}
 
    def create_index_pattern(self, pattern: str, time_field: str):
        """インデックスパターンの作成"""
        body = {
            "attributes": {
                "title": pattern,
                "timeFieldName": time_field,
            }
        }
        requests.post(
            f"{self.url}/api/saved_objects/index-pattern/{pattern}",
            json=body, auth=self.auth, headers=self.headers,
        )
 
    def create_failed_login_visualization(self) -> dict:
        """ログイン失敗の時系列グラフ"""
        return {
            "title": "Failed Login Attempts Over Time",
            "visState": json.dumps({
                "title": "Failed Login Attempts",
                "type": "line",
                "aggs": [
                    {
                        "id": "1",
                        "type": "count",
                        "schema": "metric",
                    },
                    {
                        "id": "2",
                        "type": "date_histogram",
                        "schema": "segment",
                        "params": {
                            "field": "@timestamp",
                            "interval": "5m",
                        },
                    },
                ],
                "params": {
                    "seriesParams": [{
                        "show": True,
                        "type": "line",
                        "mode": "normal",
                    }],
                },
            }),
            "kibanaSavedObjectMeta": {
                "searchSourceJSON": json.dumps({
                    "query": {
                        "bool": {
                            "must": [
                                {"match": {"event": "login_failed"}}
                            ]
                        }
                    },
                    "index": "security-events-*",
                }),
            },
        }
 
    def create_top_attackers_table(self) -> dict:
        """攻撃元 IP トップ10テーブル"""
        return {
            "title": "Top Attacking IPs",
            "visState": json.dumps({
                "title": "Top Attacking IPs",
                "type": "table",
                "aggs": [
                    {
                        "id": "1",
                        "type": "count",
                        "schema": "metric",
                        "params": {"customLabel": "Attack Count"},
                    },
                    {
                        "id": "2",
                        "type": "terms",
                        "schema": "bucket",
                        "params": {
                            "field": "sourceIp.keyword",
                            "size": 10,
                            "order": "desc",
                            "orderBy": "1",
                            "customLabel": "Source IP",
                        },
                    },
                ],
            }),
        }
 
    def create_geo_map(self) -> dict:
        """攻撃元の地理分布マップ"""
        return {
            "title": "Attack Source Geographic Distribution",
            "visState": json.dumps({
                "title": "Attack Sources Map",
                "type": "tile_map",
                "aggs": [
                    {
                        "id": "1",
                        "type": "count",
                        "schema": "metric",
                    },
                    {
                        "id": "2",
                        "type": "geohash_grid",
                        "schema": "segment",
                        "params": {
                            "field": "geoLocation",
                            "precision": 3,
                        },
                    },
                ],
            }),
        }
 
    def build_security_dashboard(self):
        """セキュリティダッシュボードの構築"""
        self.create_index_pattern("security-events-*", "@timestamp")
 
        visualizations = [
            self.create_failed_login_visualization(),
            self.create_top_attackers_table(),
            self.create_geo_map(),
        ]
 
        for viz in visualizations:
            requests.post(
                f"{self.url}/api/saved_objects/visualization",
                json={"attributes": viz},
                auth=self.auth,
                headers=self.headers,
            )
 
        print(f"Dashboard created with {len(visualizations)} visualizations")

# コード例6: S3 ライフサイクルポリシーの設定
import boto3
 
s3 = boto3.client('s3')
 
def configure_log_lifecycle(bucket_name: str):
    """セキュリティログ用のライフサイクルポリシーを設定"""
 
    lifecycle_config = {
        'Rules': [
            {
                'ID': 'SecurityLogLifecycle',
                'Status': 'Enabled',
                'Filter': {
                    'Prefix': 'logs/',
                },
                'Transitions': [
                    {
                        'Days': 30,
                        'StorageClass': 'STANDARD_IA',
                    },
                    {
                        'Days': 90,
                        'StorageClass': 'GLACIER',
                    },
                    {
                        'Days': 365,
                        'StorageClass': 'DEEP_ARCHIVE',
                    },
                ],
                # PCI DSS/SOC 2 要件: 最低1年保持
                # 法的保持要件により7年保持する場合もある
                'Expiration': {
                    'Days': 2555,  # 7年
                },
                'NoncurrentVersionExpiration': {
                    'NoncurrentDays': 30,
                },
            },
        ],
    }
 
    s3.put_bucket_lifecycle_configuration(
        Bucket=bucket_name,
        LifecycleConfiguration=lifecycle_config,
    )
 
    # バケット暗号化の設定
    s3.put_bucket_encryption(
        Bucket=bucket_name,
        ServerSideEncryptionConfiguration={
            'Rules': [{
                'ApplyServerSideEncryptionByDefault': {
                    'SSEAlgorithm': 'aws:kms',
                    'KMSMasterKeyID': 'alias/security-logs-key',
                },
                'BucketKeyEnabled': True,
            }]
        },
    )
 
    # オブジェクトロック（改竄防止）
    s3.put_object_lock_configuration(
        Bucket=bucket_name,
        ObjectLockConfiguration={
            'ObjectLockEnabled': 'Enabled',
            'Rule': {
                'DefaultRetention': {
                    'Mode': 'COMPLIANCE',
                    'Days': 365,
                }
            }
        },
    )
 
    print(f"Lifecycle policy configured for {bucket_name}")
 
configure_log_lifecycle("security-logs-archive")

# コード例7: ログ欠損検知の実装
from datetime import datetime, timedelta
from typing import Dict, Set
 
class LogGapDetector:
    """ログ欠損を検知するモニター
 
    各ログソースからのログの到着間隔を監視し、
    期待される間隔を超えた場合にアラートを発行する。
    """
 
    def __init__(self):
        # ソース名 → (最終受信時刻, 期待間隔)
        self.sources: Dict[str, tuple] = {}
        # 期待されるソースの一覧
        self.expected_sources: Set[str] = set()
 
    def register_source(self, source_name: str,
                        expected_interval: timedelta):
        """ログソースを登録"""
        self.expected_sources.add(source_name)
        self.sources[source_name] = (None, expected_interval)
 
    def record_log(self, source_name: str):
        """ログの到着を記録"""
        if source_name in self.sources:
            _, interval = self.sources[source_name]
            self.sources[source_name] = (datetime.utcnow(), interval)
 
    def check_gaps(self) -> list:
        """ログの欠損を検出"""
        alerts = []
        now = datetime.utcnow()
 
        for source_name in self.expected_sources:
            if source_name not in self.sources:
                alerts.append({
                    'source': source_name,
                    'type': 'source_not_registered',
                    'message': f"ログソース {source_name} が未登録",
                })
                continue
 
            last_seen, expected_interval = self.sources[source_name]
 
            if last_seen is None:
                alerts.append({
                    'source': source_name,
                    'type': 'no_logs_received',
                    'message': f"{source_name} からログが一度も到着していない",
                })
            elif (now - last_seen) > expected_interval * 2:
                gap_minutes = (now - last_seen).total_seconds() / 60
                alerts.append({
                    'source': source_name,
                    'type': 'log_gap',
                    'message': (
                        f"{source_name} から{gap_minutes:.0f}分間"
                        f"ログが到着していない"
                    ),
                    'last_seen': last_seen.isoformat(),
                    'gap_minutes': gap_minutes,
                })
 
        return alerts