1//! The daemon: an Iroh endpoint that forwards authorized streams to fixed local services.2//!3//! It exposes no application TCP port. Iroh establishes direct or relayed connectivity,4//! and the daemon decides for itself whether to honour a request by verifying a capability5//! signed by the control plane — the control plane never touches this traffic.67pub mod config;8pub mod verify;910use std::collections::{HashMap, HashSet};11use std::sync::atomic::{AtomicU64, Ordering};12use std::sync::Arc;13use std::time::{SystemTime, UNIX_EPOCH};1415use anyhow::{Context, Result};16use devsite_proto::capability::KeyBytes;17use devsite_proto::wire::{self, ErrorCode, Request, Response};18use devsite_proto::{AccountId, ResourceId, ALPN};19use ed25519_dalek::VerifyingKey;20use iroh::endpoint::Connection;21use iroh::{Endpoint, SecretKey};22use tokio::io::{copy, AsyncWriteExt};23use tokio::net::TcpStream;24use tokio::sync::{oneshot, Mutex, RwLock};2526use crate::config::{DaemonConfig, HostedService};27use crate::verify::{Denied, ReplayGuard};2829pub fn now_secs() -> u64 {30 SystemTime::now()31 .duration_since(UNIX_EPOCH)32 .map(|d| d.as_secs())33 .unwrap_or(0)34}3536pub struct Daemon {37 endpoint: Endpoint,38 config: RwLock<DaemonConfig>,39 control_plane_key: VerifyingKey,40 replay: Arc<Mutex<ReplayGuard>>,41 active: Mutex<HashMap<u64, ActiveStream>>,42 next_stream_id: AtomicU64,43}4445struct ActiveStream {46 viewer: AccountId,47 resource: ResourceId,48 cancel: oneshot::Sender<()>,49}5051impl Daemon {52 /// Bind the endpoint and wait until Iroh has established network connectivity.53 ///54 /// Fails if no control-plane key has been pinned: a daemon that cannot verify55 /// capabilities must refuse to start rather than run in some permissive mode.56 pub async fn bind(secret_key: SecretKey, config: DaemonConfig) -> Result<Self> {57 Self::bind_builder(58 Endpoint::builder(iroh::endpoint::presets::N0).secret_key(secret_key),59 config,60 )61 .await62 }6364 /// Bind the endpoint with relay settings from the control plane.65 pub async fn bind_with_relay(66 secret_key: SecretKey,67 config: DaemonConfig,68 relay_access: devsite_iroh::RelayAccess,69 ) -> Result<Self> {70 Self::bind_builder(71 Endpoint::builder(devsite_iroh::preset(secret_key, relay_access)?),72 config,73 )74 .await75 }7677 async fn bind_builder(builder: iroh::endpoint::Builder, config: DaemonConfig) -> Result<Self> {78 let control_plane_key = config.verifying_key()?;7980 let endpoint = builder81 .alpns(vec![ALPN.to_vec()])82 .bind()83 .await84 .context("binding daemon endpoint")?;85 endpoint.online().await;8687 Ok(Self {88 endpoint,89 config: RwLock::new(config),90 control_plane_key,91 replay: Arc::new(Mutex::new(ReplayGuard::default())),92 active: Mutex::new(HashMap::new()),93 next_stream_id: AtomicU64::new(1),94 })95 }9697 pub fn endpoint(&self) -> &Endpoint {98 &self.endpoint99 }100101 pub fn endpoint_key(&self) -> KeyBytes {102 *self.endpoint.id().as_bytes()103 }104105 /// Stop accepting new peers and close the Iroh endpoint cleanly. Service106 /// managers do not need a platform-specific integration; terminating the107 /// foreground command is still the whole lifecycle contract.108 pub async fn close(&self) {109 self.endpoint.close().await;110 }111112 /// Replace only the resource map while the endpoint keeps its identity and113 /// open connections. This is intentionally platform-neutral: service114 /// managers only need to keep `devsite daemon run` alive.115 pub async fn replace_resources(&self, resources: Vec<HostedService>) -> bool {116 let allowed = resources117 .iter()118 .map(|resource| resource.resource_id)119 .collect::<HashSet<_>>();120 {121 let mut config = self.config.write().await;122 if config.resources == resources {123 return false;124 }125 config.resources = resources;126 }127 self.cancel_where(|stream| !allowed.contains(&stream.resource))128 .await;129 true130 }131132 /// Replace the control plane's current authorization snapshot and terminate133 /// streams whose viewer/resource pair has since been revoked.134 pub async fn replace_authorizations(&self, allowed: HashSet<(AccountId, ResourceId)>) -> usize {135 self.cancel_where(|stream| !allowed.contains(&(stream.viewer, stream.resource)))136 .await137 }138139 /// Fail closed after the control plane authorization lease has gone stale.140 pub async fn revoke_all_active(&self) -> usize {141 self.cancel_where(|_| true).await142 }143144 async fn cancel_where(&self, denied: impl Fn(&ActiveStream) -> bool) -> usize {145 let mut active = self.active.lock().await;146 let ids = active147 .iter()148 .filter_map(|(id, stream)| denied(stream).then_some(*id))149 .collect::<Vec<_>>();150 let count = ids.len();151 for id in ids {152 if let Some(stream) = active.remove(&id) {153 let _ = stream.cancel.send(());154 }155 }156 count157 }158159 /// Accept connections until the endpoint closes.160 pub async fn serve(self: Arc<Self>) -> Result<()> {161 let resources = self.config.read().await.resources.len();162 tracing::info!(163 resources,164 "daemon serving; every request must present a valid capability"165 );166167 while let Some(incoming) = self.endpoint.accept().await {168 let daemon = Arc::clone(&self);169 tokio::spawn(async move {170 match incoming.await {171 Ok(connection) => daemon.handle_connection(connection).await,172 Err(err) => tracing::debug!("incoming connection failed: {err:#}"),173 }174 });175 }176 Ok(())177 }178179 /// Serve requests until the peer goes away.180 ///181 /// A connection carries many requests, so this loops rather than answering once and182 /// dropping. Closing after a single response leaves the peer holding a dead connection183 /// that its next request reuses — which surfaces as a connect failing with "closed"184 /// rather than as anything obviously wrong here.185 async fn handle_connection(self: Arc<Self>, connection: Connection) {186 // The authenticated peer. This is the only identity we trust about the other side,187 // and the capability's client binding is checked against it.188 let peer: KeyBytes = *connection.remote_id().as_bytes();189190 loop {191 let stream = match connection.accept_bi().await {192 Ok(stream) => stream,193 // The peer closed, or the connection dropped. Either way we are done.194 Err(err) => {195 tracing::trace!("connection with {} ended: {err}", hex(&peer));196 return;197 }198 };199200 let daemon = Arc::clone(&self);201 tokio::spawn(async move {202 if let Err(err) = daemon.handle_stream(stream, &peer).await {203 tracing::debug!("request from {} failed: {err:#}", hex(&peer));204 }205 });206 }207 }208209 async fn handle_stream(210 &self,211 (mut send, mut recv): (iroh::endpoint::SendStream, iroh::endpoint::RecvStream),212 peer: &KeyBytes,213 ) -> Result<()> {214 let mut prefix = [0u8; 4];215 recv.read_exact(&mut prefix)216 .await217 .context("reading request length")?;218 let len = wire::frame_len(prefix)?;219 let mut payload = vec![0u8; len];220 recv.read_exact(&mut payload)221 .await222 .context("reading request frame")?;223224 let request = match wire::decode::<Request>(&payload) {225 Ok(request) => request,226 Err(err) => {227 tracing::debug!("malformed frame from {}: {err:#}", hex(peer));228 return send_response(229 &mut send,230 Response::Error {231 code: ErrorCode::BadRequest,232 message: "malformed request".to_string(),233 },234 )235 .await;236 }237 };238 let Request::Connect(request) = request;239 let targets = self.config.read().await.targets();240 let my_endpoint = self.endpoint_key();241 let now = now_secs();242243 let authorized = {244 let mut replay = self.replay.lock().await;245 verify::authorize(246 &request.capability,247 &self.control_plane_key,248 &my_endpoint,249 peer,250 &targets,251 &mut replay,252 now,253 )254 };255256 let authorized = match authorized {257 Ok(authorized) => authorized,258 Err(denied) => {259 // Logged with detail locally, reported to the peer as a bare "denied".260 tracing::warn!(peer = %hex(peer), reason = ?denied, "denied request");261 debug_assert!(matches!(262 denied,263 Denied::Capability | Denied::UnknownResource | Denied::Replayed264 ));265 return send_response(266 &mut send,267 Response::Error {268 code: ErrorCode::Denied,269 message: "denied".to_string(),270 },271 )272 .await;273 }274 };275276 let upstream = match TcpStream::connect(authorized.target).await {277 Ok(stream) => stream,278 Err(err) => {279 tracing::warn!(target = %authorized.target, "local service did not accept TCP: {err}");280 return send_response(281 &mut send,282 Response::Error {283 code: ErrorCode::UpstreamUnavailable,284 message: "local service did not accept a connection".to_string(),285 },286 )287 .await;288 }289 };290291 tracing::info!(292 viewer = %authorized.claims.viewer,293 resource = %authorized.claims.resource,294 target = %authorized.target,295 "opened service stream"296 );297 send.write_all(&wire::encode(&Response::Connected)?)298 .await299 .context("writing connect response")?;300301 let stream_id = self.next_stream_id.fetch_add(1, Ordering::Relaxed);302 let (cancel, mut cancelled) = oneshot::channel();303 self.active.lock().await.insert(304 stream_id,305 ActiveStream {306 viewer: authorized.claims.viewer,307 resource: authorized.claims.resource,308 cancel,309 },310 );311312 let (mut upstream_read, mut upstream_write) = upstream.into_split();313 let client_to_service = async {314 copy(&mut recv, &mut upstream_write)315 .await316 .context("forwarding client bytes")?;317 upstream_write.shutdown().await.ok();318 Ok::<_, anyhow::Error>(())319 };320 let service_to_client = async {321 copy(&mut upstream_read, &mut send)322 .await323 .context("forwarding service bytes")?;324 send.finish().context("finishing service stream")?;325 Ok::<_, anyhow::Error>(())326 };327 let result = tokio::select! {328 result = async { tokio::try_join!(client_to_service, service_to_client) } => {329 result.map(|_| ())330 }331 _ = &mut cancelled => Ok(()),332 };333 self.active.lock().await.remove(&stream_id);334 result335 }336}337338async fn send_response(send: &mut iroh::endpoint::SendStream, response: Response) -> Result<()> {339 send.write_all(&wire::encode(&response)?)340 .await341 .context("writing response frame")?;342 send.finish().context("finishing response stream")?;343 send.stopped().await.ok();344 Ok(())345}346347fn hex(bytes: &KeyBytes) -> String {348 data_encoding::HEXLOWER.encode(bytes)349}