open-standards/smart-byte/smart-byte-rs/crates/smart-byte-libp2p/tests/gossip_validation.rs

//! A forged envelope stops at the first honest node; a genuine one crosses it.
//!
//! A — B — C, with A and C not connected. B runs the crate's own event loop,
//! which settles every gossip message before gossipsub may forward it. A
//! publishes three messages: an envelope with its cargo changed under the
//! genuine id, the genuine envelope padded with an extra member, and the
//! genuine envelope. Only the last may reach C.

use std::time::Duration;

use libp2p::futures::StreamExt;
use libp2p::gossipsub::{self, IdentTopic};
use libp2p::swarm::SwarmEvent;
use smart_byte_core::{Aid, Cargo, Envelope, JouleCost, Said, Signer, SigningKey};
use smart_byte_libp2p::behaviour::{SmartByteBehaviourEvent, default_gossipsub_config};
use smart_byte_libp2p::transport::{settle_gossip, subscribe_topic};
use smart_byte_libp2p::{Libp2pNode, NodeConfig, NodeEvent, run};
use tokio::time::Instant;

async fn node() -> Libp2pNode {
    Libp2pNode::new(NodeConfig {
        listen_addrs: vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()],
        bootstrap_peers: vec![],
        identity: libp2p::identity::Keypair::generate_ed25519(),
        enable_mdns: false,
        enable_relay: false,
        enable_dcutr: false,
        gossipsub_config: default_gossipsub_config(),
    })
    .await
    .expect("spawn node")
}

/// Poll `node` once, settling any gossip message; return its data if it was one.
fn received(node: &mut Libp2pNode, ev: SwarmEvent<SmartByteBehaviourEvent>) -> Option<Vec<u8>> {
    if let SwarmEvent::Behaviour(SmartByteBehaviourEvent::Gossipsub(gossipsub::Event::Message {
        propagation_source,
        message_id,
        message,
    })) = ev
    {
        settle_gossip(
            &mut node.swarm,
            &message_id,
            &propagation_source,
            &message.data,
        );
        return Some(message.data);
    }
    None
}

#[tokio::test(flavor = "multi_thread")]
async fn a_forged_envelope_is_not_relayed() {
    let (mut a, mut b, mut c) = (node().await, node().await, node().await);
    let b_addr = b.listen_addrs[0].clone();
    a.swarm.dial(b_addr.clone()).expect("dial a -> b");
    c.swarm.dial(b_addr).expect("dial c -> b");
    let topic = "smart-byte/test/relay";
    for n in [&mut a, &mut b, &mut c] {
        subscribe_topic(n, topic).expect("subscribe");
    }

    let (tx, mut rx) = tokio::sync::mpsc::channel::<NodeEvent>(1024);
    tokio::spawn(async move {
        let mut b = b;
        let _ = run(&mut b, tx).await;
    });
    tokio::spawn(async move { while rx.recv().await.is_some() {} });

    // Let the connections settle and the topic mesh form.
    let warm = Instant::now() + Duration::from_secs(4);
    while Instant::now() < warm {
        tokio::select! {
            _ = a.swarm.select_next_some() => {}
            _ = c.swarm.select_next_some() => {}
            _ = tokio::time::sleep(Duration::from_millis(20)) => {}
        }
    }
    assert!(
        !a.swarm.is_connected(&c.local_peer_id),
        "A and C must not be directly connected"
    );

    let issuer = Signer::single(
        Aid::of_inception(Said::hash(b"issuer")),
        0,
        SigningKey::from_bytes(&[0x24; 32]),
    );
    let genuine = Envelope::issue(
        &issuer,
        Aid::of_inception(Said::hash(b"holder")),
        [1; 16],
        Cargo::usd_micros(100),
        JouleCost::measured(1),
    );
    // The cargo inflated after issuance, the snapshot SAID re-stamped so only
    // the identity gives it away.
    let mut forged = genuine.clone();
    forged.provenance.cargo = Cargo::usd_micros(1_000_000);
    forged.d = forged.compute_said();
    let forged = forged.to_cbor();
    let genuine_bytes = genuine.to_cbor();
    // The genuine envelope with an extra member appended: a map of five
    // members where the envelope defines four.
    let mut padded = genuine_bytes.clone();
    assert_eq!(padded[0], 0xa4, "a four-member map");
    padded[0] = 0xa5;
    padded.extend_from_slice(&[0x61, b'x', 0x00]);

    let gossip = &mut a.swarm.behaviour_mut().gossipsub;
    for bytes in [&forged, &padded, &genuine_bytes] {
        gossip
            .publish(IdentTopic::new(topic), bytes.clone())
            .expect("publish");
    }

    // Wait for the genuine envelope, then a grace period for anything that
    // would have been relayed alongside it.
    let mut seen: Vec<Vec<u8>> = Vec::new();
    let deadline = Instant::now() + Duration::from_secs(15);
    let mut grace: Option<Instant> = None;
    loop {
        let now = Instant::now();
        if now > deadline || grace.is_some_and(|g| now > g) {
            break;
        }
        tokio::select! {
            _ = a.swarm.select_next_some() => {}
            ev = c.swarm.select_next_some() => {
                if let Some(data) = received(&mut c, ev) {
                    if data == genuine_bytes && grace.is_none() {
                        grace = Some(Instant::now() + Duration::from_millis(1500));
                    }
                    seen.push(data);
                }
            }
            _ = tokio::time::sleep(Duration::from_millis(20)) => {}
        }
    }

    assert!(
        seen.contains(&genuine_bytes),
        "the genuine envelope never reached C"
    );
    assert!(
        !seen.contains(&forged),
        "B relayed an envelope whose content does not hash to its id"
    );
    assert!(
        !seen.contains(&padded),
        "B relayed a padded copy of a genuine envelope"
    );
}