git.delta.rocks / fleet / refs/heads / push-kyumtlkprzyo

difftreelog

source

remowt/crates/remowt-link-shared/src/gateway.rs2.7 KiBsourcehistory
1use std::path::{Path, PathBuf};23use anyhow::{anyhow, Context as _};4use bifrostlink::{Rpc, Rtt};5use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};6use tokio::net::{UnixListener, UnixStream};7use tracing::{debug, warn};8use uuid::Uuid;910use crate::port::child_port;11use crate::{Address, BifConfig};1213pub const SOCKET_ENV: &str = "REMOWT_AGENT_SOCKET";1415pub const SOCKET_NAME: &str = "agent.sock";1617pub fn local_socket() -> anyhow::Result<PathBuf> {18	let dir = std::env::var_os("XDG_RUNTIME_DIR").context("XDG_RUNTIME_DIR not set")?;19	Ok(PathBuf::from(dir).join("remowt-local"))20}2122pub fn socket_path() -> anyhow::Result<PathBuf> {23	match std::env::var_os(SOCKET_ENV) {24		Some(p) => Ok(PathBuf::from(p)),25		None => local_socket(),26	}27}2829fn peer_tag(addr: &Address) -> u8 {30	match addr {31		Address::User => 0,32		Address::Agent => 1,33		Address::AgentPrivileged => 2,34		_ => unreachable!(),35	}36}37fn untag_peer(tag: u8) -> anyhow::Result<Address> {38	Ok(match tag {39		0 => Address::User,40		1 => Address::Agent,41		2 => Address::AgentPrivileged,42		_ => unreachable!(),43	})44}4546pub async fn serve(rpc: Rpc<BifConfig>, path: &Path) -> anyhow::Result<()> {47	let _ = tokio::fs::remove_file(path).await;48	let listener = UnixListener::bind(path)49		.with_context(|| format!("binding agent gateway at {}", path.display()))?;50	let tag = peer_tag(&rpc.me());51	tokio::spawn(async move {52		loop {53			let mut stream = match listener.accept().await {54				Ok((stream, _)) => stream,55				Err(e) => {56					warn!("gateway accept failed: {e}");57					continue;58				}59			};60			let id = Uuid::new_v4().as_u128();61			let mut hello = [0u8; 17];62			hello[0] = tag;63			hello[1..].copy_from_slice(&id.to_be_bytes());64			if let Err(e) = stream.write_all(&hello).await {65				warn!("gateway handshake failed: {e}");66				continue;67			}68			debug!("gateway client {id:032x}");69			let (rx, tx) = stream.into_split();70			rpc.add_direct(71				Address::Ephemeral(Uuid::from_u128(id)),72				child_port(rx, tx),73				Rtt(0),74			);75		}76	});77	Ok(())78}7980pub async fn connect(path: &Path) -> anyhow::Result<Rpc<BifConfig>> {81	let mut stream = UnixStream::connect(path)82		.await83		.with_context(|| format!("connecting to agent gateway at {}", path.display()))?;8485	let mut hello = [0u8; 17];86	stream87		.read_exact(&mut hello)88		.await89		.context("reading gateway handshake")?;90	let peer = untag_peer(hello[0])?;91	let id = u128::from_be_bytes(hello[1..].try_into().expect("16 bytes"));9293	let (rx, tx) = stream.into_split();94	let rpc = Rpc::<BifConfig>::new(Address::Ephemeral(Uuid::from_u128(id)));95	rpc.add_direct(peer, child_port(rx, tx), Rtt(0));96	rpc.wait_for_connection_to(Address::User)97		.await98		.map_err(|_| anyhow!("no route to the User through the agent"))?;99	Ok(rpc)100}