alex/distributed-network-event-detection-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

initial commit

Browse Source
This commit is contained in:
Alex
2024-03-26 16:18:13 +01:00
commit 465b3536a9
48 changed files with 11254 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
graph_based_intrusion_detection/notebooks/analysis#2.ipynb
+1642
View File
File diff suppressed because it is too large Load Diff
graph_based_intrusion_detection/notebooks/analysis.ipynb
+2339
View File
File diff suppressed because one or more lines are too long
graph_based_intrusion_detection/notebooks/annotate.ipynb
+83
View File
@@ -0,0 +1,83 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "37ac6be2-a15f-40db-ba98-7f4559cf5749",
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd"
]
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"df = pd.read_pickle('../echo-dot-activity.pickel')\n",
"\n",
"print(\"ip/ip.dst\" in df.columns)\n",
"\n",
"df['Activity'] = pd.Series(False, index=df.index)\n",
"\n",
"df.loc[2330 - 12198, 'Activity'] = True # music listening\n",
"df.loc[12744 - 13396, 'Activity'] = True # announcement\n",
"df.loc[13548 - 20533, 'Activity'] = True # key exchange? maybe not activity...\n",
"df.loc[20556 - 26757, 'Activity'] = True # music listening\n",
"df.loc[26813 - 26825, 'Activity'] = True # announcement\n",
"df.loc[26918 - 26936, 'Activity'] = True # volume adjustement \n",
"\n",
"df.to_pickle('../echo-dot-activity.pickel')\n"
],
"metadata": {
"collapsed": false
},
"id": "fe1e40dd56e78b5"
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"df = pd.read_pickle(\"../echo-dot-attack.pickel\")\n",
"\n",
"df[\"Attack\"] = pd.Series(False, index=df.index, dtype=int)\n",
"\n",
"print(\"ip/ip.dst\" in df.columns)\n",
"\n",
"df.loc[49 - 262255, \"Attack\"] = 2\n",
"df.loc[262258 - 622810, \"Attack\"] = 2\n",
"df.loc[624338 - 624363, \"Attack\"] = 1\n",
"df.loc[624541 - 627011, \"Attack\"] = 3\n",
"\n",
"df.to_pickle(\"../echo-dot-attack.pickel\")"
],
"metadata": {
"collapsed": false
},
"id": "d6a43f9140b4e1f9"
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.9"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
graph_based_intrusion_detection/notebooks/data_preparation.ipynb
+122
View File
@@ -0,0 +1,122 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "initial_id",
"metadata": {
"collapsed": true
},
"outputs": [],
"source": [
"import pandas as pd\n",
"import numpy as np"
]
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"df = pd.read_pickle(\"../echo-dot-attack.pickel\")"
],
"metadata": {
"collapsed": false
},
"id": "36cbb373e91e7b86"
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"# get unique src ip addresses, drop nans\n",
"ip_addresses = df[\"ip/ip.src\"].unique()\n",
"ip_addresses = ip_addresses[~pd.isnull(ip_addresses)]\n",
"ip_addresses"
],
"metadata": {
"collapsed": false
},
"id": "737d862bee140f41"
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"# get associated mac addresses\n",
"associated_mac_addresses = list(set(\n",
" [df[df[\"ip/ip.src\"] == ip][\"eth/eth.src\"].unique()[0] for ip in ip_addresses]))\n",
"associated_mac_addresses"
],
"metadata": {
"collapsed": false
},
"id": "f5f01e284c2fd677"
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"# get unique mac addresses, drop nans\n",
"mac_addresses = set(df[\"eth/eth.src\"].unique()).union(set(df[\"eth/eth.dst\"].unique()))\n",
"mac_addresses = [mac for mac in mac_addresses if not pd.isnull(mac)]\n",
"mac_addresses"
],
"metadata": {
"collapsed": false
},
"id": "ba1b9681e60d31b7"
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"# check if all mac addresses are associated with an ip address\n",
"for mac in mac_addresses:\n",
" if mac not in associated_mac_addresses:\n",
" print(f\"mac address {mac} is not associated with an ip address\")"
],
"metadata": {
"collapsed": false
},
"id": "f6251b2b6b4f71cf"
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"# create new columns in dataframe called ip/ip.dst with ip address associated to destination mac"
],
"metadata": {
"collapsed": false
},
"id": "1c5d264aff0955cc"
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 2
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython2",
"version": "2.7.6"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
graph_based_intrusion_detection/notebooks/eda.ipynb
+986
View File
File diff suppressed because one or more lines are too long
graph_based_intrusion_detection/notebooks/feature_extraction_v2.ipynb
+377
View File
@@ -0,0 +1,377 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": null,
"id": "55bc224f879f844a",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"import xml.etree.ElementTree\n",
"import xml.etree.ElementTree as ET\n",
"import numpy as np\n",
"import pandas as pd"
]
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [
"DEVICES_LAB = {\n",
" # 'a8:03:2a:b1:35:60': 'dev01-shelly-plus-1pm-relais', # Christoph\n",
" # '60:01:94:c7:69:ac': 'dev02-sonoff-relais', # Oliver\n",
" # 'd8:f1:5b:d8:08:0c': 'dev03-tuya-lampe', # Leonard\n",
" # '1c:d6:bd:b5:d3:bb': 'dev04-linkind-zigbee-mini-hub', # Thorsten\n",
" # '6c:5a:b0:7d:e2:25': 'dev05-tplink-tapo-l530e-birne', # Suad\n",
" # '54:af:97:7c:5e:f0': 'dev06-tapo-steckdose', # Andreas\n",
" # '74:ab:93:de:a0:7e': 'dev07-blink-sicherheitskamera', # Julian\n",
" '34:25:be:ef:91:bf': 'dev08-echo-dot-l4s3re', # Alexander\n",
" # '70:ee:50:90:64:04': 'dev09-netatmo-smart-weather-station', # Jakobus\n",
" # '68:3a:48:4b:53:c5': 'dev10-aeotec-z-wave-hub', # Andre\n",
" # 'dc:ed:83:4a:cf:76': 'dev11-aqara-presence-sensor-fp2', # Artur\n",
" # '8c:f6:81:dc:63:54': 'dev12-shelly-bewegungsmelder', # Moritz\n",
" # '24:4c:ab:43:0d:0f': 'dev13-shelly-flood', # Mohamad\n",
" # '08:b6:1f:cc:4d:c0': 'dev14-shelly-ht-temperatur-sensor', # Bastian\n",
" # '90:48:6c:17:ae:25': 'dev15-ring-door-camera' # Victor (Ring hinzugefügt, device Nummern angepasst)\n",
"}"
],
"metadata": {
"collapsed": false
},
"id": "b5715a859aec218c"
},
{
"cell_type": "code",
"execution_count": null,
"id": "39ad609a78e03fa0",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"MINIMAL_FEATURES = [\n",
" #Meta\n",
" '_ws.malformed/_ws.expert/_ws.malformed.expert',\n",
" # 'bootp/bootp.dhcp'\n",
" # 'bootp/bootp.flags.bc'\n",
" # 'bootp/bootp.flags.reserved'\n",
" # 'bootp/bootp.hops'\n",
" # 'bootp/bootp.hw.len'\n",
" # 'bootp/bootp.hw.type'\n",
" # 'bootp/bootp.secs'\n",
" # 'bootp/bootp.type'\n",
" 'classicstun.length',\n",
" 'classicstun.type',\n",
" # 'frame/frame.cap_len'\n",
" 'frame/frame.number', # Nur für Labeling\n",
" 'frame/frame.encap_type',\n",
" 'frame/frame.ignore',\n",
" 'frame/frame.len',\n",
" 'frame/frame.marked',\n",
" 'frame/frame.offset_shift',\n",
" 'frame/frame.packet_flags',\n",
" # 'geninfo/caplen'\n",
" # 'geninfo/len'\n",
" 'geninfo/timestamp',\n",
" #Data-Link\n",
" 'eth/eth.dst',\n",
" 'eth/eth.dst/eth.dst.ig',\n",
" 'eth/eth.dst/eth.dst.oui',\n",
" 'eth/eth.dst/eth.lg',\n",
" 'eth/eth.src',\n",
" 'eth/eth.src/eth.ig',\n",
" 'eth/eth.src/eth.lg',\n",
" 'eth/eth.src/eth.src.oui',\n",
" 'eth/eth.type',\n",
" 'arp/arp.hw.type', # Auffälliges Verhalten des Echo (5 Felder aus Paper zurück)\n",
" 'arp/arp.proto.type',\n",
" 'arp/arp.hw.size',\n",
" 'arp/arp.proto.size',\n",
" 'arp/arp.opcode',\n",
" #Network\n",
" 'icmp/data/data.len',\n",
" 'icmp/icmp.checksum.status',\n",
" 'icmp/icmp.code',\n",
" 'icmp/icmp.ident',\n",
" 'icmp/icmp.resp_in',\n",
" 'icmp/icmp.resp_to',\n",
" 'icmp/icmp.seq',\n",
" 'icmp/icmp.seq_le',\n",
" 'icmp/icmp.type',\n",
" 'icmp/icmp.udp/icmp.udp.dstport',\n",
" 'icmp/icmp.udp/icmp.udp.length',\n",
" 'icmp/icmp.udp/icmp.udp.srcport',\n",
" 'igmp/igmp.checksum.status',\n",
" 'igmp/igmp.max_resp',\n",
" 'igmp/igmp.maddr',\n",
" 'igmp/igmp.type',\n",
" 'ip/<>/ip.options.routeralert/ip.opt.ra',\n",
" 'ip/<>/ip.options.routeralert/ip.opt.sec_cl',\n",
" 'ip/<>/ip.options.routeralert/ip.opt.type',\n",
" 'ip/ip.checksum.status',\n",
" 'ip/ip.dsfield',\n",
" 'ip/ip.dsfield/ip.dsfield.dscp',\n",
" 'ip/ip.dsfield/ip.dsfield.ecn',\n",
" 'ip/ip.evil_packet',\n",
" 'ip/ip.flags',\n",
" 'ip/ip.flags/ip.flags.df',\n",
" 'ip/ip.flags/ip.flags.mf',\n",
" 'ip/ip.flags/ip.flags.rb',\n",
" 'ip/ip.frag_offset',\n",
" 'ip/ip.hdr_len',\n",
" 'ip/ip.id',\n",
" 'ip/ip.len',\n",
" 'ip/ip.proto',\n",
" 'ip/ip.src',\n",
" 'ip/ip.dst',\n",
" 'ip/ip.ttl',\n",
" 'ip/ip.version',\n",
" # Transport\n",
" 'tcp/tcp.ack',\n",
" 'tcp/tcp.analysis/tcp.analysis.bytes_in_flight',\n",
" 'tcp/tcp.analysis/tcp.analysis.push_bytes_sent',\n",
" 'tcp/tcp.checksum.status',\n",
" 'tcp/tcp.completeness',\n",
" 'tcp/tcp.dstport',\n",
" 'tcp/tcp.flags/tcp.flags.ack',\n",
" 'tcp/tcp.flags/tcp.flags.cwr',\n",
" 'tcp/tcp.flags/tcp.flags.ecn',\n",
" 'tcp/tcp.flags/tcp.flags.fin',\n",
" 'tcp/tcp.flags/tcp.flags.ns',\n",
" 'tcp/tcp.flags/tcp.flags.push',\n",
" 'tcp/tcp.flags/tcp.flags.res',\n",
" 'tcp/tcp.flags/tcp.flags.reset',\n",
" 'tcp/tcp.flags/tcp.flags.syn',\n",
" 'tcp/tcp.flags/tcp.flags.urg',\n",
" 'tcp/tcp.hdr_len',\n",
" 'tcp/tcp.len',\n",
" 'tcp/tcp.nxtseq',\n",
" 'tcp/tcp.options/tcp.options.mss/tcp.options.mss_val',\n",
" 'tcp/tcp.options/tcp.options.nop',\n",
" # 'tcp/tcp.payload' \n",
" 'tcp/tcp.seq',\n",
" 'tcp/tcp.srcport',\n",
" 'tcp/tcp.stream',\n",
" 'tcp/tcp.urgent_pointer',\n",
" 'tcp/tcp.window_size',\n",
" 'tcp/tcp.window_size_scalefactor',\n",
" 'tcp/tcp.window_size_value',\n",
" 'udp/udp.checksum.status',\n",
" 'udp/udp.dstport',\n",
" 'udp/udp.length',\n",
" 'udp/udp.srcport',\n",
" 'udp/udp.stream',\n",
"]"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "84b478867fed089b",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"### PDML #######################################################\n",
"PDML_FILE = '../dumps/usage_dump.pdml'\n",
"\n",
"\n",
"################################################################\n",
"\n",
"def collect_packet_fields(element: xml.etree.ElementTree.Element, prefix: str, packet_fields_: dict) -> dict:\n",
" tag_name = element.tag\n",
" name_attr = element.attrib['name']\n",
" empty_name_attr = False\n",
" if not name_attr:\n",
" empty_name_attr = True\n",
" col_name_ = f'{prefix}/<>' if prefix else '<>'\n",
" else:\n",
" col_name_ = f'{prefix}/{name_attr}' if prefix else f'{name_attr}'\n",
"\n",
" # if tag_name != 'proto' and col_name_ not in BLACKLISTED_FIELDS:\n",
" if tag_name != 'proto':\n",
" if not empty_name_attr:\n",
" total_col_names.add(col_name_)\n",
" try:\n",
" packet_fields_[col_name_] = element.attrib['show']\n",
" except KeyError:\n",
" try:\n",
" packet_fields_[col_name_] = element.attrib['value']\n",
" except KeyError:\n",
" packet_fields_[col_name_] = np.nan\n",
"\n",
" # Recursively process child elements (subfields)\n",
" for child in element:\n",
" packet_fields_.update(collect_packet_fields(child, col_name_, packet_fields_))\n",
"\n",
" return packet_fields_\n",
"\n",
"\n",
"print('Parsing PDML file...')\n",
"tree = ET.parse(PDML_FILE)\n",
"root = tree.getroot()\n",
"\n",
"total_col_names = set()\n",
"total_pkt_fields = []\n",
"\n",
"print('Collecting packet fields...')\n",
"# Loop over all packets\n",
"packets = root.findall('packet')\n",
"for packet in packets:\n",
" packet_fields = {}\n",
" # Loop over all protocols\n",
" protocols = packet.findall('proto')\n",
" for protocol in protocols:\n",
" packet_fields.update(collect_packet_fields(protocol, '', {}))\n",
" total_pkt_fields.append(packet_fields)\n",
"\n",
"total_col_names = sorted(total_col_names)\n",
"total_pkt_fields = sorted(total_pkt_fields, key=lambda x: int(x['frame/frame.number'])) # Sort by frame.number\n",
"df_data_full = pd.DataFrame(total_pkt_fields)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "b2692056de3e85b7",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"df_data_full.shape"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "263145b13cd0b10b",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"# All columns with only NaN rows\n",
"sorted(df_data_full.columns[df_data_full.isna().all()].tolist())"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "25481a29eba04c14",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"# merged_features = set(FEATURES_FROM_PAPER + CUSTOM_CHOSEN_FEATURES)\n",
"features = set(MINIMAL_FEATURES)\n",
"f_intersection = features & set(df_data_full.columns)\n",
"features_not_present_in_pdml = features - f_intersection\n",
"# features_not_present_in_pdml"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "42c984d759e144b",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"# First add all features which are present in the .pdml\n",
"df_data_final = df_data_full.loc[:, list(f_intersection)]\n",
"# Then add all features which have been picked and are from the paper, but are not in the .pdml. Fill them with NaNs.\n",
"df_data_final[list(features_not_present_in_pdml)] = np.nan\n",
"# Sort columns\n",
"# df_data_final.sort_index(axis=1, inplace=True)"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "3454b5ca3e48ad92",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"# df_data_final.shape"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "5b5914a527e2a483",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"# df_data_final.columns"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c6371cc759696ec3",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"# All columns with only NaN rows\n",
"# df_data_final.columns[df_data_final.isna().all()].tolist()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "7114a6cf74cafef8",
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"# df_data_final.to_csv(\"data_final.csv\", index=False, sep='|')\n",
"df_data_final.to_pickle(\"../echo-dot-activity.pickel\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"outputs": [],
"source": [],
"metadata": {
"collapsed": false
},
"id": "93deee78d784393a"
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3 (ipykernel)",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.9"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
graph_based_intrusion_detection/notebooks/live_ids.ipynb
+328
View File
@@ -0,0 +1,328 @@
{
"cells": [
{
"metadata": {
"ExecuteTime": {
"end_time": "2024-02-24T13:28:00.794477Z",
"start_time": "2024-02-24T13:28:00.782180Z"
}
},
"cell_type": "code",
"source": [
"import os\n",
"from typing import List, Callable\n",
"import queue\n",
"import joblib\n",
"import threading\n",
"import time\n",
"import logging\n",
"import json\n",
"\n",
"import networkx as nx\n",
"\n",
"from kafka import KafkaConsumer, KafkaProducer\n",
"import redis\n",
"\n",
"from graph_based_intrusion_detection.utils import constants\n",
"from graph_based_intrusion_detection.utils.kafka_utils import deserialize, serialize\n",
"from graph_based_intrusion_detection.utils.logging import create_logger\n",
"from graph_based_intrusion_detection.utils.state_merging import merge_states\n",
"from graph_based_intrusion_detection.packet_processing import processing_functions\n",
"from graph_based_intrusion_detection.packet_processing.processing_functions import process_packets\n",
"from graph_based_intrusion_detection.graph_processing.data_extraction import create_dataset_for_node"
],
"id": "48d278a5f5448c0e",
"outputs": [],
"execution_count": 16
},
{
"metadata": {
"ExecuteTime": {
"end_time": "2024-02-23T14:28:05.510736Z",
"start_time": "2024-02-23T14:28:05.507164Z"
}
},
"cell_type": "code",
"source": [
"# define constants\n",
"log_file_path = os.path.abspath(os.path.join(os.path.dirname(\".\"), \"logs\", \"main.log\"))\n",
"print(f\"Log file path: {log_file_path}\")\n",
"\n",
"update_interval_in_ms = 1000\n",
"\n"
],
"id": "bbed502fad42ce8f",
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Log file path: /home/alex/projects/it-security-praktikum/devices/dev08-echo-dot-l4s3re/graph_based_intrusion_detection/notebooks/logs/main.log\n"
]
}
],
"execution_count": 11
},
{
"metadata": {
"ExecuteTime": {
"end_time": "2024-02-23T14:13:07.915623Z",
"start_time": "2024-02-23T14:13:07.912275Z"
}
},
"cell_type": "code",
"source": "# define function for creating logger for each thread\n",
"id": "44cf10deef903383",
"outputs": [],
"execution_count": 8
},
{
"cell_type": "code",
"execution_count": null,
"id": "7759e6a6c4b91cd6",
"metadata": {
"collapsed": true
},
"outputs": [],
"source": [
"\n",
"packet_queue = queue.Queue()\n",
"state_queue = queue.Queue()\n",
"connection_update_queue = queue.Queue()\n",
"connection_update_done_queue = queue.Queue()\n",
"\n",
"current_state = dict()\n",
"state_lock = threading.Lock()\n",
"\n",
"connection_registry = dict()\n",
"connection_registry_lock = threading.Lock()\n"
]
},
{
"metadata": {},
"cell_type": "code",
"outputs": [],
"execution_count": null,
"source": [
"def packet_worker(packet_processing_functions: list, worker_id: int):\n",
" \"\"\"\n",
" Worker thread, processes packets from the packet queue and puts the resulting state into the state queue\n",
" :param packet_processing_functions: function for processing packets\n",
" :param worker_id: id of the worker\n",
" :return: None\n",
" \"\"\"\n",
"\n",
" logger = create_logger(f\"Packet-Worker {worker_id}\")\n",
" logger.info(f\"Worker {worker_id} started\")\n",
"\n",
" while True:\n",
" # get current batch size based on approximate queue length\n",
" n_packets = packet_queue.qsize()\n",
" batch_size = n_packets // n_workers\n",
" logger.info(f\"Worker {worker_id} current batch size: {batch_size}\")\n",
"\n",
" # fetch batch from queue\n",
" batch = list()\n",
" try:\n",
" for _ in range(batch_size):\n",
" batch.append(packet_queue.get_nowait())\n",
" except queue.Empty:\n",
" pass\n",
"\n",
" logger.info(f\"Worker {worker_id} processing {len(batch)} packets\")\n",
" # process batch\n",
" if len(batch) > 0:\n",
" state = process_packets(batch, packet_processing_functions)\n",
" state_queue.put(state)\n",
" else:\n",
" time.sleep(0.1)\n",
"\n",
" logger.info(f\"Worker {worker_id} done\")\n",
"\n",
" # mark batch as done\n",
" for _ in batch:\n",
" packet_queue.task_done()"
],
"id": "852ee0a61981f880"
},
{
"metadata": {},
"cell_type": "code",
"outputs": [],
"execution_count": null,
"source": [
"\n",
"def merger(merger_function: Callable, worker_id: int):\n",
" \"\"\"\n",
" Merger thread, merges states from the state queue into one central state representing the state of the network\n",
" :param merger_function: function for merging states\n",
" :param worker_id: id of the worker\n",
" :return: None\n",
" \"\"\"\n",
" global current_state\n",
"\n",
" logger = create_logger(f\"Merger {worker_id}\")\n",
" logger.info(f\"Merger {worker_id} started\")\n",
"\n",
" while True:\n",
" # get state lock\n",
" state_lock.acquire()\n",
"\n",
" # fetch state from queue\n",
" state = state_queue.get()\n",
"\n",
" logger.info(f\"Merging state\")\n",
" states = [current_state, state]\n",
" # merge state\n",
" new_state = merger_function(states)\n",
"\n",
" # update state\n",
" current_state = new_state\n",
"\n",
" # mark state as done\n",
" state_queue.task_done()\n",
"\n",
" # release state lock\n",
" state_lock.release()\n",
"\n",
" logger.info(f\"State merged\")\n"
],
"id": "initial_id"
},
{
"metadata": {},
"cell_type": "code",
"outputs": [],
"execution_count": null,
"source": [
"def dispatcher(packet: dict,\n",
" worker_id: int):\n",
" \"\"\"\n",
" Dispatcher thread, watches connection registry and dispatches connection to workers for analysis\n",
" :param packet: packet to dispatch\n",
" :param worker_id: id of the worker\n",
" :return: None\n",
" \"\"\"\n",
"\n",
" logger = create_logger(f\"Dispatcher {worker_id}\")\n",
" logger.info(f\"Dispatcher {worker_id} started\")\n",
"\n",
" while True:\n",
" # get connections, based on layer 3 graph -> TODO: extend functionality\n",
" # dirty read is not a problem here, modification will be picked up in the next iteration\n",
" base_graph = current_state[\"layer_3_graph\"]\n",
"\n",
" # get connections, connections are edges in the graph\n",
" connections = list(base_graph.edges())\n",
"\n",
" # get connection registry lock\n",
" connection_registry_lock.acquire()\n",
"\n",
" # get connections that are not in the registry\n",
" new_connections = [connection for connection in connections if connection not in connection_registry]\n",
"\n",
" # process connection updates since last update\n",
" while not connection_update_done_queue.empty():\n",
" # get connection from queue, wait at most update interval\n",
" try:\n",
" connection = connection_update_done_queue.get(timeout=update_interval_in_ms)\n",
" except queue.Empty:\n",
" break\n",
" # update connection in registry, connection is tuple with connection and last updated\n",
" connection_registry[connection][\"last_updated\"] = connection[1]\n",
"\n",
" # add new connections to registry\n",
" for connection in new_connections:\n",
" # add connection to registry, set last updated to None so that it is processed\n",
" connection_registry[connection] = {\"last_updated\": None}\n",
"\n",
" # add connections to work queue, if last update is longer ago than the update interval or None\n",
" for connection, connection_info in connection_registry.items():\n",
" if connection_info[\"last_updated\"] is None or time.time() - connection_info[\n",
" \"last_updated\"] > update_interval_in_ms:\n",
" connection_update_queue.put(connection)\n",
"\n",
" # release connection registry lock\n",
" connection_registry_lock.release()\n"
],
"id": "cca86765185e962c"
},
{
"metadata": {},
"cell_type": "code",
"outputs": [],
"execution_count": null,
"source": [
"\n",
"\n",
"\n",
"\n",
"\n"
],
"id": "b9b68e0a300b6b39"
},
{
"metadata": {
"ExecuteTime": {
"end_time": "2024-02-23T14:00:52.816881Z",
"start_time": "2024-02-23T14:00:52.644607Z"
}
},
"cell_type": "code",
"source": [
"\n",
"n_workers = 1\n",
"packet_workers = [threading.Thread(target=packet_worker, args=(processing_functions, worker_id))\n",
" for worker_id in range(n_workers)]\n",
"for worker_thread in packet_workers:\n",
" worker_thread.daemon = True\n",
"\n",
"merger_thread = threading.Thread(target=merger, args=(merge_states,))\n",
"merger_thread.daemon = True\n",
"\n",
"# start threads\n",
"# for worker_thread in workers:\n",
"# worker_thread.start()\n",
"# \n",
"# merger_thread.start()\n"
],
"id": "dde42074dc4deda9",
"outputs": [
{
"ename": "NameError",
"evalue": "name 'threading' is not defined",
"output_type": "error",
"traceback": [
"\u001B[0;31m---------------------------------------------------------------------------\u001B[0m",
"\u001B[0;31mNameError\u001B[0m Traceback (most recent call last)",
"Cell \u001B[0;32mIn[1], line 2\u001B[0m\n\u001B[1;32m 1\u001B[0m n_workers \u001B[38;5;241m=\u001B[39m \u001B[38;5;241m1\u001B[39m\n\u001B[0;32m----> 2\u001B[0m workers \u001B[38;5;241m=\u001B[39m \u001B[43m[\u001B[49m\u001B[43mthreading\u001B[49m\u001B[38;5;241;43m.\u001B[39;49m\u001B[43mThread\u001B[49m\u001B[43m(\u001B[49m\u001B[43mtarget\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43mworker\u001B[49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43margs\u001B[49m\u001B[38;5;241;43m=\u001B[39;49m\u001B[43m(\u001B[49m\u001B[43mprocessing_functions\u001B[49m\u001B[43m,\u001B[49m\u001B[43m \u001B[49m\u001B[43mworker_id\u001B[49m\u001B[43m)\u001B[49m\u001B[43m)\u001B[49m\n\u001B[1;32m 3\u001B[0m \u001B[43m \u001B[49m\u001B[38;5;28;43;01mfor\u001B[39;49;00m\u001B[43m \u001B[49m\u001B[43mworker_id\u001B[49m\u001B[43m \u001B[49m\u001B[38;5;129;43;01min\u001B[39;49;00m\u001B[43m \u001B[49m\u001B[38;5;28;43mrange\u001B[39;49m\u001B[43m(\u001B[49m\u001B[43mn_workers\u001B[49m\u001B[43m)\u001B[49m\u001B[43m]\u001B[49m\n\u001B[1;32m 4\u001B[0m \u001B[38;5;28;01mfor\u001B[39;00m worker_thread \u001B[38;5;129;01min\u001B[39;00m workers:\n\u001B[1;32m 5\u001B[0m worker_thread\u001B[38;5;241m.\u001B[39mdaemon \u001B[38;5;241m=\u001B[39m \u001B[38;5;28;01mTrue\u001B[39;00m\n",
"Cell \u001B[0;32mIn[1], line 2\u001B[0m, in \u001B[0;36m<listcomp>\u001B[0;34m(.0)\u001B[0m\n\u001B[1;32m 1\u001B[0m n_workers \u001B[38;5;241m=\u001B[39m \u001B[38;5;241m1\u001B[39m\n\u001B[0;32m----> 2\u001B[0m workers \u001B[38;5;241m=\u001B[39m [\u001B[43mthreading\u001B[49m\u001B[38;5;241m.\u001B[39mThread(target\u001B[38;5;241m=\u001B[39mworker, args\u001B[38;5;241m=\u001B[39m(processing_functions, worker_id))\n\u001B[1;32m 3\u001B[0m \u001B[38;5;28;01mfor\u001B[39;00m worker_id \u001B[38;5;129;01min\u001B[39;00m \u001B[38;5;28mrange\u001B[39m(n_workers)]\n\u001B[1;32m 4\u001B[0m \u001B[38;5;28;01mfor\u001B[39;00m worker_thread \u001B[38;5;129;01min\u001B[39;00m workers:\n\u001B[1;32m 5\u001B[0m worker_thread\u001B[38;5;241m.\u001B[39mdaemon \u001B[38;5;241m=\u001B[39m \u001B[38;5;28;01mTrue\u001B[39;00m\n",
"\u001B[0;31mNameError\u001B[0m: name 'threading' is not defined"
]
}
],
"execution_count": 1
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 2
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython2",
"version": "2.7.6"
}
},
"nbformat": 4,
"nbformat_minor": 5
}
graph_based_intrusion_detection/notebooks/merging.ipynb
+1105
View File
File diff suppressed because it is too large Load Diff
graph_based_intrusion_detection/notebooks/system_tester.ipynb
+1010
View File
File diff suppressed because one or more lines are too long