tocat 0.2.0

A socat-inspired relay with a config file and a plugin pipeline
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143

//! tcp.rs: `tcp:` and `tcp-listen:`.
//!
//! The connecting form is the plain case. The listening form is the one with a
//! second caller: [`TcpListen::bind`] is used both by [`TcpListen::connect`],
//! which accepts exactly one peer and relays it, and by the relay's `fork`
//! loop, which keeps the listener and accepts repeatedly. Sharing `bind` is
//! what keeps the host and port defaults in one place.

use std::num::NonZeroUsize;

use serde::{Deserialize, Serialize};
use tocat_api::normalize;
use tokio::net::{TcpListener, TcpStream};
use tracing::info;

use crate::endpoint::{
    Connection, DEFAULT_HOST, DEFAULT_PORT, EndpointStream,
    parse::{Opt, ParseEndpointError, host_port},
};

#[derive(Debug, Deserialize, Serialize)]
pub struct Tcp {
    pub addr: String,
    #[serde(default)]
    pub name: Option<String>,
}

#[derive(Debug, Deserialize, Serialize)]
pub struct TcpListen {
    #[serde(default)]
    pub host: Option<String>,
    #[serde(default)]
    pub port: Option<u16>,
    #[serde(default)]
    pub name: Option<String>,
    #[serde(default)]
    pub fork: bool,
    #[serde(default, rename = "max-connections")]
    pub max_connections: Option<NonZeroUsize>,
}

impl Tcp {
    const SCHEME: &'static str = "tcp";

    pub(super) fn parse<'a>(
        body: &str,
        opts: impl Iterator<Item = Opt<'a>>,
    ) -> Result<Self, ParseEndpointError> {
        if body.is_empty() {
            return Err(ParseEndpointError::Empty);
        }

        let mut name = None;

        for opt in opts {
            match normalize(opt.key).as_str() {
                "name" => name = Some(opt.string()?),
                _ => return Err(opt.unsupported(Self::SCHEME)),
            }
        }

        Ok(Self {
            addr: body.to_owned(),
            name,
        })
    }

    pub(super) fn label(&self) -> String {
        self.name
            .clone()
            .unwrap_or_else(|| format!("tcp://{}", self.addr))
    }

    pub(super) async fn connect(&self) -> anyhow::Result<Connection> {
        let stream = TcpStream::connect(&self.addr).await?;
        Ok(EndpointStream::tcp(stream).into_connection())
    }
}

impl TcpListen {
    const SCHEME: &'static str = "tcp-listen";

    pub(super) fn parse<'a>(
        body: &str,
        opts: impl Iterator<Item = Opt<'a>>,
    ) -> Result<Self, ParseEndpointError> {
        let (host, port) = host_port(body)?;

        let mut name = None;
        let mut fork = false;
        let mut max_connections = None;

        for opt in opts {
            match normalize(opt.key).as_str() {
                "fork" => fork = opt.flag()?,
                "maxconnections" | "maxconn" => {
                    max_connections = Some(opt.count()?);
                }
                "name" => name = Some(opt.string()?),
                _ => return Err(opt.unsupported(Self::SCHEME)),
            }
        }

        Ok(Self {
            host,
            port,
            name,
            fork,
            max_connections,
        })
    }

    pub(super) fn label(&self) -> String {
        if let Some(name) = &self.name {
            return name.clone();
        }

        let (host, port) = self.addr();
        format!("tcp://{host}:{port}")
    }

    /// Where to bind, with the defaults filled in.
    pub fn addr(&self) -> (&str, u16) {
        (
            self.host.as_deref().unwrap_or(DEFAULT_HOST),
            self.port.unwrap_or(DEFAULT_PORT),
        )
    }

    /// Bind without accepting, for callers that own the accept loop.
    pub async fn bind(&self) -> std::io::Result<TcpListener> {
        TcpListener::bind(self.addr()).await
    }

    /// Bind and take a single peer.
    pub(super) async fn connect(&self) -> anyhow::Result<Connection> {
        let listener = self.bind().await?;
        info!(local = %listener.local_addr()?, "listening");
        let (stream, peer) = listener.accept().await?;
        info!("Accepted connection from {peer}");
        Ok(EndpointStream::tcp(stream).into_connection())
    }
}