⎇ d3649bfd1a80153652c2a028315369bb4b184024
crates/core/brahman-handshake/src/peer_policy.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
feat(brahman-net+handshake): swarm-level deny via libp2p block_list
1//! Política de admisión de peers libp2p: allowlist + denylist con hot
2//! reload opcional.
3//!
4//! Capa de política sobre el trust criptográfico de Fase 3. Combina:
5//!
6//! - **Denylist**: peers explícitamente baneados. Si está, deny gana.
7//! - **Allowlist**: si está set, sólo los peers listados pasan.
8//! Si no está set, modo abierto (todo peer Ed25519-válido pasa,
9//! sujeto sólo a denylist).
10//!
11//! Sin denylist y sin allowlist → modo totalmente abierto (compat
12//! con todo lo anterior). Con allowlist y denylist a la vez, el
13//! orden de evaluación es: deny first → allow check → admit.
14//!
15//! Aplica únicamente al path libp2p — el path Unix usa SO_PEERCRED
16//! del kernel para autenticación local, no PeerId.
17//!
18//! ## Hot reload
19//!
20//! Si la política se construyó con [`PeerPolicy::watch_files`], un
21//! thread dedicado vigila los archivos de allow/deny vía `notify`.
22//! Cualquier cambio (write, create, modify, remove) dispara una
23//! recarga atómica con debounce de 250ms (los editores típicos
24//! producen varios eventos por save).
25//!
26//! Errores de reload (parse fallido, archivo eliminado) se loggean
27//! pero NO bajan la política existente — aceptamos la versión
28//! anterior hasta que el archivo vuelva a parsearse limpio. Esto
29//! evita que un error de tipeo deje al Init en modo inseguro.
30//!
31//! ## Formato del archivo
32//!
33//! Idéntico para allow y deny: PeerId base58 por línea, `#` para
34//! comentarios (línea entera o inline), líneas vacías ignoradas.
35
36use std::collections::BTreeSet;
37use std::path::{Path, PathBuf};
38use std::sync::{Arc, RwLock};
39use std::time::{Duration, Instant};
40
41use brahman_net::{BrahmanNet, PeerId};
42use tracing::{debug, info, warn};
43
44/// Política de admisión combinada (allow + deny). Clone barato (todos
45/// los campos son Arc o referencias inmutables).
46#[derive(Clone)]
47pub struct PeerPolicy {
48 inner: Arc<RwLock<PolicyInner>>,
49 paths: Arc<PolicyPaths>,
50 /// `BrahmanNet` opcional asociado vía [`Self::attach_to_net`].
51 /// Si está set, cada cambio en la denylist se sincroniza con el
52 /// `block_list` behaviour del swarm — los peers baneados son
53 /// rechazados ANTES del Noise handshake. `RwLock<Option<...>>`
54 /// para que `attach_to_net` se pueda llamar después del
55 /// constructor (típico en ente-zero: primero arma la policy,
56 /// después el net, después attach).
57 net: Arc<RwLock<Option<Arc<BrahmanNet>>>>,
58}
59
60#[derive(Default)]
61struct PolicyInner {
62 /// `Some(set)`: sólo peers en el set pasan. `None`: modo abierto.
63 allow: Option<BTreeSet<PeerId>>,
64 /// Peers baneados. Vacío = sin denylist.
65 deny: BTreeSet<PeerId>,
66}
67
68#[derive(Default)]
69struct PolicyPaths {
70 allow_path: Option<PathBuf>,
71 deny_path: Option<PathBuf>,
72}
73
74/// Decisión del gate de política para un peer dado.
75#[derive(Debug, Clone, Copy, PartialEq, Eq)]
76pub enum Decision {
77 /// El peer es admitido (no está en deny y, si hay allow, está en allow).
78 Admit,
79 /// El peer está explícitamente en la denylist.
80 DeniedByDenylist,
81 /// Hay allowlist configurada y el peer no está en ella.
82 NotInAllowlist,
83}
84
85impl Decision {
86 pub fn is_admitted(self) -> bool {
87 matches!(self, Decision::Admit)
88 }
89
90 pub fn reason(self) -> &'static str {
91 match self {
92 Decision::Admit => "admit",
93 Decision::DeniedByDenylist => "explicitly denied",
94 Decision::NotInAllowlist => "not in allowlist",
95 }
96 }
97}
98
99#[derive(Debug, thiserror::Error)]
100pub enum PolicyError {
101 #[error("leer política en {path}: {source}")]
102 Io {
103 path: PathBuf,
104 #[source]
105 source: std::io::Error,
106 },
107 #[error("línea {line_no} de {path}: PeerId inválido '{value}'")]
108 InvalidPeerId {
109 path: PathBuf,
110 line_no: usize,
111 value: String,
112 },
113}
114
115impl PeerPolicy {
116 /// Política totalmente abierta: todo peer pasa. Útil como default
117 /// cuando no hay archivos configurados.
118 pub fn open() -> Self {
119 Self {
120 inner: Arc::new(RwLock::new(PolicyInner::default())),
121 paths: Arc::new(PolicyPaths::default()),
122 net: Arc::new(RwLock::new(None)),
123 }
124 }
125
126 /// Construye una política inline con sets explícitos. Sin paths
127 /// asociados, así que `reload` y `watch_files` son no-ops.
128 pub fn from_sets(allow: Option<BTreeSet<PeerId>>, deny: BTreeSet<PeerId>) -> Self {
129 Self {
130 inner: Arc::new(RwLock::new(PolicyInner { allow, deny })),
131 paths: Arc::new(PolicyPaths::default()),
132 net: Arc::new(RwLock::new(None)),
133 }
134 }
135
136 /// Carga política desde archivos. Cada path es opcional: `None`
137 /// significa "esa lista no aplica" (allow=None ⇒ modo abierto;
138 /// deny=None ⇒ sin baneados). Asocia los paths internamente para
139 /// que `reload` y `watch_files` los re-lean.
140 pub fn from_files(
141 allow_path: Option<&Path>,
142 deny_path: Option<&Path>,
143 ) -> Result<Self, PolicyError> {
144 let allow = match allow_path {
145 Some(p) => Some(parse_peer_set(p)?),
146 None => None,
147 };
148 let deny = match deny_path {
149 Some(p) => parse_peer_set(p)?,
150 None => BTreeSet::new(),
151 };
152 Ok(Self {
153 inner: Arc::new(RwLock::new(PolicyInner { allow, deny })),
154 paths: Arc::new(PolicyPaths {
155 allow_path: allow_path.map(Path::to_path_buf),
156 deny_path: deny_path.map(Path::to_path_buf),
157 }),
158 net: Arc::new(RwLock::new(None)),
159 })
160 }
161
162 /// Evalúa si `peer` puede registrarse. Toma read lock — barato,
163 /// concurrente, sin awaits.
164 pub fn evaluate(&self, peer: &PeerId) -> Decision {
165 let inner = match self.inner.read() {
166 Ok(g) => g,
167 Err(_) => {
168 // Lock envenenado: degrada a "deny por seguridad".
169 warn!("policy lock envenenado — deny por defecto");
170 return Decision::DeniedByDenylist;
171 }
172 };
173 if inner.deny.contains(peer) {
174 return Decision::DeniedByDenylist;
175 }
176 if let Some(allow) = &inner.allow {
177 if !allow.contains(peer) {
178 return Decision::NotInAllowlist;
179 }
180 }
181 Decision::Admit
182 }
183
184 /// Tamaño actual de cada lista, para logging. Tupla `(allow_count,
185 /// deny_count)`. `allow_count = None` significa "modo abierto"
186 /// (sin allowlist).
187 pub fn sizes(&self) -> (Option<usize>, usize) {
188 match self.inner.read() {
189 Ok(g) => (g.allow.as_ref().map(|s| s.len()), g.deny.len()),
190 Err(_) => (Some(0), 0),
191 }
192 }
193
194 /// Recarga atómica desde los paths asociados. Si un archivo
195 /// falla, la versión anterior persiste y el error se devuelve.
196 /// Esto evita que un typo en el archivo deje al Init en modo
197 /// inseguro.
198 ///
199 /// Si hay un `BrahmanNet` attached vía [`Self::attach_to_net`],
200 /// el cambio de denylist se sincroniza con el `block_list` del
201 /// swarm: se calcula el diff (added/removed) y se aplican
202 /// `block_peer`/`unblock_peer` por cada cambio.
203 pub fn reload(&self) -> Result<(), PolicyError> {
204 let new_allow = match &self.paths.allow_path {
205 Some(p) => Some(parse_peer_set(p)?),
206 None => None,
207 };
208 let new_deny = match &self.paths.deny_path {
209 Some(p) => parse_peer_set(p)?,
210 None => BTreeSet::new(),
211 };
212 // Snapshot de la deny actual ANTES de mutar, para diff.
213 let prev_deny = self
214 .inner
215 .read()
216 .map(|g| g.deny.clone())
217 .unwrap_or_default();
218 if let Ok(mut inner) = self.inner.write() {
219 inner.allow = new_allow;
220 inner.deny = new_deny.clone();
221 }
222 self.sync_deny_to_swarm(&prev_deny, &new_deny);
223 Ok(())
224 }
225
226 /// Asocia esta política a un `BrahmanNet`. Sincroniza el snapshot
227 /// actual de la denylist con el `block_list` behaviour del swarm
228 /// (cada peer baneado se rechaza ANTES del Noise handshake), y
229 /// registra el net para re-sincronizarse en cada [`Self::reload`].
230 ///
231 /// Si ya había un net attached, se reemplaza (caso esperado:
232 /// un Init no debería tener dos `BrahmanNet`s).
233 pub fn attach_to_net(&self, net: Arc<BrahmanNet>) {
234 // Sync inicial: bloquear todos los peers actualmente en deny.
235 if let Ok(inner) = self.inner.read() {
236 for peer in &inner.deny {
237 net.block_peer(*peer);
238 }
239 }
240 if let Ok(mut slot) = self.net.write() {
241 *slot = Some(net);
242 }
243 }
244
245 /// Calcula el diff entre `prev` y `new` y aplica
246 /// `block_peer`/`unblock_peer` al net asociado (si hay).
247 /// No-op si no hay net attached.
248 fn sync_deny_to_swarm(&self, prev: &BTreeSet<PeerId>, new: &BTreeSet<PeerId>) {
249 let net = match self.net.read() {
250 Ok(g) => match g.as_ref() {
251 Some(n) => n.clone(),
252 None => return,
253 },
254 Err(_) => return,
255 };
256 for added in new.difference(prev) {
257 net.block_peer(*added);
258 }
259 for removed in prev.difference(new) {
260 net.unblock_peer(*removed);
261 }
262 }
263
264 /// Arranca un thread que vigila los archivos asociados con
265 /// `notify` y llama [`Self::reload`] cuando cambian. Debounce
266 /// 250ms para coalescer múltiples eventos por save (los editores
267 /// hacen Create+Modify+más).
268 ///
269 /// Devuelve un `JoinHandle` que el caller debe mantener vivo.
270 /// Drop del handle no detiene el thread (notify watcher es
271 /// sticky); para detener, terminar el proceso.
272 ///
273 /// No-op si no hay paths asociados (devuelve un handle dummy
274 /// que termina inmediatamente).
275 pub fn spawn_watcher(&self) -> std::io::Result<std::thread::JoinHandle<()>> {
276 let allow_path = self.paths.allow_path.clone();
277 let deny_path = self.paths.deny_path.clone();
278 let policy = self.clone();
279
280 if allow_path.is_none() && deny_path.is_none() {
281 // Sin archivos a vigilar: spawn un thread que termina ya.
282 return std::thread::Builder::new()
283 .name("brahman-policy-watcher-noop".into())
284 .spawn(|| {});
285 }
286
287 std::thread::Builder::new()
288 .name("brahman-policy-watcher".into())
289 .spawn(move || {
290 run_watcher(policy, allow_path, deny_path);
291 })
292 }
293}
294
295impl std::fmt::Debug for PeerPolicy {
296 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
297 let (allow, deny) = self.sizes();
298 f.debug_struct("PeerPolicy")
299 .field("allow", &allow)
300 .field("deny", &deny)
301 .field("allow_path", &self.paths.allow_path)
302 .field("deny_path", &self.paths.deny_path)
303 .finish()
304 }
305}
306
307fn parse_peer_set(path: &Path) -> Result<BTreeSet<PeerId>, PolicyError> {
308 let contents = std::fs::read_to_string(path).map_err(|e| PolicyError::Io {
309 path: path.to_path_buf(),
310 source: e,
311 })?;
312 let mut out = BTreeSet::new();
313 for (idx, raw) in contents.lines().enumerate() {
314 let line_no = idx + 1;
315 let trimmed = raw.split('#').next().unwrap_or("").trim();
316 if trimmed.is_empty() {
317 continue;
318 }
319 let peer = trimmed
320 .parse::<PeerId>()
321 .map_err(|_| PolicyError::InvalidPeerId {
322 path: path.to_path_buf(),
323 line_no,
324 value: trimmed.to_string(),
325 })?;
326 out.insert(peer);
327 }
328 Ok(out)
329}
330
331const DEBOUNCE_MS: u64 = 250;
332
333fn run_watcher(
334 policy: PeerPolicy,
335 allow_path: Option<PathBuf>,
336 deny_path: Option<PathBuf>,
337) {
338 use notify::{RecursiveMode, Watcher};
339
340 let (tx, rx) = std::sync::mpsc::channel::<notify::Result<notify::Event>>();
341 let mut watcher = match notify::recommended_watcher(move |res| {
342 let _ = tx.send(res);
343 }) {
344 Ok(w) => w,
345 Err(e) => {
346 warn!(?e, "notify watcher para policy no se pudo crear — hot reload deshabilitado");
347 return;
348 }
349 };
350
351 // Vigilamos los DIRECTORIOS de los archivos, no los archivos
352 // directos. Los editores típicos hacen rename-and-replace (escriben
353 // a tmp, rename al destino), lo que rompe el watch del archivo
354 // pero NO el del directorio. Trade-off: recibimos más eventos
355 // (cualquier archivo del dir), filtramos por path al procesar.
356 for p in [&allow_path, &deny_path].iter().filter_map(|x| x.as_ref()) {
357 if let Some(parent) = p.parent() {
358 if let Err(e) = watcher.watch(parent, RecursiveMode::NonRecursive) {
359 warn!(path = %parent.display(), ?e, "watch failed");
360 }
361 }
362 }
363
364 let debounce = Duration::from_millis(DEBOUNCE_MS);
365 let mut pending_at: Option<Instant> = None;
366
367 loop {
368 let timeout = match pending_at {
369 Some(at) => debounce.saturating_sub(at.elapsed()).max(Duration::from_millis(10)),
370 None => Duration::from_secs(60), // wakeup periódico, no esencial
371 };
372
373 match rx.recv_timeout(timeout) {
374 Ok(Ok(event)) => {
375 // Sólo nos interesan eventos sobre los paths exactos.
376 let touches_us = event.paths.iter().any(|p| {
377 Some(p) == allow_path.as_ref() || Some(p) == deny_path.as_ref()
378 });
379 if !touches_us {
380 continue;
381 }
382 debug!(?event.kind, "policy file event recibido — debounce");
383 pending_at = Some(Instant::now());
384 }
385 Ok(Err(e)) => {
386 warn!(?e, "notify error en policy watcher");
387 }
388 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
389 if let Some(at) = pending_at {
390 if at.elapsed() >= debounce {
391 match policy.reload() {
392 Ok(()) => {
393 let (a, d) = policy.sizes();
394 info!(
395 allow = ?a,
396 deny = d,
397 "policy hot-reload completo"
398 );
399 }
400 Err(e) => {
401 warn!(?e, "policy hot-reload falló — manteniendo versión anterior");
402 }
403 }
404 pending_at = None;
405 }
406 }
407 }
408 Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
409 warn!("policy watcher channel cerrado — terminando thread");
410 return;
411 }
412 }
413 }
414}
415
416#[cfg(test)]
417mod tests {
418 use super::*;
419 use brahman_net::Keypair;
420 use tempfile::TempDir;
421
422 fn fresh_peer() -> PeerId {
423 Keypair::generate_ed25519().public().to_peer_id()
424 }
425
426 #[test]
427 fn open_admits_anyone() {
428 let p = PeerPolicy::open();
429 assert_eq!(p.evaluate(&fresh_peer()), Decision::Admit);
430 }
431
432 #[test]
433 fn allow_only_admits_listed() {
434 let p1 = fresh_peer();
435 let p2 = fresh_peer();
436 let policy = PeerPolicy::from_sets(
437 Some([p1].into_iter().collect()),
438 BTreeSet::new(),
439 );
440 assert_eq!(policy.evaluate(&p1), Decision::Admit);
441 assert_eq!(policy.evaluate(&p2), Decision::NotInAllowlist);
442 }
443
444 #[test]
445 fn deny_overrides_open() {
446 let p1 = fresh_peer();
447 let p2 = fresh_peer();
448 let policy = PeerPolicy::from_sets(None, [p1].into_iter().collect());
449 assert_eq!(policy.evaluate(&p1), Decision::DeniedByDenylist);
450 assert_eq!(policy.evaluate(&p2), Decision::Admit);
451 }
452
453 #[test]
454 fn deny_overrides_allow() {
455 // Conflicto explícito: p1 está en ambas. Deny gana.
456 let p1 = fresh_peer();
457 let policy = PeerPolicy::from_sets(
458 Some([p1].into_iter().collect()),
459 [p1].into_iter().collect(),
460 );
461 assert_eq!(policy.evaluate(&p1), Decision::DeniedByDenylist);
462 }
463
464 #[test]
465 fn from_files_with_both_lists() {
466 let p1 = fresh_peer();
467 let p2 = fresh_peer();
468 let p3 = fresh_peer();
469 let tmp = TempDir::new().unwrap();
470 let allow = tmp.path().join("allow.txt");
471 let deny = tmp.path().join("deny.txt");
472 std::fs::write(&allow, format!("{}\n{}\n", p1, p2)).unwrap();
473 std::fs::write(&deny, format!("# baneado\n{}\n", p2)).unwrap();
474 let policy = PeerPolicy::from_files(Some(&allow), Some(&deny)).unwrap();
475 assert_eq!(policy.evaluate(&p1), Decision::Admit);
476 assert_eq!(policy.evaluate(&p2), Decision::DeniedByDenylist); // deny gana
477 assert_eq!(policy.evaluate(&p3), Decision::NotInAllowlist);
478 }
479
480 #[test]
481 fn from_files_only_deny() {
482 let p1 = fresh_peer();
483 let p2 = fresh_peer();
484 let tmp = TempDir::new().unwrap();
485 let deny = tmp.path().join("deny.txt");
486 std::fs::write(&deny, format!("{}\n", p1)).unwrap();
487 let policy = PeerPolicy::from_files(None, Some(&deny)).unwrap();
488 assert_eq!(policy.evaluate(&p1), Decision::DeniedByDenylist);
489 assert_eq!(policy.evaluate(&p2), Decision::Admit);
490 }
491
492 #[test]
493 fn reload_picks_up_changes() {
494 let p1 = fresh_peer();
495 let p2 = fresh_peer();
496 let tmp = TempDir::new().unwrap();
497 let allow = tmp.path().join("allow.txt");
498 std::fs::write(&allow, format!("{}\n", p1)).unwrap();
499
500 let policy = PeerPolicy::from_files(Some(&allow), None).unwrap();
501 assert_eq!(policy.evaluate(&p1), Decision::Admit);
502 assert_eq!(policy.evaluate(&p2), Decision::NotInAllowlist);
503
504 // Mutar el archivo: ahora p2 está, p1 no.
505 std::fs::write(&allow, format!("{}\n", p2)).unwrap();
506 policy.reload().unwrap();
507 assert_eq!(policy.evaluate(&p1), Decision::NotInAllowlist);
508 assert_eq!(policy.evaluate(&p2), Decision::Admit);
509 }
510
511 #[test]
512 fn reload_failure_preserves_previous_state() {
513 let p1 = fresh_peer();
514 let tmp = TempDir::new().unwrap();
515 let allow = tmp.path().join("allow.txt");
516 std::fs::write(&allow, format!("{}\n", p1)).unwrap();
517 let policy = PeerPolicy::from_files(Some(&allow), None).unwrap();
518 assert_eq!(policy.evaluate(&p1), Decision::Admit);
519
520 // Romper el archivo con basura.
521 std::fs::write(&allow, "this-is-not-a-peer-id\n").unwrap();
522 let err = policy.reload();
523 assert!(err.is_err(), "reload con typo debe fallar");
524
525 // Estado anterior se mantiene.
526 assert_eq!(
527 policy.evaluate(&p1),
528 Decision::Admit,
529 "policy debe conservar la versión anterior tras fallo de reload"
530 );
531 }
532
533 #[test]
534 fn invalid_file_rejected_at_load() {
535 let tmp = TempDir::new().unwrap();
536 let path = tmp.path().join("bad.txt");
537 std::fs::write(&path, "not-a-peer-id\n").unwrap();
538 let err = PeerPolicy::from_files(Some(&path), None).unwrap_err();
539 assert!(matches!(err, PolicyError::InvalidPeerId { .. }));
540 }
541
542 #[test]
543 fn watcher_reloads_on_file_change() {
544 // Test integración del watcher: arma policy con file, spawn
545 // watcher, modifica el archivo, espera el debounce, verifica
546 // que la policy refleja el cambio.
547 let p1 = fresh_peer();
548 let p2 = fresh_peer();
549 let tmp = TempDir::new().unwrap();
550 let allow = tmp.path().join("allow.txt");
551 std::fs::write(&allow, format!("{}\n", p1)).unwrap();
552
553 let policy = PeerPolicy::from_files(Some(&allow), None).unwrap();
554 let _watcher = policy.spawn_watcher().unwrap();
555
556 // Le damos un instante al watcher para subscribirse al dir.
557 std::thread::sleep(Duration::from_millis(100));
558
559 // Mutamos el archivo: p2 reemplaza a p1.
560 std::fs::write(&allow, format!("{}\n", p2)).unwrap();
561
562 // Esperamos > debounce + margen.
563 let deadline = Instant::now() + Duration::from_secs(3);
564 while Instant::now() < deadline {
565 if policy.evaluate(&p2) == Decision::Admit {
566 break;
567 }
568 std::thread::sleep(Duration::from_millis(50));
569 }
570
571 assert_eq!(
572 policy.evaluate(&p2),
573 Decision::Admit,
574 "watcher debería haber recargado la policy"
575 );
576 assert_eq!(
577 policy.evaluate(&p1),
578 Decision::NotInAllowlist,
579 "p1 debería haber salido tras el reload"
580 );
581 }
582}