ferrix/crates/integration-tests/tests/dialer.rs

//! The active side: the server dials its configured peers and keeps
//! retrying.

use std::net::{Ipv4Addr, SocketAddr};
use std::time::Duration;

use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::errors::{BgpError, OpenMsgSubcode};
use bgp_packet::message::Message;
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rib::{LocalRoute, LocalRouteAttrs};
use bgp_server::config::PeerConfig;
use bgp_server::family::FamilySet;
use bgp_server::policy::ImportPolicy;
use bgp_server::session::DialState;
use eyre::Result;
use tokio::net::TcpListener;

const LOCAL_ASN: u32 = 65000;
const PEER_ASN: u32 = 65001;
const TIMEOUT: Duration = Duration::from_secs(5);

/// Starts a server configured to dial `listener`'s address as peer "r1".
async fn server_dialing(listener: &TcpListener, remote_asn: u32) -> Result<TestServer> {
    let endpoint: SocketAddr = listener.local_addr()?;
    TestServer::start_with(LOCAL_ASN, |c| {
        c.connect_retry = Duration::from_millis(200);
        c.peers = vec![PeerConfig {
            graceful_restart: Default::default(),
            groups: Vec::new(),
            reflector_client: false,
            name: "r1".into(),
            endpoint: endpoint.into(),
            remote_asn,
            hold_time_secs: 90,
            local_address: None,
            families: FamilySet::supported(),
            add_path: None,
            import: ImportPolicy::permissive(),
            role: None,
            require_role: false,
            export: Default::default(),
        }];
    })
    .await
}

async fn accept(listener: &TcpListener) -> Result<TestPeer> {
    let (stream, _) = tokio::time::timeout(TIMEOUT, listener.accept()).await??;
    Ok(TestPeer::from_stream(stream))
}

#[tokio::test]
async fn test_server_dials_configured_peer_and_establishes() -> Result<()> {
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
    let server = server_dialing(&listener, PEER_ASN).await?;

    let mut peer = accept(&listener).await?;
    // Both sides send OPEN on connect; the test peer's handshake order works
    // whichever side opened TCP.
    let open = peer
        .handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90)
        .await?;
    assert_eq!(open.asn, LOCAL_ASN as u16);

    let mut events = server.registry.subscribe();
    // Either the Up event is still queued or the session is already listed.
    let listed = tokio::time::timeout(TIMEOUT, async {
        loop {
            if let Some(s) = server.registry.list().first().cloned() {
                return s;
            }
            let _ = events.recv().await;
        }
    })
    .await?;
    assert_eq!(listed.name.as_deref(), Some("r1"));
    assert_eq!(listed.peer.asn, PEER_ASN);
    assert_eq!(listed.peer.addr, listener.local_addr()?);

    // Configured view reflects it; the dialer is not in backoff.
    let (cfg, state) = server
        .registry
        .configured_peers()
        .into_iter()
        .next()
        .unwrap();
    assert_eq!(cfg.name, "r1");
    assert!(
        matches!(state, DialState::Connecting | DialState::Idle),
        "{state:?}"
    );

    // Local routes flow over the dialled session too.
    server
        .local_rib
        .announce(LocalRoute::new(
            IpPrefix::try_from("10.100.1.0/24")?,
            LocalRouteAttrs::default(),
        ))
        .await?;
    let update = tokio::time::timeout(TIMEOUT, peer.recv_update()).await??;
    assert_eq!(update.announced_routes[0].to_string(), "10.100.1.0/24");
    Ok(())
}

#[tokio::test]
async fn test_server_reconnects_after_peer_closes() -> Result<()> {
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
    let server = server_dialing(&listener, PEER_ASN).await?;

    let mut first = accept(&listener).await?;
    first
        .handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90)
        .await?;
    tokio::time::timeout(TIMEOUT, async {
        while server.registry.list().is_empty() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await?;
    drop(first);

    // Session goes away, then the dialer comes back.
    tokio::time::timeout(TIMEOUT, async {
        while !server.registry.list().is_empty() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await?;
    let (_, state) = server
        .registry
        .configured_peers()
        .into_iter()
        .next()
        .unwrap();
    assert!(matches!(state, DialState::Backoff { .. }), "{state:?}");

    let mut second = accept(&listener).await?;
    second
        .handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90)
        .await?;
    tokio::time::timeout(TIMEOUT, async {
        while server.registry.list().is_empty() {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await?;
    Ok(())
}

#[tokio::test]
async fn test_wrong_asn_is_refused_with_bad_peer_as() -> Result<()> {
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
    let _server = server_dialing(&listener, PEER_ASN).await?;

    let mut peer = accept(&listener).await?;
    peer.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90)
        .await?;
    let _ = peer.recv_open().await?;
    let notif = tokio::time::timeout(TIMEOUT, async {
        loop {
            match peer.recv().await? {
                Message::Notification(n) => return Ok::<_, eyre::Error>(n),
                Message::KeepAlive => {}
                other => eyre::bail!("unexpected {other:?}"),
            }
        }
    })
    .await??;
    assert!(
        matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)),
        "{notif:?}"
    );

    // The dialer retries rather than giving up.
    let _again = accept(&listener).await?;
    Ok(())
}

/// An inbound connection from a configured peer's address is also checked
/// against the configured ASN and gets the configured name.
#[tokio::test]
async fn test_inbound_from_configured_address_uses_config() -> Result<()> {
    // Configure a peer at the loopback address on a port nobody listens on,
    // so the dialer never succeeds; then connect inbound from loopback.
    let server = TestServer::start_with(LOCAL_ASN, |c| {
        c.connect_retry = Duration::from_secs(60);
        c.peers = vec![PeerConfig {
            graceful_restart: Default::default(),
            groups: Vec::new(),
            reflector_client: false,
            name: "r1".into(),
            endpoint: SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1).into(),
            remote_asn: PEER_ASN,
            hold_time_secs: 90,
            local_address: None,
            families: FamilySet::supported(),
            add_path: None,
            import: ImportPolicy::permissive(),
            role: None,
            require_role: false,
            export: Default::default(),
        }];
    })
    .await?;

    // Wrong ASN inbound → refused.
    let mut bad = TestPeer::connect(server.addr).await?;
    bad.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90).await?;
    let _ = bad.recv_open().await?;
    let msg = tokio::time::timeout(TIMEOUT, bad.recv()).await??;
    assert!(
        matches!(&msg, Message::Notification(n) if matches!(n.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs))),
        "{msg:?}"
    );

    // Right ASN inbound → established, named.
    let mut good = TestPeer::connect(server.addr).await?;
    good.handshake(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90)
        .await?;
    let listed = tokio::time::timeout(TIMEOUT, async {
        loop {
            if let Some(s) = server.registry.list().first().cloned() {
                return s;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    })
    .await?;
    assert_eq!(listed.name.as_deref(), Some("r1"));
    Ok(())
}

/// With `local_address` set the connection must come from that address (the
/// peer may identify us by it), and an address the host does not have is
/// reported as the reason for backing off rather than silently ignored.
#[tokio::test]
async fn test_local_address_is_used_as_the_source() -> Result<()> {
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
    let endpoint: SocketAddr = listener.local_addr()?;
    let _server = TestServer::start_with(LOCAL_ASN, |c| {
        c.connect_retry = Duration::from_millis(200);
        c.peers = vec![PeerConfig {
            graceful_restart: Default::default(),
            groups: Vec::new(),
            reflector_client: false,
            name: "r1".into(),
            endpoint: endpoint.into(),
            remote_asn: PEER_ASN,
            hold_time_secs: 90,
            local_address: Some(Ipv4Addr::LOCALHOST.into()),
            families: FamilySet::supported(),
            add_path: None,
            import: ImportPolicy::permissive(),
            role: None,
            require_role: false,
            export: Default::default(),
        }];
    })
    .await?;
    let (_stream, from) = tokio::time::timeout(TIMEOUT, listener.accept()).await??;
    assert_eq!(from.ip(), std::net::IpAddr::from(Ipv4Addr::LOCALHOST));

    // 192.0.2.1 is documentation space; no host has it.
    let server = TestServer::start_with(LOCAL_ASN, |c| {
        c.connect_retry = Duration::from_millis(200);
        c.peers = vec![PeerConfig {
            graceful_restart: Default::default(),
            groups: Vec::new(),
            reflector_client: false,
            name: "r1".into(),
            endpoint: endpoint.into(),
            remote_asn: PEER_ASN,
            hold_time_secs: 90,
            local_address: Some(Ipv4Addr::new(192, 0, 2, 1).into()),
            families: FamilySet::supported(),
            add_path: None,
            import: ImportPolicy::permissive(),
            role: None,
            require_role: false,
            export: Default::default(),
        }];
    })
    .await?;
    let reason = tokio::time::timeout(TIMEOUT, async {
        loop {
            if let Some((_, DialState::Backoff { last_error, .. })) =
                server.registry.configured_peers().into_iter().next()
            {
                return last_error;
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
    })
    .await?;
    assert!(reason.contains("local_address 192.0.2.1"), "{reason}");
    Ok(())
}

/// A peer that takes the connection and then refuses the session, here
/// because it is not the AS we are configured to expect, will refuse it
/// again. It is dialled ever more slowly, like a peer that cannot be
/// reached: connecting used to count as success and reset the wait, so
/// such a peer was dialled every few seconds for ever.
#[tokio::test]
async fn test_a_peer_that_refuses_the_session_is_dialled_ever_more_slowly() -> Result<()> {
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
    let server = server_dialing(&listener, PEER_ASN).await?;

    let mut waits = Vec::new();
    for _ in 0..3 {
        let mut peer = accept(&listener).await?;
        peer.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90)
            .await?;
        let _ = peer.recv_open().await?;
        let refusal = tokio::time::timeout(TIMEOUT, peer.recv()).await??;
        assert!(
            matches!(&refusal, Message::Notification(n) if n.error == BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)),
            "{refusal:?}"
        );
        drop(peer);

        // How long the dialer says it will wait before trying again.
        let wait = tokio::time::timeout(TIMEOUT, async {
            loop {
                if let Some((_, DialState::Backoff { until, last_error })) =
                    server.registry.configured_peers().into_iter().next()
                {
                    assert!(last_error.contains("AS65099"), "{last_error}");
                    return until
                        .duration_since(std::time::SystemTime::now())
                        .unwrap_or_default();
                }
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        })
        .await?;
        waits.push(wait);
    }
    // 200 ms, then 400, then 800, less what passed before each was read.
    assert!(
        waits[2] > Duration::from_millis(500),
        "the waits were {waits:?}"
    );
    Ok(())
}