use std::sync::Arc;
use gunnar_grpc::auth::{AuditLedger, AuditRecord};
use gunnar_grpc::{TransferGate, TransferHold, TransferPermit, TransferRefused};
use gunnar_ssh::governor::OpClass;
use gunnar_ssh::{Admission, AdmissionGuard};
pub fn control_transfer_gate(admission: Arc<Admission>) -> Arc<dyn TransferGate> {
Arc::new(SshTransferGate { admission })
}
struct SshTransferGate {
admission: Arc<Admission>,
}
impl TransferGate for SshTransferGate {
fn admit_transfer(
&self,
principal: &str,
namespace: &str,
) -> Result<TransferPermit, TransferRefused> {
match self
.admission
.try_admit(principal, namespace, OpClass::Clone)
{
Ok(guard) => Ok(TransferPermit(Box::new(GuardHold(guard)))),
Err(refusal) => {
let stats = self.admission.stats();
Err(TransferRefused {
message: refusal.client_message(),
in_flight: stats.in_flight,
peak_in_flight: stats.peak_in_flight,
limit: stats.limit,
reserved_bytes: stats.reserved_bytes,
refused_total: stats.refused(),
})
}
}
}
}
struct GuardHold(AdmissionGuard);
impl TransferHold for GuardHold {
fn entered_service(&self) {
self.0.entered_service();
}
}
pub fn control_audit_ledger(chain: Arc<gunnar_core::AuditChain>) -> Arc<dyn AuditLedger> {
control_audit_ledger_with_window(chain, READ_WINDOW)
}
pub fn control_audit_ledger_with_window(
chain: Arc<gunnar_core::AuditChain>,
window: std::time::Duration,
) -> Arc<dyn AuditLedger> {
let every = window
.min(std::time::Duration::from_secs(5))
.max(std::time::Duration::from_millis(50));
control_audit_ledger_with(chain, window, every)
}
pub fn control_audit_ledger_with(
chain: Arc<gunnar_core::AuditChain>,
window: std::time::Duration,
every: std::time::Duration,
) -> Arc<dyn AuditLedger> {
ledger_on(chain, window, every, Arc::new(gunnar_core::now_ms))
}
pub fn activity_ledger(
trail: Option<Arc<dyn AuditLedger>>,
activity: Arc<gunnar_core::PrincipalActivity>,
) -> Arc<dyn AuditLedger> {
Arc::new(ActivityLedger { trail, activity })
}
struct ActivityLedger {
trail: Option<Arc<dyn AuditLedger>>,
activity: Arc<gunnar_core::PrincipalActivity>,
}
impl AuditLedger for ActivityLedger {
fn record(&self, record: AuditRecord<'_>) {
if record.decision == "allow" && gunnar_grpc::auth::changes_state(record.method) {
self.activity
.operated(record.principal, gunnar_core::now_ms());
}
if let Some(trail) = &self.trail {
trail.record(record);
}
}
fn record_change(&self, record: AuditRecord<'_>) -> Result<(), String> {
if record.decision == "allow" && gunnar_grpc::auth::changes_state(record.method) {
self.activity
.operated(record.principal, gunnar_core::now_ms());
}
match &self.trail {
Some(trail) => trail.record_change(record),
None => Err("this server keeps no audit trail (its chain did not open)".to_owned()),
}
}
}
type Clock = Arc<dyn Fn() -> i64 + Send + Sync>;
pub const READ_SWEEPER: &str = "audit-reads";
fn ledger_on(
chain: Arc<gunnar_core::AuditChain>,
window: std::time::Duration,
every: std::time::Duration,
clock: Clock,
) -> Arc<dyn AuditLedger> {
let ledger = Arc::new(ChainLedger {
chain,
window_ms: i64::try_from(window.as_millis()).unwrap_or(i64::MAX).max(1),
reads: std::sync::Mutex::new(Reads::default()),
clock,
});
let weak = Arc::downgrade(&ledger);
let mut first = true;
let say: gunnar_common::jobs::Say = Arc::new(|l: &str| tracing::error!("{l}"));
let started = gunnar_common::jobs::start(READ_SWEEPER, every, say, move || {
if std::mem::take(&mut first) {
return gunnar_common::jobs::Round::Again;
}
match weak.upgrade() {
Some(ledger) => {
ledger.sweep((ledger.clock)(), false);
gunnar_common::jobs::Round::Again
}
None => gunnar_common::jobs::Round::Done,
}
});
if let Err(e) = started {
tracing::error!("ALERT {READ_SWEEPER} no sweeper ({e}): a read window that closes while nobody reads is written at the next read only");
}
ledger
}
pub const READ_WINDOW: std::time::Duration = std::time::Duration::from_secs(60);
pub const AGGREGATED: &str = "aggregated: ";
fn plain_read(record: &AuditRecord<'_>) -> bool {
record.level == gunnar_grpc::auth::level_name(gunnar_proto::GrantLevel::Read)
&& matches!(record.decision, "allow" | "narrow")
&& !record.reason.starts_with("OVERRIDE:")
}
#[derive(Clone, PartialEq, Eq, Hash)]
struct ReadKey {
method: String,
principal: String,
fingerprint: String,
scope: String,
level: String,
auth_mode: String,
decision: String,
reason: String,
}
impl ReadKey {
fn of(r: &AuditRecord<'_>) -> Self {
ReadKey {
method: r.method.to_owned(),
principal: r.principal.to_owned(),
fingerprint: r.fingerprint.to_owned(),
scope: r.scope.to_owned(),
level: r.level.to_owned(),
auth_mode: r.auth_mode.to_owned(),
decision: r.decision.to_owned(),
reason: r.reason.to_owned(),
}
}
}
#[derive(Clone, Copy)]
struct Window {
first_ms: i64,
last_ms: i64,
more: u64,
}
#[derive(Default)]
struct Reads {
open: std::collections::HashMap<ReadKey, Window>,
closed: Vec<(ReadKey, Window)>,
}
struct ChainLedger {
chain: Arc<gunnar_core::AuditChain>,
window_ms: i64,
reads: std::sync::Mutex<Reads>,
clock: Clock,
}
impl ChainLedger {
fn sweep(&self, now_ms: i64, all: bool) {
let mut due: Vec<(ReadKey, Window)> = {
let mut reads = self.reads.lock().unwrap_or_else(|p| p.into_inner());
let window_ms = self.window_ms;
let closed: Vec<ReadKey> = reads
.open
.iter()
.filter(|(_, w)| all || now_ms - w.first_ms >= window_ms)
.map(|(k, _)| k.clone())
.collect();
let mut due: Vec<(ReadKey, Window)> = std::mem::take(&mut reads.closed);
due.extend(
closed
.into_iter()
.filter_map(|k| reads.open.remove(&k).map(|w| (k, w)))
.filter(|(_, w)| w.more > 0),
);
due
};
due.sort_by_key(|(_, w)| w.first_ms);
for (k, w) in due {
self.write_window(&k, w);
}
}
fn write_window(&self, k: &ReadKey, w: Window) {
let mut reason = format!(
"{AGGREGATED}{} more calls, first {}, last {}, window {} s",
w.more,
w.first_ms,
w.last_ms,
self.window_ms / 1000
);
if !k.reason.is_empty() {
reason.push_str("; ");
reason.push_str(&k.reason);
}
self.write(
AuditRecord {
method: &k.method,
principal: &k.principal,
fingerprint: &k.fingerprint,
scope: &k.scope,
level: &k.level,
auth_mode: &k.auth_mode,
decision: &k.decision,
reason: &reason,
},
w.last_ms,
);
}
}
impl Drop for ChainLedger {
fn drop(&mut self) {
self.sweep((self.clock)(), true);
}
}
impl AuditLedger for ChainLedger {
fn record_change(&self, record: AuditRecord<'_>) -> Result<(), String> {
let now = (self.clock)();
self.append(record, now)
}
fn record(&self, record: AuditRecord<'_>) {
let now = (self.clock)();
if !plain_read(&record) {
self.write(record, now);
return;
}
let first = {
let mut reads = self.reads.lock().unwrap_or_else(|p| p.into_inner());
let key = ReadKey::of(&record);
let window_ms = self.window_ms;
match reads.open.get(&key).copied() {
Some(w) if now - w.first_ms < window_ms => {
if let Some(w) = reads.open.get_mut(&key) {
w.more += 1;
w.last_ms = now;
}
false
}
Some(w) => {
if w.more > 0 {
reads.closed.push((key.clone(), w));
}
reads.open.insert(
key,
Window {
first_ms: now,
last_ms: now,
more: 0,
},
);
true
}
None => {
reads.open.insert(
key,
Window {
first_ms: now,
last_ms: now,
more: 0,
},
);
true
}
}
};
if first {
self.write(record, now);
}
}
}
impl ChainLedger {
fn write(&self, record: AuditRecord<'_>, at_ms: i64) {
let _ = self.append(record, at_ms);
}
fn append(&self, record: AuditRecord<'_>, at_ms: i64) -> Result<(), String> {
let entry = gunnar_core::AuditEvent {
method: record.method,
principal: record.principal,
fingerprint: record.fingerprint,
scope: record.scope,
level: record.level,
auth_mode: record.auth_mode,
decision: record.decision,
reason: record.reason,
};
match self.chain.record(entry, at_ms) {
Ok(gunnar_core::Recorded::Written(_)) => {}
Ok(gunnar_core::Recorded::Queued { pending, .. }) => tracing::info!(
target: gunnar_grpc::AUDIT_TARGET,
method = record.method,
principal = record.principal,
decision = record.decision,
pending,
"the data set is being grown: this decision waits for the audit chain and is \
written after the remount"
),
Ok(gunnar_core::Recorded::NotOriginated) => tracing::info!(
target: gunnar_grpc::AUDIT_TARGET,
method = record.method,
principal = record.principal,
decision = record.decision,
"this box follows its primary (a twin, or restoring): it originates nothing into \
the replicated trail, so this decision is in this log only"
),
Err(err) => {
tracing::error!(
target: gunnar_grpc::AUDIT_TARGET,
error = %err,
method = record.method,
principal = record.principal,
"the control-plane audit chain could not be appended to; this decision is in \
the log and NOT in the trail"
);
return Err(err.to_string());
}
}
Ok(())
}
}
#[cfg(test)]
mod tests {
#[test]
fn the_read_sweeper_is_a_jera_job_that_outlives_a_panicking_sweep() {
let dir = tempfile::tempdir().unwrap();
let chain =
Arc::new(gunnar_core::AuditChain::open(dir.path().join("audit-chain.jsonl")).unwrap());
let calls = Arc::new(std::sync::atomic::AtomicU32::new(0));
let clock: Clock = {
let calls = Arc::clone(&calls);
Arc::new(move || {
if calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0 {
panic!("the clock broke once");
}
1_000_000
})
};
let ledger = ledger_on(
chain,
std::time::Duration::from_millis(50),
std::time::Duration::from_millis(10),
clock,
);
let t = std::time::Instant::now();
let seen = loop {
let d = gunnar_common::jobs::running().into_iter().find(|d| {
d.kind() == READ_SWEEPER
&& d.failures() >= 1
&& d.looks() >= 4
&& d.queue().stats().failed >= 1
});
if d.is_some() || t.elapsed() > std::time::Duration::from_secs(10) {
break d;
}
std::thread::sleep(std::time::Duration::from_millis(5));
};
let d = seen.expect(
"the sweeper is a jera job that went on after its panic, the panic a FAILED row",
);
assert!(d
.last_failure()
.unwrap_or_default()
.contains("the clock broke once"));
drop(ledger);
assert!(d.stop(std::time::Duration::from_secs(5)));
}
use super::*;
use std::sync::Mutex;
#[derive(Clone, Default)]
struct Heard(Arc<Mutex<Vec<u8>>>);
impl std::io::Write for Heard {
fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(b);
Ok(b.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn decision(method: &str) -> AuditRecord<'_> {
AuditRecord {
method,
principal: "monetize",
fingerprint: "SHA256:x",
scope: "server",
level: "admin",
auth_mode: "required",
decision: "allow",
reason: "",
}
}
#[test]
fn a_plain_read_writes_at_most_its_own_entry_on_the_request_thread() {
let dir = tempfile::tempdir().unwrap();
let chain =
Arc::new(gunnar_core::AuditChain::open(dir.path().join("audit-chain.jsonl")).unwrap());
let now = Arc::new(std::sync::atomic::AtomicI64::new(1_000_000));
let clock: Clock = {
let now = Arc::clone(&now);
Arc::new(move || now.load(std::sync::atomic::Ordering::SeqCst))
};
let ledger = ledger_on(
Arc::clone(&chain),
std::time::Duration::from_millis(50),
std::time::Duration::from_secs(3600),
clock,
);
let scopes: Vec<String> = (0..50).map(|i| format!("namespace:t{i}")).collect();
fn read(scope: &str) -> AuditRecord<'_> {
AuditRecord {
method: "Usage.Read",
principal: "monetize",
fingerprint: "SHA256:m",
scope,
level: "read",
auth_mode: "required",
decision: "allow",
reason: "",
}
}
for scope in &scopes {
ledger.record(read(scope));
ledger.record(read(scope));
}
assert_eq!(
chain.entries().unwrap().len(),
50,
"50 first reads, the repeats counted"
);
now.fetch_add(1200, std::sync::atomic::Ordering::SeqCst);
let before = chain.entries().unwrap().len();
ledger.record(read("namespace:someone-else"));
let inline = chain.entries().unwrap().len() - before;
assert_eq!(
inline,
1,
"the request wrote {inline} entries inline: its own first read and {} other windows",
inline.saturating_sub(1)
);
let before = chain.entries().unwrap().len();
ledger.record(read(&scopes[0]));
let inline = chain.entries().unwrap().len() - before;
assert_eq!(
inline,
1,
"a re-read wrote {inline} entries inline: its own first read and {} closed windows",
inline.saturating_sub(1)
);
drop(ledger);
let entries = chain.entries().unwrap();
let counted: u64 = entries
.iter()
.map(|e| match e.reason.strip_prefix(AGGREGATED) {
Some(rest) => rest
.split_whitespace()
.next()
.and_then(|n| n.parse().ok())
.unwrap_or(0),
None => 1,
})
.sum();
assert_eq!(counted, 102, "every read is in the trail");
}
#[test]
fn a_twin_or_a_growing_box_says_its_decisions_at_info_never_error() {
let dir = tempfile::tempdir().unwrap();
let log = Arc::new(gunnar_core::txlog::Log::open(dir.path().join("txlog")).unwrap());
gunnar_core::txlog::sink::register(dir.path(), &log);
let chain = Arc::new(
gunnar_core::AuditChain::open_framed(dir.path().join(gunnar_core::txlog::AUDIT_REL))
.unwrap(),
);
let ledger = control_audit_ledger(Arc::clone(&chain));
let heard = Heard::default();
let h = heard.clone();
let sub = tracing_subscriber::fmt()
.with_writer(move || h.clone())
.with_ansi(false)
.with_max_level(tracing::Level::TRACE)
.finish();
tracing::subscriber::with_default(sub, || {
ledger.record(decision("Repos.Create"));
log.release().unwrap();
ledger.record(decision("Admin.Capacity"));
ledger.record(decision("Admin.GrowStatus"));
log.reacquire().unwrap();
ledger.record(decision("Admin.GrowStatus"));
log.set_follower(true);
ledger.record(decision("Admin.GrowStart"));
});
let said = String::from_utf8(heard.0.lock().unwrap().clone()).unwrap();
assert!(
!said.contains("ERROR"),
"a twin or a growing box is no ERROR:\n{said}"
);
assert_eq!(
chain.verify().unwrap(),
4,
"the grow's two decisions are in the trail after the remount:\n{said}"
);
assert!(
said.contains("INFO") && said.contains("originates nothing"),
"the follower's decision is said at INFO:\n{said}"
);
}
}