ferrix/crates/server/src/advertise.rs

//! 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, AsPathSegment, CommunitiesPathAttribute, LargeCommunitiesPathAttribute,
    Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, MultiExitDiscPathAttribute,
    NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage,
};
use bgp_primitives::constants::{AddressFamilyId, Afi, Safi};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute, RibStopped};
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))
}

/// Runs until the local RIB stops. The server watches for this task ending:
/// without it nothing is announced.
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");
                    if resync_all(&local_rib, &registry, local_asn).await.is_err() {
                        return;
                    }
                }
                Err(RecvError::Closed) => return,
            },
            ev = session_events.recv() => match ev {
                Ok(SessionEvent::Up(session)) => {
                    let Ok(routes) = local_rib.snapshot().await else { return };
                    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::RefreshRequested(session)) => {
                    let Ok(routes) = local_rib.snapshot().await else { return };
                    info!(peer = %session.peer, routes = routes.len(), "Route refresh requested, re-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");
                    if resync_all(&local_rib, &registry, local_asn).await.is_err() {
                        return;
                    }
                }
                Err(RecvError::Closed) => return,
            },
        }
    }
}

async fn resync_all(
    local_rib: &LocalRib,
    registry: &SessionRegistry,
    local_asn: u32,
) -> Result<(), RibStopped> {
    let routes = local_rib.snapshot().await?;
    for session in registry.list() {
        for route in &routes {
            send_route(&session, route, local_asn).await;
        }
    }
    Ok(())
}

async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32) {
    if !session.families.carries(&route.prefix) {
        debug!(
            peer = %session.peer,
            prefix = %route.prefix,
            "Skipping route: its address family was not negotiated"
        );
        return;
    }
    match build_announce(
        route,
        local_asn,
        session.local_addr.ip(),
        session.four_octet_asn,
    ) {
        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) {
    if !session.families.carries(prefix) {
        // It was never announced on this session either.
        return;
    }
    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.
///
/// `four_octet_asn` is whether the session the UPDATE is for negotiated
/// four-octet AS numbers. If it did not, and the local AS number needs four
/// octets, the path is also sent as AS4_PATH so that it survives the
/// two-octet speaker (RFC 6793 §4.2.2).
pub fn build_announce(
    route: &LocalRoute,
    local_asn: u32,
    next_hop: IpAddr,
    four_octet_asn: bool,
) -> Option<UpdateMessage> {
    let attrs = &route.attrs;

    let as_path = AsPathAttribute {
        segments: vec![AsPathSegment {
            ordered: true,
            path: vec![local_asn; 1 + usize::from(attrs.prepend)],
        }],
    };
    let as4_path = if four_octet_asn {
        None
    } else {
        as_path.as4_path()
    };

    // Path attributes in ascending type-code order, as most implementations
    // expect them.
    let mut path_attributes = vec![
        PathAttribute::Origin(attrs.origin),
        PathAttribute::AsPath(as_path),
    ];

    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);
        }
        (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: Afi::IPV6,
            safi: Safi::UNICAST,
            next_hop: NlriNextHop::Ipv6(nh),
            prefixes: vec![route.prefix],
        }));
    }
    path_attributes.extend(as4_path);
    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],
            path_attributes: vec![],
            announced_routes: vec![],
        },
        AddressFamilyId::Ipv6 => UpdateMessage {
            withdrawn_routes: vec![],
            path_attributes: vec![PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute {
                afi: Afi::IPV6,
                safi: Safi::UNICAST,
                prefixes: vec![*prefix],
            })],
            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"),
            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)), true).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),
            ibgp: false,
        };
        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"), LocalRouteAttrs::default());
        let nh = Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1);
        let update = build_announce(&route, 65000, IpAddr::V6(nh), true).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")]
        )));
    }

    /// A two-octet session cannot carry a four-octet AS number in AS_PATH:
    /// it goes out as AS_TRANS there, and in full in AS4_PATH.
    #[test]
    fn a_wide_as_number_reaches_a_two_octet_peer_in_as4_path() {
        let route = LocalRoute::new(
            prefix("10.100.1.0/24"),
            LocalRouteAttrs {
                prepend: 1,
                ..LocalRouteAttrs::default()
            },
        );
        let next_hop = IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1));
        let path = |update: &UpdateMessage, as4: bool| {
            update.path_attributes.iter().find_map(|a| match a {
                PathAttribute::As4Path(p) if as4 => Some(p.segments[0].path.clone()),
                PathAttribute::AsPath(p) if !as4 => Some(p.segments[0].path.clone()),
                _ => None,
            })
        };

        let update = build_announce(&route, 4_200_000_001, next_hop, false).unwrap();
        assert_eq!(path(&update, true), Some(vec![4_200_000_001; 2]));

        // What the peer reads from the wire.
        let ctx = ParserContext {
            four_octet_asn: Some(false),
            ibgp: false,
        };
        let mut buf = BytesMut::new();
        Message::Update(update).to_wire(&ctx, &mut buf).unwrap();
        let (_, Message::Update(seen)) = Message::from_wire(&ctx, &buf).unwrap() else {
            panic!("expected UPDATE")
        };
        assert_eq!(path(&seen, false), Some(vec![23456; 2]), "AS_TRANS");
        assert_eq!(path(&seen, true), Some(vec![4_200_000_001; 2]));

        // No AS4_PATH when it would say nothing new: the peer handles
        // four-octet numbers, or ours fits in two.
        let update = build_announce(&route, 4_200_000_001, next_hop, true).unwrap();
        assert_eq!(path(&update, true), None);
        let update = build_announce(&route, 65000, next_hop, false).unwrap();
        assert_eq!(path(&update, true), None);
        assert_eq!(path(&update, false), Some(vec![65000; 2]));
    }

    #[test]
    fn family_mismatch_is_skipped() {
        let v6 = LocalRoute::new(prefix("2001:db8::/32"), LocalRouteAttrs::default());
        assert!(build_announce(&v6, 65000, IpAddr::V4(Ipv4Addr::LOCALHOST), true).is_none());
        let v4 = LocalRoute::new(prefix("10.0.0.0/8"), LocalRouteAttrs::default());
        assert!(build_announce(&v4, 65000, IpAddr::V6(Ipv6Addr::LOCALHOST), true).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")])
        );
    }
}