summaryrefslogtreecommitdiffstats
path: root/nym/src
diff options
context:
space:
mode:
Diffstat (limited to 'nym/src')
-rw-r--r--nym/src/main.rs320
1 files changed, 287 insertions, 33 deletions
diff --git a/nym/src/main.rs b/nym/src/main.rs
index b906fc6..cc6e4a0 100644
--- a/nym/src/main.rs
+++ b/nym/src/main.rs
@@ -4,15 +4,26 @@ use nym_sdk::mixnet::{MixnetClientBuilder, MixnetMessageSender, Recipient, Stora
use serde::{Deserialize, Serialize};
use std::env;
use std::io::{self, Read};
-use std::path::PathBuf;
+use std::os::unix::fs::{FileTypeExt, PermissionsExt};
+use std::path::{Path, PathBuf};
use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{UnixListener, UnixStream};
+use tokio::sync::mpsc;
const MAX_PAYLOAD_BYTES: usize = 64 * 1024;
+const MAX_REQUEST_BYTES: usize = 70 * 1024;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(120);
const SEND_TIMEOUT: Duration = Duration::from_secs(60);
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(240);
+const DEFAULT_QUEUE_CAPACITY: usize = 8;
+const MAX_QUEUE_CAPACITY: usize = 64;
+const DEFAULT_SOCKET_PATH: &str = "/run/yamnweb/nym-sender.sock";
// send_plain_message queues the message for the background mixnet task. Keep
// the client alive long enough for that task to flush the message before the
-// process disconnects.
+// one-shot compatibility process disconnects. Daemon mode keeps one client
+// connected continuously, so it does not need a per-message grace period.
const FLUSH_GRACE: Duration = Duration::from_secs(180);
#[derive(Debug, Deserialize)]
@@ -28,6 +39,34 @@ struct Response<'a> {
error: Option<&'a str>,
}
+fn validate_request(request: &Request) -> Result<()> {
+ if request.payload.is_empty() || request.payload.len() > MAX_PAYLOAD_BYTES {
+ bail!("invalid envelope size");
+ }
+ if request.payload.as_bytes().contains(&0) {
+ bail!("envelope contains a NUL byte");
+ }
+ if request.entry_address.is_empty()
+ || request.entry_address.len() > 320
+ || request.entry_address.chars().any(char::is_whitespace)
+ || request.entry_address.as_bytes().contains(&0)
+ {
+ bail!("invalid entry address");
+ }
+ Ok(())
+}
+
+fn ingress_payload(request: &Request) -> Result<String> {
+ validate_request(request)?;
+ let raw_payload = BASE64.encode(request.payload.as_bytes());
+ let request_payload = serde_json::json!({
+ "version": 1,
+ "entry_address": request.entry_address,
+ "payload": raw_payload,
+ });
+ serde_json::to_string(&request_payload).context("encode ingress envelope")
+}
+
fn respond(response: Response<'_>) -> Result<()> {
serde_json::to_writer(io::stdout(), &response).context("encode response")?;
println!();
@@ -35,8 +74,7 @@ fn respond(response: Response<'_>) -> Result<()> {
}
fn configured_recipient() -> Result<Recipient> {
- let value = env::var("YAMN_NYM_RECIPIENT")
- .context("YAMN_NYM_RECIPIENT is not configured")?;
+ let value = env::var("YAMN_NYM_RECIPIENT").context("YAMN_NYM_RECIPIENT is not configured")?;
value
.parse::<Recipient>()
.map_err(|_| anyhow::anyhow!("invalid configured Nym recipient"))
@@ -50,26 +88,7 @@ fn storage_paths() -> Result<StoragePaths> {
}
async fn send(request: Request) -> Result<()> {
- if request.payload.is_empty() || request.payload.len() > MAX_PAYLOAD_BYTES {
- bail!("invalid envelope size");
- }
- if request.payload.as_bytes().contains(&0)
- || request
- .entry_address
- .chars()
- .any(|c| matches!(c, '\r' | '\n' | '\0'))
- {
- bail!("envelope contains a NUL byte");
- }
-
- let raw_payload = BASE64
- .encode(request.payload.as_bytes());
- let request_payload = serde_json::json!({
- "version": 1,
- "entry_address": request.entry_address,
- "payload": raw_payload,
- });
- let request_payload = serde_json::to_string(&request_payload).context("encode ingress envelope")?;
+ let request_payload = ingress_payload(&request)?;
let recipient = configured_recipient()?;
let paths = storage_paths()?;
@@ -78,7 +97,7 @@ async fn send(request: Request) -> Result<()> {
.context("prepare Nym client")?
.build()
.context("build Nym client")?;
- let mut client = tokio::time::timeout(CONNECT_TIMEOUT, disconnected.connect_to_mixnet())
+ let client = tokio::time::timeout(CONNECT_TIMEOUT, disconnected.connect_to_mixnet())
.await
.context("Nym connection timed out")?
.context("connect to Nym mixnet")?;
@@ -96,17 +115,214 @@ async fn send(request: Request) -> Result<()> {
Ok(())
}
+fn queue_capacity() -> Result<usize> {
+ let Some(value) = env::var_os("YAMN_NYM_QUEUE_CAPACITY") else {
+ return Ok(DEFAULT_QUEUE_CAPACITY);
+ };
+ let value = value
+ .to_str()
+ .context("YAMN_NYM_QUEUE_CAPACITY is not valid UTF-8")?
+ .parse::<usize>()
+ .context("YAMN_NYM_QUEUE_CAPACITY must be an integer")?;
+ if !(1..=MAX_QUEUE_CAPACITY).contains(&value) {
+ bail!("YAMN_NYM_QUEUE_CAPACITY must be between 1 and {MAX_QUEUE_CAPACITY}");
+ }
+ Ok(value)
+}
+
+fn socket_path() -> Result<PathBuf> {
+ let path = env::var_os("YAMN_NYM_SOCKET")
+ .map(PathBuf::from)
+ .unwrap_or_else(|| PathBuf::from(DEFAULT_SOCKET_PATH));
+ if path.parent() != Some(Path::new("/run/yamnweb")) || path.file_name().is_none() {
+ bail!("YAMN_NYM_SOCKET must be directly inside /run/yamnweb");
+ }
+ Ok(path)
+}
+
+fn remove_stale_socket(path: &Path) -> Result<()> {
+ match std::fs::symlink_metadata(path) {
+ Ok(metadata) if metadata.file_type().is_socket() => {
+ std::fs::remove_file(path).context("remove stale Nym sender socket")?;
+ }
+ Ok(_) => bail!("refusing to replace a non-socket Nym sender path"),
+ Err(error) if error.kind() == io::ErrorKind::NotFound => {}
+ Err(error) => return Err(error).context("inspect Nym sender socket"),
+ }
+ Ok(())
+}
+
+struct SocketFile(PathBuf);
+
+impl Drop for SocketFile {
+ fn drop(&mut self) {
+ let _ = std::fs::remove_file(&self.0);
+ }
+}
+
+async fn write_socket_response(
+ stream: &mut UnixStream,
+ success: bool,
+ error: Option<&'static str>,
+) -> Result<()> {
+ let response =
+ serde_json::to_vec(&Response { success, error }).context("encode queue response")?;
+ stream
+ .write_all(&response)
+ .await
+ .context("write queue response")?;
+ stream
+ .write_all(b"\n")
+ .await
+ .context("finish queue response")?;
+ Ok(())
+}
+
+async fn accept_request(mut stream: UnixStream, sender: mpsc::Sender<Request>) -> Result<()> {
+ let mut input = Vec::new();
+ tokio::time::timeout(
+ REQUEST_TIMEOUT,
+ (&mut stream)
+ .take((MAX_REQUEST_BYTES + 1) as u64)
+ .read_to_end(&mut input),
+ )
+ .await
+ .context("queue request timed out")?
+ .context("read queue request")?;
+ if input.len() > MAX_REQUEST_BYTES {
+ write_socket_response(&mut stream, false, Some("Invalid Nym request")).await?;
+ return Ok(());
+ }
+ let request: Request = match serde_json::from_slice(&input) {
+ Ok(request) => request,
+ Err(_) => {
+ write_socket_response(&mut stream, false, Some("Invalid Nym request")).await?;
+ return Ok(());
+ }
+ };
+ if validate_request(&request).is_err() {
+ write_socket_response(&mut stream, false, Some("Invalid Nym request")).await?;
+ return Ok(());
+ }
+ match sender.try_send(request) {
+ Ok(()) => write_socket_response(&mut stream, true, None).await?,
+ Err(mpsc::error::TrySendError::Full(_)) => {
+ write_socket_response(&mut stream, false, Some("Nym queue is full")).await?
+ }
+ Err(mpsc::error::TrySendError::Closed(_)) => {
+ write_socket_response(&mut stream, false, Some("Nym sender is unavailable")).await?
+ }
+ }
+ Ok(())
+}
+
+async fn run_daemon() -> Result<()> {
+ let capacity = queue_capacity()?;
+ let path = socket_path()?;
+ let recipient = configured_recipient()?;
+ let paths = storage_paths()?;
+ let disconnected = MixnetClientBuilder::new_with_default_storage(paths)
+ .await
+ .context("prepare Nym client storage")?
+ .build()
+ .context("build Nym client")?;
+ let client = tokio::time::timeout(CONNECT_TIMEOUT, disconnected.connect_to_mixnet())
+ .await
+ .context("Nym connection timed out")?
+ .context("connect to Nym mixnet")?;
+
+ remove_stale_socket(&path)?;
+ let listener = UnixListener::bind(&path).context("bind Nym sender socket")?;
+ let _socket_file = SocketFile(path.clone());
+ std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))
+ .context("restrict Nym sender socket permissions")?;
+
+ let (sender, mut receiver) = mpsc::channel::<Request>(capacity);
+ let mut worker = tokio::spawn(async move {
+ while let Some(request) = receiver.recv().await {
+ let request_payload = ingress_payload(&request)?;
+ tokio::time::timeout(
+ SEND_TIMEOUT,
+ client.send_plain_message(recipient, request_payload),
+ )
+ .await
+ .context("Nym send timed out")?
+ .context("send envelope through Nym")?;
+ }
+ tokio::time::sleep(FLUSH_GRACE).await;
+ client.disconnect().await;
+ Ok::<(), anyhow::Error>(())
+ });
+
+ let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+ .context("install shutdown signal handler")?;
+ let mut worker_running = true;
+ let result: Result<()> = loop {
+ tokio::select! {
+ accepted = listener.accept() => {
+ let (stream, _) = accepted.context("accept Nym queue connection")?;
+ let request_sender = sender.clone();
+ tokio::spawn(async move {
+ let _ = accept_request(stream, request_sender).await;
+ });
+ }
+ signal = tokio::signal::ctrl_c() => {
+ signal.context("wait for shutdown signal")?;
+ break Ok(());
+ }
+ _ = terminate.recv() => {
+ break Ok(());
+ }
+ worker_result = &mut worker => {
+ worker_running = false;
+ break match worker_result {
+ Ok(Ok(())) => Err(anyhow::anyhow!("Nym sender worker stopped unexpectedly")),
+ Ok(Err(error)) => Err(error).context("process Nym sender queue"),
+ Err(error) => Err(error).context("join Nym sender worker"),
+ };
+ }
+ }
+ };
+
+ drop(listener);
+ drop(sender);
+ if worker_running {
+ match tokio::time::timeout(SHUTDOWN_TIMEOUT, &mut worker).await {
+ Err(_) => {
+ worker.abort();
+ let _ = worker.await;
+ return Err(anyhow::anyhow!(
+ "timed out draining the in-memory Nym queue"
+ ));
+ }
+ Ok(Err(error)) => return Err(error).context("join Nym sender worker"),
+ Ok(Ok(Err(error))) => return Err(error).context("process Nym sender queue"),
+ Ok(Ok(Ok(()))) => {}
+ }
+ }
+ result
+}
+
+async fn run_oneshot() -> Result<()> {
+ let mut input = String::new();
+ io::stdin()
+ .read_to_string(&mut input)
+ .context("read request")?;
+ let request: Request = serde_json::from_str(&input).context("decode request")?;
+ send(request).await
+}
+
#[tokio::main]
async fn main() {
- let result = async {
- let mut input = String::new();
- io::stdin()
- .read_to_string(&mut input)
- .context("read request")?;
- let request: Request = serde_json::from_str(&input).context("decode request")?;
- send(request).await
+ if env::args().nth(1).as_deref() == Some("--daemon") {
+ if let Err(error) = run_daemon().await {
+ eprintln!("yamn-nym-submit: daemon failed: {error:#}");
+ std::process::exit(1);
+ }
+ return;
}
- .await;
+
+ let result = run_oneshot().await;
match result {
Ok(()) => {
@@ -125,3 +341,41 @@ async fn main() {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn valid_request() -> Request {
+ Request {
+ entry_address: "yamn@example.org".to_string(),
+ payload: "opaque envelope".to_string(),
+ }
+ }
+
+ #[test]
+ fn validates_request_boundaries() {
+ assert!(validate_request(&valid_request()).is_ok());
+
+ let mut request = valid_request();
+ request.entry_address = "yamn@example.org\r\nBcc: leak@example.org".to_string();
+ assert!(validate_request(&request).is_err());
+
+ let mut request = valid_request();
+ request.payload.clear();
+ assert!(validate_request(&request).is_err());
+ }
+
+ #[test]
+ fn ingress_payload_contains_only_transport_fields() {
+ let encoded = ingress_payload(&valid_request()).expect("encode request");
+ let value: serde_json::Value = serde_json::from_str(&encoded).expect("valid JSON");
+ assert_eq!(value["version"], 1);
+ assert_eq!(value["entry_address"], "yamn@example.org");
+ assert_eq!(
+ value["payload"],
+ BASE64.encode("opaque envelope".as_bytes())
+ );
+ assert_eq!(value.as_object().expect("object").len(), 3);
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2026-08-30 06:01:58 +0000