From 501f76ac6dce6428e0111b458b42e548a5d61641 Mon Sep 17 00:00:00 2001 From: Mollusk Date: Mon, 15 Jun 2026 05:28:04 -0400 Subject: [PATCH] feat(presence): presence control transport (W7 P4 wire layer) The I/O edge of the friends-only idle listener: bind/probe/serve a ping->pong over a dedicated ALPN (peerspeak/friends/0), adapted from pixelpass's proven control plane. Request/response, one exchange per connection: probe sends a Ping and reads the Pong; serve accepts, authenticates the remote id, and asks an injected handler (which wraps presence::should_answer + builds the pong) what to reply -- None for a stranger/invisible, so the listener reveals nothing to non-friends. Verified by a loopback integration test over two real iroh endpoints (ignored by default): the allowed prober gets a Pong with the room; a fresh stranger id gets an empty, unusable reply. 256 lib tests + the loopback (run with --ignored) green, clippy clean (incl --all-targets), release builds. DEFERRED (next session, needs care + 2 machines): spawning serve on a persistent endpoint OUTSIDE the per-join room session, and the ping scheduler. Real fork noted in the module: a second always-on endpoint shares our node id with the room endpoint (possible relay collision) vs refactoring to one persistent endpoint -- intentionally not decided at 5am. Co-Authored-By: Claude Opus 4.8 --- src/lib.rs | 1 + src/presence_net.rs | 194 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 195 insertions(+) create mode 100644 src/presence_net.rs diff --git a/src/lib.rs b/src/lib.rs index 74bd579..6ca9b11 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,7 @@ pub mod config; pub mod identity; pub mod friends; pub mod presence; +pub mod presence_net; pub mod theme; pub mod notify; pub mod screenshare; diff --git a/src/presence_net.rs b/src/presence_net.rs new file mode 100644 index 0000000..2d5df55 --- /dev/null +++ b/src/presence_net.rs @@ -0,0 +1,194 @@ +//! Presence control transport (W7 P4) — the I/O edge for the friends-only idle +//! listener. The *policy* (who we answer, what we report) lives in pure form in +//! [`crate::presence`]; this module is just the wire: bind/probe/serve a +//! ping→pong over a dedicated ALPN, adapted from pixelpass's proven control plane +//! (same iroh version). +//! +//! Shape: **request/response, one exchange per connection.** A prober opens a +//! bi-stream, writes a [`ControlMsg::Ping`], finishes its send side, and reads +//! the peer's [`ControlMsg::Pong`] back. The server accepts, reads the ping, asks +//! its injected `handler` (which encapsulates [`crate::presence::should_answer`] +//! plus building the pong from current room state) what to reply, writes it, and +//! closes. A non-friend / invisible peer gets `None`, so the stream is closed +//! with no reply and the listener never reveals anything to a stranger. +//! +//! **Deferred to a later slice (the hard integration):** spawning `serve` on a +//! persistent endpoint that lives OUTSIDE the per-join room session, and the +//! outbound ping scheduler. There's a real fork there — a second always-on +//! endpoint would share our node id with the per-join room endpoint (possible +//! relay/identity collision), vs. refactoring to a single persistent endpoint. +//! That decision wants care + a 2-machine check, so it's intentionally NOT made +//! here; this module works against whatever `Endpoint` it's handed. + +use crate::presence::ControlMsg; +use anyhow::{Context, Result, bail}; +use iroh::endpoint::{Incoming, VarInt}; +use iroh::{Endpoint, EndpointAddr, EndpointId}; +use std::sync::Arc; +use std::time::Duration; + +/// ALPN for the friends presence/control plane. Separate from the audio/gossip +/// ALPNs so a control dial never lands on a bare room endpoint and vice versa. +pub const FRIENDS_ALPN: &[u8] = b"peerspeak/friends/0"; + +/// Upper bound on a single control message — generous for a Pong carrying a +/// member ticket (~300 chars), but rejects a peer trying to make us buffer a +/// huge blob. +const MAX_MSG: usize = 64 * 1024; + +/// Bound on each network phase so a half-dead peer or relay can't park us. +const IO_TIMEOUT: Duration = Duration::from_secs(10); + +fn encode(msg: &ControlMsg) -> Result> { + serde_json::to_vec(msg).context("failed to encode control message") +} + +fn decode(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes).context("failed to decode control message") +} + +/// Probe `peer` for presence: send a `Ping`, return their `Pong`. An error means +/// no usable reply (offline / unreachable / refused / malformed) — the caller +/// treats that as "appears offline". `peer` is usually a bare [`EndpointId`] +/// (friends store the stable id); a full [`EndpointAddr`] is also accepted (and +/// used by hermetic tests). +pub async fn probe(endpoint: &Endpoint, peer: impl Into) -> Result { + let conn = tokio::time::timeout(IO_TIMEOUT, endpoint.connect(peer, FRIENDS_ALPN)) + .await + .context("timed out connecting to peer")? + .context("failed to connect to peer")?; + + let io = async { + let (mut send, mut recv) = conn.open_bi().await.context("failed to open control stream")?; + send.write_all(&encode(&ControlMsg::Ping)?) + .await + .context("failed to write ping")?; + send.finish().context("failed to finish ping stream")?; + let bytes = recv + .read_to_end(MAX_MSG) + .await + .context("peer closed the stream without replying")?; + decode(&bytes) + }; + + let result = tokio::time::timeout(IO_TIMEOUT, io) + .await + .context("timed out awaiting pong")?; + conn.close(VarInt::from_u32(0), b"done"); + result +} + +/// A reply policy: given the *authenticated* remote id, decide whether and how to +/// answer a ping. `None` = don't answer (not a friend, or we're invisible). This +/// is where [`crate::presence::should_answer`] + the Pong contents are wired in +/// by the caller; keeping it injected means this transport carries no policy. +pub type Handler = Arc Option + Send + Sync>; + +/// Run the presence accept loop on `endpoint`, replying to each ping per +/// `handler`. Returns when the endpoint stops accepting (i.e. it was closed). +pub async fn serve(endpoint: Endpoint, handler: Handler) { + while let Some(incoming) = endpoint.accept().await { + let handler = handler.clone(); + tokio::spawn(async move { + if let Err(e) = handle(incoming, handler).await { + crate::log_msg(&format!("presence: inbound connection failed: {e:#}")); + } + }); + } +} + +async fn handle(incoming: Incoming, handler: Handler) -> Result<()> { + let conn = incoming.await.context("inbound connection failed")?; + // The authenticated remote id — NOT anything the peer puts in the payload. + let from = conn.remote_id(); + + let io = async { + let (mut send, mut recv) = conn.accept_bi().await.context("failed to accept stream")?; + let bytes = recv.read_to_end(MAX_MSG).await.context("failed to read ping")?; + match decode(&bytes)? { + ControlMsg::Ping => {} + other => bail!("expected a ping, got {other:?}"), + } + // Ask the policy what to send. None -> answer nothing (stranger / invisible): + // finish the stream with no bytes so the prober sees an empty (unusable) reply. + if let Some(reply) = handler(from) { + send.write_all(&encode(&reply)?) + .await + .context("failed to write pong")?; + } + send.finish().context("failed to finish reply stream")?; + Ok::<_, anyhow::Error>(()) + }; + + tokio::time::timeout(IO_TIMEOUT, io) + .await + .context("timed out handling ping")??; + let _ = tokio::time::timeout(IO_TIMEOUT, conn.closed()).await; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::presence::{ControlMsg, RoomPresence}; + + async fn bind() -> Endpoint { + iroh::Endpoint::builder(iroh::endpoint::presets::N0) + .secret_key(iroh::SecretKey::generate()) + .alpns(vec![FRIENDS_ALPN.to_vec()]) + .bind() + .await + .unwrap() + } + + /// End-to-end over two real iroh endpoints on this machine. Ignored by + /// default (binds endpoints + waits on the relay → slow, network-dependent). + /// Run with: cargo test -- --ignored presence_net + #[tokio::test] + #[ignore = "binds real iroh endpoints; run on demand"] + async fn loopback_ping_pong_and_friends_only() { + let server = bind().await; + let prober = bind().await; + server.online().await; + prober.online().await; + let server_addr = server.addr(); + let prober_id = prober.addr().id; + + // The server answers ONLY the prober's id, reporting a room. + let allowed = prober_id; + let handler: Handler = Arc::new(move |from| { + if from == allowed { + Some(ControlMsg::Pong { + room: Some(RoomPresence { name: "HangOut".into(), ticket: "t".into() }), + }) + } else { + None // stranger -> no reply + } + }); + let server_ep = server.clone(); + let serve_task = tokio::spawn(async move { serve(server_ep, handler).await }); + + // The allowed prober gets a Pong with the room. + let pong = tokio::time::timeout(Duration::from_secs(15), probe(&prober, server_addr.clone())) + .await + .expect("probe timed out") + .expect("probe failed"); + match pong { + ControlMsg::Pong { room: Some(r) } => assert_eq!(r.name, "HangOut"), + other => panic!("expected Pong with a room, got {other:?}"), + } + + // A stranger (fresh id) gets an empty reply -> decode fails -> Err. + let stranger = bind().await; + stranger.online().await; + let res = tokio::time::timeout(Duration::from_secs(15), probe(&stranger, server_addr)) + .await + .expect("stranger probe timed out"); + assert!(res.is_err(), "a non-friend must not get a usable reply"); + + server.close().await; + prober.close().await; + stranger.close().await; + serve_task.abort(); + } +}