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
|
use anyhow::{bail, Context, Result};
use base64::{engine::general_purpose::STANDARD as BASE64, Engine};
use nym_sdk::mixnet::{MixnetClientBuilder, MixnetMessageSender, Recipient, StoragePaths};
use serde::{Deserialize, Serialize};
use std::env;
use std::io::{self, Read};
use std::path::PathBuf;
use std::time::Duration;
const MAX_PAYLOAD_BYTES: usize = 64 * 1024;
const CONNECT_TIMEOUT: Duration = Duration::from_secs(120);
const SEND_TIMEOUT: Duration = Duration::from_secs(60);
// 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.
const FLUSH_GRACE: Duration = Duration::from_secs(180);
#[derive(Debug, Deserialize)]
struct Request {
entry_address: String,
payload: String,
}
#[derive(Debug, Serialize)]
struct Response<'a> {
success: bool,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<&'a str>,
}
fn respond(response: Response<'_>) -> Result<()> {
serde_json::to_writer(io::stdout(), &response).context("encode response")?;
println!();
Ok(())
}
fn configured_recipient() -> Result<Recipient> {
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"))
}
fn storage_paths() -> Result<StoragePaths> {
let directory = env::var_os("YAMN_NYM_STORAGE")
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from("/var/lib/yamnweb/nym-client"));
StoragePaths::new_from_dir(&directory).context("prepare Nym client storage")
}
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 recipient = configured_recipient()?;
let paths = storage_paths()?;
let disconnected = MixnetClientBuilder::new_with_default_storage(paths)
.await
.context("prepare Nym client")?
.build()
.context("build Nym client")?;
let mut client = tokio::time::timeout(CONNECT_TIMEOUT, disconnected.connect_to_mixnet())
.await
.context("Nym connection timed out")?
.context("connect to Nym mixnet")?;
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(())
}
#[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
}
.await;
match result {
Ok(()) => {
let _ = respond(Response {
success: true,
error: None,
});
}
Err(error) => {
eprintln!("yamn-nym-submit: {error:#}");
let _ = respond(Response {
success: false,
error: Some("Nym submission failed"),
});
std::process::exit(1);
}
}
}
|
