use std::sync::Arc;
use std::sync::Mutex;
use std::time::{Duration, Instant};
use arc_swap::{ArcSwap, Guard};
use crate::audit::{AuditLog, ReloadActor};
use crate::catalog::policy::{Config, ResolvedPolicy};
use crate::catalog::{Catalog, Pdp};
use crate::core::crypto_provider::{ProviderAuditEvent, ProviderAuditOutcome};
use crate::decision::DecisionRecord;
use crate::event::EventSource;
use crate::manager::{BackendManager, ManagerError};
use crate::reload::ReloadInputs;
use crate::revocation::JwtRevocationStore;
pub const INITIAL_GENERATION_ID: u64 = 1;
#[derive(Debug)]
pub struct Generation {
id: u64,
catalog: Arc<Catalog>,
policy: ResolvedPolicy,
config: Config,
}
impl Generation {
#[must_use]
pub fn new(
id: u64,
catalog: impl Into<Arc<Catalog>>,
policy: ResolvedPolicy,
config: Config,
) -> Self {
Self {
id,
catalog: catalog.into(),
policy,
config,
}
}
#[must_use]
pub const fn id(&self) -> u64 {
self.id
}
#[must_use]
pub fn pdp(&self) -> Pdp<'_> {
Pdp::new(&self.catalog, &self.policy, &self.config)
}
#[must_use]
pub const fn config(&self) -> &Config {
&self.config
}
#[must_use]
pub const fn policy(&self) -> &ResolvedPolicy {
&self.policy
}
#[must_use]
pub fn catalog(&self) -> &Catalog {
&self.catalog
}
}
pub const DEFAULT_MAX_ENCRYPT_SIZE: usize = 1024 * 1024;
pub const DEFAULT_MAX_PAYLOAD_SIZE: usize = 1024 * 1024;
pub const DEFAULT_ROTATION_GRACE_VERSIONS: u32 = 1;
pub const DEFAULT_SVID_TTL_SECS: u64 = 300;
#[derive(Debug, Clone, Copy)]
pub struct BrokerLimits {
pub max_encrypt_size: usize,
pub max_payload_size: usize,
pub grace_versions: u32,
pub svid_ttl_secs: u64,
pub retain_versions: Option<u32>,
}
impl Default for BrokerLimits {
fn default() -> Self {
Self {
max_encrypt_size: DEFAULT_MAX_ENCRYPT_SIZE,
max_payload_size: DEFAULT_MAX_PAYLOAD_SIZE,
grace_versions: DEFAULT_ROTATION_GRACE_VERSIONS,
svid_ttl_secs: DEFAULT_SVID_TTL_SECS,
retain_versions: None,
}
}
}
impl BrokerLimits {
#[must_use]
pub const fn grace_floor(&self, latest: u32) -> u32 {
let floor = latest.saturating_sub(self.grace_versions);
if floor < 1 { 1 } else { floor }
}
#[must_use]
pub const fn retention_floor(&self, latest: u32) -> Option<u32> {
match self.retain_versions {
Some(retain) => {
let floor = latest.saturating_sub(retain);
Some(if floor < 1 { 1 } else { floor })
}
None => None,
}
}
}
pub const READINESS_CACHE_TTL: Duration = Duration::from_secs(2);
pub const JWKS_CACHE_TTL: Duration = Duration::from_secs(2);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadinessState {
Ready,
RequiredKeyMissing,
BackendUnreachable,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReadinessOutcome {
pub state: ReadinessState,
pub keys_total: u32,
pub keys_present: u32,
pub keys_required_missing: u32,
pub keys_optional_missing: u32,
}
impl ReadinessOutcome {
#[must_use]
pub const fn ready(&self) -> bool {
matches!(self.state, ReadinessState::Ready)
}
}
#[derive(Debug, Clone, Copy)]
struct CachedReadiness {
outcome: ReadinessOutcome,
generation: u64,
expires_at: Instant,
}
#[derive(Debug, Clone)]
pub struct CachedJwks {
pub body: Vec<u8>,
pub etag: String,
generation: u64,
expires_at: Instant,
}
#[derive(Debug)]
pub struct BrokerState {
generation: ArcSwap<Generation>,
manager: BackendManager,
backend_label: String,
agent_version: String,
limits: BrokerLimits,
audit: Option<Arc<AuditLog>>,
events: EventSource,
jwt_revocations: JwtRevocationStore,
reload_inputs: Option<ReloadInputs>,
readiness_cache: Mutex<Option<CachedReadiness>>,
jwks_cache: Mutex<Option<CachedJwks>>,
reload_lock: Mutex<()>,
}
impl BrokerState {
#[must_use]
pub fn new(
catalog: Catalog,
policy: ResolvedPolicy,
config: Config,
manager: BackendManager,
backend_label: impl Into<String>,
) -> Self {
Self::with_limits(
Arc::new(catalog),
policy,
config,
manager,
backend_label,
BrokerLimits::default(),
)
}
#[must_use]
pub fn with_limits(
catalog: impl Into<Arc<Catalog>>,
policy: ResolvedPolicy,
config: Config,
manager: BackendManager,
backend_label: impl Into<String>,
limits: BrokerLimits,
) -> Self {
let generation = Generation::new(INITIAL_GENERATION_ID, catalog, policy, config);
Self {
generation: ArcSwap::from_pointee(generation),
manager,
backend_label: backend_label.into(),
agent_version: env!("CARGO_PKG_VERSION").to_string(),
limits,
audit: None,
events: EventSource::new(),
jwt_revocations: JwtRevocationStore::default(),
reload_inputs: None,
readiness_cache: Mutex::new(None),
jwks_cache: Mutex::new(None),
reload_lock: Mutex::new(()),
}
}
#[must_use]
pub fn with_audit_log(mut self, audit: Arc<AuditLog>) -> Self {
self.audit = Some(audit);
self
}
#[must_use]
pub fn with_jwt_revocations(mut self, jwt_revocations: JwtRevocationStore) -> Self {
self.jwt_revocations = jwt_revocations;
self
}
#[must_use]
pub fn with_version(mut self, version: impl Into<String>) -> Self {
self.agent_version = version.into();
self
}
#[must_use]
pub fn with_reload_inputs(mut self, inputs: ReloadInputs) -> Self {
self.reload_inputs = Some(inputs);
self
}
#[must_use]
pub const fn reload_inputs(&self) -> Option<&ReloadInputs> {
self.reload_inputs.as_ref()
}
#[must_use]
pub const fn reload_lock(&self) -> &Mutex<()> {
&self.reload_lock
}
#[must_use]
pub fn active_generation_id(&self) -> u64 {
self.load_generation().id()
}
#[must_use]
pub fn cached_readiness(&self) -> Option<ReadinessOutcome> {
let generation = self.active_generation_id();
let now = Instant::now();
let cached = {
let guard = self.readiness_cache.lock().ok()?;
(*guard)?
};
(cached.generation == generation && now < cached.expires_at).then_some(cached.outcome)
}
pub fn cache_readiness(&self, generation: u64, outcome: ReadinessOutcome) {
if let Ok(mut guard) = self.readiness_cache.lock() {
*guard = Some(CachedReadiness {
outcome,
generation,
expires_at: Instant::now() + READINESS_CACHE_TTL,
});
}
}
#[must_use]
pub fn cached_jwks(&self) -> Option<CachedJwks> {
let generation = self.active_generation_id();
let now = Instant::now();
let cached = {
let guard = self.jwks_cache.lock().ok()?;
(*guard).clone()?
};
(cached.generation == generation && now < cached.expires_at).then_some(cached)
}
pub fn cache_jwks(&self, generation: u64, body: Vec<u8>, etag: String) {
if let Ok(mut guard) = self.jwks_cache.lock() {
*guard = Some(CachedJwks {
body,
etag,
generation,
expires_at: Instant::now() + JWKS_CACHE_TTL,
});
}
}
pub fn swap_generation(&self, generation: Arc<Generation>) {
self.generation.store(generation);
}
#[must_use]
pub fn load_generation(&self) -> Guard<Arc<Generation>> {
self.generation.load()
}
pub fn record_decision(&self, record: &DecisionRecord) {
record.record();
if let Some(audit) = &self.audit {
audit.append(record);
}
}
pub fn record_provider_event(&self, event: &ProviderAuditEvent<'_>) {
match event.outcome {
ProviderAuditOutcome::Failure | ProviderAuditOutcome::Deny => tracing::warn!(
event = "basil.audit.provider_operation",
op = event.op,
key = event.key_id,
algorithm = event.algorithm,
outcome = ?event.outcome,
reason = event.reason,
"software-custody provider operation",
),
ProviderAuditOutcome::Allow | ProviderAuditOutcome::Success => tracing::info!(
event = "basil.audit.provider_operation",
op = event.op,
key = event.key_id,
algorithm = event.algorithm,
outcome = ?event.outcome,
reason = event.reason,
"software-custody provider operation",
),
}
if let Some(audit) = &self.audit {
audit.append_value(&event.to_json_value());
}
}
pub fn record_reload(
&self,
previous_generation: u64,
new_generation: u64,
outcome: &str,
reason: &str,
actor: ReloadActor,
) {
match outcome {
"applied" => tracing::info!(
event = "basil.audit.reload",
previous_generation,
generation = new_generation,
outcome,
reason,
"catalog/policy reload applied",
),
"checked" => tracing::info!(
event = "basil.audit.reload",
previous_generation,
generation = new_generation,
outcome,
reason,
"catalog/policy reload dry-run validated; no swap",
),
_ => tracing::warn!(
event = "basil.audit.reload",
previous_generation,
generation = new_generation,
outcome,
reason,
"catalog/policy reload rejected; previous generation still serving",
),
}
if let Some(audit) = &self.audit {
audit.append_reload(previous_generation, new_generation, outcome, reason, actor);
}
}
#[must_use]
pub const fn manager(&self) -> &BackendManager {
&self.manager
}
#[must_use]
pub fn backend_label(&self) -> &str {
&self.backend_label
}
#[must_use]
pub fn agent_version(&self) -> &str {
&self.agent_version
}
#[must_use]
pub const fn limits(&self) -> BrokerLimits {
self.limits
}
#[must_use]
pub const fn events(&self) -> &EventSource {
&self.events
}
#[must_use]
pub const fn jwt_revocations(&self) -> &JwtRevocationStore {
&self.jwt_revocations
}
pub async fn refresh_jwt_revocations(&self) -> Result<(), ManagerError> {
self.jwt_revocations
.refresh_from_manager(&self.manager)
.await
}
pub async fn revoke_jwt_svid(
&self,
trust_domain: &str,
jti: &str,
expires_at_unix: u64,
) -> Result<(), crate::manager::ManagerError> {
self.jwt_revocations
.insert(trust_domain, jti, expires_at_unix)?;
self.jwt_revocations.persist(&self.manager).await?;
self.events.revoked(trust_domain, jti);
Ok(())
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use arc_swap::ArcSwap;
use super::{Generation, INITIAL_GENERATION_ID};
use crate::catalog::Catalog;
use crate::catalog::policy::{Config, Op, ResolvedPolicy};
fn empty_catalog() -> Catalog {
Catalog {
schema_version: 1,
backends: BTreeMap::new(),
keys: BTreeMap::new(),
}
}
fn generation(id: u64) -> Generation {
Generation::new(
id,
empty_catalog(),
ResolvedPolicy::default(),
Config::default(),
)
}
#[test]
fn initial_generation_id_is_one() {
assert_eq!(INITIAL_GENERATION_ID, 1);
assert_eq!(generation(INITIAL_GENERATION_ID).id(), 1);
}
#[test]
fn one_snapshot_is_internally_coherent() {
let swap = ArcSwap::from_pointee(generation(7));
let pinned = swap.load();
assert_eq!(pinned.id(), 7);
assert!(
pinned
.pdp()
.explain_subject("svc.missing", Op::Get, "any.key")
.decision
.is_deny()
);
assert!(pinned.config().names.users.is_empty());
}
#[test]
fn repeated_loads_without_swap_see_same_generation() {
let swap = ArcSwap::from_pointee(generation(INITIAL_GENERATION_ID));
let first = swap.load().id();
let second = swap.load().id();
assert_eq!(first, second);
assert_eq!(first, INITIAL_GENERATION_ID);
}
#[test]
fn pinned_guard_survives_a_later_swap() {
let swap = ArcSwap::from_pointee(generation(1));
let pinned = swap.load();
assert_eq!(pinned.id(), 1);
swap.store(std::sync::Arc::new(generation(2)));
assert_eq!(pinned.id(), 1);
assert_eq!(swap.load().id(), 2);
}
}