Detect and block clients that violate the protocol

letmein-fwproto/src/lib.rs

@@ -35,6 +35,8 @@ pub enum FirewallOperation {
     Ack,
     /// Open a port.
     Open,
+    /// Block an address.
+    BlockAddr,
 }
 
 impl TryFrom<u16> for FirewallOperation {
@@ -44,10 +46,12 @@ impl TryFrom<u16> for FirewallOperation {
         const OPERATION_OPEN: u16 = FirewallOperation::Open as u16;
         const OPERATION_ACK: u16 = FirewallOperation::Ack as u16;
         const OPERATION_NACK: u16 = FirewallOperation::Nack as u16;
+        const OPERATION_BLOCKADDR: u16 = FirewallOperation::BlockAddr as u16;
         match value {
             OPERATION_OPEN => Ok(Self::Open),
             OPERATION_ACK => Ok(Self::Ack),
             OPERATION_NACK => Ok(Self::Nack),
+            OPERATION_BLOCKADDR => Ok(Self::BlockAddr),
             _ => Err(err!("Invalid FirewallMessage/Operation value")),
         }
     }
@@ -207,6 +211,18 @@ impl FirewallMessage {
         }
     }
 
+    /// Construct a new message that requests installing a firewall-block-address rule.
+    pub fn new_block_addr(addr: IpAddr) -> Self {
+        let (addr_type, addr) = addr_to_octets(addr);
+        Self {
+            operation: FirewallOperation::BlockAddr,
+            port_type: PortType::TcpUdp,
+            addr_type,
+            addr,
+            ..Default::default()
+        }
+    }
+
     /// Get the operation type from this message.
     pub fn operation(&self) -> FirewallOperation {
         self.operation
@@ -216,14 +232,16 @@ impl FirewallMessage {
     pub fn port(&self) -> Option<(PortType, u16)> {
         match self.operation {
             FirewallOperation::Open => Some((self.port_type, self.port)),
-            FirewallOperation::Ack | FirewallOperation::Nack => None,
+            FirewallOperation::Ack | FirewallOperation::Nack | FirewallOperation::BlockAddr => None,
         }
     }
 
     /// Get the `IpAddr` from this message.
     pub fn addr(&self) -> Option<IpAddr> {
         match self.operation {
-            FirewallOperation::Open => Some(octets_to_addr(self.addr_type, &self.addr)),
+            FirewallOperation::Open | FirewallOperation::BlockAddr => {
+                Some(octets_to_addr(self.addr_type, &self.addr))
+            }
             FirewallOperation::Ack | FirewallOperation::Nack => None,
         }
     }

letmeind/src/firewall_client.rs

@@ -51,7 +51,33 @@ impl FirewallClient {
         match msg_reply.operation() {
             FirewallOperation::Ack => Ok(()),
             FirewallOperation::Nack => Err(err!("The firewall rejected the port-open request")),
-            FirewallOperation::Open => Err(err!("Received invalid reply")),
+            FirewallOperation::Open | FirewallOperation::BlockAddr => {
+                Err(err!("Received invalid reply"))
+            }
+        }
+    }
+
+    pub async fn block_addr(&mut self, addr: IpAddr) -> ah::Result<()> {
+        // Send an block-address request to the firewall daemon.
+        FirewallMessage::new_block_addr(addr)
+            .send(&mut self.stream)
+            .await
+            .context("Send block-addr message")?;
+
+        // Receive the block-addr reply.
+        let Some(msg_reply) = FirewallMessage::recv(&mut self.stream)
+            .await
+            .context("Receive block-addr reply")?
+        else {
+            return Err(err!("Connection terminated"));
+        };
+
+        match msg_reply.operation() {
+            FirewallOperation::Ack => Ok(()),
+            FirewallOperation::Nack => Err(err!("The firewall rejected the block-addr request")),
+            FirewallOperation::Open | FirewallOperation::BlockAddr => {
+                Err(err!("Received invalid reply"))
+            }
         }
     }
 }

letmeind/src/main.rs

@@ -12,10 +12,11 @@
 std::compile_error!("letmeind server does not support non-Linux platforms.");
 
 mod firewall_client;
+mod monitor;
 mod protocol;
 mod server;
 
-use crate::{protocol::Protocol, server::Server};
+use crate::{monitor::Monitor, protocol::Protocol, server::Server};
 use anyhow::{self as ah, format_err as err, Context as _};
 use clap::Parser;
 use letmein_conf::{Config, ConfigVariant, Seccomp};
@@ -189,6 +190,9 @@ async fn async_main(opts: Arc<Opts>) -> ah::Result<()> {
     // Create async IPC channels.
     let (exit_sock_tx, mut exit_sock_rx) = sync::mpsc::channel(1);
 
+    // Create client monitor.
+    let monitor = Arc::new(Monitor::new(&opts.rundir));
+
     // Start the TCP control port listener.
     let srv = Server::new(&*conf.read().await, opts.no_systemd)
         .await
@@ -205,21 +209,27 @@ async fn async_main(opts: Arc<Opts>) -> ah::Result<()> {
     task::spawn({
         let conf = Arc::clone(&conf);
         let opts = Arc::clone(&opts);
+        let monitor = Arc::clone(&monitor);
 
         async move {
             let conn_semaphore = Semaphore::new(opts.num_connections);
             loop {
                 let conf = Arc::clone(&conf);
                 let opts = Arc::clone(&opts);
+                let monitor = Arc::clone(&monitor);
+
                 match srv.accept().await {
                     Ok(conn) => {
                         // Socket connection handler.
                         if let Ok(_permit) = conn_semaphore.acquire().await {
                             task::spawn(async move {
                                 let conf = conf.read().await;
                                 let mut proto = Protocol::new(conn, &conf, &opts.rundir);
+                                let ipaddr = proto.addr().ip();
                                 if let Err(e) = proto.run().await {
-                                    eprintln!("Client '{}' ERROR: {}", proto.addr().ip(), e);
+                                    monitor.log_client_error(ipaddr, e).await;
+                                } else {
+                                    monitor.log_client_auth_ok(ipaddr).await;
                                 }
                             });
                         }

letmeind/src/monitor.rs

@@ -0,0 +1,122 @@
+// -*- coding: utf-8 -*-
+//
+// Copyright (C) 2024 Michael Büsch <[email protected]>
+//
+// Licensed under the Apache License version 2.0
+// or the MIT license, at your option.
+// SPDX-License-Identifier: Apache-2.0 OR MIT
+
+use crate::firewall_client::FirewallClient;
+use anyhow as ah;
+use std::{
+    collections::HashMap,
+    net::IpAddr,
+    path::{Path, PathBuf},
+    time::{Duration, Instant},
+};
+use tokio::sync::Mutex;
+
+const MON_ERROR_THRES: u32 = 3;
+const MON_TIMEOUT: Duration = Duration::from_millis(1000 * 60 * 60 * 12);
+
+struct MonitorData {
+    errors: u32,
+    timeout: Instant,
+}
+
+impl MonitorData {
+    fn new() -> Self {
+        Self {
+            errors: 0,
+            timeout: Instant::now() + MON_TIMEOUT,
+        }
+    }
+}
+
+struct MonitorInner {
+    rundir: PathBuf,
+    clients: HashMap<IpAddr, MonitorData>,
+}
+
+impl MonitorInner {
+    fn new(rundir: PathBuf) -> Self {
+        Self {
+            rundir,
+            clients: HashMap::new(),
+        }
+    }
+
+    async fn maintenance(&mut self) {
+        // Remove timed-out monitoring entries.
+        let now = Instant::now();
+        self.clients.retain(|_, mon_data| mon_data.timeout < now);
+    }
+
+    async fn log_client_error(&mut self, addr: IpAddr, error: ah::Error) {
+        eprintln!("Client '{addr}' ERROR: {error}");
+
+        // Remove timed-out monitoring entries.
+        self.maintenance().await;
+
+        // Get the monitoring state of this client.
+        let mon_data = self.clients.entry(addr).or_insert_with(MonitorData::new);
+
+        // Increment the error count.
+        mon_data.errors = mon_data.errors.saturating_add(1);
+
+        // If the error count is above a threshold, block the client.
+        if mon_data.errors >= MON_ERROR_THRES {
+            eprintln!("Monitor: Blocking client {addr}");
+
+            // Connect to letmeinfwd unix socket.
+            let mut fw = match FirewallClient::new(&self.rundir).await {
+                Ok(fw) => fw,
+                Err(e) => {
+                    eprintln!("Monitor: Failed to connect to `letmeinfwd`: {e}");
+                    return;
+                }
+            };
+
+            // Block the IP address in the firewall.
+            if let Err(e) = fw.block_addr(addr).await {
+                eprintln!("Monitor: Failed block client {addr} via `letmeinfwd`: {e}");
+                return;
+            }
+
+            // The client has been blocked in the firewall.
+            // Do not track it in the monitoring any longer.
+            self.clients.remove(&addr);
+        }
+    }
+
+    async fn log_client_auth_ok(&mut self, addr: IpAddr) {
+        // The client has successfully authenticated.
+        // Do not track it in the monitoring any longer.
+        self.clients.remove(&addr);
+
+        // Remove timed-out monitoring entries.
+        self.maintenance().await;
+    }
+}
+
+pub struct Monitor {
+    inner: Mutex<MonitorInner>,
+}
+
+impl Monitor {
+    pub fn new(rundir: &Path) -> Self {
+        Self {
+            inner: Mutex::new(MonitorInner::new(rundir.to_path_buf())),
+        }
+    }
+
+    pub async fn log_client_error(&self, addr: IpAddr, error: ah::Error) {
+        self.inner.lock().await.log_client_error(addr, error).await;
+    }
+
+    pub async fn log_client_auth_ok(&self, addr: IpAddr) {
+        self.inner.lock().await.log_client_auth_ok(addr).await;
+    }
+}
+
+// vim: ts=4 sw=4 expandtab

letmeinfwd/src/server.rs

@@ -136,6 +136,9 @@ impl FirewallConnection {
                     self.send_msg(&FirewallMessage::new_nack()).await?;
                 }
             }
+            FirewallOperation::BlockAddr => {
+                //TODO
+            }
             FirewallOperation::Ack | FirewallOperation::Nack => {
                 return Err(err!("Received invalid message"));
             }