Thalamus support

Cargo.toml

@@ -46,7 +46,7 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde_yaml = "0.9"
 serde_ignored = "0.1"
-clap = { version = "4.6.0", features = ["derive"] }
+clap = { version = "4.5.54", features = ["derive"] }
 nix = { version = "0.31.1", features = ["net", "fs", "resource"] }
 redis = { version = "1.0", features = [
   "tokio-native-tls-comp",
@@ -55,7 +55,7 @@ redis = { version = "1.0", features = [
 ] }
 native-tls = "0.2"
 tokio-rustls = "0.26.4"
-rustls = { version = "0.23.37", default-features = false, features = [
+rustls = { version = "0.23.36", default-features = false, features = [
   "std",
   "ring",
   "logging",
@@ -93,7 +93,7 @@ log4rs = { version = "1.3", features = [
 ] }
 syslog = "7.0"
 jsonwebtoken = { version = "10.1", features = ["rust_crypto"] }
-uuid = { version = "1.22", features = ["v4", "serde"] }
+uuid = { version = "1.20", features = ["v4", "serde"] }
 url = "2.5"
 clamav-tcp = "0.2"
 multer = "3.0"
@@ -110,27 +110,28 @@ daemonize = "0.5.0"
 # pingora-memory-cache = { path = "../pingora/pingora-memory-cache" }
 
 wirefilter-engine = { git = "https://github.com/gen0sec/wirefilter", rev = "ab901470a24aad789cb9c03dd214d6c7d4cab589" }
-pingora = { git = "https://github.com/gen0sec/pingora", rev = "e82e0689ce62280f85c659dbfd862e49197133ad", features = [
+pingora = { git = "https://github.com/gen0sec/pingora", rev = "c92146d621542303dd9b93a4cb5252e1eef46c81", features = [
   "lb",
   "openssl",
   "proxy",
 ] }
-pingora-core = { git = "https://github.com/gen0sec/pingora", rev = "e82e0689ce62280f85c659dbfd862e49197133ad" }
-pingora-proxy = { git = "https://github.com/gen0sec/pingora", rev = "e82e0689ce62280f85c659dbfd862e49197133ad" }
-pingora-limits = { git = "https://github.com/gen0sec/pingora", rev = "e82e0689ce62280f85c659dbfd862e49197133ad" }
-pingora-http = { git = "https://github.com/gen0sec/pingora", rev = "e82e0689ce62280f85c659dbfd862e49197133ad" }
-pingora-memory-cache = { git = "https://github.com/gen0sec/pingora", rev = "e82e0689ce62280f85c659dbfd862e49197133ad" }
+pingora-core = { git = "https://github.com/gen0sec/pingora", rev = "c92146d621542303dd9b93a4cb5252e1eef46c81" }
+pingora-proxy = { git = "https://github.com/gen0sec/pingora", rev = "c92146d621542303dd9b93a4cb5252e1eef46c81" }
+pingora-limits = { git = "https://github.com/gen0sec/pingora", rev = "c92146d621542303dd9b93a4cb5252e1eef46c81" }
+pingora-http = { git = "https://github.com/gen0sec/pingora", rev = "c92146d621542303dd9b93a4cb5252e1eef46c81" }
+pingora-memory-cache = { git = "https://github.com/gen0sec/pingora", rev = "c92146d621542303dd9b93a4cb5252e1eef46c81" }
 
 # JA4+ fingerprinting library
 #nstealth = { path = "../nstealth" }
-nstealth = { git = "https://github.com/gen0sec/nstealth", rev = "f9a6323d41acf8b7d45e33ae7e4e863ec716ba05" }
+nstealth = { git = "https://github.com/gen0sec/nstealth", rev = "3c87751b9d9537b055a119f155a730360a7d0078" }
 
 mimalloc = { version = "0.1.48", default-features = false }
+crossbeam-channel = "0.5"
 dashmap = "7.0.0-rc2"
 ctrlc = "3.5.0"
 arc-swap = "1.8.0"
 prometheus = "0.14.0"
-once_cell = "1.21.4"
+once_cell = "1.21.3"
 maxminddb = "0.27"
 memmap2 = "0.9"
 axum-server = { version = "0.8.0", features = ["tls-openssl"] }
@@ -155,12 +156,14 @@ libbpf-rs = { version = "0.26.0", optional = true }
 [target.'cfg(target_os = "linux")'.dependencies]
 nftables = "0.6"
 iptables = "0.6"
+thalamus = { git = "https://github.com/gen0sec/thalamus", branch = "blocking", default-features = false, optional = true }
 
 [dev-dependencies]
 serial_test = "3.3"
-tempfile = "3.27.0"
+tempfile = "3.25.0"
 
 [features]
 default = ["bpf"]
 bpf = ["dep:libbpf-rs", "dep:libbpf-cargo", "dep:vmlinux"]
 disable-bpf = []
+thalamus-ids = ["dep:thalamus"]

build.rs

@@ -18,17 +18,7 @@ fn main() {
     println!("cargo:rerun-if-changed={}", JA4TS_SRC);
     println!("cargo:rerun-if-changed={}/filter.h", HEADER_DIR);
     println!("cargo:rerun-if-changed={}/xdp_maps.h", HEADER_DIR);
-    println!(
-        "cargo:rerun-if-changed={}/lib/tcp_fingerprinting.h",
-        HEADER_DIR
-    );
-    println!("cargo:rerun-if-changed={}/lib/firewall.h", HEADER_DIR);
-    println!("cargo:rerun-if-changed={}/lib/helper.h", HEADER_DIR);
-    println!(
-        "cargo:rerun-if-changed={}/lib/latency_tracking.h",
-        HEADER_DIR
-    );
-    println!("cargo:rerun-if-changed={}/include/common.h", HEADER_DIR);
+    println!("cargo:rerun-if-changed={}/lib/ids_export.h", HEADER_DIR);
 
     let bpf_disabled = env::var_os("CARGO_FEATURE_DISABLE_BPF").is_some();
     let bpf_feature_set = env::var_os("CARGO_FEATURE_BPF").is_some();

src/bpf/lib/ratelimit.h

@@ -0,0 +1,108 @@
+#pragma once
+
+#include "common.h"
+
+#include "../xdp_maps.h"
+#include "firewall.h"
+#include "vmlinux.h"
+
+struct ratelimiter_config_t {
+  __u64 TOKENS_PER_REQUEST;  // tokens consumed per request
+  __u64 REFILL_RATE;         // tokens added per sec
+  __u64 MAX_BUCKET_CAPACITY; // refill_rate * (max_bucket_capacity /
+                             // refill_rate) = allow x request burst
+};
+
+// set this before loading the script
+volatile struct ratelimiter_config_t ratelimiter_config = {1, 10, 100};
+
+static __always_inline void refill_tokens(struct ratelimit_bucket_value *rl_val,
+                                          __u64 now) {
+  __u64 elapsed_ns = now - rl_val->last_topup;
+
+  // Calculate tokens to add:  (elapsed_seconds * REFILL_RATE)
+  __u64 tokens_to_add =
+      (elapsed_ns * ratelimiter_config.REFILL_RATE) / 1000000000ULL;
+
+  if (tokens_to_add > 0) {
+    rl_val->num_of_tokens += tokens_to_add;
+
+    if (rl_val->num_of_tokens > ratelimiter_config.MAX_BUCKET_CAPACITY) {
+      rl_val->num_of_tokens = ratelimiter_config.MAX_BUCKET_CAPACITY;
+    }
+
+    rl_val->last_topup = now;
+  }
+}
+
+static __noinline __u8 ipv4_syn_ratelimit(__be32 *addr, struct tcphdr *tcph) {
+  __u64 now = bpf_ktime_get_ns();
+
+  struct ratelimit_bucket_value *bucket =
+      bpf_map_lookup_elem(&ipv4_syn_bucket_store, addr);
+
+  if (!bucket) {
+    struct ratelimit_bucket_value new_bucket = {};
+    new_bucket.last_topup = now;
+    new_bucket.num_of_tokens = ratelimiter_config.MAX_BUCKET_CAPACITY;
+    bpf_map_update_elem(&ipv4_syn_bucket_store, &addr, &new_bucket, BPF_ANY);
+    bpf_printk("Bucket created for addr: %d", addr);
+    return XDP_PASS;
+  }
+
+  refill_tokens(bucket, now);
+
+  if (bucket->num_of_tokens >= ratelimiter_config.TOKENS_PER_REQUEST) {
+    bucket->num_of_tokens -= ratelimiter_config.TOKENS_PER_REQUEST;
+    bpf_printk("Packet passed for addr: %d", addr);
+    return XDP_PASS;
+  }
+
+  bpf_printk("Packet dropped for addr: %d", addr);
+  return XDP_DROP;
+}
+
+static __noinline __u8 ipv6_syn_ratelimit(ipv6_addr *addr,
+                                          struct tcphdr *tcph) {
+  __u64 now = bpf_ktime_get_ns();
+
+  struct ratelimit_bucket_value *bucket =
+      bpf_map_lookup_elem(&ipv6_syn_bucket_store, addr);
+
+  if (!bucket) {
+    struct ratelimit_bucket_value new_bucket = {};
+    new_bucket.last_topup = now;
+    new_bucket.num_of_tokens = ratelimiter_config.MAX_BUCKET_CAPACITY;
+    bpf_map_update_elem(&ipv6_syn_bucket_store, addr, &new_bucket, BPF_ANY);
+
+    return XDP_PASS;
+  }
+
+  refill_tokens(bucket, now);
+
+  if (bucket->num_of_tokens >= ratelimiter_config.TOKENS_PER_REQUEST) {
+    bucket->num_of_tokens -= ratelimiter_config.TOKENS_PER_REQUEST;
+    return XDP_PASS;
+  }
+
+  return XDP_DROP;
+}
+
+int __noinline xdp_ratelimit(struct iphdr *iph, struct ipv6hdr *ip6h,
+                             struct tcphdr *tcph) {
+  if (tcph) {
+    if (iph) {
+      if (ipv4_syn_ratelimit(&iph->saddr, tcph) == XDP_DROP) {
+        return XDP_DROP;
+      }
+    } else if (ip6h) {
+      ipv6_addr *ipv6_saddr_ptr = (ipv6_addr *)&ip6h->saddr;
+
+      if (ipv6_syn_ratelimit(ipv6_saddr_ptr, tcph) == XDP_DROP) {
+        return XDP_DROP;
+      }
+    }
+  }
+
+  return XDP_PASS;
+}

src/cli.rs

@@ -175,7 +175,7 @@ fn default_log_sending_enabled() -> bool {
 }
 
 fn default_include_response_body() -> bool {
-    false
+    true
 }
 
 fn default_max_body_size() -> usize {
@@ -295,7 +295,7 @@ impl Config {
                 base_url: "https://api.gen0sec.com/v1".to_string(),
                 threat: GeoipDatabaseConfig::default(),
                 log_sending_enabled: true,
-                include_response_body: false,
+                include_response_body: true,
                 max_body_size: 1024 * 1024, // 1MB
                 captcha: CaptchaConfig {
                     site_key: None,
@@ -550,7 +550,7 @@ impl Config {
             }
         }
         if let Ok(val) = env::var("AX_ARXIGNIS_INCLUDE_RESPONSE_BODY") {
-            self.arxignis.include_response_body = val.parse().unwrap_or(false);
+            self.arxignis.include_response_body = val.parse().unwrap_or(true);
         }
         if let Ok(val) = env::var("AX_ARXIGNIS_MAX_BODY_SIZE") {
             self.arxignis.max_body_size = val.parse().unwrap_or(1024 * 1024);
@@ -682,8 +682,8 @@ pub struct Args {
     #[arg(long)]
     pub arxignis_log_sending_enabled: Option<bool>,
 
-    /// Include response body in access logs (may capture sensitive data)
-    #[arg(long, default_value_t = false)]
+    /// Include response body in access logs
+    #[arg(long, default_value_t = true)]
     pub arxignis_include_response_body: bool,
 
     /// Maximum size for request/response bodies in access logs (bytes)
@@ -905,15 +905,6 @@ pub struct PingoraConfig {
     pub healthcheck_interval: u16,
     #[serde(default)]
     pub proxy_protocol: ProxyProtocolConfig,
-    /// Enable HTTP/2 cleartext (h2c) on the plaintext HTTP listener
-    #[serde(default)]
-    pub h2c: bool,
-    /// Allow proxying HTTP CONNECT requests (tunneling/WebSocket upgrades)
-    #[serde(default)]
-    pub allow_connect_method_proxying: bool,
-    /// Maximum requests per connection before closing (default: no limit)
-    #[serde(default)]
-    pub keepalive_request_limit: Option<u32>,
 }
 
 fn default_pingora_tls_grade() -> String {
@@ -994,9 +985,6 @@ impl PingoraConfig {
         app_config.healthcheck_method = self.healthcheck_method.clone();
         app_config.healthcheck_interval = self.healthcheck_interval;
         app_config.proxy_protocol_enabled = self.proxy_protocol.enabled;
-        app_config.h2c = self.h2c;
-        app_config.allow_connect_method_proxying = self.allow_connect_method_proxying;
-        app_config.keepalive_request_limit = self.keepalive_request_limit;
 
         // Parse config_address to local_server
         if let Some((ip, port_str)) = self.config_address.split_once(':') {