pirohxy 0.1.0

SOCKS proxy with Iroh transport
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

pub mod config;
mod protocol;
mod stream;

use core::{fmt::Debug, net::SocketAddr};

use color_eyre::eyre::Result;
use iroh::{Endpoint, NodeId, SecretKey, Watcher, endpoint::VarInt, protocol::Router};
use tokio::{net::TcpListener, signal};
use tracing::{debug, info, warn};

use config::Auth;
use protocol::Socks;
use stream::copy_bidi_stream;

/// The ALPN (Application-Layer Protocol Negotiation) identifier for the iroh socks protocol.
/// This is used to identify the protocol.
const ALPN: &[u8] = b"/pirohxy/socks";

/// Starts the egress proxy server with the given identity and configuration.
/// 
/// # Errors
/// The server fails to start, or if there is an error during the shutdown process.
pub async fn start_egress<T>(self_key: SecretKey, cfg: T) -> Result<()>
where
	T: Auth + Debug + Send + Sync + 'static,
{
	debug!("Serving as node {}", self_key.public().fmt_short());
	let endpoint = start_iroh_node(self_key).await?;

	let iroh_socks = Socks::new(cfg);
	let router = Router::builder(endpoint.clone())
		.accept(ALPN, iroh_socks)
		.spawn();

	signal::ctrl_c().await?;

	Ok(router.shutdown().await?)
}

/// Binds a TCP listener and connects to a server using the provided keys and address.
/// 
/// # Errors
/// The listener fails to bind, or there is an error during the connection process.
pub async fn bind_and_connect(
	self_key: SecretKey,
	server_key: NodeId,
	bind_addr: &SocketAddr,
) -> Result<()> {
	let client_listener = TcpListener::bind(bind_addr).await?;

	debug!(
		"Dialing server {} as client {}",
		server_key.fmt_short(),
		self_key.public().fmt_short()
	);
	let endpoint = start_iroh_node(self_key).await?;

	loop {
		info!("Waiting for client connection on {}", bind_addr);

		tokio::select! {
			client_socket_res = client_listener.accept() => {
				let endpoint2 = endpoint.clone();
				match client_socket_res {
					Ok((client_socket, _)) => {
						debug!("Accepted client connection at {}", client_socket.peer_addr()?);
						let _task = tokio::spawn(async move {
							let conn = endpoint2.connect(server_key, ALPN).await?;

							let (to1, from1) = conn.open_bi().await?;
							let (from2, to2) = client_socket.into_split();

							copy_bidi_stream(from2, to2, from1, to1).await?;
							conn.close(VarInt::from_u32(1), b"BiDi exit");
							Result::<()>::Ok(())
						});
					}
					Err(e) => {
						warn!("Failed to accept client connection: {}", e);
					}
				}
			}
			_ = signal::ctrl_c() => {
				info!("Received Ctrl+C, exiting process loop.");
				endpoint.close().await;
				return Ok(());
			}
		}
	}
}

/// Starts an iroh node with the given secret key and returns the endpoint.
/// 
/// # Errors
/// The node fails to start or there is an error during initialization.
async fn start_iroh_node(key: SecretKey) -> Result<Endpoint> {
	debug!("Starting iroh node as {}", key.public().fmt_short());
	let endpoint = Endpoint::builder()
		.secret_key(key)
		.discovery_n0()
		.bind()
		.await?;
	let _relay_url = endpoint.home_relay().initialized().await;
	Ok(endpoint)
}