Extending NFStream and extracting features from other layers.

[neyney10]

I would like to extend this amazing tool to extract more features from other layers.
NFStream currently only extracts data related to flows/sessions, but we can look at the traffic in the prespective of Hosts (sent/recv flows/sessions/packets from host) and timed-windows.

So I've started to look into the engine's internals and came across the function of "packet_fanout"
reference

nfstream/nfstream/engine/engine_cc.c

Line 500 in ed42a79

int packet_fanout(struct nf_packet *nf_pkt, int mode, uint64_t hashval, int n_roots, int root_idx) {

which is responsible for distributing packets to meters by using a hash of the 6 tuple, i.e, packets of the same flow will go to the same meter.
reference

nfstream/nfstream/engine/engine_cc.c

Line 559 in ed42a79

hashval = nf_pkt->protocol + nf_pkt->vlan_id + iph->saddr + iph->daddr + nf_pkt->src_port + nf_pkt->dst_port;

This by itself make it hard for me to distribute all packets that are related to some host to a specific meter, and the data can be spread across meters, If I change the hash value computation to match a host, it will be problematic as well because each packet has source host (src ip) and destination host (dst ip) which means the system need to somehow remember which host has shared packets with some other hosts (as they share a flow/session), some way is to use the "direction" of the packet to separate the logic to "this host sent this packet" and "this host received this packet" which then each host can live in different meters, and a the associated flow can be in one of the meters only (or even more, should I maybe seperate bidrectional-flows to unidirectional-flows for this kind of scenario?), ofcourse, all fo this by still not maintaning a good balance of workload between meters.

I have 2 questions:

1. I noticed that each meter reads the whole pcap file (by working offline), and if it came across some packet that does not belong to it (by its hashval) then it just skip it (and use it as a time sync), what were the thoughts behind this? why not read the file in a single pass and fanout the packets to the respective meters, instead of letting each meter to read the file in separate which leads eventually that the same packets are read multiple times.
2. I really love what you have done in this project, and as I said I like to extend the amount of layers from which the tool extracts features and stats from, is it possible? and if so, would you recommend doing so? or is it better to just create a separate tool because NFStream is so tied to the idea of flows?

[aouinizied]

@neyney10 Thank you for opening this discussion. I was also thinking about host layer dynamics at some time but after focused on building a reliable flow computation engine. Now that we can tell that it's being used/tested by several users, we can indeed start to expand it to other layers.

1. Yes this strategy if adopted for offline mode (all platforms) and online mode on MacOS. On Linux platform (online mode) the fanout is done by the kernel and each meter opening a listening socket receive only a subpart of the load. Now for offline mode, things are more complicated. We are using processes instead of threads to provide real parallelism (avoid GIL limitations). Consequently, if we implement a reader process that span packets across meters, we will need an IPC channel between the reader and each meter. I tried to do it by several means (sockets, ZMQ, shared memory) and what I concluded is that when you do IPC at each packet, IPC communication overhead become a bottleneck and you lose the speed gain obtained by parallelism. So I switched a more simplistic approach (current implementation) and it turns out that it's effective. There is still IPC communication between meters and the streamer but as this communication is related to flow granularity, this is a sustainable cost. Keep in mind that whatever you are doing, you need to reduce the time spent per packet to achieve acceptable packets per second processing speed.
2. Thanks and I appreciate that you took the time to understand the internals of NFStream. The main goal of NFStream is to provide a common framework for all network data science research/projects etc. That's why it will be good if we add the host aggregation to it instead of a separate tool. What I was thinking is that we have mainly two directions that we can take. The first one is to let packets span across meters and computation is performed at a flow granularity. We can add an aggregation layer (different from NFStreamer), let's say HostStreamer that will compute received flows statistics per host and export based on inactive active time windowing. This option will be the simplest to implement as a first step however the per host statistics will be limited to informations that we can extract from flows.
   Second option is to implement a per host span of packets as you said and deal with an effective way to compute both flows and packets per host. This will need a considerable design and implementation efforts but if we are 2 working on that, this is feasable.

So what I propose is to provide us with more insights about what will be the per host event. An example:

• Host, Number of initiatedflows, TotalPackets, MeanInterFlowTime, etc?
  I think if we can define a draft of what we want as output, we can then discuss what need to be changed/implemented. What do you think about it?

Zied

[drnpkr]

Hi @neyney10,

I wonder what is the specific use case you try to achieve here? Is there a specific scenario in which NFStream turned out to come short? I agree with @aouinizied, the expected outputs must be specified in more detailed to comprehend the benefits of per-host statistics better.

I think that existing flow metering tools avoid per-host statistics for a reason. The processing demand required to keep track of flows on a per-host basis is relatively high. Those statistics that are mentioned above can be obtained currently by post-processing the observed flows. The current implementation might make this slightly limited but extending NFStream with functionality where the flows are being streamed via Kafka/Solace to a broker will likely do the trick.

[neyney10]

@aouinizied

1. Yeah, I noticed the overhead too, your approach was 2 to 4 times faster on my machine than the dispacher/fanout approach from one proccess to meters with IPC.
2. Currently I'm extracting Host features from aggregation of flows by working on the pandas dataframe created by NFStream, although it is much slower and memory consuming, but is the shortest way to do so (as you can extract a Host's feature by a single line on code with pandas), but I'm looking for a long term solution that can scale.
   I was thinking the same regarding if we can only use aggregation of flows and if it will be enough, I'm currently trying to extract features as Ivan Letteri did in his paper "MTA-KDD'19: A Dataset for Malware Traffic Detection", github: MTA-KDD-19, and the presentation can be found in ITASEC2020.pdf, as you can see, extracting features from aggregation of flows can work, the HostStreamer sounds good. (he extracts features such as, distribution of UDP, TCP, amount of connections, average, variance) however, I want a single tool for all posssible layers, there are papers that extract features from flow's clumps, which are groups of packets in the flow that goes in the same direction until the next packet which comes in goes to the other direction, and there are some papers that are working with time-series and time-windows.
   At first I thought about making a generic system that can accept a grouping function (to flows, host, clumps, time serieses and what not) that groups packets in a way that then the user can extract features from, this generic approach makes it much easier to extend and add more layers as the user may wish, although, it was slow (as It was written using Python's Scapy), So I was hoping maybe I could integrate one grouping function at a time inside the NFStream engine, it won't allow the user to extend new layers, but in time, the library would grow and the amount of layers available would be more than enough.
   Features for hosts such as you said: initiatedflows, TotalPackets, MeanInterFlowTime are good too.

@drnpkr Kafka is not even needed, for your idea to work, although, I dont agree with that it might demand more processing and memory, as NFStream does the same thing for flows, it keep flows in cache and update them, although, they do expire due to activity/inactivity which then might reduce memory, but we can do the same to hosts, i.e, row per activity time interval.

@aouinizied @drnpkr
What do you think? Am I jumping too high?

[aouinizied]

@neyney10

No, the idea sounds good for me and it will be a great contribution for the community. So let's do it.
I agree with the fact that a single class with internal cache management is enough for this task.
What I have in mind is as follow:

• We need first to write a test with a pcap file and a pandas extracted with NFStream. We can write a set of assertions (we compare with exact values) and it will allow us to have a clear view of what we want as a result per host (feature naming included). During the implementation, once a host feature implemented, we replace the exact value with the output.
• Let's say that we are going to implement a class NHStreamer (Network Host Sreamer), a sketch of it can be as follow:

NHStreamer(source=NFStreamer(classical NFStreamer parameters here), active_timeout = 999, inactive_timeout = 999, udps=HostPlugin())

for x in NFStreamer:
   print(x)

to_pandas() and to_csv() must be present too.

• Code from NFStreamer can be reused adapted to avoid renventing the wheel for cache mangement, expiration etc.
• We must provide a Plugin interface like the NFPlugin one to allow users to add their specific computation.

Finally, for features from features from flow's clumps, this can be simply done by writing an NFPlugin that will implement custom expiration strategy and based on this NFStreamer, we can pass it as source to NHStreamer.

Now the most important question is regarding the cache key. In my opinion, A host is the src_ip in the flow. It will be the cache key by default. However, we can manage to have an additional parameter defining the key strategy which can be (src, dst, both):

• src: src_ip of flow is the host. We update the entry related to this host.
• dst: dst_ip of flow is the host. We update the entry related to this host.
• both: we create/update 2 entrie per flow (this is a stressing scenario)

Is that OK for you as first steps?

Zied

[neyney10]

Yeah, sounds good, I can start writing some tests, how many features are too many? should we describe only some basic features and the rest of the features would be implemented in plugins?
similar to src2dst_, dst2src_, bidirectional_ naming convention, I thought of using sent_ and recv_.

• In case someone want features from both Flows and Hosts, I would like to make that possibility more clear, hence, I thoght that maybe it might be possible that in addition to

for host in NHStreamer:
   print(host)

There might be somehow an option to:

for flow, host in NHStreamer:
   print(flow.src_ip) 
   print(host.recv_packets) # number of packets the host received 

Instead of running NFStreamer and NHStreamer in seperate.

• The plugin interface should be similar to NFPLugin
  For example, the on_update(self, packet, flow) can be in hosts as on_update(self, flow, host), the on_init and on_expire should match the same idea.
• Regarding flow clumps, currently, a custom expiration strategy (with -1) won't work as the at the moment I receive a packet to the other direction (hence, the clump is already 'expired'), I need to forward that packet to the next clump, a new clump need to be created and initialized, hence, the problem is that it includes the current proccessed packet, although, we can maybe implement a new custom strategy we excludes current packet (of -2?), or maybe we better off creating a new streamer such as:

NHStreamer(
     source=NFStreamer(
        source=NFCStreamer(~parameters...~), # Network flow clump streamer
        ~other-parameters...~),
    active_timeout = 999, 
    inactive_timeout = 999, 
    udps=HostPlugin())

• Regarding key cache, at first I thoguht to use src_ip as key, but what about dst_ip? we don't update the destination host at all? we must update both hosts of the flow for each flow, my first implementation was to use only "ip" let it be src_ip or dst_ip, and when updating a host with a flow, I needed first to check if it was the src to dst of the flow, and then update accordingly (i.e, take src2dst or dst2src), is it a bad approach? it differs from your "both" key idea, as we store hosts in seperate keys, I'm still worrying about fanout though.

[aouinizied]

@neyney10 Yes let's go with this. An implementation with src_ip as key will simplify things for evryone and we can inprove it later. Are you still interested?

Regards,
Zied