1pub mod actions;
4pub mod wire;
5pub mod xml;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, SpawnContext};
11use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
12use crate::server::connection::ConnectionId;
13use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
14use anyhow::Result;
15use serde_json::{json, Value};
16use std::collections::BTreeMap;
17use std::{net::SocketAddr, sync::Arc, time::Duration};
18use tokio::io::{AsyncRead, AsyncWrite};
19use wire::{Command, Refusal};
20
21pub const DEFAULT_SERVER_ID: &str = "NetGet EPP";
22pub const IDLE_TIMEOUT: Duration = Duration::from_secs(600);
23const TLS_TIMEOUT: Duration = Duration::from_secs(10);
24const MAX_FAILED_LOGINS: u32 = 3;
25const OBJECTS: &[&str] = &["domain", "host", "contact"];
26
27struct Shared {
28 ctx: SpawnContext,
29 server_id: String,
30 clients: BTreeMap<String, String>,
31 idle: Duration,
32}
33
34pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
35 let p = ctx.startup_params.as_ref();
36 let get = |name| {
37 p.map(|p| p.get_optional_string(name))
38 .transpose()
39 .map(Option::flatten)
40 };
41 let server_id = get("server_id")?.unwrap_or_else(|| DEFAULT_SERVER_ID.into());
42 anyhow::ensure!(
43 (3..=64).contains(&server_id.len()),
44 "server_id is 3 to 64 characters"
45 );
46 let mut clients = BTreeMap::new();
47 if let Some(m) = p
48 .map(|p| p.get_optional_object("clients"))
49 .transpose()?
50 .flatten()
51 {
52 anyhow::ensure!(m.len() <= 64, "at most 64 clients");
53 for (k, v) in m {
54 let v = v
55 .as_str()
56 .ok_or_else(|| anyhow::anyhow!("client passwords are strings"))?;
57 clients.insert(k.clone(), v.to_owned());
58 }
59 }
60 let idle = p
61 .map(|p| p.get_optional_u64("idle_timeout_secs"))
62 .transpose()?
63 .flatten()
64 .unwrap_or(IDLE_TIMEOUT.as_secs());
65 anyhow::ensure!((1..=3600).contains(&idle), "idle_timeout_secs is 1 to 3600");
66 let tls_on = p
67 .map(|p| p.get_optional_bool("tls"))
68 .transpose()?
69 .flatten()
70 .unwrap_or(true);
71 let tls = match (tls_on, get("tls_cert_file")?, get("tls_key_file")?) {
72 (false, None, None) => None,
73 (false, _, _) => anyhow::bail!("tls is false but a certificate was given"),
74 (true, Some(c), Some(k)) => Some(tokio_rustls::TlsAcceptor::from(
75 crate::server::tls_cert_manager::load_tls_config_from_files(&c, &k)?,
76 )),
77 (true, None, None) => {
78 let spec = crate::server::tls_cert_manager::CertificateSpec {
79 common_name: "localhost".into(),
80 san_dns_names: vec!["localhost".into()],
81 validity_days: 30,
82 organization: Some("NetGet".into()),
83 organizational_unit: Some("EPP".into()),
84 };
85 let (cert, key) = crate::server::tls_cert_manager::generate_self_signed_cert(&spec)?;
86 let pem = cert.pem();
87 ctx.state
88 .with_server_mut(ctx.server_id, |s| {
89 s.set_protocol_field("certificate_pem".into(), json!(pem))
90 })
91 .await;
92 Some(tokio_rustls::TlsAcceptor::from(
93 crate::server::tls_cert_manager::create_rustls_server_config(&cert, &key)?,
94 ))
95 }
96 _ => anyhow::bail!("tls_cert_file and tls_key_file go together"),
97 };
98 let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
99 let local = listener.local_addr()?;
100 Log::new(Some(&ctx.status_tx)).info(format!(
101 "EPP registry on {local} ({})",
102 if tls.is_some() { "TLS" } else { "plain TCP" }
103 ));
104 let shared = Arc::new(Shared {
105 ctx: ctx.clone(),
106 server_id,
107 clients,
108 idle: Duration::from_secs(idle),
109 });
110 let server_id = ctx.server_id;
111 let accept = tokio::spawn(async move {
112 let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
113 loop {
114 let (stream, peer, permit) =
115 match accept_bounded(&listener, &limiter, b"", "EPP", Some(&shared.ctx.status_tx))
116 .await
117 {
118 Ok(v) => v,
119 Err(_) => break,
120 };
121 let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
122 let now = crate::utils::clock::Instant::now();
123 shared
124 .ctx
125 .state
126 .add_connection_to_server(
127 server_id,
128 ConnectionState {
129 id,
130 remote_addr: peer,
131 local_addr: local,
132 bytes_sent: 0,
133 bytes_received: 0,
134 packets_sent: 0,
135 packets_received: 0,
136 last_activity: now,
137 status: ConnectionStatus::Active,
138 status_changed_at: now,
139 protocol_info: ProtocolConnectionInfo::empty(),
140 },
141 )
142 .await;
143 let child = shared.clone();
144 let tls = tls.clone();
145 shared
146 .ctx
147 .state
148 .spawn_server_task(server_id, async move {
149 let _permit = permit;
150 let served = match tls {
151 Some(acceptor) => {
152 match tokio::time::timeout(TLS_TIMEOUT, acceptor.accept(stream)).await {
153 Ok(Ok(s)) => session(&child, id, s).await,
154 Ok(Err(e)) => Err(anyhow::anyhow!("TLS handshake: {e}")),
155 Err(_) => Err(anyhow::anyhow!("TLS handshake timed out")),
156 }
157 }
158 None => session(&child, id, stream).await,
159 };
160 if let Err(e) = served {
161 Log::new(Some(&child.ctx.status_tx))
162 .debug(format!("EPP connection {id}: {e:#}"));
163 }
164 child
165 .ctx
166 .state
167 .update_connection_status(server_id, id, ConnectionStatus::Closed)
168 .await;
169 let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
170 })
171 .await;
172 }
173 });
174 ctx.state.register_server_task(server_id, accept).await;
175 Ok(local)
176}
177
178fn log_decision(ctx: &SpawnContext, id: ConnectionId, what: &str, decision: &str, code: u16) {
179 let line = format!("EPP connection {id} {what} decision={decision} code={code}");
180 let log = Log::new(Some(&ctx.status_tx));
181 if decision.starts_with("fail_closed_") || decision == "model_silent" {
182 log.error(line);
183 } else {
184 log.info(line);
185 }
186}
187
188struct Answer {
189 code: u16,
190 reason: Option<String>,
191 res_data: String,
192}
193
194impl Answer {
195 fn of(code: u16, reason: impl Into<String>) -> Self {
196 Self {
197 code,
198 reason: Some(reason.into()).filter(|r: &String| !r.is_empty()),
199 res_data: String::new(),
200 }
201 }
202}
203
204async fn session<S: AsyncRead + AsyncWrite + Unpin>(
205 shared: &Shared,
206 id: ConnectionId,
207 stream: S,
208) -> Result<()> {
209 let ctx = &shared.ctx;
210 let (mut reader, mut writer) = tokio::io::split(stream);
211 let now = || {
212 chrono::Utc::now()
213 .format("%Y-%m-%dT%H:%M:%S%.3fZ")
214 .to_string()
215 };
216 let greeting = wire::greeting(&shared.server_id, &now(), OBJECTS);
217 wire::write_frame(&mut writer, &greeting).await?;
218 let mut client: Option<String> = None;
219 let mut failed_logins = 0u32;
220 let mut serial = 0u64;
221 loop {
222 let frame = match tokio::time::timeout(shared.idle, wire::read_frame(&mut reader)).await {
223 Err(_) => {
224 log_decision(ctx, id, "idle", "protocol_refusal", 2500);
225 return Ok(());
226 }
227 Ok(Ok(None)) => return Ok(()),
228 Ok(Ok(Some(f))) => f,
229 Ok(Err(e)) => {
230 log_decision(ctx, id, "frame", "protocol_refusal", 2500);
231 serial += 1;
232 let sv = format!("NG-{id}-{serial}");
233 let _ = wire::write_frame(
234 &mut writer,
235 &wire::response(
236 2500,
237 wire::message(2500),
238 Some(&e.to_string()),
239 "",
240 None,
241 &sv,
242 ),
243 )
244 .await;
245 return Ok(());
246 }
247 };
248 ctx.state
249 .update_connection_stats(
250 ctx.server_id,
251 id,
252 Some(frame.len() as u64 + 4),
253 None,
254 Some(1),
255 None,
256 )
257 .await;
258 serial += 1;
259 let sv_trid = format!("NG-{id}-{serial}");
260 let parsed = xml::parse(&frame)
261 .map_err(|e| Refusal {
262 code: 2001,
263 reason: e.to_string(),
264 })
265 .and_then(|root| wire::command(&root));
266 let (answer, cl_trid, close) = match parsed {
267 Err(r) => {
268 log_decision(ctx, id, "command", "protocol_refusal", r.code);
269 (Answer::of(r.code, r.reason), None, false)
270 }
271 Ok(frame) => {
272 let cl = frame.cl_trid.clone();
273 match frame.command {
274 Command::Hello => {
275 wire::write_frame(
276 &mut writer,
277 &wire::greeting(&shared.server_id, &now(), OBJECTS),
278 )
279 .await?;
280 continue;
281 }
282 Command::Login {
283 client_id,
284 password,
285 version,
286 obj_uris,
287 } => {
288 if client.is_some() {
289 (Answer::of(2002, "already logged in"), cl, false)
290 } else if version != "1.0" {
291 (Answer::of(2100, "only version 1.0 is served"), cl, false)
292 } else if let Some(u) = obj_uris
293 .iter()
294 .find(|u| wire::object_of(u).is_none() && u.as_str() != xml::EPP)
295 {
296 (Answer::of(2307, format!("{u} is not served")), cl, false)
297 } else if !shared.clients.is_empty()
298 && shared.clients.get(&client_id) != Some(&password)
299 {
300 failed_logins += 1;
301 if failed_logins >= MAX_FAILED_LOGINS {
302 log_decision(ctx, id, "login", "protocol_refusal", 2501);
303 (Answer::of(2501, "too many failed logins"), cl, true)
304 } else {
305 log_decision(ctx, id, "login", "protocol_refusal", 2200);
306 (Answer::of(2200, ""), cl, false)
307 }
308 } else {
309 Log::new(Some(&ctx.status_tx))
310 .info(format!("EPP connection {id} logged in as {client_id}"));
311 client = Some(client_id);
312 (Answer::of(1000, ""), cl, false)
313 }
314 }
315 Command::Logout => (Answer::of(1500, ""), cl, true),
316 _ if client.is_none() => (Answer::of(2002, "log in first"), cl, false),
317 Command::Poll { op, .. } if op == "req" => (Answer::of(1300, ""), cl, false),
318 Command::Poll { .. } => (Answer::of(2303, "no such message"), cl, false),
319 Command::Object(command, object, fields) => {
320 let answer = ask(
321 shared,
322 id,
323 &command,
324 &object,
325 fields,
326 client.as_deref().unwrap_or_default(),
327 cl.as_deref(),
328 )
329 .await;
330 (answer, cl, false)
331 }
332 }
333 }
334 };
335 let xml = wire::response(
336 answer.code,
337 wire::message(answer.code),
338 answer.reason.as_deref(),
339 &answer.res_data,
340 cl_trid.as_deref(),
341 &sv_trid,
342 );
343 wire::write_frame(&mut writer, &xml).await?;
344 ctx.state
345 .update_connection_stats(
346 ctx.server_id,
347 id,
348 None,
349 Some(xml.len() as u64 + 4),
350 None,
351 Some(1),
352 )
353 .await;
354 if close {
355 return Ok(());
356 }
357 }
358}
359
360async fn ask(
361 shared: &Shared,
362 id: ConnectionId,
363 command: &str,
364 object: &str,
365 fields: Value,
366 client: &str,
367 cl_trid: Option<&str>,
368) -> Answer {
369 let ctx = &shared.ctx;
370 let what = format!("{object}:{command}");
371 let event = Event::new(
372 &actions::COMMAND_EVENT,
373 json!({"command": command, "object": object, "fields": fields, "client_id": client, "cl_trid": cl_trid}),
374 );
375 let result = match call_llm(
376 &ctx.llm_client,
377 &ctx.state,
378 ctx.server_id,
379 Some(id),
380 &event,
381 &actions::EppProtocol,
382 )
383 .await
384 {
385 Ok(r) => r,
386 Err(_) => {
387 log_decision(ctx, id, &what, "fail_closed_llm_error", 2400);
388 return Answer::of(2400, "");
389 }
390 };
391 if !result.failures.is_empty() {
392 log_decision(ctx, id, &what, "fail_closed_invalid_reply", 2400);
393 return Answer::of(2400, "");
394 }
395 let Some(action) = result.protocol_results.into_iter().find_map(|r| match r {
396 ActionResult::Custom { data, .. } => Some(data),
397 _ => None,
398 }) else {
399 log_decision(ctx, id, &what, "model_silent", 2400);
400 return Answer::of(2400, "");
401 };
402 match wire::res_data(&action, command, object) {
403 Ok(res_data) => {
404 let default = if action["type"] == "epp_result" {
405 2400
406 } else {
407 1000
408 };
409 let code = action["code"].as_u64().map(|c| c as u16).unwrap_or(default);
410 log_decision(
411 ctx,
412 id,
413 &what,
414 if code >= 2000 {
415 "model_reject"
416 } else {
417 "model_answer"
418 },
419 code,
420 );
421 Answer {
422 code,
423 reason: action["reason"]
424 .as_str()
425 .map(str::to_owned)
426 .filter(|r| !r.is_empty()),
427 res_data,
428 }
429 }
430 Err(e) => {
431 log_decision(ctx, id, &what, "fail_closed_invalid_reply", 2400);
432 Log::new(Some(&ctx.status_tx)).warn(format!("EPP answer for {what} unusable: {e}"));
433 Answer::of(2400, "")
434 }
435 }
436}