feat(brahman-handshake+ente-zero): allowlist explicita de peers libp2p

Capa de politica sobre el trust criptografico de Fase 3. Hasta ahora
cualquier peer Ed25519-valido pasaba el handshake remoto; con
allowlist activa, solo los peers explicitamente listados. Aplica
unicamente al path libp2p — el path Unix sigue usando SO_PEERCRED
del kernel.

API nueva en brahman_handshake::peer_allowlist:
- PeerAllowlist::from_iter / from_file con AllowlistError tipado.
- Formato del archivo: PeerId base58 por linea, # comentarios (linea
  entera o inline), lineas vacias ignoradas. Errores de parseo
  reportan numero de linea.
- is_allowed, len, is_empty, iter.

Wire en el server:
- ServerConfig.allowlist: Option<PeerAllowlist>. None = modo abierto
  (compat). Some = solo los listados.
- Gate en do_handshake ANTES de la verificacion de firma — la
  comparacion BTreeSet O(log n) es mas barata que crypto, asi que
  rechazamos peers invalidos antes de gastar ciclos.
- HandshakeError::Unauthorized("peer X no esta en la allowlist").

Wire en Arje (ente-zero):
- Env var BRAHMAN_PEER_ALLOWLIST apuntando a un archivo.
- setup_brahman_allowlist carga al startup; degrada a None si el
  archivo falla (doctrina PID 1: no romper por subsistemas
  opcionales).

Activacion end-to-end:
  BRAHMAN_LISTEN_MULTIADDR=/ip4/0.0.0.0/tcp/4101 \\
  BRAHMAN_PEER_ALLOWLIST=/etc/brahman/allowlist.txt \\
  ente-zero

Tests: 6 unit en peer_allowlist + 1 E2E en network_libp2p
(libp2p_handshake_allowlist_admits_listed_rejects_others). 25 tests
verdes en brahman-handshake. Sin regresion en ente-zero.

Pendientes: denylist explicita, hot reload via SIGHUP/watch, aplicar
politica a nivel de swarm via libp2p_allow_block_list::Behaviour
para rechazar ANTES del Noise handshake.

crates/core/brahman-handshake/src/lib.rs

@@ -22,6 +22,7 @@ pub mod messages;
 pub mod server;
 pub mod client;
 pub mod network;
+pub mod peer_allowlist;
 pub mod signature;
 pub mod transport;
 

crates/core/brahman-handshake/src/peer_allowlist.rs

@@ -0,0 +1,198 @@
+//! Allowlist explícita de `PeerId`s para el trust gate remoto.
+//!
+//! Capa de política sobre el trust criptográfico de Fase 3
+//! (firma Ed25519 anclada al peer libp2p). Hoy cualquier peer con
+//! keypair Ed25519 válida pasa el handshake; con allowlist activa,
+//! sólo los peers explícitamente listados.
+//!
+//! Aplica únicamente al path libp2p — el path Unix sigue usando
+//! `SO_PEERCRED` del kernel para autenticación local. Sin allowlist
+//! configurada, comportamiento abierto (compatible con todo lo
+//! anterior).
+//!
+//! ## Formato del archivo
+//!
+//! Texto plano, una entrada por línea:
+//! - `PeerId` en formato base58 (canónico de libp2p, lo que muestra
+//!   `peer_id.to_string()` y lo que aparece en multiaddrs `/p2p/...`).
+//! - Líneas vacías y líneas que empiezan con `#` se ignoran.
+//!
+//! Ejemplo:
+//!
+//! ```text
+//! # allowlist brahman para máquina prod-eu-1
+//! 12D3KooWFooBarBazFooBarBazFooBarBazFooBarBazFooBarBaz
+//! # operador secundario
+//! 12D3KooWQuxQuxQuxQuxQuxQuxQuxQuxQuxQuxQuxQuxQuxQuxQux
+//! ```
+
+use std::collections::BTreeSet;
+use std::path::Path;
+
+use brahman_net::PeerId;
+
+/// Política de admisión por `PeerId` para conexiones libp2p.
+///
+/// **Nota**: la allowlist sólo se evalúa cuando el server tiene
+/// `expected_peer = Some(...)` (path libp2p). El path Unix la
+/// ignora — autenticación local va por SO_PEERCRED.
+#[derive(Debug, Clone)]
+pub struct PeerAllowlist {
+    allowed: BTreeSet<PeerId>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum AllowlistError {
+    #[error("leer allowlist en {path}: {source}")]
+    Io {
+        path: std::path::PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+    #[error("línea {line_no} de {path}: PeerId inválido '{value}'")]
+    InvalidPeerId {
+        path: std::path::PathBuf,
+        line_no: usize,
+        value: String,
+    },
+}
+
+impl PeerAllowlist {
+    /// Construye una allowlist a partir de un iterable de `PeerId`s.
+    /// Útil para tests o para listas hardcodeadas.
+    pub fn from_iter<I>(peers: I) -> Self
+    where
+        I: IntoIterator<Item = PeerId>,
+    {
+        Self {
+            allowed: peers.into_iter().collect(),
+        }
+    }
+
+    /// Carga la allowlist desde un archivo. Cada línea no-vacía y no
+    /// comentario debe ser un `PeerId` parseable. Errores incluyen
+    /// el número de línea para facilitar el debug.
+    pub fn from_file(path: impl AsRef<Path>) -> Result<Self, AllowlistError> {
+        let path = path.as_ref();
+        let contents = std::fs::read_to_string(path).map_err(|e| AllowlistError::Io {
+            path: path.to_path_buf(),
+            source: e,
+        })?;
+        let mut allowed = BTreeSet::new();
+        for (idx, raw) in contents.lines().enumerate() {
+            let line_no = idx + 1;
+            let trimmed = raw.split('#').next().unwrap_or("").trim();
+            if trimmed.is_empty() {
+                continue;
+            }
+            let peer = trimmed
+                .parse::<PeerId>()
+                .map_err(|_| AllowlistError::InvalidPeerId {
+                    path: path.to_path_buf(),
+                    line_no,
+                    value: trimmed.to_string(),
+                })?;
+            allowed.insert(peer);
+        }
+        Ok(Self { allowed })
+    }
+
+    /// Indica si `peer` está en la lista. Si la lista está vacía,
+    /// devuelve `false` (la "lista vacía" no es lo mismo que "sin
+    /// política" — para sin política, no construyas un `PeerAllowlist`
+    /// y dejá `ServerConfig.allowlist = None`).
+    pub fn is_allowed(&self, peer: &PeerId) -> bool {
+        self.allowed.contains(peer)
+    }
+
+    /// Cantidad de peers en la lista. Útil para logs.
+    pub fn len(&self) -> usize {
+        self.allowed.len()
+    }
+
+    /// `true` si no hay ningún peer en la lista.
+    pub fn is_empty(&self) -> bool {
+        self.allowed.is_empty()
+    }
+
+    /// Itera los `PeerId`s permitidos en orden determinístico.
+    pub fn iter(&self) -> impl Iterator<Item = &PeerId> {
+        self.allowed.iter()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use brahman_net::Keypair;
+    use tempfile::TempDir;
+
+    fn fresh_peer() -> PeerId {
+        Keypair::generate_ed25519().public().to_peer_id()
+    }
+
+    #[test]
+    fn from_iter_and_is_allowed() {
+        let p1 = fresh_peer();
+        let p2 = fresh_peer();
+        let p3 = fresh_peer();
+        let list = PeerAllowlist::from_iter([p1, p2]);
+        assert!(list.is_allowed(&p1));
+        assert!(list.is_allowed(&p2));
+        assert!(!list.is_allowed(&p3));
+        assert_eq!(list.len(), 2);
+    }
+
+    #[test]
+    fn from_file_parses_clean() {
+        let p1 = fresh_peer();
+        let p2 = fresh_peer();
+        let tmp = TempDir::new().unwrap();
+        let path = tmp.path().join("allow.txt");
+        let content = format!(
+            "# header comment\n\n{}\n  # indented comment ignored\n{}\n\n",
+            p1, p2
+        );
+        std::fs::write(&path, content).unwrap();
+        let list = PeerAllowlist::from_file(&path).unwrap();
+        assert_eq!(list.len(), 2);
+        assert!(list.is_allowed(&p1));
+        assert!(list.is_allowed(&p2));
+    }
+
+    #[test]
+    fn from_file_supports_inline_comment() {
+        let p1 = fresh_peer();
+        let tmp = TempDir::new().unwrap();
+        let path = tmp.path().join("allow.txt");
+        let content = format!("{}  # operador foo\n", p1);
+        std::fs::write(&path, content).unwrap();
+        let list = PeerAllowlist::from_file(&path).unwrap();
+        assert!(list.is_allowed(&p1));
+    }
+
+    #[test]
+    fn from_file_rejects_invalid_peer_id() {
+        let tmp = TempDir::new().unwrap();
+        let path = tmp.path().join("bad.txt");
+        std::fs::write(&path, "not-a-real-peer-id\n").unwrap();
+        let err = PeerAllowlist::from_file(&path).unwrap_err();
+        match err {
+            AllowlistError::InvalidPeerId { line_no, .. } => assert_eq!(line_no, 1),
+            other => panic!("wrong error: {other:?}"),
+        }
+    }
+
+    #[test]
+    fn from_file_missing_returns_io_error() {
+        let err = PeerAllowlist::from_file("/no/such/path/allow.txt").unwrap_err();
+        assert!(matches!(err, AllowlistError::Io { .. }));
+    }
+
+    #[test]
+    fn empty_list_rejects_everything() {
+        let list = PeerAllowlist::from_iter(std::iter::empty());
+        assert!(list.is_empty());
+        assert!(!list.is_allowed(&fresh_peer()));
+    }
+}

crates/core/brahman-handshake/src/server.rs

@@ -57,6 +57,13 @@ pub struct ServerConfig {
     /// locales (lo cual es correcto cuando no hay conectividad o no
     /// se desea exponer al exterior).
     pub net: Option<Arc<BrahmanNet>>,
+    /// Política de admisión de peers libp2p. Si está presente, el
+    /// trust gate del path libp2p exige además que el `peer_id`
+    /// autenticado por Noise esté en la lista. `None` → modo abierto
+    /// (cualquier peer Ed25519-válido pasa, comportamiento de Fase 3
+    /// sin restricción adicional). El path Unix la ignora — la
+    /// allowlist es a nivel libp2p, no de filesystem.
+    pub allowlist: Option<crate::peer_allowlist::PeerAllowlist>,
 }
 
 // Manual Debug porque BrahmanNet no implementa Debug (libp2p Swarm
@@ -67,6 +74,7 @@ impl std::fmt::Debug for ServerConfig {
             .field("init_attached", &self.init_attached)
             .field("broker", &self.broker.as_ref().map(|_| "<broker>"))
             .field("net", &self.net.as_ref().map(|_| "<net>"))
+            .field("allowlist", &self.allowlist.as_ref().map(|a| a.len()))
             .finish()
     }
 }
@@ -540,6 +548,26 @@ where
         return Ok(None);
     }
 
+    // Allowlist gate (path libp2p): si está configurada, el peer
+    // autenticado por Noise debe estar en la lista. Se chequea
+    // ANTES de la firma porque es comparación O(log n) sin crypto
+    // — ahorra ciclos contra peers no permitidos. La allowlist no
+    // se aplica al path Unix (autenticación por SO_PEERCRED, no
+    // por libp2p PeerId).
+    if let (Some(peer), Some(allowlist)) = (expected_peer, &config.allowlist) {
+        if !allowlist.is_allowed(&peer) {
+            write_frame(
+                stream,
+                &Frame::Error(HandshakeError::Unauthorized(format!(
+                    "peer {peer} no está en la allowlist"
+                ))),
+            )
+            .await?;
+            debug!(peer = %peer, "rechazado por allowlist");
+            return Ok(None);
+        }
+    }
+
     // Trust gate: en path libp2p (expected_peer = Some), exigir
     // firma cuya public key derive al peer autenticado por Noise.
     // En path Unix (expected_peer = None), si la firma viene se

crates/core/brahman-handshake/tests/handshake.rs

@@ -58,7 +58,7 @@ fn sock_path(name: &str) -> std::path::PathBuf {
 #[tokio::test]
 async fn full_handshake_roundtrip() {
     let path = sock_path("happy");
-    let server = Server::bind(&path, ServerConfig { init_attached: true, broker: None, net: None }).unwrap();
+    let server = Server::bind(&path, ServerConfig { init_attached: true, broker: None, net: None, allowlist: None }).unwrap();
 
     let session_handle = tokio::spawn({
         async move {
@@ -195,6 +195,7 @@ async fn broker_registers_and_unregisters_with_session() {
             init_attached: false,
             broker: Some(broker.clone()),
             net: None,
+            allowlist: None,
         },
     )
     .unwrap();
@@ -238,6 +239,7 @@ async fn broker_matches_two_live_modules() {
             init_attached: false,
             broker: Some(broker.clone()),
             net: None,
+            allowlist: None,
         },
     )
     .unwrap();
@@ -314,6 +316,7 @@ async fn match_event_pushed_on_producer_arrival() {
             init_attached: false,
             broker: Some(broker.clone()),
             net: None,
+            allowlist: None,
         },
     )
     .unwrap();

crates/core/brahman-handshake/tests/network_discovery.rs

@@ -85,6 +85,7 @@ async fn dht_discovery_finds_remote_provider() {
                 init_attached: true,
                 broker: Some(a_broker.clone()),
                 net: Some(a_net.clone()), // ← clave Fase 2: anuncia al DHT
+                allowlist: None,
             },
         )
         .unwrap(),
@@ -171,6 +172,7 @@ async fn dht_discovery_negative_unknown_flow() {
                 init_attached: true,
                 broker: Some(a_broker),
                 net: Some(a_net.clone()),
+                allowlist: None,
             },
         )
         .unwrap(),
@@ -249,6 +251,7 @@ async fn dht_discovery_withdraws_on_session_cleanup() {
                 init_attached: true,
                 broker: Some(a_broker),
                 net: Some(a_net.clone()),
+                allowlist: None,
             },
         )
         .unwrap(),

crates/core/brahman-handshake/tests/network_libp2p.rs

@@ -19,6 +19,7 @@ use brahman_card::{
     CARD_SCHEMA_VERSION,
 };
 use brahman_handshake::network::{connect_libp2p, run_libp2p_accept_loop};
+use brahman_handshake::peer_allowlist::PeerAllowlist;
 use brahman_handshake::server::{Server, ServerConfig};
 use brahman_net::{BrahmanNet, Keypair, Multiaddr, PeerId, Protocol};
 use tempfile::TempDir;
@@ -56,6 +57,7 @@ async fn libp2p_handshake_roundtrip() {
                 init_attached: true,
                 broker: Some(broker.clone()),
                 net: None,
+                allowlist: None,
             },
         )
         .unwrap(),
@@ -134,6 +136,7 @@ async fn libp2p_handshake_rejects_mismatched_signing_key() {
                 init_attached: true,
                 broker: None,
                 net: None,
+                allowlist: None,
             },
         )
         .unwrap(),
@@ -168,3 +171,79 @@ async fn libp2p_handshake_rejects_mismatched_signing_key() {
     let s = sessions.lock().await;
     assert_eq!(s.len(), 0, "no debería haber sesión registrada");
 }
+
+/// Allowlist gate: A configura `allowlist = [client_authorized_peer]`.
+/// Un cliente con peer_id en la lista pasa el handshake; otro con
+/// peer_id distinto es rechazado con `Unauthorized` ANTES de la
+/// verificación de firma (la allowlist se chequea primero, es más
+/// barata).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn libp2p_handshake_allowlist_admits_listed_rejects_others() {
+    // Pre-generamos las dos identidades cliente para que A pueda
+    // construir la allowlist conociendo cuál es la "permitida".
+    let allowed_kp = Keypair::generate_ed25519();
+    let allowed_peer = allowed_kp.public().to_peer_id();
+    let denied_kp = Keypair::generate_ed25519();
+    // (denied_peer no se necesita para la lista — sólo para clarity)
+    let _ = denied_kp.public().to_peer_id();
+
+    // ---- Server con allowlist activa ----
+    let tmp = TempDir::new().unwrap();
+    let unix_socket = tmp.path().join("brahman-init.sock");
+    let server = Arc::new(
+        Server::bind(
+            &unix_socket,
+            ServerConfig {
+                init_attached: true,
+                broker: None,
+                net: None,
+                allowlist: Some(PeerAllowlist::from_iter([allowed_peer])),
+            },
+        )
+        .unwrap(),
+    );
+    let sessions = server.sessions();
+
+    let server_net = Arc::new(BrahmanNet::new().unwrap());
+    let server_peer = server_net.peer_id;
+    let actual = server_net.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
+    let mut full = actual.clone();
+    full.push(Protocol::P2p(server_peer));
+
+    tokio::spawn(run_libp2p_accept_loop(server.clone(), server_net.clone()));
+
+    // ---- Cliente PERMITIDO ----
+    let allowed_net = BrahmanNet::with_keypair(allowed_kp.clone()).unwrap();
+    allowed_net.dial(full.clone());
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    let card_ok = sample_card("test.allowed");
+    let mut allowed_client = connect_libp2p(&allowed_net, server_peer, card_ok, None, &allowed_kp)
+        .await
+        .expect("peer en allowlist debe pasar");
+
+    {
+        let s = sessions.lock().await;
+        assert_eq!(s.len(), 1, "sesión del peer permitido registrada");
+    }
+    allowed_client.farewell().await.ok();
+    tokio::time::sleep(Duration::from_millis(100)).await;
+
+    // ---- Cliente DENEGADO ----
+    let denied_net = BrahmanNet::with_keypair(denied_kp.clone()).unwrap();
+    denied_net.dial(full.clone());
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    let card_no = sample_card("test.denied");
+    let result = connect_libp2p(&denied_net, server_peer, card_no, None, &denied_kp).await;
+
+    assert!(
+        result.is_err(),
+        "peer fuera de allowlist debe ser rechazado, got: {:?}",
+        result.is_ok()
+    );
+    {
+        let s = sessions.lock().await;
+        assert_eq!(s.len(), 0, "ninguna sesión adicional registrada tras intento denegado");
+    }
+}