工控系统(ICS/SCADA)取证深度分析

1. 区域划分确认 → 识别 Zones 与 Conduits 边界
2. 安全等级评估 → 确定各区域的 SL 等级（SL1-SL4）
3. 资产清单建立 → 记录所有工控设备的固件版本、配置基线
4. 被动数据采集 → 网络流量镜像、日志远程转发
5. 异常检测与分析 → 协议异常、配置偏差、行为基线偏离
6. 证据保全 → 哈希校验、时间戳同步、证据链完整性
7. 报告与修复建议 → 符合行业法规要求的取证报告

┌─────────────────────────────────────────────────────────┐
│                   取证前风险评估清单                       │
├─────────────────────────────────────────────────────────┤
│ □ 确认目标系统的安全等级（SIL/SL）                        │
│ □ 获取系统所有者的书面授权                                  │
│ □ 确认取证窗口（与生产团队协商）                            │
│ □ 评估取证操作对物理过程的潜在影响                          │
│ □ 准备回退方案（Rollback Plan）                            │
│ □ 确认通信路径不会干扰安全关键控制回路                      │
│ □ 验证取证工具不会向控制器发送写命令                        │
│ □ 确认取证人员具备工控安全资质                              │
│ □ 建立取证期间的应急通信渠道                               │
│ □ 记录所有取证操作的开始/结束时间                           │
└─────────────────────────────────────────────────────────┘

读取输入 → 执行程序 → 处理通信 → 执行诊断 → 写输出
  ↑                                              │
  └──────────────────────────────────────────────┘

# 使用 snap7 库（Python）连接 Siemens S7 PLC 并读取程序块信息
# 注意：仅执行只读操作，不发送任何写命令

import snap7
from snap7.util import get_int, get_real

plc = snap7.client.Client()
plc.connect('192.168.1.10', 0, 1)

cpu_info = plc.get_cpu_info()
print(f"模块类型: {cpu_info.ModuleTypeName.decode()}")
print(f"固件版本: {cpu_info.AS_Name.decode()}")
print(f"序列号: {cpu_info.AS_Index}")

block_list = plc.list_blocks()
for block_type, count in block_list.items():
    print(f"块类型 {block_type}: {count} 个块")

plc.disconnect()

# 使用 nmap 进行 OT 安全的 PLC 发现（仅使用安全脚本）
nmap -sS -p 102,502,44818,4001 \
  --script s7-info,modbus-info,ethernetip-info \
  -n --disable-arp-ping \
  192.168.1.0/24

# 输出示例：
# PORT    STATE SERVICE
# 102/tcp open  siemens-s7
# | s7-info:
# |   Module: CPU 1515-2 PN
# |   Firmware: V2.9
# |   Module Type: S7-1500
# |   Serial Number: S-C7MJ01234567
# |   Memory Size: 1048576 bytes

# 计算项目文件哈希，建立基线
find /ews/projects/ -name "*.ap17" -o -name "*.zap17" | \
  while read f; do
    sha256sum "$f" >> /evidence/project_baselines.sha256
  done

# 对比当前 PLC 上传的项目与基线
diff -rq /ews/projects/current_backup/ /ews/projects/baseline/ \
  --exclude="*.bak" --exclude="*.log"

import hashlib
import json

def compute_program_hash(plc_program_bytes):
    return hashlib.sha256(plc_program_bytes).hexdigest()

def compare_program_hashes(baseline_hash, current_hash):
    if baseline_hash != current_hash:
        return {
            "status": "TAMPERED",
            "confidence": "HIGH",
            "evidence_strength": "CONFIRMED_MALICIOUS"
                if baseline_hash
                else "REQUIRES_INVESTIGATION"
        }
    return {"status": "INTACT", "confidence": "HIGH"}

def analyze_ladder_diff(baseline_rungs, current_rungs):
    modifications = []
    for idx, (base_rung, curr_rung) in enumerate(
        zip(baseline_rungs, current_rungs)
    ):
        if base_rung.hash != curr_rung.hash:
            modifications.append({
                "rung_index": idx,
                "change_type": "MODIFIED",
                "baseline_contacts": base_rung.contacts,
                "current_contacts": curr_rung.contacts,
                "new_coils": [c for c in curr_rung.coils
                              if c not in base_rung.coils],
                "removed_coils": [c for c in base_rung.coils
                                  if c not in curr_rung.coils]
            })

    if len(current_rungs) > len(baseline_rungs):
        for idx in range(len(baseline_rungs), len(current_rungs)):
            modifications.append({
                "rung_index": idx,
                "change_type": "INSERTED",
                "content": current_rungs[idx].raw_data
            })

    return modifications

def detect_setpoint_anomalies(baseline_tags, current_tags, threshold_pct=10):
    anomalies = []
    for tag_name in baseline_tags:
        if tag_name not in current_tags:
            anomalies.append({
                "tag": tag_name,
                "anomaly_type": "TAG_MISSING",
                "severity": "HIGH"
            })
            continue

        base_val = baseline_tags[tag_name].value
        curr_val = current_tags[tag_name].value

        if base_val != 0:
            deviation = abs(curr_val - base_val) / abs(base_val) * 100
            if deviation > threshold_pct:
                anomalies.append({
                    "tag": tag_name,
                    "baseline_value": base_val,
                    "current_value": curr_val,
                    "deviation_pct": round(deviation, 2),
                    "anomaly_type": "SETPOINT_DEVIATION",
                    "severity": "CRITICAL" if deviation > 50 else "HIGH"
                })

    for tag_name in current_tags:
        if tag_name not in baseline_tags:
            anomalies.append({
                "tag": tag_name,
                "anomaly_type": "NEW_TAG",
                "severity": "MEDIUM",
                "value": current_tags[tag_name].value
            })

    return anomalies

def detect_hidden_logic(plc_program):
    findings = []

    for db in plc_program.data_blocks:
        if db.size > db.expected_size:
            findings.append({
                "type": "DB_SIZE_ANOMALY",
                "block": db.number,
                "expected_size": db.expected_size,
                "actual_size": db.size,
                "excess_bytes": db.size - db.expected_size
            })

    for ob in plc_program.organization_blocks:
        called_blocks = set(ob.call_graph.all_called())
        declared_blocks = set(ob.declared_blocks)
        undeclared = called_blocks - declared_blocks
        if undeclared:
            findings.append({
                "type": "UNDECLARED_BLOCK_CALL",
                "ob_number": ob.number,
                "undeclared_blocks": list(undeclared)
            })

    return findings

# Siemens S7 PLC 事件日志提取（使用 s7client 工具）
s7client --ip 192.168.1.10 --rack 0 --slot 1 \
  read-event-log --format csv --output /evidence/plc_events.csv

# 分析 STOP/RUN 切换事件
grep -E "STOP|RUN" /evidence/plc_events.csv | \
  awk -F',' '{print $1, $3, $4}' | sort -k1

Level 0: 无保护（可读写所有块）
Level 1: 密码保护（知道密码可读写）
Level 2: 密码保护 + 读取保护
Level 3: 密码保护 + 已知/未知密码（最高保护）
Level 4: 安全等级 4（S7-1500，支持 PROFINET 安全）

# 使用 pycomm3 库连接 ControlLogix
from pycomm3 import LogixDriver

with LogixDriver('192.168.1.20') as plc:
    print(f"设备: {plc.info['device_type']}")
    print(f"固件: {plc.info['firmware']}")
    print(f"序列号: {plc.info['serial']}")

    tags = plc.get_tag_list()
    for tag in tags:
        print(f"Tag: {tag['tag_name']}, "
              f"Type: {tag['data_type']}, "
              f"Value: {tag.get('value', 'N/A')}")

    controller_tags = plc.read('Program:MainProgram.AllTags')

现场传感器 → PLC/RTU → SCADA Server → HMI → 操作员
                ↑                              │
                └──── 操作员操作指令 ────────────┘

# HMI 面板 Windows CE 系统镜像提取
# 通过 CF 卡或 SD 卡物理提取
dd if=/dev/sdc of=/evidence/hmi_cf_card.img bs=4M conv=noerror,sync
sha256sum /evidence/hmi_cf_card.img > /evidence/hmi_cf_card.img.sha256

# 挂载分析（只读）
mkdir -p /mnt/hmi_evidence
mount -o ro,loop /evidence/hmi_cf_card.img /mnt/hmi_evidence

# 提取关键日志文件
find /mnt/hmi_evidence -name "*.log" -o -name "*.evt" -o -name "*.csv" | \
  while read f; do
    cp --preserve=timestamps "$f" /evidence/hmi_logs/
  done

import os
import hashlib
import json

def audit_hmi_screens(project_dir, baseline_dir):
    tampered_screens = []

    for root, dirs, files in os.walk(project_dir):
        for fname in files:
            if fname.endswith(('.pdl', '.pdlx', '.hmi', '.fpt')):
                filepath = os.path.join(root, fname)
                rel_path = os.path.relpath(filepath, project_dir)
                baseline_path = os.path.join(baseline_dir, rel_path)

                with open(filepath, 'rb') as f:
                    current_hash = hashlib.sha256(f.read()).hexdigest()

                if os.path.exists(baseline_path):
                    with open(baseline_path, 'rb') as f:
                        baseline_hash = hashlib.sha256(
                            f.read()
                        ).hexdigest()

                    if current_hash != baseline_hash:
                        tampered_screens.append({
                            "screen": rel_path,
                            "baseline_hash": baseline_hash,
                            "current_hash": current_hash,
                            "change_type": "MODIFIED"
                        })
                else:
                    tampered_screens.append({
                        "screen": rel_path,
                        "current_hash": current_hash,
                        "change_type": "NEW_SCREEN"
                    })

    return tampered_screens

def detect_alarm_config_anomalies(alarm_config, baseline_config):
    anomalies = []

    for alarm_id in baseline_config:
        if alarm_id not in alarm_config:
            anomalies.append({
                "alarm_id": alarm_id,
                "issue": "ALARM_REMOVED",
                "severity": "CRITICAL",
                "description": "报警已被删除，可能导致操作员无法收到关键告警"
            })
            continue

        base = baseline_config[alarm_id]
        curr = alarm_config[alarm_id]

        if curr.get('priority') != base.get('priority'):
            anomalies.append({
                "alarm_id": alarm_id,
                "issue": "PRIORITY_CHANGED",
                "baseline_priority": base.get('priority'),
                "current_priority": curr.get('priority'),
                "severity": "HIGH"
            })

        if curr.get('threshold') != base.get('threshold'):
            anomalies.append({
                "alarm_id": alarm_id,
                "issue": "THRESHOLD_CHANGED",
                "baseline_threshold": base.get('threshold'),
                "current_threshold": curr.get('threshold'),
                "severity": "CRITICAL"
            })

        if curr.get('suppressed') and not base.get('suppressed'):
            anomalies.append({
                "alarm_id": alarm_id,
                "issue": "ALARM_SUPPRESSED",
                "severity": "CRITICAL",
                "description": "报警被抑制，操作员将无法收到此告警"
            })

    return anomalies

import re

def scan_hmi_scripts(script_dir):
    suspicious_patterns = [
        (r'shell\s*\(', "Shell 命令执行"),
        (r'WScript\.Shell', "WScript 对象实例化"),
        (r'CreateObject\s*\(', "COM 对象创建"),
        (r'InternetExplorer\.Application', "浏览器对象（可能用于数据外传）"),
        (r'ADODB\.Stream', "文件流操作"),
        (r'SaveToFile', "文件写入操作"),
        (r'WinHttp|XMLHTTP|ServerXMLHTTP', "HTTP 请求（数据外传风险）"),
        (r'FileSystemObject', "文件系统操作"),
        (r'RegWrite|RegRead', "注册表操作"),
        (r'socket|connect|send|recv', "网络通信"),
        (r'powershell|cmd\.exe|regsvr32', "命令执行"),
        (r'base64|decode|decode64', "编码/解码操作"),
    ]

    findings = []
    for root, _, files in os.walk(script_dir):
        for fname in files:
            if fname.endswith(('.vbs', '.bs', '.c', '.py', '.lua', '.macro')):
                filepath = os.path.join(root, fname)
                with open(filepath, 'r', errors='ignore') as f:
                    content = f.read()
                    for pattern, desc in suspicious_patterns:
                        matches = re.findall(pattern, content, re.IGNORECASE)
                        if matches:
                            findings.append({
                                "file": filepath,
                                "pattern": pattern,
                                "description": desc,
                                "match_count": len(matches),
                                "severity": "HIGH"
                            })

    return findings

-- 检测异常操作模式（非工作时间的写操作）
SELECT operator_id, tag_name, old_value, new_value, timestamp
FROM operation_log
WHERE HOUR(timestamp) NOT BETWEEN 8 AND 18
   OR DAYOFWEEK(timestamp) IN (1, 7)
ORDER BY timestamp DESC
LIMIT 100;

-- 检测短时间内大量写操作（可能的批量篡改）
SELECT operator_id, COUNT(*) as write_count,
       MIN(timestamp) as first_write,
       MAX(timestamp) as last_write
FROM operation_log
WHERE action_type = 'WRITE'
GROUP BY operator_id, DATE(timestamp)
HAVING write_count > 50
ORDER BY write_count DESC;

-- 检测关键安全标签的修改
SELECT * FROM operation_log
WHERE tag_name IN (
    SELECT tag_name FROM critical_tags
    WHERE safety_related = TRUE
)
ORDER BY timestamp DESC;

def detect_tag_anomalies(current_tags, baseline_tags):
    anomalies = []

    for tag in current_tags:
        if tag.name not in {t.name for t in baseline_tags}:
            anomalies.append({
                "tag": tag.name,
                "issue": "NEW_TAG",
                "value": tag.value,
                "engineering_units": tag.units,
                "severity": "MEDIUM"
            })

    for tag in baseline_tags:
        if tag.name not in {t.name for t in current_tags}:
            anomalies.append({
                "tag": tag.name,
                "issue": "TAG_REMOVED",
                "severity": "HIGH"
            })

    baseline_map = {t.name: t for t in baseline_tags}
    for tag in current_tags:
        if tag.name in baseline_map:
            base = baseline_map[tag.name]
            if tag.scale_low != base.scale_low or tag.scale_high != base.scale_high:
                anomalies.append({
                    "tag": tag.name,
                    "issue": "SCALING_CHANGED",
                    "baseline_low": base.scale_low,
                    "baseline_high": base.scale_high,
                    "current_low": tag.scale_low,
                    "current_high": tag.scale_high,
                    "severity": "CRITICAL"
                })

    return anomalies

import hashlib
from datetime import datetime, timedelta

def verify_historian_integrity(data_points, expected_interval_sec=60):
    gaps = []
    duplicates = []
    out_of_order = []

    sorted_points = sorted(data_points, key=lambda x: x.timestamp)

    for i in range(1, len(sorted_points)):
        prev = sorted_points[i - 1]
        curr = sorted_points[i]

        delta = (curr.timestamp - prev.timestamp).total_seconds()

        if delta > expected_interval_sec * 2:
            gaps.append({
                "gap_start": prev.timestamp.isoformat(),
                "gap_end": curr.timestamp.isoformat(),
                "gap_duration_sec": delta,
                "severity": "HIGH" if delta > 3600 else "MEDIUM"
            })

        if delta == 0 and prev.value == curr.value:
            duplicates.append({
                "timestamp": curr.timestamp.isoformat(),
                "value": curr.value,
                "tag": curr.tag_name
            })

        if curr.timestamp < prev.timestamp:
            out_of_order.append({
                "prev_ts": prev.timestamp.isoformat(),
                "curr_ts": curr.timestamp.isoformat(),
                "tag": curr.tag_name
            })

    return {
        "total_points": len(sorted_points),
        "gaps": gaps,
        "duplicates": duplicates,
        "out_of_order": out_of_order,
        "integrity_score": 1.0 - (
            len(gaps) + len(duplicates) + len(out_of_order)
        ) / len(sorted_points)
    }

def analyze_timestamp_consistency(historian_data, plc_event_log):
    discrepancies = []

    hist_timestamps = {p.tag_name: p.timestamp for p in historian_data}
    plc_timestamps = {e.tag_name: e.timestamp for e in plc_event_log}

    for tag in set(hist_timestamps.keys()) & set(plc_timestamps.keys()):
        delta = abs(
            (hist_timestamps[tag] - plc_timestamps[tag]).total_seconds()
        )
        if delta > 5:
            discrepancies.append({
                "tag": tag,
                "historian_ts": hist_timestamps[tag].isoformat(),
                "plc_ts": plc_timestamps[tag].isoformat(),
                "delta_sec": delta,
                "possible_cause": "CLOCK_SKEW" if delta < 60
                    else "POTENTIAL_TAMPERING"
            })

    return discrepancies

#!/usr/bin/env python3
import os
import sys
import hashlib
import json
import csv
from datetime import datetime

def full_hmi_audit(hmi_project_dir, baseline_dir, output_dir):
    report = {
        "audit_time": datetime.now().isoformat(),
        "project_dir": hmi_project_dir,
        "baseline_dir": baseline_dir,
        "findings": []
    }

    for root, dirs, files in os.walk(hmi_project_dir):
        for fname in files:
            filepath = os.path.join(root, fname)
            rel_path = os.path.relpath(filepath, hmi_project_dir)
            baseline_path = os.path.join(baseline_dir, rel_path)

            with open(filepath, 'rb') as f:
                current_hash = hashlib.sha256(f.read()).hexdigest()

            if os.path.exists(baseline_path):
                with open(baseline_path, 'rb') as f:
                    baseline_hash = hashlib.sha256(f.read()).hexdigest()

                if current_hash != baseline_hash:
                    report["findings"].append({
                        "file": rel_path,
                        "status": "MODIFIED",
                        "baseline_hash": baseline_hash,
                        "current_hash": current_hash,
                        "severity": "HIGH"
                    })
            else:
                report["findings"].append({
                    "file": rel_path,
                    "status": "NEW_FILE",
                    "current_hash": current_hash,
                    "severity": "MEDIUM"
                })

    for baseline_root, _, baseline_files in os.walk(baseline_dir):
        for fname in baseline_files:
            baseline_path = os.path.join(baseline_root, fname)
            rel_path = os.path.relpath(baseline_path, baseline_dir)
            current_path = os.path.join(hmi_project_dir, rel_path)

            if not os.path.exists(current_path):
                report["findings"].append({
                    "file": rel_path,
                    "status": "DELETED",
                    "severity": "CRITICAL"
                })

    os.makedirs(output_dir, exist_ok=True)
    report_path = os.path.join(output_dir, "hmi_audit_report.json")
    with open(report_path, 'w') as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    return report

if __name__ == "__main__":
    if len(sys.argv) != 4:
        print(f"Usage: {sys.argv[0]} <hmi_project> <baseline> <output>")
        sys.exit(1)

    result = full_hmi_audit(sys.argv[1], sys.argv[2], sys.argv[3])
    print(f"审计完成，发现 {len(result['findings'])} 个异常项")
    for finding in result["findings"]:
        print(f"  [{finding['severity']}] {finding['file']}: "
              f"{finding['status']}")

+-------------------+-------------------+-------------------+
| MBAP Header       | PDU               | Data              |
| (7 bytes)         |                   |                   |
+-------------------+-------------------+-------------------+
| Transaction ID    | Function Code     | Register Data     |
| Protocol ID       | Reference Number  |                   |
| Length             |                   |                   |
| Unit ID           |                   |                   |
+-------------------+-------------------+-------------------+

from scapy.all import *
from collections import defaultdict, Counter

def analyze_modbus_traffic(pcap_file):
    packets = rdpcap(pcap_file)

    write_operations = []
    func_code_dist = Counter()
    src_dst_writes = defaultdict(list)
    unusual_transactions = []

    for pkt in packets:
        if TCP in pkt and pkt[TCP].dport == 502:
            payload = bytes(pkt[TCP].payload)
            if len(payload) >= 9:
                trans_id = int.from_bytes(payload[0:2], 'big')
                func_code = payload[7]
                func_code_dist[func_code] += 1

                if func_code in (0x05, 0x06, 0x0F, 0x10):
                    src_ip = pkt[IP].src if IP in pkt else "unknown"
                    dst_ip = pkt[IP].dst if IP in pkt else "unknown"
                    ref_num = int.from_bytes(payload[8:10], 'big')

                    write_operations.append({
                        "timestamp": float(pkt.time),
                        "src": src_ip,
                        "dst": dst_ip,
                        "func_code": func_code,
                        "reference": ref_num,
                        "transaction_id": trans_id
                    })
                    src_dst_writes[src_ip].append({
                        "timestamp": float(pkt.time),
                        "func_code": func_code,
                        "reference": ref_num
                    })

                if func_code not in (0x01, 0x02, 0x03, 0x04,
                                      0x05, 0x06, 0x0F, 0x10,
                                      0x08, 0x0B, 0x0C, 0x11,
                                      0x17, 0x2B):
                    unusual_transactions.append({
                        "timestamp": float(pkt.time),
                        "func_code": f"0x{func_code:02X}",
                        "src": pkt[IP].src if IP in pkt else "unknown",
                        "transaction_id": trans_id
                    })

    return {
        "total_writes": len(write_operations),
        "func_code_distribution": dict(func_code_dist),
        "write_operations": write_operations,
        "unusual_transactions": unusual_transactions,
        "suspicious_sources": {
            src: len(ops) for src, ops in src_dst_writes.items()
            if len(ops) > 100
        }
    }

# 过滤 Modbus 写操作
mbtcp.trans_id >= 0 && modbus.func_code in {5 6 15 16}

# 过滤非法事务标识符
mbtcp.trans_id > 60000

# 过滤特定源 IP 的 Modbus 通信
modbus && ip.src == 192.168.1.100

# 过滤诊断功能码
modbus.func_code == 8

# 过滤异常功能码
modbus.func_code > 43

def analyze_dnp3_operations(pcap_file):
    packets = rdpcap(pcap_file)

    operations = []
    time_sync_events = []
    select_operate_pairs = defaultdict(dict)

    for pkt in packets:
        if TCP in pkt and (pkt[TCP].dport == 20000 or pkt[TCP].sport == 20000):
            payload = bytes(pkt[TCP].payload)
            if len(payload) < 10:
                continue

            if pkt[TCP].dport == 20000:
                direction = "master_to_outstation"
            else:
                direction = "outstation_to_master"

            for offset in range(len(payload) - 2):
                if payload[offset] == 0x64:
                    al_func = payload[offset + 2] if offset + 2 < len(payload) else 0

                    if al_func in (0x05, 0x03, 0x04):
                        operations.append({
                            "timestamp": float(pkt.time),
                            "direction": direction,
                            "function_code": al_func,
                            "function_name": {
                                0x03: "SELECT",
                                0x04: "OPERATE",
                                0x05: "DIRECT_OPERATE"
                            }.get(al_func, f"UNKNOWN_0x{al_func:02X}"),
                            "src": pkt[IP].src if IP in pkt else "unknown"
                        })

                    if al_func == 0x0C:
                        time_sync_events.append({
                            "timestamp": float(pkt.time),
                            "direction": direction,
                            "src": pkt[IP].src if IP in pkt else "unknown"
                        })

    return {
        "operations": operations,
        "time_sync_events": time_sync_events,
        "direct_operate_count": sum(
            1 for op in operations if op["function_code"] == 0x05
        )
    }

from scapy.all import *

def analyze_goose_traffic(interface_or_pcap):
    goose_events = []
    state_change_count = 0
    sequence_anomalies = []
    prev_st_num = {}

    def goose_callback(pkt):
        nonlocal state_change_count
        if pkt.haslayer('GSE'):
            gse_layer = pkt['GSE']
            go_id = gse_layer.goID.decode() if hasattr(gse_layer, 'goID') else "unknown"
            st_num = gse_layer.stNum if hasattr(gse_layer, 'stNum') else 0
            sq_num = gse_layer.sqNum if hasattr(gse_layer, 'sqNum') else 0

            goose_events.append({
                "timestamp": float(pkt.time),
                "go_id": go_id,
                "st_num": st_num,
                "sq_num": sq_num,
                "src_mac": pkt.src
            })

            if go_id in prev_st_num:
                if st_num > prev_st_num[go_id]:
                    state_change_count += 1
                elif st_num == prev_st_num[go_id] and sq_num == 0:
                    sequence_anomalies.append({
                        "go_id": go_id,
                        "timestamp": float(pkt.time),
                        "issue": "SEQUENCE_RESET"
                    })

            prev_st_num[go_id] = st_num

    if os.path.isfile(interface_or_pcap):
        packets = rdpcap(interface_or_pcap)
        for pkt in packets:
            goose_callback(pkt)
    else:
        sniff(iface=interface_or_pcap, filter="ether proto 0x88B8",
              prn=goose_callback, count=10000, timeout=300)

    return {
        "total_goose_events": len(goose_events),
        "state_changes": state_change_count,
        "sequence_anomalies": sequence_anomalies,
        "unique_go_ids": list(set(e["go_id"] for e in goose_events))
    }

# 计算配置文件哈希
sha256sum *.scd *.icd *.cid > config_baseline.sha256

# 验证配置文件完整性
sha256sum -c config_baseline.sha256

# 对比 SCD 文件差异（XML 格式）
xmldiff baseline.scd current.scd --output scd_diff.xml

# Wireshark 过滤器：S7comm 协议
s7comm || s7commplus

# 过滤 S7 写操作
s7comm.param.func == 26

# 过滤块上传/下载
s7comm.param.func in {1 2 3}

# 过滤 CPU 控制命令（STOP/RUN/RESET）
s7comm.param.func == 28

def audit_opcua_security(opcua_audit_log):
    findings = []

    for entry in opcua_audit_log:
        if entry.get('event_type') == 'Write':
            if entry.get('user') not in KNOWN_AUTHORIZED_USERS:
                findings.append({
                    "timestamp": entry['timestamp'],
                    "event": "UNAUTHORIZED_WRITE",
                    "user": entry['user'],
                    "node_id": entry['node_id'],
                    "value": entry['value'],
                    "severity": "CRITICAL"
                })

        if entry.get('event_type') == 'CreateSession':
            if entry.get('security_policy') == 'None':
                findings.append({
                    "timestamp": entry['timestamp'],
                    "event": "INSECURE_SESSION",
                    "client": entry.get('client_address'),
                    "severity": "HIGH"
                })

        if entry.get('event_type') == 'Call':
            method_id = entry.get('method_id', '')
            if 'reset' in method_id.lower() or 'delete' in method_id.lower():
                findings.append({
                    "timestamp": entry['timestamp'],
                    "event": "DANGEROUS_METHOD_CALL",
                    "method": method_id,
                    "user": entry['user'],
                    "severity": "HIGH"
                })

    return findings

#!/usr/bin/env python3
import sys
import json
from collections import defaultdict, Counter
from datetime import datetime
from scapy.all import rdpcap, TCP, IP

PROTOCOL_PORTS = {
    'modbus': 502,
    'dnp3': 20000,
    's7comm': 102,
    'ethernetip': 44818,
    'opcua': 4840,
    'iec104': 2404
}

WRITE_FUNC_CODES = {
    'modbus': [0x05, 0x06, 0x0F, 0x10],
    'dnp3': [0x03, 0x04, 0x05],
}

def detect_protocol_anomalies(pcap_file, protocol, baseline_file=None):
    port = PROTOCOL_PORTS.get(protocol)
    if not port:
        print(f"不支持的协议: {protocol}")
        return None

    packets = rdpcap(pcap_file)
    write_ops = []
    new_sources = set()
    func_counter = Counter()
    time_anomalies = []
    prev_timestamp = None

    baseline_sources = set()
    if baseline_file:
        baseline_pkts = rdpcap(baseline_file)
        for pkt in baseline_pkts:
            if IP in pkt and TCP in pkt and pkt[TCP].dport == port:
                baseline_sources.add(pkt[IP].src)

    for pkt in packets:
        if IP not in pkt or TCP not in pkt:
            continue

        if pkt[TCP].dport != port:
            continue

        src_ip = pkt[IP].src
        payload = bytes(pkt[TCP].payload)

        if baseline_sources and src_ip not in baseline_sources:
            new_sources.add(src_ip)

        if len(payload) > 7:
            if protocol == 'modbus':
                fc = payload[7]
                func_counter[fc] += 1

                if fc in WRITE_FUNC_CODES['modbus']:
                    write_ops.append({
                        "timestamp": float(pkt.time),
                        "src": src_ip,
                        "dst": pkt[IP].dst,
                        "func_code": f"0x{fc:02X}"
                    })

        if prev_timestamp and float(pkt.time) < prev_timestamp:
            time_anomalies.append({
                "timestamp": float(pkt.time),
                "prev_timestamp": prev_timestamp,
                "issue": "OUT_OF_ORDER"
            })
        prev_timestamp = float(pkt.time)

    report = {
        "protocol": protocol,
        "pcap_file": pcap_file,
        "analysis_time": datetime.now().isoformat(),
        "total_packets": len(packets),
        "write_operations": len(write_ops),
        "new_sources": list(new_sources),
        "func_code_distribution": dict(func_counter),
        "time_anomalies": len(time_anomalies),
        "findings": []
    }

    if new_sources:
        report["findings"].append({
            "type": "NEW_SOURCE_DETECTED",
            "sources": list(new_sources),
            "severity": "HIGH"
        })

    if time_anomalies:
        report["findings"].append({
            "type": "TIMESTAMP_ANOMALY",
            "count": len(time_anomalies),
            "severity": "MEDIUM"
        })

    if len(write_ops) > 1000:
        report["findings"].append({
            "type": "EXCESSIVE_WRITES",
            "count": len(write_ops),
            "severity": "CRITICAL"
        })

    return report

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]} <pcap_file> <protocol> [baseline_pcap]")
        print(f"Protocols: {', '.join(PROTOCOL_PORTS.keys())}")
        sys.exit(1)

    pcap = sys.argv[1]
    proto = sys.argv[2]
    baseline = sys.argv[3] if len(sys.argv) > 3 else None

    result = detect_protocol_anomalies(pcap, proto, baseline)
    if result:
        print(json.dumps(result, indent=2, ensure_ascii=False))

# 使用 flashrom 从 SPI Flash 读取固件
flashrom -p ch341a_spi -c "W25Q128BV" -r firmware_dump.bin

# 验证读取完整性（多次读取并比对）
flashrom -p ch341a_spi -c "W25Q128BV" -r firmware_dump_verify1.bin
flashrom -p ch341a_spi -c "W25Q128BV" -r firmware_dump_verify2.bin
sha256sum firmware_dump.bin firmware_dump_verify1.bin firmware_dump_verify2.bin

# 使用 OpenOCD 通过 JTAG 提取固件
openocd -f interface/jlink.cfg -f target/stm32f4x.cfg \
  -c "init" -c "halt" \
  -c "dump_image firmware_jtag.bin 0x08000000 0x100000" \
  -c "exit"

# 通过固件更新接口提取（某些设备允许下载当前固件）
# Siemens S7-1500 固件备份（通过 TIA Portal）
# 操作路径：TIA Portal → 项目视图 → 在线 → 固件备份

# 使用 s7fwextract 工具提取 S7 固件
python3 s7fwextract.py --ip 192.168.1.10 --output s7_firmware.bin

# ABB AC 800M 固件提取（通过 Control Builder M）
# 操作路径：Controller → Online → Create Firmware Backup

# 固件类型识别
file firmware_dump.bin

# binwalk 扫描
binwalk firmware_dump.bin

# 提取固件中的文件系统
binwalk -e firmware_dump.bin

# 递归提取
binwalk -eM firmware_dump.bin

# 熵分析（检测加密/压缩区域）
binwalk -E firmware_dump.bin

# 提取 SquashFS 文件系统
unsquashfs firmware_dump.bin.extracted/*.squashfs

# 提取 CramFS
cramfsck firmware_dump.bin.extracted/*.cramfs -x extracted_cramfs/

# 提取 JFFS2 文件系统
jefferson firmware_dump.bin.extracted/*.jffs2 -d extracted_jffs2/

# 提取 UBI/UBIFS
ubireader_extract_images -o ubi_output/ firmware_dump.bin

# 使用 strings 和正则检测硬编码凭据
strings firmware_dump.bin | grep -iE \
  "(password|passwd|pwd|admin|root|default|secret|key|token)" | \
  sort -u > hardcoded_credentials.txt

# 使用 truffleHog 扫描固件中的密钥
trufflehog filesystem --directory extracted_firmware/ \
  --json > secrets_report.json

# 使用 credsfinder 检测凭据
python3 -m credsfinder scan -d extracted_firmware/ -o creds_report.json

import os
import re
import hashlib

def scan_firmware_backdoors(extracted_dir):
    backdoor_indicators = [
        rb'\x7fELF.*\x00/bin/sh',
        rb'nc\s+-[el]+\s+\d+',
        rb'wget\s+http',
        rb'curl\s+http',
        rb'/dev/tcp/',
        rb'mkfifo\s+/tmp/',
        rb'iptables\s+-F',
        rb'chmod\s+777',
        rb'suid|sgid',
    ]

    suspicious_strings = [
        "backdoor", "reverse_shell", "bind_shell",
        "rootkit", "keylog", "exfil", "c2_beacon",
        "telnetd", "ftpd", "dropbear", "ncat"
    ]

    findings = []

    for root, dirs, files in os.walk(extracted_dir):
        for fname in files:
            filepath = os.path.join(root, fname)
            try:
                with open(filepath, 'rb') as f:
                    content = f.read()

                for pattern in backdoor_indicators:
                    matches = re.findall(pattern, content)
                    if matches:
                        findings.append({
                            "file": filepath,
                            "type": "BINARY_PATTERN",
                            "pattern": pattern.decode('unicode_escape'),
                            "match_count": len(matches)
                        })

                content_lower = content.lower()
                for s in suspicious_strings:
                    if s.encode() in content_lower:
                        findings.append({
                            "file": filepath,
                            "type": "SUSPICIOUS_STRING",
                            "string": s,
                            "severity": "HIGH"
                        })

                if os.path.getsize(filepath) > 0:
                    file_hash = hashlib.sha256(content).hexdigest()
                    findings.append({
                        "file": filepath,
                        "type": "FILE_HASH",
                        "sha256": file_hash,
                        "size": os.path.getsize(filepath)
                    })

            except (PermissionError, IsADirectoryError):
                continue

    return findings

# 建立固件基线哈希库
find /firmware_baseline/ -type f | while read f; do
  sha256sum "$f"
done > firmware_baseline_hashes.txt

# 对比当前固件
find /firmware_current/ -type f | while read f; do
  sha256sum "$f"
done > firmware_current_hashes.txt

diff firmware_baseline_hashes.txt firmware_current_hashes.txt

Level 5: 企业网络（ERP/MES）
  │
Level 4: DMZ（工业 DMZ）
  │
Level 3: 生产运营管理（Historian/OPC Server/Engineering）
  │
Level 2: 监控层（HMI/SCADA Server）
  │
Level 1: 基本控制层（PLC/DCS Controller）
  │
Level 0: 物理过程层（传感器/执行器）

def verify_purdue_compliance(network_topology_file):
    violations = []

    with open(network_topology_file) as f:
        topology = json.load(f)

    level_ranges = {
        "Level_5_Enterprise": ["10.0.0.0/8"],
        "Level_4_DMZ": ["172.16.0.0/12"],
        "Level_3_Operations": ["192.168.10.0/24"],
        "Level_2_Supervisory": ["192.168.20.0/24"],
        "Level_1_Control": ["192.168.30.0/24"],
        "Level_0_Physical": ["192.168.40.0/24"]
    }

    for connection in topology.get('connections', []):
        src_level = connection.get('source_level')
        dst_level = connection.get('destination_level')

        allowed_connections = [
            ("Level_5_Enterprise", "Level_4_DMZ"),
            ("Level_4_DMZ", "Level_3_Operations"),
            ("Level_3_Operations", "Level_2_Supervisory"),
            ("Level_2_Supervisory", "Level_1_Control"),
            ("Level_1_Control", "Level_0_Physical"),
        ]

        pair = (src_level, dst_level)
        reverse_pair = (dst_level, src_level)

        if pair not in allowed_connections and reverse_pair not in allowed_connections:
            violations.append({
                "source": connection.get('source_ip'),
                "source_level": src_level,
                "destination": connection.get('destination_ip'),
                "destination_level": dst_level,
                "violation": "CROSS_LEVEL_CONNECTION",
                "severity": "CRITICAL"
            })

        if src_level == dst_level and src_level in ("Level_1_Control", "Level_0_Physical"):
            pass
        elif src_level == dst_level:
            violations.append({
                "source": connection.get('source_ip'),
                "destination": connection.get('destination_ip'),
                "level": src_level,
                "violation": "INTRA_LEVEL_CONNECTION",
                "severity": "MEDIUM"
            })

    return violations

# 从 IT 网络测试到 OT 网络的连通性（仅 ICMP，不发送侵入性探测）
for subnet in 192.168.10.0/24 192.168.20.0/24 192.168.30.0/24; do
  echo "Testing $subnet..."
  nmap -sn -PE --max-retries 1 -T1 $subnet 2>/dev/null | \
    grep "report" | awk '{print $5}'
done

# 检查防火墙规则（工业 DMZ 防火墙）
ssh dmz_firewall_admin@firewall_ip "show access-list" | \
  grep -E "permit|deny" | sort

# 分析工业防火墙日志中的异常连接尝试
grep "DENY" /var/log/firewall/ics_fw.log | \
  awk '{print $3, $5, $7}' | \
  sort | uniq -c | sort -rn | head -20

# 检测从 IT 到 OT 的异常端口访问
grep -E "102|502|2404|44818|4001" /var/log/firewall/ics_fw.log | \
  grep "src_zone=IT" | grep "dst_zone=OT"

# 检测防火墙规则变更
grep -iE "rule.*add|rule.*delete|rule.*modify|config.*change" \
  /var/log/firewall/ics_fw.log | tail -50

from collections import defaultdict
import statistics

def build_network_baseline(pcap_files, duration_hours=168):
    baseline = {
        "protocol_distribution": defaultdict(int),
        "source_ips": set(),
        "destination_ips": set(),
        "connections": defaultdict(int),
        "bytes_per_hour": defaultdict(int),
        "packets_per_protocol_per_hour": defaultdict(lambda: defaultdict(int))
    }

    for pcap_file in pcap_files:
        packets = rdpcap(pcap_file)
        for pkt in packets:
            if IP not in pkt:
                continue

            src = pkt[IP].src
            dst = pkt[IP].dst
            proto = "unknown"

            if TCP in pkt:
                dport = pkt[TCP].dport
                if dport == 502:
                    proto = "modbus"
                elif dport == 102:
                    proto = "s7comm"
                elif dport == 2404:
                    proto = "iec104"
                elif dport == 44818:
                    proto = "ethernetip"
                elif dport == 20000:
                    proto = "dnp3"
                else:
                    proto = f"tcp_{dport}"
            elif UDP in pkt:
                proto = f"udp_{pkt[UDP].dport}"

            baseline["protocol_distribution"][proto] += 1
            baseline["source_ips"].add(src)
            baseline["destination_ips"].add(dst)
            baseline["connections"][(src, dst, proto)] += 1

    baseline["source_ips"] = list(baseline["source_ips"])
    baseline["destination_ips"] = list(baseline["destination_ips"])

    return baseline

def detect_baseline_deviations(current_traffic, baseline, threshold_sigma=3):
    deviations = []

    for src, dst, proto, count in current_traffic:
        key = (src, dst, proto)
        baseline_count = baseline["connections"].get(key, 0)

        if src not in baseline["source_ips"]:
            deviations.append({
                "type": "NEW_SOURCE",
                "source": src,
                "destination": dst,
                "protocol": proto,
                "severity": "HIGH"
            })

        if proto not in baseline["protocol_distribution"]:
            deviations.append({
                "type": "NEW_PROTOCOL",
                "protocol": proto,
                "source": src,
                "destination": dst,
                "severity": "HIGH"
            })

    return deviations

# 检测工控网络中的远程访问工具
nmap -sV -p 22,3389,5900,5985,5986 --open 192.168.0.0/16

# 检测异常的出站连接（可能的 C2 通道）
tcpdump -i any -nn 'dst port 443 or dst port 8443 or dst port 53' \
  -w outbound_suspicious.pcap -c 10000

# 检测 VPN 通道
nmap -sU -p 500,4500 --open 192.168.0.0/16

初始感染（USB 快捷方式漏洞 CVE-2010-0466）
  → 内网传播（MS08-067 / MS10-046 / MS10-061）
  → 对等网络建立
  → Siemens Step7 软件检测
  → S7-315 PLC 感染
  → 读取特定配置（Profibus 拓扑）
  → 载荷注入（修改 OB1/OB35 块）
  → 中间人攻击（重放正常传感器数据）
  → 离心机转子超速/低速交替破坏
  → 同时向监控系统回传正常数据

def detect_stuxnet_indicators(system_path):
    indicators = []

    known_hashes = {
        "6229986940390658392b9e77a9498629": "wmi.dll backdoor",
        "384723719d7f4e4688792f8506264841": "s7otbxdx.dll MITM",
        "c5159690c7ffc0164a4b0b1bf89b52c6": "payload DLL"
    }

    for root, _, files in os.walk(system_path):
        for fname in files:
            fpath = os.path.join(root, fname)
            try:
                with open(fpath, 'rb') as f:
                    md5 = hashlib.md5(f.read()).hexdigest()
                if md5 in known_hashes:
                    indicators.append({
                        "file": fpath,
                        "hash": md5,
                        "description": known_hashes[md5],
                        "severity": "CRITICAL"
                    })
            except (PermissionError, OSError):
                continue

    return indicators

正常流程:
  传感器 → BCS(基本控制) → 正常调节
                ↓ (失控时)
            SIS(安全仪表) → 紧急停车(ESD)

Triton 攻击流程:
  传感器 → BCS(正常) → 正常调节
                ↓
            SIS(被篡改) → 不执行安全停车 → 物理损坏/爆炸

def detect_triton_indicators(log_data, controller_config):
    findings = []

    for event in log_data:
        if event.get('protocol') == 'triston' or event.get('port') == 1513:
            if event.get('function') in ('WRITE_ALL', 'DOWNLOAD'):
                findings.append({
                    "timestamp": event['timestamp'],
                    "type": "SAFETY_LOGIC_MODIFICATION",
                    "severity": "CRITICAL",
                    "detail": f"安全逻辑修改操作: {event.get('function')}",
                    "source": event.get('source_ip', 'unknown')
                })

        if event.get('function') == 'STOP' and event.get('target') == 'SIS':
            findings.append({
                "timestamp": event['timestamp'],
                "type": "SIS_STOP_COMMAND",
                "severity": "CRITICAL",
                "detail": "安全仪表系统被停止"
            })

    for block in controller_config.get('function_blocks', []):
        if block.get('type') not in EXPECTED_SIS_BLOCK_TYPES:
            findings.append({
                "type": "UNKNOWN_SIS_BLOCK",
                "block_name": block.get('name'),
                "block_type": block.get('type'),
                "severity": "CRITICAL",
                "detail": "SIS 中发现非预期的功能块"
            })

    return findings

def detect_iec104_anomalies(pcap_file):
    suspicious_commands = []

    iec104_cause_types = {
        6: "激活（Activation）",
        7: "激活确认（Activation Confirmation）",
        8: "停止激活（Deactivation）",
        44: "未知类型（Unknown Type）",
        45: "未知传送原因",
        46: "未知公共地址",
        47: "未知信息对象地址"
    }

    packets = rdpcap(pcap_file)
    for pkt in packets:
        if TCP in pkt and pkt[TCP].dport == 2404:
            payload = bytes(pkt[TCP].payload)
            if len(payload) < 6:
                continue

            cause_type = payload[2] & 0x3F if len(payload) > 2 else 0

            if cause_type in (44, 45, 46, 47):
                suspicious_commands.append({
                    "timestamp": float(pkt.time),
                    "type": "IEC104_UNKNOWN_CAUSE",
                    "cause_type": cause_type,
                    "description": iec104_cause_types.get(cause_type, "Unknown"),
                    "severity": "HIGH"
                })

            type_id = payload[0] if len(payload) > 0 else 0
            if type_id in (45, 46, 47):
                suspicious_commands.append({
                    "timestamp": float(pkt.time),
                    "type": "IEC104_COMMAND",
                    "type_id": type_id,
                    "description": "单命令/双命令操作",
                    "severity": "CRITICAL"
                })

    return suspicious_commands

BlackEnergy 3 架构:
├── 初始感染载体（鱼叉式钓鱼邮件 / 水坑攻击）
├── C2 通信模块（HTTP/HTTPS）
├── 破坏模块（KillDisk - 磁盘擦除）
├── 远程桌面模块（远程操作工控系统）
└── 工控攻击模块
    ├── 断路器控制脚本
    ├── UPS 配置（防止自身断电）
    └── 通信中断工具

class ICSThreatDetector:
    def __init__(self):
        self.ioc_database = {
            "file_hashes": {},
            "network_signatures": [],
            "protocol_anomalies": [],
            "behavioral_patterns": []
        }
        self.findings = []

    def load_ioc_database(self, ioc_file):
        with open(ioc_file) as f:
            data = json.load(f)
            self.ioc_database.update(data)

    def scan_file_system(self, path):
        for root, _, files in os.walk(path):
            for fname in files:
                fpath = os.path.join(root, fname)
                try:
                    with open(fpath, 'rb') as f:
                        content = f.read()
                    file_hash = hashlib.sha256(content).hexdigest()
                    if file_hash in self.ioc_database["file_hashes"]:
                        self.findings.append({
                            "type": "KNOWN_MALWARE",
                            "file": fpath,
                            "hash": file_hash,
                            "malware_family": self.ioc_database["file_hashes"][file_hash],
                            "severity": "CRITICAL"
                        })
                except (PermissionError, OSError):
                    continue

    def analyze_network_traffic(self, pcap_file):
        packets = rdpcap(pcap_file)
        for pkt in packets:
            if TCP in pkt:
                payload = bytes(pkt[TCP].payload)
                for sig in self.ioc_database["network_signatures"]:
                    if sig["pattern"] in payload:
                        self.findings.append({
                            "type": "NETWORK_SIGNATURE_MATCH",
                            "timestamp": float(pkt.time),
                            "signature": sig["name"],
                            "severity": sig.get("severity", "HIGH")
                        })

    def generate_report(self):
        return {
            "scan_time": datetime.now().isoformat(),
            "total_findings": len(self.findings),
            "critical_count": sum(1 for f in self.findings if f["severity"] == "CRITICAL"),
            "high_count": sum(1 for f in self.findings if f["severity"] == "HIGH"),
            "findings": self.findings
        }

rule Stuxnet_LNK_Exploit {
    meta:
        description = "Detects Stuxnet LNK exploit (CVE-2010-0466)"
        author = "ICS Forensics Team"
        date = "2026-07-02"
    strings:
        $lnk_magic = { 4C 00 00 00 }
        $clsid_1 = { 00 02 14 01 00 00 }
        $shellcode_pattern = { 8B FF 55 8B EC 83 EC ?? B8 }
    condition:
        $lnk_magic at 0 and ($clsid_1 or $shellcode_pattern)
}

rule Triton_TriStation_Protocol {
    meta:
        description = "Detects Triton TriStation protocol abuse"
        author = "ICS Forensics Team"
        date = "2026-07-02"
    strings:
        $triston_port = { 05 E9 }
        $write_all = { 0D 00 }
        $download_cmd = { 0B 00 }
    condition:
        $triston_port and ($write_all or $download_cmd)
}

rule Industroyer_IEC104_Command {
    meta:
        description = "Detects IEC 104 command manipulation"
        author = "ICS Forensics Team"
        date = "2026-07-02"
    strings:
        $start_byte = { 68 }
        $single_command = { 2D }
        $double_command = { 2E }
        $cause_activate = { 06 }
    condition:
        $start_byte at 0 and ($single_command or $double_command) and $cause_activate
}

rule Havex_Scan_Pattern {
    meta:
        description = "Detects Havex ICS scanning behavior"
        author = "ICS Forensics Team"
        date = "2026-07-02"
    strings:
        $modbus_scan = { 01 03 00 00 00 01 }
        $dnp3_scan = { 05 64 05 C0 }
        $banner_strings = "Modbus/TCP" "DNP3" "IEC 104"
    condition:
        ($modbus_scan or $dnp3_scan) and any of ($banner_strings)
}

rule Generic_ICS_Backdoor {
    meta:
        description = "Generic ICS backdoor detection"
        author = "ICS Forensics Team"
        date = "2026-07-02"
    strings:
        $reverse_shell = /(?:nc|ncat|netcat)\s+-[el]+\s+\d+/ nocase
        $hidden_process = /\/dev\/\.hidden/ nocase
        $plc_manipulation = /(?:write|modify).*(?:register|coil|ladder|PLC)/ nocase
        $data_exfil = /(?:curl|wget|ftp)\s+.*(?:upload|post|send)/ nocase
    condition:
        any of them
}

SIL 1: 低风险 - 年失效概率 0.1 ~ 0.01
SIL 2: 中风险 - 年失效概率 0.01 ~ 0.001
SIL 3: 高风险 - 年失效概率 0.001 ~ 0.0001
SIL 4: 极高风险 - 年失效概率 0.0001 ~ 0.00001（极少使用）

def detect_sis_logic_tampering(current_logic, baseline_logic):
    tampering_indicators = []

    for func_block_id in baseline_logic:
        if func_block_id not in current_logic:
            tampering_indicators.append({
                "block_id": func_block_id,
                "type": "SAFETY_BLOCK_REMOVED",
                "severity": "CRITICAL",
                "description": "安全功能块被移除"
            })

    for func_block_id in current_logic:
        if func_block_id not in baseline_logic:
            tampering_indicators.append({
                "block_id": func_block_id,
                "type": "NEW_SAFETY_BLOCK",
                "severity": "CRITICAL",
                "description": "发现非预期的安全功能块"
            })
            continue

        base_block = baseline_logic[func_block_id]
        curr_block = current_logic[func_block_id]

        if curr_block.get('parameters') != base_block.get('parameters'):
            tampering_indicators.append({
                "block_id": func_block_id,
                "type": "PARAMETER_MODIFIED",
                "original_params": base_block.get('parameters'),
                "modified_params": curr_block.get('parameters'),
                "severity": "CRITICAL"
            })

        if curr_block.get('logic_type') != base_block.get('logic_type'):
            tampering_indicators.append({
                "block_id": func_block_id,
                "type": "LOGIC_TYPE_CHANGED",
                "original_type": base_block.get('logic_type'),
                "modified_type": curr_block.get('logic_type'),
                "severity": "CRITICAL"
            })

    return tampering_indicators

def analyze_sis_setpoints(current_setpoints, design_specifications):
    anomalies = []

    for tag, value in current_setpoints.items():
        if tag in design_specifications:
            spec = design_specifications[tag]
            min_allowed = spec.get('min_safe_value')
            max_allowed = spec.get('max_safe_value')

            if value < min_allowed or value > max_allowed:
                anomalies.append({
                    "tag": tag,
                    "current_setpoint": value,
                    "safe_range": f"[{min_allowed}, {max_allowed}]",
                    "deviation": value - spec.get('nominal_value', 0),
                    "severity": "CRITICAL",
                    "risk": "安全设定值超出设计安全范围"
                })

            if spec.get('change_requires_approval') and tag not in spec.get('approved_changes', []):
                anomalies.append({
                    "tag": tag,
                    "type": "UNAUTHORIZED_CHANGE",
                    "severity": "HIGH",
                    "description": "设定值变更未经审批流程"
                })

    return anomalies

def detect_bypass_suppression(sis_status_data):
    bypass_events = []

    for channel in sis_status_data.get('channels', []):
        if channel.get('bypassed'):
            bypass_events.append({
                "channel": channel['id'],
                "tag": channel.get('tag_name'),
                "bypass_start": channel.get('bypass_timestamp'),
                "bypass_duration": channel.get('bypass_duration'),
                "authorized_by": channel.get('bypass_authorizer', 'UNKNOWN'),
                "severity": "CRITICAL" if channel.get('safety_critical') else "HIGH"
            })

        if channel.get('suppressed'):
            bypass_events.append({
                "channel": channel['id'],
                "tag": channel.get('tag_name'),
                "type": "ALARM_SUPPRESSED",
                "suppress_time": channel.get('suppress_timestamp'),
                "severity": "CRITICAL"
            })

    for vote in sis_status_data.get('voting_configs', []):
        original = vote.get('original_voting')
        current = vote.get('current_voting')
        if original != current:
            bypass_events.append({
                "type": "VOTING_CHANGED",
                "channel": vote['id'],
                "original": original,
                "current": current,
                "severity": "CRITICAL",
                "description": f"投票逻辑从 {original} 变更为 {current}"
            })

    return bypass_events

def detect_sensor_actuator_anomalies(sensor_data, actuator_data):
    anomalies = []

    for sensor in sensor_data:
        if sensor.get('stuck_value'):
            anomalies.append({
                "type": "SENSOR_STUCK",
                "sensor_id": sensor['id'],
                "stuck_value": sensor['value'],
                "duration": sensor.get('stuck_duration'),
                "severity": "HIGH",
                "description": "传感器值长时间未变化，可能被欺骗或故障"
            })

        if sensor.get('range_violation'):
            anomalies.append({
                "type": "SENSOR_RANGE_VIOLATION",
                "sensor_id": sensor['id'],
                "value": sensor['value'],
                "valid_range": sensor.get('valid_range'),
                "severity": "CRITICAL"
            })

    for actuator in actuator_data:
        if actuator.get('command_mismatch'):
            anomalies.append({
                "type": "ACTUATOR_COMMAND_MISMATCH",
                "actuator_id": actuator['id'],
                "expected_position": actuator.get('expected'),
                "actual_position": actuator.get('actual'),
                "severity": "CRITICAL",
                "description": "执行器实际位置与控制命令不一致"
            })

    return anomalies

# 提取 SIS 事件日志（以 Triconex 为例）
# 通过 TriStation 1131 软件导出事件日志
triston_export --controller 192.168.30.50 --output sis_events.csv --format csv

# 分析安全事件时间线
python3 -c "
import csv
from datetime import datetime

events = []
with open('sis_events.csv') as f:
    reader = csv.DictReader(f)
    for row in reader:
        events.append(row)

for e in sorted(events, key=lambda x: x['timestamp']):
    if e['event_type'] in ('BYPASS', 'SUPPRESS', 'TRIP', 'RESET'):
        print(f\"{e['timestamp']} [{e['severity']}] {e['event_type']}: {e['description']}\")
"

def verify_sil_compliance(sis_config, process_requirements):
    verification_results = []

    for safety_function in sis_config.get('safety_functions', []):
        sf_id = safety_function['id']
        required_sil = process_requirements.get(sf_id, {}).get('required_sil')
        achieved_sil = safety_function.get('achieved_sil')

        sil_levels = {'SIL_1': 1, 'SIL_2': 2, 'SIL_3': 3, 'SIL_4': 4}
        required_level = sil_levels.get(required_sil, 0)
        achieved_level = sil_levels.get(achieved_sil, 0)

        result = {
            "safety_function": sf_id,
            "required_sil": required_sil,
            "achieved_sil": achieved_sil,
            "compliant": achieved_level >= required_level
        }

        if not result["compliant"]:
            result["severity"] = "CRITICAL"
            result["description"] = f"SIL 等级不满足要求: 需要 {required_sil}, 实际 {achieved_sil}"

        pfh_data = safety_function.get('pfh_data', {})
        if pfh_data:
            result["pfh"] = pfh_data.get('value')
            result["pfh_range"] = pfh_data.get('valid_range')

        verification_results.append(result)

    return verification_results

物理影响证据等级评估:

Level 5 - 直接物理损害证据:
  ├── 设备物理损坏记录
  ├── 传感器读数超出安全范围
  └── 执行器异常动作日志

Level 4 - 物理过程异常:
  ├── 工艺参数偏离正常范围
  ├── 非计划停车事件
  └── 安全系统触发记录

Level 3 - 控制逻辑异常:
  ├── PLC 程序与基线不一致
  ├── 设定点被修改
  └── 控制输出异常

Level 2 - 系统配置异常:
  ├── 固件版本不匹配
  ├── 配置参数偏差
  └── 通信配置变更

Level 1 - 信息性指标:
  ├── 系统日志中的异常条目
  ├── 用户登录记录
  └── 网络流量统计变化

def correlate_it_ot_evidence(it_events, ot_events, time_window_sec=300):
    correlations = []

    for it_event in it_events:
        it_ts = it_event['timestamp']
        related_ot_events = []

        for ot_event in ot_events:
            ot_ts = ot_event['timestamp']
            time_diff = abs((it_ts - ot_ts).total_seconds())

            if time_diff <= time_window_sec:
                related_ot_events.append({
                    "event": ot_event,
                    "time_diff_sec": time_diff
                })

        if related_ot_events:
            correlations.append({
                "it_event": it_event,
                "related_ot_events": sorted(
                    related_ot_events, key=lambda x: x['time_diff_sec']
                ),
                "correlation_strength": "STRONG"
                    if any(e['time_diff_sec'] < 30 for e in related_ot_events)
                    else "MODERATE",
                "attack_phase": determine_attack_phase(
                    it_event, related_ot_events
                )
            })

    return correlations

def determine_attack_phase(it_event, ot_events):
    it_type = it_event.get('event_type', '')
    phase_mapping = {
        'lateral_movement': 'INITIAL_ACCESS',
        'credential_theft': 'CREDENTIAL_COMPROMISE',
        'privilege_escalation': 'PRIVILEGE_ESCALATION',
        'file_transfer': 'PAYLOAD_DELIVERY',
        'remote_access': 'PERSISTENCE',
    }

    for ot_event in ot_events:
        ot_type = ot_event['event'].get('event_type', '')
        if 'write' in ot_type.lower() or 'modify' in ot_type.lower():
            return 'IMPACT'
        if 'scan' in ot_type.lower() or 'enum' in ot_type.lower():
            return 'RECONNAISSANCE'

    return phase_mapping.get(it_type, 'UNKNOWN')

工控取证报告模板:
├── 1. 执行摘要
│   ├── 事件概述
│   ├── 关键发现
│   └── 风险评级
├── 2. 取证范围与方法
│   ├── 取证时间范围
│   ├── 目标系统清单
│   ├── 取证工具与方法
│   └── 安全约束说明
├── 3. 技术发现
│   ├── 网络层分析结果
│   ├── 主机层分析结果
│   ├── 工控设备分析结果
│   ├── 协议分析结果
│   └── 恶意软件分析结果
├── 4. 攻击链重建
│   ├── 时间线
│   ├── 攻击路径
│   └── 影响评估
├── 5. 证据清单
│   ├── 证据编号与描述
│   ├── 证据强度评级
│   └── 证据保全记录
├── 6. 修复建议
│   ├── 短期修复措施
│   ├── 长期安全加固
│   └── 监控增强建议
└── 7. 附录
    ├── IOC 列表
    ├── 工具输出详情
    └── 参考资料

title: Modbus 异常写操作
id: a1b2c3d4-e5f6-7890-abcd-ef1234567890
status: experimental
description: 检测 Modbus TCP 协议中的异常写操作
author: ICS Security Team
date: 2026/07/02
references:
    - https://www.iana.org/assignments/modbus-application-protocol/
tags:
    - attack.ics
    - attack.command_and_control
logsource:
    product: network
    service: modbus
detection:
    selection_write:
        protocol: modbus
        func_code:
            - 5
            - 6
            - 15
            - 16
    filter_known_masters:
        src_ip:
            - '192.168.20.10'
            - '192.168.20.11'
    condition: selection_write and not filter_known_masters
level: high
falsepositives:
    - 合法的工程师站操作
    - 维护窗口期间的配置变更

title: S7comm PLC STOP 命令
id: b2c3d4e5-f6a7-8901-bcde-f12345678901
status: experimental
description: 检测针对 Siemens S7 PLC 的 STOP 命令
author: ICS Security Team
date: 2026/07/02
tags:
    - attack.ics
    - attack.impair_process_control
logsource:
    product: network
    service: s7comm
detection:
    selection:
        protocol: s7comm
        param_func: 28
        cpu_state: STOP
    condition: selection
level: critical
falsepositives:
    - 计划内维护操作

title: DNP3 直接操作命令异常
id: c3d4e5f6-a7b8-9012-cdef-123456789012
status: experimental
description: 检测 DNP3 协议中非授权来源的直接操作命令
author: ICS Security Team
date: 2026/07/02
tags:
    - attack.ics
    - attack.command_and_control
logsource:
    product: network
    service: dnp3
detection:
    selection:
        protocol: dnp3
        function_code: 5
        direction: master_to_outstation
    filter_authorized:
        src_ip:
            - '192.168.20.5'
    condition: selection and not filter_authorized
level: critical
falsepositives:
    - 备用主站切换
    - 维护期间操作

#!/usr/bin/env python3
from scapy.all import sniff, TCP, IP, UDP
from collections import defaultdict
import json
import time
from datetime import datetime

class ICSProtocolMonitor:
    def __init__(self, interface, output_file=None):
        self.interface = interface
        self.output_file = output_file
        self.stats = defaultdict(int)
        self.anomalies = []
        self.known_masters = set()
        self.known_slaves = set()
        self.write_operations = []
        self.start_time = time.time()

    def packet_handler(self, pkt):
        if IP not in pkt:
            return

        if TCP in pkt:
            dport = pkt[TCP].dport
            sport = pkt[TCP].sport
            payload = bytes(pkt[TCP].payload)

            if dport == 502 or sport == 502:
                self._handle_modbus(pkt, payload, dport == 502)
            elif dport == 102 or sport == 102:
                self._handle_s7comm(pkt, payload)
            elif dport == 2404 or sport == 2404:
                self._handle_iec104(pkt, payload, dport == 2404)
            elif dport == 20000 or sport == 20000:
                self._handle_dnp3(pkt, payload, dport == 20000)
            elif dport == 44818 or sport == 44818:
                self._handle_ethernetip(pkt, payload)

    def _handle_modbus(self, pkt, payload, is_request):
        self.stats['modbus_total'] += 1
        if is_request and len(payload) >= 8:
            fc = payload[7]
            self.stats[f'modbus_fc_{fc}'] += 1

            if fc in (0x05, 0x06, 0x0F, 0x10):
                src = pkt[IP].src
                self.write_operations.append({
                    "timestamp": datetime.now().isoformat(),
                    "protocol": "modbus",
                    "src": src,
                    "dst": pkt[IP].dst,
                    "func_code": fc,
                    "reference": int.from_bytes(payload[8:10], 'big')
                        if len(payload) >= 10 else 0
                })

                if src not in self.known_masters:
                    self.anomalies.append({
                        "timestamp": datetime.now().isoformat(),
                        "type": "UNKNOWN_MODBUS_MASTER",
                        "source": src,
                        "severity": "HIGH"
                    })

    def _handle_s7comm(self, pkt, payload):
        self.stats['s7comm_total'] += 1
        if len(payload) > 10:
            pdu_type = payload[5] if len(payload) > 5 else 0
            if pdu_type == 1:
                param_func = payload[10] if len(payload) > 10 else 0
                self.stats[f's7comm_func_{param_func}'] += 1

                if param_func == 28:
                    self.anomalies.append({
                        "timestamp": datetime.now().isoformat(),
                        "type": "S7_CPU_CONTROL",
                        "source": pkt[IP].src,
                        "severity": "CRITICAL"
                    })

    def _handle_iec104(self, pkt, payload, is_request):
        self.stats['iec104_total'] += 1
        if is_request and len(payload) > 2:
            type_id = payload[0]
            cause = payload[2] & 0x3F if len(payload) > 2 else 0

            if type_id in (45, 46, 47):
                self.write_operations.append({
                    "timestamp": datetime.now().isoformat(),
                    "protocol": "iec104",
                    "src": pkt[IP].src,
                    "dst": pkt[IP].dst,
                    "type_id": type_id,
                    "cause": cause
                })

    def _handle_dnp3(self, pkt, payload, is_request):
        self.stats['dnp3_total'] += 1

    def _handle_ethernetip(self, pkt, payload):
        self.stats['ethernetip_total'] += 1

    def get_report(self):
        return {
            "monitor_duration_sec": time.time() - self.start_time,
            "statistics": dict(self.stats),
            "anomalies": self.anomalies,
            "write_operations": self.write_operations,
            "anomaly_count": len(self.anomalies),
            "critical_count": sum(
                1 for a in self.anomalies if a['severity'] == 'CRITICAL'
            )
        }

    def start(self, packet_count=0, timeout=0):
        sniff(
            iface=self.interface,
            prn=self.packet_handler,
            count=packet_count if packet_count else 0,
            timeout=timeout if timeout else None,
            store=False
        )
        return self.get_report()

#!/usr/bin/env python3
import hashlib
import json
import time
from datetime import datetime

class PLCIntegrityMonitor:
    def __init__(self, plc_connections, check_interval_sec=300):
        self.plc_connections = plc_connections
        self.check_interval = check_interval_sec
        self.baselines = {}
        self.alerts = []

    def establish_baseline(self, plc_id, program_data):
        self.baselines[plc_id] = {
            "program_hash": hashlib.sha256(
                json.dumps(program_data, sort_keys=True).encode()
            ).hexdigest(),
            "block_count": len(program_data.get('blocks', [])),
            "timestamp": datetime.now().isoformat(),
            "block_hashes": {
                block['id']: hashlib.sha256(
                    json.dumps(block, sort_keys=True).encode()
                ).hexdigest()
                for block in program_data.get('blocks', [])
            }
        }

    def check_integrity(self, plc_id, current_program):
        if plc_id not in self.baselines:
            return {"status": "NO_BASELINE", "severity": "MEDIUM"}

        baseline = self.baselines[plc_id]
        current_hash = hashlib.sha256(
            json.dumps(current_program, sort_keys=True).encode()
        ).hexdigest()

        if current_hash != baseline["program_hash"]:
            changed_blocks = []
            for block in current_program.get('blocks', []):
                block_hash = hashlib.sha256(
                    json.dumps(block, sort_keys=True).encode()
                ).hexdigest()
                if block['id'] not in baseline["block_hashes"]:
                    changed_blocks.append({
                        "block_id": block['id'],
                        "change": "NEW_BLOCK",
                        "severity": "CRITICAL"
                    })
                elif block_hash != baseline["block_hashes"][block['id']]:
                    changed_blocks.append({
                        "block_id": block['id'],
                        "change": "MODIFIED",
                        "severity": "CRITICAL"
                    })

            baseline_blocks = set(baseline["block_hashes"].keys())
            current_blocks = set(
                b['id'] for b in current_program.get('blocks', [])
            )
            removed = baseline_blocks - current_blocks
            for block_id in removed:
                changed_blocks.append({
                    "block_id": block_id,
                    "change": "REMOVED",
                    "severity": "CRITICAL"
                })

            alert = {
                "timestamp": datetime.now().isoformat(),
                "plc_id": plc_id,
                "type": "PROGRAM_MODIFIED",
                "changed_blocks": changed_blocks,
                "severity": "CRITICAL"
            }
            self.alerts.append(alert)
            return alert

        return {"status": "INTACT", "severity": "NONE"}