From 36ab272bd413357fe184327bb1b0cf140bd3513b Mon Sep 17 00:00:00 2001 From: onelrian Date: Fri, 7 Aug 2026 11:40:58 +0100 Subject: [PATCH 1/3] ci: add test gate, cargo-deny, Trivy scan, multi-arch, releases Closes #9 Reproducible multi-stage Dockerfile (docker build . now works standalone), digest-pinned base images, and a test job gating the build. Bumped bytes/anyhow to clear two real RUSTSEC advisories cargo-deny caught. Tag pushes now cut a GitHub Release. --- .dockerignore | 8 +- .github/workflows/ci.yml | 146 +++++++++++++++++++++++++++ .github/workflows/docker-publish.yml | 85 ---------------- Cargo.lock | 8 +- Dockerfile | 29 +++++- README.md | 2 +- deny.toml | 30 ++++++ 7 files changed, 209 insertions(+), 99 deletions(-) create mode 100644 .github/workflows/ci.yml delete mode 100644 .github/workflows/docker-publish.yml create mode 100644 deny.toml diff --git a/.dockerignore b/.dockerignore index ed5e0c9..62447e7 100644 --- a/.dockerignore +++ b/.dockerignore @@ -2,11 +2,9 @@ .git .gitignore -# Rust build output (Exclude debug artifacts, keep release binary for CI COPY) -target/debug -target/doc -target/package -target/test +# Rust build output: the image now compiles from source in a builder +# stage, so none of the host's target/ is needed in the build context. +target # IDE settings .vscode diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..9b1936b --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,146 @@ +name: CI + +on: + push: + branches: [ "main" ] + tags: [ 'v*.*.*' ] + pull_request: + branches: [ "main" ] + +env: + REGISTRY_GHCR: ghcr.io + IMAGE_NAME: ${{ github.repository }} + +jobs: + test: + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v7 + + - name: Set up Rust + uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - name: Cache Cargo Registry + uses: Swatinem/rust-cache@v2 + + - name: Check formatting + run: cargo fmt --check + + - name: Clippy + run: cargo clippy --all-targets -- -D warnings + + - name: Test + run: cargo test --locked + + - name: Dependency advisories, licenses, and bans + uses: EmbarkStudios/cargo-deny-action@v2 + + build: + needs: test + runs-on: ubuntu-latest + env: + DOCKER_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }} + permissions: + contents: read + packages: write + id-token: write + + steps: + - name: Checkout repository + uses: actions/checkout@v7 + + # Login to Docker Hub + - name: Log into Docker Hub + if: github.event_name != 'pull_request' && env.DOCKER_USERNAME != '' + uses: docker/login-action@v4 + with: + username: ${{ secrets.DOCKERHUB_USERNAME }} + password: ${{ secrets.DOCKERHUB_TOKEN }} + + # Login to GitHub Container Registry + - name: Log into GHCR + if: github.event_name != 'pull_request' + uses: docker/login-action@v4 + with: + registry: ${{ env.REGISTRY_GHCR }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Extract Docker metadata + id: meta + uses: docker/metadata-action@v6 + with: + images: | + ${{ env.REGISTRY_GHCR }}/${{ env.IMAGE_NAME }} + ${{ secrets.DOCKERHUB_USERNAME != '' && env.IMAGE_NAME || '' }} + tags: | + # 1. Tag 'latest' ONLY when pushing to main + type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }} + + # 2. Tag with the version number (1.0.0) when pushing a tag + type=semver,pattern={{version}} + + # 3. (Optional) Also tag 'main' literally if you want image:main + type=ref,event=branch + + - name: Set up QEMU (for arm64 builds) + uses: docker/setup-qemu-action@v4 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v4 + + # Build a single-arch image locally first so Trivy has a concrete + # image to scan; the multi-arch manifest list below isn't scannable. + - name: Build image for vulnerability scan + uses: docker/build-push-action@v7 + with: + context: . + platforms: linux/amd64 + load: true + tags: auditbridge:scan + cache-from: type=gha + cache-to: type=gha,mode=max + + # ignore-unfixed: the debian:bookworm-slim base carries a handful of + # HIGH/CRITICAL OS-package CVEs with no upstream fix available + # (will_not_fix/fix_deferred); gating on those would leave CI + # permanently red for nothing this project can act on. Anything with + # an actual available fix still fails the build. + - name: Scan image for vulnerabilities + uses: aquasecurity/trivy-action@v0.36.0 + with: + image-ref: 'auditbridge:scan' + format: 'table' + exit-code: '1' + ignore-unfixed: true + severity: 'HIGH,CRITICAL' + + - name: Build and push multi-arch image + id: build-and-push + uses: docker/build-push-action@v7 + with: + context: . + platforms: linux/amd64,linux/arm64 + push: ${{ github.event_name != 'pull_request' }} + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha + cache-to: type=gha,mode=max + + release: + needs: build + if: startsWith(github.ref, 'refs/tags/v') + runs-on: ubuntu-latest + permissions: + contents: write + steps: + - name: Checkout repository + uses: actions/checkout@v7 + + - name: Create GitHub Release + uses: softprops/action-gh-release@v3 + with: + generate_release_notes: true diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml deleted file mode 100644 index d21a367..0000000 --- a/.github/workflows/docker-publish.yml +++ /dev/null @@ -1,85 +0,0 @@ -name: Docker - -on: - push: - branches: [ "main" ] - tags: [ 'v*.*.*' ] - pull_request: - branches: [ "main" ] - -env: - REGISTRY_GHCR: ghcr.io - IMAGE_NAME: ${{ github.repository }} - -jobs: - build: - runs-on: ubuntu-latest - env: - DOCKER_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }} - permissions: - contents: read - packages: write - id-token: write - - steps: - - name: Checkout repository - uses: actions/checkout@v4 - - # Login to Docker Hub - - name: Log into Docker Hub - if: github.event_name != 'pull_request' && env.DOCKER_USERNAME != '' - uses: docker/login-action@v3 - with: - username: ${{ secrets.DOCKERHUB_USERNAME }} - password: ${{ secrets.DOCKERHUB_TOKEN }} - - # Login to GitHub Container Registry - - name: Log into GHCR - if: github.event_name != 'pull_request' - uses: docker/login-action@v3 - with: - registry: ${{ env.REGISTRY_GHCR }} - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - - name: Extract Docker metadata - id: meta - uses: docker/metadata-action@v5 - with: - images: | - ${{ env.REGISTRY_GHCR }}/${{ env.IMAGE_NAME }} - ${{ secrets.DOCKERHUB_USERNAME != '' && env.IMAGE_NAME || '' }} - tags: | - # 1. Tag 'latest' ONLY when pushing to main - type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }} - - # 2. Tag with the version number (1.0.0) when pushing a tag - type=semver,pattern={{version}} - - # 3. (Optional) Also tag 'main' literally if you want image:main - type=ref,event=branch - - - name: Set up Rust - uses: dtolnay/rust-toolchain@stable - - - name: Cache Cargo Registry - uses: swatinem/rust-cache@v2 - - # Note: We use the strategy of building in the runner and copying the binary. - # This utilizes the rust-cache efficiently. - - name: Build Binary - run: cargo build --release --locked - - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v3 - - - name: Build and push Docker image - id: build-and-push - uses: docker/build-push-action@v5 - with: - context: . - push: ${{ github.event_name != 'pull_request' }} - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max diff --git a/Cargo.lock b/Cargo.lock index 496640e..99d2b38 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -43,9 +43,9 @@ dependencies = [ [[package]] name = "anyhow" -version = "1.0.100" +version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61" +checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" [[package]] name = "assert-json-diff" @@ -211,9 +211,9 @@ checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43" [[package]] name = "bytes" -version = "1.11.0" +version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" diff --git a/Dockerfile b/Dockerfile index d06a2f2..77b33ef 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,5 +1,28 @@ +# syntax=docker/dockerfile:1 + +# Builder stage: compiles the release binary inside the image, so `docker +# build .` works standalone with no separate CI pre-build step. +FROM rust:1-slim-bookworm@sha256:96c0af8cf054fd006435089f0076729716784ec9be485bd655de59c55df105ce AS builder + +# native-tls links against system OpenSSL at build time. +RUN apt-get update && \ + apt-get install -y --no-install-recommends pkg-config libssl-dev && \ + rm -rf /var/lib/apt/lists/* + +WORKDIR /build + +# Cache dependency compilation separately from source changes: this layer +# only invalidates when Cargo.toml/Cargo.lock change, not on every edit. +COPY Cargo.toml Cargo.lock ./ +RUN mkdir src && echo "fn main() {}" > src/main.rs && \ + cargo build --release --locked && \ + rm -rf src + +COPY src ./src +RUN touch src/main.rs && cargo build --release --locked + # Runtime stage (Debian Bookworm slim - provides glibc 2.36+ and openssl 3) -FROM debian:bookworm-slim +FROM debian:bookworm-slim@sha256:abd67ffcfa541b485a3dff59865ab629aa048a6c613e639d36e7456b0b229241 # Install only runtime dependencies RUN apt-get update && \ @@ -14,9 +37,7 @@ RUN groupadd -g 1000 exporter && \ WORKDIR /app -# Copy binary directly from CI workspace (target/release) -# Note: CI must run `cargo build --release` before this -COPY target/release/signal /app/exporter +COPY --from=builder /build/target/release/signal /app/exporter # Verify binary RUN chmod +x /app/exporter diff --git a/README.md b/README.md index 5159cec..082e2b2 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Signal -[![Build Status](https://img.shields.io/github/actions/workflow/status/onelrian/signal/docker.yml?branch=main)](https://github.com/onelrian/signal/actions) +[![Build Status](https://img.shields.io/github/actions/workflow/status/onelrian/signal/ci.yml?branch=main)](https://github.com/onelrian/signal/actions) [![Docker Pulls](https://img.shields.io/docker/pulls/onelrian/signal)](https://hub.docker.com/r/onelrian/signal) [![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE) diff --git a/deny.toml b/deny.toml new file mode 100644 index 0000000..4ac8e3e --- /dev/null +++ b/deny.toml @@ -0,0 +1,30 @@ +[graph] +all-features = false + +[advisories] +version = 2 +yanked = "deny" + +[licenses] +version = 2 +allow = [ + "MIT", + "Apache-2.0", + "Apache-2.0 WITH LLVM-exception", + "BSD-2-Clause", + "BSD-3-Clause", + "ISC", + "Unicode-3.0", + "Zlib", + "CDLA-Permissive-2.0", + "MPL-2.0", +] +confidence-threshold = 0.8 + +[bans] +multiple-versions = "warn" +wildcards = "deny" + +[sources] +unknown-registry = "deny" +unknown-git = "deny" From 98630c1b5bd92bded10785215be0bc72b73f9db4 Mon Sep 17 00:00:00 2001 From: onelrian Date: Fri, 7 Aug 2026 11:46:17 +0100 Subject: [PATCH 2/3] fix(tests): remove redundant mod tests nesting inside tests.rs Fixes the module_inception warning clippy -D warnings now gates on in CI (added in this same PR), harmless before but a hard build failure now that the lint is enforced instead of just noted. --- src/tests.rs | 1495 +++++++++++++++++++++++++------------------------- 1 file changed, 746 insertions(+), 749 deletions(-) diff --git a/src/tests.rs b/src/tests.rs index a4a377c..bdcc869 100644 --- a/src/tests.rs +++ b/src/tests.rs @@ -1,817 +1,814 @@ -#[cfg(test)] -mod tests { - use crate::config::Config; - use crate::cursor; - use crate::metrics::{router, Metrics}; - use crate::models::Event; - use crate::netbird::NetbirdClient; - use crate::retry::{with_retry, RetryConfig}; - use crate::sinks::encoding::Encoding; - use crate::sinks::http::HttpSink; - use crate::sinks::syslog::{SyslogProtocol, SyslogSink}; - use crate::sinks::Sink; - use crate::{build_initial_cursors, process_cycle, run}; - use chrono::{DateTime, Utc}; - use reqwest::Method; - use std::collections::HashMap; - use std::sync::atomic::{AtomicU32, Ordering}; - use std::time::Duration; - use wiremock::matchers::{method, path}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - - // No retries: keeps process_cycle tests focused on watermark behavior, - // not on how many times a deliberately-failing mock gets hit. - fn no_retry() -> RetryConfig { - RetryConfig { - max_attempts: 1, - base_delay: Duration::from_millis(1), - max_delay: Duration::from_millis(1), - } +use crate::config::Config; +use crate::cursor; +use crate::metrics::{router, Metrics}; +use crate::models::Event; +use crate::netbird::NetbirdClient; +use crate::retry::{with_retry, RetryConfig}; +use crate::sinks::encoding::Encoding; +use crate::sinks::http::HttpSink; +use crate::sinks::syslog::{SyslogProtocol, SyslogSink}; +use crate::sinks::Sink; +use crate::{build_initial_cursors, process_cycle, run}; +use chrono::{DateTime, Utc}; +use reqwest::Method; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::Duration; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +// No retries: keeps process_cycle tests focused on watermark behavior, +// not on how many times a deliberately-failing mock gets hit. +fn no_retry() -> RetryConfig { + RetryConfig { + max_attempts: 1, + base_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(1), } +} - fn test_config() -> Config { - Config { - netbird_api_url: "http://localhost".to_string(), - netbird_api_token: "token".to_string(), - check_interval: Duration::from_millis(10), - sinks: vec![], - cursor_file: None, - retry: no_retry(), - metrics_port: 0, - } +fn test_config() -> Config { + Config { + netbird_api_url: "http://localhost".to_string(), + netbird_api_token: "token".to_string(), + check_interval: Duration::from_millis(10), + sinks: vec![], + cursor_file: None, + retry: no_retry(), + metrics_port: 0, } +} - fn sample_event(id: &str) -> Event { - Event { - id: id.to_string(), - timestamp: "2023-01-01T00:00:00Z".to_string(), - activity: "test_activity".to_string(), - activity_code: "test.activity".to_string(), - initiator_id: Some("init1".to_string()), - initiator_email: Some("admin@example.com".to_string()), - initiator_name: Some("Admin".to_string()), - target_id: Some("target1".to_string()), - account_id: Some("acc1".to_string()), - meta: None, - } +fn sample_event(id: &str) -> Event { + Event { + id: id.to_string(), + timestamp: "2023-01-01T00:00:00Z".to_string(), + activity: "test_activity".to_string(), + activity_code: "test.activity".to_string(), + initiator_id: Some("init1".to_string()), + initiator_email: Some("admin@example.com".to_string()), + initiator_name: Some("Admin".to_string()), + target_id: Some("target1".to_string()), + account_id: Some("acc1".to_string()), + meta: None, } +} - #[tokio::test] - async fn test_netbird_fetch_events() { - let mock_server = MockServer::start().await; +#[tokio::test] +async fn test_netbird_fetch_events() { + let mock_server = MockServer::start().await; - Mock::given(method("GET")) - .and(path("/api/events/audit")) - .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) - .mount(&mock_server) - .await; + Mock::given(method("GET")) + .and(path("/api/events/audit")) + .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) + .mount(&mock_server) + .await; - let client = NetbirdClient::new(mock_server.uri(), "fake_token".to_string()); - let events = client.fetch_events().await.expect("Failed to fetch events"); + let client = NetbirdClient::new(mock_server.uri(), "fake_token".to_string()); + let events = client.fetch_events().await.expect("Failed to fetch events"); - assert_eq!(events.len(), 1); - assert_eq!(events[0].id, "1"); - assert_eq!(events[0].activity, "test_activity"); - } + assert_eq!(events.len(), 1); + assert_eq!(events[0].id, "1"); + assert_eq!(events[0].activity, "test_activity"); +} - #[tokio::test] - async fn test_http_sink_loki_encoding() { - let mock_server = MockServer::start().await; +#[tokio::test] +async fn test_http_sink_loki_encoding() { + let mock_server = MockServer::start().await; - Mock::given(method("POST")) - .and(path("/loki/api/v1/push")) - .respond_with(ResponseTemplate::new(204)) - .mount(&mock_server) - .await; + Mock::given(method("POST")) + .and(path("/loki/api/v1/push")) + .respond_with(ResponseTemplate::new(204)) + .mount(&mock_server) + .await; - let sink = HttpSink::new( - "loki".to_string(), - format!("{}/loki/api/v1/push", mock_server.uri()), - Method::POST, - vec![], - Encoding::Loki, - ); - sink.send(&[sample_event("1")]) - .await - .expect("Failed to send events"); - } + let sink = HttpSink::new( + "loki".to_string(), + format!("{}/loki/api/v1/push", mock_server.uri()), + Method::POST, + vec![], + Encoding::Loki, + ); + sink.send(&[sample_event("1")]) + .await + .expect("Failed to send events"); +} - #[tokio::test] - async fn test_http_sink_json_encoding_with_custom_headers() { - let mock_server = MockServer::start().await; +#[tokio::test] +async fn test_http_sink_json_encoding_with_custom_headers() { + let mock_server = MockServer::start().await; - Mock::given(method("POST")) - .and(path("/ingest")) - .respond_with(ResponseTemplate::new(200)) - .mount(&mock_server) - .await; + Mock::given(method("POST")) + .and(path("/ingest")) + .respond_with(ResponseTemplate::new(200)) + .mount(&mock_server) + .await; - let sink = HttpSink::new( - "generic-webhook".to_string(), - format!("{}/ingest", mock_server.uri()), - Method::POST, - vec![("X-Api-Key".to_string(), "secret".to_string())], - Encoding::Json, - ); - sink.send(&[sample_event("1")]) - .await - .expect("Failed to send events"); - } + let sink = HttpSink::new( + "generic-webhook".to_string(), + format!("{}/ingest", mock_server.uri()), + Method::POST, + vec![("X-Api-Key".to_string(), "secret".to_string())], + Encoding::Json, + ); + sink.send(&[sample_event("1")]) + .await + .expect("Failed to send events"); +} - #[tokio::test] - async fn test_syslog_sink_rfc3164_over_tcp() { - use tokio::io::AsyncReadExt; - use tokio::net::TcpListener; +#[tokio::test] +async fn test_syslog_sink_rfc3164_over_tcp() { + use tokio::io::AsyncReadExt; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut buf = Vec::new(); + socket.read_to_end(&mut buf).await.ok(); + buf + }); + + let sink = SyslogSink::new( + "wazuh".to_string(), + addr.to_string(), + SyslogProtocol::Tcp, + Encoding::Syslog3164, + ); + sink.send(&[sample_event("1")]) + .await + .expect("Failed to send events"); + drop(sink); + + let received = tokio::time::timeout(std::time::Duration::from_secs(2), server) + .await + .expect("syslog mock server timed out") + .expect("syslog mock server task panicked"); + let text = String::from_utf8(received).unwrap(); + + assert!( + text.starts_with("<134>"), + "expected syslog PRI framing, got: {}", + text + ); + assert!(text.contains("test.activity")); +} - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); +#[tokio::test] +async fn test_process_cycle_does_not_advance_watermark_on_send_failure() { + let nb_mock = MockServer::start().await; + let sink_mock = MockServer::start().await; - let server = tokio::spawn(async move { - let (mut socket, _) = listener.accept().await.unwrap(); - let mut buf = Vec::new(); - socket.read_to_end(&mut buf).await.ok(); - buf - }); + Mock::given(method("GET")) + .and(path("/api/events/audit")) + .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) + .mount(&nb_mock) + .await; - let sink = SyslogSink::new( - "wazuh".to_string(), - addr.to_string(), - SyslogProtocol::Tcp, - Encoding::Syslog3164, - ); - sink.send(&[sample_event("1")]) - .await - .expect("Failed to send events"); - drop(sink); - - let received = tokio::time::timeout(std::time::Duration::from_secs(2), server) - .await - .expect("syslog mock server timed out") - .expect("syslog mock server task panicked"); - let text = String::from_utf8(received).unwrap(); - - assert!( - text.starts_with("<134>"), - "expected syslog PRI framing, got: {}", - text - ); - assert!(text.contains("test.activity")); - } + Mock::given(method("POST")) + .and(path("/ingest")) + .respond_with(ResponseTemplate::new(500)) + .mount(&sink_mock) + .await; - #[tokio::test] - async fn test_process_cycle_does_not_advance_watermark_on_send_failure() { - let nb_mock = MockServer::start().await; - let sink_mock = MockServer::start().await; - - Mock::given(method("GET")) - .and(path("/api/events/audit")) - .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) - .mount(&nb_mock) - .await; - - Mock::given(method("POST")) - .and(path("/ingest")) - .respond_with(ResponseTemplate::new(500)) - .mount(&sink_mock) - .await; - - let nb_client = NetbirdClient::new(nb_mock.uri(), "fake_token".to_string()); - let sinks: Vec> = vec![Box::new(HttpSink::new( - "flaky".to_string(), - format!("{}/ingest", sink_mock.uri()), - Method::POST, - vec![], - Encoding::Json, - ))]; - let mut cursors: HashMap>> = HashMap::new(); - - process_cycle( - &nb_client, - &sinks, - &mut cursors, - &no_retry(), - &Metrics::default(), - ) + let nb_client = NetbirdClient::new(nb_mock.uri(), "fake_token".to_string()); + let sinks: Vec> = vec![Box::new(HttpSink::new( + "flaky".to_string(), + format!("{}/ingest", sink_mock.uri()), + Method::POST, + vec![], + Encoding::Json, + ))]; + let mut cursors: HashMap>> = HashMap::new(); + + process_cycle( + &nb_client, + &sinks, + &mut cursors, + &no_retry(), + &Metrics::default(), + ) + .await; + + assert_eq!( + cursors.get("flaky").copied().flatten(), + None, + "watermark must not advance when the sink write fails" + ); +} + +#[tokio::test] +async fn test_process_cycle_advances_watermark_on_send_success() { + let nb_mock = MockServer::start().await; + let sink_mock = MockServer::start().await; + + Mock::given(method("GET")) + .and(path("/api/events/audit")) + .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) + .mount(&nb_mock) .await; - assert_eq!( - cursors.get("flaky").copied().flatten(), - None, - "watermark must not advance when the sink write fails" - ); - } + Mock::given(method("POST")) + .and(path("/ingest")) + .respond_with(ResponseTemplate::new(200)) + .mount(&sink_mock) + .await; - #[tokio::test] - async fn test_process_cycle_advances_watermark_on_send_success() { - let nb_mock = MockServer::start().await; - let sink_mock = MockServer::start().await; - - Mock::given(method("GET")) - .and(path("/api/events/audit")) - .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) - .mount(&nb_mock) - .await; - - Mock::given(method("POST")) - .and(path("/ingest")) - .respond_with(ResponseTemplate::new(200)) - .mount(&sink_mock) - .await; - - let nb_client = NetbirdClient::new(nb_mock.uri(), "fake_token".to_string()); - let sinks: Vec> = vec![Box::new(HttpSink::new( - "generic".to_string(), - format!("{}/ingest", sink_mock.uri()), - Method::POST, - vec![], - Encoding::Json, - ))]; - let mut cursors: HashMap>> = HashMap::new(); - - process_cycle( - &nb_client, - &sinks, - &mut cursors, - &no_retry(), - &Metrics::default(), - ) + let nb_client = NetbirdClient::new(nb_mock.uri(), "fake_token".to_string()); + let sinks: Vec> = vec![Box::new(HttpSink::new( + "generic".to_string(), + format!("{}/ingest", sink_mock.uri()), + Method::POST, + vec![], + Encoding::Json, + ))]; + let mut cursors: HashMap>> = HashMap::new(); + + process_cycle( + &nb_client, + &sinks, + &mut cursors, + &no_retry(), + &Metrics::default(), + ) + .await; + + assert!( + cursors.get("generic").copied().flatten().is_some(), + "watermark must advance once delivery is confirmed" + ); +} + +#[tokio::test] +async fn test_process_cycle_one_failing_sink_does_not_block_the_other() { + let nb_mock = MockServer::start().await; + let down_mock = MockServer::start().await; + let up_mock = MockServer::start().await; + + Mock::given(method("GET")) + .and(path("/api/events/audit")) + .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) + .mount(&nb_mock) .await; - assert!( - cursors.get("generic").copied().flatten().is_some(), - "watermark must advance once delivery is confirmed" - ); - } + Mock::given(method("POST")) + .and(path("/ingest")) + .respond_with(ResponseTemplate::new(500)) + .mount(&down_mock) + .await; - #[tokio::test] - async fn test_process_cycle_one_failing_sink_does_not_block_the_other() { - let nb_mock = MockServer::start().await; - let down_mock = MockServer::start().await; - let up_mock = MockServer::start().await; - - Mock::given(method("GET")) - .and(path("/api/events/audit")) - .respond_with(ResponseTemplate::new(200).set_body_json(vec![sample_event("1")])) - .mount(&nb_mock) - .await; - - Mock::given(method("POST")) - .and(path("/ingest")) - .respond_with(ResponseTemplate::new(500)) - .mount(&down_mock) - .await; - - Mock::given(method("POST")) - .and(path("/ingest")) - .respond_with(ResponseTemplate::new(200)) - .mount(&up_mock) - .await; - - let nb_client = NetbirdClient::new(nb_mock.uri(), "fake_token".to_string()); - let sinks: Vec> = vec![ - Box::new(HttpSink::new( - "down".to_string(), - format!("{}/ingest", down_mock.uri()), - Method::POST, - vec![], - Encoding::Json, - )), - Box::new(HttpSink::new( - "up".to_string(), - format!("{}/ingest", up_mock.uri()), - Method::POST, - vec![], - Encoding::Json, - )), - ]; - let mut cursors: HashMap>> = HashMap::new(); - - process_cycle( - &nb_client, - &sinks, - &mut cursors, - &no_retry(), - &Metrics::default(), - ) + Mock::given(method("POST")) + .and(path("/ingest")) + .respond_with(ResponseTemplate::new(200)) + .mount(&up_mock) .await; - assert_eq!(cursors.get("down").copied().flatten(), None); - assert!(cursors.get("up").copied().flatten().is_some()); - } + let nb_client = NetbirdClient::new(nb_mock.uri(), "fake_token".to_string()); + let sinks: Vec> = vec![ + Box::new(HttpSink::new( + "down".to_string(), + format!("{}/ingest", down_mock.uri()), + Method::POST, + vec![], + Encoding::Json, + )), + Box::new(HttpSink::new( + "up".to_string(), + format!("{}/ingest", up_mock.uri()), + Method::POST, + vec![], + Encoding::Json, + )), + ]; + let mut cursors: HashMap>> = HashMap::new(); + + process_cycle( + &nb_client, + &sinks, + &mut cursors, + &no_retry(), + &Metrics::default(), + ) + .await; + + assert_eq!(cursors.get("down").copied().flatten(), None); + assert!(cursors.get("up").copied().flatten().is_some()); +} - #[test] - fn test_config_defaults_to_loki_sink() { - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", Some("test_token")), - ("SINKS", None), - ("LOKI_URL", None), - ], - || { - let config = Config::from_env().unwrap(); - assert_eq!(config.netbird_api_token, "test_token"); - assert_eq!(config.sinks.len(), 1); - assert_eq!(config.sinks[0].name, "loki"); - }, - ); - } +#[test] +fn test_config_defaults_to_loki_sink() { + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", Some("test_token")), + ("SINKS", None), + ("LOKI_URL", None), + ], + || { + let config = Config::from_env().unwrap(); + assert_eq!(config.netbird_api_token, "test_token"); + assert_eq!(config.sinks.len(), 1); + assert_eq!(config.sinks[0].name, "loki"); + }, + ); +} - #[test] - fn test_config_wazuh_requires_addr() { - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", Some("test_token")), - ("SINKS", Some("wazuh")), - ("SINK_WAZUH_ADDR", None), - ("WAZUH_ADDR", None), - ], - || { - let result = Config::from_env(); - assert!( - result.is_err(), - "expected an error when no Wazuh address is set" - ); - }, - ); - } +#[test] +fn test_config_wazuh_requires_addr() { + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", Some("test_token")), + ("SINKS", Some("wazuh")), + ("SINK_WAZUH_ADDR", None), + ("WAZUH_ADDR", None), + ], + || { + let result = Config::from_env(); + assert!( + result.is_err(), + "expected an error when no Wazuh address is set" + ); + }, + ); +} - #[test] - fn test_config_generic_sink_requires_explicit_transport_and_encoding() { - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", Some("test_token")), - ("SINKS", Some("datadog")), - ( - "SINK_DATADOG_URL", - Some("https://http-intake.example/v1/logs"), - ), - ("SINK_DATADOG_ENCODING", Some("json")), - // no SINK_DATADOG_TRANSPORT on purpose: unknown sink names get no defaults. - ], - || { - let result = Config::from_env(); - assert!( - result.is_err(), - "a sink name with no built-in preset must require an explicit transport" - ); - }, - ); - } +#[test] +fn test_config_generic_sink_requires_explicit_transport_and_encoding() { + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", Some("test_token")), + ("SINKS", Some("datadog")), + ( + "SINK_DATADOG_URL", + Some("https://http-intake.example/v1/logs"), + ), + ("SINK_DATADOG_ENCODING", Some("json")), + // no SINK_DATADOG_TRANSPORT on purpose: unknown sink names get no defaults. + ], + || { + let result = Config::from_env(); + assert!( + result.is_err(), + "a sink name with no built-in preset must require an explicit transport" + ); + }, + ); +} - #[test] - fn test_config_generic_http_sink_via_env_only() { - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", Some("test_token")), - ("SINKS", Some("loki,datadog")), - ("SINK_DATADOG_TRANSPORT", Some("http")), - ( - "SINK_DATADOG_URL", - Some("https://http-intake.example/v1/logs"), - ), - ("SINK_DATADOG_ENCODING", Some("json")), - ("SINK_DATADOG_HEADERS", Some("DD-API-KEY:secret,X-Extra:1")), - ], - || { - let config = Config::from_env().unwrap(); - assert_eq!(config.sinks.len(), 2); - let datadog = config.sinks.iter().find(|s| s.name == "datadog").unwrap(); - assert_eq!(datadog.headers.len(), 2); - }, - ); - } +#[test] +fn test_config_generic_http_sink_via_env_only() { + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", Some("test_token")), + ("SINKS", Some("loki,datadog")), + ("SINK_DATADOG_TRANSPORT", Some("http")), + ( + "SINK_DATADOG_URL", + Some("https://http-intake.example/v1/logs"), + ), + ("SINK_DATADOG_ENCODING", Some("json")), + ("SINK_DATADOG_HEADERS", Some("DD-API-KEY:secret,X-Extra:1")), + ], + || { + let config = Config::from_env().unwrap(); + assert_eq!(config.sinks.len(), 2); + let datadog = config.sinks.iter().find(|s| s.name == "datadog").unwrap(); + assert_eq!(datadog.headers.len(), 2); + }, + ); +} - #[test] - fn test_config_wazuh_syslog_encoding_defaults_to_rfc3164() { - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", Some("test_token")), - ("SINKS", Some("wazuh")), - ("SINK_WAZUH_ADDR", Some("127.0.0.1:1514")), - ], - || { - let config = Config::from_env().unwrap(); - assert_eq!(config.sinks[0].encoding, Encoding::Syslog3164); - }, - ); - } +#[test] +fn test_config_wazuh_syslog_encoding_defaults_to_rfc3164() { + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", Some("test_token")), + ("SINKS", Some("wazuh")), + ("SINK_WAZUH_ADDR", Some("127.0.0.1:1514")), + ], + || { + let config = Config::from_env().unwrap(); + assert_eq!(config.sinks[0].encoding, Encoding::Syslog3164); + }, + ); +} - fn temp_secret_file(name: &str, contents: &str) -> String { - let path = std::env::temp_dir() - .join(format!( - "auditbridge-test-secret-{}-{}", - name, - std::process::id() - )) - .to_string_lossy() - .to_string(); - std::fs::write(&path, contents).unwrap(); - path - } +fn temp_secret_file(name: &str, contents: &str) -> String { + let path = std::env::temp_dir() + .join(format!( + "auditbridge-test-secret-{}-{}", + name, + std::process::id() + )) + .to_string_lossy() + .to_string(); + std::fs::write(&path, contents).unwrap(); + path +} - #[test] - fn test_config_netbird_token_file_reads_and_trims_file_contents() { - let path = temp_secret_file("token", "nbp_from_file\n"); - - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", None), - ("NETBIRD_API_TOKEN_FILE", Some(path.as_str())), - ], - || { - let config = Config::from_env().unwrap(); - assert_eq!(config.netbird_api_token, "nbp_from_file"); - }, - ); - - std::fs::remove_file(&path).ok(); - } +#[test] +fn test_config_netbird_token_file_reads_and_trims_file_contents() { + let path = temp_secret_file("token", "nbp_from_file\n"); + + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", None), + ("NETBIRD_API_TOKEN_FILE", Some(path.as_str())), + ], + || { + let config = Config::from_env().unwrap(); + assert_eq!(config.netbird_api_token, "nbp_from_file"); + }, + ); + + std::fs::remove_file(&path).ok(); +} - #[test] - fn test_config_netbird_token_both_direct_and_file_is_an_error() { - let path = temp_secret_file("token-conflict", "nbp_from_file"); - - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", Some("nbp_direct")), - ("NETBIRD_API_TOKEN_FILE", Some(path.as_str())), - ], - || { - let result = Config::from_env(); - assert!( - result.is_err(), - "setting both the direct var and _FILE is ambiguous" - ); - }, - ); - - std::fs::remove_file(&path).ok(); - } +#[test] +fn test_config_netbird_token_both_direct_and_file_is_an_error() { + let path = temp_secret_file("token-conflict", "nbp_from_file"); + + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", Some("nbp_direct")), + ("NETBIRD_API_TOKEN_FILE", Some(path.as_str())), + ], + || { + let result = Config::from_env(); + assert!( + result.is_err(), + "setting both the direct var and _FILE is ambiguous" + ); + }, + ); + + std::fs::remove_file(&path).ok(); +} - #[test] - fn test_config_sink_headers_file_reads_credentials_from_file() { - let path = temp_secret_file("headers", "Authorization:Bearer secret-token\n"); - - temp_env::with_vars( - [ - ("NETBIRD_API_TOKEN", Some("test_token")), - ("SINKS", Some("datadog")), - ( - "SINK_DATADOG_URL", - Some("https://http-intake.example/v1/logs"), - ), - ("SINK_DATADOG_ENCODING", Some("json")), - ("SINK_DATADOG_TRANSPORT", Some("http")), - ("SINK_DATADOG_HEADERS_FILE", Some(path.as_str())), - ], - || { - let config = Config::from_env().unwrap(); - assert_eq!( - config.sinks[0].headers, - vec![( - "Authorization".to_string(), - "Bearer secret-token".to_string() - )] - ); - }, - ); - - std::fs::remove_file(&path).ok(); - } +#[test] +fn test_config_sink_headers_file_reads_credentials_from_file() { + let path = temp_secret_file("headers", "Authorization:Bearer secret-token\n"); + + temp_env::with_vars( + [ + ("NETBIRD_API_TOKEN", Some("test_token")), + ("SINKS", Some("datadog")), + ( + "SINK_DATADOG_URL", + Some("https://http-intake.example/v1/logs"), + ), + ("SINK_DATADOG_ENCODING", Some("json")), + ("SINK_DATADOG_TRANSPORT", Some("http")), + ("SINK_DATADOG_HEADERS_FILE", Some(path.as_str())), + ], + || { + let config = Config::from_env().unwrap(); + assert_eq!( + config.sinks[0].headers, + vec![( + "Authorization".to_string(), + "Bearer secret-token".to_string() + )] + ); + }, + ); + + std::fs::remove_file(&path).ok(); +} - fn temp_cursor_path(name: &str) -> String { - std::env::temp_dir() - .join(format!( - "auditbridge-test-{}-{}.json", - name, - std::process::id() - )) - .to_string_lossy() - .to_string() - } +fn temp_cursor_path(name: &str) -> String { + std::env::temp_dir() + .join(format!( + "auditbridge-test-{}-{}.json", + name, + std::process::id() + )) + .to_string_lossy() + .to_string() +} - #[test] - fn test_cursor_load_missing_file_returns_empty() { - let path = temp_cursor_path("missing"); - let loaded = cursor::load(&path); - assert!(loaded.is_empty()); - } +#[test] +fn test_cursor_load_missing_file_returns_empty() { + let path = temp_cursor_path("missing"); + let loaded = cursor::load(&path); + assert!(loaded.is_empty()); +} - #[test] - fn test_cursor_load_corrupt_file_returns_empty() { - let path = temp_cursor_path("corrupt"); - std::fs::write(&path, "not valid json").unwrap(); +#[test] +fn test_cursor_load_corrupt_file_returns_empty() { + let path = temp_cursor_path("corrupt"); + std::fs::write(&path, "not valid json").unwrap(); - let loaded = cursor::load(&path); + let loaded = cursor::load(&path); - assert!( - loaded.is_empty(), - "a corrupt cursor file must not crash the load" - ); - std::fs::remove_file(&path).ok(); - } + assert!( + loaded.is_empty(), + "a corrupt cursor file must not crash the load" + ); + std::fs::remove_file(&path).ok(); +} - #[test] - fn test_cursor_save_and_load_round_trip() { - let path = temp_cursor_path("roundtrip"); - let mut cursors = HashMap::new(); - cursors.insert( +#[test] +fn test_cursor_save_and_load_round_trip() { + let path = temp_cursor_path("roundtrip"); + let mut cursors = HashMap::new(); + cursors.insert( + "loki".to_string(), + "2023-01-01T00:00:00Z".parse::>().unwrap(), + ); + cursors.insert( + "wazuh".to_string(), + "2023-06-15T12:30:00Z".parse::>().unwrap(), + ); + + cursor::save(&path, &cursors).expect("save should succeed"); + let loaded = cursor::load(&path); + + assert_eq!(loaded, cursors); + std::fs::remove_file(&path).ok(); +} + +#[test] +fn test_build_initial_cursors_resumes_from_persisted_state() { + let sinks: Vec> = vec![ + Box::new(HttpSink::new( "loki".to_string(), - "2023-01-01T00:00:00Z".parse::>().unwrap(), - ); - cursors.insert( + "http://loki/loki/api/v1/push".to_string(), + Method::POST, + vec![], + Encoding::Loki, + )), + Box::new(SyslogSink::new( "wazuh".to_string(), - "2023-06-15T12:30:00Z".parse::>().unwrap(), - ); + "127.0.0.1:1514".to_string(), + SyslogProtocol::Tcp, + Encoding::Syslog3164, + )), + ]; - cursor::save(&path, &cursors).expect("save should succeed"); - let loaded = cursor::load(&path); + let mut persisted = HashMap::new(); + let loki_ts: DateTime = "2023-01-01T00:00:00Z".parse().unwrap(); + persisted.insert("loki".to_string(), loki_ts); + // no entry for "wazuh": never persisted (e.g. first run for that sink) - assert_eq!(loaded, cursors); - std::fs::remove_file(&path).ok(); - } + let cursors = build_initial_cursors(&sinks, &persisted); - #[test] - fn test_build_initial_cursors_resumes_from_persisted_state() { - let sinks: Vec> = vec![ - Box::new(HttpSink::new( - "loki".to_string(), - "http://loki/loki/api/v1/push".to_string(), - Method::POST, - vec![], - Encoding::Loki, - )), - Box::new(SyslogSink::new( - "wazuh".to_string(), - "127.0.0.1:1514".to_string(), - SyslogProtocol::Tcp, - Encoding::Syslog3164, - )), - ]; - - let mut persisted = HashMap::new(); - let loki_ts: DateTime = "2023-01-01T00:00:00Z".parse().unwrap(); - persisted.insert("loki".to_string(), loki_ts); - // no entry for "wazuh": never persisted (e.g. first run for that sink) - - let cursors = build_initial_cursors(&sinks, &persisted); - - assert_eq!(cursors.get("loki").copied().flatten(), Some(loki_ts)); - assert_eq!(cursors.get("wazuh").copied().flatten(), None); - } + assert_eq!(cursors.get("loki").copied().flatten(), Some(loki_ts)); + assert_eq!(cursors.get("wazuh").copied().flatten(), None); +} - fn fast_retry(max_attempts: u32) -> RetryConfig { - RetryConfig { - max_attempts, - base_delay: Duration::from_millis(1), - max_delay: Duration::from_millis(5), - } +fn fast_retry(max_attempts: u32) -> RetryConfig { + RetryConfig { + max_attempts, + base_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(5), } +} - #[tokio::test] - async fn test_with_retry_succeeds_after_transient_failures() { - let attempts = AtomicU32::new(0); - let cfg = fast_retry(5); - - let result = with_retry("test op", &cfg, || { - let n = attempts.fetch_add(1, Ordering::SeqCst); - async move { - if n < 2 { - Err(anyhow::anyhow!("transient failure")) - } else { - Ok(42) - } +#[tokio::test] +async fn test_with_retry_succeeds_after_transient_failures() { + let attempts = AtomicU32::new(0); + let cfg = fast_retry(5); + + let result = with_retry("test op", &cfg, || { + let n = attempts.fetch_add(1, Ordering::SeqCst); + async move { + if n < 2 { + Err(anyhow::anyhow!("transient failure")) + } else { + Ok(42) } - }) - .await; + } + }) + .await; - assert_eq!(result.unwrap(), 42); - assert_eq!(attempts.load(Ordering::SeqCst), 3); - } + assert_eq!(result.unwrap(), 42); + assert_eq!(attempts.load(Ordering::SeqCst), 3); +} - #[tokio::test] - async fn test_with_retry_gives_up_after_max_attempts() { - let attempts = AtomicU32::new(0); - let cfg = fast_retry(3); +#[tokio::test] +async fn test_with_retry_gives_up_after_max_attempts() { + let attempts = AtomicU32::new(0); + let cfg = fast_retry(3); + + let result: anyhow::Result<()> = with_retry("test op", &cfg, || { + attempts.fetch_add(1, Ordering::SeqCst); + async { Err(anyhow::anyhow!("permanent failure")) } + }) + .await; + + assert!(result.is_err()); + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "must stop at max_attempts, not retry forever" + ); +} - let result: anyhow::Result<()> = with_retry("test op", &cfg, || { - attempts.fetch_add(1, Ordering::SeqCst); - async { Err(anyhow::anyhow!("permanent failure")) } - }) - .await; +#[test] +fn test_metrics_not_ready_until_fetch_and_a_sink_both_succeed() { + let metrics = Metrics::default(); + assert!( + !metrics.is_ready(), + "must not be ready before the first cycle" + ); + + metrics.record_fetch_success(3); + assert!( + !metrics.is_ready(), + "fetch alone isn't enough, no sink has delivered yet" + ); + + metrics.record_sink_success("loki", 3); + assert!(metrics.is_ready()); +} - assert!(result.is_err()); - assert_eq!( - attempts.load(Ordering::SeqCst), - 3, - "must stop at max_attempts, not retry forever" - ); - } +#[test] +fn test_metrics_becomes_unready_again_after_fetch_failure() { + let metrics = Metrics::default(); + metrics.record_fetch_success(1); + metrics.record_sink_success("loki", 1); + assert!(metrics.is_ready()); + + metrics.record_fetch_error(); + assert!( + !metrics.is_ready(), + "a failed fetch must flip readiness back off" + ); +} - #[test] - fn test_metrics_not_ready_until_fetch_and_a_sink_both_succeed() { - let metrics = Metrics::default(); - assert!( - !metrics.is_ready(), - "must not be ready before the first cycle" - ); - - metrics.record_fetch_success(3); - assert!( - !metrics.is_ready(), - "fetch alone isn't enough, no sink has delivered yet" - ); - - metrics.record_sink_success("loki", 3); - assert!(metrics.is_ready()); - } +#[test] +fn test_metrics_render_prometheus_includes_recorded_values() { + let metrics = Metrics::default(); + metrics.record_fetch_success(5); + metrics.record_sink_success("loki", 5); + metrics.record_sink_error("wazuh"); - #[test] - fn test_metrics_becomes_unready_again_after_fetch_failure() { - let metrics = Metrics::default(); - metrics.record_fetch_success(1); - metrics.record_sink_success("loki", 1); - assert!(metrics.is_ready()); - - metrics.record_fetch_error(); - assert!( - !metrics.is_ready(), - "a failed fetch must flip readiness back off" - ); - } + let output = metrics.render_prometheus(); + + assert!(output.contains("auditbridge_events_fetched_total 5")); + assert!(output.contains("auditbridge_events_delivered_total{sink=\"loki\"} 5")); + assert!(output.contains("auditbridge_delivery_errors_total{sink=\"wazuh\"} 1")); +} - #[test] - fn test_metrics_render_prometheus_includes_recorded_values() { - let metrics = Metrics::default(); - metrics.record_fetch_success(5); - metrics.record_sink_success("loki", 5); - metrics.record_sink_error("wazuh"); +#[tokio::test] +async fn test_metrics_server_exposes_healthz_readyz_metrics() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let metrics = Metrics::new(); + let server_metrics = metrics.clone(); + + tokio::spawn(async move { + axum::serve(listener, router(server_metrics)).await.unwrap(); + }); + + let client = reqwest::Client::new(); + + let healthz = client + .get(format!("http://{}/healthz", addr)) + .send() + .await + .unwrap(); + assert_eq!(healthz.status(), reqwest::StatusCode::OK); + + let readyz_before = client + .get(format!("http://{}/readyz", addr)) + .send() + .await + .unwrap(); + assert_eq!( + readyz_before.status(), + reqwest::StatusCode::SERVICE_UNAVAILABLE + ); + + metrics.record_fetch_success(2); + metrics.record_sink_success("loki", 2); + + let readyz_after = client + .get(format!("http://{}/readyz", addr)) + .send() + .await + .unwrap(); + assert_eq!(readyz_after.status(), reqwest::StatusCode::OK); + + let metrics_resp = client + .get(format!("http://{}/metrics", addr)) + .send() + .await + .unwrap(); + assert_eq!(metrics_resp.status(), reqwest::StatusCode::OK); + let body = metrics_resp.text().await.unwrap(); + assert!(body.contains("auditbridge_events_fetched_total 2")); +} - let output = metrics.render_prometheus(); +#[tokio::test] +async fn test_run_exits_immediately_if_shutdown_already_signaled() { + let nb_mock = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/events/audit")) + .respond_with(ResponseTemplate::new(200).set_body_json(Vec::::new())) + .expect(0) + .mount(&nb_mock) + .await; - assert!(output.contains("auditbridge_events_fetched_total 5")); - assert!(output.contains("auditbridge_events_delivered_total{sink=\"loki\"} 5")); - assert!(output.contains("auditbridge_delivery_errors_total{sink=\"wazuh\"} 1")); - } + let nb_client = NetbirdClient::new(nb_mock.uri(), "token".to_string()); + let sinks: Vec> = vec![]; + let mut cursors = HashMap::new(); + let config = test_config(); + let metrics = Metrics::default(); + let (_tx, rx) = tokio::sync::watch::channel(true); - #[tokio::test] - async fn test_metrics_server_exposes_healthz_readyz_metrics() { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - let metrics = Metrics::new(); - let server_metrics = metrics.clone(); - - tokio::spawn(async move { - axum::serve(listener, router(server_metrics)).await.unwrap(); - }); - - let client = reqwest::Client::new(); - - let healthz = client - .get(format!("http://{}/healthz", addr)) - .send() - .await - .unwrap(); - assert_eq!(healthz.status(), reqwest::StatusCode::OK); - - let readyz_before = client - .get(format!("http://{}/readyz", addr)) - .send() - .await - .unwrap(); - assert_eq!( - readyz_before.status(), - reqwest::StatusCode::SERVICE_UNAVAILABLE - ); - - metrics.record_fetch_success(2); - metrics.record_sink_success("loki", 2); - - let readyz_after = client - .get(format!("http://{}/readyz", addr)) - .send() - .await - .unwrap(); - assert_eq!(readyz_after.status(), reqwest::StatusCode::OK); - - let metrics_resp = client - .get(format!("http://{}/metrics", addr)) - .send() - .await - .unwrap(); - assert_eq!(metrics_resp.status(), reqwest::StatusCode::OK); - let body = metrics_resp.text().await.unwrap(); - assert!(body.contains("auditbridge_events_fetched_total 2")); - } + let start = std::time::Instant::now(); + run(&nb_client, &sinks, &mut cursors, &config, &metrics, rx).await; + + assert!(start.elapsed() < Duration::from_millis(500)); + nb_mock.verify().await; +} - #[tokio::test] - async fn test_run_exits_immediately_if_shutdown_already_signaled() { - let nb_mock = MockServer::start().await; - Mock::given(method("GET")) - .and(path("/api/events/audit")) - .respond_with(ResponseTemplate::new(200).set_body_json(Vec::::new())) - .expect(0) - .mount(&nb_mock) - .await; - - let nb_client = NetbirdClient::new(nb_mock.uri(), "token".to_string()); - let sinks: Vec> = vec![]; - let mut cursors = HashMap::new(); - let config = test_config(); - let metrics = Metrics::default(); - let (_tx, rx) = tokio::sync::watch::channel(true); - - let start = std::time::Instant::now(); +#[tokio::test] +async fn test_run_exits_promptly_when_shutdown_fires_during_idle_wait() { + let nb_mock = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/events/audit")) + .respond_with(ResponseTemplate::new(200).set_body_json(Vec::::new())) + .mount(&nb_mock) + .await; + + let nb_client = NetbirdClient::new(nb_mock.uri(), "token".to_string()); + let sinks: Vec> = vec![]; + let cursors = HashMap::new(); + let mut config = test_config(); + config.check_interval = Duration::from_secs(60); + let metrics = Metrics::default(); + let (tx, rx) = tokio::sync::watch::channel(false); + + let handle = tokio::spawn(async move { + let mut cursors = cursors; run(&nb_client, &sinks, &mut cursors, &config, &metrics, rx).await; + }); - assert!(start.elapsed() < Duration::from_millis(500)); - nb_mock.verify().await; - } + tokio::time::sleep(Duration::from_millis(50)).await; + tx.send(true).unwrap(); - #[tokio::test] - async fn test_run_exits_promptly_when_shutdown_fires_during_idle_wait() { - let nb_mock = MockServer::start().await; - Mock::given(method("GET")) - .and(path("/api/events/audit")) - .respond_with(ResponseTemplate::new(200).set_body_json(Vec::::new())) - .mount(&nb_mock) - .await; - - let nb_client = NetbirdClient::new(nb_mock.uri(), "token".to_string()); - let sinks: Vec> = vec![]; - let cursors = HashMap::new(); - let mut config = test_config(); - config.check_interval = Duration::from_secs(60); - let metrics = Metrics::default(); - let (tx, rx) = tokio::sync::watch::channel(false); - - let handle = tokio::spawn(async move { - let mut cursors = cursors; - run(&nb_client, &sinks, &mut cursors, &config, &metrics, rx).await; - }); - - tokio::time::sleep(Duration::from_millis(50)).await; - tx.send(true).unwrap(); - - let result = tokio::time::timeout(Duration::from_secs(2), handle).await; - assert!( - result.is_ok(), - "run() must exit promptly once shutdown fires during the idle wait, not wait out check_interval" - ); - } + let result = tokio::time::timeout(Duration::from_secs(2), handle).await; + assert!( + result.is_ok(), + "run() must exit promptly once shutdown fires during the idle wait, not wait out check_interval" + ); +} - #[tokio::test] - async fn test_run_lets_in_flight_cycle_finish_before_exiting() { - let nb_mock = MockServer::start().await; - Mock::given(method("GET")) - .and(path("/api/events/audit")) - .respond_with( - ResponseTemplate::new(200) - .set_body_json(vec![sample_event("1")]) - .set_delay(Duration::from_millis(150)), - ) - .mount(&nb_mock) - .await; - - let sink_mock = MockServer::start().await; - Mock::given(method("POST")) - .and(path("/ingest")) - .respond_with(ResponseTemplate::new(200)) - .mount(&sink_mock) - .await; - - let nb_client = NetbirdClient::new(nb_mock.uri(), "token".to_string()); - let sinks: Vec> = vec![Box::new(HttpSink::new( - "test".to_string(), - format!("{}/ingest", sink_mock.uri()), - Method::POST, - vec![], - Encoding::Json, - ))]; - let cursors = HashMap::new(); - let mut config = test_config(); - config.check_interval = Duration::from_secs(60); - let metrics = Metrics::default(); - let (tx, rx) = tokio::sync::watch::channel(false); - - let handle = tokio::spawn(async move { - let mut cursors = cursors; - run(&nb_client, &sinks, &mut cursors, &config, &metrics, rx).await; - cursors - }); - - // Fires while the 150ms-delayed fetch is still in flight. - tokio::time::sleep(Duration::from_millis(30)).await; - tx.send(true).unwrap(); - - let result = tokio::time::timeout(Duration::from_secs(2), handle) - .await - .expect("run() did not exit within the timeout") - .expect("run() task panicked"); - - assert!( - result.get("test").copied().flatten().is_some(), - "the in-flight cycle must finish and its result apply even though shutdown fired mid-fetch" - ); - } +#[tokio::test] +async fn test_run_lets_in_flight_cycle_finish_before_exiting() { + let nb_mock = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/api/events/audit")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(vec![sample_event("1")]) + .set_delay(Duration::from_millis(150)), + ) + .mount(&nb_mock) + .await; + + let sink_mock = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/ingest")) + .respond_with(ResponseTemplate::new(200)) + .mount(&sink_mock) + .await; + + let nb_client = NetbirdClient::new(nb_mock.uri(), "token".to_string()); + let sinks: Vec> = vec![Box::new(HttpSink::new( + "test".to_string(), + format!("{}/ingest", sink_mock.uri()), + Method::POST, + vec![], + Encoding::Json, + ))]; + let cursors = HashMap::new(); + let mut config = test_config(); + config.check_interval = Duration::from_secs(60); + let metrics = Metrics::default(); + let (tx, rx) = tokio::sync::watch::channel(false); + + let handle = tokio::spawn(async move { + let mut cursors = cursors; + run(&nb_client, &sinks, &mut cursors, &config, &metrics, rx).await; + cursors + }); + + // Fires while the 150ms-delayed fetch is still in flight. + tokio::time::sleep(Duration::from_millis(30)).await; + tx.send(true).unwrap(); + + let result = tokio::time::timeout(Duration::from_secs(2), handle) + .await + .expect("run() did not exit within the timeout") + .expect("run() task panicked"); + + assert!( + result.get("test").copied().flatten().is_some(), + "the in-flight cycle must finish and its result apply even though shutdown fired mid-fetch" + ); } From 9abd4791776326fc0a76cf69259b72285e448b2d Mon Sep 17 00:00:00 2001 From: onelrian Date: Fri, 7 Aug 2026 12:02:35 +0100 Subject: [PATCH 3/3] ci: only publish images on version tags, not every main push Building and pushing a multi-arch manifest on every merge (and every PR, unpublished) wasted CI time on QEMU arm64 compiles nobody used. Now PRs/main only build+scan single-arch; publishing (and latest) only happens on a real vX.Y.Z tag. --- .github/workflows/ci.yml | 80 ++++++++++++++++++++-------------------- 1 file changed, 39 insertions(+), 41 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9b1936b..0c0f6cf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,17 +52,47 @@ jobs: - name: Checkout repository uses: actions/checkout@v7 - # Login to Docker Hub + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v4 + + # Every push/PR builds and scans a single-arch image for fast + # feedback. Nothing here is pushed, so no registry login, no QEMU, + # no multi-arch build for a manifest list that gets thrown away. + - name: Build image for vulnerability scan + uses: docker/build-push-action@v7 + with: + context: . + platforms: linux/amd64 + load: true + tags: auditbridge:scan + cache-from: type=gha + cache-to: type=gha,mode=max + + # ignore-unfixed: debian:bookworm-slim carries HIGH/CRITICAL CVEs with + # no upstream fix (will_not_fix/fix_deferred). A fixable one still + # fails the build; an unfixable base-image gap would leave CI red + # forever for nothing this project can act on. + - name: Scan image for vulnerabilities + uses: aquasecurity/trivy-action@v0.36.0 + with: + image-ref: 'auditbridge:scan' + format: 'table' + exit-code: '1' + ignore-unfixed: true + severity: 'HIGH,CRITICAL' + + # Everything below only runs on a vX.Y.Z tag, the only event that + # actually publishes. main-branch pushes build and scan above but + # ship nothing, so `latest` only ever points at a real tagged release. - name: Log into Docker Hub - if: github.event_name != 'pull_request' && env.DOCKER_USERNAME != '' + if: startsWith(github.ref, 'refs/tags/v') && env.DOCKER_USERNAME != '' uses: docker/login-action@v4 with: username: ${{ secrets.DOCKERHUB_USERNAME }} password: ${{ secrets.DOCKERHUB_TOKEN }} - # Login to GitHub Container Registry - name: Log into GHCR - if: github.event_name != 'pull_request' + if: startsWith(github.ref, 'refs/tags/v') uses: docker/login-action@v4 with: registry: ${{ env.REGISTRY_GHCR }} @@ -70,6 +100,7 @@ jobs: password: ${{ secrets.GITHUB_TOKEN }} - name: Extract Docker metadata + if: startsWith(github.ref, 'refs/tags/v') id: meta uses: docker/metadata-action@v6 with: @@ -77,54 +108,21 @@ jobs: ${{ env.REGISTRY_GHCR }}/${{ env.IMAGE_NAME }} ${{ secrets.DOCKERHUB_USERNAME != '' && env.IMAGE_NAME || '' }} tags: | - # 1. Tag 'latest' ONLY when pushing to main - type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }} - - # 2. Tag with the version number (1.0.0) when pushing a tag + type=raw,value=latest type=semver,pattern={{version}} - # 3. (Optional) Also tag 'main' literally if you want image:main - type=ref,event=branch - - name: Set up QEMU (for arm64 builds) + if: startsWith(github.ref, 'refs/tags/v') uses: docker/setup-qemu-action@v4 - - name: Set up Docker Buildx - uses: docker/setup-buildx-action@v4 - - # Build a single-arch image locally first so Trivy has a concrete - # image to scan; the multi-arch manifest list below isn't scannable. - - name: Build image for vulnerability scan - uses: docker/build-push-action@v7 - with: - context: . - platforms: linux/amd64 - load: true - tags: auditbridge:scan - cache-from: type=gha - cache-to: type=gha,mode=max - - # ignore-unfixed: the debian:bookworm-slim base carries a handful of - # HIGH/CRITICAL OS-package CVEs with no upstream fix available - # (will_not_fix/fix_deferred); gating on those would leave CI - # permanently red for nothing this project can act on. Anything with - # an actual available fix still fails the build. - - name: Scan image for vulnerabilities - uses: aquasecurity/trivy-action@v0.36.0 - with: - image-ref: 'auditbridge:scan' - format: 'table' - exit-code: '1' - ignore-unfixed: true - severity: 'HIGH,CRITICAL' - - name: Build and push multi-arch image + if: startsWith(github.ref, 'refs/tags/v') id: build-and-push uses: docker/build-push-action@v7 with: context: . platforms: linux/amd64,linux/arm64 - push: ${{ github.event_name != 'pull_request' }} + push: true tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} cache-from: type=gha