performing-oil-gas-cybersecurity-assessment

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

适用场景

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

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

前置条件

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

工作流

步骤 1：基于设施类型进行范围评估

石油和天然气设施因其运营环节而具有独特特征，这些特征会影响评估方法。

# Oil & Gas Cybersecurity Assessment Scope
facility:
  name: "Gulf Coast Refinery"
  segment: "Downstream"
  capacity: "250,000 barrels per day"
  regulatory: ["TSA SD-02", "API 1164", "IEC 62443", "NIST CSF"]

assessment_areas:
  process_control:
    description: "Refinery DCS and SCADA systems"
    systems:
      - "Honeywell Experion DCS - main process control"
      - "Yokogawa CENTUM VP - hydrocracker unit"
      - "Triconex SIS - emergency shutdown systems"
      - "Allen-Bradley PLCs - utilities and tank farm"
    protocols: ["Modbus/TCP", "OPC UA", "HART", "Foundation Fieldbus"]

  pipeline_scada:
    description: "Pipeline SCADA for crude receipt and product dispatch"
    systems:
      - "ABB RTU560 - pipeline RTUs at pump stations"
      - "GE iFIX SCADA - pipeline control center"
      - "Flow computers - custody transfer metering"
    protocols: ["DNP3", "Modbus RTU over serial", "IEC 60870-5-104"]
    communications: ["Licensed radio", "Leased line", "Satellite (VSAT)"]

  safety_systems:
    description: "Safety Instrumented Systems and fire/gas detection"
    systems:
      - "Schneider Triconex 3008 - process SIS"
      - "Honeywell FSC - fire and gas"
      - "Combustion turbine protection systems"
    criticality: "SIL 2/3 rated - highest priority"

  remote_access:
    description: "Vendor and operator remote access to OT"
    methods:
      - "Citrix-based remote access to SCADA terminals"
      - "VPN to vendor support for DCS maintenance"
      - "Satellite communication to remote pump stations"

  physical_security:
    description: "Physical security integration with cyber"
    systems:
      - "Access control systems (badge readers)"
      - "CCTV with IP network connectivity"
      - "Perimeter intrusion detection"

compliance_mapping:
  tsa_sd_02:
    - "Implement network segmentation between IT and OT"
    - "Develop and maintain a Cybersecurity Implementation Plan (CIP)"
    - "Establish a Cybersecurity Assessment Program"
    - "Report cybersecurity incidents to CISA within 24 hours"
    - "Implement access control measures for critical OT systems"
  api_1164:
    - "Risk-based cybersecurity program for pipeline SCADA"
    - "Asset identification and classification"
    - "Network security and access control"
    - "Personnel security and training"
    - "Incident response and recovery"

步骤 2：评估管道 SCADA 安全

管道 SCADA 系统面临独特挑战，包括通过不可信介质进行的长距离通信、无人值守远程站点，以及交接计量完整性要求。

#!/usr/bin/env python3
"""Pipeline SCADA Security Assessment Tool.

Evaluates security of pipeline SCADA systems against
API 1164 and TSA Pipeline Security Directive requirements.
"""

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:
    """Pipeline SCADA security assessment per API 1164 / TSA SD-02."""

    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):
        """Evaluate pipeline SCADA network architecture."""
        checks = []

        # TSA SD-02: Network segmentation between IT and OT
        if not architecture_data.get("it_ot_segmentation"):
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="Network Architecture",
                severity="critical",
                title="No IT/OT Network Segmentation",
                description=(
                    "Pipeline SCADA network is not segmented from corporate IT. "
                    "An attacker compromising the corporate network could pivot "
                    "directly to pipeline control systems."
                ),
                affected_systems=["Pipeline SCADA servers", "RTU communications"],
                regulatory_reference="TSA SD-02 Section 2.1; API 1164 Section 7",
                remediation="Deploy DMZ with industrial firewall between IT and pipeline SCADA",
                timeline="30 days",
            ))
            self.finding_counter += 1

        # Check for encrypted RTU communications
        if architecture_data.get("rtu_comm_encrypted") is False:
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="Communication Security",
                severity="high",
                title="Unencrypted Pipeline RTU Communications",
                description=(
                    "DNP3 communications between control center and remote RTUs "
                    "traverse radio/satellite links without encryption. An attacker "
                    "with radio access could intercept or inject SCADA commands."
                ),
                affected_systems=["Pipeline RTUs", "SCADA master station"],
                regulatory_reference="API 1164 Section 7.3; IEC 62351",
                remediation="Deploy DNP3 Secure Authentication or VPN tunnel for RTU links",
                timeline="90 days",
            ))
            self.finding_counter += 1

        # Check remote pump station physical security
        if not architecture_data.get("remote_site_intrusion_detection"):
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="Physical-Cyber Convergence",
                severity="high",
                title="Remote Pump Stations Lack Physical Intrusion Detection",
                description=(
                    "Unmanned pump stations along the pipeline corridor lack physical "
                    "intrusion detection systems. An attacker could gain physical access "
                    "to RTUs and SCADA communication equipment without detection."
                ),
                affected_systems=["Remote pump station RTUs and networking equipment"],
                regulatory_reference="TSA SD-02 Section 2.3; API 1164 Section 10",
                remediation="Install intrusion detection with cellular alerting at remote sites",
                timeline="60 days",
            ))
            self.finding_counter += 1

    def assess_custody_transfer(self, metering_data):
        """Assess security of custody transfer metering systems."""
        if not metering_data.get("flow_computer_auth"):
            self.findings.append(AssessmentFinding(
                finding_id=f"OG-{self.finding_counter:03d}",
                category="Custody Transfer Integrity",
                severity="critical",
                title="Flow Computer Lacks Authentication",
                description=(
                    "Custody transfer flow computers accept unauthenticated Modbus "
                    "commands allowing meter factor and flow calculation parameters "
                    "to be modified. This could enable financial fraud through "
                    "manipulation of custody transfer measurements."
                ),
                affected_systems=["Flow computers at custody transfer points"],
                regulatory_reference="API 1164 Section 8; API MPMS Chapter 21",
                remediation="Implement authenticated access to flow computers; deploy audit logging",
                timeline="45 days",
            ))
            self.finding_counter += 1

    def check_tsa_compliance(self):
        """Evaluate compliance with TSA Pipeline Security Directives."""
        tsa_requirements = [
            ComplianceCheck("TSA-01", "Cybersecurity Coordinator designated",
                           "TSA SD-01", "compliant", "CySec coordinator appointed"),
            ComplianceCheck("TSA-02", "Incident reporting to CISA within 24 hours",
                           "TSA SD-01", "partial", "Process exists but not tested",
                           gap="Need tabletop exercise to validate reporting timeline"),
            ComplianceCheck("TSA-03", "Cybersecurity Implementation Plan",
                           "TSA SD-02", "non-compliant", "No formal CIP exists",
                           gap="Develop and submit CIP to TSA within 90 days"),
            ComplianceCheck("TSA-04", "Network segmentation between IT and OT",
                           "TSA SD-02", "non-compliant", "Flat network observed",
                           gap="Implement DMZ and zone-based segmentation"),
            ComplianceCheck("TSA-05", "Access control for critical OT systems",
                           "TSA SD-02", "partial", "Shared accounts on SCADA",
                           gap="Implement individual accounts with role-based access"),
            ComplianceCheck("TSA-06", "Continuous monitoring and detection",
                           "TSA SD-02", "non-compliant", "No OT IDS deployed",
                           gap="Deploy OT intrusion detection (Dragos/Nozomi/Claroty)"),
            ComplianceCheck("TSA-07", "Patch management for critical systems",
                           "TSA SD-02", "partial", "Ad hoc patching only",
                           gap="Establish formal OT patch management program"),
        ]
        self.compliance_checks.extend(tsa_requirements)

    def generate_report(self):
        """Generate comprehensive assessment report."""
        report = []
        report.append("=" * 70)
        report.append(f"OIL & GAS CYBERSECURITY ASSESSMENT REPORT")
        report.append(f"Facility: {self.facility}")
        report.append(f"Date: {datetime.now().strftime('%Y-%m-%d')}")
        report.append("=" * 70)

        # Findings summary
        report.append(f"\nFINDINGS: {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"    Category: {f.category}")
            report.append(f"    {f.description}")
            report.append(f"    Regulatory: {f.regulatory_reference}")
            report.append(f"    Remediation: {f.remediation} ({f.timeline})")

        # Compliance summary
        report.append(f"\n{'='*70}")
        report.append("TSA PIPELINE SECURITY DIRECTIVE COMPLIANCE")
        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"      Status: {c.status.upper()}")
            if c.gap:
                report.append(f"      Gap: {c.gap}")

        return "\n".join(report)


if __name__ == "__main__":
    assessment = PipelineSCADAAssessment("Gulf Coast Refinery")

    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（实施）
交接计量	石油产品在各方之间的所有权转移，要求计量系统完整性以防止财务欺诈
DCS	集散控制系统，用于炼油厂连续过程控制，配备冗余控制器和操作员站
远程终端单元（RTU）	远程管道现场设备，采集传感器数据并执行控制命令，通过无线电/卫星通信
安全完整性等级（SIL）	IEC 61511 对安全仪表功能的评级，SIL 1-4 定义了要求时的失效概率
HAZOP	危险与可操作性研究，识别工艺设计中的潜在危险；网络安全应与 HAZOP 结果相结合

工具与系统

• Dragos Platform：OT 网络安全平台，针对石油和天然气威胁组织（XENOTIME、KAMACITE、ERYTHRITE）提供特定检测
• Claroty xDome：面向石油和天然气 OT 环境的全面资产发现与漏洞管理
• Nozomi Guardian：网络监控，支持管道协议（DNP3、Modbus、IEC 60870-5-104）
• Honeywell Forge Cybersecurity：专为炼油厂常见的 Honeywell DCS 环境设计的 OT 安全平台

输出格式

Oil & Gas Cybersecurity Assessment Report
==========================================
Facility: [Name]
Segment: Upstream / Midstream / Downstream
Date: YYYY-MM-DD
Standards: API 1164, TSA SD-02, IEC 62443

FINDINGS:
  Critical: [N]  High: [N]  Medium: [N]  Low: [N]

COMPLIANCE STATUS:
  TSA SD-02: [N]% compliant
  API 1164: [N]% compliant
  IEC 62443: [N]% compliant