use bytes::BytesMut; use futures::Future; use hyper::header::{ACCEPT_ENCODING, CACHE_CONTROL, CONTENT_ENCODING, CONTENT_TYPE}; use hyper::rt::Executor; use hyper::service::{make_service_fn, service_fn}; use hyper::{Body, Method, Request, Response, Server, StatusCode}; use std::convert::Infallible; use std::io::Read; use std::net::SocketAddr; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; use tokio::sync::Mutex; use tokio_stream::wrappers::{IntervalStream, WatchStream}; use tokio_stream::StreamExt; use tracing::error; use zerocopy::{AsBytes, FromBytes, LittleEndian, Unaligned}; use tracing::{debug, instrument}; use crate::constants::DEBUG_SERVER_PING_INTERVAL; use crate::packets::{Header, Packet}; use crate::ports::PortHandler; use crate::spawn; #[derive(Clone)] struct NamedExecutor; impl + Send + 'static> Executor for NamedExecutor { fn execute(&self, fut: Fut) { spawn("http worker", fut); } } const COMPRESSED_HTML: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/minified.html.gz")); async fn index(req: &Request) -> Result, hyper::http::Error> { let response = Response::builder(); let accepts_gzip = req .headers() .get(ACCEPT_ENCODING) .map_or(false, |accept_encoding| { accept_encoding .as_bytes() .split(|x| *x == b',') .filter_map(|x| x.split(|x| *x == b';').next()) .filter_map(|x| std::str::from_utf8(x).ok()) .any(|x| x.trim() == "gzip") }); if accepts_gzip { response .header(CONTENT_ENCODING, "gzip") .body(Body::from(COMPRESSED_HTML)) } else { let (mut sender, body) = Body::channel(); spawn("gunzip task", async move { let mut decoder = flate2::bufread::GzDecoder::new(std::io::Cursor::new(COMPRESSED_HTML)); let mut done = false; while !done { let mut chunk = BytesMut::zeroed(256); let mut i = 0; loop { let dst = &mut chunk.as_bytes_mut()[i..]; if dst.is_empty() { break; // we are done } match decoder.read(dst) { Ok(n) => { if n == 0 { done = true; break; } i += n; } Err(err) => unreachable!("failed to read from gzip decode: {err}"), } } chunk.truncate(i); if sender.send_data(chunk.freeze()).await.is_err() { break; } } }); response.body(body) } } async fn data( _req: &Request, port_handler: Arc>, ) -> Result, hyper::http::Error> { let res = Response::builder().header(CACHE_CONTROL, "no-store"); match serde_json::to_string(&*port_handler.lock().await) { Ok(data) => res .header(CONTENT_TYPE, "application/json") .body(Body::from(data)), Err(err) => { error!(%err, "failed to serialize data for debug server"); res.status(StatusCode::INTERNAL_SERVER_ERROR) .body(Body::from("")) } } } fn events( _req: &Request, change_receiver: tokio::sync::watch::Receiver, ) -> Result, hyper::http::Error> { Response::builder() .status(StatusCode::OK) .header(CACHE_CONTROL, "no-store") .header(CONTENT_TYPE, "text/event-stream") .body(Body::wrap_stream({ WatchStream::new(change_receiver) .map(|x| ("change", x)) .merge( IntervalStream::new(tokio::time::interval(DEBUG_SERVER_PING_INTERVAL)) .map(|x| ("ping", x.into_std())), ) .filter_map(|(kind, time)| { let timestamp = (SystemTime::now() + time.elapsed()) .duration_since(UNIX_EPOCH) .ok()? .as_secs(); Some(Ok::<_, Infallible>(format!( "event:{kind}\ndata: {timestamp}\n\n" ))) }) })) } pub async fn debug_server( addr: SocketAddr, port_handler: Arc>, change_receiver: tokio::sync::watch::Receiver, ) { let server = Server::bind(&addr) .executor(NamedExecutor) .serve(make_service_fn(move |_conn| { let port_handler = port_handler.clone(); let change_receiver = change_receiver.clone(); async move { Ok::<_, Infallible>(service_fn(move |req| { let port_handler = port_handler.clone(); let change_receiver = change_receiver.clone(); async move { match (req.method(), req.uri().path()) { (&Method::GET, "/") => index(&req).await, (&Method::GET, "/data") => data(&req, port_handler).await, (&Method::GET, "/events") => events(&req, change_receiver), _ => Response::builder() .status(StatusCode::NOT_FOUND) .body(Body::empty()), } } })) } })); if let Err(error) = server.await { error!(%error, "debug server error"); } } type U16 = zerocopy::U16; type U32 = zerocopy::U32; #[derive(AsBytes)] #[repr(transparent)] #[allow(dead_code)] struct PeerQuery { number: U32, } #[derive(FromBytes, Unaligned, Debug)] #[repr(packed)] #[allow(dead_code)] struct PeerReply { number: U32, name: [u8; 40], flags: U16, kind: u8, hostname: [u8; 40], ipaddress: [u8; 4], port: U16, extension: u8, pin: U16, timestamp: U32, } #[instrument] pub async fn peer_query(server: &SocketAddr, number: u32) -> eyre::Result> { debug!(%number, "looking up"); let mut packet = Packet { header: Header { kind: 3, // Peer Query length: 4, }, data: Vec::new(), }; packet.data.clear(); packet.data.resize(packet.header.length as usize, 0); PeerQuery { number: number.into(), } .write_to(packet.data.as_mut_slice()) .unwrap(); let mut socket = tokio::net::TcpStream::connect(server).await?; let (mut reader, mut writer) = socket.split(); packet.send(&mut writer).await?; packet.recv_into(&mut reader).await?; Ok(if packet.kind().raw() == 5 { // PeerReply PeerReply::read_from(packet.data.as_slice()).and_then(|x| { let i = x .name .iter() .enumerate() .find_map(|(i, c)| (*c == 0).then_some(i)) .unwrap_or(x.name.len()); Some(std::str::from_utf8(&x.name[..i]).ok()?.to_owned()) }) } else { None }) }