Merge pull request #23 from afmurillo/dev-concealment-mitm

Dev concealment mitm

Author: afmurillo

dhalsim/network_attacks/concealment_netfilter_queue.py

@@ -5,6 +5,8 @@
 import os
 import sys
 
+import pandas as pd
+
 from scapy.layers.inet import IP, TCP
 from scapy.packet import Raw
 
@@ -19,6 +21,8 @@ def __init__(self,  intermediate_yaml_path: Path, yaml_index: int, queue_number:
         self.scada_session_ids = []
         self.attack_session_ids = []
 
+        self.concealment_data_pd = pd.read_csv(self.intermediate_attack['concealment_data'])
+
     def capture(self, packet):
         """
         This function is the function that will run in the thread started in the setup function.
@@ -34,26 +38,17 @@ def capture(self, packet):
                 if len(p) == 116:
                     this_session =  int.from_bytes(p[Raw].load[4:8], sys.byteorder)
                     tag_name = p[Raw].load.decode(encoding='latin-1')[54:56]
-                    self.logger.debug('ENIP TCP Session ID: ' + str(this_session))
-                    self.logger.debug('Received tag is: ' + tag_name)
-                    self.logger.debug('Attack tag is: ' + self.attacked_tag)
                     if self.attacked_tag == tag_name:
                         # This is a packet being sent to SCADA server, conceal the manipulation
-                        self.logger.debug('Packet source: ' + p[IP].src )
-                        self.logger.debug('SCADA IP: ' + self.intermediate_yaml['scada']['public_ip'])
                         if p[IP].src == self.intermediate_yaml['scada']['public_ip']:
-                            self.logger.debug('SCADA session: ' + str(this_session))
                             self.scada_session_ids.append(this_session)
                         else:
-                            self.logger.debug('PLC session: ' + str(this_session))
                             self.attack_session_ids.append(this_session)
 
                 if len(p) == 102:
                     this_session = int.from_bytes(p[Raw].load[4:8], sys.byteorder)
                     if this_session in self.attack_session_ids:
-                        self.logger.debug('Modifying because session is: ' + str(this_session))
                         value = translate_payload_to_float(p[Raw].load)
-                        self.logger.debug('tag value is:' + str(value))
 
                         if 'value' in self.intermediate_attack.keys():
                             p[Raw].load = translate_float_to_payload(
@@ -72,8 +67,10 @@ def capture(self, packet):
 
                     elif this_session in self.scada_session_ids:
                         self.logger.debug('Concealing to SCADA: ' + str(this_session))
-                        p[Raw].load = translate_float_to_payload(
-                            self.intermediate_attack['concealment_value'], p[Raw].load)
+                        exp = (self.concealment_data_pd['iteration'] == self.get_master_clock())
+                        concealment_value = float(self.concealment_data_pd.loc[exp][self.attacked_tag].values[-1])
+                        self.logger.debug('Concealing with value: ' + str(concealment_value))
+                        p[Raw].load = translate_float_to_payload(concealment_value, p[Raw].load)
 
                         del p[IP].chksum
                         del p[TCP].chksum

dhalsim/network_attacks/mitm_netfilter_queue.py

@@ -33,19 +33,13 @@ def capture(self, packet):
                 if len(p) == 116:
                     this_session =  int.from_bytes(p[Raw].load[4:8], sys.byteorder)
                     tag_name = p[Raw].load.decode(encoding='latin-1')[54:56]
-                    self.logger.debug('ENIP TCP Session ID: ' + str(this_session))
-                    self.logger.debug('Received tag is: ' + tag_name)
-                    self.logger.debug('Attack tag is: ' + self.attacked_tag)
                     if self.attacked_tag == tag_name:
-                        self.logger.debug('Modifying tag: ' + tag_name)
                         self.session_ids.append(this_session)
 
                 if len(p) == 102:
                     this_session = int.from_bytes(p[Raw].load[4:8], sys.byteorder)
                     if this_session in self.session_ids:
-                        self.logger.debug('Modifying because session is: ' + str(this_session))
                         value = translate_payload_to_float(p[Raw].load)
-                        self.logger.debug('tag value is:' + str(value))
 
                         if 'value' in self.intermediate_attack.keys():
                             p[Raw].load = translate_float_to_payload(

dhalsim/network_attacks/mitm_netfilter_queue_subprocess.py

@@ -1,20 +1,32 @@
-from pathlib import Path
+import yaml
+import sqlite3
 import sys
 import signal
+import time
+import random
 
-from netfilterqueue import NetfilterQueue
-import yaml
 from dhalsim.py3_logger import get_logger
-
+from pathlib import Path
+from netfilterqueue import NetfilterQueue
 from abc import ABCMeta, abstractmethod
 
+class Error(Exception):
+    """Base class for exceptions in this module."""
+
+
+class DatabaseError(Error):
+    """Raised when not being able to connect to the database"""
+
 
 class PacketQueue(metaclass=ABCMeta):
     """
     Currently, the Netfilterqueue library in Python3 does not support running in threads, using blocking calls.
     We will use this class to launch a subprocess that handles the packets in the queue.
     """
 
+    DB_TRIES = 10
+    """Amount of times a db query will retry on a exception"""
+
     def __init__(self,  intermediate_yaml_path: Path, yaml_index: int, queue_number: int):
 
         signal.signal(signal.SIGINT, self.sigint_handler)
@@ -31,6 +43,7 @@ def __init__(self,  intermediate_yaml_path: Path, yaml_index: int, queue_number:
 
         # Get the attack that we are from the intermediate YAML
         self.intermediate_attack = self.intermediate_yaml["network_attacks"][self.yaml_index]
+        self.db_sleep_time = random.uniform(0.01, 0.1)
 
     def main_loop(self):
         self.logger.debug('Parent NF Class launched')
@@ -44,6 +57,60 @@ def main_loop(self):
                 self.nfqueue.unbind()
             sys.exit(0)
 
+    def db_query(self, query, write=False, parameters=None):
+        """
+        Execute a query on the database
+        On a :code:`sqlite3.OperationalError` it will retry with a max of :code:`DB_TRIES` tries.
+        Before it reties, it will sleep for :code:`DB_SLEEP_TIME` seconds.
+        This is necessary because of the limited concurrency in SQLite.
+
+        :param query: The SQL query to execute in the db
+        :type query: str
+
+        :param write: Boolean flag to indicate if this query will write into the database
+
+        :param parameters: The parameters to put in the query. This must be a tuple.
+
+        :raise DatabaseError: When a :code:`sqlite3.OperationalError` is still raised after
+           :code:`DB_TRIES` tries.
+        """
+        for i in range(self.DB_TRIES):
+            try:
+                with sqlite3.connect(self.intermediate_yaml["db_path"]) as conn:
+                    cur = conn.cursor()
+                    if parameters:
+                        cur.execute(query, parameters)
+                    else:
+                        cur.execute(query)
+                    conn.commit()
+
+                    if not write:
+                        return cur.fetchone()[0]
+                    else:
+                        return
+            except sqlite3.OperationalError as exc:
+                self.logger.info(
+                    "Failed to connect to db with exception {exc}. Trying {i} more times.".format(
+                        exc=exc, i=self.DB_TRIES - i - 1))
+                time.sleep(self.db_sleep_time)
+        self.logger.error(
+            "Failed to connect to db. Tried {i} times.".format(i=self.DB_TRIES))
+        raise DatabaseError("Failed to get master clock from database")
+
+    def get_master_clock(self):
+        """
+        Get the value of the master clock of the physical process through the database.
+        On a :code:`sqlite3.OperationalError` it will retry with a max of :code:`DB_TRIES` tries.
+        Before it reties, it will sleep for :code:`DB_SLEEP_TIME` seconds.
+
+        :return: Iteration in the physical process.
+
+        :raise DatabaseError: When a :code:`sqlite3.OperationalError` is still raised after
+           :code:`DB_TRIES` tries.
+        """
+        master_time = self.db_query("SELECT time FROM master_time WHERE id IS 1", False, None)
+        return master_time
+
     @abstractmethod
     def capture(self, pkt):
         """

dhalsim/parser/config_parser.py

@@ -239,9 +239,14 @@ class SchemaParser:
                 'value': And(
                     Or(float, And(int, Use(float)))
                 ),
-                'concealment_value': And(
-                    Or(float, And(int, Use(float)))
-                )
+                'concealment_data': And(
+                    str,
+                    Use(Path),
+                    Use(lambda p: p.absolute().parent / p),
+                    Schema(lambda l: Path.is_file, error="'inp_file' could not be found."),
+                    Schema(lambda f: f.suffix == '.csv',
+                           error="Suffix of 'inp_file' should be .inp.")
+                ),
             },
             {
                 'type': And(
@@ -630,7 +635,7 @@ def generate_device_attacks(self, yaml_data):
 
     def generate_network_attacks(self):
         """
-        This function will add device attacks to the appropriate PLCs in the intermediate yaml
+        This function will add network attacks to the appropriate PLCs in the intermediate yaml
 
         :param network_attacks: The YAML data of the network attacks
         """
@@ -661,6 +666,9 @@ def generate_network_attacks(self):
                         raise NoSuchTag(
                             f"PLC {target_plc['name']} does not have all the tags specified.")
 
+                if network_attack['type'] == 'concealment_mitm':
+                    network_attack['concealment_data'] = str(network_attack['concealment_data'])
+
             return network_attacks
         return []
 
@@ -766,6 +774,7 @@ def generate_intermediate_yaml(self):
         yaml_data = self.generate_device_attacks(yaml_data)
         yaml_data["network_attacks"] = self.generate_network_attacks()
 
+
         # Parse network events from the config file
         yaml_data["network_events"] = self.generate_network_events()
 

examples/ctown_topology/ctown_concealment_mitm.yaml

@@ -3,9 +3,9 @@ network_attacks:
   type: concealment_mitm
   tag: T3
   target: PLC4
-  value: 42.0
-  concealment_value: 84
+  value: 8.0
+  concealment_data: concealment_test.csv
   trigger:
-    start: 5
-    end: 15
+    start: 648
+    end: 936
     type: time

examples/ctown_topology/ctown_config.yaml

@@ -1,5 +1,5 @@
 inp_file: ctown_map.inp
-iterations: 20
+iterations: 2880
 network_topology_type: complex
 plcs: !include ctown_plcs.yaml