1use std::collections::{BTreeMap, BTreeSet};
46use std::path::Path;
47use std::sync::Arc;
48
49use crate::catalog::loader::LoadError;
50use crate::catalog::schema::{BackendKind, Capability, Class, Engine, KeyAlgorithm};
51use crate::catalog::{Catalog, Config, ResolvedPolicy, load};
52use crate::state::{BrokerState, Generation};
53
54#[derive(Debug, Clone)]
60pub struct ReloadInputs {
61 pub catalog_path: std::path::PathBuf,
63 pub policy_path: std::path::PathBuf,
65}
66
67#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72pub struct ReloadOutcome {
73 pub previous_generation: u64,
75 pub new_generation: u64,
77 pub key_count: usize,
79 pub grant_count: usize,
81}
82
83#[derive(Debug, thiserror::Error)]
86pub enum ReloadError {
87 #[error("reading catalog from {path}: {source}")]
89 ReadCatalog {
90 path: String,
92 source: std::io::Error,
94 },
95
96 #[error("reading policy from {path}: {source}")]
98 ReadPolicy {
99 path: String,
101 source: std::io::Error,
103 },
104
105 #[error("catalog/policy reload input changed while reading {path}; retry reload")]
109 TornSnapshot {
110 path: String,
112 },
113
114 #[error("validating reloaded catalog/policy: {0}")]
117 Validate(#[from] LoadError),
118
119 #[error("reload touches a restart-only routing dimension: {0}")]
124 RoutingShapeChanged(String),
125
126 #[error("reload unavailable: broker has no configured catalog/policy paths")]
130 NoInputs,
131}
132
133impl ReloadError {
134 #[must_use]
136 pub const fn audit_reason(&self) -> &'static str {
137 match self {
138 Self::ReadCatalog { .. } => "catalog_read_failed",
139 Self::ReadPolicy { .. } => "policy_read_failed",
140 Self::TornSnapshot { .. } => "inputs_changed_during_read",
141 Self::Validate(_) => "validation_failed",
142 Self::RoutingShapeChanged(_) => "routing_shape_changed",
143 Self::NoInputs => "no_reload_inputs",
144 }
145 }
146}
147
148#[derive(Debug, PartialEq, Eq)]
153struct BackendShape {
154 kind: BackendKind,
155 addr: String,
156 engines: Vec<Engine>,
157 capabilities: Vec<Capability>,
158 requires: Vec<Capability>,
159}
160
161#[derive(Debug, PartialEq, Eq)]
166struct KeyShape {
167 class: Class,
168 key_type: Option<KeyAlgorithm>,
169 backend: String,
170 engine: Option<Engine>,
171 path: String,
172 public_path: Option<String>,
173}
174
175fn routing_shape(
180 catalog: &Catalog,
181) -> (BTreeMap<String, BackendShape>, BTreeMap<String, KeyShape>) {
182 let backends = catalog
183 .backends
184 .iter()
185 .map(|(name, b)| {
186 (
187 name.clone(),
188 BackendShape {
189 kind: b.kind,
190 addr: b.addr.clone(),
191 engines: b.engines.clone(),
192 capabilities: b.capabilities.clone(),
193 requires: b.requires.clone(),
194 },
195 )
196 })
197 .collect();
198 let keys = catalog
199 .keys
200 .iter()
201 .map(|(name, k)| {
202 (
203 name.clone(),
204 KeyShape {
205 class: k.class,
206 key_type: k.key_type,
207 backend: k.backend.clone(),
208 engine: k.engine,
209 path: k.path.clone(),
210 public_path: k.public_path.clone(),
211 },
212 )
213 })
214 .collect();
215 (backends, keys)
216}
217
218fn ensure_reloadable(current: &Catalog, candidate: &Catalog) -> Result<(), ReloadError> {
225 let (cur_backends, cur_keys) = routing_shape(current);
226 let (new_backends, new_keys) = routing_shape(candidate);
227 if cur_backends != new_backends {
228 return Err(ReloadError::RoutingShapeChanged(
229 "the backend set or a backend's kind/addr/engines/capabilities/requires changed"
230 .to_string(),
231 ));
232 }
233 if cur_keys != new_keys {
234 return Err(ReloadError::RoutingShapeChanged(
235 "a key was added/removed or a key's class/backend/path/engine/key_type/public_path changed"
236 .to_string(),
237 ));
238 }
239 Ok(())
240}
241
242fn spiffe_bundle_publishers(catalog: &Catalog) -> BTreeMap<String, (String, String)> {
243 catalog
244 .keys
245 .iter()
246 .filter_map(|(name, entry)| {
247 let svid_kind = entry.labels.get("svid_kind")?;
248 if !matches!(svid_kind, "jwt" | "x509") {
249 return None;
250 }
251 let trust_domain = entry.labels.get("trust_domain")?;
252 Some((
253 name.clone(),
254 (svid_kind.to_string(), trust_domain.to_string()),
255 ))
256 })
257 .collect()
258}
259
260fn bundle_changed_trust_domains(current: &Catalog, candidate: &Catalog) -> Vec<String> {
261 let current_publishers = spiffe_bundle_publishers(current);
262 let candidate_publishers = spiffe_bundle_publishers(candidate);
263 if current_publishers == candidate_publishers {
264 return Vec::new();
265 }
266
267 current_publishers
268 .values()
269 .chain(candidate_publishers.values())
270 .map(|(_, trust_domain)| trust_domain.clone())
271 .collect::<BTreeSet<_>>()
272 .into_iter()
273 .collect()
274}
275
276#[cfg(unix)]
277#[derive(Debug, Clone, Copy, PartialEq, Eq)]
278struct FileFingerprint {
279 dev: u64,
280 ino: u64,
281 len: u64,
282 mtime_sec: i64,
283 mtime_nsec: i64,
284 ctime_sec: i64,
285 ctime_nsec: i64,
286}
287
288#[cfg(unix)]
289fn file_fingerprint(path: &Path) -> std::io::Result<FileFingerprint> {
290 use std::os::unix::fs::MetadataExt;
291
292 let metadata = std::fs::metadata(path)?;
293 Ok(FileFingerprint {
294 dev: metadata.dev(),
295 ino: metadata.ino(),
296 len: metadata.len(),
297 mtime_sec: metadata.mtime(),
298 mtime_nsec: metadata.mtime_nsec(),
299 ctime_sec: metadata.ctime(),
300 ctime_nsec: metadata.ctime_nsec(),
301 })
302}
303
304#[cfg(not(unix))]
305#[derive(Debug, Clone, PartialEq, Eq)]
306struct FileFingerprint {
307 len: u64,
308 modified: Option<std::time::SystemTime>,
309}
310
311#[cfg(not(unix))]
312fn file_fingerprint(path: &Path) -> std::io::Result<FileFingerprint> {
313 let metadata = std::fs::metadata(path)?;
314 Ok(FileFingerprint {
315 len: metadata.len(),
316 modified: metadata.modified().ok(),
317 })
318}
319
320fn read_reload_inputs_with_observer(
321 inputs: &ReloadInputs,
322 observer: impl FnOnce(),
323) -> Result<(String, String), ReloadError> {
324 let catalog_before =
325 file_fingerprint(&inputs.catalog_path).map_err(|source| ReloadError::ReadCatalog {
326 path: inputs.catalog_path.display().to_string(),
327 source,
328 })?;
329 let policy_before =
330 file_fingerprint(&inputs.policy_path).map_err(|source| ReloadError::ReadPolicy {
331 path: inputs.policy_path.display().to_string(),
332 source,
333 })?;
334
335 let catalog_json = std::fs::read_to_string(&inputs.catalog_path).map_err(|source| {
336 ReloadError::ReadCatalog {
337 path: inputs.catalog_path.display().to_string(),
338 source,
339 }
340 })?;
341 let policy_json =
342 std::fs::read_to_string(&inputs.policy_path).map_err(|source| ReloadError::ReadPolicy {
343 path: inputs.policy_path.display().to_string(),
344 source,
345 })?;
346
347 observer();
348
349 let catalog_after =
350 file_fingerprint(&inputs.catalog_path).map_err(|source| ReloadError::ReadCatalog {
351 path: inputs.catalog_path.display().to_string(),
352 source,
353 })?;
354 if catalog_before != catalog_after {
355 return Err(ReloadError::TornSnapshot {
356 path: inputs.catalog_path.display().to_string(),
357 });
358 }
359 let policy_after =
360 file_fingerprint(&inputs.policy_path).map_err(|source| ReloadError::ReadPolicy {
361 path: inputs.policy_path.display().to_string(),
362 source,
363 })?;
364 if policy_before != policy_after {
365 return Err(ReloadError::TornSnapshot {
366 path: inputs.policy_path.display().to_string(),
367 });
368 }
369
370 Ok((catalog_json, policy_json))
371}
372
373fn read_reload_inputs(inputs: &ReloadInputs) -> Result<(String, String), ReloadError> {
374 read_reload_inputs_with_observer(inputs, || {})
375}
376
377struct ValidatedCandidate {
382 catalog: Catalog,
383 policy: ResolvedPolicy,
384 config: Config,
385 outcome: ReloadOutcome,
386 bundle_changed_trust_domains: Vec<String>,
387}
388
389fn validate_candidate(state: &BrokerState) -> Result<ValidatedCandidate, ReloadError> {
408 let inputs = state.reload_inputs().ok_or(ReloadError::NoInputs)?;
409 let (catalog_json, policy_json) = read_reload_inputs(inputs)?;
410
411 let (catalog, policy, config, warnings) = load(&catalog_json, &policy_json)?;
414 for w in &warnings {
415 tracing::warn!(warning = %w, "reload: catalog/policy load warning");
416 }
417
418 let current = state.load_generation();
421 ensure_reloadable(current.catalog(), &catalog)?;
422
423 let previous_generation = current.id();
424 let new_generation = previous_generation.saturating_add(1);
425 let bundle_changed_trust_domains = bundle_changed_trust_domains(current.catalog(), &catalog);
426 let outcome = ReloadOutcome {
427 previous_generation,
428 new_generation,
429 key_count: catalog.keys.len(),
430 grant_count: policy.grant_count(),
431 };
432
433 Ok(ValidatedCandidate {
434 catalog,
435 policy,
436 config,
437 outcome,
438 bundle_changed_trust_domains,
439 })
440}
441
442pub fn check_reload(state: &BrokerState) -> Result<ReloadOutcome, ReloadError> {
456 validate_candidate(state).map(|c| c.outcome)
457}
458
459pub fn reload_generation(state: &BrokerState) -> Result<ReloadOutcome, ReloadError> {
477 let _reload_guard = state
483 .reload_lock()
484 .lock()
485 .unwrap_or_else(std::sync::PoisonError::into_inner);
486 let candidate = validate_candidate(state)?;
487 let ValidatedCandidate {
488 catalog,
489 policy,
490 config,
491 outcome,
492 bundle_changed_trust_domains,
493 } = candidate;
494
495 let next = Generation::new(outcome.new_generation, Arc::new(catalog), policy, config);
496 state.swap_generation(Arc::new(next));
497 for trust_domain in bundle_changed_trust_domains {
498 state.events().bundle_changed(trust_domain);
499 }
500
501 Ok(outcome)
502}
503
504#[cfg(test)]
505mod tests {
506 use std::collections::BTreeMap;
507 use std::sync::Arc;
508
509 use async_trait::async_trait;
510 use basil_proto::KeyType;
511
512 use super::{
513 ReloadError, ReloadInputs, check_reload, read_reload_inputs_with_observer,
514 reload_generation,
515 };
516 use crate::backend::{Backend, BackendError, NewKey};
517 use crate::catalog::load;
518 use crate::manager::BackendManager;
519 use crate::state::{BrokerState, INITIAL_GENERATION_ID};
520
521 struct NoopBackend;
525
526 #[async_trait]
527 impl Backend for NoopBackend {
528 fn kind(&self) -> &'static str {
529 "noop"
530 }
531 async fn new_key(&self, _key_type: KeyType) -> Result<NewKey, BackendError> {
532 Err(BackendError::Unsupported("new_key"))
533 }
534 async fn public_key(&self, _key_id: &str) -> Result<Vec<u8>, BackendError> {
535 Err(BackendError::Unsupported("public_key"))
536 }
537 async fn sign(&self, _key_id: &str, _message: &[u8]) -> Result<Vec<u8>, BackendError> {
538 Err(BackendError::Unsupported("sign"))
539 }
540 async fn verify(
541 &self,
542 _key_id: &str,
543 _message: &[u8],
544 _signature: &[u8],
545 ) -> Result<bool, BackendError> {
546 Err(BackendError::Unsupported("verify"))
547 }
548 }
549
550 fn catalog_json(writable: bool) -> String {
553 format!(
554 r#"{{
555 "schemaVersion": 1,
556 "backends": {{ "bao": {{ "kind": "vault", "addr": "http://127.0.0.1:8200" }} }},
557 "keys": {{
558 "web.signer": {{
559 "class": "asymmetric", "keyType": "ed25519", "backend": "bao",
560 "path": "signer", "writable": {writable}, "description": "a signer"
561 }}
562 }}
563 }}"#
564 )
565 }
566
567 fn catalog_json_repathed() -> String {
569 r#"{
570 "schemaVersion": 1,
571 "backends": { "bao": { "kind": "vault", "addr": "http://127.0.0.1:8200" } },
572 "keys": {
573 "web.signer": {
574 "class": "asymmetric", "keyType": "ed25519", "backend": "bao",
575 "path": "signer-v2", "writable": true, "description": "a signer"
576 }
577 }
578 }"#
579 .to_string()
580 }
581
582 fn policy_json(grant_sign: bool) -> String {
583 let rules = if grant_sign {
584 r#"[ { "id": "r1", "subjects": ["svc.web"], "action": ["op:sign"], "target": ["web.signer"] } ]"#
585 } else {
586 "[]"
587 };
588 format!(
589 r#"{{
590 "schemaVersion": 2,
591 "subjects": {{ "svc.web": {{ "allOf": [ {{ "kind": "unix", "uid": 1000 }} ] }} }},
592 "roles": {{}},
593 "rules": {rules},
594 "config": {{}}
595 }}"#
596 )
597 }
598
599 fn state_with_files(catalog: &str, policy: &str) -> (Arc<BrokerState>, ReloadInputs) {
602 let dir = std::env::temp_dir().join(format!(
603 "basil-reload-test-{}-{}",
604 std::process::id(),
605 uuid::Uuid::new_v4()
606 ));
607 std::fs::create_dir_all(&dir).expect("create temp dir");
608 let catalog_path = dir.join("catalog.json");
609 let policy_path = dir.join("policy.json");
610 std::fs::write(&catalog_path, catalog).expect("write catalog");
611 std::fs::write(&policy_path, policy).expect("write policy");
612
613 let (cat, pol, cfg, warnings) = load(catalog, policy).expect("fixture loads");
614 assert!(warnings.is_empty());
615 let mut backends: BTreeMap<String, Box<dyn Backend>> = BTreeMap::new();
616 backends.insert("bao".into(), Box::new(NoopBackend));
617 let manager = BackendManager::new(cat.clone(), backends).expect("manager builds");
618 let inputs = ReloadInputs {
619 catalog_path,
620 policy_path,
621 };
622 let state = Arc::new(
623 BrokerState::new(cat, pol, cfg, manager, "noop").with_reload_inputs(inputs.clone()),
624 );
625 (state, inputs)
626 }
627
628 fn write_files(inputs: &ReloadInputs, catalog: &str, policy: &str) {
629 std::fs::write(&inputs.catalog_path, catalog).expect("rewrite catalog");
630 std::fs::write(&inputs.policy_path, policy).expect("rewrite policy");
631 }
632
633 #[test]
638 fn valid_reload_swaps_generation_and_stays_coherent() {
639 let (state, inputs) = state_with_files(&catalog_json(false), &policy_json(false));
640 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
641
642 let pinned = state.load_generation();
644 assert_eq!(pinned.id(), INITIAL_GENERATION_ID);
645
646 write_files(&inputs, &catalog_json(true), &policy_json(true));
648 let outcome = reload_generation(&state).expect("valid reload applies");
649
650 assert_eq!(outcome.previous_generation, INITIAL_GENERATION_ID);
651 assert_eq!(outcome.new_generation, INITIAL_GENERATION_ID + 1);
652 assert_eq!(outcome.key_count, 1);
653 assert_eq!(outcome.grant_count, 1);
654
655 assert_eq!(pinned.id(), INITIAL_GENERATION_ID);
658 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID + 1);
659 }
660
661 #[test]
664 fn invalid_policy_is_rejected_and_previous_generation_keeps_serving() {
665 let (state, inputs) = state_with_files(&catalog_json(true), &policy_json(true));
666
667 write_files(
671 &inputs,
672 &catalog_json(true),
673 r#"{ "schemaVersion": 2, "subjects": { "svc.web": { "allOf": [ { "kind": "unix", "uid": 1000 } ] } }, "roles": {}, "rules": [ { "id": "bad", "subjects": ["svc.web"], "action": ["role:nonexistent"], "target": ["web.signer"] } ], "config": {} }"#,
674 );
675
676 let err = reload_generation(&state).expect_err("malformed policy rejected");
677 assert!(matches!(err, ReloadError::Validate(_)));
678 assert_eq!(err.audit_reason(), "validation_failed");
679 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
681 }
682
683 #[test]
684 fn reload_input_change_during_read_is_rejected() {
685 let (state, inputs) = state_with_files(&catalog_json(true), &policy_json(true));
686
687 let err = read_reload_inputs_with_observer(&inputs, || {
688 std::fs::write(
689 &inputs.policy_path,
690 policy_json(true).replace("\"rules\"", "\"rules_changed\""),
691 )
692 .expect("race policy rewrite");
693 })
694 .expect_err("changed policy fingerprint rejects torn read");
695
696 assert!(matches!(err, ReloadError::TornSnapshot { .. }));
697 assert_eq!(err.audit_reason(), "inputs_changed_during_read");
698 assert_eq!(
699 state.active_generation_id(),
700 INITIAL_GENERATION_ID,
701 "helper rejection leaves the serving generation untouched"
702 );
703 }
704
705 #[test]
709 fn non_profile_jwt_svid_issuer_is_rejected_on_reload() {
710 let base_catalog = r#"{
712 "schemaVersion": 1,
713 "backends": { "bao": { "kind": "vault", "addr": "http://127.0.0.1:8200" } },
714 "keys": {
715 "spiffe.jwt": {
716 "class": "asymmetric", "keyType": "rsa-2048", "backend": "bao", "path": "jwt",
717 "labels": ["svid_kind=jwt", "trust_domain=example.org"],
718 "writable": false, "description": "jwt issuer"
719 }
720 }
721 }"#;
722 let (state, inputs) = state_with_files(base_catalog, &policy_json(false));
723
724 let bad_catalog = base_catalog.replace("rsa-2048", "ed25519");
726 write_files(&inputs, &bad_catalog, &policy_json(false));
727
728 let err = reload_generation(&state).expect_err("non-profile jwt issuer rejected");
729 assert!(matches!(
733 err,
734 ReloadError::Validate(_) | ReloadError::RoutingShapeChanged(_)
735 ));
736 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
737 }
738
739 #[test]
742 fn restart_only_routing_change_is_rejected() {
743 let (state, inputs) = state_with_files(&catalog_json(true), &policy_json(true));
744 write_files(&inputs, &catalog_json_repathed(), &policy_json(true));
745
746 let err = reload_generation(&state).expect_err("repath rejected");
747 assert!(matches!(err, ReloadError::RoutingShapeChanged(_)));
748 assert_eq!(err.audit_reason(), "routing_shape_changed");
749 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
750 }
751
752 #[test]
756 fn check_reload_validates_without_swapping() {
757 let (state, inputs) = state_with_files(&catalog_json(false), &policy_json(false));
758 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
759
760 write_files(&inputs, &catalog_json(true), &policy_json(true));
761 let dry = check_reload(&state).expect("dry-run validates");
762 assert_eq!(dry.previous_generation, INITIAL_GENERATION_ID);
763 assert_eq!(dry.new_generation, INITIAL_GENERATION_ID + 1);
764 assert_eq!(dry.key_count, 1);
765 assert_eq!(dry.grant_count, 1);
766 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
768
769 let applied = reload_generation(&state).expect("real reload applies");
771 assert_eq!(applied, dry);
772 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID + 1);
773 }
774
775 #[test]
778 fn check_reload_rejects_what_real_reload_rejects() {
779 let (state, inputs) = state_with_files(&catalog_json(true), &policy_json(true));
780 write_files(&inputs, &catalog_json_repathed(), &policy_json(true));
781
782 let dry = check_reload(&state).expect_err("dry-run rejects repath");
783 assert!(matches!(dry, ReloadError::RoutingShapeChanged(_)));
784 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
785
786 let real = reload_generation(&state).expect_err("real reload rejects repath");
787 assert!(matches!(real, ReloadError::RoutingShapeChanged(_)));
788 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
789 }
790
791 #[test]
795 fn concurrent_reloads_are_serialized_with_monotonic_generations() {
796 let (state, inputs) = state_with_files(&catalog_json(false), &policy_json(false));
797 write_files(&inputs, &catalog_json(true), &policy_json(true));
798
799 let outcomes = std::thread::scope(|scope| {
800 let first = scope.spawn(|| reload_generation(&state));
803 let second = scope.spawn(|| reload_generation(&state));
804 [first, second].map(|h| h.join().expect("reload thread panicked"))
805 });
806
807 let mut transitions: Vec<(u64, u64)> = outcomes
808 .into_iter()
809 .map(|o| {
810 let o = o.expect("both concurrent reloads apply");
811 (o.previous_generation, o.new_generation)
812 })
813 .collect();
814 transitions.sort_unstable();
815 assert_eq!(
818 transitions,
819 vec![
820 (INITIAL_GENERATION_ID, INITIAL_GENERATION_ID + 1),
821 (INITIAL_GENERATION_ID + 1, INITIAL_GENERATION_ID + 2),
822 ]
823 );
824 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID + 2);
825 }
826
827 #[test]
830 fn reload_without_inputs_fails_closed() {
831 let (cat, pol, cfg, _) =
832 load(&catalog_json(true), &policy_json(true)).expect("fixture loads");
833 let mut backends: BTreeMap<String, Box<dyn Backend>> = BTreeMap::new();
834 backends.insert("bao".into(), Box::new(NoopBackend));
835 let manager = BackendManager::new(cat.clone(), backends).expect("manager builds");
836 let state = BrokerState::new(cat, pol, cfg, manager, "noop");
837
838 let err = reload_generation(&state).expect_err("no inputs → fail closed");
839 assert!(matches!(err, ReloadError::NoInputs));
840 assert_eq!(state.active_generation_id(), INITIAL_GENERATION_ID);
841 }
842}