new: tracking per-interface network usage

dyana/cli.py

@@ -17,6 +17,7 @@
     view_gpus,
     view_header,
     view_network_events,
+    view_network_usage,
     view_process_executions,
     view_ram,
     view_security_events,
@@ -153,6 +154,7 @@ def summary(trace_path: pathlib.Path = typer.Option(help="Path to the trace file
     view_disk_usage(trace["run"])
 
     view_process_executions(trace)
+    view_network_usage(trace["run"])
     view_network_events(trace)
     view_disk_events(trace)
     view_security_events(trace)

dyana/loaders/base/dyana.py

@@ -30,6 +30,7 @@ def __init__(self, gpu: bool = False):
         self._disk: dict[str, int] = {"start": get_disk_usage()}
         self._ram: dict[str, int] = {"start": get_peak_rss()}
         self._gpu: dict[str, list[dict[str, t.Any]]] = {"start": get_gpu_usage()} if gpu else {}
+        self._network: dict[str, dict[str, int]] = {"start": get_network_stats()}
         self._imports_at_start = get_current_imports()
         self._additionals: dict[str, t.Any] = {}
 
@@ -41,6 +42,9 @@ def track_memory(self, event: str) -> None:
     def track_disk(self, event: str) -> None:
         self._disk[event] = get_disk_usage()
 
+    def track_network(self, event: str) -> None:
+        self._network[event] = get_network_stats()
+
     def track_error(self, event: str, error: str) -> None:
         self._errors[event] = error
 
@@ -51,9 +55,13 @@ def as_dict(self) -> dict[str, t.Any]:
         imports_at_end = get_current_imports()
         imported = {k: imports_at_end[k] for k in imports_at_end if k not in self._imports_at_start}
 
+        if len(self._network.keys()) == 1:
+            self.track_network("end")
+
         as_dict: dict[str, t.Any] = {
             "ram": self._ram,
             "disk": self._disk,
+            "network": self._network,
             "errors": self._errors,
             "extra": {"imports": imported},
         } | self._additionals
@@ -142,3 +150,29 @@ def get_current_imports() -> dict[str, str | None]:
             imports[module_name] = module.__dict__["__file__"] if "__file__" in module.__dict__ else None
 
     return imports
+
+
+def get_network_stats() -> dict[str, dict[str, int]]:
+    """
+    Parse /proc/net/dev and return a dictionary of network interface statistics.
+    Returns a dictionary where each key is an interface name and each value is
+    a dictionary containing bytes_received and bytes_sent.
+    """
+    stats: dict[str, dict[str, int]] = {}
+
+    with open("/proc/net/dev") as f:
+        # skip the first two header lines
+        next(f)
+        next(f)
+
+        for line in f:
+            # split the line into interface name and statistics
+            parts = line.strip().split(":")
+            if len(parts) != 2:
+                continue
+
+            interface = parts[0].strip()
+            values = parts[1].split()
+            stats[interface] = {"rx": int(values[0]), "tx": int(values[8])}
+
+    return stats

dyana/loaders/loader.py

@@ -23,6 +23,11 @@ class GpuDeviceUsage(BaseModel):
     free_memory: int
 
 
+class NetworkDeviceUsage(BaseModel):
+    rx: int
+    tx: int
+
+
 class Run(BaseModel):
     loader_name: str | None = None
     build_platform: str | None = None
@@ -33,6 +38,7 @@ class Run(BaseModel):
     ram: dict[str, int] | None = None
     gpu: dict[str, list[GpuDeviceUsage]] | None = None
     disk: dict[str, int] | None = None
+    network: dict[str, dict[str, NetworkDeviceUsage]] | None = None
     stdout: str | None = None
     stderr: str | None = None
     exit_code: int | None = None

dyana/view.py

@@ -184,11 +184,55 @@ def view_process_executions(trace: dict[str, t.Any]) -> None:
         print()
 
 
+def view_network_usage(run: dict[str, t.Any]) -> None:
+    has_network_usage = "network" in run and run["network"]
+    if has_network_usage:
+        network = run["network"]
+        stages = list(network.keys())
+        interfaces = list(network[stages[0]].keys())
+        any_change = False
+
+        for interface in interfaces:
+            for stage in stages:
+                if network[stage][interface]["rx"] > 0 or network[stage][interface]["tx"] > 0:
+                    any_change = True
+                    break
+
+        if any_change:
+            print("[bold yellow]Network Usage:[/]")
+
+            for interface in interfaces:
+                # Check if there were any network changes across stages
+                had_network_activity = False
+                for stage in stages:
+                    if network[stage][interface]["rx"] > 0 or network[stage][interface]["tx"] > 0:
+                        had_network_activity = True
+                        break
+
+                if not had_network_activity:
+                    continue
+
+                print(f"  [bold]{interface}[/]")
+                prev_stage = None
+                for stage in stages:
+                    if prev_stage is None:
+                        print(
+                            f"    {stage} : rx={sizeof_fmt(network[stage][interface]["rx"])} tx={sizeof_fmt(network[stage][interface]["tx"])}"
+                        )
+                    else:
+                        print(
+                            f"    {stage} : rx={delta_fmt(network[prev_stage][interface]["rx"], network[stage][interface]["rx"])} tx={delta_fmt(network[prev_stage][interface]["tx"], network[stage][interface]["tx"])}"
+                        )
+                    prev_stage = stage
+
+                print()
+
+
 def view_network_events(trace: dict[str, t.Any]) -> None:
     connects = [event for event in trace["events"] if event["eventName"] == "security_socket_connect"]
     dns_queries = [event for event in trace["events"] if event["eventName"] == "net_packet_dns"]
     if connects or dns_queries:
-        print("[bold yellow]Network:[/]")
+        print("[bold yellow]Network Activity:[/]")
 
         all = connects + dns_queries
         all.sort(key=lambda e: e["timestamp"])
@@ -229,6 +273,7 @@ def view_network_events(trace: dict[str, t.Any]) -> None:
                     print(line)
 
         print()
+        quit()
 
 
 def view_disk_events(trace: dict[str, t.Any]) -> None: