securing-remote-access-to-ot-environment

OT 環境へのリモートアクセスの保護

使用する場面

• OT 環境のリモートアクセスアーキテクチャを導入またはアップグレードする場合
• サポートや保守のために OT システムへのリモートアクセスを必要とするベンダーを受け入れる場合
• MFA を含むリモートアクセス管理に関する CIP-005-7 R2 の要件を実装する場合
• OT ネットワークへの従来型の直接 VPN アクセスを、セキュアなジャンプサーバーアーキテクチャに置き換える場合
• 産業用制御システムへの不正なリモートアクセスを伴うインシデントに対応する場合

使用しないでください。OT コンポーネントを含まない IT 専用のリモートアクセスを保護する場合、社内の従業員向けに VPN を設定する場合（一般的な VPN ガイドを参照）、または OT 施設への物理アクセスを制御する場合には使用しないでください。

前提条件

• 企業の IT ネットワークと OT ネットワークの間にある DMZ インフラストラクチャ（Level 3.5）
• ジャンプサーバー／踏み台ホストのプラットフォーム（CyberArk、BeyondTrust、または強化された Windows/Linux サーバー）
• 多要素認証ソリューション（Duo、RSA SecurID、YubiKey、スマートカード）
• 監査証跡のコンプライアンスに対応するセッション記録機能
• DMZ の中間システム経由でのみリモートアクセスを許可するファイアウォールルール

ワークフロー

ステップ 1：セキュアなリモートアクセスアーキテクチャを設計する

外部ユーザーと OT システム間の直接的なネットワーク接続を一切防ぐため、DMZ に中間システムを配置した多層防御のリモートアクセスアーキテクチャを実装します。

# OT Remote Access Architecture
# Key principle: NO direct connection from external networks to OT systems

architecture:
  external_access_point:
    location: "Internet-facing firewall"
    service: "VPN gateway (IKEv2/IPsec or SSL VPN)"
    authentication: "Certificate + MFA (CIP-005-7 R2.4)"
    controls:
      - "Source IP allowlisting for vendor access"
      - "Time-based access windows"
      - "Bandwidth rate limiting"

  dmz_intermediate_system:
    location: "Level 3.5 DMZ"
    platform: "CyberArk Privileged Access Security or hardened jump server"
    function: "Session broker - terminates external connection, initiates new internal connection"
    controls:
      - "All sessions terminate at jump server (no pass-through)"
      - "Session recording with keystroke logging"
      - "Clipboard and file transfer restrictions"
      - "Session timeout after 30 minutes inactivity"
      - "Concurrent session limits per user"

  ot_internal_access:
    location: "Level 3 / Level 2 OT network"
    method: "Jump server initiates RDP/SSH/VNC to OT systems"
    controls:
      - "Firewall allows only jump server IP to reach OT"
      - "Protocol restricted (RDP 3389, SSH 22, VNC 5900)"
      - "Destination-specific per user role"

  vendor_access:
    description: "Third-party vendor remote access"
    additional_controls:
      - "Vendor account disabled by default; enabled on request"
      - "Time-limited access windows (enable/disable per session)"
      - "OT operator must co-attend vendor sessions"
      - "Real-time session monitoring by OT security"
      - "No persistent credentials - one-time access tokens"

# Network flow:
# User -> VPN -> Firewall -> DMZ Jump Server -> Internal FW -> OT System
# (Two separate authenticated connections; no direct routing)

ステップ 2：特権アクセス管理を備えたジャンプサーバーを設定する

セッション管理、記録、ロールベースのアクセス制御を備えたジャンプサーバーを DMZ にデプロイし、セキュリティを強化します。

#!/usr/bin/env python3
"""OT Remote Access Session Manager.

Manages authorized remote access sessions to OT environments
including session creation, monitoring, recording, and termination.
Integrates with PAM solutions for credential vaulting.
"""

import json
import hashlib
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from enum import Enum


class SessionState(str, Enum):
    PENDING_APPROVAL = "pending_approval"
    APPROVED = "approved"
    ACTIVE = "active"
    TERMINATED = "terminated"
    EXPIRED = "expired"
    DENIED = "denied"


class UserRole(str, Enum):
    OT_OPERATOR = "ot_operator"
    OT_ENGINEER = "ot_engineer"
    VENDOR = "vendor"
    SECURITY_ANALYST = "security_analyst"


@dataclass
class RemoteAccessSession:
    session_id: str
    user_id: str
    user_role: str
    source_ip: str
    target_system: str
    target_ip: str
    protocol: str
    purpose: str
    state: str = SessionState.PENDING_APPROVAL
    mfa_verified: bool = False
    approved_by: str = ""
    created_at: str = ""
    started_at: str = ""
    ended_at: str = ""
    max_duration_minutes: int = 120
    recording_path: str = ""
    actions_logged: list = field(default_factory=list)


class OTRemoteAccessManager:
    """Manages remote access sessions to OT environment."""

    def __init__(self):
        self.sessions = {}
        self.active_sessions = {}
        self.access_policies = {}
        self.audit_log = []

    def define_access_policy(self, role, allowed_targets, protocols, max_duration):
        """Define access policy for a user role."""
        self.access_policies[role] = {
            "allowed_targets": allowed_targets,
            "allowed_protocols": protocols,
            "max_duration_minutes": max_duration,
            "requires_co_attendance": role == UserRole.VENDOR,
            "requires_approval": role == UserRole.VENDOR,
        }

    def request_session(self, user_id, user_role, source_ip,
                        target_system, target_ip, protocol, purpose):
        """Request a new remote access session."""
        # Generate unique session ID
        session_id = hashlib.sha256(
            f"{user_id}{target_ip}{datetime.now().isoformat()}".encode()
        ).hexdigest()[:16]

        # Check access policy
        policy = self.access_policies.get(user_role)
        if not policy:
            return None, "No access policy defined for role"

        if target_system not in policy["allowed_targets"]:
            self._audit("ACCESS_DENIED", user_id, target_system,
                       f"Target not in allowed list for role {user_role}")
            return None, f"Target {target_system} not authorized for role {user_role}"

        if protocol not in policy["allowed_protocols"]:
            self._audit("ACCESS_DENIED", user_id, target_system,
                       f"Protocol {protocol} not allowed for role {user_role}")
            return None, f"Protocol {protocol} not authorized"

        session = RemoteAccessSession(
            session_id=session_id,
            user_id=user_id,
            user_role=user_role,
            source_ip=source_ip,
            target_system=target_system,
            target_ip=target_ip,
            protocol=protocol,
            purpose=purpose,
            max_duration_minutes=policy["max_duration_minutes"],
            created_at=datetime.now().isoformat(),
        )

        # Vendor sessions require approval
        if policy.get("requires_approval"):
            session.state = SessionState.PENDING_APPROVAL
        else:
            session.state = SessionState.APPROVED

        self.sessions[session_id] = session
        self._audit("SESSION_REQUESTED", user_id, target_system, purpose)

        return session_id, "Session created"

    def approve_session(self, session_id, approver_id):
        """Approve a pending vendor session."""
        session = self.sessions.get(session_id)
        if not session:
            return False, "Session not found"
        if session.state != SessionState.PENDING_APPROVAL:
            return False, "Session not in pending approval state"

        session.state = SessionState.APPROVED
        session.approved_by = approver_id
        self._audit("SESSION_APPROVED", approver_id, session.target_system,
                    f"Approved session {session_id} for {session.user_id}")
        return True, "Session approved"

    def activate_session(self, session_id, mfa_token):
        """Activate an approved session after MFA verification."""
        session = self.sessions.get(session_id)
        if not session or session.state != SessionState.APPROVED:
            return False, "Session not approved"

        # Verify MFA (simplified - real implementation calls MFA provider)
        session.mfa_verified = True
        session.state = SessionState.ACTIVE
        session.started_at = datetime.now().isoformat()
        session.recording_path = f"/recordings/{session_id}_{datetime.now().strftime('%Y%m%d')}.mp4"

        self.active_sessions[session_id] = session
        self._audit("SESSION_ACTIVATED", session.user_id, session.target_system,
                    f"MFA verified, recording to {session.recording_path}")

        return True, "Session active"

    def terminate_session(self, session_id, reason="manual"):
        """Terminate an active session."""
        session = self.active_sessions.pop(session_id, None)
        if not session:
            session = self.sessions.get(session_id)
        if not session:
            return False

        session.state = SessionState.TERMINATED
        session.ended_at = datetime.now().isoformat()
        self._audit("SESSION_TERMINATED", session.user_id, session.target_system, reason)
        return True

    def check_expired_sessions(self):
        """Terminate sessions that have exceeded their maximum duration."""
        now = datetime.now()
        expired = []
        for sid, session in list(self.active_sessions.items()):
            started = datetime.fromisoformat(session.started_at)
            if (now - started).total_seconds() > session.max_duration_minutes * 60:
                self.terminate_session(sid, "maximum_duration_exceeded")
                expired.append(sid)
        return expired

    def _audit(self, event_type, user_id, target, detail):
        """Write to audit log."""
        self.audit_log.append({
            "timestamp": datetime.now().isoformat(),
            "event": event_type,
            "user": user_id,
            "target": target,
            "detail": detail,
        })

    def generate_report(self):
        """Generate remote access session report."""
        report = []
        report.append("=" * 70)
        report.append("OT REMOTE ACCESS SESSION REPORT")
        report.append(f"Date: {datetime.now().isoformat()}")
        report.append("=" * 70)

        # Session summary
        total = len(self.sessions)
        active = sum(1 for s in self.sessions.values() if s.state == SessionState.ACTIVE)
        report.append(f"\nTotal Sessions: {total}")
        report.append(f"Active Sessions: {active}")

        # Active sessions detail
        if self.active_sessions:
            report.append("\nACTIVE SESSIONS:")
            for sid, s in self.active_sessions.items():
                report.append(f"  [{sid[:8]}] {s.user_id} ({s.user_role})")
                report.append(f"    Target: {s.target_system} ({s.target_ip})")
                report.append(f"    Started: {s.started_at}")
                report.append(f"    MFA: {'Verified' if s.mfa_verified else 'NOT VERIFIED'}")

        return "\n".join(report)


if __name__ == "__main__":
    manager = OTRemoteAccessManager()

    # Define policies
    manager.define_access_policy(
        UserRole.OT_ENGINEER,
        ["HMI-01", "HMI-02", "EWS-01", "HISTORIAN-01"],
        ["RDP", "SSH"],
        max_duration=240,
    )
    manager.define_access_policy(
        UserRole.VENDOR,
        ["DCS-EWS-01"],
        ["RDP"],
        max_duration=120,
    )

    # Request vendor session
    sid, msg = manager.request_session(
        "vendor_honeywell_01", UserRole.VENDOR,
        "203.0.113.50", "DCS-EWS-01", "10.30.1.20",
        "RDP", "DCS firmware update per change request CR-2026-0045")
    print(f"Session request: {msg} ({sid})")

    if sid:
        manager.approve_session(sid, "ot_manager_01")
        manager.activate_session(sid, "123456")
        print(manager.generate_report())

主な概念

用語	定義
中間システム	外部接続を終端し、OT への新たな接続を仲介する DMZ 内のシステム。CIP-005 に従い、直接的なネットワークアクセスを防ぎます
ジャンプサーバー	セッション記録とアクセス制御を備え、OT システムへのリモートアクセスセッションに使用される、DMZ 内の強化された踏み台ホスト
セッション記録	セキュリティ監査やインシデント調査に備え、すべてのリモートアクセスセッションのアクティビティ（画面、キー入力、コマンド）を記録すること
特権アクセス管理（PAM）	重要な OT システムに対する認証情報の保管、アクセス制御、特権セッションの監査を行うシステム
立ち会い	ベンダーのリモートアクセスセッションを OT オペレーターがリアルタイムで監視する要件
時間制限付きアクセス	特定の保守時間帯のみベンダーアカウントを有効にし、時間帯終了後に自動的に無効化すること

ツールとシステム

• CyberArk Privileged Access Security：セッション管理、認証情報の保管、OT へのリモートアクセスの記録に対応するエンタープライズ PAM
• BeyondTrust Privileged Remote Access：セッション記録ときめ細かなアクセス制御ポリシーを備えた、専用のリモートアクセスソリューション
• Claroty Secure Remote Access (SRA)：プロトコルを認識するセッション制御機能を備えた、OT 専用のリモートアクセスソリューション
• Duo Security：OT アクセスの認証確認に対応するプッシュ通知、ハードウェアトークン、生体認証をサポートする MFA プロバイダー

出力形式

OT Remote Access Security Report
===================================
Active Sessions: [N]
Pending Approval: [N]
Sessions Today: [N]

MFA COMPLIANCE:
  All sessions MFA verified: [Yes/No]
  Sessions without MFA: [N]

VENDOR ACCESS:
  Active vendor sessions: [N]
  Co-attended: [N]
  Recorded: [N]