server/lib/src/audit/siem.rs
//! **SIEM export** — the audit log shipped, line by line, to where an operator's
//! SIEM reads it: JSON lines, syslog (RFC 5424 over UDP, TCP or TLS), CEF, and
//! OpenTelemetry logs (OTLP/HTTP, JSON encoding).
//!
//! ## The rules
//!
//! * **Off by default.** No `audit.siem` block, or a sink without
//! `enabled: true`, exports nothing.
//! * **The export never blocks the audit write.** An exporter is a follower of
//! `audit.jsonl`: the writer appends and bumps a [`TailSignal`], and never
//! waits on anything an exporter does. A dead SIEM costs the writer nothing.
//! * **Retried from a cursor.** Each sink keeps `<audit dir>/siem/<name>.cursor`
//! (the byte offset and `seq` of the next line to send). The cursor moves only
//! after a send succeeded, so a failed send is retried, from the same line,
//! after `retry_ms` — and after a restart.
//! * **Local only on the airgap side.** `local_only: true` (forced in a build
//! without `online`, the zero-egress airgap tiers) refuses every sink that is
//! not a file or an IP literal on a loopback, private, link-local or unique
//! local address (`holger-audit-9`). A host name is refused there: what it
//! resolves to is not this config's to decide.
//!
//! ## The formats
//!
//! | format | targets | one record is |
//! |---|---|---|
//! | `JsonLines` | `File`, `Udp`, `Tcp`, `Tls` | the audit line exactly as the log holds it, chain fields included, so a SIEM can re-verify the chain |
//! | `Syslog` | `File`, `Udp`, `Tcp`, `Tls` | an RFC 5424 message: `<PRI>1 TIMESTAMP HOST holger PID MSGID [holger@32473 ...] BOM detail` |
//! | `Cef` | `File`, `Udp`, `Tcp`, `Tls` | `CEF:0|Holger|holger-server|VERSION|SIGNATURE|NAME|SEVERITY|EXTENSIONS`; on a network target, the MSG of an RFC 5424 message |
//! | `Otlp` | `Http` | a `LogRecord` in an OTLP/HTTP JSON `ExportLogsServiceRequest`, POSTed to `/v1/logs` |
//!
//! Framing: UDP is one message per datagram (RFC 5426); TCP and TLS use octet
//! counting, `LEN SP MSG` (RFC 6587 §3.4.1, RFC 5425 §4.3); JSON lines over TCP
//! or TLS are newline-delimited; a file gets one message per line.
//!
//! OTLP is written by hand over the same TCP/TLS transport (HTTP/1.1, one POST
//! per batch, `Connection: close`): the airgap tiers link no HTTP client, and an
//! OTLP SDK would add a tree of crates for one POST.
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::net::{IpAddr, TcpStream, ToSocketAddrs, UdpSocket};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use serde::{Deserialize, Serialize};
use super::chain::TailSignal;
/// The `audit.siem` block.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct SiemSettings {
/// The sinks. Each is off until it says `enabled: true`.
#[serde(default)]
pub sinks: Vec<SinkSettings>,
/// The airgap side: only files and local addresses ([module docs](self)).
#[serde(default)]
pub local_only: bool,
/// Records per send (one datagram or framed message each; one POST for OTLP).
#[serde(default = "default_batch")]
pub batch: usize,
/// Wait after a failed send before the retry, milliseconds.
#[serde(default = "default_retry_ms")]
pub retry_ms: u64,
}
fn default_batch() -> usize {
100
}
fn default_retry_ms() -> u64 {
2000
}
/// One sink.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SinkSettings {
/// A name, unique among the sinks: it names the cursor file.
pub name: String,
/// Off unless `true`.
#[serde(default)]
pub enabled: bool,
/// What one record looks like on the wire.
pub format: SinkFormat,
/// Where it goes.
pub target: SinkTarget,
/// Syslog/CEF: the facility (default 13, "log audit").
#[serde(default = "default_facility")]
pub facility: u8,
/// Syslog: the SD-ID of the structured data element.
#[serde(default = "default_sd_id")]
pub sd_id: String,
/// Syslog/CEF: the HOSTNAME field; default this host's name.
#[serde(default)]
pub hostname: Option<String>,
/// OTLP: extra request headers (an API key a collector wants).
#[serde(default)]
pub headers: Vec<(String, String)>,
}
fn default_facility() -> u8 {
13
}
fn default_sd_id() -> String {
"holger@32473".into()
}
/// The wire format of a sink.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum SinkFormat {
/// The audit line itself.
JsonLines,
/// RFC 5424.
Syslog,
/// ArcSight Common Event Format.
Cef,
/// OpenTelemetry logs, OTLP/HTTP JSON.
Otlp,
}
/// Where a sink sends.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum SinkTarget {
/// Append to a local file.
File(String),
/// `host:port`, one message per datagram.
Udp(String),
/// `host:port`, framed.
Tcp(String),
/// `host:port` over TLS; `ca` is the PEM file the server certificate must
/// chain to, `server_name` the name it must carry (default: the host).
Tls {
addr: String,
ca: String,
#[serde(default)]
server_name: Option<String>,
},
/// An OTLP/HTTP endpoint, `http://host:port[/path]` or `https://…` (then
/// `ca` is required). An empty path means `/v1/logs`.
Http {
url: String,
#[serde(default)]
ca: Option<String>,
},
}
/// A sink this config may not have (`holger-audit-9`).
fn refused(why: impl std::fmt::Display) -> String {
holger_errcode::AUDIT_SIEM_REFUSED.tag(why)
}
/// Whether `ip` is an address that stays inside the site.
pub fn is_local_ip(ip: &IpAddr) -> bool {
match ip {
IpAddr::V4(v) => v.is_loopback() || v.is_private() || v.is_link_local(),
IpAddr::V6(v) => {
let s = v.segments()[0];
v.is_loopback() || (s & 0xfe00) == 0xfc00 || (s & 0xffc0) == 0xfe80
}
}
}
fn host_of(addr: &str) -> &str {
let a = addr.trim();
if let Some(rest) = a.strip_prefix('[') {
return rest.split(']').next().unwrap_or(rest);
}
match a.rsplit_once(':') {
Some((h, _)) => h,
None => a,
}
}
fn local_host(host: &str) -> bool {
host == "localhost"
|| host
.parse::<IpAddr>()
.map(|ip| is_local_ip(&ip))
.unwrap_or(false)
}
impl SinkSettings {
/// Refuse a sink that cannot mean what it says, or that leaves the site
/// when `local_only` is set.
pub fn validate(&self, local_only: bool) -> Result<(), String> {
use SinkFormat::*;
use SinkTarget::*;
if self.name.trim().is_empty()
|| !self
.name
.chars()
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
{
return Err(refused(format!(
"sink name '{}' must be non-empty [A-Za-z0-9._-]",
self.name
)));
}
let ok = matches!(
(self.format, &self.target),
(
JsonLines | Syslog | Cef,
File(_) | Udp(_) | Tcp(_) | Tls { .. }
) | (Otlp, Http { .. })
);
if !ok {
return Err(refused(format!(
"sink '{}': format {:?} cannot go to {:?} (OTLP goes to Http; the others to File, Udp, Tcp or Tls)",
self.name, self.format, self.target
)));
}
if self.facility > 23 {
return Err(refused(format!(
"sink '{}': syslog facility {} is not 0..=23",
self.name, self.facility
)));
}
if let Http { url, ca } = &self.target {
let u =
HttpUrl::parse(url).map_err(|e| refused(format!("sink '{}': {e}", self.name)))?;
if u.tls && ca.is_none() {
return Err(refused(format!(
"sink '{}': an https endpoint needs `ca`",
self.name
)));
}
}
if local_only {
let host = match &self.target {
File(_) => None,
Udp(a) | Tcp(a) | Tls { addr: a, .. } => Some(host_of(a).to_string()),
Http { url, .. } => HttpUrl::parse(url).ok().map(|u| u.host),
};
if let Some(h) = host {
if !local_host(&h) {
return Err(refused(format!(
"sink '{}' sends to '{h}', and this side exports to local sinks only (a file, \
or a loopback, private, link-local or unique-local IP address)",
self.name
)));
}
}
}
Ok(())
}
}
impl SiemSettings {
/// Whether this build forces `local_only` (the zero-egress airgap tiers).
pub const FORCED_LOCAL: bool = cfg!(not(feature = "online"));
/// The enabled sinks, each validated. Any refusal refuses all: an operator
/// who configured an export must not boot believing it runs.
pub fn enabled_sinks(&self) -> Result<Vec<SinkSettings>, String> {
// HOL-NCSC: `network: Offline` implies `local_only`.
let local = self.local_only || Self::FORCED_LOCAL || crate::egress::is_offline();
let mut seen = std::collections::BTreeSet::new();
let mut out = Vec::new();
for s in self.sinks.iter().filter(|s| s.enabled) {
s.validate(local)?;
if !seen.insert(s.name.clone()) {
return Err(refused(format!("two sinks are named '{}'", s.name)));
}
out.push(s.clone());
}
Ok(out)
}
}
// ── the formats ────────────────────────────────────────────────────────────
/// One audit line as the formatters see it: the raw text and its fields.
#[derive(Debug, Clone)]
pub struct Line {
/// The exact text in the log, without the newline.
pub raw: String,
/// Its fields.
pub obj: serde_json::Map<String, serde_json::Value>,
}
impl Line {
fn s(&self, k: &str) -> String {
match self.obj.get(k) {
Some(serde_json::Value::String(s)) => s.clone(),
Some(serde_json::Value::Null) | None => String::new(),
Some(v) => v.to_string(),
}
}
fn n(&self, k: &str) -> i64 {
self.obj.get(k).and_then(|v| v.as_i64()).unwrap_or(0)
}
fn is_checkpoint(&self) -> bool {
self.s("kind") == super::chain::CHECKPOINT_KIND
}
/// The step name: the trail kind, else the action, else `checkpoint`.
fn step(&self) -> String {
if self.is_checkpoint() {
return "checkpoint".into();
}
let t = self.s("trail");
if t.is_empty() {
self.s("action")
} else {
t
}
}
fn status(&self) -> i64 {
self.n("status")
}
}
/// `YYYY-MM-DDTHH:MM:SS.ffffffZ` (RFC 3339, UTC, microseconds — RFC 5424 §6.2.3).
pub fn rfc3339_micros(ts_nanos: i64) -> String {
let secs = ts_nanos.div_euclid(1_000_000_000);
let micros = ts_nanos.rem_euclid(1_000_000_000) / 1000;
let days = secs.div_euclid(86_400);
let sod = secs.rem_euclid(86_400);
// Howard Hinnant's civil_from_days.
let z = days + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let y = if m <= 2 { y + 1 } else { y };
format!(
"{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}.{micros:06}Z",
sod / 3600,
(sod % 3600) / 60,
sod % 60
)
}
fn this_host() -> String {
std::fs::read_to_string("/proc/sys/kernel/hostname")
.or_else(|_| std::fs::read_to_string("/etc/hostname"))
.map(|s| s.trim().to_string())
.ok()
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "-".into())
}
/// RFC 5424 printable US-ASCII, no spaces, at most `max` characters; `-` if empty.
fn header_token(s: &str, max: usize) -> String {
let t: String = s
.chars()
.filter(|c| ('!'..='~').contains(c))
.take(max)
.collect();
if t.is_empty() {
"-".into()
} else {
t
}
}
/// RFC 5424 §6.3.3: `"`, `\` and `]` are escaped inside a PARAM-VALUE.
pub fn sd_escape(v: &str) -> String {
let mut o = String::with_capacity(v.len());
for c in v.chars() {
if matches!(c, '"' | '\\' | ']') {
o.push('\\');
}
o.push(c);
}
o
}
fn syslog_severity(l: &Line) -> u8 {
match l.status() {
s if s >= 500 => 3,
s if s >= 400 => 4,
_ if l.is_checkpoint() => 6,
_ => 5,
}
}
const SD_PARAMS: [&str; 12] = [
"seq", "hash", "prev", "ident", "action", "repo", "artifact", "sha256", "trail", "ip",
"status", "bytes",
];
/// **One RFC 5424 message** for one audit line (no framing).
pub fn to_syslog(l: &Line, sink: &SinkSettings, host: &str) -> String {
let pri = sink.facility as u16 * 8 + syslog_severity(l) as u16;
let mut sd = format!("[{}", header_token(&sink.sd_id, 32));
for k in SD_PARAMS {
let v = l.s(k);
if !v.is_empty() {
sd.push_str(&format!(" {k}=\"{}\"", sd_escape(&v)));
}
}
if l.is_checkpoint() {
sd.push_str(&format!(
" kind=\"checkpoint\" head=\"{}\"",
sd_escape(&l.s("head"))
));
}
sd.push(']');
let msg = if l.is_checkpoint() {
format!("audit checkpoint at head {}", l.s("head"))
} else {
l.s("detail")
};
format!(
"<{pri}>1 {} {} holger {} {} {sd} \u{FEFF}{msg}",
rfc3339_micros(l.n("ts")),
header_token(host, 255),
std::process::id(),
header_token(&l.step(), 32),
)
}
/// CEF header escaping: `\` and `|`; a newline becomes a space.
pub fn cef_header_escape(v: &str) -> String {
v.replace('\\', "\\\\")
.replace('|', "\\|")
.replace(['\r', '\n'], " ")
}
/// CEF extension escaping: `\`, `=`, and newlines as `\n` / `\r`.
pub fn cef_ext_escape(v: &str) -> String {
v.replace('\\', "\\\\")
.replace('=', "\\=")
.replace('\n', "\\n")
.replace('\r', "\\r")
}
fn cef_severity(l: &Line) -> u8 {
match (l.status(), l.s("trail").as_str()) {
(s, _) if s >= 500 => 8,
(_, "policy_denied") => 7,
(s, _) if s >= 400 => 6,
_ => 3,
}
}
/// **One CEF event** (`CEF:0|…`) for one audit line.
pub fn to_cef(l: &Line) -> String {
let step = l.step();
let name = if l.is_checkpoint() {
"audit checkpoint".to_string()
} else if l.s("trail").is_empty() {
format!("audit {}", l.s("action"))
} else {
format!("artifact {}", step.replace('_', " "))
};
let mut ext: Vec<(&str, String)> = vec![
("rt", (l.n("ts") / 1_000_000).to_string()),
("suser", l.s("ident")),
("act", l.s("action")),
("cs1Label", "repo".into()),
("cs1", l.s("repo")),
("fname", l.s("artifact")),
("fileHash", l.s("sha256")),
("outcome", l.s("status")),
("cs2Label", "trail".into()),
("cs2", l.s("trail")),
("cn1Label", "seq".into()),
("cn1", l.s("seq")),
("cs3Label", "chainHash".into()),
("cs3", l.s("hash")),
("in", l.s("bytes")),
(
"msg",
if l.is_checkpoint() {
l.s("head")
} else {
l.s("detail")
},
),
];
let ip = l.s("ip");
let host = host_of(&ip);
if host.parse::<IpAddr>().is_ok() {
ext.push(("src", host.to_string()));
}
let ext: Vec<String> = ext
.into_iter()
.filter(|(_, v)| !v.is_empty())
.map(|(k, v)| format!("{k}={}", cef_ext_escape(&v)))
.collect();
format!(
"CEF:0|Holger|holger-server|{}|{}|{}|{}|{}",
cef_header_escape(env!("CARGO_PKG_VERSION")),
cef_header_escape(&step),
cef_header_escape(&name),
cef_severity(l),
ext.join(" ")
)
}
/// CEF carried as the MSG of an RFC 5424 message (no structured data).
fn cef_over_syslog(l: &Line, sink: &SinkSettings, host: &str) -> String {
let pri = sink.facility as u16 * 8 + syslog_severity(l) as u16;
format!(
"<{pri}>1 {} {} holger {} {} - {}",
rfc3339_micros(l.n("ts")),
header_token(host, 255),
std::process::id(),
header_token(&l.step(), 32),
to_cef(l)
)
}
fn otlp_attr_str(k: &str, v: String) -> serde_json::Value {
serde_json::json!({ "key": k, "value": { "stringValue": v } })
}
fn otlp_attr_int(k: &str, v: i64) -> serde_json::Value {
serde_json::json!({ "key": k, "value": { "intValue": v.to_string() } })
}
/// **One OTLP `LogRecord`** (JSON encoding) for one audit line.
pub fn to_otlp_record(l: &Line) -> serde_json::Value {
let (num, text) = match l.status() {
s if s >= 500 => (17, "ERROR"),
s if s >= 400 => (13, "WARN"),
_ => (9, "INFO"),
};
let mut attrs = vec![
otlp_attr_int("holger.audit.seq", l.n("seq")),
otlp_attr_str("holger.audit.hash", l.s("hash")),
otlp_attr_str("holger.audit.prev", l.s("prev")),
otlp_attr_str("holger.audit.step", l.step()),
];
for (k, f) in [
("enduser.id", "ident"),
("holger.audit.action", "action"),
("holger.repository", "repo"),
("holger.artifact", "artifact"),
("holger.artifact.sha256", "sha256"),
("holger.audit.trail", "trail"),
("client.address", "ip"),
("holger.audit.head", "head"),
] {
let v = l.s(f);
if !v.is_empty() {
attrs.push(otlp_attr_str(k, v));
}
}
if !l.is_checkpoint() {
attrs.push(otlp_attr_int("holger.status", l.status()));
attrs.push(otlp_attr_int("holger.bytes", l.n("bytes")));
}
serde_json::json!({
"timeUnixNano": l.n("ts").to_string(),
"observedTimeUnixNano": super::now_nanos().to_string(),
"severityNumber": num,
"severityText": text,
"body": { "stringValue": if l.is_checkpoint() { format!("audit checkpoint at head {}", l.s("head")) } else { l.s("detail") } },
"attributes": attrs,
})
}
/// **The OTLP/HTTP JSON body** (`ExportLogsServiceRequest`) for a batch.
pub fn to_otlp_request(lines: &[Line], host: &str) -> String {
let records: Vec<serde_json::Value> = lines.iter().map(to_otlp_record).collect();
serde_json::json!({
"resourceLogs": [{
"resource": { "attributes": [
otlp_attr_str("service.name", "holger".into()),
otlp_attr_str("service.version", env!("CARGO_PKG_VERSION").into()),
otlp_attr_str("host.name", host.into()),
]},
"scopeLogs": [{
"scope": { "name": "holger.audit", "version": env!("CARGO_PKG_VERSION") },
"logRecords": records,
}],
}]
})
.to_string()
}
// ── the transports ─────────────────────────────────────────────────────────
/// A parsed OTLP endpoint.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HttpUrl {
pub tls: bool,
pub host: String,
pub port: u16,
pub path: String,
}
impl HttpUrl {
pub fn parse(url: &str) -> Result<Self, String> {
let (tls, rest) = if let Some(r) = url.strip_prefix("https://") {
(true, r)
} else if let Some(r) = url.strip_prefix("http://") {
(false, r)
} else {
return Err(format!("'{url}' is not an http:// or https:// URL"));
};
let (auth, path) = match rest.find('/') {
Some(i) => (&rest[..i], &rest[i..]),
None => (rest, ""),
};
let path = if path.is_empty() || path == "/" {
"/v1/logs".to_string()
} else {
path.to_string()
};
let (host, port) = if let Some(r) = auth.strip_prefix('[') {
let (h, p) = r
.split_once(']')
.ok_or_else(|| format!("'{url}': unclosed IPv6 literal"))?;
(h.to_string(), p.strip_prefix(':').map(|p| p.to_string()))
} else {
match auth.rsplit_once(':') {
Some((h, p)) => (h.to_string(), Some(p.to_string())),
None => (auth.to_string(), None),
}
};
let port = match port {
Some(p) => p.parse().map_err(|_| format!("'{url}': bad port"))?,
None if tls => 443,
None => 80,
};
if host.is_empty() {
return Err(format!("'{url}' names no host"));
}
Ok(Self {
tls,
host,
port,
path,
})
}
fn addr(&self) -> String {
if self.host.contains(':') {
format!("[{}]:{}", self.host, self.port)
} else {
format!("{}:{}", self.host, self.port)
}
}
}
const IO_TIMEOUT: Duration = Duration::from_secs(10);
fn connect_tcp(addr: &str) -> Result<TcpStream, String> {
crate::egress::check_site_local("siem", addr).map_err(|e| e.to_string())?;
let mut last = format!("'{addr}' resolves to nothing");
for a in addr
.to_socket_addrs()
.map_err(|e| format!("'{addr}': {e}"))?
{
match TcpStream::connect_timeout(&a, IO_TIMEOUT) {
Ok(s) => {
let _ = s.set_write_timeout(Some(IO_TIMEOUT));
let _ = s.set_read_timeout(Some(IO_TIMEOUT));
return Ok(s);
}
Err(e) => last = format!("connecting to {a}: {e}"),
}
}
Err(last)
}
fn tls_config(ca_pem_file: &str) -> Result<Arc<rustls::ClientConfig>, String> {
use rustls::pki_types::pem::PemObject as _;
use rustls::pki_types::CertificateDer;
let pem =
std::fs::read(ca_pem_file).map_err(|e| format!("reading the CA {ca_pem_file}: {e}"))?;
let mut roots = rustls::RootCertStore::empty();
for c in CertificateDer::pem_slice_iter(&pem) {
let c = c.map_err(|e| format!("{ca_pem_file}: {e}"))?;
roots.add(c).map_err(|e| format!("{ca_pem_file}: {e}"))?;
}
if roots.is_empty() {
return Err(format!("{ca_pem_file} holds no certificate"));
}
let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(
rustls::crypto::ring::default_provider(),
))
.with_safe_default_protocol_versions()
.map_err(|e| format!("TLS: {e}"))?
.with_root_certificates(roots)
.with_no_client_auth();
Ok(Arc::new(cfg))
}
type TlsStream = rustls::StreamOwned<rustls::ClientConnection, TcpStream>;
fn connect_tls(addr: &str, ca: &str, server_name: Option<&str>) -> Result<TlsStream, String> {
let cfg = tls_config(ca)?;
let name = server_name
.map(str::to_string)
.unwrap_or_else(|| host_of(addr).to_string());
let sn = rustls::pki_types::ServerName::try_from(name.clone())
.map_err(|e| format!("server name '{name}': {e}"))?;
let conn = rustls::ClientConnection::new(cfg, sn).map_err(|e| format!("TLS: {e}"))?;
let tcp = connect_tcp(addr)?;
let mut s = rustls::StreamOwned::new(conn, tcp);
while s.conn.is_handshaking() {
s.conn
.complete_io(&mut s.sock)
.map_err(|e| format!("TLS handshake with {addr}: {e}"))?;
}
Ok(s)
}
enum Conn {
None,
Tcp(TcpStream),
Tls(Box<TlsStream>),
}
/// **A sink, ready to send.** Holds the connection across sends and drops it
/// on any error, so the next try reconnects.
pub struct Sink {
settings: SinkSettings,
host: String,
conn: Conn,
}
impl Sink {
pub fn new(settings: SinkSettings) -> Self {
let host = settings.hostname.clone().unwrap_or_else(this_host);
Self {
settings,
host,
conn: Conn::None,
}
}
pub fn name(&self) -> &str {
&self.settings.name
}
/// The messages one batch becomes, unframed (one per line; one body for OTLP).
pub fn render(&self, lines: &[Line]) -> Vec<String> {
let net = !matches!(self.settings.target, SinkTarget::File(_));
match self.settings.format {
SinkFormat::JsonLines => lines.iter().map(|l| l.raw.clone()).collect(),
SinkFormat::Syslog => lines
.iter()
.map(|l| to_syslog(l, &self.settings, &self.host))
.collect(),
SinkFormat::Cef if net => lines
.iter()
.map(|l| cef_over_syslog(l, &self.settings, &self.host))
.collect(),
SinkFormat::Cef => lines.iter().map(to_cef).collect(),
SinkFormat::Otlp => vec![to_otlp_request(lines, &self.host)],
}
}
/// Send one batch. `Ok` only when every message of it was handed over.
pub fn send(&mut self, lines: &[Line]) -> Result<(), String> {
if lines.is_empty() {
return Ok(());
}
let msgs = self.render(lines);
let out = self.send_msgs(&msgs);
if out.is_err() {
self.conn = Conn::None;
}
out
}
fn stream(&mut self) -> Result<&mut dyn Write, String> {
if matches!(self.conn, Conn::None) {
self.conn = match &self.settings.target {
SinkTarget::Tcp(a) => Conn::Tcp(connect_tcp(a)?),
SinkTarget::Tls {
addr,
ca,
server_name,
} => Conn::Tls(Box::new(connect_tls(addr, ca, server_name.as_deref())?)),
_ => return Err("not a stream target".into()),
};
}
let w: &mut dyn Write = match &mut self.conn {
Conn::Tcp(s) => s,
Conn::Tls(s) => s.as_mut(),
Conn::None => return Err("no connection".into()),
};
Ok(w)
}
fn send_msgs(&mut self, msgs: &[String]) -> Result<(), String> {
let json = self.settings.format == SinkFormat::JsonLines;
match self.settings.target.clone() {
SinkTarget::File(p) => {
let mut f = OpenOptions::new()
.create(true)
.append(true)
.open(&p)
.map_err(|e| format!("{p}: {e}"))?;
let mut buf = String::new();
for m in msgs {
buf.push_str(m);
buf.push('\n');
}
f.write_all(buf.as_bytes()).map_err(|e| format!("{p}: {e}"))
}
SinkTarget::Udp(a) => {
let dest = a
.to_socket_addrs()
.map_err(|e| format!("'{a}': {e}"))?
.next()
.ok_or_else(|| format!("'{a}' resolves to nothing"))?;
let bind = if dest.is_ipv6() {
"[::]:0"
} else {
"0.0.0.0:0"
};
crate::egress::check_site_local("siem", a.as_str()).map_err(|e| e.to_string())?;
let sock =
UdpSocket::bind(bind).map_err(|e| format!("binding a UDP socket: {e}"))?;
for m in msgs {
sock.send_to(m.as_bytes(), dest)
.map_err(|e| format!("sending to {dest}: {e}"))?;
}
Ok(())
}
SinkTarget::Tcp(_) | SinkTarget::Tls { .. } => {
let mut buf = Vec::new();
for m in msgs {
if json {
buf.extend_from_slice(m.as_bytes());
buf.push(b'\n');
} else {
buf.extend_from_slice(format!("{} ", m.len()).as_bytes());
buf.extend_from_slice(m.as_bytes());
}
}
let w = self.stream()?;
w.write_all(&buf).map_err(|e| format!("writing: {e}"))?;
w.flush().map_err(|e| format!("flushing: {e}"))
}
SinkTarget::Http { url, ca } => {
let u = HttpUrl::parse(&url)?;
let body = &msgs[0];
let mut req = format!(
"POST {} HTTP/1.1\r\nHost: {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\nUser-Agent: holger/{}\r\n",
u.path,
u.addr(),
body.len(),
env!("CARGO_PKG_VERSION")
);
for (k, v) in &self.settings.headers {
req.push_str(&format!("{}: {}\r\n", k.trim(), v.trim()));
}
req.push_str("\r\n");
req.push_str(body);
let mut reply = Vec::new();
if u.tls {
let ca = ca.ok_or("an https endpoint needs `ca`")?;
let mut s = connect_tls(&u.addr(), &ca, Some(&u.host))?;
s.write_all(req.as_bytes())
.map_err(|e| format!("writing: {e}"))?;
s.flush().map_err(|e| format!("flushing: {e}"))?;
let _ = s.read_to_end(&mut reply);
} else {
let mut s = connect_tcp(&u.addr())?;
s.write_all(req.as_bytes())
.map_err(|e| format!("writing: {e}"))?;
let _ = s.read_to_end(&mut reply);
}
let head = String::from_utf8_lossy(&reply);
let status = head
.lines()
.next()
.unwrap_or_default()
.split_whitespace()
.nth(1)
.unwrap_or_default()
.to_string();
if status.starts_with('2') {
Ok(())
} else {
Err(format!(
"the collector at {url} answered '{}'",
head.lines().next().unwrap_or("nothing")
))
}
}
}
}
}
// ── the cursor and the exporter ────────────────────────────────────────────
/// Where a sink stands: the byte offset of the next line to send, and its seq.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Cursor {
pub offset: u64,
pub sent: u64,
}
/// The directory the cursors live in, under the audit directory.
pub fn cursor_dir(audit_dir: &Path) -> PathBuf {
audit_dir.join("siem")
}
impl Cursor {
fn path(audit_dir: &Path, sink: &str) -> PathBuf {
cursor_dir(audit_dir).join(format!("{sink}.cursor"))
}
pub fn load(audit_dir: &Path, sink: &str) -> Self {
let text = std::fs::read_to_string(Self::path(audit_dir, sink)).unwrap_or_default();
let mut it = text
.split_whitespace()
.map(|t| t.parse::<u64>().unwrap_or(0));
Self {
offset: it.next().unwrap_or(0),
sent: it.next().unwrap_or(0),
}
}
pub fn store(&self, audit_dir: &Path, sink: &str) -> Result<(), String> {
let dir = cursor_dir(audit_dir);
std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
let p = Self::path(audit_dir, sink);
let tmp = p.with_extension("cursor.tmp");
std::fs::write(&tmp, format!("{} {}\n", self.offset, self.sent))
.map_err(|e| format!("{}: {e}", tmp.display()))?;
std::fs::rename(&tmp, &p).map_err(|e| format!("{}: {e}", p.display()))
}
}
/// Up to `max` whole lines from `offset`, and the offset after them. A last
/// line without its newline is still being written and is left for later.
pub fn read_from(path: &Path, offset: u64, max: usize) -> Result<(Vec<Line>, u64), String> {
let mut f = match File::open(path) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((Vec::new(), offset)),
Err(e) => return Err(format!("{}: {e}", path.display())),
};
let len = f.metadata().map_err(|e| e.to_string())?.len();
let offset = if len < offset {
log::warn!("audit export: {} is shorter than the cursor ({len} < {offset}); exporting it again from the start", path.display());
0
} else {
offset
};
f.seek(SeekFrom::Start(offset)).map_err(|e| e.to_string())?;
let mut r = BufReader::new(f);
let mut out = Vec::new();
let mut at = offset;
let mut buf = String::new();
while out.len() < max {
buf.clear();
let n = r.read_line(&mut buf).map_err(|e| e.to_string())?;
if n == 0 || !buf.ends_with('\n') {
break;
}
at += n as u64;
let raw = buf.trim_end_matches(['\n', '\r']).to_string();
if raw.trim().is_empty() {
continue;
}
let obj = match serde_json::from_str::<serde_json::Value>(&raw) {
Ok(serde_json::Value::Object(m)) => m,
_ => {
// An unreadable line is still evidence: send it as it is.
let mut m = serde_json::Map::new();
m.insert("detail".into(), raw.clone().into());
m
}
};
out.push(Line { raw, obj });
}
Ok((out, at))
}
/// **One sink's follower of the audit file.**
pub struct Exporter {
sink: Sink,
audit_dir: PathBuf,
batch: usize,
retry: Duration,
}
impl Exporter {
pub fn new(
settings: SinkSettings,
audit_dir: impl Into<PathBuf>,
batch: usize,
retry_ms: u64,
) -> Self {
Self {
sink: Sink::new(settings),
audit_dir: audit_dir.into(),
batch: batch.max(1),
retry: Duration::from_millis(retry_ms),
}
}
pub fn name(&self) -> &str {
self.sink.name()
}
/// Send one batch from the cursor. `Ok(n)` lines sent (0 = caught up);
/// `Err` leaves the cursor where it was.
pub fn step(&mut self) -> Result<usize, String> {
let mut cur = Cursor::load(&self.audit_dir, self.sink.name());
let (lines, next) = read_from(
&self.audit_dir.join(super::AUDIT_FILE),
cur.offset,
self.batch,
)?;
if lines.is_empty() {
if next != cur.offset {
cur.offset = next;
cur.store(&self.audit_dir, self.sink.name())?;
}
return Ok(0);
}
self.sink.send(&lines)?;
cur.offset = next;
cur.sent += lines.len() as u64;
cur.store(&self.audit_dir, self.sink.name())?;
Ok(lines.len())
}
/// Send until caught up, or until a send fails. Lines sent.
pub fn catch_up(&mut self) -> Result<u64, String> {
let mut total = 0u64;
loop {
match self.step()? {
0 => return Ok(total),
n => total += n as u64,
}
}
}
/// Follow the file until `stop`: send what is there, wait on `signal` (or
/// a poll interval), retry a failure after `retry`.
pub fn run(mut self, signal: Option<Arc<TailSignal>>, stop: Arc<AtomicBool>) {
let poll = Duration::from_millis(500);
let mut seen = signal.as_ref().map(|s| s.now()).unwrap_or(0);
while !stop.load(Ordering::Relaxed) {
match self.catch_up() {
Ok(_) => {
seen = match &signal {
Some(s) => s.wait_past(seen, poll),
None => {
std::thread::sleep(poll);
seen
}
};
}
Err(e) => {
log::warn!(
"audit export '{}': {e}; retrying from the cursor in {} ms",
self.name(),
self.retry.as_millis()
);
let until = std::time::Instant::now() + self.retry;
while !stop.load(Ordering::Relaxed) && std::time::Instant::now() < until {
std::thread::sleep(Duration::from_millis(50).min(self.retry));
}
}
}
}
}
}
/// The running exporters; dropping it stops them (they finish their send).
pub struct ExporterHandles {
stop: Arc<AtomicBool>,
threads: Vec<std::thread::JoinHandle<()>>,
/// The sink names, in config order.
pub names: Vec<String>,
}
impl ExporterHandles {
/// Stop every exporter and wait for it.
pub fn stop(mut self) {
self.stop.store(true, Ordering::Relaxed);
for t in self.threads.drain(..) {
let _ = t.join();
}
}
}
impl Drop for ExporterHandles {
fn drop(&mut self) {
self.stop.store(true, Ordering::Relaxed);
}
}
/// **Start one follower per enabled sink.** Each is a named thread of its own,
/// not a job of the queue: it runs in every tier (the airgap tiers carry no job
/// queue), its whole state is the cursor file, and it must never contend with
/// the writer for anything but the page cache.
pub fn start(
settings: &SiemSettings,
audit_dir: &Path,
signal: Option<Arc<TailSignal>>,
) -> Result<ExporterHandles, String> {
let sinks = settings.enabled_sinks()?;
let stop = Arc::new(AtomicBool::new(false));
let mut threads = Vec::new();
let mut names = Vec::new();
for s in sinks {
names.push(s.name.clone());
let ex = Exporter::new(s.clone(), audit_dir, settings.batch, settings.retry_ms);
let (sig, st) = (signal.clone(), Arc::clone(&stop));
// A32-14, stays a task: one exporter per SIEM sink; its handle is kept and it ends at its stop flag.
let t = std::thread::Builder::new()
.name(format!("holger-siem-{}", s.name))
.spawn(move || ex.run(sig, st))
.map_err(|e| format!("starting the '{}' exporter: {e}", s.name))?;
threads.push(t);
log::info!(
"audit export '{}' ENABLED: {:?} to {:?}",
s.name,
s.format,
s.target
);
}
Ok(ExporterHandles {
stop,
threads,
names,
})
}
/// Each sink's name with the lines it sent, or why it stopped.
pub type SinkResults = Vec<(String, Result<u64, String>)>;
/// **The batch export**: every enabled sink sent from its cursor to the end of
/// the log, once. One result per sink: the lines sent, or why it stopped.
pub fn export_once(settings: &SiemSettings, audit_dir: &Path) -> Result<SinkResults, String> {
Ok(settings
.enabled_sinks()?
.into_iter()
.map(|s| {
let name = s.name.clone();
let mut ex = Exporter::new(s, audit_dir, settings.batch, settings.retry_ms);
(name, ex.catch_up())
})
.collect())
}
#[cfg(test)]
mod tests {
use super::*;
fn line(obj: serde_json::Value) -> Line {
let serde_json::Value::Object(m) = obj.clone() else {
panic!()
};
Line {
raw: obj.to_string(),
obj: m,
}
}
#[test]
fn rfc3339_is_exact() {
assert_eq!(rfc3339_micros(0), "1970-01-01T00:00:00.000000Z");
// 2026-10-19T08:30:05.123456Z
assert_eq!(
rfc3339_micros(1_792_398_605_123_456_789),
"2026-10-19T08:30:05.123456Z"
);
assert_eq!(
rfc3339_micros(951_782_400_000_000_000),
"2000-02-29T00:00:00.000000Z"
);
}
#[test]
fn local_only_refuses_what_leaves_the_site() {
let s = |t: SinkTarget, f: SinkFormat| SinkSettings {
name: "x".into(),
enabled: true,
format: f,
target: t,
facility: 13,
sd_id: default_sd_id(),
hostname: None,
headers: vec![],
};
assert!(s(SinkTarget::File("/tmp/x".into()), SinkFormat::JsonLines)
.validate(true)
.is_ok());
assert!(
s(SinkTarget::Udp("127.0.0.1:514".into()), SinkFormat::Syslog)
.validate(true)
.is_ok()
);
assert!(s(SinkTarget::Tcp("10.1.2.3:514".into()), SinkFormat::Cef)
.validate(true)
.is_ok());
assert!(s(SinkTarget::Tcp("[fd00::1]:514".into()), SinkFormat::Cef)
.validate(true)
.is_ok());
let e = s(SinkTarget::Tcp("8.8.8.8:514".into()), SinkFormat::Syslog)
.validate(true)
.unwrap_err();
assert!(e.starts_with("[holger-audit-9]"), "{e}");
assert!(s(
SinkTarget::Tcp("siem.example.com:514".into()),
SinkFormat::Syslog
)
.validate(true)
.is_err());
assert!(s(
SinkTarget::Http {
url: "http://1.2.3.4:4318".into(),
ca: None
},
SinkFormat::Otlp
)
.validate(true)
.is_err());
assert!(s(SinkTarget::Tcp("8.8.8.8:514".into()), SinkFormat::Syslog)
.validate(false)
.is_ok());
assert!(
s(SinkTarget::Udp("127.0.0.1:514".into()), SinkFormat::Otlp)
.validate(false)
.is_err(),
"OTLP goes to Http"
);
}
#[test]
fn escaping_follows_the_specs() {
assert_eq!(sd_escape(r#"a"b\c]d"#), r#"a\"b\\c\]d"#);
assert_eq!(cef_header_escape("a|b\\c"), "a\\|b\\\\c");
assert_eq!(cef_ext_escape("a=b\\c\nd"), "a\\=b\\\\c\\nd");
let l = line(
serde_json::json!({"ts": 0, "action": "upload", "status": 200, "detail": "x=y", "ip": "10.0.0.1:5"}),
);
let c = to_cef(&l);
assert!(c.starts_with("CEF:0|Holger|holger-server|"), "{c}");
assert!(c.contains("msg=x\\=y"), "{c}");
assert!(c.contains("src=10.0.0.1"), "{c}");
}
}