Add integration test for events generated in containers

Because capturing information for a process running in a container is
not fully possible from `/proc`, some information is passed to the
Process constructor directly to overwrite it. This should be fine
though, because we are using a specific container that should not have
major changes on how it executes the command we are expecting.

Author: Molter73

Cargo.toml

@@ -24,7 +24,7 @@ log = { version = "0.4.22", default-features = false }
 prometheus-client = { version = "0.24.0", default-features = false }
 prost = "0.13.5"
 prost-types = "0.13.5"
-serde = { version = "1.0.219", features = ["derive"] }
+serde = { version = "1.0.219", features = ["derive", "rc"] }
 serde_json = "1.0.142"
 tokio = { version = "1.40.0", default-features = false, features = [
     "macros",

fact/src/cgroup.rs

@@ -17,7 +17,7 @@ use crate::host_info::get_cgroup_paths;
 
 #[derive(Debug)]
 struct ContainerIdEntry {
-    container_id: Option<String>,
+    container_id: Option<Arc<String>>,
     pub last_seen: SystemTime,
 }
 
@@ -51,7 +51,7 @@ impl ContainerIdCache {
         })
     }
 
-    pub async fn get_container_id(&self, cgroup_id: u64) -> Option<String> {
+    pub async fn get_container_id(&self, cgroup_id: u64) -> Option<Arc<String>> {
         let mut map = self.0.lock().await;
         match map.get(&cgroup_id) {
             Some(entry) => entry.container_id.clone(),
@@ -82,7 +82,7 @@ impl ContainerIdCache {
         })
     }
 
-    fn walk_cgroupfs(path: &PathBuf, map: &mut ContainerIdMap, parent_id: Option<&str>) {
+    fn walk_cgroupfs(path: &PathBuf, map: &mut ContainerIdMap, parent_id: Option<Arc<String>>) {
         for entry in std::fs::read_dir(path).unwrap() {
             let entry = match entry {
                 Ok(entry) => entry,
@@ -109,8 +109,8 @@ impl ContainerIdCache {
                         .unwrap_or("");
                     let container_id = match ContainerIdCache::extract_container_id(last_component)
                     {
-                        Some(cid) => Some(cid),
-                        None => parent_id.map(|f| f.to_owned()),
+                        Some(cid) => Some(Arc::new(cid)),
+                        None => parent_id.clone(),
                     };
                     let last_seen = SystemTime::now();
                     map.insert(
@@ -123,7 +123,7 @@ impl ContainerIdCache {
                     container_id
                 }
             };
-            ContainerIdCache::walk_cgroupfs(&p, map, container_id.as_deref());
+            ContainerIdCache::walk_cgroupfs(&p, map, container_id);
         }
     }
 

fact/src/event.rs

@@ -1,6 +1,6 @@
 #[cfg(test)]
 use std::time::{SystemTime, UNIX_EPOCH};
-use std::{ffi::CStr, os::raw::c_char, path::PathBuf};
+use std::{ffi::CStr, os::raw::c_char, path::PathBuf, sync::Arc};
 
 use fact_api::FileActivity;
 use serde::Serialize;
@@ -60,7 +60,7 @@ pub struct Process {
     comm: String,
     args: Vec<String>,
     exe_path: String,
-    container_id: Option<String>,
+    container_id: Option<Arc<String>>,
     uid: u32,
     username: &'static str,
     gid: u32,
@@ -126,7 +126,7 @@ impl Process {
             .unwrap();
         let args = std::env::args().collect::<Vec<_>>();
         let cgroup = std::fs::read_to_string("/proc/self/cgroup").expect("Failed to read cgroup");
-        let container_id = ContainerIdCache::extract_container_id(&cgroup);
+        let container_id = ContainerIdCache::extract_container_id(&cgroup).map(Arc::new);
         let uid = unsafe { libc::getuid() };
         let gid = unsafe { libc::getgid() };
         let pid = std::process::id();
@@ -182,19 +182,16 @@ impl From<Process> for fact_api::ProcessSignal {
             lineage,
         } = value;
 
-        let container_id = container_id.unwrap_or("".to_string());
-
-        let args = args
-            .into_iter()
-            .reduce(|acc, i| acc + " " + &i)
-            .unwrap_or("".to_owned());
+        let container_id = container_id
+            .map(Arc::unwrap_or_clone)
+            .unwrap_or("".to_string());
 
         Self {
             id: Uuid::new_v4().to_string(),
             container_id,
             creation_time: None,
             name: comm,
-            args,
+            args: args.join(" "),
             exec_file_path: exe_path,
             pid,
             uid,

tests/conftest.py

@@ -9,6 +9,7 @@
 import requests
 
 from server import FileActivityService
+from logs import dump_logs
 
 
 @pytest.fixture
@@ -46,6 +47,18 @@ def docker_client():
     return docker.from_env()
 
 
+@pytest.fixture(scope='session', autouse=True)
+def docker_api_client():
+    """
+    Create a docker API client, which is a lower level object and has
+    access to more methods than the regular client.
+
+    Returns:
+        A docker.APIClient object created with default values.
+    """
+    return docker.APIClient()
+
+
 @pytest.fixture
 def server():
     """
@@ -74,12 +87,6 @@ def get_image(request, docker_client):
         docker_client.images.pull(image)
 
 
-def dump_logs(container, file):
-    logs = container.logs().decode('utf-8')
-    with open(file, 'w') as f:
-        f.write(logs)
-
-
 @pytest.fixture
 def fact(request, docker_client, monitored_dir, server, logs_dir):
     """
@@ -120,6 +127,10 @@ def fact(request, docker_client, monitored_dir, server, logs_dir):
                 'bind': '/host/usr/lib/os-release',
                 'mode': 'ro',
             },
+            '/sys/fs/cgroup/': {
+                'bind': '/host/sys/fs/cgroup',
+                'mode': 'ro',
+            }
         },
     )
 

tests/event.py

@@ -36,7 +36,12 @@ class Process:
     Represents a process with its attributes.
     """
 
-    def __init__(self, pid: int | None = None):
+    def __init__(self,
+                 pid: int | None = None,
+                 comm: str | None = None,
+                 exe_path: str | None = None,
+                 args: list[str] | None = None,
+                 ):
         self._pid: int = pid if pid is not None else os.getpid()
         proc_dir = os.path.join('/proc', str(self._pid))
 
@@ -54,16 +59,23 @@ def get_id(line: str, wanted_id: str) -> int | None:
                 elif (gid := get_id(line, 'Gid')) is not None:
                     self._gid: int = gid
 
-        self._exe_path: str = os.path.realpath(os.path.join(proc_dir, 'exe'))
-
-        with open(os.path.join(proc_dir, 'cmdline'), 'rb') as f:
-            content = f.read(4096)
-            args = [arg.decode('utf-8')
-                    for arg in content.split(b'\x00') if arg]
-            self._args: str = ' '.join(args)
-
-        with open(os.path.join(proc_dir, 'comm'), 'r') as f:
-            self._name: str = f.read().strip()
+        self._exe_path: str = os.path.realpath(os.path.join(
+            proc_dir, 'exe')) if exe_path is None else exe_path
+
+        if args is None:
+            with open(os.path.join(proc_dir, 'cmdline'), 'rb') as f:
+                content = f.read(4096)
+                args = [arg.decode('utf-8')
+                        for arg in content.split(b'\x00') if arg]
+                self._args: str = ' '.join(args)
+        else:
+            self._args = ' '.join(args)
+
+        if comm is None:
+            with open(os.path.join(proc_dir, 'comm'), 'r') as f:
+                self._name: str = f.read().strip()
+        else:
+            self._name = comm
 
         with open(os.path.join(proc_dir, 'cgroup'), 'r') as f:
             self._container_id: str = extract_container_id(f.read())

tests/logs.py

@@ -0,0 +1,4 @@
+def dump_logs(container, file):
+    logs = container.logs().decode('utf-8')
+    with open(file, 'w') as f:
+        f.write(logs)

tests/test_file_open.py

@@ -1,8 +1,12 @@
+import json
 import multiprocessing as mp
 import os
 import subprocess
 
+import pytest
+
 from event import Event, EventType, Process
+from logs import dump_logs
 
 
 def test_open(fact, monitored_dir, server):
@@ -145,3 +149,40 @@ def do_test(fut: str, stop_event: mp.Event):
     finally:
         stop_event.set()
         proc.join(1)
+
+
+CONTAINER_CMD = 'mkdir -p {monitored_dir}; echo "Some content" > {monitored_dir}/test.txt ; sleep 5'
+
+
+@pytest.fixture(scope='function')
+def test_container(fact, docker_client, monitored_dir, logs_dir):
+    image = 'fedora:42'
+    command = f"bash -c '{CONTAINER_CMD.format(monitored_dir=monitored_dir)}'"
+    container_log = os.path.join(logs_dir, 'fedora.log')
+    container = docker_client.containers.run(
+        image,
+        detach=True,
+        command=command,
+    )
+    yield container
+    container.stop(timeout=1)
+    container.wait(timeout=1)
+    dump_logs(container, container_log)
+    container.remove()
+
+
+def test_container_event(fact, monitored_dir, server, test_container, docker_api_client):
+    fut = os.path.join(monitored_dir, 'test.txt')
+
+    inspect = docker_api_client.inspect_container(test_container.id)
+    p = Process(pid=inspect['State']['Pid'],
+                comm='bash',
+                exe_path='/usr/bin/bash',
+                args=['bash', '-c',
+                      CONTAINER_CMD.format(monitored_dir=monitored_dir)]
+                )
+
+    creation = Event(process=p, event_type=EventType.CREATION, file=fut)
+    print(f'Waiting for event: {creation}')
+
+    server.wait_events([creation])