mirror of
https://github.com/nfstream/nfstream.git
synced 2026-08-06 23:47:26 +00:00
427ba6e151
Windows spawns meter processes instead of forking them, so plugin instances must be pickled to reach them. The plugin was defined inside the test function, which cannot be pickled, and the Windows job failed with AttributeError: Can't pickle local object 'NFStreamTest.test_anonymize_absent_values.<locals>.MixedValues' Linux and macOS fork, so the plugin was inherited rather than pickled and the test passed there. The failure was also masked at first because the Windows jobs were failing earlier, at the Npcap download, and never reached the tests. Moving the class to module level makes it picklable, matching how the other plugins used in the suite are defined.
1027 lines
38 KiB
Python
1027 lines
38 KiB
Python
"""
|
|
------------------------------------------------------------------------------------------------------------------------
|
|
tests.py
|
|
Copyright (C) 2019-22 - NFStream Developers
|
|
This file is part of NFStream, a Flexible Network Data Analysis Framework (https://www.nfstream.org/).
|
|
NFStream is free software: you can redistribute it and/or modify it under the terms of the GNU Lesser General Public
|
|
License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later
|
|
version.
|
|
NFStream is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
|
|
of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more details.
|
|
You should have received a copy of the GNU Lesser General Public License along with NFStream.
|
|
If not, see <http://www.gnu.org/licenses/>.
|
|
------------------------------------------------------------------------------------------------------------------------
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
import json
|
|
import os
|
|
from nfstream import NFPlugin, NFStreamer
|
|
from nfstream.plugins import SPLT, DHCP, FlowSlicer, MDNS
|
|
|
|
|
|
class AnonymizerMixedValues(NFPlugin):
|
|
"""Must remain module-level because Windows spawn pickles plugins."""
|
|
|
|
def on_init(self, packet, flow):
|
|
flow.udps.absent = ""
|
|
flow.udps.populated = "value"
|
|
flow.udps.zero = 0
|
|
flow.udps.false = False
|
|
flow.udps.array = np.zeros(3)
|
|
|
|
|
|
def get_files_list(path):
|
|
files = []
|
|
for r, d, f in os.walk(path):
|
|
for file in f:
|
|
if (
|
|
".pcap" == file[-5:] or ".pcapng" == file[-7:]
|
|
): # Pick out only pcaps files
|
|
files.append(os.path.join(r, file))
|
|
files.sort()
|
|
return files
|
|
|
|
|
|
class NFStreamTest(object):
|
|
@staticmethod
|
|
def test_source_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
source = ["inexisting.pcap", "lo", 11]
|
|
for x in source:
|
|
try:
|
|
NFStreamer(source=x).to_pandas()
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 3
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_source_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_decode_tunnels_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
decode_tunnels = [33, "True"]
|
|
for x in decode_tunnels:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
decode_tunnels=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_decode_tunnels_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_bpf_filter_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
bpf_filter = ["my filter", 11]
|
|
for x in bpf_filter:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
bpf_filter=x,
|
|
).to_pandas()
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_bpf_filter_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_promiscuous_mode_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
promiscuous_mode = ["yes", 89]
|
|
for x in promiscuous_mode:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
promiscuous_mode=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_promiscuous_mode_parameter".ljust(60, " "),
|
|
"OK",
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_snapshot_length_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
snapshot_length = ["largest", -1]
|
|
for x in snapshot_length:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
snapshot_length=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_snapshot_length_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_socket_buffer_size_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
socket_buffer_size = ["largest", -1, 2**31]
|
|
for x in socket_buffer_size:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
socket_buffer_size=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 3
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_socket_buffer_size_parameter".ljust(60, " "),
|
|
"OK",
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_idle_timeout_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
idle_timeout = [-1, "idle"]
|
|
for x in idle_timeout:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
idle_timeout=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_idle_timeout_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_active_timeout_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
active_timeout = [-1, "active"]
|
|
for x in active_timeout:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
active_timeout=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_active_timeout_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_accounting_mode_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
accounting_mode = [-1, 5, "ip"]
|
|
for x in accounting_mode:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
accounting_mode=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 3
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_accounting_mode_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_udps_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
udps = [lambda y: y + 1, "NFPlugin"]
|
|
for x in udps:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"), udps=x
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_udps_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_n_dissections_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
n_dissections = ["yes", -1, 256]
|
|
for x in n_dissections:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
n_dissections=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 3
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_n_dissections_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_system_visibility_mode_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
system_visibility_mode = ["yes", -1, 3]
|
|
for x in system_visibility_mode:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
system_visibility_mode=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 3
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_system_visibility_mode_parameter".ljust(60, " "),
|
|
"OK",
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_system_visibility_poll_ms():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
system_visibility_mode = ["yes", -1]
|
|
for x in system_visibility_mode:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
system_visibility_poll_ms=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_system_visibility_poll_ms".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_statistical_analysis_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
statistical_analysis = ["yes", 89]
|
|
for x in statistical_analysis:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
statistical_analysis=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_statistical_analysis_parameter".ljust(60, " "),
|
|
"OK",
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_splt_analysis_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
splt_analysis = [-1, 70000, "yes"]
|
|
for x in splt_analysis:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
splt_analysis=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 3
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_splt_analysis_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_n_meters_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
n_meters = ["yes", -1]
|
|
for x in n_meters:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"), n_meters=x
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_n_meters_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_max_nflows_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
max_nflows = ["yes", -1]
|
|
for x in max_nflows:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
max_nflows=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_max_nflows_parameter".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_performance_report_parameter():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
performance_report = ["yes", -1]
|
|
for x in performance_report:
|
|
try:
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
performance_report=x,
|
|
)
|
|
except ValueError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 2
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_performance_report_parameter".ljust(60, " "),
|
|
"OK",
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_expiration_management():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
# Idle expiration
|
|
streamer_expiration = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"), idle_timeout=0
|
|
)
|
|
last_id = 0
|
|
for flow in streamer_expiration:
|
|
last_id = flow.id
|
|
assert last_id == 27
|
|
# Active expiration
|
|
streamer_expiration = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"), active_timeout=0
|
|
)
|
|
last_id = 0
|
|
for flow in streamer_expiration:
|
|
last_id = flow.id
|
|
assert last_id == 27
|
|
# Custom expiration
|
|
streamer_expiration = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
udps=FlowSlicer(limit=1),
|
|
)
|
|
last_id = 0
|
|
for flow in streamer_expiration:
|
|
last_id = flow.id
|
|
assert last_id == 27
|
|
streamer_expiration = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
udps=FlowSlicer(limit=4),
|
|
)
|
|
last_id = 0
|
|
for flow in streamer_expiration:
|
|
last_id = flow.id
|
|
assert last_id == 6
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_expiration_management".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_tunnel_decoding():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
n_exceptions = 0
|
|
decode_streamer = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "gtp-u.pcap"),
|
|
statistical_analysis=True,
|
|
decode_tunnels=True,
|
|
)
|
|
for flow in decode_streamer:
|
|
assert flow.tunnel_id == 1
|
|
decode_streamer.decode_tunnels = False
|
|
for flow in decode_streamer:
|
|
try:
|
|
getattr(flow, "tunnel_id")
|
|
except AttributeError:
|
|
n_exceptions += 1
|
|
assert n_exceptions == 1
|
|
del decode_streamer
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_tunnel_decoding".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_statistical():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
statistical_streamer = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
statistical_analysis=True,
|
|
accounting_mode=1,
|
|
)
|
|
for flow in statistical_streamer:
|
|
assert flow.id == 0
|
|
assert flow.expiration_id == 0
|
|
assert flow.src_ip == "172.31.3.224"
|
|
assert flow.src_mac == "80:c6:ca:00:9e:9f"
|
|
assert flow.src_oui == "80:c6:ca"
|
|
assert flow.src_port == 42835
|
|
assert flow.dst_ip == "216.58.212.100"
|
|
assert flow.dst_mac == "00:0e:8e:4d:b4:a8"
|
|
assert flow.dst_oui == "00:0e:8e"
|
|
assert flow.dst_port == 443
|
|
assert flow.protocol == 6
|
|
assert flow.ip_version == 4
|
|
assert flow.vlan_id == 0
|
|
assert flow.tunnel_id == 0
|
|
assert flow.bidirectional_first_seen_ms == 1434443394683
|
|
assert flow.bidirectional_last_seen_ms == 1434443401353
|
|
assert flow.bidirectional_duration_ms == 6670
|
|
assert flow.bidirectional_packets == 28
|
|
assert flow.bidirectional_bytes == 8696
|
|
assert flow.src2dst_first_seen_ms == 1434443394683
|
|
assert flow.src2dst_last_seen_ms == 1434443401353
|
|
assert flow.src2dst_duration_ms == 6670
|
|
assert flow.src2dst_packets == 16
|
|
assert flow.src2dst_bytes == 1288
|
|
assert flow.dst2src_first_seen_ms == 1434443394717
|
|
assert flow.dst2src_last_seen_ms == 1434443401308
|
|
assert flow.dst2src_duration_ms == 6591
|
|
assert flow.dst2src_packets == 12
|
|
assert flow.dst2src_bytes == 7408
|
|
assert flow.bidirectional_min_ps == 40
|
|
assert (flow.bidirectional_mean_ps - 310.571) < 0.001
|
|
assert (flow.bidirectional_stddev_ps - 500.546) < 0.001
|
|
assert flow.bidirectional_max_ps == 1470
|
|
assert flow.src2dst_min_ps == 40
|
|
assert (flow.src2dst_mean_ps - 80.499) < 0.001
|
|
assert (flow.src2dst_stddev_ps - 89.555) < 0.001
|
|
assert flow.src2dst_max_ps == 354
|
|
assert flow.dst2src_min_ps == 40
|
|
assert (flow.dst2src_mean_ps - 617.333) < 0.001
|
|
assert (flow.dst2src_stddev_ps - 651.452) < 0.001
|
|
assert flow.dst2src_max_ps == 1470
|
|
assert flow.bidirectional_min_piat_ms == 0
|
|
assert (flow.bidirectional_mean_piat_ms - 247.037) < 0.001
|
|
assert (flow.bidirectional_stddev_piat_ms - 324.045) < 0.001
|
|
assert flow.bidirectional_max_piat_ms == 995
|
|
assert flow.src2dst_min_piat_ms == 76
|
|
assert (flow.src2dst_mean_piat_ms - 444.666) < 0.001
|
|
assert (flow.src2dst_stddev_piat_ms - 397.603) < 0.001
|
|
assert flow.src2dst_max_piat_ms == 1185
|
|
assert flow.dst2src_min_piat_ms == 66
|
|
assert (flow.dst2src_mean_piat_ms - 599.181) < 0.001
|
|
assert (flow.dst2src_stddev_piat_ms - 384.784) < 0.001
|
|
assert flow.dst2src_max_piat_ms == 1213
|
|
assert flow.bidirectional_syn_packets == 2
|
|
assert flow.bidirectional_cwr_packets == 0
|
|
assert flow.bidirectional_ece_packets == 0
|
|
assert flow.bidirectional_urg_packets == 0
|
|
assert flow.bidirectional_ack_packets == 27
|
|
assert flow.bidirectional_psh_packets == 8
|
|
assert flow.bidirectional_rst_packets == 0
|
|
assert flow.bidirectional_fin_packets == 2
|
|
assert flow.src2dst_syn_packets == 1
|
|
assert flow.src2dst_cwr_packets == 0
|
|
assert flow.src2dst_ece_packets == 0
|
|
assert flow.src2dst_urg_packets == 0
|
|
assert flow.src2dst_ack_packets == 15
|
|
assert flow.src2dst_psh_packets == 4
|
|
assert flow.src2dst_rst_packets == 0
|
|
assert flow.src2dst_fin_packets == 1
|
|
assert flow.dst2src_syn_packets == 1
|
|
assert flow.dst2src_cwr_packets == 0
|
|
assert flow.dst2src_ece_packets == 0
|
|
assert flow.dst2src_urg_packets == 0
|
|
assert flow.dst2src_ack_packets == 12
|
|
assert flow.dst2src_psh_packets == 4
|
|
assert flow.dst2src_rst_packets == 0
|
|
assert flow.dst2src_fin_packets == 1
|
|
del statistical_streamer
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_statistical".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_fingerprint_extraction():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
fingerprint_streamer = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "facebook.pcap"),
|
|
statistical_analysis=True,
|
|
accounting_mode=1,
|
|
)
|
|
for flow in fingerprint_streamer:
|
|
assert flow.application_name == "TLS.Facebook"
|
|
assert flow.application_category_name == "SocialNetwork"
|
|
assert flow.application_is_guessed == 0
|
|
assert flow.application_confidence == 6
|
|
requested_server_name = flow.requested_server_name in [
|
|
"facebook.com",
|
|
"www.facebook.com",
|
|
]
|
|
assert int(requested_server_name) == 1
|
|
# nDPI 5.0: client_fingerprint now contains JA4 (not JA3), server_fingerprint still JA3S
|
|
client_fingerprint = flow.client_fingerprint in [
|
|
"t12d1311h2_27a29bd8d6e6_c4623e4f4474", # JA4 for www.facebook.com
|
|
"t12d1310h2_27a29bd8d6e6_85173d161f9a", # JA4 for facebook.com
|
|
]
|
|
assert int(client_fingerprint) == 1
|
|
server_fingerprint = flow.server_fingerprint in [
|
|
"2d1eb5817ece335c24904f516ad5da12", # JA3S unchanged
|
|
"96681175a9547081bf3d417f1a572091", # JA3S unchanged
|
|
]
|
|
assert int(server_fingerprint) == 1
|
|
del fingerprint_streamer
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_fingerprint_extraction".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_export():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
df = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "steam.pcapng"),
|
|
statistical_analysis=True,
|
|
n_dissections=20,
|
|
).to_pandas()
|
|
df_anon = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "steam.pcapng"),
|
|
statistical_analysis=True,
|
|
n_dissections=20,
|
|
).to_pandas(columns_to_anonymize=["src_ip", "dst_ip"])
|
|
assert df_anon.shape[0] == df.shape[0]
|
|
assert df_anon.shape[1] == df.shape[1]
|
|
assert df_anon["src_ip"].nunique() == df["src_ip"].nunique()
|
|
assert df_anon["dst_ip"].nunique() == df["dst_ip"].nunique()
|
|
total_flows = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "steam.pcapng"),
|
|
statistical_analysis=True,
|
|
n_dissections=20,
|
|
).to_csv()
|
|
df_from_csv = pd.read_csv(os.path.join("tests", "pcaps", "steam.pcapng.csv"))
|
|
total_flows_anon = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "steam.pcapng"),
|
|
statistical_analysis=True,
|
|
n_dissections=20,
|
|
).to_csv()
|
|
df_anon_from_csv = pd.read_csv(os.path.join("tests", "pcaps", "steam.pcapng.csv"))
|
|
os.remove(os.path.join("tests", "pcaps", "steam.pcapng.csv"))
|
|
assert total_flows == total_flows_anon
|
|
assert total_flows == df_from_csv.shape[0]
|
|
assert total_flows_anon == df_anon_from_csv.shape[0]
|
|
assert total_flows == df.shape[0]
|
|
assert total_flows_anon == df_anon.shape[0]
|
|
print("{}\t: {}".format(".test_export".ljust(60, " "), "OK"))
|
|
|
|
@staticmethod
|
|
def test_bpf():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
streamer_test = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "facebook.pcap"),
|
|
bpf_filter="src port 52066 or dst port 52066",
|
|
)
|
|
last_id = 0
|
|
for flow in streamer_test:
|
|
last_id = flow.id
|
|
assert flow.src_port == 52066
|
|
assert last_id == 0
|
|
print("{}\t: {}".format(".test_bpf".ljust(60, " "), "OK"))
|
|
|
|
@staticmethod
|
|
def test_ndpi_integration():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
pcap_files = get_files_list(os.path.join("tests", "pcaps"))
|
|
result_files = get_files_list(os.path.join("tests", "results"))
|
|
failures = 0
|
|
print(".Test nDPI integration on {} applications:".format(len(pcap_files)))
|
|
for file_idx, test_file in enumerate(pcap_files):
|
|
test_case_name = os.path.basename(test_file)
|
|
try:
|
|
test = (
|
|
NFStreamer(source=test_file, n_dissections=20, n_meters=1)
|
|
.to_pandas()[
|
|
[
|
|
"id",
|
|
"bidirectional_packets",
|
|
"bidirectional_bytes",
|
|
"application_name",
|
|
"application_category_name",
|
|
"application_is_guessed",
|
|
"application_confidence",
|
|
]
|
|
]
|
|
.to_dict()
|
|
)
|
|
|
|
true = pd.read_csv(result_files[file_idx]).to_dict()
|
|
assert test == true
|
|
print(
|
|
"{}\t: {}".format(
|
|
test_case_name.ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
except AssertionError:
|
|
failures += 1
|
|
print(
|
|
"{}\t: {}".format(
|
|
test_case_name.ljust(60, " "), "KO"
|
|
)
|
|
)
|
|
# Everything must be OK
|
|
assert failures == 0
|
|
|
|
@staticmethod
|
|
def test_splt():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
splt_df = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
splt_analysis=5,
|
|
udps=SPLT(sequence_length=5, accounting_mode=0),
|
|
).to_pandas()
|
|
direction = json.loads(splt_df["udps.splt_direction"][0])
|
|
ps = json.loads(splt_df["udps.splt_ps"][0])
|
|
piat = json.loads(splt_df["udps.splt_piat_ms"][0])
|
|
ndirection = json.loads(splt_df["splt_direction"][0])
|
|
nps = json.loads(splt_df["splt_ps"][0])
|
|
npiat = json.loads(splt_df["splt_piat_ms"][0])
|
|
assert direction == [0, 1, 0, 0, 1]
|
|
assert ps == [58, 60, 54, 180, 60]
|
|
assert piat == [0, 34, 134, 144, 35]
|
|
assert direction == ndirection
|
|
assert ps == nps
|
|
assert piat == npiat
|
|
print("{}\t: {}".format(".test_splt".ljust(60, " "), "OK"))
|
|
|
|
@staticmethod
|
|
def test_dhcp():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
dhcp_df = (
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "dhcp.pcap"),
|
|
n_dissections=0,
|
|
udps=DHCP(),
|
|
)
|
|
.to_pandas()
|
|
.sort_values(by=["src_ip"])
|
|
.reset_index(drop=True)
|
|
)
|
|
assert dhcp_df["udps.dhcp_msg_type"][0] == "MsgType.DISCOVER"
|
|
assert dhcp_df["udps.dhcp_50"][1] == "192.168.0.10"
|
|
assert dhcp_df["udps.dhcp_55"][1] == "1,3,6,42"
|
|
assert dhcp_df["udps.dhcp_options"][1] == "[53, 61, 50, 54, 55]"
|
|
assert dhcp_df["udps.dhcp_msg_type"][1] == "MsgType.REQUEST"
|
|
assert dhcp_df["udps.dhcp_oui"][1] == "00:0b:82"
|
|
assert dhcp_df.shape[0] == 3
|
|
print("{}\t: {}".format(".test_dhcp".ljust(60, " "), "OK"))
|
|
|
|
@staticmethod
|
|
def test_mdns():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
mdns_df = (
|
|
NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "mdns.pcap"),
|
|
n_dissections=0,
|
|
udps=MDNS(),
|
|
)
|
|
.to_pandas()
|
|
.sort_values(by=["src_ip"])
|
|
.reset_index(drop=True)
|
|
)
|
|
assert (
|
|
mdns_df["udps.mdns_ptr"][0] == "['skynet.local', "
|
|
"'skynet [00:1a:ef:17:c3:05]._workstation._tcp.local', "
|
|
"'recombinator_mpd._mpd._tcp.local', '_mpd._tcp.local', "
|
|
"'skynet._udisks-ssh._tcp.local', '_udisks-ssh._tcp.local', "
|
|
"'_workstation._tcp.local']"
|
|
)
|
|
print("{}\t: {}".format(".test_mdns".ljust(60, " "), "OK"))
|
|
|
|
@staticmethod
|
|
def test_multi_files():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
multi_files = [
|
|
os.path.join("tests", "pcaps", "one_flow_1_5.pcap"),
|
|
os.path.join("tests", "pcaps", "one_flow_6_10.pcap"),
|
|
os.path.join("tests", "pcaps", "one_flow_11_15.pcap"),
|
|
os.path.join("tests", "pcaps", "one_flow_16_19.pcap"),
|
|
]
|
|
for flow in NFStreamer(source=multi_files):
|
|
assert flow.id == 0
|
|
assert flow.expiration_id == 0
|
|
assert flow.src_ip == "192.168.43.18"
|
|
assert flow.src_mac == "30:52:cb:6c:9c:1b"
|
|
assert flow.src_oui == "30:52:cb"
|
|
assert flow.src_port == 52066
|
|
assert flow.dst_ip == "66.220.156.68"
|
|
assert flow.dst_mac == "98:0c:82:d3:3c:7c"
|
|
assert flow.dst_oui == "98:0c:82"
|
|
assert flow.dst_port == 443
|
|
assert flow.protocol == 6
|
|
assert flow.ip_version == 4
|
|
assert flow.vlan_id == 0
|
|
assert flow.tunnel_id == 0
|
|
assert flow.bidirectional_first_seen_ms == 1472393122365
|
|
assert flow.bidirectional_last_seen_ms == 1472393123665
|
|
assert flow.bidirectional_duration_ms == 1300
|
|
assert flow.bidirectional_packets == 19
|
|
assert flow.bidirectional_bytes == 5745
|
|
assert flow.src2dst_first_seen_ms == 1472393122365
|
|
assert flow.src2dst_last_seen_ms == 1472393123408
|
|
assert flow.src2dst_duration_ms == 1043
|
|
assert flow.src2dst_packets == 9
|
|
assert flow.src2dst_bytes == 1345
|
|
assert flow.dst2src_first_seen_ms == 1472393122668
|
|
assert flow.dst2src_last_seen_ms == 1472393123665
|
|
assert flow.dst2src_duration_ms == 997
|
|
assert flow.dst2src_packets == 10
|
|
assert flow.dst2src_bytes == 4400
|
|
assert flow.application_name == "TLS.Facebook"
|
|
assert flow.application_category_name == "SocialNetwork"
|
|
assert flow.application_is_guessed == 0
|
|
assert flow.application_confidence == 6
|
|
assert flow.requested_server_name == "facebook.com"
|
|
# nDPI 5.0: client_fingerprint is now JA4 instead of JA3
|
|
assert flow.client_fingerprint == "t12d1310h2_27a29bd8d6e6_85173d161f9a"
|
|
assert flow.server_fingerprint == "2d1eb5817ece335c24904f516ad5da12"
|
|
assert flow.user_agent is None
|
|
assert flow.content_type is None
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_multi_files".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_optional_string_fields():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
optional_fields = [
|
|
"requested_server_name",
|
|
"client_fingerprint",
|
|
"server_fingerprint",
|
|
"user_agent",
|
|
"content_type",
|
|
]
|
|
|
|
def collect(**kwargs):
|
|
# n_meters=1 keeps flow ordering deterministic across both runs.
|
|
streamer = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "facebook.pcap"),
|
|
n_meters=1,
|
|
**kwargs
|
|
)
|
|
collected = [
|
|
{name: getattr(flow, name) for name in optional_fields}
|
|
for flow in streamer
|
|
]
|
|
del streamer
|
|
return collected
|
|
|
|
# Plugins switch the meter to sync mode, no plugin means non sync mode.
|
|
non_plugin_flows = collect()
|
|
plugin_flows = collect(udps=SPLT(sequence_length=5, accounting_mode=0))
|
|
assert len(non_plugin_flows) > 0
|
|
assert non_plugin_flows == plugin_flows
|
|
for flow_values in non_plugin_flows:
|
|
for name in optional_fields:
|
|
value = flow_values[name]
|
|
# Absent values are None, populated ones are non empty strings.
|
|
assert value is None or (isinstance(value, str) and value != "")
|
|
# facebook.pcap carries TLS metadata but no HTTP metadata.
|
|
assert flow_values["requested_server_name"] is not None
|
|
assert flow_values["client_fingerprint"] is not None
|
|
assert flow_values["server_fingerprint"] is not None
|
|
assert flow_values["user_agent"] is None
|
|
assert flow_values["content_type"] is None
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_optional_string_fields".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_anonymize_absent_values():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
|
|
df = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "google_ssl.pcap"),
|
|
n_meters=1,
|
|
udps=AnonymizerMixedValues(),
|
|
).to_pandas(
|
|
columns_to_anonymize=[
|
|
"udps.absent",
|
|
"udps.populated",
|
|
"udps.zero",
|
|
"udps.false",
|
|
"udps.array",
|
|
]
|
|
)
|
|
assert df.shape[0] > 0
|
|
for _, row in df.iterrows():
|
|
# An absent value must not be replaced by the digest of "", which would be
|
|
# constant across the export and indistinguishable from a real value.
|
|
assert pd.isna(row["udps.absent"])
|
|
# A populated value is anonymized.
|
|
assert isinstance(row["udps.populated"], str)
|
|
assert len(row["udps.populated"]) == 128
|
|
assert row["udps.populated"] != "value"
|
|
# 0 and False are legitimate values, not absent ones, so they are still
|
|
# anonymized: the guard must not degrade into a truthiness test.
|
|
assert isinstance(row["udps.zero"], str)
|
|
assert len(row["udps.zero"]) == 128
|
|
assert isinstance(row["udps.false"], str)
|
|
assert len(row["udps.false"]) == 128
|
|
# A plugin may store a numpy array, whose comparison does not yield a
|
|
# bool; the guard must not raise on it.
|
|
assert isinstance(row["udps.array"], str)
|
|
assert len(row["udps.array"]) == 128
|
|
print(
|
|
"{}\t: {}".format(
|
|
".test_anonymize_absent_values".ljust(60, " "), "OK"
|
|
)
|
|
)
|
|
|
|
@staticmethod
|
|
def test_max_nflows():
|
|
print(
|
|
"\n----------------------------------------------------------------------"
|
|
)
|
|
df = NFStreamer(source=os.path.join("tests", "pcaps", "skype.pcap")).to_pandas()
|
|
assert df.shape[0] == 294
|
|
df = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "skype.pcap"), max_nflows=100
|
|
).to_pandas()
|
|
assert df.shape[0] == 100
|
|
df = NFStreamer(
|
|
source=os.path.join("tests", "pcaps", "skype.pcap"), max_nflows=0
|
|
).to_pandas()
|
|
assert df.shape[0] == 294
|
|
print(
|
|
"{}\t: {}".format(".test_max_nflows".ljust(60, " "), "OK")
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# IMPORTANT: As NFStream input is network bytes, we rely on fuzzing techniques to ensure robustness.
|
|
# Fuzzing testing is part of Google OSS-Fuzz project.
|
|
# Github: https://github.com/google/oss-fuzz/tree/master/projects/nfstream
|
|
# Build status: https://oss-fuzz-build-logs.storage.googleapis.com/index.html#nfstream
|
|
|
|
NFStreamTest.test_source_parameter()
|
|
NFStreamTest.test_decode_tunnels_parameter()
|
|
NFStreamTest.test_bpf_filter_parameter()
|
|
NFStreamTest.test_promiscuous_mode_parameter()
|
|
NFStreamTest.test_snapshot_length_parameter()
|
|
NFStreamTest.test_socket_buffer_size_parameter()
|
|
NFStreamTest.test_idle_timeout_parameter()
|
|
NFStreamTest.test_active_timeout_parameter()
|
|
NFStreamTest.test_accounting_mode_parameter()
|
|
NFStreamTest.test_udps_parameter()
|
|
NFStreamTest.test_n_dissections_parameter()
|
|
NFStreamTest.test_system_visibility_mode_parameter()
|
|
NFStreamTest.test_system_visibility_poll_ms()
|
|
NFStreamTest.test_statistical_analysis_parameter()
|
|
NFStreamTest.test_splt_analysis_parameter()
|
|
NFStreamTest.test_n_meters_parameter()
|
|
NFStreamTest.test_max_nflows_parameter()
|
|
NFStreamTest.test_performance_report_parameter()
|
|
NFStreamTest.test_expiration_management()
|
|
NFStreamTest.test_tunnel_decoding()
|
|
NFStreamTest.test_statistical()
|
|
NFStreamTest.test_fingerprint_extraction()
|
|
NFStreamTest.test_export()
|
|
NFStreamTest.test_bpf()
|
|
NFStreamTest.test_ndpi_integration()
|
|
NFStreamTest.test_splt()
|
|
NFStreamTest.test_dhcp()
|
|
NFStreamTest.test_mdns()
|
|
NFStreamTest.test_multi_files()
|
|
NFStreamTest.test_optional_string_fields()
|
|
NFStreamTest.test_anonymize_absent_values()
|
|
NFStreamTest.test_max_nflows()
|