| name | securing-remote-access-to-ot-environment |
| description | 本技能涵盖为操作员、工程师和供应商实施OT/ICS环境的安全远程访问，同时防止可能危害工业运营的未授权访问。涉及跳板服务器架构、多因素认证、会话记录、特权访问管理、供应商远程访问控制，以及符合IEC 62443和NERC CIP-005远程访问要求。
|
| domain | cybersecurity |
| subdomain | ot-ics-security |
| tags | ["ot-security","ics","scada","industrial-control","iec62443","remote-access","jump-server","mfa"] |
| version | 1.0.0 |
| author | mahipal |
| license | Apache-2.0 |
保护OT环境的远程访问
适用场景
- 为OT环境实施或升级远程访问架构时
- 需要远程访问OT系统进行支持和维护的供应商入驻时
- 实施CIP-005-7 R2中包括MFA在内的远程访问管理要求时
- 用安全跳板服务器架构替换对OT网络的旧版直接VPN访问时
- 响应涉及对工业控制系统未授权远程访问的事件时
不适用于保护没有OT组件的纯IT远程访问、为企业员工配置VPN（参见通用VPN指南），或OT设施的物理访问控制。
前置条件
- 企业IT网络与OT网络之间的DMZ基础设施（第3.5层）
- 跳板服务器/堡垒主机平台（CyberArk、BeyondTrust或加固的Windows/Linux服务器）
- 多因素认证解决方案（Duo、RSA SecurID、YubiKey、智能卡）
- 用于审计跟踪合规的会话记录功能
- 仅允许通过DMZ中间系统进行远程访问的防火墙规则
工作流程
步骤 1：设计安全远程访问架构
在DMZ中实施具有中间系统的深度防御远程访问架构，防止外部用户与OT系统之间的任何直接网络连接。
architecture:
external_access_point:
location: "面向互联网的防火墙"
service: "VPN网关（IKEv2/IPsec或SSL VPN）"
authentication: "证书 + MFA（CIP-005-7 R2.4）"
controls:
- "供应商访问的源IP白名单"
- "基于时间的访问窗口"
- "带宽速率限制"
dmz_intermediate_system:
location: "第3.5层DMZ"
platform: "CyberArk特权访问安全或加固的跳板服务器"
function: "会话代理——终止外部连接，发起新的内部连接"
controls:
- "所有会话在跳板服务器终止（不直通）"
- "带键盘记录的会话录制"
- "剪贴板和文件传输限制"
- "不活跃30分钟后会话超时"
- "每用户并发会话限制"
ot_internal_access:
location: "第3层/第2层OT网络"
method: "跳板服务器向OT系统发起RDP/SSH/VNC连接"
controls:
- "防火墙仅允许跳板服务器IP到达OT"
- "协议受限（RDP 3389、SSH 22、VNC 5900）"
- "按用户角色限定目标"
vendor_access:
description: "第三方供应商远程访问"
additional_controls:
步骤 2：配置带特权访问管理的跳板服务器
在DMZ中部署和加固跳板服务器，配置会话管理、记录和基于角色的访问控制。
"""OT远程访问会话管理器。
管理对OT环境的授权远程访问会话，
包括会话创建、监控、录制和终止。
与PAM解决方案集成以进行凭据存管。
"""
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: =
max_duration_minutes: =
recording_path: =
actions_logged: = field(default_factory=)
:
():
.sessions = {}
.active_sessions = {}
.access_policies = {}
.audit_log = []
():
.access_policies[role] = {
: allowed_targets,
: protocols,
: max_duration,
: role == UserRole.VENDOR,
: role == UserRole.VENDOR,
}
():
session_id = hashlib.sha256(
.encode()
).hexdigest()[:]
policy = .access_policies.get(user_role)
policy:
,
target_system policy[]:
._audit(, user_id, target_system,
)
,
protocol policy[]:
._audit(, user_id, target_system,
)
,
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[],
created_at=datetime.now().isoformat(),
)
policy.get():
session.state = SessionState.PENDING_APPROVAL
:
session.state = SessionState.APPROVED
.sessions[session_id] = session
._audit(, user_id, target_system, purpose)
session_id,
():
session = .sessions.get(session_id)
session:
,
session.state != SessionState.PENDING_APPROVAL:
,
session.state = SessionState.APPROVED
session.approved_by = approver_id
._audit(, approver_id, session.target_system,
)
,
():
session = .sessions.get(session_id)
session session.state != SessionState.APPROVED:
,
session.mfa_verified =
session.state = SessionState.ACTIVE
session.started_at = datetime.now().isoformat()
session.recording_path =
.active_sessions[session_id] = session
._audit(, session.user_id, session.target_system,
)
,
():
session = .active_sessions.pop(session_id, )
session:
session = .sessions.get(session_id)
session:
session.state = SessionState.TERMINATED
session.ended_at = datetime.now().isoformat()
._audit(, session.user_id, session.target_system, reason)
():
now = datetime.now()
expired = []
sid, session (.active_sessions.items()):
started = datetime.fromisoformat(session.started_at)
(now - started).total_seconds() > session.max_duration_minutes * :
.terminate_session(sid, )
expired.append(sid)
expired
():
.audit_log.append({
: datetime.now().isoformat(),
: event_type,
: user_id,
: target,
: detail,
})
():
report = []
report.append( * )
report.append()
report.append()
report.append( * )
total = (.sessions)
active = ( s .sessions.values() s.state == SessionState.ACTIVE)
report.append()
report.append()
.active_sessions:
report.append()
sid, s .active_sessions.items():
report.append()
report.append()
report.append()
report.append()
.join(report)
__name__ == :
manager = OTRemoteAccessManager()
manager.define_access_policy(
UserRole.OT_ENGINEER,
[, , , ],
[, ],
max_duration=,
)
manager.define_access_policy(
UserRole.VENDOR,
[],
[],
max_duration=,
)
sid, msg = manager.request_session(
, UserRole.VENDOR,
, , ,
, )
()
sid:
manager.approve_session(sid, )
manager.activate_session(sid, )
(manager.generate_report())
核心概念
| 术语 | 定义 |
|---|
| 中间系统（Intermediate System） | DMZ中终止外部连接并代理到OT的新连接的系统，按CIP-005要求防止直接网络访问 |
| 跳板服务器（Jump Server） | DMZ中用于到OT系统的远程访问会话的加固堡垒主机，具备会话记录和访问控制 |
| 会话记录（Session Recording） | 捕获所有远程访问会话活动（屏幕、键盘操作、命令）用于安全审计和事件调查 |
| 特权访问管理（PAM） | 用于存管凭据、控制访问和审计对关键OT系统特权会话的系统 |
| 陪同（Co-Attendance） | OT操作员实时监控供应商远程访问会话的要求 |
| 时间限制访问（Time-Limited Access） | 供应商账户仅在特定维护窗口期间启用，窗口关闭后自动禁用 |
工具和系统
- CyberArk特权访问安全: 企业PAM，具备OT远程访问的会话管理、凭据存管和录制功能
- BeyondTrust特权远程访问（PRA）: 专为目的设计的远程访问解决方案，具备会话记录和细粒度访问策略
- Claroty安全远程访问（SRA）: OT专用远程访问解决方案，具备协议感知会话控制
- Duo Security: MFA提供商，支持推送通知、硬件令牌和生物识别用于OT访问验证
输出格式
OT远程访问安全报告
===================================
活跃会话: [N]
待审批: [N]
今日会话: [N]
MFA合规性:
所有会话MFA已验证: [是/否]
无MFA的会话: [N]
供应商访问:
活跃供应商会话: [N]
已陪同: [N]
已录制: [N]
