performing-ot-vulnerability-scanning-safely

OT の脆弱性スキャンを安全に実施する

使用する場面

• レガシーコントローラーがある OT 環境で脆弱性評価を実施する場合
• プロセスの可用性に影響を与えずに、継続的な脆弱性監視を実施する場合
• 脆弱性データが必要となる IEC 62443 または NERC CIP のコンプライアンス監査に備える場合
• OT 資産のリスクベースのパッチ適用優先度を評価する場合
• 補完的な制御によって、パッチ未適用の ICS デバイスが保護されていることを検証する場合

次の場合には使用しないでください：本番 PLC に対する積極的なアクティブスキャン（レガシーコントローラーがクラッシュする可能性があります）、OT ネットワーク上で標準の Nessus プロファイルを使用した IT 脆弱性スキャン、または稼働中の OT システムへの侵入テスト（performing-ics-penetration-testing を参照）。

前提条件

• Tenable OT Security（旧称 Tenable.ot／Indegy）、または同等の OT セーフなスキャンプラットフォーム
• OT ネットワークセグメントの SPAN／TAP に設置されたパッシブ監視センサー
• 本番環境で使用する前に、各デバイスタイプに対して検証済みのラボテスト済みスキャンプロファイル
• アクティブスキャンを行う場合の変更管理承認とメンテナンス時間枠
• スキャンによってサポート契約が無効にならないことを確認するための、ベンダー保証の確認

ワークフロー

ステップ 1：パッシブ脆弱性検出を導入する

パッシブ監視では、OT デバイスにパケットを送信せずに脆弱性を特定します。

#!/usr/bin/env python3
"""OT Safe Vulnerability Scanner Orchestrator.

Coordinates passive monitoring, native protocol queries, and carefully
controlled active scanning for OT vulnerability assessment without
disrupting industrial operations.
"""

import json
import csv
import sys
from datetime import datetime
from typing import Dict, List, Optional

try:
    import requests
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)


class OTVulnerabilityScanner:
    """Safe OT vulnerability scanning orchestrator."""

    SCAN_SAFETY_LEVELS = {
        "passive": {
            "description": "Observe network traffic only, zero risk to devices",
            "risk_level": "NONE",
            "methods": ["traffic_fingerprinting", "protocol_analysis", "version_detection"],
            "requires_window": False,
        },
        "native_query": {
            "description": "Query devices using native industrial protocols",
            "risk_level": "MINIMAL",
            "methods": ["modbus_device_id", "s7_szl_read", "cip_identity", "bacnet_whois"],
            "requires_window": True,
        },
        "controlled_active": {
            "description": "Standard vulnerability checks with OT-safe profiles",
            "risk_level": "LOW-MODERATE",
            "methods": ["credentialed_scan", "banner_grab", "service_detection"],
            "requires_window": True,
        },
    }

    def __init__(self, tenable_url: str, api_key: str, verify_ssl: bool = True):
        self.tenable_url = tenable_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "X-ApiKeys": f"accessKey={api_key}",
            "Content-Type": "application/json",
        })
        self.session.verify = verify_ssl
        self.findings = []

    def check_safety_prerequisites(self, scan_level: str, target_subnet: str) -> dict:
        """Verify safety prerequisites before scanning."""
        checks = {
            "scan_level": scan_level,
            "target": target_subnet,
            "safety_level": self.SCAN_SAFETY_LEVELS[scan_level],
            "checks_passed": [],
            "checks_failed": [],
            "approved": False,
        }

        prerequisites = [
            {
                "name": "Lab validation complete",
                "description": "Scan profile tested against each device type in lab environment",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "Vendor warranty verified",
                "description": "Scanning will not void vendor support agreements",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "Change management approved",
                "description": "Change ticket approved for scanning activity",
                "required_for": ["native_query", "controlled_active"],
            },
            {
                "name": "Maintenance window confirmed",
                "description": "Operations team confirms acceptable scanning window",
                "required_for": ["controlled_active"],
            },
            {
                "name": "Rollback plan documented",
                "description": "Procedure to stop scan and recover if device becomes unresponsive",
                "required_for": ["controlled_active"],
            },
            {
                "name": "SIS excluded from scope",
                "description": "Safety Instrumented Systems are never actively scanned",
                "required_for": ["passive", "native_query", "controlled_active"],
            },
        ]

        for prereq in prerequisites:
            if scan_level in prereq["required_for"]:
                checks["checks_passed"].append(prereq["name"])

        return checks

    def run_passive_assessment(self, site_id: str):
        """Run passive vulnerability assessment using traffic analysis."""
        print(f"[*] Running passive vulnerability assessment for site {site_id}")
        print(f"[*] Safety Level: NONE - no packets sent to OT devices")

        try:
            resp = self.session.get(
                f"{self.tenable_url}/api/v1/assets",
                params={"site_id": site_id}
            )
            resp.raise_for_status()
            assets = resp.json().get("assets", [])

            for asset in assets:
                asset_id = asset.get("id")
                vuln_resp = self.session.get(
                    f"{self.tenable_url}/api/v1/assets/{asset_id}/vulnerabilities"
                )
                if vuln_resp.status_code == 200:
                    vulns = vuln_resp.json().get("vulnerabilities", [])
                    for vuln in vulns:
                        self.findings.append({
                            "asset": asset.get("name", "Unknown"),
                            "ip": asset.get("ip_address", ""),
                            "type": asset.get("type", ""),
                            "vendor": asset.get("vendor", ""),
                            "cve": vuln.get("cve_id", ""),
                            "severity": vuln.get("severity", ""),
                            "cvss": vuln.get("cvss_score", 0),
                            "description": vuln.get("description", ""),
                            "detection_method": "passive",
                            "remediation": vuln.get("remediation", ""),
                        })

            print(f"[+] Passive assessment complete: {len(self.findings)} vulnerabilities found")
        except requests.RequestException as e:
            print(f"[!] API error: {e}")

    def generate_prioritized_report(self, output_file: str):
        """Generate risk-prioritized vulnerability report for OT environment."""
        self.findings.sort(key=lambda x: x.get("cvss", 0), reverse=True)

        print(f"\n{'='*70}")
        print("OT VULNERABILITY ASSESSMENT REPORT")
        print(f"{'='*70}")
        print(f"Date: {datetime.now().isoformat()}")
        print(f"Total Findings: {len(self.findings)}")

        severity_counts = {}
        for f in self.findings:
            sev = f.get("severity", "Unknown")
            severity_counts[sev] = severity_counts.get(sev, 0) + 1

        print(f"\nSeverity Distribution:")
        for sev in ["Critical", "High", "Medium", "Low"]:
            print(f"  {sev}: {severity_counts.get(sev, 0)}")

        # Risk-based prioritization considering OT context
        print(f"\n--- RISK-PRIORITIZED FINDINGS ---")
        print(f"(Prioritized by CVSS score and OT impact)")
        for i, finding in enumerate(self.findings[:20], 1):
            print(f"\n  {i}. [{finding['severity']}] {finding['cve']}")
            print(f"     Asset: {finding['asset']} ({finding['ip']})")
            print(f"     Vendor: {finding['vendor']} | Type: {finding['type']}")
            print(f"     CVSS: {finding['cvss']}")
            print(f"     Detection: {finding['detection_method']}")
            print(f"     Description: {finding['description'][:100]}")
            if finding.get("remediation"):
                print(f"     Remediation: {finding['remediation'][:100]}")

        # Export to CSV
        if output_file:
            with open(output_file, "w", newline="") as f:
                writer = csv.DictWriter(f, fieldnames=self.findings[0].keys())
                writer.writeheader()
                writer.writerows(self.findings)
            print(f"\n[+] Report exported to {output_file}")


if __name__ == "__main__":
    scanner = OTVulnerabilityScanner(
        tenable_url="https://tenable-ot.plant.local",
        api_key="your-api-key-here",
        verify_ssl=True,
    )

    # Always start with passive assessment
    safety_check = scanner.check_safety_prerequisites("passive", "10.10.0.0/16")
    print(f"Safety prerequisites: {json.dumps(safety_check, indent=2)}")

    scanner.run_passive_assessment(site_id="plant-01")
    scanner.generate_prioritized_report("ot_vulnerabilities.csv")

主な概念

用語	定義
パッシブ脆弱性検出	OT デバイスにパケットを送信せず、ミラーリングされたトラフィックを分析して脆弱性を特定すること
ネイティブプロトコルクエリ	産業用プロトコル（Modbus FC43、S7 SZL Read、CIP Get Attribute）を使用して、デバイス情報を安全に抽出すること
OT セーフなスキャンプロファイル	産業用コントローラーのクラッシュを避けるよう設計され、ラボでテストされた脆弱性スキャナーの設定
補完的な制御	パッチ未適用の OT 資産を保護する代替セキュリティ対策（ファイアウォールの DPI、ネットワーク分離）
OT 環境における CVSS	安全性、可用性、物理的な影響を考慮して調整された標準の CVSS スコア
Tenable OT Security	パッシブ検出とネイティブプロトコルベースの検出を使用する、OT 向けに特化した脆弱性管理プラットフォーム

出力形式

OT VULNERABILITY ASSESSMENT REPORT
=====================================
Date: YYYY-MM-DD
Scope: [network segments]
Method: [Passive/Native Query/Controlled Active]

VULNERABILITY SUMMARY:
  Critical: [count]
  High: [count]
  Medium: [count]
  Low: [count]

TOP RISK FINDINGS:
  1. [CVE] - [CVSS] - [Asset] - [Description]

UNPATACHABLE ASSETS REQUIRING COMPENSATING CONTROLS:
  [Asset] - [Reason] - [Recommended Control]

PATCH PRIORITIZATION:
  Immediate: [list]
  Next Window: [list]
  Acceptable Risk: [list with justification]