Merge pull request #1395 from stratosphereips/alya/use_the_label_from_given_labeled_zeek_logs

use the label from given labeled zeek logs

Author: AlyaGomaa

modules/flowalerts/ssl.py

@@ -220,6 +220,8 @@ async def check_non_ssl_port_443_conns(
             self.keep_track_of_ssl_flow(flow, key)
             return False
 
+        flow.starttime = float(flow.starttime)
+
         # in seconds
         five_mins = 5 * 60
 

modules/flowmldetection/flowmldetection.py

@@ -316,6 +316,8 @@ def detect(self, x_flow) -> Optional[numpy.ndarray]:
                 "endtime",
                 "bytes",
                 "flow_source",
+                "ground_truth_label",  # todo now we can use them
+                "detailed_ground_truth_label",
             ]
             for field in fields_to_drop:
                 try:

modules/http_analyzer/http_analyzer.py

@@ -485,8 +485,8 @@ async def check_non_http_port_80_conns(
         we discard the evidence. if not found, we check for future 5 mins
         of matching zeek flows that were detected as http by zeek.
          if found, we dont set an evidence, if not found, we set an evidence
-        :kwarg timeout_reached: did we wait 5 mins in future and in the
-        past for the http of the given flow to arrive or not?
+        :kwarg timeout_reached: did we wait 5 mins in future AND in the
+        past for the http of the given flow to arrive?
         """
         if not self.is_tcp_established_port_80_non_empty_flow(flow):
             # we're not interested in that flow
@@ -500,6 +500,7 @@ async def check_non_http_port_80_conns(
             self.keep_track_of_http_flow(flow, key)
             return False
 
+        flow.starttime = float(flow.starttime)
         # in seconds
         five_mins = 5 * 60
 
@@ -522,7 +523,6 @@ async def check_non_http_port_80_conns(
 
         if matching_http_flows:
             # awesome! discard evidence. FP dodged.
-            # clear these timestamps as we dont need them anymore?
             return False
 
         # reaching here means we looked in the past 5 mins and
@@ -538,8 +538,8 @@ async def check_non_http_port_80_conns(
         # within that time?
         await self.wait_for_new_flows_or_timeout(five_mins)
         # we can safely await here without blocking the main thread because
-        # once the above await returns, this function will never sleep
-        # again, it'll either set the evidence or discard it
+        # once we run this func with timeout_reached=True, this function will
+        # never sleep again, it'll either set the evidence or discard it
         await self.check_non_http_port_80_conns(
             twid, flow, timeout_reached=True
         )
@@ -548,7 +548,9 @@ async def check_non_http_port_80_conns(
     async def wait_for_new_flows_or_timeout(self, timeout: float):
         """
         waits for new incoming flows, but interrupts the wait if profiler
-        stop sending new flows within within the timeout period.
+        stop sending new flows within the timeout period.
+        because that means no more flows are coming during the wait period,
+        no need to wait.
 
         :param timeout: the maximum time to wait before resuming execution.
         """
@@ -568,6 +570,7 @@ async def will_slips_have_new_incoming_flows():
             await asyncio.wait_for(
                 will_slips_have_new_incoming_flows(), timeout
             )
+
         except asyncio.TimeoutError:
             pass  # timeout reached
 
@@ -669,3 +672,5 @@ async def main(self):
             twid = msg["twid"]
             flow = self.classifier.convert_to_flow_obj(msg["flow"])
             self.create_task(self.check_non_http_port_80_conns, twid, flow)
+
+        self.remove_old_entries_from_http_recognized_flows()

modules/ip_info/ip_info.py

@@ -232,9 +232,13 @@ def get_vendor_offline(self, mac_addr, profileid):
 
     def get_vendor(self, mac_addr: str, profileid: str) -> dict:
         """
-        Returns the vendor info  of a MAC address and stores it in slips db
+        Returns the vendor info of a MAC address and stores it in slips db
         either from an offline or an online database
         """
+        if not utils.is_ignored_ip(profileid.split("_")[-1]):
+            # dont try to get the MAC vendor of private profiles, the MAC
+            # here is irrelevant (might be the gateway's)
+            return False
 
         if (
             "ff:ff:ff:ff:ff:ff" in mac_addr.lower()

slips_files/common/slips_utils.py

@@ -310,8 +310,8 @@ def convert_format(self, ts, required_format: str):
             return datetime_obj.astimezone(tz=self.local_tz).isoformat()
         elif required_format == "unixtimestamp":
             return datetime_obj.timestamp()
-        else:
-            return datetime_obj.strftime(required_format)
+
+        return datetime_obj.strftime(required_format)
 
     def get_local_timezone(self):
         """

slips_files/core/flows/zeek.py

@@ -34,11 +34,17 @@ class Conn:
     sbytes: int
     dbytes: int
 
-    smac: str
-    dmac: str
-
     state: str
     history: str
+
+    smac: str = ""
+    dmac: str = ""
+
+    # this is for when you give flows labeled by the netflow labeler
+    # https://github.com/stratosphereips/netflowlabeler
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "conn"
     dir_: str = "->"
 
@@ -74,6 +80,8 @@ class DNS:
 
     answers: List[str]
     TTLs: str
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
 
     type_: str = "dns"
 
@@ -108,6 +116,9 @@ class HTTP:
     resp_mime_types: str
     resp_fuids: str
 
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "http"
 
     def __post_init__(self) -> None:
@@ -143,6 +154,9 @@ class SSL:
     ja3s: str
     is_DoH: str
 
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "ssl"
 
 
@@ -168,29 +182,35 @@ class SSH:
     host_key_alg: str
     host_key: str
 
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "ssh"
 
 
 @dataclass
 class DHCP:
     starttime: float
     uids: List[str]
-    saddr: str
-    daddr: str
-
     client_addr: str
     server_addr: str
     host_name: str
 
     smac: str  # this is the client mac
     requested_addr: str
 
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "dhcp"
 
     def __post_init__(self) -> None:
         # Some zeek flow don't have saddr or daddr,
         # seen in dhcp.log and notice.log use the mac
         # address instead
+        self.saddr = self.client_addr
+        self.daddr = self.server_addr
+        # if the client_addr is empty, use the mac address
         if not self.saddr and not self.daddr:
             self.saddr = self.smac
 
@@ -203,6 +223,10 @@ class FTP:
     daddr: str
 
     used_port: int
+
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "ftp"
 
 
@@ -214,6 +238,10 @@ class SMTP:
     daddr: str
 
     last_reply: str
+
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "smtp"
 
 
@@ -230,6 +258,9 @@ class Tunnel:
     tunnel_type: str
     action: str
 
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "tunnel"
 
 
@@ -250,6 +281,10 @@ class Notice:
 
     # TODO srsly what is this?
     dst: str
+
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     # every evidence needs a uid, notice.log flows dont have one by
     # default, slips adds one to them to be able to deal with it.
     type_: str = "notice"
@@ -293,6 +328,9 @@ class Files:
     tx_hosts: List[str]
     rx_hosts: List[str]
 
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "files"
 
     def __post_init__(self) -> None:
@@ -338,6 +376,9 @@ class ARP:
     dpkts: str = ""
     appproto: str = ""
 
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "arp"
 
 
@@ -346,13 +387,18 @@ class Software:
     starttime: str
     uid: str
     saddr: str
-    daddr: str
+    sport: int
 
     software: str
 
     unparsed_version: str
     version_major: str
     version_minor: str
+    # software log lines dont have daddr
+    daddr: str = ""
+    ground_truth_label: str = ""
+    detailed_ground_truth_label: str = ""
+
     type_: str = "software"
 
     def __post_init__(self) -> None:

slips_files/core/input.py

@@ -271,9 +271,11 @@ def cache_nxt_line_in_file(self, filename: str):
         if not file_handle:
             return False
 
-        # Only read the next line if the previous line from this file was sent to profiler
+        # Only read the next line if the previous line from this file was sent
+        # to profiler
         if filename in self.cache_lines:
-            # We have still something to send, do not read the next line from this file
+            # We have still something to send, do not read the next line from
+            # this file
             return False
 
         # We don't have any waiting line for this file, so proceed
@@ -286,14 +288,21 @@ def cache_nxt_line_in_file(self, filename: str):
             return False
 
         # Did the file end?
-        if not zeek_line or zeek_line.startswith("#"):
+        if not zeek_line or zeek_line.startswith("#close"):
             # We reached the end of one of the files that we were reading.
             # Wait for more lines to come from another file
             return False
 
-        timestamp, nline = self.get_ts_from_line(zeek_line)
-        if not timestamp:
-            return False
+        if zeek_line.startswith("#fields"):
+            # this line contains the zeek fields, we want to cache it and
+            # send it to the profiler normally
+            nline = zeek_line
+            # to send the line as early as possible
+            timestamp = -1
+        else:
+            timestamp, nline = self.get_ts_from_line(zeek_line)
+            if not timestamp:
+                return False
 
         self.file_time[filename] = timestamp
         # Store the line in the cache