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::{ServerWithPeer, TestServer};
use bgp_integration_tests::{TIMEOUT, peer_config, prefix, wait_for, wait_until};
use bgp_packet::errors::{BgpError, OpenMsgSubcode};
use bgp_packet::message::Message;
use bgp_rib::{LocalRoute, LocalRouteAttrs};
use bgp_server::config::PeerConfig;
use bgp_server::policy::ImportPolicy;
use bgp_server::session::DialState;
use eyre::Result;

const LOCAL_ASN: u32 = 65000;
const PEER_ASN: u32 = 65001;
const PEER_ID: Ipv4Addr = Ipv4Addr::new(10, 0, 0, 2);

/// A server that dials its peer "r1", in `remote_asn`, every 200 ms.
async fn server_dialing(remote_asn: u32) -> Result<ServerWithPeer> {
    server_with_peer(remote_asn, false).await
}

/// A server with peer "r1" in `remote_asn`, which it dials every 200 ms
/// unless the peer is `passive`.
async fn server_with_peer(remote_asn: u32, passive: bool) -> Result<ServerWithPeer> {
    TestServer::with_peer(LOCAL_ASN, remote_asn, |c, peer| {
        c.connect_retry = Duration::from_millis(200);
        peer.import = ImportPolicy::permissive();
        peer.passive = passive;
    })
    .await
}

#[tokio::test]
async fn test_server_dials_configured_peer_and_establishes() -> Result<()> {
    let ServerWithPeer { server, listener } = server_dialing(PEER_ASN).await?;

    let mut peer = TestPeer::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, PEER_ID, 90).await?;
    assert_eq!(open.asn, LOCAL_ASN as u16);

    let listed = server.wait_for_session().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(
            prefix("10.100.1.0/24"),
            LocalRouteAttrs::default(),
        ))
        .await?;
    let update = 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 ServerWithPeer { server, listener } = server_dialing(PEER_ASN).await?;

    let mut first = TestPeer::accept(&listener).await?;
    first.handshake(PEER_ASN, PEER_ID, 90).await?;
    server.wait_for_session().await?;
    drop(first);

    // Session goes away, then the dialer comes back.
    server.wait_for_sessions(0).await?;
    let (_, state) = server
        .registry
        .configured_peers()
        .into_iter()
        .next()
        .unwrap();
    assert!(matches!(state, DialState::Backoff { .. }), "{state:?}");

    let mut second = TestPeer::accept(&listener).await?;
    second.handshake(PEER_ASN, PEER_ID, 90).await?;
    server.wait_for_session().await?;
    Ok(())
}

#[tokio::test]
async fn test_wrong_asn_is_refused_with_bad_peer_as() -> Result<()> {
    let ServerWithPeer {
        server: _server,
        listener,
    } = server_dialing(PEER_ASN).await?;

    let mut peer = TestPeer::accept(&listener).await?;
    peer.send_open(65099, PEER_ID, 90).await?;
    let _ = peer.recv_open().await?;
    let notif = peer.expect_notification().await?;
    assert!(
        matches!(notif.error, BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)),
        "{notif:?}"
    );

    // The dialer retries rather than giving up.
    let _again = TestPeer::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);
        let at = SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1);
        c.peers = vec![PeerConfig {
            import: ImportPolicy::permissive(),
            ..peer_config("r1", at, PEER_ASN)
        }];
    })
    .await?;

    // Wrong ASN inbound → refused.
    let mut bad = TestPeer::connect(server.addr).await?;
    bad.send_open(65099, PEER_ID, 90).await?;
    let _ = bad.recv_open().await?;
    let msg = 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, PEER_ID, 90).await?;
    let listed = server.wait_for_session().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 ServerWithPeer {
        server: _server,
        listener,
    } = TestServer::with_peer(LOCAL_ASN, PEER_ASN, |c, peer| {
        c.connect_retry = Duration::from_millis(200);
        peer.import = ImportPolicy::permissive();
        peer.local_address = Some(Ipv4Addr::LOCALHOST.into());
    })
    .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 endpoint = listener.local_addr()?;
    let server = TestServer::start_with(LOCAL_ASN, |c| {
        c.connect_retry = Duration::from_millis(200);
        c.peers = vec![PeerConfig {
            local_address: Some(Ipv4Addr::new(192, 0, 2, 1).into()),
            import: ImportPolicy::permissive(),
            ..peer_config("r1", endpoint, PEER_ASN)
        }];
    })
    .await?;
    let reason = wait_for("the dialer is backing off", || {
        match server.registry.configured_peers().into_iter().next() {
            Some((_, DialState::Backoff { last_error, .. })) => Some(last_error),
            _ => None,
        }
    })
    .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 ServerWithPeer { server, listener } = server_dialing(PEER_ASN).await?;

    let mut waits = Vec::new();
    for _ in 0..3 {
        let mut peer = TestPeer::accept(&listener).await?;
        peer.send_open(65099, PEER_ID, 90).await?;
        let _ = peer.recv_open().await?;
        let refusal = 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 = wait_for("the dialer is backing off", || {
            match server.registry.configured_peers().into_iter().next() {
                Some((_, DialState::Backoff { until, last_error })) => {
                    assert!(last_error.contains("AS65099"), "{last_error}");
                    Some(
                        until
                            .duration_since(std::time::SystemTime::now())
                            .unwrap_or_default(),
                    )
                }
                _ => None,
            }
        })
        .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(())
}

/// A passive peer is not dialled: its session comes up when it connects,
/// and until then, and after, it is shown as waiting, with how its last
/// session ended.
#[tokio::test]
async fn test_a_passive_peer_is_not_dialled_and_comes_up_when_it_connects() -> Result<()> {
    let ServerWithPeer { server, listener } = server_with_peer(PEER_ASN, true).await?;
    let state = || server.registry.configured_peers()[0].1.clone();

    // Many retry intervals pass, and nothing connects.
    assert!(
        tokio::time::timeout(Duration::from_secs(1), listener.accept())
            .await
            .is_err(),
        "a passive peer was dialled"
    );
    assert_eq!(state(), DialState::Passive { last_error: None });

    let mut peer = TestPeer::connect(server.addr).await?;
    peer.handshake(PEER_ASN, PEER_ID, 90).await?;
    let listed = server.wait_for_session().await?;
    assert_eq!(listed.name.as_deref(), Some("r1"));

    drop(peer);
    wait_until("the peer is shown as waiting, with how it ended", || {
        matches!(
            state(),
            DialState::Passive {
                last_error: Some(_)
            }
        )
    })
    .await?;
    assert!(server.registry.list().is_empty());
    assert!(
        tokio::time::timeout(Duration::from_millis(600), listener.accept())
            .await
            .is_err(),
        "a passive peer was dialled after its session ended"
    );
    Ok(())
}