performing-oil-gas-cybersecurity-assessment

name	performing-oil-gas-cybersecurity-assessment
description	本技能涵盖针对石油和天然气设施（包括上游探采、中游管道运输和下游炼化分销）进行网络安全评估。内容涉及控制管道运营的SCADA系统、炼厂过程控制DCS、危险工艺安全仪表系统、无人井口现场RTU，以及对API 1164、TSA管道安全指令、IEC 62443和NIST网络安全框架关键基础设施合规性评估。
domain	cybersecurity
subdomain	ot-ics-security
tags	["ot-security","ics","scada","industrial-control","iec62443","oil-gas","pipeline-security","api1164"]
version	1.0.0
author	mahipal
license	Apache-2.0

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

适用场景

• 对炼厂、管道或生产设施进行网络安全评估时
• 准备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: "安全仪表系统和火气检测"
    
       
       
       
     

  
     
    
       
       
       

  
     
    
       
       
       


  
     
     
     
     
     
  
     
     
     
     
     

步骤 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):
        
        checks = []

        
          architecture_data.get():
            .findings.append(AssessmentFinding(
                finding_id=,
                category=,
                severity=,
                title=,
                description=(
                    
                    
                    
                ),
                affected_systems=[, ],
                regulatory_reference=,
                remediation=,
                timeline=,
            ))
            .finding_counter += 

        
         architecture_data.get()  :
            .findings.append(AssessmentFinding(
                finding_id=,
                category=,
                severity=,
                title=,
                description=(
                    
                    
                    
                    
                ),
                affected_systems=[, ],
                regulatory_reference=,
                remediation=,
                timeline=,
            ))
            .finding_counter += 

        
          architecture_data.get():
            .findings.append(AssessmentFinding(
                finding_id=,
                category=,
                severity=,
                title=,
                description=(
                    
                    
                    
                ),
                affected_systems=[],
                regulatory_reference=,
                remediation=,
                timeline=,
            ))
            .finding_counter += 

     ():
        
          metering_data.get():
            .findings.append(AssessmentFinding(
                finding_id=,
                category=,
                severity=,
                title=,
                description=(
                    
                    
                    
                ),
                affected_systems=[],
                regulatory_reference=,
                remediation=,
                timeline=,
            ))
            .finding_counter += 

     ():
        
        tsa_requirements = [
            ComplianceCheck(, ,
                           , , ),
            ComplianceCheck(, ,
                           , , ,
                           gap=),
            ComplianceCheck(, ,
                           , , ,
                           gap=),
            ComplianceCheck(, ,
                           , , ,
                           gap=),
            ComplianceCheck(, ,
                           , , ,
                           gap=),
            ComplianceCheck(, ,
                           , , ,
                           gap=),
            ComplianceCheck(, ,
                           , , ,
                           gap=),
        ]
        .compliance_checks.extend(tsa_requirements)

     ():
        
        report = []
        report.append( * )
        report.append()
        report.append()
        report.append()
        report.append( * )

        
        report.append()
         sev  [, , , ]:
            count = (  f  .findings  f.severity == sev)
             count:
                report.append()

         f  .findings:
            report.append()
            report.append()
            report.append()
            report.append()
            report.append()

        
        report.append()
        report.append()
        report.append( * )
         c  .compliance_checks:
            icon =   c.status ==     c.status ==   
            report.append()
            report.append()
             c.gap:
                report.append()

         .join(report)


 __name__ == :
    assessment = PipelineSCADAAssessment()

    assessment.assess_network_architecture({
        : ,
        : ,
        : ,
    })

    assessment.assess_custody_transfer({
        : ,
    })

    assessment.check_tsa_compliance()
    (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]%合规