chore: update packet filter api

fleek-network · Jun 1, 2024 · 5fd83e8 · 5fd83e8
1 parent bb95e23
commit 5fd83e8
Show file tree

Hide file tree

Showing 4 changed files with 25 additions and 35 deletions.
diff --git a/etc/ebpf/service/src/config.rs b/etc/ebpf/service/src/config.rs
@@ -143,7 +143,7 @@ impl ConfigSource {
 
             let mut tmp_path = PathBuf::new();
             tmp_path.push(self.paths.tmp_dir.as_path());
-            tmp_path.push(&fname);
+            tmp_path.push(fname);
 
             let mut tmp = fs::File::create(tmp_path.as_path()).await?;
             let bytes = serde_json::to_string(&profile)?;
@@ -152,7 +152,7 @@ impl ConfigSource {
 
             let mut dst = PathBuf::new();
             dst.push(self.paths.profiles_dir.as_path());
-            dst.push(&fname);
+            dst.push(fname);
 
             fs::rename(tmp_path, dst).await?;
         }

diff --git a/etc/ebpf/service/src/filter.rs b/etc/ebpf/service/src/filter.rs
@@ -8,43 +8,31 @@ use crate::frame::{IpcServiceFrame, Pf, PACKET_FILTER_SERVICE};
 use crate::utils::write;
 
 /// Client to update dynamic list of packet filter rules.
-#[derive(Default, Debug)]
+#[derive(Debug)]
 pub struct PacketFilter {
-    inner: Option<UnixStream>,
+    inner: UnixStream,
 }
 
 impl PacketFilter {
-    pub fn new() -> Self {
-        PacketFilter::default()
+    pub fn new(&mut self, stream: UnixStream) -> Self {
+        Self { inner: stream }
     }
 
-    // Todo: remove this method.
-    pub fn init(&mut self, stream: UnixStream) {
-        self.inner = Some(stream);
-    }
-
-    pub async fn connect(&mut self) -> io::Result<()> {
-        let sock = self.inner.as_ref().expect("To be initialized");
+    pub async fn connect(&self) -> io::Result<()> {
         let service = PACKET_FILTER_SERVICE.to_le_bytes();
-        write(sock, Bytes::from(service.to_vec())).await
+        write(&self.inner, Bytes::from(service.to_vec())).await
     }
 
     pub async fn add(&self, addr: SocketAddrV4) -> io::Result<()> {
-        if let Some(stream) = &self.inner {
-            let frame = IpcServiceFrame::Pf(Pf { op: Pf::ADD, addr });
-            write(stream, frame.serialize_len_delimit()).await?;
-        }
-        Ok(())
+        let frame = IpcServiceFrame::Pf(Pf { op: Pf::ADD, addr });
+        write(&self.inner, frame.serialize_len_delimit()).await
     }
 
     pub async fn remove(&self, addr: SocketAddrV4) -> io::Result<()> {
-        if let Some(stream) = &self.inner {
-            let frame = IpcServiceFrame::Pf(Pf {
-                op: Pf::REMOVE,
-                addr,
-            });
-            write(stream, frame.serialize_len_delimit()).await?;
-        }
-        Ok(())
+        let frame = IpcServiceFrame::Pf(Pf {
+            op: Pf::REMOVE,
+            addr,
+        });
+        write(&self.inner, frame.serialize_len_delimit()).await
     }
 }
diff --git a/etc/ebpf/service/src/map/schema.rs b/etc/ebpf/service/src/map/schema.rs
@@ -148,23 +148,23 @@ impl FileRule {
         let mut result = String::new();
 
         if self.operations & Self::OPEN_MASK == Self::OPEN_MASK {
-            result.push_str("o");
+            result.push('o');
         }
 
         if self.operations & Self::READ_MASK == Self::READ_MASK {
-            result.push_str("r");
+            result.push('r');
         }
 
         if self.operations & Self::WRITE_MASK == Self::WRITE_MASK {
-            result.push_str("w");
+            result.push('w');
         }
 
         if self.operations & Self::EXEC_MASK == Self::EXEC_MASK {
-            result.push_str("x");
+            result.push('x');
         }
 
         if result.is_empty() {
-            result.push_str("-");
+            result.push('-');
         }
 
         result

diff --git a/etc/ebpf/service/src/utils.rs b/etc/ebpf/service/src/utils.rs
@@ -18,7 +18,7 @@ pub async fn write(socket: &UnixStream, bytes: Bytes) -> io::Result<()> {
                     break 'write;
                 },
                 Err(e) => {
-                    return Err(e.into());
+                    return Err(e);
                 },
             }
         }
@@ -31,6 +31,8 @@ pub async fn read(socket: &UnixStream) -> io::Result<Option<Bytes>> {
     let mut frame_len = 0;
     loop {
         socket.ready(Interest::READABLE).await?;
+        // Todo: address this.
+        #[allow(clippy::never_loop)]
         'read: loop {
             while frame_len == 0 && bytes_read < 8 {
                 match socket.try_read(&mut read_buf[bytes_read..]) {
@@ -45,7 +47,7 @@ pub async fn read(socket: &UnixStream) -> io::Result<Option<Bytes>> {
                         break 'read;
                     },
                     Err(e) => {
-                        return Err(e.into());
+                        return Err(e);
                     },
                 }
             }
@@ -75,7 +77,7 @@ pub async fn read(socket: &UnixStream) -> io::Result<Option<Bytes>> {
                         break 'read;
                     },
                     Err(e) => {
-                        return Err(e.into());
+                        return Err(e);
                     },
                 }
             }