# コード例1: STRIDE-per-Element 分析ツール
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional, Dict, Set
 
class StrideCategory(Enum):
    SPOOFING = "S: なりすまし"
    TAMPERING = "T: 改ざん"
    REPUDIATION = "R: 否認"
    INFO_DISCLOSURE = "I: 情報漏洩"
    DENIAL_OF_SERVICE = "D: サービス妨害"
    ELEVATION_OF_PRIVILEGE = "E: 権限昇格"
 
class DFDElementType(Enum):
    EXTERNAL_ENTITY = "外部エンティティ"
    PROCESS = "プロセス"
    DATA_FLOW = "データフロー"
    DATA_STORE = "データストア"
 
# DFD要素ごとに該当するSTRIDEカテゴリを定義
STRIDE_PER_ELEMENT: Dict[DFDElementType, Set[StrideCategory]] = {
    DFDElementType.EXTERNAL_ENTITY: {
        StrideCategory.SPOOFING,
    },
    DFDElementType.PROCESS: {
        StrideCategory.SPOOFING,
        StrideCategory.TAMPERING,
        StrideCategory.REPUDIATION,
        StrideCategory.INFO_DISCLOSURE,
        StrideCategory.DENIAL_OF_SERVICE,
        StrideCategory.ELEVATION_OF_PRIVILEGE,
    },
    DFDElementType.DATA_FLOW: {
        StrideCategory.TAMPERING,
        StrideCategory.INFO_DISCLOSURE,
        StrideCategory.DENIAL_OF_SERVICE,
    },
    DFDElementType.DATA_STORE: {
        StrideCategory.TAMPERING,
        StrideCategory.REPUDIATION,
        StrideCategory.INFO_DISCLOSURE,
        StrideCategory.DENIAL_OF_SERVICE,
    },
}
 
@dataclass
class StrideThreat:
    """特定された脅威"""
    id: str
    category: StrideCategory
    element_name: str
    element_type: DFDElementType
    description: str
    attack_vector: str
    mitigations: List[str] = field(default_factory=list)
    priority: str = "未評価"  # Critical/High/Medium/Low/未評価
 
@dataclass
class StrideAnalyzer:
    """STRIDE-per-Element分析エンジン"""
 
    system_name: str
    threats: List[StrideThreat] = field(default_factory=list)
    _threat_counter: int = 0
 
    def analyze_element(self, name: str, element_type: DFDElementType,
                        description: str = "") -> List[str]:
        """DFD要素に対してSTRIDE分析を実行し、検討すべき質問を返す"""
        applicable = STRIDE_PER_ELEMENT.get(element_type, set())
        questions = []
 
        for category in applicable:
            question = self._generate_question(category, name, element_type)
            questions.append(question)
 
        return questions
 
    def _generate_question(self, category: StrideCategory,
                           element_name: str,
                           element_type: DFDElementType) -> str:
        """脅威カテゴリに基づいた分析質問を生成"""
        templates = {
            StrideCategory.SPOOFING:
                f"[{element_name}] に対して、なりすましは可能か？"
                f"正当な{element_type.value}であることをどう検証するか？",
            StrideCategory.TAMPERING:
                f"[{element_name}] のデータを攻撃者が改ざんできるか？"
                f"改ざんをどう検出するか？",
            StrideCategory.REPUDIATION:
                f"[{element_name}] の操作を後から否認できるか？"
                f"証跡は十分に記録されているか？",
            StrideCategory.INFO_DISCLOSURE:
                f"[{element_name}] から機密情報が漏洩する経路はあるか？"
                f"データは適切に暗号化されているか？",
            StrideCategory.DENIAL_OF_SERVICE:
                f"[{element_name}] のサービスを妨害する方法はあるか？"
                f"リソース制限は設定されているか？",
            StrideCategory.ELEVATION_OF_PRIVILEGE:
                f"[{element_name}] で権限昇格は可能か？"
                f"最小権限の原則は適用されているか？",
        }
        return templates.get(category, "")
 
    def add_threat(self, category: StrideCategory, element_name: str,
                   element_type: DFDElementType, description: str,
                   attack_vector: str,
                   mitigations: Optional[List[str]] = None) -> str:
        """脅威を登録"""
        self._threat_counter += 1
        threat_id = f"T-{self._threat_counter:03d}"
        threat = StrideThreat(
            id=threat_id,
            category=category,
            element_name=element_name,
            element_type=element_type,
            description=description,
            attack_vector=attack_vector,
            mitigations=mitigations or [],
        )
        self.threats.append(threat)
        return threat_id
 
    def generate_report(self) -> str:
        """分析レポートを生成"""
        lines = [
            f"# STRIDE分析レポート: {self.system_name}",
            f"## 脅威総数: {len(self.threats)}件\n",
        ]
        for cat in StrideCategory:
            cat_threats = [t for t in self.threats if t.category == cat]
            lines.append(f"### {cat.value} ({len(cat_threats)}件)")
            if not cat_threats:
                lines.append("  脅威なし\n")
                continue
            for t in cat_threats:
                lines.append(f"- **[{t.id}]** [{t.element_name}] {t.description}")
                lines.append(f"  - 攻撃ベクトル: {t.attack_vector}")
                lines.append(f"  - 優先度: {t.priority}")
                for m in t.mitigations:
                    lines.append(f"  - 緩和策: {m}")
            lines.append("")
        return "\n".join(lines)
 
# 使用例
analyzer = StrideAnalyzer("ECサイト決済システム")
 
# DFD要素に対するSTRIDE分析質問の生成
questions = analyzer.analyze_element(
    "決済API", DFDElementType.PROCESS, "クレジットカード決済処理"
)
for q in questions:
    print(f"  検討: {q}")
 
# 脅威の登録
analyzer.add_threat(
    StrideCategory.SPOOFING, "ログインAPI", DFDElementType.PROCESS,
    "ブルートフォースによるアカウント乗っ取り",
    "POST /api/login に大量の認証試行",
    ["レートリミット(5回/分)", "アカウントロックアウト(10分)", "MFA導入"],
)
analyzer.add_threat(
    StrideCategory.TAMPERING, "注文データフロー", DFDElementType.DATA_FLOW,
    "価格パラメータの改ざんによる不正な低額購入",
    "MITM/プロキシによるリクエストボディの改ざん",
    ["サーバーサイドでの価格再計算", "リクエスト署名検証", "TLS必須化"],
)
analyzer.add_threat(
    StrideCategory.INFO_DISCLOSURE, "ユーザーDB", DFDElementType.DATA_STORE,
    "SQLインジェクションによる全顧客データ漏洩",
    "検索フォームに ' OR 1=1 -- を入力",
    ["パラメータ化クエリ", "WAF導入", "DB最小権限"],
)
 
print(analyzer.generate_report())

# コード例2: DREAD + CVSSハイブリッドスコアリング
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class DreadScore:
    """DREADリスクスコアの計算"""
    damage: int           # 1-10: 被害の大きさ
    reproducibility: int  # 1-10: 再現性
    exploitability: int   # 1-10: 攻撃の容易さ
    affected_users: int   # 1-10: 影響範囲
    discoverability: int  # 1-10: 発見可能性
 
    def __post_init__(self):
        fields = {
            "damage": self.damage,
            "reproducibility": self.reproducibility,
            "exploitability": self.exploitability,
            "affected_users": self.affected_users,
            "discoverability": self.discoverability,
        }
        for name, value in fields.items():
            if not 1 <= value <= 10:
                raise ValueError(f"{name}は1-10の範囲: {value}")
 
    @property
    def total(self) -> float:
        """DREADスコア（平均値）"""
        return (self.damage + self.reproducibility +
                self.exploitability + self.affected_users +
                self.discoverability) / 5.0
 
    @property
    def risk_level(self) -> str:
        score = self.total
        if score >= 8:
            return "Critical"
        elif score >= 6:
            return "High"
        elif score >= 4:
            return "Medium"
        return "Low"
 
    @property
    def risk_color(self) -> str:
        """リスクレベルに対応する色コード"""
        colors = {
            "Critical": "RED",
            "High": "ORANGE",
            "Medium": "YELLOW",
            "Low": "GREEN",
        }
        return colors[self.risk_level]
 
    def breakdown(self) -> str:
        """スコアの内訳を表示"""
        items = [
            ("Damage", self.damage),
            ("Reproducibility", self.reproducibility),
            ("Exploitability", self.exploitability),
            ("Affected Users", self.affected_users),
            ("Discoverability", self.discoverability),
        ]
        lines = []
        for name, value in items:
            bar = "█" * value + "░" * (10 - value)
            lines.append(f"  {name:<20s} [{bar}] {value}/10")
        lines.append(f"  {'Total':<20s}  => {self.total:.1f}/10 ({self.risk_level})")
        return "\n".join(lines)
 
# 代表的な脅威のDREADスコア比較
threats = {
    "SQLインジェクション": DreadScore(9, 10, 7, 10, 8),
    "XSS (Stored)": DreadScore(7, 9, 6, 8, 7),
    "CSRF": DreadScore(6, 8, 5, 7, 6),
    "ブルートフォース": DreadScore(5, 10, 8, 3, 9),
    "DDoS": DreadScore(4, 10, 9, 10, 10),
}
 
print("=== 脅威 DREAD スコア比較 ===\n")
for name, score in sorted(
    threats.items(), key=lambda x: x[1].total, reverse=True
):
    print(f"--- {name} ---")
    print(score.breakdown())
    print()

# コード例3: アタックツリーの構築と最小コスト経路分析
from dataclasses import dataclass, field
from typing import List, Optional, Tuple
 
@dataclass
class AttackNode:
    """アタックツリーのノード"""
    name: str
    description: str = ""
    cost: int = 0             # 攻撃コスト（低いほど容易）
    difficulty: str = "Medium"  # Low/Medium/High
    probability: float = 0.5  # 成功確率（0.0-1.0）
    children: List['AttackNode'] = field(default_factory=list)
    is_and: bool = False      # True=AND条件, False=OR条件
    mitigations: List[str] = field(default_factory=list)
 
    def add_child(self, child: 'AttackNode') -> 'AttackNode':
        self.children.append(child)
        return self  # メソッドチェーン
 
    def min_cost(self) -> int:
        """最小攻撃コストを再帰的に計算"""
        if not self.children:
            return self.cost
        child_costs = [c.min_cost() for c in self.children]
        if self.is_and:
            return sum(child_costs)  # AND: 全子ノードのコスト合計
        return min(child_costs)      # OR: 最安経路
 
    def max_probability(self) -> float:
        """最大成功確率を再帰的に計算"""
        if not self.children:
            return self.probability
        child_probs = [c.max_probability() for c in self.children]
        if self.is_and:
            # AND: 全子ノードの確率の積
            result = 1.0
            for p in child_probs:
                result *= p
            return result
        return max(child_probs)  # OR: 最高確率の経路
 
    def find_cheapest_path(self) -> List[str]:
        """最もコストの低い攻撃経路を特定"""
        if not self.children:
            return [self.name]
 
        if self.is_and:
            # AND: すべての子を含む
            path = [self.name]
            for child in self.children:
                path.extend(child.find_cheapest_path())
            return path
 
        # OR: 最安の子を選択
        cheapest = min(self.children, key=lambda c: c.min_cost())
        return [self.name] + cheapest.find_cheapest_path()
 
    def display(self, indent: int = 0) -> None:
        """ツリーを視覚的に表示"""
        prefix = "  " * indent
        cost_str = f"${self.cost}K" if self.cost else ""
        prob_str = f"P={self.probability:.0%}" if not self.children else ""
        connector = "AND" if self.is_and else "OR"
 
        print(f"{prefix}[{self.name}] {cost_str} {prob_str}")
        if self.mitigations:
            for m in self.mitigations:
                print(f"{prefix}  >> 対策: {m}")
        if self.children:
            print(f"{prefix}  ({connector})")
            for child in self.children:
                child.display(indent + 2)
 
# アタックツリーの構築例
root = AttackNode("顧客情報の窃取", "ECサイトの顧客データを不正取得")
 
# Webアプリ経由
web_attack = AttackNode("Webアプリ経由", "アプリ脆弱性を利用")
sqli = AttackNode("SQLインジェクション", "入力フォーム経由",
                  cost=3, difficulty="Low", probability=0.7,
                  mitigations=["パラメータ化クエリ", "WAF", "入力バリデーション"])
xss = AttackNode("XSS→セッション窃取", "Stored XSSでCookie窃取",
                 cost=5, difficulty="Medium", probability=0.4,
                 mitigations=["CSP", "HttpOnly Cookie", "出力エスケープ"])
web_attack.add_child(sqli)
web_attack.add_child(xss)
 
# 内部者経由
insider = AttackNode("内部者経由", "組織内部からの攻撃")
db_direct = AttackNode("DB直接アクセス", "認証情報の不正利用",
                       cost=10, difficulty="High", probability=0.2,
                       mitigations=["最小権限", "特権アクセス管理", "監査ログ"])
backup_theft = AttackNode("バックアップ窃取", "暗号化なしバックアップ",
                          cost=2, difficulty="Low", probability=0.3,
                          mitigations=["バックアップ暗号化", "アクセス制限", "DLP"])
insider.add_child(db_direct)
insider.add_child(backup_theft)
 
root.add_child(web_attack)
root.add_child(insider)
 
# 分析結果
root.display()
print(f"\n最小攻撃コスト: ${root.min_cost()}K")
print(f"最大成功確率: {root.max_probability():.0%}")
print(f"最安経路: {' → '.join(root.find_cheapest_path())}")

# コード例4: DFD解析と信頼境界を越えるフローの脅威自動検出
from dataclasses import dataclass, field
from typing import List, Set, Dict
 
@dataclass
class DFDComponent:
    """DFDのコンポーネント"""
    name: str
    component_type: str  # external_entity, process, data_store
    trust_zone: str       # untrusted, dmz, trusted, highly_trusted
    description: str = ""
 
@dataclass
class DataFlow:
    """データフロー"""
    source: str
    destination: str
    data_type: str
    protocol: str
    encrypted: bool
    authenticated: bool
    data_classification: str = "internal"  # public, internal, confidential, restricted
 
@dataclass
class TrustBoundary:
    """信頼境界"""
    name: str
    trust_level: str  # untrusted, dmz, trusted, highly_trusted
    components: Set[str] = field(default_factory=set)
 
class DFDThreatAnalyzer:
    """DFDに基づく自動脅威検出エンジン"""
 
    TRUST_HIERARCHY = ["untrusted", "dmz", "trusted", "highly_trusted"]
 
    def __init__(self, system_name: str):
        self.system_name = system_name
        self.components: Dict[str, DFDComponent] = {}
        self.flows: List[DataFlow] = []
        self.boundaries: List[TrustBoundary] = []
 
    def add_component(self, component: DFDComponent) -> None:
        self.components[component.name] = component
 
    def add_flow(self, flow: DataFlow) -> None:
        self.flows.append(flow)
 
    def add_boundary(self, boundary: TrustBoundary) -> None:
        self.boundaries.append(boundary)
 
    def _get_trust_level(self, component_name: str) -> str:
        """コンポーネントの信頼レベルを取得"""
        comp = self.components.get(component_name)
        return comp.trust_zone if comp else "untrusted"
 
    def _crosses_boundary(self, flow: DataFlow) -> bool:
        """フローが信頼境界を越えるか判定"""
        src_trust = self._get_trust_level(flow.source)
        dst_trust = self._get_trust_level(flow.destination)
        return src_trust != dst_trust
 
    def _trust_direction(self, flow: DataFlow) -> str:
        """信頼方向を判定（inbound/outbound/lateral）"""
        src_idx = self.TRUST_HIERARCHY.index(
            self._get_trust_level(flow.source)
        )
        dst_idx = self.TRUST_HIERARCHY.index(
            self._get_trust_level(flow.destination)
        )
        if src_idx < dst_idx:
            return "inbound"   # 低信頼→高信頼（要注意）
        elif src_idx > dst_idx:
            return "outbound"  # 高信頼→低信頼（情報漏洩リスク）
        return "lateral"
 
    def detect_threats(self) -> List[Dict]:
        """脅威を自動検出"""
        threats = []
 
        for flow in self.flows:
            flow_threats = []
 
            # 脅威1: 信頼境界を越える非暗号化フロー
            if self._crosses_boundary(flow) and not flow.encrypted:
                flow_threats.append({
                    "type": "TAMPERING/INFO_DISCLOSURE",
                    "severity": "High",
                    "description": (
                        f"信頼境界を越えるフロー "
                        f"({flow.source} → {flow.destination}) "
                        f"が暗号化されていない"
                    ),
                    "mitigation": "TLS/mTLS の導入",
                })
 
            # 脅威2: 認証なしの境界越えフロー
            if self._crosses_boundary(flow) and not flow.authenticated:
                direction = self._trust_direction(flow)
                if direction == "inbound":
                    flow_threats.append({
                        "type": "SPOOFING",
                        "severity": "Critical",
                        "description": (
                            f"低信頼ゾーンから高信頼ゾーンへの"
                            f"認証なしフロー "
                            f"({flow.source} → {flow.destination})"
                        ),
                        "mitigation": "認証メカニズムの実装",
                    })
 
            # 脅威3: 機密データの非暗号化フロー
            if flow.data_classification in ("confidential", "restricted"):
                if not flow.encrypted:
                    flow_threats.append({
                        "type": "INFO_DISCLOSURE",
                        "severity": "Critical",
                        "description": (
                            f"機密データ({flow.data_type}) が "
                            f"暗号化なしで転送されている"
                        ),
                        "mitigation": "転送時暗号化(TLS)と保存時暗号化の実装",
                    })
 
            for t in flow_threats:
                t["flow"] = f"{flow.source} → {flow.destination}"
                t["data_type"] = flow.data_type
                threats.append(t)
 
        return threats
 
    def generate_report(self) -> str:
        """脅威分析レポートを生成"""
        threats = self.detect_threats()
        lines = [
            f"# DFD脅威分析レポート: {self.system_name}",
            f"## コンポーネント数: {len(self.components)}",
            f"## データフロー数: {len(self.flows)}",
            f"## 検出された脅威: {len(threats)}件\n",
        ]
 
        by_severity = {}
        for t in threats:
            by_severity.setdefault(t["severity"], []).append(t)
 
        for severity in ["Critical", "High", "Medium", "Low"]:
            sev_threats = by_severity.get(severity, [])
            if sev_threats:
                lines.append(f"### {severity} ({len(sev_threats)}件)")
                for t in sev_threats:
                    lines.append(f"- [{t['type']}] {t['description']}")
                    lines.append(f"  フロー: {t['flow']}")
                    lines.append(f"  対策: {t['mitigation']}")
                lines.append("")
 
        return "\n".join(lines)
 
# 使用例
analyzer = DFDThreatAnalyzer("ECサイト")
 
# コンポーネント定義
analyzer.add_component(DFDComponent(
    "ブラウザ", "external_entity", "untrusted", "ユーザーのWebブラウザ"
))
analyzer.add_component(DFDComponent(
    "APIサーバー", "process", "trusted", "バックエンドAPI"
))
analyzer.add_component(DFDComponent(
    "データベース", "data_store", "highly_trusted", "PostgreSQL"
))
 
# データフロー定義
analyzer.add_flow(DataFlow(
    "ブラウザ", "APIサーバー", "認証情報",
    "HTTPS", encrypted=True, authenticated=True,
    data_classification="confidential",
))
analyzer.add_flow(DataFlow(
    "APIサーバー", "データベース", "SQLクエリ",
    "TCP", encrypted=False, authenticated=True,
    data_classification="internal",
))
 
threats = analyzer.detect_threats()
print(analyzer.generate_report())

# コード例5: 脅威モデリングワークシートの完全実装
import json
from datetime import datetime
from typing import List, Dict, Optional
from dataclasses import dataclass, field
 
@dataclass
class ThreatModelWorksheet:
    """脅威モデリングワークシートの生成・管理"""
 
    project: str
    version: str
    author: str = ""
    created: str = field(default_factory=lambda: datetime.now().isoformat())
    components: List[Dict] = field(default_factory=list)
    threats: List[Dict] = field(default_factory=list)
    mitigations: List[Dict] = field(default_factory=list)
    assumptions: List[str] = field(default_factory=list)
    scope: str = ""
 
    def add_component(self, name: str, component_type: str,
                      trust_level: str, description: str = "") -> None:
        """DFDコンポーネントを追加"""
        self.components.append({
            "name": name,
            "type": component_type,
            "trust_level": trust_level,
            "description": description,
        })
 
    def add_threat(self, stride_cat: str, component: str,
                   description: str, attack_scenario: str,
                   dread: Optional[Dict] = None) -> str:
        """脅威を追加"""
        threat_id = f"T-{len(self.threats) + 1:03d}"
        dread_score = 0
        if dread:
            dread_score = sum(dread.values()) / len(dread)
        self.threats.append({
            "id": threat_id,
            "stride_category": stride_cat,
            "component": component,
            "description": description,
            "attack_scenario": attack_scenario,
            "dread_score": round(dread_score, 1) if dread else None,
            "dread_detail": dread,
            "status": "identified",
        })
        return threat_id
 
    def add_mitigation(self, threat_id: str, strategy: str,
                       description: str, effort: str = "medium",
                       assigned_to: str = "") -> None:
        """対策を追加"""
        self.mitigations.append({
            "threat_id": threat_id,
            "strategy": strategy,
            "description": description,
            "effort": effort,
            "assigned_to": assigned_to,
            "implemented": False,
        })
 
    def get_summary(self) -> Dict:
        """サマリーを生成"""
        severity_counts = {"Critical": 0, "High": 0, "Medium": 0, "Low": 0}
        for t in self.threats:
            score = t.get("dread_score", 0) or 0
            if score >= 8:
                severity_counts["Critical"] += 1
            elif score >= 6:
                severity_counts["High"] += 1
            elif score >= 4:
                severity_counts["Medium"] += 1
            else:
                severity_counts["Low"] += 1
 
        return {
            "project": self.project,
            "version": self.version,
            "total_threats": len(self.threats),
            "severity_breakdown": severity_counts,
            "total_mitigations": len(self.mitigations),
            "mitigated": sum(1 for m in self.mitigations if m["implemented"]),
            "pending": sum(1 for m in self.mitigations if not m["implemented"]),
            "coverage": (
                f"{sum(1 for m in self.mitigations if m['implemented'])}"
                f"/{len(self.mitigations)}"
            ),
        }
 
    def export_json(self) -> str:
        """JSON形式でエクスポート"""
        return json.dumps({
            "project": self.project,
            "version": self.version,
            "author": self.author,
            "created": self.created,
            "scope": self.scope,
            "assumptions": self.assumptions,
            "components": self.components,
            "threats": self.threats,
            "mitigations": self.mitigations,
            "summary": self.get_summary(),
        }, indent=2, ensure_ascii=False)
 
# 使用例
ws = ThreatModelWorksheet(
    project="ECサイト決済システム",
    version="v2.0",
    author="セキュリティチーム",
    scope="決済フロー（カート → 注文確定 → 決済処理 → 完了通知）",
)
ws.assumptions = [
    "外部決済ゲートウェイ（Stripe）は信頼できるものとする",
    "TLS 1.3が全通信で使用されている",
    "管理者は信頼できるが、最小権限の原則は適用する",
]
 
ws.add_component("ブラウザ", "external_entity", "untrusted", "顧客のブラウザ")
ws.add_component("APIサーバー", "process", "trusted", "Node.js Express")
ws.add_component("PostgreSQL", "data_store", "highly_trusted", "注文・顧客DB")
ws.add_component("Stripe API", "external_entity", "semi-trusted", "決済処理")
 
t1 = ws.add_threat(
    "Tampering", "APIサーバー",
    "注文金額の改ざんによる不正な低額決済",
    "MITM/プロキシで /api/checkout のPOSTボディを改ざん",
    {"D": 9, "R": 10, "E": 6, "A": 8, "D2": 7},
)
ws.add_mitigation(t1, "mitigate",
    "サーバーサイドでの金額再計算（カートIDから商品価格を再取得）",
    effort="low", assigned_to="バックエンドチーム")
 
t2 = ws.add_threat(
    "Spoofing", "ブラウザ",
    "他ユーザーの注文情報閲覧（IDOR）",
    "GET /api/orders/{id} のidを変更して他者の注文を閲覧",
    {"D": 7, "R": 10, "E": 9, "A": 5, "D2": 8},
)
ws.add_mitigation(t2, "mitigate",
    "オーナーシップチェック: order.user_id == current_user.id",
    effort="low", assigned_to="バックエンドチーム")
 
summary = ws.get_summary()
print(f"プロジェクト: {summary['project']}")
print(f"脅威数: {summary['total_threats']}")
print(f"深刻度: {summary['severity_breakdown']}")

# NG: 脅威モデリングを省略してリリース
class ReleaseProcess:
    def ship(self, feature):
        # 脅威モデリングなし → 本番で脆弱性が見つかる
        # 修正コスト: $100,000+
        deploy(feature)
 
# OK: 最小限の脅威モデリングを必ず実施
class SecureReleaseProcess:
    def ship(self, feature):
        # 最低限のSTRIDE分析（30分〜1時間）
        threats = self.quick_stride_analysis(feature)
        if any(t.risk_level == "Critical" for t in threats):
            raise SecurityGateError("Critical脅威が未対策です")
        # 対策完了後にリリース
        # 修正コスト: $100（設計段階）
        deploy(feature)

# NG: 初回のみ実施
threat_model = create_threat_model()  # v1.0でのみ実施
# v2.0で新機能追加 → 脅威モデル未更新 → 新しい脆弱性が放置
 
# OK: 変更のたびに差分分析
class ContinuousThreatModeling:
    def on_feature_change(self, feature_spec):
        """機能変更時に脅威モデルを差分更新"""
        # 1. 変更されたDFD要素を特定
        changed_elements = self.diff_dfd(feature_spec)
        # 2. 変更要素に対するSTRIDE分析
        new_threats = self.analyze_stride(changed_elements)
        # 3. 既存の脅威モデルに統合
        self.merge_threats(new_threats)
        # 4. レビューとサインオフ
        self.request_review(new_threats)