performing-oil-gas-cybersecurity-assessment

执行石油和天然气网络安全评估

适用场景

• 对炼厂、管道或生产设施进行网络安全评估时
• 准备TSA管道安全指令合规（SD-01、SD-02）时
• 根据API标准1164（管道SCADA安全）评估网络安全态势时
• 评估远程井口SCADA系统和卫星通信安全性时
• 并购、收购或监管审计需要全面OT安全评估时

不适用于石油和天然气公司仅IT企业网络评估、无网络组件的物理安全评估，或环境合规评估。

前置条件

• 设施管理层和运营团队的书面授权
• 了解石油和天然气运营（上游、中游、下游工艺）
• 熟悉API 1164、TSA SD-01/SD-02、IEC 62443和NIST CSF
• 用于OT网络流量捕获的被动监控工具
• 可访问网络图、SCADA架构文档和安全研究（HAZOP）

工作流程

步骤 1：基于设施类型界定评估范围

石油和天然气设施因运营细分而具有独特特征，影响评估方法。

# 石油和天然气网络安全评估范围
facility:
  name: "墨西哥湾沿岸炼厂"
  segment: "下游"
  capacity: "每日25万桶"
  regulatory: ["TSA SD-02", "API 1164", "IEC 62443", "NIST CSF"]

assessment_areas:
  process_control:
    description: "炼厂DCS和SCADA系统"
    systems:
      - "Honeywell Experion DCS - 主过程控制"
      - "Yokogawa CENTUM VP - 加氢裂化装置"
      - "Triconex SIS - 紧急停车系统"
      - "Allen-Bradley PLC - 公用工程和储罐区"
    protocols: ["Modbus/TCP", "OPC UA", "HART", "Foundation Fieldbus"]

  pipeline_scada:
    description: "原油接收和产品分发的管道SCADA"
    systems:
      - "ABB RTU560 - 泵站管道RTU"
      - "GE iFIX SCADA - 管道控制中心"
      - "流量计算机 - 交接计量"
    protocols: ["DNP3", "串口Modbus RTU", "IEC 60870-5-104"]
    communications: ["授权无线电", "租用线路", "卫星（VSAT）"]

  safety_systems:
    description: "安全仪表系统和火气检测"
    systems:
      - "Schneider Triconex 3008 - 过程SIS"
      - "Honeywell FSC - 火气系统"
      - "燃气轮机保护系统"
    criticality: "SIL 2/3等级 - 最高优先级"

  remote_access:
    description: "供应商和操作员对OT的远程访问"
    methods:
      - "基于Citrix的SCADA终端远程访问"
      - "VPN供应商支持DCS维护"
      - "远程泵站卫星通信"

  physical_security:
    description: "物理安全与网络安全集成"
    systems:
      - "门禁控制系统（门禁读卡器）"
      - "具有IP网络连接的CCTV"
      - "周界入侵检测"

compliance_mapping:
  tsa_sd_02:
    - "实施IT和OT之间的网络隔离"
    - "制定并维护网络安全实施计划（CIP）"
    - "建立网络安全评估项目"
    - "24小时内向CISA报告网络安全事件"
    - "为关键OT系统实施访问控制措施"
  api_1164:
    - "基于风险的管道SCADA网络安全计划"
    - "资产识别和分类"
    - "网络安全和访问控制"
    - "人员安全和培训"
    - "事件响应和恢复"

步骤 2：评估管道SCADA安全性

管道SCADA系统面临独特挑战，包括通过不受信媒介的长距离通信、无人远程站点和交接计量完整性要求。

#!/usr/bin/env python3
"""管道SCADA安全评估工具。

根据API 1164和TSA管道安全指令要求
评估管道SCADA系统的安全性。
"""

import json
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime


@dataclass
class AssessmentFinding:
    finding_id: str
    category: str
    severity: str
    title: str
    description: str
    affected_systems: list
    regulatory_reference: str
    remediation: str
    timeline: str


@dataclass
class ComplianceCheck:
    requirement_id: str
    description: str
    standard: str
    status: str  # compliant, partial, non-compliant
    evidence: str
    gap: str = ""


class PipelineSCADAAssessment:
    """按API 1164 / TSA SD-02进行管道SCADA安全评估。"""

    def __init__(self, facility_name):
        self.facility = facility_name
        self.findings = []
        self.compliance_checks = []
        self.finding_counter = 1

    def assess_network_architecture(self, architecture_data):
        """评估管道SCADA网络架构。"""
        checks = []

        # TSA SD-02: IT和OT之间的网络隔离
        if not architecture_data.get("it_ot_segmentation"):
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="网络架构",
                severity="critical",
                title="无IT/OT网络隔离",
                description=(
                    "管道SCADA网络未与企业IT隔离。"
                    "攻击者一旦入侵企业网络，可直接"
                    "横向移动到管道控制系统。"
                ),
                affected_systems=["管道SCADA服务器", "RTU通信"],
                regulatory_reference="TSA SD-02第2.1节; API 1164第7节",
                remediation="在IT和管道SCADA之间部署带工业防火墙的DMZ",
                timeline="30天",
            ))
            self.finding_counter += 1

        # 检查RTU通信加密
        if architecture_data.get("rtu_comm_encrypted") is False:
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="通信安全",
                severity="high",
                title="管道RTU通信未加密",
                description=(
                    "控制中心与远程RTU之间的DNP3通信"
                    "通过无线电/卫星链路传输，未加密。"
                    "具有无线电访问权限的攻击者可拦截或"
                    "注入SCADA命令。"
                ),
                affected_systems=["管道RTU", "SCADA主站"],
                regulatory_reference="API 1164第7.3节; IEC 62351",
                remediation="为RTU链路部署DNP3安全认证或VPN隧道",
                timeline="90天",
            ))
            self.finding_counter += 1

        # 检查远程泵站物理安全
        if not architecture_data.get("remote_site_intrusion_detection"):
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="物理-网络融合",
                severity="high",
                title="远程泵站缺乏物理入侵检测",
                description=(
                    "管道沿线无人泵站缺乏物理入侵检测系统。"
                    "攻击者可在不被察觉的情况下物理访问"
                    "RTU和SCADA通信设备。"
                ),
                affected_systems=["远程泵站RTU和网络设备"],
                regulatory_reference="TSA SD-02第2.3节; API 1164第10节",
                remediation="在远程站点安装带蜂窝告警的入侵检测系统",
                timeline="60天",
            ))
            self.finding_counter += 1

    def assess_custody_transfer(self, metering_data):
        """评估交接计量系统的安全性。"""
        if not metering_data.get("flow_computer_auth"):
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="交接计量完整性",
                severity="critical",
                title="流量计算机缺乏认证",
                description=(
                    "交接计量流量计算机接受未经认证的Modbus命令，"
                    "允许修改计量系数和流量计算参数。"
                    "这可能通过操纵交接计量数据实现财务欺诈。"
                ),
                affected_systems=["交接计量点的流量计算机"],
                regulatory_reference="API 1164第8节; API MPMS第21章",
                remediation="实施流量计算机认证访问；部署审计日志",
                timeline="45天",
            ))
            self.finding_counter += 1

    def check_tsa_compliance(self):
        """评估TSA管道安全指令的合规性。"""
        tsa_requirements = [
            ComplianceCheck("TSA-01", "已指定网络安全协调员",
                           "TSA SD-01", "compliant", "已任命网络安全协调员"),
            ComplianceCheck("TSA-02", "24小时内向CISA报告事件",
                           "TSA SD-01", "partial", "流程存在但未经测试",
                           gap="需要桌面演练以验证报告时限"),
            ComplianceCheck("TSA-03", "网络安全实施计划",
                           "TSA SD-02", "non-compliant", "无正式CIP",
                           gap="在90天内制定并提交CIP给TSA"),
            ComplianceCheck("TSA-04", "IT和OT之间的网络隔离",
                           "TSA SD-02", "non-compliant", "观察到扁平网络",
                           gap="实施DMZ和基于区域的隔离"),
            ComplianceCheck("TSA-05", "关键OT系统的访问控制",
                           "TSA SD-02", "partial", "SCADA上使用共享账户",
                           gap="实施具有基于角色访问的个人账户"),
            ComplianceCheck("TSA-06", "持续监控和检测",
                           "TSA SD-02", "non-compliant", "未部署OT IDS",
                           gap="部署OT入侵检测（Dragos/Nozomi/Claroty）"),
            ComplianceCheck("TSA-07", "关键系统的补丁管理",
                           "TSA SD-02", "partial", "仅有临时补丁",
                           gap="建立正式的OT补丁管理程序"),
        ]
        self.compliance_checks.extend(tsa_requirements)

    def generate_report(self):
        """生成综合评估报告。"""
        report = []
        report.append("=" * 70)
        report.append(f"石油和天然气网络安全评估报告")
        report.append(f"设施: {self.facility}")
        report.append(f"日期: {datetime.now().strftime('%Y-%m-%d')}")
        report.append("=" * 70)

        # 发现摘要
        report.append(f"\n发现: {len(self.findings)}")
        for sev in ["critical", "high", "medium", "low"]:
            count = sum(1 for f in self.findings if f.severity == sev)
            if count:
                report.append(f"  {sev.upper()}: {count}")

        for f in self.findings:
            report.append(f"\n  [{f.finding_id}] [{f.severity.upper()}] {f.title}")
            report.append(f"    类别: {f.category}")
            report.append(f"    {f.description}")
            report.append(f"    法规: {f.regulatory_reference}")
            report.append(f"    修复措施: {f.remediation}（{f.timeline}）")

        # 合规摘要
        report.append(f"\n{'='*70}")
        report.append("TSA管道安全指令合规性")
        report.append("=" * 70)
        for c in self.compliance_checks:
            icon = "+" if c.status == "compliant" else "~" if c.status == "partial" else "-"
            report.append(f"  [{icon}] {c.requirement_id}: {c.description}")
            report.append(f"      状态: {c.status.upper()}")
            if c.gap:
                report.append(f"      差距: {c.gap}")

        return "\n".join(report)


if __name__ == "__main__":
    assessment = PipelineSCADAAssessment("墨西哥湾沿岸炼厂")

    assessment.assess_network_architecture({
        "it_ot_segmentation": False,
        "rtu_comm_encrypted": False,
        "remote_site_intrusion_detection": False,
    })

    assessment.assess_custody_transfer({
        "flow_computer_auth": False,
    })

    assessment.check_tsa_compliance()
    print(assessment.generate_report())

核心概念

术语	定义
API 1164	美国石油学会管道SCADA安全标准，为管道控制系统网络安全提供基于风险的框架
TSA管道安全指令	TSA针对管道运营商发布的强制性网络安全要求，包括SD-01（报告）和SD-02（实施）
交接计量（Custody Transfer）	石油产品所有权的交接转让，需要计量系统完整性以防止财务欺诈
分布式控制系统（DCS）	炼厂用于连续过程控制的分布式控制系统，具有冗余控制器和操作站
远程终端单元（RTU）	远程管道站点的现场设备，通过无线电/卫星通信采集传感器数据并执行控制命令
安全完整性等级（SIL）	IEC 61511对安全仪表功能的评级，SIL 1-4定义了按需失效概率
HAZOP	危险与可操作性研究，识别工艺设计中的潜在危险；网络安全应与HAZOP结果集成

工具和系统

• Dragos平台: 专注于石油和天然气威胁组织（XENOTIME、KAMACITE、ERYTHRITE）检测的OT网络安全平台
• Claroty xDome: 石油和天然气OT环境的综合资产发现和漏洞管理
• Nozomi Guardian: 支持管道协议（DNP3、Modbus、IEC 60870-5-104）的网络监控
• Honeywell Forge Cybersecurity: 专为炼厂常用Honeywell DCS环境设计的OT安全平台

输出格式

石油和天然气网络安全评估报告
==========================================
设施: [名称]
细分市场: 上游/中游/下游
日期: YYYY-MM-DD
标准: API 1164, TSA SD-02, IEC 62443

发现:
  严重: [N]  高: [N]  中: [N]  低: [N]

合规状态:
  TSA SD-02: [N]%合规
  API 1164: [N]%合规
  IEC 62443: [N]%合规