yog 0.0.75

//! A fake DHT node on loopback UDP — the far end every walk in this corpus
//! talks to (REMOTE §13.5: designed for from the first line). One thread per
//! node, a scripted routing answer, a BEP 44 store that checks what a real
//! node checks (token, signature, sequence), and a `Mood` for each way a
//! node on the commons misbehaves.

use super::super::bencode::{Dict, Value, bytes, entry};
use super::super::krpc::{Node, NodeId};
use super::super::mutable::Mutable;
use std::net::{SocketAddr, UdpSocket};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::JoinHandle;
use std::time::Duration;
use wire::{claim, error, reply, routing};

mod wire;

pub(crate) const TOKEN: &[u8] = b"tok";

#[derive(Clone, Copy)]
pub(crate) enum Mood {
    Answer,
    Silent,
    Garbage,
    Refuse,
    /// Replies without stating an id.
    Anonymous,
    /// Replies under a transaction nobody opened.
    Stray,
    /// Answers, and says (BEP 42's `ip`) the query came from this address —
    /// true or not, which is the point: one node's word is only a claim.
    Claim(SocketAddr),
    /// A mainline bootstrap router as measured (REMOTE §13.7 ruling 3):
    /// answers `find_node`, silent to BEP 44's `get` and `put`.
    Router,
    /// The one router that answers from the deployed engine box, as measured
    /// (bl-d00f): a `Router` whose every `find_node` answer is ONE of its
    /// peers repeated eight times, the next peer on the next query.
    Rotor,
    /// Answers everything but `put`: offers a token and never spends it.
    Mute,
    /// Answers, but every second `get` for an item it holds comes back
    /// without it — an inbox that alternates between a stale item and a
    /// quiet read, as a thin swarm's does (bl-1633).
    Blink,
}

pub(crate) struct FakeNode {
    socket: Option<UdpSocket>,
    pub(crate) addr: SocketAddr,
    pub(crate) id: NodeId,
    stop: Arc<AtomicBool>,
    thread: Option<JoinHandle<()>>,
}

impl FakeNode {
    pub(crate) fn bind(id: NodeId) -> FakeNode {
        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
        socket
            .set_read_timeout(Some(Duration::from_millis(20)))
            .unwrap();
        let addr = socket.local_addr().unwrap();
        FakeNode {
            socket: Some(socket),
            addr,
            id,
            stop: Arc::new(AtomicBool::new(false)),
            thread: None,
        }
    }

    pub(crate) fn node(&self) -> Node {
        Node {
            id: self.id,
            addr: self.addr,
        }
    }

    pub(crate) fn serve(&mut self, peers: Vec<Node>, mood: Mood, items: Vec<Mutable>) {
        let socket = self.socket.take().unwrap();
        let (id, stop) = (self.id, Arc::clone(&self.stop));
        self.thread = Some(std::thread::spawn(move || {
            run(&socket, id, &peers, mood, items, &stop);
        }));
    }
}

impl Drop for FakeNode {
    fn drop(&mut self) {
        self.stop.store(true, Ordering::Relaxed);
        if let Some(t) = self.thread.take() {
            t.join().unwrap();
        }
    }
}

fn run(
    socket: &UdpSocket,
    id: NodeId,
    peers: &[Node],
    mood: Mood,
    mut items: Vec<Mutable>,
    stop: &AtomicBool,
) {
    let mut buf = vec![0u8; 8192];
    let mut turn = 0usize;
    while !stop.load(Ordering::Relaxed) {
        let Ok((n, from)) = socket.recv_from(&mut buf) else {
            continue;
        };
        let q = Value::decode(&buf[..n]).unwrap();
        let tid = q.get("t").unwrap().as_bytes().unwrap().to_vec();
        let datagram = match mood {
            Mood::Silent => continue,
            Mood::Router | Mood::Rotor
                if q.get("q").unwrap().as_bytes().unwrap() != b"find_node" =>
            {
                continue;
            }
            Mood::Mute if q.get("q").unwrap().as_bytes().unwrap() == b"put" => continue,
            Mood::Rotor => {
                turn += 1;
                let one = vec![peers[(turn - 1) % peers.len()]; 8];
                answer(&tid, id, &one, &mut items, &q)
            }
            Mood::Blink if hits(&q, &items) => {
                turn += 1;
                let mut none = Vec::new();
                let shown = if turn.is_multiple_of(2) {
                    &mut none
                } else {
                    &mut items
                };
                answer(&tid, id, peers, shown, &q)
            }
            Mood::Garbage => b"not bencode".to_vec(),
            Mood::Refuse => error(&tid, 201, "refused"),
            Mood::Anonymous => reply(&tid, Dict::new()),
            Mood::Stray => reply(b"stray", Dict::from([entry("id", bytes(&id.0))])),
            Mood::Answer | Mood::Router | Mood::Mute | Mood::Blink => {
                answer(&tid, id, peers, &mut items, &q)
            }
            Mood::Claim(ip) => claim(answer(&tid, id, peers, &mut items, &q), ip),
        };
        socket.send_to(&datagram, from).unwrap();
    }
}

/// A `get` for an item this node holds.
fn hits(q: &Value, items: &[Mutable]) -> bool {
    q.get("q").unwrap().as_bytes().unwrap() == b"get"
        && items.iter().any(|i| {
            let target = q
                .get("a")
                .unwrap()
                .get("target")
                .unwrap()
                .as_bytes()
                .unwrap();
            i.target().0 == target
        })
}

fn answer(tid: &[u8], id: NodeId, peers: &[Node], items: &mut Vec<Mutable>, q: &Value) -> Vec<u8> {
    let a = q.get("a").unwrap();
    let mut r = Dict::from([entry("id", bytes(&id.0))]);
    match q.get("q").unwrap().as_bytes().unwrap() {
        b"find_node" => {
            r.extend(routing(peers));
            reply(tid, r)
        }
        b"get" => {
            r.extend(routing(peers));
            r.insert(b"token".to_vec(), bytes(TOKEN));
            let target = NodeId::parse(a.get("target").unwrap().as_bytes().unwrap()).unwrap();
            if let Some(item) = items.iter().find(|i| i.target() == target) {
                r.insert(b"k".to_vec(), bytes(&item.key));
                r.insert(b"seq".to_vec(), Value::Int(item.seq));
                r.insert(b"sig".to_vec(), bytes(&item.sig));
                r.insert(b"v".to_vec(), bytes(&item.value));
            }
            reply(tid, r)
        }
        b"put" => {
            if a.get("token").unwrap().as_bytes().unwrap() != TOKEN {
                return error(tid, 203, "bad token");
            }
            let item = Mutable {
                key: a.get("k").unwrap().as_bytes().unwrap().try_into().unwrap(),
                salt: a
                    .get("salt")
                    .map(|s| s.as_bytes().unwrap().to_vec())
                    .unwrap_or_default(),
                seq: a.get("seq").unwrap().as_int().unwrap(),
                value: a.get("v").unwrap().as_bytes().unwrap().to_vec(),
                sig: a
                    .get("sig")
                    .unwrap()
                    .as_bytes()
                    .unwrap()
                    .try_into()
                    .unwrap(),
            };
            if !item.verify() {
                return error(tid, 206, "invalid signature");
            }
            let target = item.target();
            if items
                .iter()
                .any(|i| i.target() == target && i.seq >= item.seq)
            {
                return error(tid, 302, "sequence number less than current");
            }
            items.retain(|i| i.target() != target);
            items.push(item);
            reply(tid, r)
        }
        _ => error(tid, 204, "unknown method"),
    }
}