netget/client/epp/
mod.rs

1//! EPP registrar client: TLS, the greeting, an optional login, then the handler's commands.
2pub mod actions;
3use crate::client::llm_budget::call_llm_for_client;
4use crate::llm::actions::client_trait::Client;
5use crate::logging::emit::Log;
6use crate::protocol::{ConnectContext, Event};
7use crate::server::epp::{wire, xml};
8use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
9use crate::state::{AccessLogOwner, ClientStatus};
10pub use actions::EppClientProtocol;
11use anyhow::{bail, ensure, Context, Result};
12use serde_json::{json, Value};
13use std::net::SocketAddr;
14use std::sync::Arc;
15use std::time::Duration;
16use tokio::io::{AsyncRead, AsyncWrite, ReadHalf, WriteHalf};
17use tokio::sync::mpsc;
18
19const TIMEOUT: Duration = Duration::from_secs(30);
20/// handler → command → response → handler … stops here (a provisioning flow is about ten deep);
21/// an injected command starts afresh.
22const MAX_FOLLOWUP_DEPTH: usize = 16;
23
24trait Stream: AsyncRead + AsyncWrite + Unpin + Send {}
25impl<T: AsyncRead + AsyncWrite + Unpin + Send> Stream for T {}
26
27struct Conn {
28    reader: ReadHalf<Box<dyn Stream>>,
29    writer: WriteHalf<Box<dyn Stream>>,
30    serial: u64,
31}
32
33impl Conn {
34    async fn read(&mut self) -> Result<Value> {
35        let frame = tokio::time::timeout(TIMEOUT, wire::read_frame(&mut self.reader))
36            .await
37            .context("the server did not answer in time")??
38            .context("the server closed the session")?;
39        wire::read_response(&xml::parse(&frame)?)
40    }
41
42    /// Send one rendered command and read its answer.
43    async fn exchange(&mut self, command: &str, element: &str) -> Result<Value> {
44        self.serial += 1;
45        let cl_trid = format!("NG-C-{:06}", self.serial);
46        wire::write_frame(
47            &mut self.writer,
48            &actions::document(command, element, &cl_trid),
49        )
50        .await?;
51        self.read().await
52    }
53}
54
55async fn open(ctx: &ConnectContext) -> Result<(Box<dyn Stream>, SocketAddr)> {
56    let p = ctx.startup_params.as_ref();
57    let get = |name| {
58        p.map(|p| p.get_optional_string(name))
59            .transpose()
60            .map(Option::flatten)
61    };
62    let tls = p
63        .map(|p| p.get_optional_bool("tls"))
64        .transpose()?
65        .flatten()
66        .unwrap_or(true);
67    let tcp = tokio::time::timeout(TIMEOUT, tokio::net::TcpStream::connect(&ctx.remote_addr))
68        .await
69        .context("connecting timed out")??;
70    let local = tcp.local_addr()?;
71    if !tls {
72        return Ok((Box::new(tcp), local));
73    }
74    let mut roots = rustls::RootCertStore::empty();
75    match get("ca_cert_path")? {
76        Some(ca) => {
77            let pem = tokio::fs::read(&ca)
78                .await
79                .with_context(|| format!("reading {ca}"))?;
80            for cert in rustls_pemfile::certs(&mut pem.as_slice()) {
81                roots.add(cert.with_context(|| format!("{ca} holds no PEM certificate"))?)?;
82            }
83            ensure!(!roots.is_empty(), "{ca} holds no PEM certificate");
84        }
85        None => roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()),
86    }
87    let _ = rustls::crypto::ring::default_provider().install_default();
88    let config = rustls::ClientConfig::builder()
89        .with_root_certificates(roots)
90        .with_no_client_auth();
91    let host = match get("server_name")? {
92        Some(n) => n,
93        None => ctx
94            .remote_addr
95            .rsplit_once(':')
96            .map(|(h, _)| h.trim_matches(['[', ']']).to_owned())
97            .context("remote_addr is host:port")?,
98    };
99    let name = rustls::pki_types::ServerName::try_from(host.clone())
100        .with_context(|| format!("{host} is not a TLS name"))?;
101    let stream = tokio::time::timeout(
102        TIMEOUT,
103        tokio_rustls::TlsConnector::from(Arc::new(config)).connect(name, tcp),
104    )
105    .await
106    .context("the TLS handshake timed out")??;
107    Ok((Box::new(stream), local))
108}
109
110pub async fn connect(ctx: ConnectContext) -> Result<SocketAddr> {
111    let p = ctx.startup_params.as_ref();
112    let get = |name| {
113        p.map(|p| p.get_optional_string(name))
114            .transpose()
115            .map(Option::flatten)
116    };
117    let credentials = match (get("client_id")?, get("password")?) {
118        (Some(c), Some(pw)) => Some((c, pw)),
119        (None, None) => None,
120        _ => bail!("client_id and password go together"),
121    };
122    let (stream, local) = open(&ctx).await?;
123    let (reader, writer) = tokio::io::split(stream);
124    let mut conn = Conn {
125        reader,
126        writer,
127        serial: 0,
128    };
129    let greeting = conn.read().await?;
130    ensure!(greeting["greeting"] == true, "the server did not greet");
131    let mut info = json!({"sv_id": greeting["sv_id"], "obj_uris": greeting["obj_uris"], "ext_uris": greeting["ext_uris"], "login": null});
132    if let Some((client_id, password)) = &credentials {
133        let r = conn
134            .exchange("login", &actions::login(client_id, password))
135            .await?;
136        let code = r["code"].as_u64().unwrap_or(0);
137        ensure!(
138            code == 1000,
139            "login refused with {code}: {}",
140            r["reason"]
141                .as_str()
142                .or(r["message"].as_str())
143                .unwrap_or_default()
144        );
145        info["login"] = json!({"code": code, "message": r["message"], "client_id": client_id});
146    }
147    Log::new(Some(&ctx.status_tx)).info(format!(
148        "EPP session with {} ({})",
149        ctx.remote_addr, greeting["sv_id"]
150    ));
151    ctx.state
152        .update_client_status(ctx.client_id, ClientStatus::Connected)
153        .await;
154    let external =
155        crate::client::command_support::register_command_channel(&ctx.state, ctx.client_id).await;
156    let (internal_tx, internal_rx) = mpsc::channel::<Value>(16);
157    let (event_tx, mut event_rx) = mpsc::channel::<Event>(16);
158    event_tx.try_send(Event::new(&actions::CONNECTED_EVENT, info))?;
159    let events_ctx = ctx.clone();
160    let dispatcher = tokio::spawn(async move {
161        let protocol = EppClientProtocol;
162        while let Some(event) = event_rx.recv().await {
163            let instruction = events_ctx
164                .state
165                .get_instruction_for_client(events_ctx.client_id)
166                .await
167                .unwrap_or_default();
168            let memory = events_ctx
169                .state
170                .get_memory_for_client(events_ctx.client_id)
171                .await
172                .unwrap_or_default();
173            match call_llm_for_client(
174                &events_ctx.llm_client,
175                &events_ctx.state,
176                events_ctx.client_id.to_string(),
177                &instruction,
178                &memory,
179                Some(&event),
180                &protocol,
181                &events_ctx.status_tx,
182            )
183            .await
184            {
185                Ok(result) => {
186                    if let Some(m) = result.memory_updates {
187                        events_ctx
188                            .state
189                            .set_memory_for_client(events_ctx.client_id, m)
190                            .await;
191                    }
192                    for action in result.actions {
193                        if internal_tx.send(action).await.is_err() {
194                            return;
195                        }
196                    }
197                }
198                Err(e) => {
199                    Log::new(Some(&events_ctx.status_tx)).warn(format!("EPP client handler: {e}"))
200                }
201            }
202        }
203    });
204    ctx.state
205        .register_client_task(ctx.client_id, dispatcher)
206        .await;
207    let session_ctx = ctx.clone();
208    let task = tokio::spawn(async move {
209        if let Err(e) = run(&session_ctx, &mut conn, external, internal_rx, &event_tx).await {
210            Log::new(Some(&session_ctx.status_tx)).warn(format!("EPP session ended: {e:#}"));
211        }
212        session_ctx
213            .state
214            .update_client_status(session_ctx.client_id, ClientStatus::Disconnected)
215            .await;
216        session_ctx
217            .state
218            .remove_client_handle(session_ctx.client_id)
219            .await;
220        let _ = session_ctx.status_tx.send("__UPDATE_UI__".into());
221    });
222    ctx.state.register_client_task(ctx.client_id, task).await;
223    Ok(local)
224}
225
226async fn run(
227    ctx: &ConnectContext,
228    conn: &mut Conn,
229    mut external: mpsc::Receiver<ClientCommand>,
230    mut internal: mpsc::Receiver<Value>,
231    events: &mpsc::Sender<Event>,
232) -> Result<()> {
233    let mut depth = 0usize;
234    loop {
235        let (action, command) = tokio::select! {
236            c = external.recv() => match c { Some(c) => (c.action.clone(), Some(c)), None => return Ok(()) },
237            a = internal.recv() => match a { Some(a) => (a, None), None => return Ok(()) },
238        };
239        let rendered = EppClientProtocol
240            .execute_action(action.clone())
241            .and_then(|_| actions::render(&action));
242        let (outcome, done) = match rendered {
243            Err(e) => (
244                ClientSendOutcome::Rejected {
245                    error: e.to_string(),
246                },
247                false,
248            ),
249            Ok((verb, object, element)) => {
250                let answer = conn.exchange(&verb, &element).await;
251                let answer = match answer {
252                    Ok(a) => a,
253                    Err(e) => {
254                        if let Some(c) = command {
255                            crate::client::command_support::reply(
256                                c,
257                                Ok(ClientSendOutcome::Rejected {
258                                    error: format!("{e:#}"),
259                                }),
260                            );
261                        }
262                        return Err(e);
263                    }
264                };
265                let mut response = answer;
266                if response["greeting"] == true {
267                    response = json!({"code": 1000, "message": "greeting", "data": response});
268                }
269                response["command"] = json!(verb);
270                response["object"] = json!(object);
271                depth = if command.is_some() { 0 } else { depth + 1 };
272                if depth <= MAX_FOLLOWUP_DEPTH {
273                    events
274                        .send(Event::new(&actions::RESPONSE_EVENT, response))
275                        .await
276                        .ok();
277                } else {
278                    Log::new(Some(&ctx.status_tx)).warn(format!("EPP client: follow-up depth {MAX_FOLLOWUP_DEPTH} reached; the response is not raised"));
279                }
280                (
281                    ClientSendOutcome::Sent {
282                        bytes_sent: element.len(),
283                    },
284                    verb == "logout",
285                )
286            }
287        };
288        if let Some(c) = command {
289            ctx.state
290                .record_access_log(
291                    AccessLogOwner::Client(ctx.client_id.as_u32()),
292                    "EPP",
293                    None,
294                    "injected_action",
295                    json!({"type": action["type"]}),
296                    vec![serde_json::to_value(&outcome).unwrap_or(Value::Null)],
297                )
298                .await;
299            crate::client::command_support::reply(c, Ok(outcome));
300        }
301        if done {
302            return Ok(());
303        }
304    }
305}
netget/client/epp/mod.rs

netget/client/epp/
mod.rs
