1//! The connecting-client side of the dev.site data plane.2//!3//! Each accepted local TCP connection becomes one authenticated Iroh bidirectional stream.45use std::collections::HashMap;6use std::sync::Mutex;7use std::time::Duration;89use anyhow::{Context, Result};10use devsite_proto::capability::SignedCapability;11use devsite_proto::wire::{self, ConnectRequest, ErrorCode, Request, Response};12use devsite_proto::ALPN;13use iroh::endpoint::Connection;14use iroh::{Endpoint, EndpointAddr, EndpointId};15use n0_future::time::timeout;1617/// How long to wait for a daemon before giving up on it.18///19/// There is no presence service telling us in advance whether the far end is20/// running, so this is what "offline" means now: we asked, and nobody answered.21/// Long enough to cover address lookup and a relay handshake, short enough that22/// a dead daemon does not leave someone watching a spinner.23pub const CONNECT_TIMEOUT: Duration = Duration::from_secs(8);2425/// Why a service stream could not be opened.26///27/// Transport failure and refusal are kept distinct so a caller — and especially a test —28/// can tell "the daemon said no" from "we never reached the daemon". Collapsing the two29/// would let a network timeout masquerade as a successful authorization check.3031pub enum ConnectError {32 /// The daemon reached a verdict and refused.33 34 Denied,35 /// The daemon responded, but with some other refusal.36 37 Rejected(ErrorCode),38 /// We never got a verdict: connection, relay, or protocol failure.39 40 Transport( anyhow::Error),41}4243impl From<wire::CodecError> for ConnectError {44 fn from(err: wire::CodecError) -> Self {45 ConnectError::Transport(err.into())46 }47}4849/// The two halves of an authorized service byte stream.50pub struct ServiceStream {51 pub send: iroh::endpoint::SendStream,52 pub recv: iroh::endpoint::RecvStream,53}5455/// An ephemeral client endpoint. A capability is bound to this key and therefore cannot56/// be redeemed by another devsite client.57pub struct ClientEndpoint {58 endpoint: Endpoint,59 /// One live connection per daemon, reused across service streams.60 ///61 /// Connecting per request does not work: dropping a `Connection` closes it, and the62 /// endpoint hands the same closed connection back to the next connection for that peer,63 /// which then fails with "closed". Holding it open also lets many local connections64 /// share one authenticated QUIC connection.65 connections: Mutex<HashMap<EndpointId, Connection>>,66}6768impl ClientEndpoint {69 /// Bind a new endpoint with a freshly generated key and wait until it is online.70 pub async fn create() -> Result<Self> {71 let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)72 .bind()73 .await74 .context("binding client endpoint")?;75 endpoint.online().await;76 Ok(Self {77 endpoint,78 connections: Mutex::new(HashMap::new()),79 })80 }8182 /// Bind an endpoint with a caller-provided key. Brokered grants use this so83 /// the endpoint that signed the request is also the endpoint that connects.84 pub async fn create_with_secret(secret_key: iroh::SecretKey) -> Result<Self> {85 let endpoint = Endpoint::builder(iroh::endpoint::presets::N0)86 .secret_key(secret_key)87 .bind()88 .await89 .context("binding client endpoint")?;90 endpoint.online().await;91 Ok(Self {92 endpoint,93 connections: Mutex::new(HashMap::new()),94 })95 }9697 /// Bind an endpoint with relay settings from the control plane.98 pub async fn create_with_relay(99 secret_key: iroh::SecretKey,100 relay_access: devsite_iroh::RelayAccess,101 ) -> Result<Self> {102 let endpoint = Endpoint::builder(devsite_iroh::preset(secret_key, relay_access)?)103 .bind()104 .await105 .context("binding client endpoint")?;106 endpoint.online().await;107 Ok(Self {108 endpoint,109 connections: Mutex::new(HashMap::new()),110 })111 }112113 /// This client's public key. The control plane binds capabilities to it, and the114 /// daemon checks that binding against the authenticated peer of the connection.115 pub fn endpoint_id(&self) -> EndpointId {116 self.endpoint.id()117 }118119 /// Close the ephemeral endpoint and every cached daemon connection cleanly.120 pub async fn close(&self) {121 self.endpoint.close().await;122 }123124 /// Open one authorized byte stream to a service behind a daemon.125 ///126 /// `daemon` is normally a bare [`EndpointId`]: the daemon publishes its own127 /// address through iroh's address lookup, and the client resolves it. Nothing has to128 /// tell us where it is. Callers that already know an129 /// address — such as tests that avoid address lookup — can pass a fuller130 /// [`EndpointAddr`].131 ///132 /// The capability must have been issued for this endpoint's key; the daemon checks133 /// that binding against the connection and will refuse otherwise.134 pub async fn connect(135 &self,136 daemon: impl Into<EndpointAddr>,137 capability: SignedCapability,138 ) -> Result<ServiceStream, ConnectError> {139 let addr = daemon.into();140 let connection = self.connection_to(&addr).await?;141 request_stream(&connection, capability).await142 }143144 /// A live connection to `daemon`, reusing the cached one when there is one.145 async fn connection_to(&self, daemon: &EndpointAddr) -> Result<Connection, ConnectError> {146 // Scoped so the guard is released before the await below. An `if let` scrutinee147 // temporary would otherwise live to the end of the whole statement, and the148 // second lock in this function would deadlock against it.149 let existing = {150 let cache = self.connections.lock().unwrap();151 cache.get(&daemon.id).cloned()152 };153 if let Some(existing) = existing {154 return Ok(existing);155 }156 // Bounded explicitly. Without a presence service there is nothing to tell157 // us the far end is down before we try, so an unreachable daemon has to158 // surface as a timeout rather than as a connect that waits on the relay.159 let connection = timeout(CONNECT_TIMEOUT, self.endpoint.connect(daemon.clone(), ALPN))160 .await161 .map_err(|_| anyhow::anyhow!("no answer from the daemon"))?162 .context("connecting to daemon")?;163 self.connections164 .lock()165 .unwrap()166 .insert(daemon.id, connection.clone());167 Ok(connection)168 }169}170171async fn request_stream(172 connection: &Connection,173 capability: SignedCapability,174) -> Result<ServiceStream, ConnectError> {175 let (mut send, mut recv) = connection176 .open_bi()177 .await178 .context("opening bidirectional stream")?;179180 let request = Request::Connect(ConnectRequest { capability });181 send.write_all(&wire::encode(&request)?)182 .await183 .context("writing connect frame")?;184185 let mut prefix = [0u8; 4];186 recv.read_exact(&mut prefix)187 .await188 .context("reading response length")?;189 let len = wire::frame_len(prefix)?;190 let mut payload = vec![0u8; len];191 recv.read_exact(&mut payload)192 .await193 .context("reading response frame")?;194195 let response = wire::decode::<Response>(&payload).context("decoding response frame")?;196 match response {197 Response::Connected => Ok(ServiceStream { send, recv }),198 Response::Error {199 code: ErrorCode::Denied,200 ..201 } => Err(ConnectError::Denied),202 Response::Error { code, .. } => Err(ConnectError::Rejected(code)),203 }204}