OT/ICS Security Mastery: Advanced Monitoring for Industrial Control Systems with Wazuh#

1
# Industrial Cyber Security Framework
2
class IndustrialCyberSecurityFramework:
3
    def __init__(self):
4
        self.industrial_kill_chain = {
5
            'stage_1_reconnaissance': {
6
                'techniques': [
7
                    'network_scanning',
8
                    'protocol_enumeration',
9
                    'hmi_reconnaissance',
10
                    'vendor_identification'
11
                ],
12
                'detection_difficulty': 'medium',
13
                'business_impact': 'low'
14
            },
15
            'stage_2_initial_access': {
16
                'techniques': [
17
                    'spear_phishing_engineering',
18
                    'removable_media_infection',
19
                    'remote_access_compromise',
20
                    'supply_chain_infiltration'
21
                ],
22
                'detection_difficulty': 'high',
23
                'business_impact': 'medium'
24
            },
25
            'stage_3_lateral_movement': {
26
                'techniques': [
27
                    'credential_harvesting',
28
                    'protocol_exploitation',
29
                    'network_traversal',
30
                    'trust_relationship_abuse'
31
                ],
32
                'detection_difficulty': 'medium',
33
                'business_impact': 'high'
34
            },
35
            'stage_4_persistence': {
36
                'techniques': [
37
                    'firmware_modification',
38
                    'configuration_changes',
39
                    'backdoor_installation',
40
                    'legitimate_tool_abuse'
41
                ],
42
                'detection_difficulty': 'high',
43
                'business_impact': 'high'
44
            },
45
            'stage_5_evasion': {
46
                'techniques': [
47
                    'legitimate_protocol_abuse',
48
                    'timing_manipulation',
49
                    'process_mimicry',
50
                    'logging_disruption'
51
                ],
52
                'detection_difficulty': 'very_high',
53
                'business_impact': 'medium'
54
            },
55
            'stage_6_collection': {
56
                'techniques': [
57
                    'process_data_harvesting',
58
                    'configuration_extraction',
59
                    'credential_dumping',
60
                    'system_information_gathering'
61
                ],
62
                'detection_difficulty': 'medium',
63
                'business_impact': 'high'
64
            },
65
            'stage_7_command_and_control': {
66
                'techniques': [
67
                    'covert_channel_establishment',
68
                    'protocol_tunneling',
69
                    'legitimate_service_abuse',
70
                    'periodic_beaconing'
71
                ],
72
                'detection_difficulty': 'high',
73
                'business_impact': 'high'
74
            },
75
            'stage_8_impact': {
76
                'techniques': [
77
                    'process_manipulation',
78
                    'safety_system_compromise',
79
                    'data_destruction',
80
                    'physical_damage'
81
                ],
82
                'detection_difficulty': 'low',
83
                'business_impact': 'critical'
84
            }
85
        }
86
        self.ot_protocols = self.initialize_ot_protocols()
87
        self.safety_systems = SafetySystemMonitor()
88

89
    def assess_industrial_security_posture(self, ot_environment):
90
        """Assess OT/ICS security posture across kill chain stages"""
91
        security_assessment = {
92
            'overall_risk_score': 0,
93
            'stage_vulnerabilities': {},
94
            'critical_gaps': [],
95
            'safety_impact_analysis': {},
96
            'recommended_controls': []
97
        }
98

99
        # Analyze each kill chain stage
100
        for stage, stage_config in self.industrial_kill_chain.items():
101
            stage_analysis = self.analyze_kill_chain_stage(
102
                stage,
103
                stage_config,
104
                ot_environment
105
            )
106

107
            security_assessment['stage_vulnerabilities'][stage] = stage_analysis
108

109
            # Weight by business impact and detection difficulty
110
            impact_weight = self.get_impact_weight(stage_config['business_impact'])
111
            detection_weight = self.get_detection_weight(stage_config['detection_difficulty'])
112

113
            stage_risk = stage_analysis['vulnerability_score'] * impact_weight * detection_weight
114
            security_assessment['overall_risk_score'] += stage_risk
115

116
            # Identify critical gaps
117
            if stage_analysis['vulnerability_score'] > 0.7 and impact_weight > 0.8:
118
                security_assessment['critical_gaps'].append({
119
                    'stage': stage,
120
                    'vulnerability_score': stage_analysis['vulnerability_score'],
121
                    'impact': stage_config['business_impact'],
122
                    'urgent_actions': stage_analysis['urgent_actions']
123
                })
124

125
        # Safety system impact analysis
126
        security_assessment['safety_impact_analysis'] = self.safety_systems.analyze_safety_impact(
127
            security_assessment['stage_vulnerabilities']
128
        )
129

130
        # Generate prioritized recommendations
131
        security_assessment['recommended_controls'] = self.generate_ot_security_recommendations(
132
            security_assessment
133
        )
134

135
        return security_assessment
136

137
    def initialize_ot_protocols(self):
138
        """Initialize OT protocol monitoring capabilities"""
139
        return {
140
            'modbus': {
141
                'port': 502,
142
                'functions': ['read_coils', 'read_discrete_inputs', 'read_holding_registers', 'write_single_coil'],
143
                'security_risks': ['unauthorized_commands', 'data_manipulation', 'device_enumeration'],
144
                'monitoring_focus': ['function_code_anomalies', 'unauthorized_writes', 'timing_attacks']
145
            },
146
            'dnp3': {
147
                'port': 20000,
148
                'functions': ['data_polling', 'control_operations', 'time_synchronization'],
149
                'security_risks': ['spoofed_responses', 'replay_attacks', 'unauthorized_control'],
150
                'monitoring_focus': ['authentication_bypass', 'unsolicited_responses', 'control_anomalies']
151
            },
152
            'iec61850': {
153
                'port': 102,
154
                'functions': ['goose_messaging', 'sampled_values', 'mms_communication'],
155
                'security_risks': ['goose_spoofing', 'sv_manipulation', 'configuration_tampering'],
156
                'monitoring_focus': ['goose_anomalies', 'sv_timing_attacks', 'unauthorized_configuration']
157
            },
158
            'opcua': {
159
                'port': 4840,
160
                'functions': ['data_access', 'alarms_events', 'historical_data'],
161
                'security_risks': ['certificate_attacks', 'session_hijacking', 'data_tampering'],
162
                'monitoring_focus': ['certificate_validation', 'session_anomalies', 'data_integrity']
163
            },
164
            'ethernet_ip': {
165
                'port': 44818,
166
                'functions': ['implicit_messaging', 'explicit_messaging', 'io_data'],
167
                'security_risks': ['cip_attacks', 'configuration_changes', 'io_manipulation'],
168
                'monitoring_focus': ['cip_anomalies', 'unauthorized_configuration', 'io_data_tampering']
169
            }
170
        }

1
<!-- OT/ICS Protocol Monitoring Rules -->
2
<group name="ot_ics_security,protocol_monitoring">
3
  <!-- Modbus Unauthorized Write -->
4
  <rule id="850001" level="14">
5
    <if_sid>86001</if_sid>
6
    <field name="protocol">modbus</field>
7
    <field name="modbus.function_code">05,06,15,16</field>
8
    <field name="modbus.authorized_source" negate="yes">true</field>
9
    <description>OT Security Critical: Unauthorized Modbus write command from untrusted source</description>
10
    <group>ot_security,modbus_unauthorized_write</group>
11
    <mitre>
12
      <id>T0832</id>
13
    </mitre>
14
  </rule>
15

16
  <!-- DNP3 Authentication Bypass -->
17
  <rule id="850002" level="15">
18
    <if_sid>86001</if_sid>
19
    <field name="protocol">dnp3</field>
20
    <field name="dnp3.auth_enabled">true</field>
21
    <field name="dnp3.auth_success">false</field>
22
    <field name="dnp3.command_executed">true</field>
23
    <description>OT Security Critical: DNP3 command executed without authentication</description>
24
    <group>ot_security,dnp3_auth_bypass</group>
25
  </rule>
26

27
  <!-- IEC 61850 GOOSE Spoofing -->
28
  <rule id="850003" level="13">
29
    <if_sid>86001</if_sid>
30
    <field name="protocol">iec61850</field>
31
    <field name="iec61850.message_type">goose</field>
32
    <field name="iec61850.sequence_number" type="pcre2">^(0|[1-9]\d*)$</field>
33
    <field name="iec61850.expected_sequence" negate="yes">match</field>
34
    <description>OT Security Alert: IEC 61850 GOOSE message sequence anomaly detected</description>
35
    <group>ot_security,goose_spoofing</group>
36
  </rule>
37

38
  <!-- OPC UA Certificate Violation -->
39
  <rule id="850004" level="12">
40
    <if_sid>86001</if_sid>
41
    <field name="protocol">opcua</field>
42
    <field name="opcua.certificate_valid">false</field>
43
    <field name="opcua.connection_established">true</field>
44
    <description>OT Security Alert: OPC UA connection with invalid certificate</description>
45
    <group>ot_security,opcua_certificate</group>
46
  </rule>
47

48
  <!-- EtherNet/IP CIP Anomaly -->
49
  <rule id="850005" level="11">
50
    <if_sid>86001</if_sid>
51
    <field name="protocol">ethernet_ip</field>
52
    <field name="cip.service_code" type="pcre2">^(0x[8-9a-fA-F][0-9a-fA-F]|0x[a-fA-F][0-9a-fA-F])$</field>
53
    <description>OT Security Alert: EtherNet/IP unusual CIP service code detected</description>
54
    <group>ot_security,cip_anomaly</group>
55
  </rule>
56

57
  <!-- Cross-Protocol Communication -->
58
  <rule id="850006" level="10" frequency="3" timeframe="300">
59
    <if_sid>850001,850002,850003,850004,850005</if_sid>
60
    <same_source_ip />
61
    <different_protocol />
62
    <description>OT Security Alert: Multi-protocol reconnaissance from single source</description>
63
    <group>ot_security,protocol_scanning</group>
64
  </rule>
65
</group>

1
class OTProtocolAnalyzer:
2
    def __init__(self):
3
        self.protocol_parsers = {
4
            'modbus': ModbusProtocolParser(),
5
            'dnp3': DNP3ProtocolParser(),
6
            'iec61850': IEC61850ProtocolParser(),
7
            'opcua': OPCUAProtocolParser(),
8
            'ethernet_ip': EtherNetIPProtocolParser()
9
        }
10
        self.behavioral_baselines = {}
11
        self.anomaly_detectors = {}
12

13
    def analyze_ot_traffic(self, network_packet):
14
        """Analyze OT network traffic for security anomalies"""
15
        analysis_result = {
16
            'packet_id': network_packet['id'],
17
            'timestamp': network_packet['timestamp'],
18
            'protocol_identified': None,
19
            'security_findings': [],
20
            'anomaly_score': 0,
21
            'risk_level': 'low'
22
        }
23

24
        # Identify OT protocol
25
        protocol = self.identify_ot_protocol(network_packet)
26
        analysis_result['protocol_identified'] = protocol
27

28
        if protocol and protocol in self.protocol_parsers:
29
            parser = self.protocol_parsers[protocol]
30

31
            # Parse protocol-specific content
32
            parsed_data = parser.parse_packet(network_packet)
33

34
            # Analyze against baseline behavior
35
            if protocol in self.behavioral_baselines:
36
                baseline_analysis = self.analyze_against_baseline(
37
                    parsed_data,
38
                    self.behavioral_baselines[protocol]
39
                )
40

41
                if baseline_analysis['anomalous']:
42
                    analysis_result['security_findings'].append({
43
                        'type': 'behavioral_anomaly',
44
                        'protocol': protocol,
45
                        'details': baseline_analysis['details'],
46
                        'severity': baseline_analysis['severity']
47
                    })
48
                    analysis_result['anomaly_score'] += baseline_analysis['score']
49

50
            # Protocol-specific security checks
51
            security_checks = parser.perform_security_analysis(parsed_data)
52
            analysis_result['security_findings'].extend(security_checks)
53

54
            # Calculate risk level
55
            analysis_result['risk_level'] = self.calculate_risk_level(
56
                analysis_result['security_findings'],
57
                analysis_result['anomaly_score']
58
            )
59

60
        return analysis_result
61

62
    def build_protocol_baseline(self, historical_traffic, protocol, days=30):
63
        """Build behavioral baseline for specific OT protocol"""
64
        if protocol not in self.protocol_parsers:
65
            raise ValueError(f"Unsupported protocol: {protocol}")
66

67
        parser = self.protocol_parsers[protocol]
68
        baseline_data = {
69
            'protocol': protocol,
70
            'analysis_period': days,
71
            'total_packets': 0,
72
            'function_patterns': {},
73
            'timing_patterns': {},
74
            'data_patterns': {},
75
            'communication_patterns': {},
76
            'anomaly_threshold': 0.05
77
        }
78

79
        parsed_packets = []
80

81
        # Parse historical traffic
82
        for packet in historical_traffic:
83
            if self.identify_ot_protocol(packet) == protocol:
84
                parsed = parser.parse_packet(packet)
85
                if parsed:
86
                    parsed_packets.append(parsed)
87
                    baseline_data['total_packets'] += 1
88

89
        if not parsed_packets:
90
            return None
91

92
        # Analyze function code patterns
93
        function_codes = [p.get('function_code') for p in parsed_packets if 'function_code' in p]
94
        baseline_data['function_patterns'] = self.analyze_function_patterns(function_codes)
95

96
        # Analyze timing patterns
97
        timestamps = [p.get('timestamp') for p in parsed_packets if 'timestamp' in p]
98
        baseline_data['timing_patterns'] = self.analyze_timing_patterns(timestamps)
99

100
        # Analyze data patterns
101
        data_values = [p.get('data_values') for p in parsed_packets if 'data_values' in p]
102
        baseline_data['data_patterns'] = self.analyze_data_patterns(data_values)
103

104
        # Analyze communication patterns
105
        comm_pairs = [(p.get('source'), p.get('destination')) for p in parsed_packets]
106
        baseline_data['communication_patterns'] = self.analyze_communication_patterns(comm_pairs)
107

108
        # Store baseline
109
        self.behavioral_baselines[protocol] = baseline_data
110

111
        return baseline_data
112

113
class ModbusProtocolParser:
114
    def __init__(self):
115
        self.function_codes = {
116
            1: 'read_coils',
117
            2: 'read_discrete_inputs',
118
            3: 'read_holding_registers',
119
            4: 'read_input_registers',
120
            5: 'write_single_coil',
121
            6: 'write_single_register',
122
            15: 'write_multiple_coils',
123
            16: 'write_multiple_registers'
124
        }
125

126
    def parse_packet(self, packet):
127
        """Parse Modbus packet for security analysis"""
128
        try:
129
            modbus_data = packet.get('payload', {})
130

131
            parsed = {
132
                'protocol': 'modbus',
133
                'timestamp': packet['timestamp'],
134
                'source': packet['source_ip'],
135
                'destination': packet['destination_ip'],
136
                'transaction_id': modbus_data.get('transaction_id'),
137
                'unit_id': modbus_data.get('unit_id'),
138
                'function_code': modbus_data.get('function_code'),
139
                'function_name': self.function_codes.get(
140
                    modbus_data.get('function_code'),
141
                    'unknown'
142
                ),
143
                'data_values': modbus_data.get('data'),
144
                'exception_code': modbus_data.get('exception_code')
145
            }
146

147
            return parsed
148

149
        except Exception as e:
150
            logger.error(f"Failed to parse Modbus packet: {e}")
151
            return None
152

153
    def perform_security_analysis(self, parsed_data):
154
        """Perform Modbus-specific security analysis"""
155
        security_findings = []
156

157
        # Check for unauthorized write operations
158
        if parsed_data['function_code'] in [5, 6, 15, 16]:
159
            if not self.is_authorized_writer(parsed_data['source']):
160
                security_findings.append({
161
                    'type': 'unauthorized_write',
162
                    'severity': 'critical',
163
                    'description': f"Unauthorized Modbus write from {parsed_data['source']}",
164
                    'function_code': parsed_data['function_code'],
165
                    'mitigation': 'Block source IP and investigate'
166
                })
167

168
        # Check for unusual function codes
169
        if parsed_data['function_code'] not in self.function_codes:
170
            security_findings.append({
171
                'type': 'unknown_function_code',
172
                'severity': 'medium',
173
                'description': f"Unknown Modbus function code: {parsed_data['function_code']}",
174
                'mitigation': 'Investigate for protocol exploitation'
175
            })
176

177
        # Check for exception responses (potential probing)
178
        if parsed_data['exception_code']:
179
            security_findings.append({
180
                'type': 'exception_response',
181
                'severity': 'low',
182
                'description': f"Modbus exception code: {parsed_data['exception_code']}",
183
                'mitigation': 'Monitor for reconnaissance patterns'
184
            })
185

186
        return security_findings
187

188
    def is_authorized_writer(self, source_ip):
189
        """Check if source IP is authorized for write operations"""
190
        # This should be configured based on the industrial network architecture
191
        authorized_writers = [
192
            '192.168.1.10',  # HMI station
193
            '192.168.1.20',  # Engineering workstation
194
            '192.168.1.30'   # SCADA server
195
        ]
196

197
        return source_ip in authorized_writers

1
class SafetySystemMonitor:
2
    def __init__(self):
3
        self.safety_systems = {
4
            'sis': SafetyInstrumentedSystem(),
5
            'fire_gas': FireGasSystem(),
6
            'emergency_shutdown': EmergencyShutdownSystem(),
7
            'process_safety': ProcessSafetySystem()
8
        }
9
        self.safety_integrity_levels = {
10
            'SIL4': {'pfd': 1e-5, 'priority': 'critical'},
11
            'SIL3': {'pfd': 1e-4, 'priority': 'high'},
12
            'SIL2': {'pfd': 1e-3, 'priority': 'medium'},
13
            'SIL1': {'pfd': 1e-2, 'priority': 'low'}
14
        }
15

16
    def monitor_safety_functions(self, ot_data):
17
        """Monitor critical safety functions for security compromise"""
18
        safety_analysis = {
19
            'timestamp': datetime.now(),
20
            'safety_status': 'normal',
21
            'compromised_functions': [],
22
            'integrity_violations': [],
23
            'safety_recommendations': []
24
        }
25

26
        # Analyze each safety system
27
        for system_name, system in self.safety_systems.items():
28
            system_analysis = self.analyze_safety_system_security(
29
                system,
30
                ot_data
31
            )
32

33
            if system_analysis['compromised']:
34
                safety_analysis['safety_status'] = 'compromised'
35
                safety_analysis['compromised_functions'].append({
36
                    'system': system_name,
37
                    'functions': system_analysis['affected_functions'],
38
                    'severity': system_analysis['severity'],
39
                    'impact': system_analysis['safety_impact']
40
                })
41

42
            # Check safety integrity level violations
43
            sil_violations = self.check_sil_violations(system, system_analysis)
44
            if sil_violations:
45
                safety_analysis['integrity_violations'].extend(sil_violations)
46

47
        # Generate safety-specific recommendations
48
        if safety_analysis['compromised_functions']:
49
            safety_analysis['safety_recommendations'] = self.generate_safety_recommendations(
50
                safety_analysis['compromised_functions']
51
            )
52

53
        return safety_analysis
54

55
    def analyze_safety_system_security(self, safety_system, ot_data):
56
        """Analyze security status of individual safety system"""
57
        analysis = {
58
            'system_id': safety_system.system_id,
59
            'compromised': False,
60
            'affected_functions': [],
61
            'severity': 'low',
62
            'safety_impact': 'none',
63
            'evidence': []
64
        }
65

66
        # Check for unauthorized access to safety functions
67
        safety_commands = [
68
            event for event in ot_data
69
            if event.get('target_system') == safety_system.system_id
70
        ]
71

72
        for command in safety_commands:
73
            # Verify command authorization
74
            if not self.verify_safety_command_authorization(command, safety_system):
75
                analysis['compromised'] = True
76
                analysis['affected_functions'].append(command.get('function'))
77
                analysis['evidence'].append({
78
                    'type': 'unauthorized_safety_command',
79
                    'command': command,
80
                    'timestamp': command.get('timestamp')
81
                })
82

83
            # Check for safety system bypass attempts
84
            if self.detect_safety_bypass_attempt(command, safety_system):
85
                analysis['compromised'] = True
86
                analysis['severity'] = 'critical'
87
                analysis['safety_impact'] = 'high'
88
                analysis['evidence'].append({
89
                    'type': 'safety_bypass_attempt',
90
                    'command': command,
91
                    'bypass_method': self.identify_bypass_method(command)
92
                })
93

94
        # Analyze safety system communication patterns
95
        comm_analysis = self.analyze_safety_communications(safety_system, ot_data)
96
        if comm_analysis['anomalous']:
97
            analysis['compromised'] = True
98
            analysis['evidence'].extend(comm_analysis['anomalies'])
99

100
        return analysis
101

102
    def detect_safety_bypass_attempt(self, command, safety_system):
103
        """Detect attempts to bypass safety systems"""
104
        bypass_indicators = [
105
            'force_override',
106
            'bypass_enable',
107
            'safety_disable',
108
            'maintenance_mode',
109
            'test_mode_permanent'
110
        ]
111

112
        command_text = str(command.get('command_data', '')).lower()
113

114
        # Check for bypass keywords
115
        if any(indicator in command_text for indicator in bypass_indicators):
116
            # Verify if bypass is authorized
117
            if not self.is_authorized_bypass(command, safety_system):
118
                return True
119

120
        # Check for configuration changes that could disable safety functions
121
        if command.get('function_type') == 'configuration_change':
122
            config_changes = command.get('configuration_data', {})
123

124
            # Look for safety-critical parameter changes
125
            safety_params = ['trip_setpoint', 'delay_time', 'voting_logic', 'enable_status']
126

127
            for param in safety_params:
128
                if param in config_changes:
129
                    # Verify change is within safe operating limits
130
                    if not self.verify_safety_parameter_limits(param, config_changes[param]):
131
                        return True
132

133
        return False
134

135
    def is_authorized_bypass(self, command, safety_system):
136
        """Verify if safety system bypass is authorized"""
137
        # Check authorization requirements for safety bypasses
138
        required_authorizations = safety_system.get_bypass_authorization_requirements()
139

140
        command_authorizations = command.get('authorizations', [])
141

142
        # Verify all required authorizations are present
143
        for required_auth in required_authorizations:
144
            if not any(auth['type'] == required_auth for auth in command_authorizations):
145
                return False
146

147
        # Check authorization validity
148
        for auth in command_authorizations:
149
            if not self.verify_authorization_validity(auth):
150
                return False
151

152
        # Check if bypass duration is within limits
153
        bypass_duration = command.get('bypass_duration', 0)
154
        max_allowed_duration = safety_system.get_max_bypass_duration()
155

156
        if bypass_duration > max_allowed_duration:
157
            return False
158

159
        return True

1
<!-- Safety System Monitoring Rules -->
2
<group name="ot_safety_systems">
3
  <!-- Safety System Bypass -->
4
  <rule id="850010" level="15">
5
    <if_sid>86001</if_sid>
6
    <field name="system_type">safety_instrumented</field>
7
    <field name="command_type">bypass_enable</field>
8
    <field name="authorization_level" negate="yes">safety_engineer</field>
9
    <description>OT Safety Critical: Unauthorized safety system bypass attempt</description>
10
    <group>ot_safety,unauthorized_bypass</group>
11
  </rule>
12

13
  <!-- Safety Setpoint Manipulation -->
14
  <rule id="850011" level="14">
15
    <if_sid>86001</if_sid>
16
    <field name="parameter_type">trip_setpoint</field>
17
    <field name="value_change" compare=">">10</field>
18
    <field name="safety_approval">false</field>
19
    <description>OT Safety Critical: Safety trip setpoint modified without approval</description>
20
    <group>ot_safety,setpoint_manipulation</group>
21
  </rule>
22

23
  <!-- Emergency Shutdown System Compromise -->
24
  <rule id="850012" level="15">
25
    <if_sid>86001</if_sid>
26
    <field name="system_type">emergency_shutdown</field>
27
    <field name="function">disable</field>
28
    <description>OT Safety Critical: Emergency shutdown system disabled</description>
29
    <group>ot_safety,esd_compromise</group>
30
  </rule>
31

32
  <!-- Fire & Gas System Tampering -->
33
  <rule id="850013" level="14">
34
    <if_sid>86001</if_sid>
35
    <field name="system_type">fire_gas</field>
36
    <field name="detector_status">disabled</field>
37
    <field name="maintenance_mode">false</field>
38
    <description>OT Safety Alert: Fire & gas detector disabled outside maintenance</description>
39
    <group>ot_safety,fire_gas_tampering</group>
40
  </rule>
41

42
  <!-- Safety Logic Solver Anomaly -->
43
  <rule id="850014" level="13">
44
    <if_sid>86001</if_sid>
45
    <field name="device_type">safety_logic_solver</field>
46
    <field name="diagnostic_alarm">true</field>
47
    <field name="alarm_type">security_violation</field>
48
    <description>OT Safety Alert: Safety logic solver security violation</description>
49
    <group>ot_safety,sls_security</group>
50
  </rule>
51
</group>

1
class OTAssetDiscoveryEngine:
2
    def __init__(self):
3
        self.discovery_methods = {
4
            'passive_monitoring': PassiveNetworkDiscovery(),
5
            'protocol_scanning': ProtocolBasedScanning(),
6
            'device_enumeration': DeviceEnumeration(),
7
            'firmware_fingerprinting': FirmwareFingerprintingService()
8
        }
9
        self.asset_classifier = OTAssetClassifier()
10
        self.vulnerability_scanner = OTVulnerabilityScanner()
11

12
    def discover_ot_assets(self, network_ranges):
13
        """Comprehensive OT asset discovery and inventory"""
14
        discovery_results = {
15
            'discovered_assets': [],
16
            'asset_categories': {},
17
            'security_assessment': {},
18
            'network_topology': {},
19
            'recommendations': []
20
        }
21

22
        # Execute discovery methods
23
        all_discovered_assets = []
24

25
        for method_name, discovery_method in self.discovery_methods.items():
26
            try:
27
                method_results = discovery_method.discover(network_ranges)
28
                all_discovered_assets.extend(method_results)
29

30
                logger.info(f"{method_name} discovered {len(method_results)} assets")
31

32
            except Exception as e:
33
                logger.error(f"Discovery method {method_name} failed: {e}")
34

35
        # Deduplicate and merge asset information
36
        unique_assets = self.merge_asset_information(all_discovered_assets)
37

38
        # Classify assets
39
        for asset in unique_assets:
40
            classification = self.asset_classifier.classify_asset(asset)
41
            asset.update(classification)
42

43
        discovery_results['discovered_assets'] = unique_assets
44

45
        # Categorize assets
46
        discovery_results['asset_categories'] = self.categorize_assets(unique_assets)
47

48
        # Perform security assessment
49
        discovery_results['security_assessment'] = self.assess_asset_security(unique_assets)
50

51
        # Map network topology
52
        discovery_results['network_topology'] = self.map_network_topology(unique_assets)
53

54
        # Generate recommendations
55
        discovery_results['recommendations'] = self.generate_asset_security_recommendations(
56
            discovery_results
57
        )
58

59
        return discovery_results
60

61
    def classify_asset(self, asset):
62
        """Classify OT asset based on discovered characteristics"""
63
        classification = {
64
            'asset_type': 'unknown',
65
            'criticality': 'medium',
66
            'safety_impact': 'low',
67
            'function': 'unknown',
68
            'vendor': 'unknown',
69
            'model': 'unknown',
70
            'firmware_version': 'unknown'
71
        }
72

73
        # Identify asset type based on protocols and services
74
        protocols = asset.get('supported_protocols', [])
75
        services = asset.get('services', [])
76

77
        # HMI identification
78
        if any(port in services for port in [3389, 5900, 5901]):
79
            if any(proto in protocols for proto in ['modbus', 'opcua', 'dnp3']):
80
                classification['asset_type'] = 'hmi'
81
                classification['criticality'] = 'high'
82

83
        # PLC identification
84
        elif 'modbus' in protocols or 'ethernet_ip' in protocols:
85
            classification['asset_type'] = 'plc'
86
            classification['criticality'] = 'critical'
87
            classification['safety_impact'] = 'high'
88

89
        # RTU identification
90
        elif 'dnp3' in protocols or 'modbus_rtu' in protocols:
91
            classification['asset_type'] = 'rtu'
92
            classification['criticality'] = 'high'
93

94
        # SCADA server identification
95
        elif len(protocols) > 3 and 'database' in services:
96
            classification['asset_type'] = 'scada_server'
97
            classification['criticality'] = 'critical'
98

99
        # Engineering workstation
100
        elif 'windows' in asset.get('os_family', '').lower():
101
            if any(proto in protocols for proto in ['modbus', 'opcua']):
102
                classification['asset_type'] = 'engineering_workstation'
103
                classification['criticality'] = 'high'
104

105
        # IED identification
106
        elif 'iec61850' in protocols or 'goose' in protocols:
107
            classification['asset_type'] = 'ied'
108
            classification['criticality'] = 'high'
109
            classification['safety_impact'] = 'medium'
110

111
        # Safety system identification
112
        if any(keyword in asset.get('device_description', '').lower()
113
               for keyword in ['safety', 'sis', 'emergency', 'fire', 'gas']):
114
            classification['safety_impact'] = 'critical'
115
            classification['criticality'] = 'critical'
116

117
        # Extract vendor and model information
118
        device_info = asset.get('device_identification', {})
119
        classification['vendor'] = device_info.get('vendor', 'unknown')
120
        classification['model'] = device_info.get('model', 'unknown')
121
        classification['firmware_version'] = device_info.get('firmware_version', 'unknown')
122

123
        return classification
124

125
    def assess_asset_security(self, assets):
126
        """Assess security posture of discovered OT assets"""
127
        security_assessment = {
128
            'total_assets': len(assets),
129
            'vulnerability_summary': {},
130
            'risk_distribution': {'critical': 0, 'high': 0, 'medium': 0, 'low': 0},
131
            'common_vulnerabilities': [],
132
            'security_recommendations': []
133
        }
134

135
        # Scan assets for vulnerabilities
136
        for asset in assets:
137
            vuln_scan_result = self.vulnerability_scanner.scan_asset(asset)
138
            asset['vulnerabilities'] = vuln_scan_result
139

140
            # Update risk distribution
141
            if vuln_scan_result['risk_level']:
142
                security_assessment['risk_distribution'][vuln_scan_result['risk_level']] += 1
143

144
        # Identify common vulnerabilities
145
        vuln_counts = {}
146
        for asset in assets:
147
            for vuln in asset.get('vulnerabilities', {}).get('vulnerabilities', []):
148
                vuln_id = vuln.get('cve_id') or vuln.get('title')
149
                vuln_counts[vuln_id] = vuln_counts.get(vuln_id, 0) + 1
150

151
        # Sort by frequency
152
        common_vulns = sorted(vuln_counts.items(), key=lambda x: x[1], reverse=True)[:10]
153
        security_assessment['common_vulnerabilities'] = [
154
            {'vulnerability': vuln, 'affected_assets': count}
155
            for vuln, count in common_vulns
156
        ]
157

158
        return security_assessment

1
class OTBehavioralAnalytics:
2
    def __init__(self):
3
        self.process_models = {}
4
        self.anomaly_detectors = {}
5
        self.baseline_period = 30  # days
6

7
    def build_process_behavioral_model(self, process_data, process_id):
8
        """Build behavioral model for industrial process"""
9
        model = {
10
            'process_id': process_id,
11
            'model_type': 'industrial_process',
12
            'parameters': {},
13
            'operating_ranges': {},
14
            'control_patterns': {},
15
            'alarm_patterns': {},
16
            'maintenance_patterns': {}
17
        }
18

19
        # Extract process variables
20
        process_variables = self.extract_process_variables(process_data)
21

22
        # Analyze operating ranges
23
        for var_name, var_data in process_variables.items():
24
            model['operating_ranges'][var_name] = {
25
                'min': np.min(var_data),
26
                'max': np.max(var_data),
27
                'mean': np.mean(var_data),
28
                'std': np.std(var_data),
29
                'percentiles': {
30
                    'p5': np.percentile(var_data, 5),
31
                    'p95': np.percentile(var_data, 95),
32
                    'p99': np.percentile(var_data, 99)
33
                }
34
            }
35

36
        # Analyze control patterns
37
        control_events = [event for event in process_data if event.get('event_type') == 'control']
38
        model['control_patterns'] = self.analyze_control_patterns(control_events)
39

40
        # Analyze alarm patterns
41
        alarm_events = [event for event in process_data if event.get('event_type') == 'alarm']
42
        model['alarm_patterns'] = self.analyze_alarm_patterns(alarm_events)
43

44
        # Build anomaly detection model
45
        features = self.extract_behavioral_features(process_data)
46
        anomaly_model = IsolationForest(contamination=0.1, random_state=42)
47
        anomaly_model.fit(features)
48

49
        self.process_models[process_id] = model
50
        self.anomaly_detectors[process_id] = anomaly_model
51

52
        return model
53

54
    def detect_process_anomalies(self, current_data, process_id):
55
        """Detect anomalies in industrial process behavior"""
56
        if process_id not in self.process_models:
57
            return {'error': f'No model available for process {process_id}'}
58

59
        model = self.process_models[process_id]
60
        anomaly_detector = self.anomaly_detectors[process_id]
61

62
        anomaly_analysis = {
63
            'process_id': process_id,
64
            'timestamp': datetime.now(),
65
            'anomalies_detected': [],
66
            'overall_anomaly_score': 0,
67
            'process_status': 'normal'
68
        }
69

70
        # Extract current features
71
        current_features = self.extract_behavioral_features([current_data])
72

73
        # Detect statistical anomalies
74
        anomaly_scores = anomaly_detector.decision_function(current_features)
75
        predictions = anomaly_detector.predict(current_features)
76

77
        if predictions[0] == -1:  # Anomaly detected
78
            anomaly_analysis['anomalies_detected'].append({
79
                'type': 'behavioral_anomaly',
80
                'score': float(anomaly_scores[0]),
81
                'description': 'Process behavior deviates from established baseline',
82
                'severity': self.calculate_anomaly_severity(anomaly_scores[0])
83
            })
84

85
        # Check for parameter range violations
86
        process_variables = self.extract_process_variables([current_data])
87

88
        for var_name, current_value in process_variables.items():
89
            if var_name in model['operating_ranges']:
90
                operating_range = model['operating_ranges'][var_name]
91

92
                # Check if value is outside normal operating range
93
                if (current_value < operating_range['percentiles']['p5'] or
94
                    current_value > operating_range['percentiles']['p95']):
95

96
                    anomaly_analysis['anomalies_detected'].append({
97
                        'type': 'parameter_range_violation',
98
                        'variable': var_name,
99
                        'current_value': current_value,
100
                        'normal_range': {
101
                            'min': operating_range['percentiles']['p5'],
102
                            'max': operating_range['percentiles']['p95']
103
                        },
104
                        'severity': 'medium' if current_value < operating_range['percentiles']['p99'] else 'high'
105
                    })
106

107
        # Analyze control action patterns
108
        control_anomalies = self.detect_control_anomalies(current_data, model)
109
        anomaly_analysis['anomalies_detected'].extend(control_anomalies)
110

111
        # Calculate overall anomaly score
112
        if anomaly_analysis['anomalies_detected']:
113
            anomaly_analysis['overall_anomaly_score'] = np.mean([
114
                a.get('score', 0.5) for a in anomaly_analysis['anomalies_detected']
115
            ])
116

117
            if anomaly_analysis['overall_anomaly_score'] > 0.8:
118
                anomaly_analysis['process_status'] = 'critical_anomaly'
119
            elif anomaly_analysis['overall_anomaly_score'] > 0.6:
120
                anomaly_analysis['process_status'] = 'significant_anomaly'
121
            else:
122
                anomaly_analysis['process_status'] = 'minor_anomaly'
123

124
        return anomaly_analysis
125

126
    def detect_control_anomalies(self, current_data, model):
127
        """Detect anomalies in control actions"""
128
        control_anomalies = []
129

130
        if current_data.get('event_type') != 'control':
131
            return control_anomalies
132

133
        control_patterns = model.get('control_patterns', {})
134

135
        # Check for unusual control sequences
136
        control_action = current_data.get('control_action')
137
        if control_action:
138
            # Verify against expected control patterns
139
            expected_patterns = control_patterns.get('action_sequences', [])
140

141
            if control_action not in expected_patterns:
142
                control_anomalies.append({
143
                    'type': 'unusual_control_action',
144
                    'action': control_action,
145
                    'description': 'Control action not seen in baseline period',
146
                    'severity': 'medium',
147
                    'score': 0.7
148
                })
149

150
        # Check for control timing anomalies
151
        control_timing = current_data.get('control_timing')
152
        if control_timing:
153
            expected_timing = control_patterns.get('timing_patterns', {})
154

155
            if abs(control_timing - expected_timing.get('mean', 0)) > 3 * expected_timing.get('std', 1):
156
                control_anomalies.append({
157
                    'type': 'control_timing_anomaly',
158
                    'timing': control_timing,
159
                    'expected_range': {
160
                        'mean': expected_timing.get('mean'),
161
                        'std': expected_timing.get('std')
162
                    },
163
                    'description': 'Control action timing outside normal parameters',
164
                    'severity': 'high',
165
                    'score': 0.8
166
                })
167

168
        return control_anomalies