From 29a1b13e7ac8f20512b7a066c351bad614998f83 Mon Sep 17 00:00:00 2001
From: Toni Uhlig <matzeton@googlemail.com>
Date: Tue, 25 Jan 2022 11:16:41 +0100
Subject: Improved Plotly/Dash example. It is now somehow informative.
* TCP timeout after FIN/RST: switched back to the value from a35fc1d5ea8570609cc0c8cf6edadc81f8f5bb76
* py-flow-info: reset 'guessed' flag after detection/detection-update received
Signed-off-by: Toni Uhlig <matzeton@googlemail.com>
---
examples/py-flow-dashboard/flow-dash.py | 232 +++++++++++++++-----------------
1 file changed, 105 insertions(+), 127 deletions(-)
(limited to 'examples/py-flow-dashboard/flow-dash.py')
diff --git a/examples/py-flow-dashboard/flow-dash.py b/examples/py-flow-dashboard/flow-dash.py
index 8e49ed020..283a97bb9 100755
--- a/examples/py-flow-dashboard/flow-dash.py
+++ b/examples/py-flow-dashboard/flow-dash.py
@@ -1,16 +1,11 @@
#!/usr/bin/env python3
-from collections import deque
-import dash
-from dash.dependencies import Output, Input
-import dash_core_components as dcc
-import dash_html_components as html
import multiprocessing
import os
-import plotly
-import plotly.graph_objs as go
import sys
+import plotly_dash
+
sys.path.append(os.path.dirname(sys.argv[0]) + '/../share/nDPId')
sys.path.append(os.path.dirname(sys.argv[0]) + '/../usr/share/nDPId')
try:
@@ -21,146 +16,109 @@ except ImportError:
import nDPIsrvd
from nDPIsrvd import nDPIsrvdSocket
-mgr = multiprocessing.Manager()
-
-global shared_flow_dict
-shared_flow_dict = mgr.dict()
-
-FLOW_COUNT_DATAPOINTS = 50
-
-global live_flow_count_X
-live_flow_count_X = deque(maxlen=FLOW_COUNT_DATAPOINTS)
-live_flow_count_X.append(1)
-global live_flow_count_Y
-live_flow_count_Y = deque(maxlen=FLOW_COUNT_DATAPOINTS)
-live_flow_count_Y.append(1)
-
-live_flow_bars = ['risky', 'midstream', 'detected', 'guessed', 'not-detected']
-fig = go.Figure()
-
-app = dash.Dash(__name__)
-app.layout = html.Div(
- [
- dcc.Graph(id='live-flow-count', animate=True),
- dcc.Graph(id='live-flow-bars', animate=True, figure=fig),
- dcc.Interval(
- id='graph-update',
- interval=1000,
- n_intervals=0
- ),
- ]
-)
-
-
-@app.callback(
- Output('live-flow-count', 'figure'),
- [Input('graph-update', 'n_intervals')]
-)
-def update_graph_scatter(n):
- live_flow_count_X.append(live_flow_count_X[-1]+1)
- live_flow_count_Y.append(len(shared_flow_dict))
-
- data = plotly.graph_objs.Scatter(
- x=list(live_flow_count_X),
- y=list(live_flow_count_Y),
- name='Scatter',
- mode='lines+markers'
- )
-
- return {
- 'data': [data],
- 'layout':
- go.Layout(
- xaxis=dict(
- range=[min(live_flow_count_X), max(live_flow_count_X)]
- ),
- yaxis=dict(
- range=[min(live_flow_count_Y), max(live_flow_count_Y)]
- ),
- )}
-
-
-@app.callback(
- Output('live-flow-bars', 'figure'),
- [Input('graph-update', 'n_intervals')]
-)
-def update_pie(n):
- values = [0, 0, 0, 0, 0]
-
- for flow_id in shared_flow_dict.keys():
- try:
- flow = shared_flow_dict[flow_id]
- except KeyError:
- continue
-
- if flow['is_risky'] is True:
- values[0] += 1
-
- if flow['is_midstream'] is True:
- values[1] += 1
-
- if flow['is_detected'] is True:
- values[2] += 1
-
- if flow['is_guessed'] is True:
- values[3] += 1
-
- if flow['is_not_detected'] is True:
- values[4] += 1
-
- # print(values)
-
- return {
- 'data': [
- go.Bar(name='', x=live_flow_bars, y=values)
- ],
- 'layout': go.Layout(yaxis=dict(range=[0, max(values)]))
- }
-
-
-def web_worker():
- app.run_server()
-
def nDPIsrvd_worker_onFlowCleanup(instance, current_flow, global_user_data):
+ _, shared_flow_dict = global_user_data
+
+ flow_id = current_flow.flow_id
+
+ shared_flow_dict['current-flows'] -= 1
+
+ if shared_flow_dict[flow_id]['is_detected'] is True:
+ shared_flow_dict['current-detected-flows'] -= 1
+
+ if shared_flow_dict[flow_id]['is_guessed'] is True:
+ shared_flow_dict['current-guessed-flows'] -= 1
+
+ if shared_flow_dict[flow_id]['is_not_detected'] is True:
+ shared_flow_dict['current-detected-flows'] -= 1
+
+ if shared_flow_dict[flow_id]['is_midstream'] is True:
+ shared_flow_dict['current-midstream-flows'] -= 1
+
+ if shared_flow_dict[flow_id]['is_risky'] is True:
+ shared_flow_dict['current-risky-flows'] -= 1
+
del shared_flow_dict[current_flow.flow_id]
return True
def nDPIsrvd_worker_onJsonLineRecvd(json_dict, instance, current_flow, global_user_data):
+ nsock, shared_flow_dict = global_user_data
+
+ shared_flow_dict['total-events'] += 1
+ shared_flow_dict['total-bytes'] = nsock.received_bytes
+
if 'flow_id' not in json_dict:
return True
+ else:
+ if current_flow.flow_id != json_dict['flow_id']:
+ return False
+ flow_id = current_flow.flow_id
# print(json_dict)
- if json_dict['flow_id'] not in shared_flow_dict:
- shared_flow_dict[json_dict['flow_id']] = mgr.dict()
- shared_flow_dict[json_dict['flow_id']]['is_detected'] = False
- shared_flow_dict[json_dict['flow_id']]['is_guessed'] = False
- shared_flow_dict[json_dict['flow_id']]['is_not_detected'] = False
- shared_flow_dict[json_dict['flow_id']]['is_midstream'] = False
- shared_flow_dict[json_dict['flow_id']]['is_risky'] = False
+ if flow_id not in shared_flow_dict:
+ shared_flow_dict[flow_id] = mgr.dict()
+ shared_flow_dict[flow_id]['is_detected'] = False
+ shared_flow_dict[flow_id]['is_guessed'] = False
+ shared_flow_dict[flow_id]['is_not_detected'] = False
+ shared_flow_dict[flow_id]['is_midstream'] = False
+ shared_flow_dict[flow_id]['is_risky'] = False
+
+ shared_flow_dict['total-flows'] += 1
+ shared_flow_dict['current-flows'] += 1
+
+ if 'midstream' in json_dict and json_dict['midstream'] != 0:
+ if shared_flow_dict[flow_id]['is_midstream'] is False:
+ shared_flow_dict['total-midstream-flows'] += 1
+ shared_flow_dict['current-midstream-flows'] += 1
+ shared_flow_dict[flow_id]['is_midstream'] = True
+
+ if 'ndpi' in json_dict and 'flow_risk' in json_dict['ndpi']:
+ if shared_flow_dict[flow_id]['is_risky'] is False:
+ shared_flow_dict['total-risky-flows'] += 1
+ shared_flow_dict['current-risky-flows'] += 1
+ shared_flow_dict[flow_id]['is_risky'] = True
if 'flow_event_name' not in json_dict:
return True
if json_dict['flow_event_name'] == 'new':
- if 'midstream' in json_dict and json_dict['midstream'] != 0:
- shared_flow_dict[json_dict['flow_id']]['is_midstream'] = True
+
+ pass
+
elif json_dict['flow_event_name'] == 'guessed':
- shared_flow_dict[json_dict['flow_id']]['is_guessed'] = True
+
+ if shared_flow_dict[flow_id]['is_guessed'] is False:
+ shared_flow_dict['total-guessed-flows'] += 1
+ shared_flow_dict['current-guessed-flows'] += 1
+ shared_flow_dict[flow_id]['is_guessed'] = True
+
elif json_dict['flow_event_name'] == 'not-detected':
- shared_flow_dict[json_dict['flow_id']]['is_not_detected'] = True
- elif json_dict['flow_event_name'] == 'detected':
- shared_flow_dict[json_dict['flow_id']]['is_detected'] = True
- shared_flow_dict[json_dict['flow_id']]['is_guessed'] = False
- if 'ndpi' in json_dict and 'flow_risk' in json_dict['ndpi']:
- shared_flow_dict[json_dict['flow_id']]['is_risky'] = True
+
+ if shared_flow_dict[flow_id]['is_not_detected'] is False:
+ shared_flow_dict['total-not-detected-flows'] += 1
+ shared_flow_dict['current-not-detected-flows'] += 1
+ shared_flow_dict[flow_id]['is_not_detected'] = True
+
+ elif json_dict['flow_event_name'] == 'detected' or \
+ json_dict['flow_event_name'] == 'detection-update':
+
+ if shared_flow_dict[flow_id]['is_detected'] is False:
+ shared_flow_dict['total-detected-flows'] += 1
+ shared_flow_dict['current-detected-flows'] += 1
+ shared_flow_dict[flow_id]['is_detected'] = True
+
+ if shared_flow_dict[flow_id]['is_guessed'] is True:
+ shared_flow_dict['total-guessed-flows'] -= 1
+ shared_flow_dict['current-guessed-flows'] -= 1
+ shared_flow_dict[flow_id]['is_guessed'] = False
return True
-def nDPIsrvd_worker(address, nDPIsrvd_global_user_data):
+def nDPIsrvd_worker(address, shared_flow_dict):
sys.stderr.write('Recv buffer size: {}\n'
.format(nDPIsrvd.NETWORK_BUFFER_MAX_SIZE))
sys.stderr.write('Connecting to {} ..\n'
@@ -171,7 +129,7 @@ def nDPIsrvd_worker(address, nDPIsrvd_global_user_data):
nsock.connect(address)
nsock.loop(nDPIsrvd_worker_onJsonLineRecvd,
nDPIsrvd_worker_onFlowCleanup,
- nDPIsrvd_global_user_data)
+ (nsock, shared_flow_dict))
if __name__ == '__main__':
@@ -179,11 +137,31 @@ if __name__ == '__main__':
args = argparser.parse_args()
address = nDPIsrvd.validateAddress(args)
+ mgr = multiprocessing.Manager()
+ shared_flow_dict = mgr.dict()
+
+ shared_flow_dict['total-events'] = 0
+ shared_flow_dict['total-bytes'] = 0
+ shared_flow_dict['total-flows'] = 0
+ shared_flow_dict['total-detected-flows'] = 0
+ shared_flow_dict['total-risky-flows'] = 0
+ shared_flow_dict['total-midstream-flows'] = 0
+ shared_flow_dict['total-guessed-flows'] = 0
+ shared_flow_dict['total-not-detected-flows'] = 0
+
+ shared_flow_dict['current-flows'] = 0
+ shared_flow_dict['current-detected-flows'] = 0
+ shared_flow_dict['current-risky-flows'] = 0
+ shared_flow_dict['current-midstream-flows'] = 0
+ shared_flow_dict['current-guessed-flows'] = 0
+ shared_flow_dict['current-not-detected-flows'] = 0
+
nDPIsrvd_job = multiprocessing.Process(target=nDPIsrvd_worker,
- args=(address, None))
+ args=(address, shared_flow_dict))
nDPIsrvd_job.start()
- web_job = multiprocessing.Process(target=web_worker, args=())
+ web_job = multiprocessing.Process(target=plotly_dash.web_worker,
+ args=(shared_flow_dict,))
web_job.start()
nDPIsrvd_job.join()
--
cgit v1.2.3