脅威軽減マッピング

効果的なセキュリティ計画のために脅威をコントロールに結びつけます。

このスキルを使用するタイミング

• セキュリティ投資の優先順位付け
• 修復ロードマップの作成
• コントロール対象範囲の検証
• 多層防御設計
• セキュリティアーキテクチャレビュー
• リスク処理計画

コアコンセプト

1. コントロールカテゴリ

予防的 ────► 攻撃が発生する前に停止
   │         (ファイアウォール、入力検証)
   │
検知的 ─────► 進行中の攻撃を特定
   │         (IDS、ログ監視)
   │
是正的 ────► 攻撃から回復・対応
            (インシデント対応、バックアップ復元)

2. コントロール層

層	例
ネットワーク	ファイアウォール、WAF、DDoS保護
アプリケーション	入力検証、認証
データ	暗号化、アクセス制御
エンドポイント	EDR、パッチ管理
プロセス	セキュリティトレーニング、インシデント対応

3. 多層防御

                    ┌──────────────────────┐
                    │      境界部分         │ ← ファイアウォール、WAF
                    │   ┌──────────────┐   │
                    │   │   ネットワーク │   │ ← セグメンテーション、IDS
                    │   │  ┌────────┐  │   │
                    │   │  │  ホスト  │  │   │ ← EDR、硬化
                    │   │  │ ┌────┐ │  │   │
                    │   │  │ │アプリ │  │   │ ← 認証、検証
                    │   │  │ │データ │  │   │ ← 暗号化
                    │   │  │ └────┘ │  │   │
                    │   │  └────────┘  │   │
                    │   └──────────────┘   │
                    └──────────────────────┘

テンプレート

テンプレート 1: 軽減モデル

from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Optional, Set
from datetime import datetime

class ControlType(Enum):
    PREVENTIVE = "preventive"
    DETECTIVE = "detective"
    CORRECTIVE = "corrective"


class ControlLayer(Enum):
    NETWORK = "network"
    APPLICATION = "application"
    DATA = "data"
    ENDPOINT = "endpoint"
    PROCESS = "process"
    PHYSICAL = "physical"


class ImplementationStatus(Enum):
    NOT_IMPLEMENTED = "not_implemented"
    PARTIAL = "partial"
    IMPLEMENTED = "implemented"
    VERIFIED = "verified"


class Effectiveness(Enum):
    NONE = 0
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    VERY_HIGH = 4


@dataclass
class SecurityControl:
    id: str
    name: str
    description: str
    control_type: ControlType
    layer: ControlLayer
    effectiveness: Effectiveness
    implementation_cost: str  # Low, Medium, High
    maintenance_cost: str
    status: ImplementationStatus = ImplementationStatus.NOT_IMPLEMENTED
    mitigates_threats: List[str] = field(default_factory=list)
    dependencies: List[str] = field(default_factory=list)
    technologies: List[str] = field(default_factory=list)
    compliance_refs: List[str] = field(default_factory=list)

    def coverage_score(self) -> float:
        """ステータスと有効性に基づいてカバレッジスコアを計算します。"""
        status_multiplier = {
            ImplementationStatus.NOT_IMPLEMENTED: 0.0,
            ImplementationStatus.PARTIAL: 0.5,
            ImplementationStatus.IMPLEMENTED: 0.8,
            ImplementationStatus.VERIFIED: 1.0,
        }
        return self.effectiveness.value * status_multiplier[self.status]


@dataclass
class Threat:
    id: str
    name: str
    category: str  # STRIDE category
    description: str
    impact: str  # Critical, High, Medium, Low
    likelihood: str
    risk_score: float


@dataclass
class MitigationMapping:
    threat: Threat
    controls: List[SecurityControl]
    residual_risk: str = "Unknown"
    notes: str = ""

    def calculate_coverage(self) -> float:
        """コントロールがどの程度脅威をカバーしているかを計算します。"""
        if not self.controls:
            return 0.0

        total_score = sum(c.coverage_score() for c in self.controls)
        max_possible = len(self.controls) * Effectiveness.VERY_HIGH.value

        return (total_score / max_possible) * 100 if max_possible > 0 else 0

    def has_defense_in_depth(self) -> bool:
        """複数層がカバーされているかチェックします。"""
        layers = set(c.layer for c in self.controls if c.status != ImplementationStatus.NOT_IMPLEMENTED)
        return len(layers) >= 2

    def has_control_diversity(self) -> bool:
        """複数のコントロール種類が存在するかチェックします。"""
        types = set(c.control_type for c in self.controls if c.status != ImplementationStatus.NOT_IMPLEMENTED)
        return len(types) >= 2


@dataclass
class MitigationPlan:
    name: str
    threats: List[Threat] = field(default_factory=list)
    controls: List[SecurityControl] = field(default_factory=list)
    mappings: List[MitigationMapping] = field(default_factory=list)

    def get_unmapped_threats(self) -> List[Threat]:
        """軽減策なしの脅威を見つけます。"""
        mapped_ids = {m.threat.id for m in self.mappings}
        return [t for t in self.threats if t.id not in mapped_ids]

    def get_control_coverage(self) -> Dict[str, float]:
        """各脅威のカバレッジパーセンテージを取得します。"""
        return {
            m.threat.id: m.calculate_coverage()
            for m in self.mappings
        }

    def get_gaps(self) -> List[Dict]:
        """軽減ギャップを特定します。"""
        gaps = []
        for mapping in self.mappings:
            coverage = mapping.calculate_coverage()
            if coverage < 50:
                gaps.append({
                    "threat": mapping.threat.id,
                    "threat_name": mapping.threat.name,
                    "coverage": coverage,
                    "issue": "コントロール対象範囲が不十分",
                    "recommendation": "コントロールの追加または既存のものの改善"
                })
            if not mapping.has_defense_in_depth():
                gaps.append({
                    "threat": mapping.threat.id,
                    "threat_name": mapping.threat.name,
                    "coverage": coverage,
                    "issue": "多層防御がない",
                    "recommendation": "異なる層のコントロールを追加"
                })
            if not mapping.has_control_diversity():
                gaps.append({
                    "threat": mapping.threat.id,
                    "threat_name": mapping.threat.name,
                    "coverage": coverage,
                    "issue": "コントロール多様性がない",
                    "recommendation": "検知的・是正的コントロールを追加"
                })
        return gaps

テンプレート 2: コントロールライブラリ

class ControlLibrary:
    """標準的なセキュリティコントロールのライブラリ。"""

    STANDARD_CONTROLS = {
        # 認証コントロール
        "AUTH-001": SecurityControl(
            id="AUTH-001",
            name="多要素認証",
            description="すべてのユーザー認証に MFA を要求",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.APPLICATION,
            effectiveness=Effectiveness.HIGH,
            implementation_cost="Medium",
            maintenance_cost="Low",
            mitigates_threats=["SPOOFING"],
            technologies=["TOTP", "WebAuthn", "SMS OTP"],
            compliance_refs=["PCI-DSS 8.3", "NIST 800-63B"]
        ),
        "AUTH-002": SecurityControl(
            id="AUTH-002",
            name="アカウントロックアウトポリシー",
            description="認証失敗後にアカウントをロック",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.APPLICATION,
            effectiveness=Effectiveness.MEDIUM,
            implementation_cost="Low",
            maintenance_cost="Low",
            mitigates_threats=["SPOOFING"],
            technologies=["Custom implementation"],
            compliance_refs=["PCI-DSS 8.1.6"]
        ),

        # 入力検証コントロール
        "VAL-001": SecurityControl(
            id="VAL-001",
            name="入力検証フレームワーク",
            description="すべてのユーザー入力を検証・サニタイズ",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.APPLICATION,
            effectiveness=Effectiveness.HIGH,
            implementation_cost="Medium",
            maintenance_cost="Medium",
            mitigates_threats=["TAMPERING", "INJECTION"],
            technologies=["Joi", "Yup", "Pydantic"],
            compliance_refs=["OWASP ASVS V5"]
        ),
        "VAL-002": SecurityControl(
            id="VAL-002",
            name="Web Application Firewall",
            description="悪意のあるリクエストをフィルタリングする WAF を配置",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.NETWORK,
            effectiveness=Effectiveness.MEDIUM,
            implementation_cost="Medium",
            maintenance_cost="Medium",
            mitigates_threats=["TAMPERING", "INJECTION", "DOS"],
            technologies=["AWS WAF", "Cloudflare", "ModSecurity"],
            compliance_refs=["PCI-DSS 6.6"]
        ),

        # 暗号化コントロール
        "ENC-001": SecurityControl(
            id="ENC-001",
            name="保存時データ暗号化",
            description="ストレージ内の機密データを暗号化",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.DATA,
            effectiveness=Effectiveness.HIGH,
            implementation_cost="Medium",
            maintenance_cost="Low",
            mitigates_threats=["INFORMATION_DISCLOSURE"],
            technologies=["AES-256", "KMS", "HSM"],
            compliance_refs=["PCI-DSS 3.4", "GDPR Art. 32"]
        ),
        "ENC-002": SecurityControl(
            id="ENC-002",
            name="TLS 暗号化",
            description="TLS 1.3 を使用して転送中のデータを暗号化",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.NETWORK,
            effectiveness=Effectiveness.HIGH,
            implementation_cost="Low",
            maintenance_cost="Low",
            mitigates_threats=["INFORMATION_DISCLOSURE", "TAMPERING"],
            technologies=["TLS 1.3", "Certificate management"],
            compliance_refs=["PCI-DSS 4.1", "HIPAA"]
        ),

        # ロギングコントロール
        "LOG-001": SecurityControl(
            id="LOG-001",
            name="セキュリティイベントログ",
            description="すべてのセキュリティ関連イベントをログに記録",
            control_type=ControlType.DETECTIVE,
            layer=ControlLayer.APPLICATION,
            effectiveness=Effectiveness.MEDIUM,
            implementation_cost="Low",
            maintenance_cost="Medium",
            mitigates_threats=["REPUDIATION"],
            technologies=["ELK Stack", "Splunk", "CloudWatch"],
            compliance_refs=["PCI-DSS 10.2", "SOC2"]
        ),
        "LOG-002": SecurityControl(
            id="LOG-002",
            name="ログ整合性保護",
            description="ログを改ざんから保護",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.DATA,
            effectiveness=Effectiveness.MEDIUM,
            implementation_cost="Medium",
            maintenance_cost="Low",
            mitigates_threats=["REPUDIATION", "TAMPERING"],
            technologies=["Immutable storage", "Log signing"],
            compliance_refs=["PCI-DSS 10.5"]
        ),

        # アクセスコントロール
        "ACC-001": SecurityControl(
            id="ACC-001",
            name="ロールベースアクセス制御",
            description="認可のために RBAC を実装",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.APPLICATION,
            effectiveness=Effectiveness.HIGH,
            implementation_cost="Medium",
            maintenance_cost="Medium",
            mitigates_threats=["ELEVATION_OF_PRIVILEGE", "INFORMATION_DISCLOSURE"],
            technologies=["RBAC", "ABAC", "Policy engines"],
            compliance_refs=["PCI-DSS 7.1", "SOC2"]
        ),

        # 可用性コントロール
        "AVL-001": SecurityControl(
            id="AVL-001",
            name="レート制限",
            description="リクエストレートを制限して悪用を防止",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.APPLICATION,
            effectiveness=Effectiveness.MEDIUM,
            implementation_cost="Low",
            maintenance_cost="Low",
            mitigates_threats=["DENIAL_OF_SERVICE"],
            technologies=["API Gateway", "Redis", "Token bucket"],
            compliance_refs=["OWASP API Security"]
        ),
        "AVL-002": SecurityControl(
            id="AVL-002",
            name="DDoS 対策",
            description="DDoS 軽減サービスを配置",
            control_type=ControlType.PREVENTIVE,
            layer=ControlLayer.NETWORK,
            effectiveness=Effectiveness.HIGH,
            implementation_cost="High",
            maintenance_cost="Medium",
            mitigates_threats=["DENIAL_OF_SERVICE"],
            technologies=["Cloudflare", "AWS Shield", "Akamai"],
            compliance_refs=["NIST CSF"]
        ),
    }

    def get_controls_for_threat(self, threat_category: str) -> List[SecurityControl]:
        """脅威カテゴリを軽減するすべてのコントロールを取得します。"""
        return [
            c for c in self.STANDARD_CONTROLS.values()
            if threat_category in c.mitigates_threats
        ]

    def get_controls_by_layer(self, layer: ControlLayer) -> List[SecurityControl]:
        """特定の層のコントロールを取得します。"""
        return [c for c in self.STANDARD_CONTROLS.values() if c.layer == layer]

    def get_control(self, control_id: str) -> Optional[SecurityControl]:
        """ID で特定のコントロールを取得します。"""
        return self.STANDARD_CONTROLS.get(control_id)

    def recommend_controls(
        self,
        threat: Threat,
        existing_controls: List[str]
    ) -> List[SecurityControl]:
        """脅威に対する追加コントロールを推奨します。"""
        available = self.get_controls_for_threat(threat.category)
        return [c for c in available if c.id not in existing_controls]

テンプレート 3: 軽減分析

class MitigationAnalyzer:
    """軽減戦略を分析・最適化します。"""

    def __init__(self, plan: MitigationPlan, library: ControlLibrary):
        self.plan = plan
        self.library = library

    def calculate_overall_risk_reduction(self) -> float:
        """全体的なリスク低減パーセンテージを計算します。"""
        if not self.plan.mappings:
            return 0.0

        weighted_coverage = 0
        total_weight = 0

        for mapping in self.plan.mappings:
            # 脅威リスクスコアで重み付け
            weight = mapping.threat.risk_score
            coverage = mapping.calculate_coverage()
            weighted_coverage += weight * coverage
            total_weight += weight

        return weighted_coverage / total_weight if total_weight > 0 else 0

    def get_critical_gaps(self) -> List[Dict]:
        """すぐに対応が必要な重大なギャップを見つけます。"""
        gaps = self.plan.get_gaps()
        critical_threats = {t.id for t in self.plan.threats if t.impact == "Critical"}

        return [g for g in gaps if g["threat"] in critical_threats]

    def optimize_budget(
        self,
        budget: float,
        cost_map: Dict[str, float]
    ) -> List[SecurityControl]:
        """予算内でリスク低減を最大化するコントロールを選択します。"""
        # シンプルなグリーディアプローチ - 最適化アルゴリズムで置き換え可能
        recommended = []
        remaining_budget = budget
        unmapped = self.plan.get_unmapped_threats()

        # 効果性/コスト比でコントロールをソート
        all_controls = list(self.library.STANDARD_CONTROLS.values())
        controls_with_value = []

        for control in all_controls:
            if control.status == ImplementationStatus.NOT_IMPLEMENTED:
                cost = cost_map.get(control.id, float('inf'))
                if cost <= remaining_budget:
                    # 値を脅威カバー数 * 有効性 / コストとして計算
                    threats_covered = len([
                        t for t in unmapped
                        if t.category in control.mitigates_threats
                    ])
                    if threats_covered > 0:
                        value = (threats_covered * control.effectiveness.value) / cost
                        controls_with_value.append((control, value, cost))

        # 値でソート（高いほうが良い）
        controls_with_value.sort(key=lambda x: x[1], reverse=True)

        for control, value, cost in controls_with_value:
            if cost <= remaining_budget:
                recommended.append(control)
                remaining_budget -= cost

        return recommended

    def generate_roadmap(self) -> List[Dict]:
        """優先度順の実装ロードマップを生成します。"""
        roadmap = []
        gaps = self.plan.get_gaps()

        # フェーズ 1: カバレッジが低い重大な脅威
        phase1 = []
        for gap in gaps:
            mapping = next(
                (m for m in self.plan.mappings if m.threat.id == gap["threat"]),
                None
            )
            if mapping and mapping.threat.impact == "Critical":
                controls = self.library.get_controls_for_threat(mapping.threat.category)
                phase1.extend([
                    {
                        "threat": gap["threat"],
                        "control": c.id,
                        "control_name": c.name,
                        "phase": 1,
                        "priority": "Critical"
                    }
                    for c in controls
                    if c.status == ImplementationStatus.NOT_IMPLEMENTED
                ])

        roadmap.extend(phase1[:5])  # フェーズ 1 向け上位 5 つ

        # フェーズ 2: 高影響の脅威
        phase2 = []
        for gap in gaps:
            mapping = next(
                (m for m in self.plan.mappings if m.threat.id == gap["threat"]),
                None
            )
            if mapping and mapping.threat.impact == "High":
                controls = self.library.get_controls_for_threat(mapping.threat.category)
                phase2.extend([
                    {
                        "threat": gap["threat"],
                        "control": c.id,
                        "control_name": c.name,
                        "phase": 2,
                        "priority": "High"
                    }
                    for c in controls
                    if c.status == ImplementationStatus.NOT_IMPLEMENTED
                ])

        roadmap.extend(phase2[:5])  # フェーズ 2 向け上位 5 つ

        return roadmap

    def defense_in_depth_analysis(self) -> Dict[str, List[str]]:
        """多層防御対象範囲を分析します。"""
        layer_coverage = {layer.value: [] for layer in ControlLayer}

        for mapping in self.plan.mappings:
            for control in mapping.controls:
                if control.status in [ImplementationStatus.IMPLEMENTED, ImplementationStatus.VERIFIED]:
                    layer_coverage[control.layer.value].append(control.id)

        return layer_coverage

    def generate_report(self) -> str:
        """包括的な軽減レポートを生成します。"""
        risk_reduction = self.calculate_overall_risk_reduction()
        gaps = self.plan.get_gaps()
        critical_gaps = self.get_critical_gaps()
        layer_coverage = self.defense_in_depth_analysis()

        report = f"""
# 脅威軽減レポート

## エグゼクティブサマリー
- **全体的なリスク低減:** {risk_reduction:.1f}%
- **総脅威数:** {len(self.plan.threats)}
- **総コントロール数:** {len(self.plan.controls)}
- **特定されたギャップ:** {len(gaps)}
- **重大なギャップ:** {len(critical_gaps)}

## 多層防御対象範囲
{self._format_layer_coverage(layer_coverage)}

## 即座の対応が必要な重大なギャップ
{self._format_gaps(critical_gaps)}

## 推奨事項
{self._format_recommendations()}

## 実装ロードマップ
{self._format_roadmap()}
"""
        return report

    def _format_layer_coverage(self, coverage: Dict[str, List[str]]) -> str:
        lines = []
        for layer, controls in coverage.items():
            status = "✓" if controls else "✗"
            lines.append(f"- {layer}: {status} ({len(controls)} controls)")
        return "\n".join(lines)

    def _format_gaps(self, gaps: List[Dict]) -> str:
        if not gaps:
            return "重大なギャップは特定されていません。"
        lines = []
        for gap in gaps:
            lines.append(f"- **{gap['threat_name']}**: {gap['issue']}")
            lines.append(f"  - カバレッジ: {gap['coverage']:.1f}%")
            lines.append(f"  - 推奨: {gap['recommendation']}")
        return "\n".join(lines)

    def _format_recommendations(self) -> str:
        recommendations = []
        layer_coverage = self.defense_in_depth_analysis()

        for layer, controls in layer_coverage.items():
            if not controls:
                recommendations.append(f"- {layer} 層コントロールを追加")

        gaps = self.plan.get_gaps()
        if any(g["issue"] == "No control diversity" for g in gaps):
            recommendations.append("- より多くの検知的および是正的コントロールを追加")

        return "\n".join(recommendations) if recommendations else "現在の対象範囲は適切です。"

    def _format_roadmap(self) -> str:
        roadmap = self.generate_roadmap()
        if not roadmap:
            return "現時点では追加のコントロール推奨は不要です。"

        lines = []
        current_phase = 0
        for item in roadmap:
            if item["phase"] != current_phase:
                current_phase = item["phase"]
                lines.append(f"\n### フェーズ {current_phase}")
            lines.append(f"- [{item['priority']}] {item['control_name']} ({item['threat']} 向け)")

        return "\n".join(lines)

テンプレート 4: コントロール有効性テスト

from dataclasses import dataclass
from typing import List, Callable, Any
import asyncio

@dataclass
class ControlTest:
    control_id: str
    test_name: str
    test_function: Callable[[], bool]
    expected_result: bool
    description: str


class ControlTester:
    """コントロール有効性をテストします。"""

    def __init__(self):
        self.tests: List[ControlTest] = []
        self.results: List[Dict] = []

    def add_test(self, test: ControlTest) -> None:
        self.tests.append(test)

    async def run_tests(self) -> List[Dict]:
        """すべてのコントロールテストを実行します。"""
        self.results = []

        for test in self.tests:
            try:
                result = test.test_function()
                passed = result == test.expected_result
                self.results.append({
                    "control_id": test.control_id,
                    "test_name": test.test_name,
                    "passed": passed,
                    "actual_result": result,
                    "expected_result": test.expected_result,
                    "description": test.description,
                    "error": None
                })
            except Exception as e:
                self.results.append({
                    "control_id": test.control_id,
                    "test_name": test.test_name,
                    "passed": False,
                    "actual_result": None,
                    "expected_result": test.expected_result,
                    "description": test.description,
                    "error": str(e)
                })

        return self.results

    def get_effectiveness_score(self, control_id: str) -> float:
        """コントロールの有効性スコアを計算します。"""
        control_results = [r for r in self.results if r["control_id"] == control_id]
        if not control_results:
            return 0.0

        passed = sum(1 for r in control_results if r["passed"])
        return (passed / len(control_results)) * 100

    def generate_test_report(self) -> str:
        """テスト結果レポートを生成します。"""
        if not self.results:
            return "テストが実行されていません。"

        total = len(self.results)
        passed = sum(1 for r in self.results if r["passed"])

        report = f"""
# コントロール有効性テストレポート

## サマリー
- **総テスト数:** {total}
- **合格:** {passed}
- **不合格:** {total - passed}
- **合格率:** {(passed/total)*100:.1f}%

## コントロール別結果
"""
        # コントロール別にグループ化
        controls = {}
        for result in self.results:
            cid = result["control_id"]
            if cid not in controls:
                controls[cid] = []
            controls[cid].append(result)

        for control_id, results in controls.items():
            score = self.get_effectiveness_score(control_id)
            report += f"\n### {control_id} (有効性: {score:.1f}%)\n"
            for r in results:
                status = "✓" if r["passed"] else "✗"
                report += f"- {status} {r['test_name']}\n"
                if r["error"]:
                    report += f"  - エラー: {r['error']}\n"

        return report

ベストプラクティス

すべきこと

• すべての脅威をマッピング - 脅威が未マッピングになってはいけません
• コントロール層を重ねる - 多層防御は必須です
• コントロール種類を組み合わせ - 予防的、検知的、是正的
• 有効性を追跡 - 測定して改善します
• 定期的にレビュー - コントロールは時間とともに低下します

してはいけないこと

• 単一のコントロールに依存しない - 単一障害点です
• コストを無視しない - ROI が大切です
• テストをスキップしない - テストされていないコントロールは失敗する可能性があります
• 設定後に放置しない - 継続的な改善が必要です
• 人・プロセスを無視しない - テクノロジーだけでは不十分です