//! Route origin validation, end to end: a server that validates against a
//! cache, a peer that announces routes, and the validation data changing
//! under both.

use std::collections::BTreeSet;
use std::net::Ipv4Addr;
use std::time::Duration;

use bgp_integration_tests::default_policy;
use bgp_integration_tests::test_cache::TestCache;
use bgp_integration_tests::test_peer::TestPeer;
use bgp_integration_tests::test_server::TestServer;
use bgp_packet::message::{
    AsPathAttribute, Message, NextHopPathAttribute, OriginPathAttribute, PathAttribute,
    UpdateMessage,
};
use bgp_primitives::ip_prefix::IpPrefix;
use bgp_rpki::{CacheState, RpkiConfig, Vrp};
use bgp_server::config::PeerConfig;
use bgp_server::family::FamilySet;
use bgp_server::policy::ImportPolicy;
use eyre::Result;
use serde_json::Value;
use tokio::net::TcpListener;

const LOCAL_ASN: u32 = 1299;
const PEER_ASN: u32 = 3333;
const TIMEOUT: Duration = Duration::from_secs(5);

fn prefix(s: &str) -> IpPrefix {
    IpPrefix::try_from(s).unwrap()
}

fn vrp(p: &str, max_length: u8, asn: u32) -> Vrp {
    Vrp {
        prefix: prefix(p),
        max_length,
        asn,
    }
}

/// What a test holds on to: the server, and the listener its dialling
/// goes to.
struct Setup {
    server: TestServer,
    _listener: TcpListener,
}

/// A server that validates against `cache`, with one configured peer at
/// the address the test connects from. Returns once the server has the
/// cache's data.
async fn server_with(
    cache: &TestCache,
    expire: Option<Duration>,
    import: ImportPolicy,
) -> Result<Setup> {
    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?;
    let endpoint = listener.local_addr()?;
    let cache = cache.address();
    let server = TestServer::start_with(LOCAL_ASN, |c| {
        c.accept_unconfigured_peers = false;
        c.rpki = Some(RpkiConfig {
            caches: vec![cache],
            // Data never expires sooner than a refresh and a retry take.
            refresh: expire.map(|expire| expire / 2),
            retry: Some(Duration::from_millis(100)),
            expire,
        });
        c.peers = vec![PeerConfig {
            name: "r1".into(),
            endpoint: endpoint.into(),
            remote_asn: PEER_ASN,
            hold_time_secs: 90,
            local_address: None,
            families: FamilySet::supported(),
            add_path: None,
            import,
            role: None,
            require_role: false,
        }];
    })
    .await?;
    let rpki = server.rpki.clone().expect("validation is configured");
    tokio::time::timeout(TIMEOUT, async {
        while rpki.status().state != CacheState::Synchronised {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    })
    .await?;
    Ok(Setup {
        server,
        _listener: listener,
    })
}

async fn established(server: &TestServer) -> Result<TestPeer> {
    let mut peer = TestPeer::connect(server.addr).await?;
    peer.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(5)).await;
        }
    })
    .await?;
    Ok(peer)
}

/// Announces `prefixes` as originated by `origin`, reached through the
/// peer.
async fn announce(peer: &mut TestPeer, prefixes: &[&str], origin: u32) -> Result<()> {
    let path = if origin == PEER_ASN {
        vec![PEER_ASN]
    } else {
        vec![PEER_ASN, origin]
    };
    peer.send(&Message::Update(UpdateMessage {
        withdrawn_routes: vec![],
        path_attributes: vec![
            PathAttribute::Origin(OriginPathAttribute::IGP),
            AsPathAttribute::from_asns(path),
            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
        ],
        announced_routes: prefixes.iter().map(|p| prefix(p).into()).collect(),
    }))
    .await
}

fn set(prefixes: &[&str]) -> BTreeSet<String> {
    prefixes.iter().map(|p| p.to_string()).collect()
}

/// Waits until the table holds exactly `expected` and the peer has `held`
/// routes held aside.
async fn table_becomes(server: &TestServer, expected: &[&str], held: u64) -> Result<()> {
    let expected = set(expected);
    let table = || -> BTreeSet<String> {
        server
            .received_routes()
            .iter()
            .map(|route| route.prefix.to_string())
            .collect()
    };
    let held_now = || {
        server
            .registry
            .list()
            .first()
            .map_or(0, |session| session.counters.held())
    };
    tokio::time::timeout(TIMEOUT, async {
        while table() != expected || held_now() != held {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    })
    .await
    .map_err(|_| {
        eyre::eyre!(
            "the table has {:?} with {} held; expected {expected:?} with {held} held",
            table(),
            held_now()
        )
    })
}

/// A route whose origin is not the one the holder of the addresses
/// authorised is refused. One that is authorised is taken, and so is one
/// that nothing is known about.
#[tokio::test]
async fn test_an_invalid_route_is_refused_and_the_rest_are_taken() -> Result<()> {
    let cache = TestCache::start(vec![vrp("193.0.0.0/21", 24, PEER_ASN)]).await;
    let Setup { server, .. } = server_with(&cache, None, default_policy()).await?;
    let (base, _http) = server.serve_http().await?;
    let mut peer = established(&server).await?;

    announce(&mut peer, &["193.0.0.0/21", "194.0.0.0/24"], PEER_ASN).await?;
    // Someone else's AS as the origin of a covered prefix; and the right
    // AS, more specific than it may announce.
    announce(&mut peer, &["193.0.4.0/24"], 15169).await?;
    announce(&mut peer, &["193.0.6.0/25"], PEER_ASN).await?;
    table_becomes(&server, &["193.0.0.0/21", "194.0.0.0/24"], 1).await?;

    let client = reqwest::Client::new();
    let get = |path: &str| {
        let request = client.get(format!("{base}{path}"));
        async move { Ok::<Value, eyre::Report>(request.send().await?.error_for_status()?.json().await?) }
    };
    // Each route says how it stands.
    let routes = get("/api/routes/received").await?["routes"].take();
    let validity: Vec<(&str, &str)> = routes
        .as_array()
        .unwrap()
        .iter()
        .map(|r| (r["prefix"].as_str().unwrap(), r["rpki"].as_str().unwrap()))
        .collect();
    assert_eq!(
        validity,
        [("193.0.0.0/21", "valid"), ("194.0.0.0/24", "not found")]
    );
    // The refused one is on record with its reason. The too-specific one
    // was refused before validation came to it.
    let refused = get("/api/routes/rejected").await?;
    assert_eq!(
        refused[0]["by_reason"],
        serde_json::json!({"RPKI invalid": 1, "prefix too specific": 1})
    );
    let peers = get("/api/peers").await?;
    assert_eq!(peers[0]["routes"], 2);
    assert_eq!(peers[0]["held_routes"], 1);

    let rpki = get("/api/rpki").await?;
    assert_eq!(rpki["enabled"], true);
    assert_eq!(rpki["state"], "synchronised");
    assert_eq!(rpki["vrps_ipv4"], 1);
    assert_eq!(rpki["cache"], cache.address().to_string());
    Ok(())
}

/// Validity changes with the validation data, and the routes follow: one
/// that was taken goes when it becomes invalid, and one that was refused
/// as invalid is taken when it no longer is. The peer sends nothing more
/// meanwhile; it has said all it has to say.
#[tokio::test]
async fn test_routes_follow_the_validation_data() -> Result<()> {
    let ripe = vrp("193.0.0.0/21", 24, PEER_ASN);
    let cache = TestCache::start(vec![ripe]).await;
    let Setup { server, .. } = server_with(&cache, None, default_policy()).await?;
    let mut peer = established(&server).await?;

    // Nothing is known about 8.8.8.0/24, so it is taken. 193.0.4.0/24 is
    // covered, and not for this origin.
    announce(&mut peer, &["8.8.8.0/24", "193.0.4.0/24"], 15169).await?;
    table_becomes(&server, &["8.8.8.0/24"], 1).await?;

    // The holder of 8.8.8.0/24 authorises another AS: the route in the
    // table is now invalid, and goes.
    cache.set(vec![ripe, vrp("8.8.8.0/24", 24, 13335)]);
    table_becomes(&server, &[], 2).await?;
    let refused = &server.refusals.peers()[0].1;
    assert_eq!(refused.total(), 2);
    assert_eq!(refused.recent()[0].prefix, prefix("8.8.8.0/24"));
    assert_eq!(&refused.recent()[0].as_path[..], [PEER_ASN, 15169]);

    // Both are authorised for the origin they have: both come back, with
    // the attributes they were announced with.
    cache.set(vec![
        ripe,
        vrp("8.8.8.0/24", 24, 15169),
        vrp("193.0.4.0/24", 24, 15169),
    ]);
    table_becomes(&server, &["8.8.8.0/24", "193.0.4.0/24"], 0).await?;
    for route in server.received_routes() {
        assert!(
            route
                .attributes
                .contains(&AsPathAttribute::from_asns(vec![PEER_ASN, 15169])),
            "{route:?}"
        );
    }
    assert_eq!(server.registry.list()[0].counters.routes(), 2);
    Ok(())
}

/// A held route is the peer's to withdraw or replace like any other. One
/// it has withdrawn does not appear when it would have become valid.
#[tokio::test]
async fn test_a_held_route_the_peer_withdrew_does_not_come_back() -> Result<()> {
    let ripe = vrp("193.0.0.0/21", 24, PEER_ASN);
    let cache = TestCache::start(vec![ripe]).await;
    let Setup { server, .. } = server_with(&cache, None, default_policy()).await?;
    let mut peer = established(&server).await?;

    announce(&mut peer, &["193.0.4.0/24", "193.0.5.0/24"], 15169).await?;
    table_becomes(&server, &[], 2).await?;

    // One is withdrawn, the other announced again with the right origin.
    peer.send(&Message::Update(UpdateMessage {
        withdrawn_routes: vec![prefix("193.0.4.0/24").into()],
        path_attributes: vec![],
        announced_routes: vec![],
    }))
    .await?;
    announce(&mut peer, &["193.0.5.0/24"], PEER_ASN).await?;
    table_becomes(&server, &["193.0.5.0/24"], 0).await?;

    // Now the old origin would be valid for both. Neither old route is
    // there to be brought back.
    cache.set(vec![ripe, vrp("193.0.4.0/23", 24, 15169)]);
    tokio::time::sleep(Duration::from_millis(500)).await;
    table_becomes(&server, &["193.0.5.0/24"], 0).await?;
    assert!(
        server.received_routes()[0]
            .attributes
            .contains(&AsPathAttribute::from_asns(vec![PEER_ASN])),
        "the route the peer has now"
    );
    Ok(())
}

/// A peer can be exempt: its invalid routes are taken, and shown for what
/// they are.
#[tokio::test]
async fn test_a_peer_can_be_exempt_from_validation() -> Result<()> {
    let cache = TestCache::start(vec![vrp("193.0.0.0/21", 24, PEER_ASN)]).await;
    let import = ImportPolicy {
        accept_rpki_invalid: true,
        ..default_policy()
    };
    let Setup { server, .. } = server_with(&cache, None, import).await?;
    let (base, _http) = server.serve_http().await?;
    let mut peer = established(&server).await?;

    announce(&mut peer, &["193.0.4.0/24"], 15169).await?;
    table_becomes(&server, &["193.0.4.0/24"], 0).await?;
    let routes: Value = reqwest::get(format!("{base}/api/routes/received"))
        .await?
        .json()
        .await?;
    assert_eq!(routes["routes"][0]["rpki"], "invalid");
    assert!(server.refusals.peers().is_empty());
    Ok(())
}

/// Validation data that no cache has confirmed for too long is dropped,
/// and with it every verdict: routes held as invalid are taken. Refusing
/// routes on the word of data that may be long out of date would be worse.
#[tokio::test]
async fn test_held_routes_are_taken_when_the_validation_data_expires() -> Result<()> {
    let cache = TestCache::start(vec![vrp("193.0.0.0/21", 24, PEER_ASN)]).await;
    let Setup { server, .. } =
        server_with(&cache, Some(Duration::from_millis(1500)), default_policy()).await?;
    let mut peer = established(&server).await?;
    announce(&mut peer, &["193.0.4.0/24"], 15169).await?;
    table_becomes(&server, &[], 1).await?;

    drop(cache);
    table_becomes(&server, &["193.0.4.0/24"], 0).await?;
    assert!(server.rpki.as_ref().unwrap().current().vrps.is_empty());
    Ok(())
}

/// A hijack that was possible: on a session with two-octet AS numbers a
/// route has two paths, AS_PATH and AS4_PATH, and a peer can make them
/// say different things. Here AS_PATH is the peer alone, and an AS4_PATH
/// too long to be used ends at the AS that holds the addresses. The route
/// is the peer's and invalid, when it arrives and when it is looked at
/// again; nothing may take the unused path's word for where it comes from.
#[tokio::test]
async fn test_an_unused_as4_path_does_not_make_a_route_valid() -> Result<()> {
    use bgp_packet::message::AsPathSegment;
    const HOLDER: u32 = 15169;
    let cache = TestCache::start(vec![vrp("193.0.0.0/21", 24, HOLDER)]).await;
    let Setup { server, .. } = server_with(&cache, None, default_policy()).await?;

    // No capabilities: AS numbers are two octets wide on this session.
    let mut peer = TestPeer::connect(server.addr).await?;
    peer.handshake_with(PEER_ASN, Ipv4Addr::new(10, 0, 0, 2), 90, vec![])
        .await?;
    tokio::time::timeout(TIMEOUT, async {
        while server.registry.list().is_empty() {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    })
    .await?;
    assert!(!server.registry.list()[0].four_octet_asn);

    peer.send(&Message::Update(UpdateMessage {
        withdrawn_routes: vec![],
        path_attributes: vec![
            PathAttribute::Origin(OriginPathAttribute::IGP),
            AsPathAttribute::from_asns(vec![PEER_ASN]),
            PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))),
            PathAttribute::As4Path(AsPathAttribute {
                segments: vec![AsPathSegment {
                    ordered: true,
                    path: vec![3356, HOLDER],
                }],
            }),
        ],
        announced_routes: vec![prefix("193.0.4.0/24").into()],
    }))
    .await?;
    table_becomes(&server, &[], 1).await?;

    // The data changes, and every held route is looked at again.
    cache.set(vec![
        vrp("193.0.0.0/21", 24, HOLDER),
        vrp("198.51.100.0/24", 24, 64500),
    ]);
    let rpki = server.rpki.clone().unwrap();
    tokio::time::timeout(TIMEOUT, async {
        while rpki.current().vrps.len() != 2 {
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
    })
    .await?;
    // Time enough for the look, and then some.
    tokio::time::sleep(Duration::from_millis(600)).await;
    table_becomes(&server, &[], 1).await?;
    Ok(())
}