tachyon-i2p 0.0.4

//! [`Destination`]: a local I2P destination (an eepsite's identity), reachable at a
//! `.b32.i2p` address.

use crate::error::I2pError;
use crate::router::I2pRouter;
use crate::stream::I2pStream;
use std::collections::HashMap;
use std::ffi::c_void;
use std::os::raw::c_int;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, LazyLock, Mutex};

/// How many not-yet-[`accept`](Destination::accept)ed inbound streams may queue before new ones
/// are refused (closed on libi2pd's accept thread, never reaching this side). Bounded so a
/// destination nobody accepts on can't grow memory without limit; the accept callback must never
/// block (see `shim.h`), so refusing is the only option when this is full.
const ACCEPT_QUEUE_CAPACITY: usize = 128;

/// Length of the `[u8; 32]` in [`Destination::ident_hash`] and [`Destination::connect`].
///
/// Not verifiable from Rust: i2pd-sys's `I2PD_IDENT_HASH_LEN` is a C macro and its bindgen run
/// allowlists functions only. The shim `static_assert`s it against libi2pd's own `IdentHash`
/// instead, which is the side that would change.
const IDENT_HASH_LEN: usize = 32;

/// An owned stream pointer in transit: from libi2pd's accept thread into the queue
/// [`accept`](Destination::accept) drains, or out of the `spawn_blocking` worker in
/// [`connect`](Destination::connect).
///
/// Dropping it without [`into_raw`](RawStream::into_raw) releases the stream back to libi2pd.
/// That covers the two paths that would otherwise leak the handle and leave the peer's stream
/// hanging open until it timed out: a `Destination` dropped with inbound streams still queued,
/// and a `connect` cancelled after the blocking worker already produced a stream.
struct RawStream(*mut i2pd_sys::I2pdStream);

// SAFETY: the pointer is treated as an opaque, movable handle here; see `shim.h`'s thread-safety
// note -- libi2pd's own Stream API is designed to be driven from an arbitrary thread.
unsafe impl Send for RawStream {}

impl RawStream {
    /// Relinquishes ownership of the pointer to the caller, suppressing the releasing [`Drop`].
    fn into_raw(self) -> *mut i2pd_sys::I2pdStream {
        let ptr = self.0;
        std::mem::forget(self);
        ptr
    }
}

impl Drop for RawStream {
    fn drop(&mut self) {
        // SAFETY: this type owns `self.0` -- moved, never copied, from the moment libi2pd hands
        // it over to the moment `into_raw` gives it up -- so this is its last use.
        // `i2pd_destroy_stream` closes the stream itself (releasing the handle alone would leave
        // it half-open on its destination), and is null-tolerant and non-blocking per `shim.h`,
        // so this is safe to run from the accept callback too.
        unsafe {
            i2pd_sys::i2pd_destroy_stream(self.0);
        }
    }
}

/// Registry mapping an opaque token to the queue its destination's inbound streams go into.
///
/// The accept callback gets a token as its `ctx` rather than a pointer to a `Box<Sender>`,
/// because a pointer here could not be freed safely: libi2pd's acceptor slot
/// (`StreamingDestination::m_Acceptor`, a bare `std::function` with no lock around it) is cleared
/// by `ResetAcceptor` on *our* thread inside `i2pd_destroy_destination` while the io_service
/// thread may be invoking that same `std::function`, so freeing a context the callback
/// dereferences is a use-after-free in that window.
///
/// A token is an integer, so the callback dereferences nothing -- it looks the token up under
/// this mutex, and a deregistered destination simply isn't found, releasing the stream instead of
/// delivering it. The lock covers only a `HashMap` lookup and a non-blocking `try_send`, which
/// satisfies the callback's must-never-block contract.
static ACCEPT_REGISTRY: LazyLock<Mutex<HashMap<usize, tokio::sync::mpsc::Sender<RawStream>>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));

/// Source of the tokens above. Monotonic, never reused, so a token freed by one destination can
/// never be resolved to a later one.
static NEXT_ACCEPT_TOKEN: AtomicUsize = AtomicUsize::new(1);

/// Nothing under this lock can panic (a `HashMap` lookup and a `try_send`), so poisoning is
/// unreachable; recovering rather than unwrapping keeps a panic elsewhere from escalating into
/// every subsequent inbound stream on every destination being dropped.
fn accept_registry()
-> std::sync::MutexGuard<'static, HashMap<usize, tokio::sync::mpsc::Sender<RawStream>>> {
    ACCEPT_REGISTRY.lock().unwrap_or_else(|e| e.into_inner())
}

/// Invoked on libi2pd's own thread for every inbound stream, and must never block per `shim.h`.
/// The lookup and `try_send` are both non-blocking; if either fails -- unknown token or full
/// queue -- `stream` is dropped here, which closes and destroys it via [`RawStream`]'s `Drop`.
extern "C" fn on_accept(ctx: *mut c_void, stream: *mut i2pd_sys::I2pdStream) {
    if stream.is_null() {
        return;
    }
    // Own it immediately, so every path out of here releases the stream unless it was handed off.
    let stream = RawStream(stream);
    // `ctx` is a token, never a real address -- read its integer value, don't dereference it.
    let token = ctx.addr();

    let rejected = {
        let registry = accept_registry();
        match registry.get(&token) {
            // On success the queue owns the stream; on failure (queue full) it comes back.
            Some(tx) => tx.try_send(stream).err().map(|e| e.into_inner()),
            // Unknown token: the destination is already gone.
            None => Some(stream),
        }
    };
    // Outside the lock: dropping a `RawStream` calls into libi2pd to close and destroy the
    // stream, which has no business running inside this crate's critical section.
    drop(rejected);
}

pub(crate) struct DestinationHandle {
    pub(crate) ptr: *mut i2pd_sys::I2pdDestination,
    accept_token: usize,
}

// SAFETY: every libi2pd entry point this handle's pointer is passed to is documented (`shim.h`)
// as safe to call from an arbitrary thread.
unsafe impl Send for DestinationHandle {}
unsafe impl Sync for DestinationHandle {}

impl Drop for DestinationHandle {
    fn drop(&mut self) {
        // SAFETY: `self.ptr` is a valid destination handle owned by this struct, and this is its
        // last use.
        unsafe {
            i2pd_sys::i2pd_destroy_destination(self.ptr);
        }
        // Deregister *after* destroying, never before: `i2pd_destroy_destination` can invoke the
        // accept callback synchronously on this thread (`StopAcceptingStreams`/`Stop` ->
        // `StreamingDestination::ResetAcceptor` calls the acceptor one last time), and it takes
        // this same lock. A callback still racing in from the io_service thread afterwards finds
        // no entry and releases its stream instead.
        let mut registry = accept_registry();
        drop(registry.remove(&self.accept_token));
    }
}

/// A local I2P destination -- an eepsite's identity, reachable at a stable `.b32.i2p` address for
/// as long as this value is alive.
///
/// Create one via [`I2pRouter::create_transient_destination`],
/// [`I2pRouter::create_persistent_destination`], or
/// [`I2pRouter::destination_from_keys_file`].
#[derive(Debug)]
pub struct Destination {
    handle: Arc<DestinationHandle>,
    b32_address: Box<str>,
    ident_hash: [u8; IDENT_HASH_LEN],
    accept_rx: tokio::sync::mpsc::Receiver<RawStream>,
    // Keeps libi2pd's global context alive at least as long as this destination. Declared *last*
    // because fields drop in declaration order: `handle`'s `Drop` touches the global tunnel
    // pool/netDb, so it must run before the router's `Drop` tears those globals down.
    router: I2pRouter,
}

impl std::fmt::Debug for DestinationHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DestinationHandle").finish_non_exhaustive()
    }
}

impl std::fmt::Debug for RawStream {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RawStream").finish_non_exhaustive()
    }
}

impl Destination {
    /// Takes ownership of a possibly-null destination pointer fresh out of libi2pd, fetches its
    /// address, and registers the acceptor. Called only from a `spawn_blocking` worker.
    pub(crate) fn from_raw(
        router: I2pRouter,
        ptr: *mut i2pd_sys::I2pdDestination,
    ) -> Result<Self, I2pError> {
        if ptr.is_null() {
            return Err(I2pError::DestinationCreationFailed);
        }

        // SAFETY: `ptr` was just null-checked and came straight from libi2pd; the returned
        // string pointer (if non-null) is freed immediately after copying it out.
        let b32_address = unsafe {
            let raw = i2pd_sys::i2pd_destination_b32_address(ptr);
            if raw.is_null() {
                i2pd_sys::i2pd_destroy_destination(ptr);
                return Err(I2pError::DestinationCreationFailed);
            }
            let s = std::ffi::CStr::from_ptr(raw.cast())
                .to_string_lossy()
                .into_owned();
            i2pd_sys::i2pd_free_string(raw);
            s
        };

        // SAFETY: `ptr` was just null-checked and came straight from libi2pd; `out` is a valid
        // local buffer of exactly the length the shim writes.
        let mut ident_hash = [0u8; IDENT_HASH_LEN];
        let ok = unsafe { i2pd_sys::i2pd_destination_ident_hash(ptr, ident_hash.as_mut_ptr()) };
        if ok == 0 {
            // SAFETY: `ptr` is still valid and not yet handed to any `DestinationHandle`.
            unsafe { i2pd_sys::i2pd_destroy_destination(ptr) };
            return Err(I2pError::DestinationCreationFailed);
        }

        let (tx, accept_rx) = tokio::sync::mpsc::channel(ACCEPT_QUEUE_CAPACITY);
        let accept_token = NEXT_ACCEPT_TOKEN.fetch_add(1, Ordering::Relaxed);
        // Register before arming the callback, so an inbound stream arriving on the very first
        // instant the acceptor is live already resolves to this queue.
        drop(accept_registry().insert(accept_token, tx));

        // SAFETY: `ptr` is a live destination. The `ctx` argument is an integer token, not a
        // pointer -- `on_accept` never dereferences it -- so there is nothing here for a
        // late-firing callback to dangle on (see `ACCEPT_REGISTRY`).
        unsafe {
            i2pd_sys::i2pd_accept_stream(
                ptr,
                Some(on_accept),
                std::ptr::without_provenance_mut::<c_void>(accept_token),
            );
        }

        Ok(Self {
            handle: Arc::new(DestinationHandle { ptr, accept_token }),
            b32_address: b32_address.into_boxed_str(),
            ident_hash,
            accept_rx,
            router,
        })
    }

    /// This destination's `<52 chars>.b32.i2p` address.
    #[must_use]
    pub fn b32_address(&self) -> &str {
        &self.b32_address
    }

    /// This destination's raw 32-byte `IdentHash`, the form [`connect`](Self::connect) expects
    /// for a remote destination. `b32_address` is derived from the same hash.
    #[must_use]
    pub const fn ident_hash(&self) -> [u8; 32] {
        self.ident_hash
    }

    /// Waits for the next inbound stream. Unlike the underlying libi2pd accept callback this is
    /// not a one-shot -- call it in the usual `loop { let stream = dest.accept().await?; }` shape
    /// for as long as the destination lives.
    ///
    /// Takes `&mut self`, so accepts are drawn one at a time from a single owner. Streams
    /// arriving while no call is pending are queued (up to 128) rather than dropped, so a caller
    /// that hands each one to a spawned task doesn't need concurrent accepts to keep up.
    ///
    /// # Errors
    /// Returns [`I2pError::DestinationClosed`] if this destination is dropped while a call is
    /// pending.
    pub async fn accept(&mut self) -> Result<I2pStream, I2pError> {
        let raw = self
            .accept_rx
            .recv()
            .await
            .ok_or(I2pError::DestinationClosed)?;
        // SAFETY: `raw` is a freshly-accepted stream handle owned solely by this `RawStream`;
        // `into_raw` transfers that ownership on to `I2pStream::from_raw`.
        Ok(unsafe { I2pStream::from_raw(self.router.clone(), raw.into_raw()) })
    }

    /// Opens an outbound stream to `remote_ident_hash` (the raw 32-byte `IdentHash`, *not* the
    /// `.b32.i2p` string). Waits up to `timeout`, off the runtime thread, for the remote's lease
    /// set to become known and a tunnel to be built -- seconds to minutes on a cold start.
    ///
    /// `timeout` truncates to whole seconds, libi2pd's retry-loop granularity. Anything under a
    /// second, [`Duration::ZERO`](std::time::Duration::ZERO) included, becomes zero, which
    /// libi2pd treats as try-once rather than as almost a second of waiting.
    ///
    /// # Errors
    /// Returns [`I2pError::ConnectFailed`] on timeout or failure.
    pub async fn connect(
        &self,
        remote_ident_hash: [u8; 32],
        timeout: std::time::Duration,
    ) -> Result<I2pStream, I2pError> {
        let handle = self.handle.clone();
        let router = self.router.clone();
        let timeout_secs = c_int::try_from(timeout.as_secs()).unwrap_or(c_int::MAX);
        let raw = tokio::task::spawn_blocking(move || {
            // SAFETY: `handle.ptr` is a live destination for the duration of this blocking call
            // (the `Arc<DestinationHandle>` clone above keeps it alive even if `self` is
            // dropped concurrently); `remote_ident_hash` is a valid 32-byte buffer. Wrapping the
            // result in `RawStream` here means a cancelled `connect` (this future dropped while
            // the blocking worker was still retrying) releases the stream the worker went on to
            // produce, instead of leaking it.
            RawStream(unsafe {
                i2pd_sys::i2pd_create_stream(handle.ptr, remote_ident_hash.as_ptr(), timeout_secs)
            })
        })
        .await
        .map_err(|_| I2pError::WorkerPanicked)?;

        if raw.0.is_null() {
            return Err(I2pError::ConnectFailed);
        }
        // SAFETY: `raw` is a freshly-created stream handle owned solely by this `RawStream`;
        // `into_raw` transfers that ownership on to `I2pStream::from_raw`.
        Ok(unsafe { I2pStream::from_raw(router, raw.into_raw()) })
    }
}