refactors

src/auth.rs

@@ -64,19 +64,13 @@ pub async fn dyn_ip_update(
             })?
             .into()),
         PacketKind::Error => {
-            let first_zero = packet
-                .data
-                .iter()
-                .enumerate()
-                .find_map(|(i, x)| (*x == 0).then_some(i));
-
             return Err(eyre!(
                 "{}",
-                std::str::from_utf8(first_zero.map_or(&packet.data, |i| &packet.data[..i]),)?
+                packet.as_string().unwrap_or("unknown dyn auth error")
             ));
         }
 
-        _ => return Err(eyre!("server returned unexpected packet")),
+        _ => return Err(eyre!("auth server returned unexpected packet")),
     };
 
     debug!(?result, "finished dyn ip update");

src/client.rs

@@ -15,7 +15,7 @@ use tracing::{info, instrument, trace};
 use crate::{
     auth::dyn_ip_update,
     constants::{AUTH_TIMEOUT, CALL_ACK_TIMEOUT, CALL_TIMEOUT, PING_TIMEOUT, SEND_PING_INTERVAL},
-    debug_server::peer_query,
+    http::peer_query,
     packets::{Header, Packet, PacketKind, RemConnect, REJECT_OOP, REJECT_TIMEOUT},
     ports::{PortHandler, PortStatus},
     Config, HandlerMetadata,

src/debug_server.rs

@@ -1,175 +0,0 @@
-use futures::Future;
-use tokio_stream::StreamExt;
-
-use hyper::rt::Executor;
-use hyper::service::{make_service_fn, service_fn};
-use hyper::{Body, Method, Response, Server, StatusCode};
-use std::convert::Infallible;
-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 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<T: Send + 'static, Fut: Future<Output = T> + Send + 'static> Executor<Fut> for NamedExecutor {
-    fn execute(&self, fut: Fut) {
-        spawn("http worker", fut);
-    }
-}
-
-pub async fn debug_server(
-    addr: SocketAddr,
-    port_handler: Arc<Mutex<PortHandler>>,
-    change_receiver: tokio::sync::watch::Receiver<std::time::Instant>,
-) {
-    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 = WatchStream::new(change_receiver.clone());
-                    async move {
-                        match (req.method(), req.uri().path()) {
-                            (&Method::GET, "/") => Ok(Response::new(Body::from(include_str!(
-                                concat!(env!("OUT_DIR"), "/minified.html")
-                            )))),
-
-                            (&Method::GET, "/data") => {
-                                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(""))
-                                    }
-                                }
-                            }
-
-                            (&Method::GET, "/events") => Response::builder()
-                                .status(StatusCode::OK)
-                                .header("Cache-Control", "no-store")
-                                .header("Content-Type", "text/event-stream")
-                                .body(Body::wrap_stream({
-                                    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"
-                                            )))
-                                        })
-                                })),
-                            _ => Response::builder()
-                                .status(StatusCode::NOT_FOUND)
-                                .body(Body::empty()),
-                        }
-                    }
-                }))
-            }
-        }));
-
-    if let Err(error) = server.await {
-        error!(%error, "debug server error");
-    }
-}
-
-type U16 = zerocopy::U16<LittleEndian>;
-type U32 = zerocopy::U32<LittleEndian>;
-
-#[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<Option<String>> {
-    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
-    })
-}

src/main.rs

@@ -10,8 +10,8 @@ use std::{
     time::Duration,
 };
 
-use debug_server::debug_server;
 use futures::Future;
+use http::debug_server;
 use packets::{Header, Packet};
 use serde::{Deserialize, Deserializer};
 use time::format_description::OwnedFormatItem;
@@ -31,7 +31,7 @@ pub mod auth;
 pub mod client;
 pub mod constants;
 #[cfg(feature = "debug_server")]
-pub mod debug_server;
+pub mod http;
 pub mod packets;
 pub mod ports;
 
@@ -184,7 +184,7 @@ fn setup_tracing(config: &Config) {
                 Level::ERROR => write!(writer, " {:>5} ", level.red())?,
             }
 
-            write!(writer, "{:23}{}", meta.target().dimmed(), ":".bold())?;
+            write!(writer, "{:18}{}", meta.target().dimmed(), ":".bold())?;
 
             /*
             if let Some(filename) = meta.file() {
@@ -249,8 +249,8 @@ async fn connection_handler(
         Err(_) => Some("internal server error".to_owned()),
         Ok(Err(err)) => match err.downcast_ref::<std::io::Error>() {
             Some(io_error) if io_error.kind() == std::io::ErrorKind::UnexpectedEof => {
-                debug!(%addr, "Client dropped their connection");
                 // don't print an error on dropped connections
+                debug!(%addr, "Client dropped their connection");
                 None
             }
             _ => Some(err.to_string()),

src/packets.rs

@@ -1,4 +1,4 @@
-use std::{ffi::CStr, fmt::Debug};
+use std::fmt::Debug;
 
 use bytemuck::{Pod, Zeroable};
 use eyre::eyre;
@@ -82,19 +82,40 @@ pub struct Packet {
     pub data: Vec<u8>,
 }
 
+impl Packet {
+    #[must_use]
+    pub fn data(&self) -> &[u8] {
+        &self.data[..self.header.length as usize]
+    }
+
+    #[must_use]
+    pub fn as_string(&self) -> Option<&str> {
+        let data = self.data();
+        let nul = data.iter().enumerate().find(|(_i, c)| **c == 0);
+
+        let data = if let Some((i, _)) = nul {
+            &data[..i]
+        } else {
+            data
+        };
+
+        std::str::from_utf8(data).ok()
+    }
+}
+
 impl Debug for Packet {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let data = &self.data;
-
         let mut debugger = f.debug_struct("Packet");
 
         debugger.field("kind", &PacketKind::from_u8(self.header.kind));
 
-        let c_str = CStr::from_bytes_until_nul(data).ok();
+        match self.as_string() {
-        if let Some(str_data) = c_str.as_ref().and_then(|x| x.to_str().ok()) {
+            Some(string) if string.chars().all(|c| !c.is_control()) => {
-            debugger.field("data", &str_data);
+                debugger.field("data", &string);
-        } else {
+            }
-            debugger.field("data", &data);
+            _ => {
+                debugger.field("data", &self.data());
+            }
         }
 
         debugger.finish()
@@ -132,7 +153,7 @@ impl Packet {
     pub async fn recv_into_cancelation_safe(
         &mut self,
         stream: &mut ReadHalf<'_>,
-    ) -> std::io::Result<()> {
+    ) -> eyre::Result<()> {
         // Makes sure all data is available before reading
         let header_bytes = bytemuck::bytes_of_mut(&mut self.header);
         stream.peek(header_bytes).await?;
@@ -144,7 +165,7 @@ impl Packet {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn recv_into(&mut self, stream: &mut ReadHalf<'_>) -> std::io::Result<()> {
+    pub async fn recv_into(&mut self, stream: &mut ReadHalf<'_>) -> eyre::Result<()> {
         let header_bytes = bytemuck::bytes_of_mut(&mut self.header);
 
         stream.read_exact(header_bytes).await?;
@@ -153,13 +174,20 @@ impl Packet {
 
         stream.read_exact(&mut self.data).await?;
 
+        if self.header.kind == PacketKind::Error.raw() {
+            return Err(eyre!(
+                "client reported error: {:?}",
+                self.as_string().unwrap_or("unknown dyn auth error")
+            ));
+        }
+
         Ok(())
     }
 
     #[allow(clippy::missing_errors_doc)]
     pub async fn send(&self, stream: &mut WriteHalf<'_>) -> std::io::Result<()> {
         stream.write_all(bytemuck::bytes_of(&self.header)).await?;
-        stream.write_all(&self.data).await?;
+        stream.write_all(self.data()).await?;
         Ok(())
     }
 

src/ports.rs

@@ -21,7 +21,7 @@ use tracing::{debug, error, info, instrument, warn};
 
 use crate::{
     constants::{CACHE_STORE_INTERVAL, PORT_OWNERSHIP_TIMEOUT, PORT_RETRY_TIME},
-    packets::Packet,
+    packets::{Packet, PacketKind},
     spawn, Config, Number, Port, UnixTimestamp,
 };
 
@@ -35,7 +35,7 @@ pub struct PortHandler {
     pub change_sender: Option<tokio::sync::watch::Sender<std::time::Instant>>,
 
     #[serde(skip_deserializing)]
-    port_guards: HashMap<Port, Rejector>,
+    rejectors: HashMap<Port, Rejector>,
 
     allowed_ports: AllowedList,
 
@@ -176,7 +176,7 @@ impl PortHandler {
 
         let value_object = value.as_object_mut().unwrap();
 
-        value_object.remove("port_guards").unwrap();
+        value_object.remove("rejectors").unwrap();
         value_object.remove("last_update").unwrap();
 
         serde_json::to_writer(BufWriter::new(File::create(&temp_file)?), &value)?;
@@ -248,7 +248,7 @@ impl PortHandler {
 
         let port_guard = Rejector::start(listener, packet);
 
-        if self.port_guards.insert(port, port_guard).is_some() {
+        if self.rejectors.insert(port, port_guard).is_some() {
             unreachable!("Tried to start rejector that is already running. This should have been impossible since it requires two listeners on the same port.");
         }
     }
@@ -257,7 +257,7 @@ impl PortHandler {
     pub async fn stop_rejector(&mut self, port: Port) -> Option<(TcpListener, Packet)> {
         info!(port, "stopping rejector");
 
-        Some(self.port_guards.remove(&port)?.stop().await)
+        Some(self.rejectors.remove(&port)?.stop().await)
     }
 
     /// # Errors
@@ -290,7 +290,19 @@ impl Serialize for Rejector {
     where
         S: Serializer,
     {
-        self.state.1.serialize(serializer)
+        let packet = &self.state.1;
+        let kind = match packet.kind() {
+            PacketKind::End => "end",
+            PacketKind::Reject => "reject",
+            _ => unreachable!(),
+        };
+
+        match packet.as_string() {
+            Some(string) if string.chars().all(|c| !c.is_control()) => {
+                (kind, string).serialize(serializer)
+            }
+            _ => (kind, packet.data()).serialize(serializer),
+        }
     }
 }
 

web/index.html

@@ -9,6 +9,9 @@
 </head>
 
 <body>
+
+    <p id="last_update" />
+    <pre id="data"></pre>
     <list id="list" />
 </body>
 

web/main.css

@@ -1,3 +1,3 @@
-.body {
+body {
-    background-color: gray;
+    background-color: #eee;
 }

web/main.js

@@ -1,19 +1,25 @@
 window.onload = () => {
     const evtSource = new EventSource("/events");
+    const data = document.getElementById("data");
+    const last_update = document.getElementById("last_update");
     evtSource.addEventListener("change", event => {
         console.log(event);
 
+        last_update.innerText = `last update at ${new Date(+event.data * 1000)}`;
+
         const newElement = document.createElement("li");
         const eventList = document.getElementById("list");
         newElement.textContent = `change at ${+event.data}`;
         eventList.appendChild(newElement);
 
-        fetch("/data").then(res => res.json().then(res => console.log(res)));
+        fetch("/data").then(res => res.json().then(res => data.innerText = JSON.stringify(res, null, 1)));
     });
 
     evtSource.addEventListener("ping", event => {
         console.log(event);
 
+        last_update.innerText = `last update at ${new Date(+event.data * 1000)}`;
+
         const newElement = document.createElement("li");
         const eventList = document.getElementById("list");
         newElement.textContent = `ping at ${+event.data}`;