diff options
Diffstat (limited to 'ingress/src')
| -rw-r--r-- | ingress/src/main.rs | 245 |
1 files changed, 245 insertions, 0 deletions
diff --git a/ingress/src/main.rs b/ingress/src/main.rs new file mode 100644 index 0000000..4f8dff6 --- /dev/null +++ b/ingress/src/main.rs @@ -0,0 +1,245 @@ +use anyhow::{bail, Context, Result}; +use base64::{engine::general_purpose::STANDARD as BASE64, Engine}; +use nym_sdk::mixnet::{MixnetClientBuilder, StoragePaths}; +use serde::Deserialize; +use std::env; +use std::path::PathBuf; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader}; +use tokio_socks::tcp::Socks5Stream; + +const MAX_REQUEST_BYTES: usize = 128 * 1024; +const MAX_PAYLOAD_BYTES: usize = 96 * 1024; +const SMTP_TIMEOUT: Duration = Duration::from_secs(60); + +#[derive(Debug, Deserialize)] +struct Envelope { + version: u8, + entry_address: String, + payload: String, +} + +fn storage_paths() -> Result<StoragePaths> { + let path = env::var_os("YAMN_NYM_INGRESS_STORAGE") + .map(PathBuf::from) + .unwrap_or_else(|| PathBuf::from("/var/lib/yamnweb/nym-ingress")); + StoragePaths::new_from_dir(&path).context("prepare Nym ingress storage") +} + +fn allowed_entry(address: &str) -> bool { + env::var("YAMN_ALLOWED_ENTRY_ADDRESSES") + .ok() + .map(|list| list.split(',').any(|item| item.trim() == address)) + .unwrap_or(false) +} + +fn dry_run() -> bool { + matches!( + env::var("YAMN_DRY_RUN").as_deref(), + Ok("1") | Ok("true") | Ok("yes") + ) +} + +fn split_address(address: &str) -> Result<(&str, &str)> { + if address.len() > 320 || address.chars().any(|c| matches!(c, '\r' | '\n' | '\0' | ' ')) { + bail!("invalid entry address"); + } + let (local, domain) = address + .rsplit_once('@') + .context("entry address has no domain")?; + if local.is_empty() || domain.is_empty() || domain.contains('@') { + bail!("invalid entry address"); + } + Ok((local, domain)) +} + +fn validate_envelope(envelope: &Envelope) -> Result<(String, Vec<u8>)> { + if envelope.version != 1 { + bail!("unsupported envelope version"); + } + split_address(&envelope.entry_address)?; + if !allowed_entry(&envelope.entry_address) { + bail!("entry address is not allowlisted"); + } + let payload = BASE64 + .decode(&envelope.payload) + .context("invalid base64 YAMN payload")?; + if payload.is_empty() || payload.len() > MAX_PAYLOAD_BYTES || payload.contains(&0) { + bail!("invalid YAMN payload size or contents"); + } + Ok((envelope.entry_address.clone(), payload)) +} + +async fn smtp_read_response<R: AsyncRead + Unpin>(reader: &mut BufReader<R>) -> Result<u16> { + let mut line = String::new(); + loop { + line.clear(); + let read = reader.read_line(&mut line).await.context("read SMTP response")?; + if read == 0 { + bail!("SMTP connection closed"); + } + if line.len() < 3 { + bail!("malformed SMTP response"); + } + let parsed: u16 = line[..3].parse().context("invalid SMTP response code")?; + if line.as_bytes().get(3) != Some(&b'-') { + return Ok(parsed); + } + } +} + +async fn smtp_command<S: AsyncRead + AsyncWrite + Unpin>( + reader: &mut BufReader<S>, + command: &[u8], + expected_class: u16, +) -> Result<()> { + reader + .get_mut() + .write_all(command) + .await + .context("write SMTP command")?; + reader + .get_mut() + .flush() + .await + .context("flush SMTP command")?; + let code = smtp_read_response(reader).await?; + if code / 100 != expected_class { + bail!("SMTP command rejected with code {code}"); + } + Ok(()) +} + +async fn deliver(entry_address: &str, payload: &[u8]) -> Result<()> { + let (_, host) = split_address(entry_address)?; + let proxy = env::var("YAMN_TOR_SOCKS").unwrap_or_else(|_| "127.0.0.1:9050".to_string()); + let (proxy_host, proxy_port) = proxy + .rsplit_once(':') + .context("invalid YAMN_TOR_SOCKS")?; + let proxy_port: u16 = proxy_port.parse().context("invalid YAMN_TOR_SOCKS port")?; + let target = format!("{host}:25"); + let stream = tokio::time::timeout( + SMTP_TIMEOUT, + Socks5Stream::connect((proxy_host, proxy_port), target.as_str()), + ) + .await + .context("Tor SMTP connection timed out")? + .context("connect to YAMN SMTP entry through Tor")?; + let mut reader = BufReader::new(stream); + let greeting = smtp_read_response(&mut reader).await?; + if greeting / 100 != 2 { + bail!("YAMN SMTP greeting rejected with code {greeting}"); + } + + let from = env::var("YAMN_INGRESS_FROM").unwrap_or_else(|_| "<>".to_string()); + if from.chars().any(|c| matches!(c, '\r' | '\n' | ' ')) { + bail!("invalid YAMN_INGRESS_FROM"); + } + smtp_command(&mut reader, b"EHLO yamn-nym-ingress\r\n", 2).await?; + smtp_command( + &mut reader, + format!("MAIL FROM:{from}\r\n").as_bytes(), + 2, + ) + .await?; + smtp_command( + &mut reader, + format!("RCPT TO:<{entry_address}>\r\n").as_bytes(), + 2, + ) + .await?; + smtp_command(&mut reader, b"DATA\r\n", 3).await?; + + for line in payload.split_inclusive(|byte| *byte == b'\n') { + if line.starts_with(b".") { + reader + .get_mut() + .write_all(b".") + .await + .context("dot-stuff SMTP payload")?; + } + reader + .get_mut() + .write_all(line) + .await + .context("write SMTP payload")?; + } + if !payload.ends_with(b"\n") { + reader + .get_mut() + .write_all(b"\r\n") + .await + .context("terminate SMTP payload")?; + } + reader + .get_mut() + .write_all(b".\r\n") + .await + .context("finish SMTP payload")?; + reader + .get_mut() + .flush() + .await + .context("flush SMTP payload")?; + let accepted = smtp_read_response(&mut reader).await?; + if accepted / 100 != 2 { + bail!("YAMN SMTP server rejected payload with code {accepted}"); + } + let _ = smtp_command(&mut reader, b"QUIT\r\n", 2).await; + Ok(()) +} + +async fn run() -> Result<()> { + let disconnected = MixnetClientBuilder::new_with_default_storage(storage_paths()?) + .await + .context("prepare Nym client")? + .build() + .context("build Nym client")?; + let mut client = disconnected + .connect_to_mixnet() + .await + .context("connect ingress to Nym mixnet")?; + eprintln!("yamn-nym-ingress address: {}", client.nym_address()); + + while let Some(messages) = client.wait_for_messages().await { + for message in messages { + if message.message.len() > MAX_REQUEST_BYTES { + eprintln!("discarded oversized Nym message"); + continue; + } + let envelope: Envelope = match serde_json::from_slice(&message.message) { + Ok(value) => value, + Err(_) => { + eprintln!("discarded malformed Nym envelope"); + continue; + } + }; + let (entry_address, payload) = match validate_envelope(&envelope) { + Ok(value) => value, + Err(error) => { + eprintln!("discarded invalid YAMN envelope: {error}"); + continue; + } + }; + if dry_run() { + eprintln!("accepted YAMN envelope in dry-run mode"); + continue; + } + if let Err(error) = deliver(&entry_address, &payload).await { + eprintln!("YAMN delivery failed: {error:#}"); + } else { + eprintln!("delivered YAMN envelope to allowlisted entry"); + } + } + } + client.disconnect().await; + Ok(()) +} + +#[tokio::main] +async fn main() { + if let Err(error) = run().await { + eprintln!("yamn-nym-ingress: {error:#}"); + std::process::exit(1); + } +} |