智慧能源与电网系统安全取证深度分析

sudo apt install wireshark tshark zeek
git clone https://github.com/automayt/ICSmap.git
git clone https://github.com/hslatman/icsmeaner.git
pip install pymodbus python-snap7 scapy

tshark -r modbus_capture.pcap -Y "modbus" -T fields -e frame.time -e ip.src -e ip.dst -e modbus.unit_id -e modbus.func_code -e modbus.data

from pymodbus.client import ModbusTcpClient

client = ModbusTcpClient('192.168.1.100', port=502)
client.connect()

result = client.read_holding_registers(address=0, count=10, unit=1)
print(f"Registers 0-9: {result.registers}")

result2 = client.read_coils(address=0, count=16, unit=1)
print(f"Coils 0-15: {result2.bits[:16]}")

client.close()

tshark -r dnp3_capture.pcap -Y "dnp3" -T fields -e frame.time -e ip.src -e ip.dst -e dnp3.ise -e dnp3.func -e dnp3.i -e dnp3.object.group -e dnp3.object.variation

tshark -r goose_capture.pcap -Y "goose" -T fields -e frame.time -e eth.src -e eth.dst -e goose.gocbref -e goose.dataset -e goose.datsetnum -e goose.stnum -e goose.sqnum -e goose.test -e goose.confrev

import pandas as pd
import numpy as np

def detect_meter_tampering(meter_data, window_size=24, threshold=2.0):
    meter_data['timestamp'] = pd.to_datetime(meter_data['timestamp'])
    meter_data = meter_data.sort_values('timestamp')
    meter_data['consumption_diff'] = meter_data['reading'].diff()
    rolling_mean = meter_data['consumption_diff'].rolling(window=window_size).mean()
    rolling_std = meter_data['consumption_diff'].rolling(window=window_size).std()
    z_scores = (meter_data['consumption_diff'] - rolling_mean) / rolling_std
    anomalies = meter_data[np.abs(z_scores) > threshold]
    return anomalies

tshark -r smart_meter.pcap -Y "dl645" -T fields -e frame.time -e dl645.address -e dl645.control -e dl645.data_id -e dl645.data

tshark -r ami_traffic.pcap -Y "tcp.port == 8080 || tls" -T fields -e frame.time -e ip.src -e ip.dst -e tcp.srcport -e tcp.dstport -e tls.handshake.type -e tls.record.version

nmap -sS -sV -p 502,102,4840,44818,80,443,22,3389 10.10.0.0/16 -oX wind_farm_scan.xml

binwalk inverter_firmware.bin
binwalk -e inverter_firmware.bin
ls -la _inverter_firmware.bin.extracted/
strings _inverter_firmware.bin.extracted/squashfs-root/usr/bin/* | grep -i "password\|key\|token\|api"

import numpy as np
from scipy import signal

def analyze_frequency_disturbance(freq_data, fs=50):
    nyq = 0.5 * fs
    b, a = signal.butter(4, 0.5/nyq, btype='low')
    filtered = signal.filtfilt(b, a, freq_data)
    deviation = np.abs(filtered - 50.0)
    disturbance_indices = np.where(deviation > 0.2)[0]
    rocof = np.gradient(filtered)
    rapid_change = np.where(np.abs(rocof) > 1.0)[0]
    return {
        'max_deviation': np.max(deviation),
        'disturbance_count': len(disturbance_indices),
        'rocof_max': np.max(np.abs(rocof)),
        'rapid_change_count': len(rapid_change)
    }

tshark -r pmu_capture.pcap -Y "c37.118" -T fields -e frame.time -e c37.118.fgen -e c37.118.freq -e c37.118.phim -e c37.118.mag -e c37.118.ang

volatility -f ems_memory.raw --profile=Win7SP1x64 pslist
volatility -f ems_memory.raw --profile=Win7SP1x64 mftparser
volatility -f ems_memory.raw --profile=Win7SP1x64 netscan

cat /etc/passwd | awk -F: '$3==0{print $1}'
find / -perm -4000 -type f 2>/dev/null
find /var/www -name "*.php" -newer /var/www/index.html
crontab -l && for user in $(cut -f1 -d: /etc/passwd); do crontab -u $user -l 2>/dev/null; done
netstat -tlnp | grep -v "127.0.0.1"

tshark -r substation_traffic.pcap -Y "eth.dst == 01:0c:cd:01:00:00" -T fields -e frame.time -e eth.src -e goose.gocbref -e goose.stnum -e goose.sqnum -e goose.test -e goose.datset

hackrf_transfer -r plc_signal.raw -f 100000 -s 2000000 -a 1
python3 plc_decoder.py --input plc_signal.raw --freq 100000 --protocol prime

title: Suspicious Modbus Function Code Enumeration
id: 8f4a2b3c-5d6e-7f8a-9b0c-1d2e3f4a5b6c
status: experimental
description: Detects sequential Modbus function code probing indicative of device enumeration or vulnerability scanning
references:
  - https://attack.mitre.org/techniques/T1046/
  - https://attack.mitre.org/techniques/T1595/
tags:
  - attack.discovery
  - attack.t1046
  - attack.t1595
  - attack.ics
logsource:
  product: ics
  service: scada
detection:
  selection_modbus_scan:
    protocol: modbus
    function_code|endswith:
      - '01'
      - '02'
      - '03'
      - '04'
      - '05'
      - '06'
      - '15'
      - '16'
  timeframe: 60s
  condition: selection_modbus_scan | count(function_code) by src_ip > 8
  falsepositives:
    - Legitimate engineering maintenance
    - PLC programming sessions
level: high

title: Anomalous GOOSE State Number Jump
id: 2a3b4c5d-6e7f-8a9b-0c1d-2e3f4a5b6c7d
status: experimental
description: Detects abnormal GOOSE stNum changes outside of expected operational periods
references:
  - https://attack.mitre.org/techniques/T0836/
tags:
  - attack.impact
  - attack.t0836
logsource:
  product: ics
  service: substation_automation
detection:
  selection_goose_anomaly:
    protocol: goose
    stnum_delta: 0
    timeframe: 300s
  filter_business_hours:
    timestamp|time: '08:00-18:00'
  condition: selection_goose_anomaly and not filter_business_hours
  falsepositives:
    - Planned switching operations
    - Protection device testing
level: critical

title: Power System Frequency Deviation Attack
id: 3b4c5d6e-7f8a-9b0c-1d2e-3f4a5b6c7d8e
status: experimental
description: Detects frequency deviations beyond normal operating range that may indicate cyber-physical attack
tags:
  - attack.impact
  - attack.t0831
logsource:
  product: ics
  service: power_grid_monitoring
detection:
  selection_freq_anomaly:
    parameter: system_frequency
    value_outside: [49.5, 50.5]
  condition: selection_freq_anomaly
  falsepositives:
    - Generator trip events
    - Large load rejection
    - Grid islanding events
level: critical

import struct
import datetime
from collections import defaultdict

def parse_modbus_pcap(pcap_file):
    from scapy.all import rdpcap, TCP
    packets = rdpcap(pcap_file)
    modbus_events = []
    function_code_names = {
        1: 'Read Coils', 2: 'Read Discrete Inputs',
        3: 'Read Holding Registers', 4: 'Read Input Registers',
        5: 'Write Single Coil', 6: 'Write Single Register',
        15: 'Write Multiple Coils', 16: 'Write Multiple Registers'
    }
    for pkt in packets:
        if pkt.haslayer(TCP) and pkt.dport == 502:
            payload = bytes(pkt[TCP].payload)
            if len(payload) >= 12:
                unit_id = payload[6]
                func_code = payload[7]
                event = {
                    'timestamp': str(pkt.time),
                    'src_ip': pkt[1].src,
                    'dst_ip': pkt[1].dst,
                    'unit_id': unit_id,
                    'function_code': func_code,
                    'function_name': function_code_names.get(func_code, f'Unknown({func_code})'),
                    'is_write': func_code in [5, 6, 15, 16]
                }
                modbus_events.append(event)
    return modbus_events

def analyze_modbus_anomalies(events):
    write_ops_by_ip = defaultdict(list)
    scan_candidates = defaultdict(lambda: defaultdict(int))
    for event in events:
        if event['is_write']:
            write_ops_by_ip[event['src_ip']].append(event)
        scan_candidates[event['src_ip']][event['function_code']] += 1
    alerts = []
    for ip, writes in write_ops_by_ip.items():
        if len(writes) > 10:
            alerts.append({
                'type': 'Excessive Write Operations',
                'severity': 'HIGH',
                'src_ip': ip,
                'count': len(writes),
                'time_range': f"{writes[0]['timestamp']} - {writes[-1]['timestamp']}"
            })
    for ip, fc_dict in scan_candidates.items():
        if len(fc_dict) >= 6:
            alerts.append({
                'type': 'Function Code Enumeration',
                'severity': 'MEDIUM',
                'src_ip': ip,
                'unique_function_codes': list(fc_dict.keys())
            })
    return alerts

events = parse_modbus_pcap('capture.pcap')
anomalies = analyze_modbus_anomalies(events)
for alert in anomalies:
    print(f"[{alert['severity']}] {alert['type']}: {alert}")

cat > /opt/zeek/site/ics-baseline.zeek ====

module ICS;

export {
    redef enum Log::ID += { PROTOCOL_LOG };

    type Connection: record {
        ts:           time    &log;
        uid:          string  &log;
        id:           conn::id &log;
        proto:        string  &log &optional;
        modbus_fc:    count   &log &optional;
        modbus_unit:  count   &log &optional;
        is_write_op:  bool    &log &default=F;
    };
}

event tcp_packet(c: connection, is_orig: bool, flags: string, seq: count, ack: count, len: count, payload: string)
{
    if (c$id$resp_p == 502/tcp && len > 7) {
        local unit_id = payload[6];
        local func_code = payload[7];
        local is_write = func_code in set(5, 6, 15, 16);

        if (is_write) {
            NOTICE([
                $note=ICS::WriteOperationDetected,
                $msg=fmt("Modbus write operation from %s to %s (FC:%d, Unit:%d)",
                         c$id$orig_h, c$id$resp_h, func_code, unit_id),
                $conn=c
            ]);
        }
    }
}

cat > /etc/rsyslog.d/scada_audit.conf << 'RULES'
$ModLoad imudp
$UDPServerAddress 0.0.0.0
$UDPServerRun 514
local0.* /var/log/scada/scada_audit.log
local1.* /var/log/scada/protected_action.log
local2.* /var/log/scada/operator_action.log
& stop
RULES
systemctl restart rsyslog