feature: locally originated routes, session registry, outbound UPDATEs

Phase 1 of the BGP teaching demo: the daemon can now originate routes and
advertise them to every established peer.

- rib: add LocalRib actor holding locally originated routes with owner,
  origin, AS-path prepend, MED, standard and large communities. Enforces a
  LocalRibPolicy (prefix allowlist, per-owner limit, owner match on
  replace/withdraw) and emits LocalRibEvents.
- server: add SessionRegistry with per-session outbound queues; sessions
  register on Established and deregister via a drop guard. Emits
  SessionEvent::{Up,Down}.
- server: advertiser task fans LocalRib changes out to all sessions and
  replays the table to new sessions. UPDATE builders handle IPv4 (legacy
  NLRI) and IPv6 (MP_REACH/MP_UNREACH).
- fsm: merge the timer and timer-less established loops into one select!
  with an outbound branch; reset the hold timer on any received message
  per RFC 4271 §8.2.2 (previously only on KEEPALIVE).
- config: new local_route_policy proto message.
- packet: fix Communitites typo, expose CommunitiesPathAttribute inner vec.
- primitives: IpPrefix::contains.
- integration tests: announce/withdraw/replay/fan-out/policy coverage.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Cargo.lock

@@ -159,6 +159,8 @@ version = "0.0.0"
 dependencies = [
  "bgp-packet",
  "bgp-primitives",
+ "serde",
+ "thiserror 2.0.12",
  "tokio",
  "tracing",
 ]
@@ -177,6 +179,7 @@ dependencies = [
  "prometheus",
  "protobuf",
  "protobuf-codegen",
+ "thiserror 2.0.12",
  "tokio",
  "tracing",
  "tracing-subscriber",

crates/integration-tests/src/test_peer.rs

@@ -1,6 +1,6 @@
 use std::net::{Ipv4Addr, SocketAddr};
 
-use bgp_packet::message::{BGP_HEADER_LEN, Message};
+use bgp_packet::message::{BGP_HEADER_LEN, Message, UpdateMessage};
 use bgp_packet::open::{Capability, OpenMessage, OpenOption};
 use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi};
 use bgp_primitives::parser::ParserContext;
@@ -49,6 +49,18 @@ impl TestPeer {
         Ok(msg)
     }
 
+    /// Read messages until an UPDATE arrives, skipping KEEPALIVEs. Any other
+    /// message type is an error.
+    pub async fn recv_update(&mut self) -> Result<UpdateMessage> {
+        loop {
+            match self.recv().await? {
+                Message::Update(u) => return Ok(u),
+                Message::KeepAlive => {}
+                other => bail!("Expected UPDATE, got {:?}", other),
+            }
+        }
+    }
+
     /// Receive and unwrap as `OpenMessage`, failing if a different type arrives.
     pub async fn recv_open(&mut self) -> Result<OpenMessage> {
         match self.recv().await? {

crates/integration-tests/src/test_server.rs

@@ -1,8 +1,9 @@
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 
-use bgp_rib::AdjRibIn;
+use bgp_rib::{AdjRibIn, LocalRib, LocalRibPolicy};
 use bgp_server::config::Config;
 use bgp_server::server::BgpServerBuilder;
+use bgp_server::session::SessionRegistry;
 use eyre::Result;
 use tokio::task::JoinHandle;
 
@@ -14,6 +15,10 @@ pub struct TestServer {
     pub addr: SocketAddr,
     /// Handle to the server's Adj-RIB-In for test introspection.
     pub adj_rib_in: AdjRibIn,
+    /// Handle to the server's locally originated routes.
+    pub local_rib: LocalRib,
+    /// Live view of the server's established sessions.
+    pub registry: SessionRegistry,
 }
 
 impl TestServer {
@@ -24,16 +29,22 @@ impl TestServer {
 
     /// Start a server with an explicit hold time (useful for timer tests).
     pub async fn start_with_hold_time(local_asn: u32, hold_time_secs: u16) -> Result<Self> {
-        Self::start_inner(local_asn, hold_time_secs).await
+        Self::start_inner(local_asn, hold_time_secs, LocalRibPolicy::default()).await
     }
 
-    async fn start_inner(local_asn: u32, hold_time_secs: u16) -> Result<Self> {
+    /// Start a server with a policy restricting locally originated routes.
+    pub async fn start_with_policy(local_asn: u32, policy: LocalRibPolicy) -> Result<Self> {
+        Self::start_inner(local_asn, 90, policy).await
+    }
+
+    async fn start_inner(local_asn: u32, hold_time_secs: u16, policy: LocalRibPolicy) -> Result<Self> {
         let config = Config {
             router_id: Ipv4Addr::new(127, 0, 0, 1),
             local_asn,
             hold_time_secs,
             listen_endpoints: vec![SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0)],
             peers: vec![],
+            local_route_policy: policy,
         };
 
         let bound = BgpServerBuilder::new(config).bind().await?;
@@ -43,9 +54,11 @@ impl TestServer {
             .next()
             .expect("at least one listener");
         let adj_rib_in = bound.adj_rib_in.clone();
+        let local_rib = bound.local_rib.clone();
+        let registry = bound.registry.clone();
 
         let handle = tokio::spawn(bound.run());
-        Ok(Self { handle, addr, adj_rib_in })
+        Ok(Self { handle, addr, adj_rib_in, local_rib, registry })
     }
 }
 

crates/integration-tests/tests/announce.rs

@@ -0,0 +1,256 @@
+//! End-to-end tests for locally originated routes: announcing via the
+//! `LocalRib` handle must produce UPDATEs on every established session.
+
+use std::net::Ipv4Addr;
+use std::time::Duration;
+
+use bgp_integration_tests::test_peer::TestPeer;
+use bgp_integration_tests::test_server::TestServer;
+use bgp_packet::message::{
+    AsPathAttribute, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity,
+    NextHopPathAttribute, OriginPathAttribute, PathAttribute, UpdateMessage,
+};
+use bgp_primitives::ip_prefix::IpPrefix;
+use bgp_rib::{LocalRibError, LocalRibPolicy, LocalRoute, LocalRouteAttrs};
+use bgp_server::session::SessionEvent;
+use eyre::Result;
+
+const LOCAL_ASN: u32 = 65000;
+const PEER_ASN: u32 = 65001;
+const TIMEOUT: Duration = Duration::from_secs(5);
+
+fn prefix(s: &str) -> IpPrefix {
+    IpPrefix::try_from(s).unwrap()
+}
+
+fn route(p: &str, attrs: LocalRouteAttrs) -> LocalRoute {
+    LocalRoute::new(prefix(p), "alice", attrs)
+}
+
+async fn established_peer(server: &TestServer) -> Result<TestPeer> {
+    let mut peer = TestPeer::connect(server.addr).await?;
+    peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 90).await?;
+    Ok(peer)
+}
+
+async fn next_update(peer: &mut TestPeer) -> Result<UpdateMessage> {
+    tokio::time::timeout(TIMEOUT, peer.recv_update()).await?
+}
+
+/// Waits until the registry reports `n` established sessions.
+async fn wait_for_sessions(server: &TestServer, n: usize) -> Result<()> {
+    tokio::time::timeout(TIMEOUT, async {
+        while server.registry.list().len() != n {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await?;
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------
+// Session registry
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+async fn test_registry_tracks_session_lifecycle() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut events = server.registry.subscribe();
+    assert!(server.registry.list().is_empty());
+
+    let peer = established_peer(&server).await?;
+
+    let up = tokio::time::timeout(TIMEOUT, events.recv()).await??;
+    let SessionEvent::Up(handle) = up else { panic!("expected Up, got {up:?}") };
+    assert_eq!(handle.peer.asn, PEER_ASN);
+    assert_eq!(handle.local_addr, server.addr);
+    assert!(handle.four_octet_asn);
+    assert_eq!(server.registry.list().len(), 1);
+
+    drop(peer);
+
+    let down = tokio::time::timeout(TIMEOUT, events.recv()).await??;
+    assert!(matches!(down, SessionEvent::Down(p) if p == handle.peer), "got {down:?}");
+    assert!(server.registry.list().is_empty());
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------
+// Announce / withdraw
+// ---------------------------------------------------------------------------
+
+/// A route announced while the session is up is sent immediately, with the
+/// attributes the announcer chose and our address as the next hop.
+#[tokio::test]
+async fn test_announce_after_session_up() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut peer = established_peer(&server).await?;
+    wait_for_sessions(&server, 1).await?;
+
+    server
+        .local_rib
+        .announce(route(
+            "10.100.1.0/24",
+            LocalRouteAttrs {
+                origin: OriginPathAttribute::IGP,
+                prepend: 0,
+                med: None,
+                communities: vec![(65000, 100), (65535, 65281)],
+                large_communities: vec![LargeCommunity { global_admin: 65000, data_1: 7, data_2: 8 }],
+            },
+        ))
+        .await?;
+
+    let update = next_update(&mut peer).await?;
+    assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
+    assert!(update.withdrawn_routes.is_empty());
+    assert_eq!(
+        update.path_attributes,
+        vec![
+            PathAttribute::Origin(OriginPathAttribute::IGP),
+            AsPathAttribute::from_asns(vec![LOCAL_ASN]),
+            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::LOCALHOST)),
+            PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 100), (65535, 65281)])),
+            PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute {
+                communities: vec![LargeCommunity { global_admin: 65000, data_1: 7, data_2: 8 }],
+            }),
+        ]
+    );
+    Ok(())
+}
+
+/// Routes announced before any session exists are replayed when a session
+/// reaches Established.
+#[tokio::test]
+async fn test_routes_replayed_on_session_up() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?;
+    server.local_rib.announce(route("10.100.2.0/24", LocalRouteAttrs::default())).await?;
+
+    let mut peer = established_peer(&server).await?;
+
+    let mut seen = vec![next_update(&mut peer).await?, next_update(&mut peer).await?]
+        .into_iter()
+        .flat_map(|u| u.announced_routes)
+        .map(|p| p.to_string())
+        .collect::<Vec<_>>();
+    seen.sort();
+    assert_eq!(seen, vec!["10.100.1.0/24", "10.100.2.0/24"]);
+    Ok(())
+}
+
+/// An announcement is fanned out to every established session.
+#[tokio::test]
+async fn test_announce_reaches_all_sessions() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut a = established_peer(&server).await?;
+    let mut b = established_peer(&server).await?;
+    wait_for_sessions(&server, 2).await?;
+
+    server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?;
+
+    for peer in [&mut a, &mut b] {
+        let update = next_update(peer).await?;
+        assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
+    }
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_withdraw_sends_withdrawn_route() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut peer = established_peer(&server).await?;
+    wait_for_sessions(&server, 1).await?;
+
+    server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?;
+    let _ = next_update(&mut peer).await?;
+
+    server.local_rib.withdraw(prefix("10.100.1.0/24"), Some("alice")).await?;
+    let update = next_update(&mut peer).await?;
+    assert_eq!(update.withdrawn_routes, vec![prefix("10.100.1.0/24")]);
+    assert!(update.announced_routes.is_empty());
+    assert!(update.path_attributes.is_empty());
+    Ok(())
+}
+
+/// Re-announcing with new attributes sends a fresh UPDATE (implicit replace).
+#[tokio::test]
+async fn test_reannounce_with_prepend_and_med() -> Result<()> {
+    let server = TestServer::start(LOCAL_ASN).await?;
+    let mut peer = established_peer(&server).await?;
+    wait_for_sessions(&server, 1).await?;
+
+    server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?;
+    let first = next_update(&mut peer).await?;
+    assert_eq!(first.path_attributes[1], AsPathAttribute::from_asns(vec![LOCAL_ASN]));
+
+    server
+        .local_rib
+        .announce(route(
+            "10.100.1.0/24",
+            LocalRouteAttrs { prepend: 3, med: Some(42), ..LocalRouteAttrs::default() },
+        ))
+        .await?;
+    let second = next_update(&mut peer).await?;
+    assert_eq!(second.announced_routes, vec![prefix("10.100.1.0/24")]);
+    assert_eq!(second.path_attributes[1], AsPathAttribute::from_asns(vec![LOCAL_ASN; 4]));
+    assert!(second
+        .path_attributes
+        .iter()
+        .any(|a| matches!(a, PathAttribute::MultiExitDisc(m) if m.0 == 42)));
+    Ok(())
+}
+
+/// The session stays healthy (keepalives keep flowing) while UPDATEs are sent.
+#[tokio::test]
+async fn test_keepalives_continue_alongside_updates() -> Result<()> {
+    let server = TestServer::start_with_hold_time(LOCAL_ASN, 3).await?;
+    let mut peer = TestPeer::connect(server.addr).await?;
+    peer.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 1), 3).await?;
+    wait_for_sessions(&server, 1).await?;
+
+    server.local_rib.announce(route("10.100.1.0/24", LocalRouteAttrs::default())).await?;
+
+    let mut saw_update = false;
+    let mut saw_keepalive = false;
+    tokio::time::timeout(TIMEOUT, async {
+        while !(saw_update && saw_keepalive) {
+            match peer.recv().await? {
+                bgp_packet::message::Message::Update(_) => saw_update = true,
+                bgp_packet::message::Message::KeepAlive => saw_keepalive = true,
+                other => eyre::bail!("unexpected {other:?}"),
+            }
+        }
+        Ok::<_, eyre::Error>(())
+    })
+    .await??;
+    Ok(())
+}
+
+// ---------------------------------------------------------------------------
+// Policy
+// ---------------------------------------------------------------------------
+
+#[tokio::test]
+async fn test_policy_rejects_prefix_outside_allowlist() -> Result<()> {
+    let server = TestServer::start_with_policy(
+        LOCAL_ASN,
+        LocalRibPolicy { allowed_prefixes: vec![prefix("10.100.0.0/16")], max_routes_per_owner: 2 },
+    )
+    .await?;
+    let mut peer = established_peer(&server).await?;
+    wait_for_sessions(&server, 1).await?;
+
+    let err = server
+        .local_rib
+        .announce(route("192.0.2.0/24", LocalRouteAttrs::default()))
+        .await
+        .unwrap_err();
+    assert_eq!(err, LocalRibError::NotAllowed(prefix("192.0.2.0/24")));
+
+    // An allowed prefix still goes through on the same session.
+    server.local_rib.announce(route("10.100.9.0/24", LocalRouteAttrs::default())).await?;
+    let update = next_update(&mut peer).await?;
+    assert_eq!(update.announced_routes, vec![prefix("10.100.9.0/24")]);
+    Ok(())
+}

crates/packet/src/message.rs

@@ -313,7 +313,7 @@ pub enum PathAttribute {
     LocalPref(LocalPrefPathAttribute) = 5,
     AtomicAggregate(AtomicAggregatePathAttribute) = 6,
     Aggregator(AggregatorPathAttribute) = 7,
-    Communitites(CommunitiesPathAttribute) = 8,
+    Communities(CommunitiesPathAttribute) = 8,
     MpReachNlri(MpReachNlriPathAttribute) = 14,
     MpUnreachNlri(MpUnreachNlriPathAttribute) = 15,
     ExtendedCommunities(ExtendedCommunitiesPathAttribute) = 16,
@@ -335,7 +335,7 @@ impl From<u8> for PathAttributeDiscriminants {
             v if v == Self::LocalPref as u8 => Self::LocalPref,
             v if v == Self::AtomicAggregate as u8 => Self::AtomicAggregate,
             v if v == Self::Aggregator as u8 => Self::Aggregator,
-            v if v == Self::Communitites as u8 => Self::Communitites,
+            v if v == Self::Communities as u8 => Self::Communities,
             v if v == Self::MpReachNlri as u8 => Self::MpReachNlri,
             v if v == Self::MpUnreachNlri as u8 => Self::MpUnreachNlri,
             v if v == Self::ExtendedCommunities as u8 => Self::ExtendedCommunities,
@@ -355,7 +355,7 @@ impl PathAttribute {
             PathAttribute::LocalPref(_) => PathAttributeDiscriminants::LocalPref as u8,
             PathAttribute::AtomicAggregate(_) => PathAttributeDiscriminants::AtomicAggregate as u8,
             PathAttribute::Aggregator(_) => PathAttributeDiscriminants::Aggregator as u8,
-            PathAttribute::Communitites(_) => PathAttributeDiscriminants::Communitites as u8,
+            PathAttribute::Communities(_) => PathAttributeDiscriminants::Communities as u8,
             PathAttribute::MpReachNlri(_) => PathAttributeDiscriminants::MpReachNlri as u8,
             PathAttribute::MpUnreachNlri(_) => PathAttributeDiscriminants::MpUnreachNlri as u8,
             PathAttribute::ExtendedCommunities(_) => {
@@ -377,7 +377,7 @@ impl PathAttribute {
             PathAttribute::LocalPref(_) => false,
             PathAttribute::AtomicAggregate(_) => true,
             PathAttribute::Aggregator(_) => true,
-            PathAttribute::Communitites(_) => true,
+            PathAttribute::Communities(_) => true,
             PathAttribute::MpReachNlri(_) => true,
             PathAttribute::MpUnreachNlri(_) => true,
             PathAttribute::ExtendedCommunities(_) => true,
@@ -395,7 +395,7 @@ impl PathAttribute {
             PathAttribute::LocalPref(_) => true,
             PathAttribute::AtomicAggregate(_) => true,
             PathAttribute::Aggregator(_) => false,
-            PathAttribute::Communitites(_) => true,
+            PathAttribute::Communities(_) => true,
             PathAttribute::MpReachNlri(_) => false,
             PathAttribute::MpUnreachNlri(_) => false,
             PathAttribute::ExtendedCommunities(_) => true,
@@ -467,8 +467,8 @@ impl PathAttribute {
             PathAttributeDiscriminants::Aggregator => {
                 PathAttribute::Aggregator(AggregatorPathAttribute::from_wire(ctx, pa_buf)?.1)
             }
-            PathAttributeDiscriminants::Communitites => {
+            PathAttributeDiscriminants::Communities => {
-                PathAttribute::Communitites(CommunitiesPathAttribute::from_wire(ctx, pa_buf)?.1)
+                PathAttribute::Communities(CommunitiesPathAttribute::from_wire(ctx, pa_buf)?.1)
             }
             PathAttributeDiscriminants::MpReachNlri => {
                 PathAttribute::MpReachNlri(MpReachNlriPathAttribute::from_wire(ctx, pa_buf)?.1)
@@ -538,7 +538,7 @@ impl PathAttribute {
             PathAttribute::Aggregator(aggregator_path_attribute) => {
                 write_path_attribute!(out, aggregator_path_attribute);
             }
-            PathAttribute::Communitites(communities_path_attribute) => {
+            PathAttribute::Communities(communities_path_attribute) => {
                 write_path_attribute!(out, communities_path_attribute);
             }
             PathAttribute::MpReachNlri(mp_reach_nlri_path_attribute) => {
@@ -583,7 +583,7 @@ impl PathAttribute {
             PathAttribute::LocalPref(a) => a.wire_len(ctx)?,
             PathAttribute::AtomicAggregate(a) => a.wire_len(ctx)?,
             PathAttribute::Aggregator(a) => a.wire_len(ctx)?,
-            PathAttribute::Communitites(a) => a.wire_len(ctx)?,
+            PathAttribute::Communities(a) => a.wire_len(ctx)?,
             PathAttribute::MpReachNlri(a) => a.wire_len(ctx)?,
             PathAttribute::MpUnreachNlri(a) => a.wire_len(ctx)?,
             PathAttribute::ExtendedCommunities(a) => a.wire_len(ctx)?,
@@ -903,7 +903,7 @@ impl AggregatorPathAttribute {
 }
 
 #[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
-pub struct CommunitiesPathAttribute(Vec<(u16, u16)>);
+pub struct CommunitiesPathAttribute(pub Vec<(u16, u16)>);
 
 impl CommunitiesPathAttribute {
     pub fn from_wire<'a>(
@@ -1624,7 +1624,7 @@ mod tests {
     test_path_attribute_roundtrip!(
         test_communities,
         &[0xc0, 0x08, 0x04, 0xca, 0xfe, 0xba, 0xbe],
-        PathAttribute::Communitites(CommunitiesPathAttribute(vec![(0xcafe, 0xbabe)]))
+        PathAttribute::Communities(CommunitiesPathAttribute(vec![(0xcafe, 0xbabe)]))
     );
 
     // MP Reach NLRI Path Attribute.

crates/primitives/src/ip_prefix.rs

@@ -34,6 +34,25 @@ impl IpPrefix {
         })
     }
 
+    /// Returns true if `other` is equal to or more specific than `self`, i.e.
+    /// `self` covers `other`. Prefixes of different address families never
+    /// contain each other.
+    pub fn contains(&self, other: &IpPrefix) -> bool {
+        if self.address_family != other.address_family || other.length < self.length {
+            return false;
+        }
+        let full_bytes = (self.length / 8) as usize;
+        if self.prefix[..full_bytes] != other.prefix[..full_bytes] {
+            return false;
+        }
+        let rem_bits = self.length % 8;
+        if rem_bits == 0 {
+            return true;
+        }
+        let mask = u8::MAX << (8 - rem_bits);
+        (self.prefix[full_bytes] & mask) == (other.prefix[full_bytes] & mask)
+    }
+
     pub fn to_wire(&self, _: &ParserContext, out: &mut BytesMut) -> Result<(), ToWireError> {
         // Verify that there is enough space to write the IpPrefix.
         if out.remaining() < (self.prefix.len() + 1) {
@@ -205,6 +224,22 @@ mod tests {
         32
     );
 
+    #[test]
+    fn contains_covers_more_specifics() {
+        let p = |s: &str| IpPrefix::try_from(s).unwrap();
+        assert!(p("10.100.0.0/16").contains(&p("10.100.7.0/24")));
+        assert!(p("10.100.0.0/16").contains(&p("10.100.0.0/16")));
+        assert!(!p("10.100.0.0/16").contains(&p("10.101.0.0/24")));
+        assert!(!p("10.100.7.0/24").contains(&p("10.100.0.0/16")));
+        assert!(p("0.0.0.0/0").contains(&p("192.0.2.0/24")));
+        // Non-byte-aligned boundary.
+        assert!(p("10.100.0.0/19").contains(&p("10.100.31.0/24")));
+        assert!(!p("10.100.0.0/19").contains(&p("10.100.32.0/24")));
+        // Cross-family never matches.
+        assert!(!p("::/0").contains(&p("10.0.0.0/8")));
+        assert!(p("2001:db8::/32").contains(&p("2001:db8:cafe::/48")));
+    }
+
     verify_roundtrip!(
         verify_roundtrip_ipv6_0,
         "::/0",

crates/rib/Cargo.toml

@@ -5,5 +5,7 @@ edition.workspace = true
 [dependencies]
 bgp-packet.workspace = true
 bgp-primitives.workspace = true
+serde.workspace = true
+thiserror.workspace = true
 tokio = { workspace = true, features = ["full"] }
 tracing.workspace = true

crates/rib/src/lib.rs

@@ -1,7 +1,9 @@
 mod adj_rib_in;
 mod entry;
 mod event;
+mod local_rib;
 
 pub use adj_rib_in::AdjRibIn;
 pub use entry::{PeerId, RibEntry, SnapshotEntry};
 pub use event::RibEvent;
+pub use local_rib::{LocalRib, LocalRibError, LocalRibEvent, LocalRibPolicy, LocalRoute, LocalRouteAttrs};

crates/rib/src/local_rib.rs

@@ -0,0 +1,305 @@
+//! Locally originated routes (the "Loc-RIB" of routes this router announces
+//! on its own behalf, as opposed to routes learned from peers).
+//!
+//! The actor stores one route per prefix together with the attributes chosen
+//! by whoever announced it. It does not know about peers or sessions: the
+//! server subscribes to [`LocalRibEvent`]s and turns them into UPDATEs.
+
+use std::collections::HashMap;
+use std::time::SystemTime;
+
+use bgp_packet::message::{LargeCommunity, OriginPathAttribute};
+use bgp_primitives::ip_prefix::IpPrefix;
+use serde::{Deserialize, Serialize};
+use tokio::sync::{broadcast, mpsc, oneshot};
+use tracing::warn;
+
+const EVENT_CHANNEL_CAPACITY: usize = 1024;
+const CMD_CHANNEL_CAPACITY: usize = 256;
+
+/// User-selectable attributes for a locally originated route.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LocalRouteAttrs {
+    pub origin: OriginPathAttribute,
+    /// Number of *additional* copies of the local ASN to prepend to the
+    /// AS_PATH. `0` yields the plain `[local_asn]` path.
+    pub prepend: u8,
+    pub med: Option<u32>,
+    /// RFC 1997 standard communities as `(high, low)` 16-bit halves.
+    pub communities: Vec<(u16, u16)>,
+    /// RFC 8092 large communities.
+    pub large_communities: Vec<LargeCommunity>,
+}
+
+impl Default for LocalRouteAttrs {
+    fn default() -> Self {
+        Self {
+            origin: OriginPathAttribute::IGP,
+            prepend: 0,
+            med: None,
+            communities: vec![],
+            large_communities: vec![],
+        }
+    }
+}
+
+/// A route originated by this router.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LocalRoute {
+    pub prefix: IpPrefix,
+    /// Free-form identity of whoever announced the route (e.g. a student
+    /// name). Used to stop one owner replacing or withdrawing another's route.
+    pub owner: String,
+    pub attrs: LocalRouteAttrs,
+    pub announced_at: SystemTime,
+}
+
+impl LocalRoute {
+    pub fn new(prefix: IpPrefix, owner: impl Into<String>, attrs: LocalRouteAttrs) -> Self {
+        Self { prefix, owner: owner.into(), attrs, announced_at: SystemTime::now() }
+    }
+}
+
+/// Limits on what may be announced. Enforced inside the actor so every caller
+/// (HTTP API, tests, future CLI) goes through the same checks.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct LocalRibPolicy {
+    /// A route is accepted only if it is covered by at least one of these
+    /// prefixes. Empty means no restriction.
+    pub allowed_prefixes: Vec<IpPrefix>,
+    /// Maximum number of routes a single owner may hold. `0` means unlimited.
+    pub max_routes_per_owner: usize,
+}
+
+impl LocalRibPolicy {
+    fn allows(&self, prefix: &IpPrefix) -> bool {
+        self.allowed_prefixes.is_empty() || self.allowed_prefixes.iter().any(|p| p.contains(prefix))
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum LocalRibError {
+    #[error("prefix {0} is not covered by the allowed prefixes")]
+    NotAllowed(IpPrefix),
+    #[error("owner {owner:?} already has {limit} routes")]
+    OwnerLimit { owner: String, limit: usize },
+    #[error("prefix {prefix} is owned by {owner:?}")]
+    OwnedByOther { prefix: IpPrefix, owner: String },
+    #[error("local RIB actor has stopped")]
+    Closed,
+}
+
+#[derive(Debug, Clone)]
+pub enum LocalRibEvent {
+    /// A route was added or its attributes replaced.
+    Announced(LocalRoute),
+    Withdrawn { prefix: IpPrefix },
+}
+
+enum Command {
+    Announce { route: LocalRoute, reply: oneshot::Sender<Result<(), LocalRibError>> },
+    Withdraw {
+        prefix: IpPrefix,
+        requester: Option<String>,
+        reply: oneshot::Sender<Result<Option<LocalRoute>, LocalRibError>>,
+    },
+    Snapshot { reply: oneshot::Sender<Vec<LocalRoute>> },
+}
+
+/// Handle to the local RIB actor. Cheap to clone; all clones share one table.
+#[derive(Clone)]
+pub struct LocalRib {
+    tx: mpsc::Sender<Command>,
+    events: broadcast::Sender<LocalRibEvent>,
+}
+
+impl LocalRib {
+    pub fn spawn(policy: LocalRibPolicy) -> Self {
+        let (tx, rx) = mpsc::channel(CMD_CHANNEL_CAPACITY);
+        let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
+        tokio::spawn(run_actor(rx, events.clone(), policy));
+        Self { tx, events }
+    }
+
+    /// Adds a route, or replaces the attributes of an existing route with the
+    /// same prefix and owner.
+    pub async fn announce(&self, route: LocalRoute) -> Result<(), LocalRibError> {
+        let (reply, rx) = oneshot::channel();
+        self.tx
+            .send(Command::Announce { route, reply })
+            .await
+            .map_err(|_| LocalRibError::Closed)?;
+        rx.await.map_err(|_| LocalRibError::Closed)?
+    }
+
+    /// Removes a route. `requester` must match the route's owner; pass `None`
+    /// to withdraw regardless of owner (administrative override). Returns the
+    /// removed route, or `None` if the prefix was not present.
+    pub async fn withdraw(
+        &self,
+        prefix: IpPrefix,
+        requester: Option<&str>,
+    ) -> Result<Option<LocalRoute>, LocalRibError> {
+        let (reply, rx) = oneshot::channel();
+        self.tx
+            .send(Command::Withdraw { prefix, requester: requester.map(str::to_owned), reply })
+            .await
+            .map_err(|_| LocalRibError::Closed)?;
+        rx.await.map_err(|_| LocalRibError::Closed)?
+    }
+
+    /// Point-in-time copy of every route, in no particular order.
+    pub async fn snapshot(&self) -> Vec<LocalRoute> {
+        let (reply, rx) = oneshot::channel();
+        let _ = self.tx.send(Command::Snapshot { reply }).await;
+        rx.await.unwrap_or_default()
+    }
+
+    /// Subscribe to mutations. See [`crate::AdjRibIn::subscribe`] for lag
+    /// semantics.
+    pub fn subscribe(&self) -> broadcast::Receiver<LocalRibEvent> {
+        self.events.subscribe()
+    }
+}
+
+async fn run_actor(
+    mut rx: mpsc::Receiver<Command>,
+    events: broadcast::Sender<LocalRibEvent>,
+    policy: LocalRibPolicy,
+) {
+    let mut table: HashMap<IpPrefix, LocalRoute> = HashMap::new();
+
+    while let Some(cmd) = rx.recv().await {
+        match cmd {
+            Command::Announce { route, reply } => {
+                let result = check_announce(&table, &policy, &route).map(|()| {
+                    table.insert(route.prefix.clone(), route.clone());
+                    let _ = events.send(LocalRibEvent::Announced(route));
+                });
+                let _ = reply.send(result);
+            }
+            Command::Withdraw { prefix, requester, reply } => {
+                let result = match (table.get(&prefix), requester) {
+                    (Some(existing), Some(req)) if existing.owner != req => {
+                        Err(LocalRibError::OwnedByOther {
+                            prefix: prefix.clone(),
+                            owner: existing.owner.clone(),
+                        })
+                    }
+                    _ => {
+                        let removed = table.remove(&prefix);
+                        if removed.is_some() {
+                            let _ = events.send(LocalRibEvent::Withdrawn { prefix });
+                        }
+                        Ok(removed)
+                    }
+                };
+                let _ = reply.send(result);
+            }
+            Command::Snapshot { reply } => {
+                if reply.send(table.values().cloned().collect()).is_err() {
+                    warn!("Snapshot reply channel closed before send");
+                }
+            }
+        }
+    }
+}
+
+fn check_announce(
+    table: &HashMap<IpPrefix, LocalRoute>,
+    policy: &LocalRibPolicy,
+    route: &LocalRoute,
+) -> Result<(), LocalRibError> {
+    if !policy.allows(&route.prefix) {
+        return Err(LocalRibError::NotAllowed(route.prefix.clone()));
+    }
+    if let Some(existing) = table.get(&route.prefix) {
+        if existing.owner != route.owner {
+            return Err(LocalRibError::OwnedByOther {
+                prefix: route.prefix.clone(),
+                owner: existing.owner.clone(),
+            });
+        }
+        // Replacing our own route never changes the per-owner count.
+        return Ok(());
+    }
+    if policy.max_routes_per_owner > 0 {
+        let held = table.values().filter(|r| r.owner == route.owner).count();
+        if held >= policy.max_routes_per_owner {
+            return Err(LocalRibError::OwnerLimit {
+                owner: route.owner.clone(),
+                limit: policy.max_routes_per_owner,
+            });
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn prefix(s: &str) -> IpPrefix {
+        IpPrefix::try_from(s).unwrap()
+    }
+
+    fn route(p: &str, owner: &str) -> LocalRoute {
+        LocalRoute::new(prefix(p), owner, LocalRouteAttrs::default())
+    }
+
+    #[tokio::test]
+    async fn announce_and_withdraw_emit_events() {
+        let rib = LocalRib::spawn(LocalRibPolicy::default());
+        let mut events = rib.subscribe();
+
+        rib.announce(route("10.0.0.0/24", "alice")).await.unwrap();
+        assert!(matches!(events.recv().await.unwrap(), LocalRibEvent::Announced(r) if r.prefix == prefix("10.0.0.0/24")));
+        assert_eq!(rib.snapshot().await.len(), 1);
+
+        let removed = rib.withdraw(prefix("10.0.0.0/24"), Some("alice")).await.unwrap();
+        assert_eq!(removed.unwrap().owner, "alice");
+        assert!(matches!(events.recv().await.unwrap(), LocalRibEvent::Withdrawn { prefix: p } if p == prefix("10.0.0.0/24")));
+        assert!(rib.snapshot().await.is_empty());
+
+        // Withdrawing an absent prefix is a no-op, not an error.
+        assert_eq!(rib.withdraw(prefix("10.0.0.0/24"), None).await.unwrap(), None);
+    }
+
+    #[tokio::test]
+    async fn policy_allowlist_is_enforced() {
+        let rib = LocalRib::spawn(LocalRibPolicy {
+            allowed_prefixes: vec![prefix("10.100.0.0/16")],
+            max_routes_per_owner: 0,
+        });
+        rib.announce(route("10.100.3.0/24", "alice")).await.unwrap();
+        let err = rib.announce(route("192.0.2.0/24", "alice")).await.unwrap_err();
+        assert_eq!(err, LocalRibError::NotAllowed(prefix("192.0.2.0/24")));
+    }
+
+    #[tokio::test]
+    async fn owner_limit_ignores_replacements() {
+        let rib = LocalRib::spawn(LocalRibPolicy { allowed_prefixes: vec![], max_routes_per_owner: 1 });
+        rib.announce(route("10.0.0.0/24", "alice")).await.unwrap();
+        // Replacing the same prefix is fine.
+        rib.announce(route("10.0.0.0/24", "alice")).await.unwrap();
+        let err = rib.announce(route("10.0.1.0/24", "alice")).await.unwrap_err();
+        assert!(matches!(err, LocalRibError::OwnerLimit { limit: 1, .. }));
+        // Other owners are unaffected.
+        rib.announce(route("10.0.1.0/24", "bob")).await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn other_owner_cannot_replace_or_withdraw() {
+        let rib = LocalRib::spawn(LocalRibPolicy::default());
+        rib.announce(route("10.0.0.0/24", "alice")).await.unwrap();
+
+        let err = rib.announce(route("10.0.0.0/24", "bob")).await.unwrap_err();
+        assert!(matches!(err, LocalRibError::OwnedByOther { .. }));
+
+        let err = rib.withdraw(prefix("10.0.0.0/24"), Some("bob")).await.unwrap_err();
+        assert!(matches!(err, LocalRibError::OwnedByOther { .. }));
+
+        // Administrative override works.
+        assert!(rib.withdraw(prefix("10.0.0.0/24"), None).await.unwrap().is_some());
+    }
+}

crates/server/Cargo.toml

@@ -15,6 +15,7 @@ tokio.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true
 protobuf.workspace = true
+thiserror.workspace = true
 
 [build-dependencies]
 protobuf-codegen.workspace = true

crates/server/proto/config.proto

@@ -17,6 +17,20 @@ message BgpConfig {
 
   // Peers configured on this server.
   repeated Peer peers = 4;
+
+  // Limits on routes originated locally (e.g. via the demo UI).
+  LocalRoutePolicy local_route_policy = 5;
+}
+
+// Limits on locally originated routes.
+message LocalRoutePolicy {
+  // A locally originated route is accepted only if it is covered by one of
+  // these prefixes, written in CIDR notation (e.g. "10.100.0.0/16").
+  // Empty means any prefix may be announced.
+  repeated string allowed_prefixes = 1;
+
+  // Maximum number of routes a single owner may announce. 0 means unlimited.
+  uint32 max_routes_per_owner = 2;
 }
 
 // An IP address and port pair.

crates/server/src/advertise.rs

@@ -0,0 +1,267 @@
+//! Turns locally originated routes into UPDATE messages for every established
+//! session.
+//!
+//! The advertiser task listens for [`LocalRibEvent`]s and [`SessionEvent`]s.
+//! A route change is fanned out to all live sessions; a new session receives
+//! the full local table. Sending the same route twice is harmless (BGP treats
+//! a re-announcement as an implicit replace), so the two event streams need
+//! no coordination beyond being handled by one task.
+
+use std::net::IpAddr;
+
+use bgp_packet::message::{
+    AsPathAttribute, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, Message,
+    MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, MultiExitDiscPathAttribute,
+    NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
+};
+use bgp_primitives::constants::{AddressFamilyId, SubsequentAfi};
+use bgp_primitives::ip_prefix::IpPrefix;
+use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute};
+use tokio::sync::broadcast::error::RecvError;
+use tokio::task::JoinHandle;
+use tracing::{debug, info, warn};
+
+use crate::session::{SessionEvent, SessionHandle, SessionRegistry};
+
+pub fn spawn_advertiser(
+    local_rib: LocalRib,
+    registry: SessionRegistry,
+    local_asn: u32,
+) -> JoinHandle<()> {
+    tokio::spawn(run(local_rib, registry, local_asn))
+}
+
+async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) {
+    let mut rib_events = local_rib.subscribe();
+    let mut session_events = registry.subscribe();
+
+    loop {
+        tokio::select! {
+            ev = rib_events.recv() => match ev {
+                Ok(LocalRibEvent::Announced(route)) => {
+                    for session in registry.list() {
+                        send_route(&session, &route, local_asn).await;
+                    }
+                }
+                Ok(LocalRibEvent::Withdrawn { prefix }) => {
+                    for session in registry.list() {
+                        send_withdraw(&session, &prefix).await;
+                    }
+                }
+                Err(RecvError::Lagged(n)) => {
+                    warn!(missed = n, "Advertiser lagged on local RIB events, resyncing all sessions");
+                    resync_all(&local_rib, &registry, local_asn).await;
+                }
+                Err(RecvError::Closed) => return,
+            },
+            ev = session_events.recv() => match ev {
+                Ok(SessionEvent::Up(session)) => {
+                    let routes = local_rib.snapshot().await;
+                    info!(peer = %session.peer, routes = routes.len(), "Session up, sending local routes");
+                    for route in &routes {
+                        send_route(&session, route, local_asn).await;
+                    }
+                }
+                Ok(SessionEvent::Down(_)) => {}
+                Err(RecvError::Lagged(n)) => {
+                    warn!(missed = n, "Advertiser lagged on session events, resyncing all sessions");
+                    resync_all(&local_rib, &registry, local_asn).await;
+                }
+                Err(RecvError::Closed) => return,
+            },
+        }
+    }
+}
+
+async fn resync_all(local_rib: &LocalRib, registry: &SessionRegistry, local_asn: u32) {
+    let routes = local_rib.snapshot().await;
+    for session in registry.list() {
+        for route in &routes {
+            send_route(&session, route, local_asn).await;
+        }
+    }
+}
+
+async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32) {
+    match build_announce(route, local_asn, session.local_addr.ip()) {
+        Some(update) => {
+            debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route");
+            if let Err(e) = session.send(Message::Update(update)).await {
+                warn!(peer = %session.peer, err = %e, "Failed to queue announce");
+            }
+        }
+        None => debug!(
+            peer = %session.peer,
+            prefix = %route.prefix,
+            "Skipping route: address family does not match session transport"
+        ),
+    }
+}
+
+async fn send_withdraw(session: &SessionHandle, prefix: &IpPrefix) {
+    debug!(peer = %session.peer, %prefix, "Withdrawing local route");
+    if let Err(e) = session.send(Message::Update(build_withdraw(prefix))).await {
+        warn!(peer = %session.peer, err = %e, "Failed to queue withdraw");
+    }
+}
+
+/// Builds the UPDATE announcing `route` with `next_hop` as the next hop.
+///
+/// Returns `None` when the prefix family does not match the next-hop family:
+/// carrying IPv4 NLRI over an IPv6 next hop (RFC 8950) or vice versa is not
+/// supported yet.
+pub fn build_announce(route: &LocalRoute, local_asn: u32, next_hop: IpAddr) -> Option<UpdateMessage> {
+    let attrs = &route.attrs;
+
+    // Path attributes in ascending type-code order, as most implementations
+    // expect them.
+    let mut path_attributes = vec![
+        PathAttribute::Origin(attrs.origin),
+        AsPathAttribute::from_asns(vec![local_asn; 1 + attrs.prepend as usize]),
+    ];
+
+    let mut announced_routes = vec![];
+    match (route.prefix.address_family, next_hop) {
+        (AddressFamilyId::Ipv4, IpAddr::V4(nh)) => {
+            path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(nh)));
+            announced_routes.push(route.prefix.clone());
+        }
+        (AddressFamilyId::Ipv6, IpAddr::V6(_)) => {}
+        _ => return None,
+    }
+
+    if let Some(med) = attrs.med {
+        path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(med)));
+    }
+    if !attrs.communities.is_empty() {
+        path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute(
+            attrs.communities.clone(),
+        )));
+    }
+    if let (AddressFamilyId::Ipv6, IpAddr::V6(nh)) = (route.prefix.address_family, next_hop) {
+        path_attributes.push(PathAttribute::MpReachNlri(MpReachNlriPathAttribute {
+            afi: AddressFamilyId::Ipv6,
+            safi: SubsequentAfi::Unicast,
+            next_hop: NlriNextHop::Ipv6(nh),
+            prefixes: vec![route.prefix.clone()],
+        }));
+    }
+    if !attrs.large_communities.is_empty() {
+        path_attributes.push(PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute {
+            communities: attrs.large_communities.clone(),
+        }));
+    }
+
+    Some(UpdateMessage { withdrawn_routes: vec![], path_attributes, announced_routes })
+}
+
+/// Builds the UPDATE withdrawing `prefix`.
+pub fn build_withdraw(prefix: &IpPrefix) -> UpdateMessage {
+    match prefix.address_family {
+        AddressFamilyId::Ipv4 => UpdateMessage {
+            withdrawn_routes: vec![prefix.clone()],
+            path_attributes: vec![],
+            announced_routes: vec![],
+        },
+        AddressFamilyId::Ipv6 => UpdateMessage {
+            withdrawn_routes: vec![],
+            path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
+                afi: AddressFamilyId::Ipv6,
+                safi: SubsequentAfi::Unicast,
+                prefixes: vec![prefix.clone()],
+            })],
+            announced_routes: vec![],
+        },
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::net::{Ipv4Addr, Ipv6Addr};
+
+    use bgp_packet::message::{LargeCommunity, OriginPathAttribute};
+    use bgp_primitives::parser::ParserContext;
+    use bgp_rib::LocalRouteAttrs;
+    use bytes::BytesMut;
+
+    use super::*;
+
+    fn prefix(s: &str) -> IpPrefix {
+        IpPrefix::try_from(s).unwrap()
+    }
+
+    #[test]
+    fn ipv4_announce_has_expected_attributes_and_roundtrips() {
+        let route = LocalRoute::new(
+            prefix("10.100.1.0/24"),
+            "alice",
+            LocalRouteAttrs {
+                origin: OriginPathAttribute::IGP,
+                prepend: 2,
+                med: Some(50),
+                communities: vec![(65000, 100), (65535, 65281)],
+                large_communities: vec![LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 }],
+            },
+        );
+        let update = build_announce(&route, 65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1))).unwrap();
+
+        assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]);
+        assert!(update.withdrawn_routes.is_empty());
+        assert_eq!(
+            update.path_attributes,
+            vec![
+                PathAttribute::Origin(OriginPathAttribute::IGP),
+                AsPathAttribute::from_asns(vec![65000, 65000, 65000]),
+                PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
+                PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute(50)),
+                PathAttribute::Communities(CommunitiesPathAttribute(vec![(65000, 100), (65535, 65281)])),
+                PathAttribute::LargeCommunities(LargeCommunitiesPathAttribute {
+                    communities: vec![LargeCommunity { global_admin: 65000, data_1: 1, data_2: 2 }],
+                }),
+            ]
+        );
+
+        // Serialise and parse back to make sure the wire encoding is sound.
+        let ctx = ParserContext { four_octet_asn: Some(true), address_family: None };
+        let msg = Message::Update(update.clone());
+        let mut buf = BytesMut::new();
+        msg.to_wire(&ctx, &mut buf).unwrap();
+        let (_, parsed) = Message::from_wire(&ctx, &buf).unwrap();
+        let Message::Update(parsed) = parsed else { panic!("expected UPDATE") };
+        assert_eq!(parsed, update);
+    }
+
+    #[test]
+    fn ipv6_announce_uses_mp_reach() {
+        let route = LocalRoute::new(prefix("2001:db8:1::/48"), "alice", LocalRouteAttrs::default());
+        let nh = Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1);
+        let update = build_announce(&route, 65000, IpAddr::V6(nh)).unwrap();
+
+        assert!(update.announced_routes.is_empty());
+        assert!(!update.path_attributes.iter().any(|a| matches!(a, PathAttribute::NextHop(_))));
+        assert!(update.path_attributes.iter().any(|a| matches!(
+            a,
+            PathAttribute::MpReachNlri(MpReachNlriPathAttribute { next_hop: NlriNextHop::Ipv6(h), prefixes, .. })
+                if *h == nh && *prefixes == vec![prefix("2001:db8:1::/48")]
+        )));
+    }
+
+    #[test]
+    fn family_mismatch_is_skipped() {
+        let v6 = LocalRoute::new(prefix("2001:db8::/32"), "a", LocalRouteAttrs::default());
+        assert!(build_announce(&v6, 65000, IpAddr::V4(Ipv4Addr::LOCALHOST)).is_none());
+        let v4 = LocalRoute::new(prefix("10.0.0.0/8"), "a", LocalRouteAttrs::default());
+        assert!(build_announce(&v4, 65000, IpAddr::V6(Ipv6Addr::LOCALHOST)).is_none());
+    }
+
+    #[test]
+    fn withdraw_by_family() {
+        let v4 = build_withdraw(&prefix("10.0.0.0/8"));
+        assert_eq!(v4.withdrawn_routes, vec![prefix("10.0.0.0/8")]);
+        assert!(v4.path_attributes.is_empty());
+
+        let v6 = build_withdraw(&prefix("2001:db8::/32"));
+        assert!(v6.withdrawn_routes.is_empty());
+        assert!(matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")]));
+    }
+}

crates/server/src/config.rs

@@ -1,7 +1,9 @@
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::str::FromStr;
 
-use eyre::{Result, bail, eyre};
+use bgp_primitives::ip_prefix::IpPrefix;
+use bgp_rib::LocalRibPolicy;
+use eyre::{Result, WrapErr, bail, eyre};
 
 use crate::proto::config::BgpConfig;
 
@@ -16,6 +18,8 @@ pub struct Config {
     pub hold_time_secs: u16,
     pub listen_endpoints: Vec<SocketAddr>,
     pub peers: Vec<PeerConfig>,
+    /// Limits on locally originated routes.
+    pub local_route_policy: LocalRibPolicy,
 }
 
 pub struct PeerConfig {
@@ -74,7 +78,26 @@ impl Config {
             })
             .collect::<Result<_>>()?;
 
-        Ok(Self { router_id, local_asn: proto.local_asn, hold_time_secs: DEFAULT_HOLD_TIME_SECS, listen_endpoints, peers })
+        let local_route_policy = match proto.local_route_policy.into_option() {
+            Some(policy) => LocalRibPolicy {
+                allowed_prefixes: policy
+                    .allowed_prefixes
+                    .iter()
+                    .map(|p| IpPrefix::try_from(p.as_str()).wrap_err_with(|| format!("Invalid allowed prefix {p:?}")))
+                    .collect::<Result<_>>()?,
+                max_routes_per_owner: policy.max_routes_per_owner as usize,
+            },
+            None => LocalRibPolicy::default(),
+        };
+
+        Ok(Self {
+            router_id,
+            local_asn: proto.local_asn,
+            hold_time_secs: DEFAULT_HOLD_TIME_SECS,
+            listen_endpoints,
+            peers,
+            local_route_policy,
+        })
     }
 }
 

crates/server/src/fsm.rs

@@ -12,11 +12,18 @@ use eyre::{Result, bail, eyre};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
+use tokio::sync::mpsc;
 use tokio::time::{Instant, interval_at, sleep_until};
 use tracing::{info, warn};
 
 use bgp_packet::message::{AS_TRANS, BGP4_VERSION};
 
+use crate::session::{SessionHandle, SessionRegistry};
+
+/// Number of outbound messages that may be queued for a session before
+/// senders are back-pressured.
+const OUTBOUND_QUEUE_CAPACITY: usize = 1024;
+
 /// Minimal per-session config extracted from the server config.
 #[derive(Clone)]
 pub struct SessionConfig {
@@ -24,6 +31,7 @@ pub struct SessionConfig {
     pub local_asn: u32,
     pub hold_time_secs: u16,
     pub adj_rib_in: AdjRibIn,
+    pub registry: SessionRegistry,
 }
 
 /// Drive the BGP FSM for a single accepted TCP connection.
@@ -34,6 +42,7 @@ pub async fn run_session(
     peer_addr: SocketAddr,
     config: SessionConfig,
 ) -> Result<()> {
+    let local_addr = stream.local_addr()?;
     let (mut reader, mut writer) = stream.into_split();
 
     // --- OpenSent -----------------------------------------------------------
@@ -136,6 +145,16 @@ pub async fn run_session(
     let peer_id = PeerId { addr: peer_addr, asn: peer_asn };
 
     // --- Established --------------------------------------------------------
+    // Register so the rest of the server can see this session and queue
+    // messages for it. The guard deregisters on any exit path.
+    let (outbound_tx, mut outbound_rx) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY);
+    let _guard = config.registry.register(SessionHandle::new(
+        peer_id,
+        local_addr,
+        peer_four_byte_asn,
+        outbound_tx,
+    ));
+
     established_loop(
         &mut reader,
         &mut writer,
@@ -143,6 +162,7 @@ pub async fn run_session(
         negotiated_hold_time,
         peer_id,
         &config.adj_rib_in,
+        &mut outbound_rx,
     )
     .await
 }
@@ -154,14 +174,15 @@ async fn established_loop(
     hold_time_secs: u16,
     peer_id: PeerId,
     adj_rib_in: &AdjRibIn,
+    outbound: &mut mpsc::Receiver<Message>,
 ) -> Result<()> {
-    // When hold_time is 0, keepalives and the hold timer are disabled (RFC 4271 §4.4).
+    // When hold_time is 0, keepalives and the hold timer are disabled
-    if hold_time_secs == 0 {
+    // (RFC 4271 §4.4). The timer branches below are gated off in that case;
-        return no_timer_loop(reader, writer, ctx, peer_id, adj_rib_in).await;
+    // the placeholder durations are never observed.
-    }
+    let timers_enabled = hold_time_secs != 0;
-
+    let secs = if timers_enabled { hold_time_secs as u64 } else { 3 };
-    let hold_duration = Duration::from_secs(hold_time_secs as u64);
+    let hold_duration = Duration::from_secs(secs);
-    let keepalive_interval = Duration::from_secs(hold_time_secs as u64 / 3);
+    let keepalive_interval = Duration::from_secs(secs / 3);
 
     // interval_at avoids firing immediately; first keepalive is sent after one interval.
     let mut keepalive_timer = interval_at(Instant::now() + keepalive_interval, keepalive_interval);
@@ -178,6 +199,7 @@ async fn established_loop(
                         hold_timer.as_mut().reset(Instant::now() + hold_duration);
                     }
                     Message::Update(update) => {
+                        hold_timer.as_mut().reset(Instant::now() + hold_duration);
                         info!(
                             peer = %peer_id,
                             withdrawn = update.withdrawn_routes.len(),
@@ -195,13 +217,26 @@ async fn established_loop(
                         send_notification(writer, BgpError::FsmError(FsmSubcode::UnexpectedEstablished), ctx).await;
                         bail!("Received OPEN in Established state from {}", peer_id);
                     }
-                    Message::RouteRefresh => {}
+                    Message::RouteRefresh => {
+                        hold_timer.as_mut().reset(Instant::now() + hold_duration);
+                    }
                 }
             }
-            _ = keepalive_timer.tick() => {
+            Some(msg) = outbound.recv() => {
+                if let Message::Update(update) = &msg {
+                    info!(
+                        peer = %peer_id,
+                        withdrawn = update.withdrawn_routes.len(),
+                        announced = update.announced_routes.len(),
+                        "Sending UPDATE",
+                    );
+                }
+                send_message(writer, &msg, ctx).await?;
+            }
+            _ = keepalive_timer.tick(), if timers_enabled => {
                 send_message(writer, &Message::KeepAlive, ctx).await?;
             }
-            _ = &mut hold_timer => {
+            _ = &mut hold_timer, if timers_enabled => {
                 warn!(peer = %peer_id, "Hold timer expired");
                 send_notification(writer, BgpError::HoldTimer, ctx).await;
                 bail!("Hold timer expired for peer {}", peer_id);
@@ -210,45 +245,6 @@ async fn established_loop(
     }
 }
 
-/// Hold-timer-disabled path for hold_time == 0.
-async fn no_timer_loop(
-    reader: &mut OwnedReadHalf,
-    writer: &mut OwnedWriteHalf,
-    ctx: &ParserContext,
-    peer_id: PeerId,
-    adj_rib_in: &AdjRibIn,
-) -> Result<()> {
-    loop {
-        match read_message(reader, ctx).await? {
-            Message::KeepAlive => {}
-            Message::Update(update) => {
-                info!(
-                    peer = %peer_id,
-                    withdrawn = update.withdrawn_routes.len(),
-                    announced = update.announced_routes.len(),
-                    "Received UPDATE",
-                );
-                apply_update(adj_rib_in, peer_id, &update.path_attributes,
-                    &update.withdrawn_routes, &update.announced_routes).await;
-            }
-            Message::Notification(notif) => {
-                info!(peer = %peer_id, error = %notif.error, "Received NOTIFICATION, closing session");
-                return Ok(());
-            }
-            Message::Open(_) => {
-                send_notification(
-                    writer,
-                    BgpError::FsmError(FsmSubcode::UnexpectedEstablished),
-                    ctx,
-                )
-                .await;
-                bail!("Received OPEN in Established state from {}", peer_id);
-            }
-            Message::RouteRefresh => {}
-        }
-    }
-}
-
 // ---------------------------------------------------------------------------
 // Wire helpers
 // ---------------------------------------------------------------------------

crates/server/src/lib.rs

@@ -1,7 +1,9 @@
+pub mod advertise;
 pub mod config;
 pub mod connection;
 pub mod fsm;
 pub mod server;
+pub mod session;
 pub mod tracing;
 
 pub mod proto {

crates/server/src/server.rs

@@ -7,11 +7,13 @@ use futures::stream::{FuturesUnordered, StreamExt};
 use tokio::net::{TcpListener, TcpStream};
 use tracing::{error, info};
 
-use bgp_rib::AdjRibIn;
+use bgp_rib::{AdjRibIn, LocalRib};
 
+use crate::advertise::spawn_advertiser;
 use crate::config::Config;
 use crate::connection::handle_connection;
 use crate::fsm::SessionConfig;
+use crate::session::SessionRegistry;
 
 pub struct BgpServerBuilder {
     config: Config,
@@ -25,6 +27,10 @@ pub struct BgpServer {
     /// Shared RIB handle — created at bind time so callers can clone it before
     /// the accept loop starts (e.g. tests that want to introspect the RIB).
     pub adj_rib_in: AdjRibIn,
+    /// Routes this router originates; announced to every established peer.
+    pub local_rib: LocalRib,
+    /// Live view of established sessions.
+    pub registry: SessionRegistry,
 }
 
 impl BgpServerBuilder {
@@ -46,9 +52,11 @@ impl BgpServerBuilder {
             listeners.push((bound_addr, listener));
         }
         Ok(BgpServer {
-            config: self.config,
             listeners,
             adj_rib_in: AdjRibIn::spawn(),
+            local_rib: LocalRib::spawn(self.config.local_route_policy.clone()),
+            registry: SessionRegistry::new(),
+            config: self.config,
         })
     }
 
@@ -77,8 +85,15 @@ impl BgpServer {
             local_asn: self.config.local_asn,
             hold_time_secs: self.config.hold_time_secs,
             adj_rib_in: self.adj_rib_in.clone(),
+            registry: self.registry.clone(),
         });
 
+        let _advertiser = spawn_advertiser(
+            self.local_rib.clone(),
+            self.registry.clone(),
+            self.config.local_asn,
+        );
+
         let mut accepting: FuturesUnordered<_> = self
             .listeners
             .into_iter()

crates/server/src/session.rs

@@ -0,0 +1,118 @@
+//! Registry of established BGP sessions.
+//!
+//! A session registers itself when it reaches Established and is removed when
+//! its [`SessionGuard`] drops, so a session that errors out is always
+//! deregistered. Other parts of the server (the advertiser, the HTTP API) use
+//! the registry to enumerate live sessions and queue outbound messages.
+
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::sync::{Arc, RwLock};
+use std::time::SystemTime;
+
+use bgp_packet::message::Message;
+use bgp_rib::PeerId;
+use tokio::sync::{broadcast, mpsc};
+
+const EVENT_CHANNEL_CAPACITY: usize = 256;
+
+/// Handle to one established session. Cloning is cheap.
+#[derive(Debug, Clone)]
+pub struct SessionHandle {
+    pub peer: PeerId,
+    /// Our side of the TCP connection; used as the next hop for routes we
+    /// originate to this peer.
+    pub local_addr: SocketAddr,
+    pub established_at: SystemTime,
+    /// Whether the peer negotiated RFC 6793 four-octet ASNs. Determines how
+    /// AS_PATH is encoded on the wire for this session.
+    pub four_octet_asn: bool,
+    outbound: mpsc::Sender<Message>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+#[error("session with {0} has closed")]
+pub struct SessionClosed(pub PeerId);
+
+impl SessionHandle {
+    pub fn new(
+        peer: PeerId,
+        local_addr: SocketAddr,
+        four_octet_asn: bool,
+        outbound: mpsc::Sender<Message>,
+    ) -> Self {
+        Self { peer, local_addr, established_at: SystemTime::now(), four_octet_asn, outbound }
+    }
+
+    /// Queues a message for transmission on this session. Waits for queue
+    /// space if the session's writer is behind.
+    pub async fn send(&self, msg: Message) -> Result<(), SessionClosed> {
+        self.outbound.send(msg).await.map_err(|_| SessionClosed(self.peer))
+    }
+}
+
+#[derive(Debug, Clone)]
+pub enum SessionEvent {
+    Up(SessionHandle),
+    Down(PeerId),
+}
+
+#[derive(Clone)]
+pub struct SessionRegistry {
+    inner: Arc<RwLock<HashMap<PeerId, SessionHandle>>>,
+    events: broadcast::Sender<SessionEvent>,
+}
+
+impl Default for SessionRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SessionRegistry {
+    pub fn new() -> Self {
+        let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
+        Self { inner: Arc::new(RwLock::new(HashMap::new())), events }
+    }
+
+    /// Adds a session and emits [`SessionEvent::Up`]. The returned guard
+    /// removes the session again when dropped.
+    pub fn register(&self, handle: SessionHandle) -> SessionGuard {
+        let peer = handle.peer;
+        self.inner.write().expect("session registry poisoned").insert(peer, handle.clone());
+        let _ = self.events.send(SessionEvent::Up(handle));
+        SessionGuard { registry: self.clone(), peer }
+    }
+
+    pub fn get(&self, peer: &PeerId) -> Option<SessionHandle> {
+        self.inner.read().expect("session registry poisoned").get(peer).cloned()
+    }
+
+    /// All currently established sessions, in no particular order.
+    pub fn list(&self) -> Vec<SessionHandle> {
+        self.inner.read().expect("session registry poisoned").values().cloned().collect()
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<SessionEvent> {
+        self.events.subscribe()
+    }
+
+    fn remove(&self, peer: &PeerId) {
+        let removed = self.inner.write().expect("session registry poisoned").remove(peer);
+        if removed.is_some() {
+            let _ = self.events.send(SessionEvent::Down(*peer));
+        }
+    }
+}
+
+/// Removes the session from its registry on drop.
+pub struct SessionGuard {
+    registry: SessionRegistry,
+    peer: PeerId,
+}
+
+impl Drop for SessionGuard {
+    fn drop(&mut self) {
+        self.registry.remove(&self.peer);
+    }
+}

docs/RFCs.md

@@ -27,7 +27,10 @@ The base protocol specification.
 | §8 | FSM — passive (inbound) side: OpenSent → OpenConfirm → Established | ✅ |
 | §8 | FSM — active (outbound) side: Idle → Connect → Active | ❌ |
 | §9 | Decision process / best-path selection | ❌ |
-| §9 | RIB (Adj-RIB-In, Loc-RIB, Adj-RIB-Out) | ❌ |
+| §9 | Adj-RIB-In (per-peer table of received routes) | ✅ |
+| §9 | Locally originated routes, advertised to all established peers | ✅ |
+| §9 | Loc-RIB / Adj-RIB-Out for routes learned from peers (re-advertisement) | ❌ |
+| §9.2 | Hold timer reset on any received message (not only KEEPALIVE) | ✅ |
 
 ---
 
@@ -122,6 +125,7 @@ https://datatracker.ietf.org/doc/html/rfc1997
 | Feature | Status |
 |---------|--------|
 | `COMMUNITIES` path attribute (parse, serialise) | ✅ |
+| Attach to locally originated routes | ✅ |
 
 ### RFC 4360 — BGP Extended Communities Attribute
 https://datatracker.ietf.org/doc/html/rfc4360
@@ -137,6 +141,7 @@ https://datatracker.ietf.org/doc/html/rfc8092
 | Feature | Status |
 |---------|--------|
 | `LARGE_COMMUNITIES` path attribute (parse, serialise) | ✅ |
+| Attach to locally originated routes | ✅ |
 
 ---
 

testdata/server.textpb

@@ -8,3 +8,8 @@ listen_endpoints {
     address: "[::1]"
     port: 179
 }
+local_route_policy {
+    allowed_prefixes: "10.100.0.0/16"
+    allowed_prefixes: "2001:db8:100::/48"
+    max_routes_per_owner: 4
+}