soruh/centralex
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

document possible errors

Browse Source
This commit is contained in:
soruh 2023-03-19 17:38:31 +01:00
parent d917afe58c
commit ac72742c2a
5 changed files with 60 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/auth.rs
View File

@ -5,6 +5,9 @@ use tracing::debug;
use crate::packets::{Header, Packet, PacketKind};
/// # Errors
/// - the dyn ip server returns a malformed response or is unreachable
/// - the authentication fails
pub async fn dyn_ip_update(
server: &SocketAddr,
number: u32,

68
src/client.rs
View File

@ -20,6 +20,8 @@ use crate::{
Config, HandlerMetadata,
};
/// # Errors
/// - the client authentication fails
async fn authenticate(
config: &Config,
port_handler: &Mutex<PortHandler>,
@ -57,37 +59,33 @@ async fn authenticate(
TcpListener::bind((config.listen_addr.ip(), port)).await
};
match listener {
Ok(listener) => {
// make sure that if we have an error, we still have access
// to the listener in the error handler.
handler_metadata.listener = Some(listener);
if let Ok(listener) = listener {
// make sure that if we have an error, we still have access
// to the listener in the error handler.
handler_metadata.listener = Some(listener);
// if we authenticated a client for a port we then failed to open
// we need to update the server here once a port that can be opened
// has been found
if !updated_server {
let _ip = dyn_ip_update(&config.dyn_ip_server, number, pin, port)
.await
.context("dy-ip update")?;
}
port_handler.register_update();
port_handler
.port_state
.entry(port)
.or_default()
.new_state(PortStatus::Idle);
handler_metadata.port = Some(port);
break Ok(Some(port));
// if we authenticated a client for a port we then failed to open
// we need to update the server here once a port that can be opened
// has been found
if !updated_server {
let _ip = dyn_ip_update(&config.dyn_ip_server, number, pin, port)
.await
.context("dy-ip update")?;
}
Err(_err) => {
port_handler.mark_port_error(number, port);
continue;
}
};
port_handler.register_update();
port_handler
.port_state
.entry(port)
.or_default()
.new_state(PortStatus::Idle);
handler_metadata.port = Some(port);
break Ok(Some(port));
}
port_handler.mark_port_error(number, port);
}
}
@ -183,7 +181,7 @@ async fn notify_or_disconnect(
.take()
.expect("tried to start rejector twice"),
packet,
)?;
);
Ok(None)
} else {
Err(anyhow!("unexpected packet: {:?}", packet.kind()))
@ -248,7 +246,7 @@ async fn connect(
.take()
.expect("tried to start rejector twice"),
packet,
)?;
);
}
stream.set_nodelay(true)?;
@ -282,6 +280,12 @@ async fn connect(
Ok(())
}
/// # Errors
/// - the connection to the client or the caller is interupted
/// - the clients sends unexpected or malformed packets
/// - accepting a tcp connection fails
/// - settings tcp socket properties fails
/// - the client authentication fails
pub async fn handler(
stream: &mut TcpStream,
addr: SocketAddr,
@ -354,7 +358,7 @@ pub async fn handler(
.take()
.expect("tried to start rejector twice"),
packet,
)?;
);
Ok(())
}

11
src/main.rs
View File

@ -1,5 +1,5 @@
#![warn(clippy::pedantic)]
#![allow(clippy::missing_errors_doc)]
// #![allow(clippy::missing_errors_doc)]
use std::{
fmt::Debug,
@ -42,10 +42,13 @@ type UnixTimestamp = u64;
#[derive(Debug, Deserialize)]
pub struct Config {
allowed_ports: AllowedList,
#[serde(deserialize_with = "parse_socket_addr")]
listen_addr: SocketAddr,
#[serde(deserialize_with = "parse_socket_addr")]
dyn_ip_server: SocketAddr,
#[cfg(feature = "debug_server")]
#[serde(deserialize_with = "maybe_parse_socket_addr")]
#[serde(default)]
@ -220,7 +223,7 @@ async fn connection_handler(
}
if let Some(listener) = handler_metadata.listener.take() {
let res = port_handler.start_rejector(
port_handler.start_rejector(
port,
listener,
Packet {
@ -231,10 +234,6 @@ async fn connection_handler(
data: b"nc\0".to_vec(),
},
);
if let Err(error) = res {
error!(%port, %error, "failed to start rejector");
}
}
}

7
src/packets.rs
View File

@ -108,12 +108,14 @@ pub struct RemConnect {
}
impl Packet {
#[allow(clippy::missing_errors_doc)]
pub async fn peek_packet_kind(stream: &mut ReadHalf<'_>) -> std::io::Result<PacketKind> {
Self::peek_packet_kind_raw(stream)
.await
.map(PacketKind::from_u8)
}
#[allow(clippy::missing_errors_doc)]
pub async fn peek_packet_kind_raw(stream: &mut ReadHalf<'_>) -> std::io::Result<u8> {
let mut kind = 0;
let n = stream.peek(std::slice::from_mut(&mut kind)).await?;
@ -125,6 +127,7 @@ impl Packet {
}
}
#[allow(clippy::missing_errors_doc)]
pub async fn recv_into_cancelation_safe(
&mut self,
stream: &mut ReadHalf<'_>,
@ -139,6 +142,7 @@ impl Packet {
self.recv_into(stream).await
}
#[allow(clippy::missing_errors_doc)]
pub async fn recv_into(&mut self, stream: &mut ReadHalf<'_>) -> std::io::Result<()> {
let header_bytes = bytemuck::bytes_of_mut(&mut self.header);
@ -151,6 +155,7 @@ impl Packet {
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?;
@ -162,6 +167,8 @@ impl Packet {
PacketKind::from_u8(self.header.kind)
}
/// # Errors
/// the packet must be a `RemConnect` packet and must contain at least 6 bytes of data
pub fn as_rem_connect(&self) -> anyhow::Result<RemConnect> {
if self.kind() != PacketKind::RemConnect {
bail!("Unexpected Packet: {:?} expected RemConnect", self.kind());

18
src/ports.rs
View File

@ -278,7 +278,8 @@ impl PortHandler {
.expect("failed to notify cache writer");
}
pub fn store(&self, cache: &Path) -> anyhow::Result<()> {
#[allow(clippy::missing_errors_doc)]
pub fn store(&self, cache: &Path) -> std::io::Result<()> {
debug!("storing cache");
let temp_file = cache.with_extension(".temp");
@ -288,6 +289,7 @@ impl PortHandler {
Ok(())
}
#[allow(clippy::missing_errors_doc)]
pub fn load(
cache: &Path,
change_sender: tokio::sync::watch::Sender<Instant>,
@ -344,12 +346,7 @@ impl PortHandler {
});
}
pub fn start_rejector(
&mut self,
port: Port,
listener: TcpListener,
packet: Packet,
) -> anyhow::Result<()> {
pub fn start_rejector(&mut self, port: Port, listener: TcpListener, packet: Packet) {
info!(port, ?packet, "starting rejector");
let port_guard = Rejector::start(listener, packet);
@ -357,7 +354,6 @@ impl PortHandler {
if self.port_guards.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.");
}
Ok(())
}
pub async fn stop_rejector(&mut self, port: Port) -> Option<(TcpListener, Packet)> {
@ -366,6 +362,8 @@ impl PortHandler {
Some(self.port_guards.remove(&port)?.stop().await)
}
/// # Errors
/// - the rejector must be running
pub async fn change_rejector(
&mut self,
port: Port,
@ -378,7 +376,9 @@ impl PortHandler {
f(&mut packet);
self.start_rejector(port, listener, packet)
self.start_rejector(port, listener, packet);
Ok(())
}
}