netget/server/rdap/
mod.rs

1//! RDAP over HTTP/1.1: Rust parses and normalizes every RFC 9082 query and enforces the
2//! RFC 9083 envelope on every answer; the handler supplies the registration data.
3pub mod actions;
4pub mod query;
5
6use crate::llm::action_helper::call_llm;
7use crate::llm::actions::protocol_trait::ActionResult;
8use crate::logging::emit::Log;
9use crate::protocol::{Event, SpawnContext};
10use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
11use crate::server::connection::ConnectionId;
12use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
13use anyhow::{bail, ensure, Context, Result};
14use bytes::Bytes;
15use http_body_util::Full;
16use hyper::{
17    body::Incoming, header, server::conn::http1, service::service_fn, Method, Request, Response,
18    StatusCode,
19};
20use hyper_util::rt::{TokioIo, TokioTimer};
21use serde_json::Value;
22use std::{convert::Infallible, net::SocketAddr, sync::Arc, time::Duration};
23
24pub const HEADER_TIMEOUT: Duration = Duration::from_secs(30);
25pub const MAX_HEADER_BYTES: usize = 16 * 1024;
26pub const MAX_HEADERS: usize = 64;
27const CAP_REFUSAL: &[u8] = b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\nRetry-After: 5\r\n\r\n";
28
29struct Shared {
30    ctx: SpawnContext,
31    base: String,
32}
33
34pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
35    let base = ctx
36        .startup_params
37        .as_ref()
38        .map(|p| p.get_optional_string("base_path"))
39        .transpose()?
40        .flatten()
41        .unwrap_or_else(|| query::DEFAULT_BASE_PATH.into());
42    ensure!(
43        base.starts_with('/') && base.len() <= 128 && !base.contains(['?', '#', ' ']),
44        "base_path must be an absolute path without query"
45    );
46    let base = base.trim_end_matches('/').to_owned();
47    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
48    let local = listener.local_addr()?;
49    Log::new(Some(&ctx.status_tx)).info(format!("RDAP listening on http://{local}{base}/"));
50    let shared = Arc::new(Shared {
51        ctx: ctx.clone(),
52        base,
53    });
54    let server_id = ctx.server_id;
55    let accept = tokio::spawn(async move {
56        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
57        loop {
58            let (stream, peer, permit) = match accept_bounded(
59                &listener,
60                &limiter,
61                CAP_REFUSAL,
62                "RDAP",
63                Some(&shared.ctx.status_tx),
64            )
65            .await
66            {
67                Ok(v) => v,
68                Err(_) => break,
69            };
70            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
71            let now = crate::utils::clock::Instant::now();
72            shared
73                .ctx
74                .state
75                .add_connection_to_server(
76                    server_id,
77                    ConnectionState {
78                        id,
79                        remote_addr: peer,
80                        local_addr: local,
81                        bytes_sent: 0,
82                        bytes_received: 0,
83                        packets_sent: 0,
84                        packets_received: 0,
85                        last_activity: now,
86                        status: ConnectionStatus::Active,
87                        status_changed_at: now,
88                        protocol_info: ProtocolConnectionInfo::empty(),
89                    },
90                )
91                .await;
92            let child = shared.clone();
93            shared
94                .ctx
95                .state
96                .spawn_server_task(server_id, async move {
97                    let _permit = permit;
98                    let service_shared = child.clone();
99                    let service = service_fn(move |request| {
100                        let shared = service_shared.clone();
101                        async move { Ok::<_, Infallible>(handle(&shared, id, request).await) }
102                    });
103                    let mut builder = http1::Builder::new();
104                    builder
105                        .timer(TokioTimer::new())
106                        .header_read_timeout(HEADER_TIMEOUT)
107                        .max_headers(MAX_HEADERS)
108                        .max_buf_size(MAX_HEADER_BYTES);
109                    if let Err(e) = builder
110                        .serve_connection(TokioIo::new(stream), service)
111                        .await
112                    {
113                        Log::new(Some(&child.ctx.status_tx))
114                            .debug(format!("RDAP connection {id}: {e}"));
115                    }
116                    child
117                        .ctx
118                        .state
119                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
120                        .await;
121                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
122                })
123                .await;
124        }
125    });
126    ctx.state.register_server_task(server_id, accept).await;
127    Ok(local)
128}
129
130fn json_response(status: u16, body: &Value, head: bool) -> Response<Full<Bytes>> {
131    let bytes = serde_json::to_vec(body).unwrap_or_default();
132    let mut r = Response::new(Full::new(if head {
133        Bytes::new()
134    } else {
135        Bytes::from(bytes)
136    }));
137    *r.status_mut() = StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);
138    let h = r.headers_mut();
139    h.insert(
140        header::CONTENT_TYPE,
141        header::HeaderValue::from_static(query::MEDIA_TYPE),
142    );
143    // RFC 7480 §5.6: RDAP is meant to be readable from browser scripts.
144    h.insert(
145        header::ACCESS_CONTROL_ALLOW_ORIGIN,
146        header::HeaderValue::from_static("*"),
147    );
148    r
149}
150
151fn error_response(status: u16, title: &str, head: bool) -> Response<Full<Bytes>> {
152    json_response(status, &query::error_body(status, title, &[]), head)
153}
154
155fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
156    let summary = format!("RDAP connection {id} query={operation} decision={decision}");
157    let log = Log::new(Some(&ctx.status_tx));
158    if decision.starts_with("fail_closed_") || decision == "model_silent" {
159        log.error(summary);
160    } else {
161        log.info(summary);
162    }
163}
164
165async fn decide(shared: &Shared, id: ConnectionId, event: Event) -> Result<Value> {
166    let ctx = &shared.ctx;
167    let result = match call_llm(
168        &ctx.llm_client,
169        &ctx.state,
170        ctx.server_id,
171        Some(id),
172        &event,
173        &actions::RdapProtocol,
174    )
175    .await
176    {
177        Ok(r) => r,
178        Err(e) => {
179            outcome(ctx, id, event.id(), "fail_closed_llm_error");
180            return Err(e);
181        }
182    };
183    if !result.failures.is_empty() {
184        outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
185        bail!("RDAP handler supplied an invalid action");
186    }
187    let mut found = None;
188    let mut pending = result.protocol_results;
189    while let Some(r) = pending.pop() {
190        match r {
191            ActionResult::Custom { name, data } if name == "rdap_response" => {
192                if found.replace(data).is_some() {
193                    outcome(ctx, id, event.id(), "fail_closed_invalid_reply");
194                    bail!("RDAP handler supplied more than one response");
195                }
196            }
197            ActionResult::Multiple(items) => pending.extend(items),
198            _ => {}
199        }
200    }
201    found
202        .context("RDAP handler did not answer")
203        .inspect_err(|_| outcome(ctx, id, event.id(), "model_silent"))
204}
205
206async fn handle(
207    shared: &Shared,
208    id: ConnectionId,
209    request: Request<Incoming>,
210) -> Response<Full<Bytes>> {
211    let ctx = &shared.ctx;
212    let received = request.uri().to_string().len()
213        + request
214            .headers()
215            .iter()
216            .map(|(k, v)| k.as_str().len() + v.len() + 4)
217            .sum::<usize>();
218    ctx.state
219        .update_connection_stats(
220            ctx.server_id,
221            id,
222            Some(received as u64),
223            None,
224            Some(1),
225            None,
226        )
227        .await;
228    let head = request.method() == Method::HEAD;
229    if request.method() != Method::GET && !head {
230        let mut r = error_response(405, "Method Not Allowed", false);
231        r.headers_mut()
232            .insert(header::ALLOW, header::HeaderValue::from_static("GET, HEAD"));
233        return r;
234    }
235    let path = request.uri().path().to_owned();
236    let Some(relative) = path
237        .strip_prefix(shared.base.as_str())
238        .filter(|r| shared.base.is_empty() || r.is_empty() || r.starts_with('/'))
239    else {
240        return error_response(404, "Not Found", head);
241    };
242    let query = match query::parse(relative, request.uri().query()) {
243        Ok(q) => q,
244        // RFC 7480 §5.4: a malformed query is 400 and costs no handler call.
245        Err(e) => {
246            outcome(ctx, id, relative, "protocol_refusal");
247            return json_response(
248                400,
249                &query::error_body(400, "Bad Request", &[e.to_string()]),
250                head,
251            );
252        }
253    };
254    let mut data = query.to_event();
255    data["method"] = Value::from(if head { "HEAD" } else { "GET" });
256    let label = data["query_type"].as_str().unwrap_or("rdap").to_owned();
257    let response = match decide(shared, id, Event::new(&actions::QUERY_EVENT, data)).await {
258        Err(e) => {
259            let (status, title) = if crate::utils::WireFailure::classify(&e).is_overloaded() {
260                (503, "Service Unavailable")
261            } else {
262                (500, "Internal Server Error")
263            };
264            let mut r = error_response(status, title, head);
265            if status == 503 {
266                r.headers_mut()
267                    .insert(header::RETRY_AFTER, header::HeaderValue::from_static("5"));
268            }
269            r
270        }
271        Ok(v) => match query::answer(&query, &v) {
272            Ok(query::Answer::Body { status, body }) => {
273                outcome(
274                    ctx,
275                    id,
276                    &label,
277                    if status < 300 {
278                        "model_answer"
279                    } else {
280                        "model_reject"
281                    },
282                );
283                json_response(status, &body, head)
284            }
285            Ok(query::Answer::Redirect(url)) => {
286                outcome(ctx, id, &label, "model_redirect");
287                let mut r = Response::new(Full::new(Bytes::new()));
288                *r.status_mut() = StatusCode::FOUND;
289                if let Ok(v) = header::HeaderValue::from_str(&url) {
290                    r.headers_mut().insert(header::LOCATION, v);
291                }
292                r.headers_mut().insert(
293                    header::ACCESS_CONTROL_ALLOW_ORIGIN,
294                    header::HeaderValue::from_static("*"),
295                );
296                r
297            }
298            Err(e) => {
299                outcome(ctx, id, &label, "fail_closed_invalid_reply");
300                Log::new(Some(&ctx.status_tx))
301                    .warn(format!("RDAP connection {id}: handler answer refused: {e}"));
302                error_response(500, "Internal Server Error", head)
303            }
304        },
305    };
306    let sent = hyper::body::Body::size_hint(response.body())
307        .exact()
308        .unwrap_or(0);
309    ctx.state
310        .update_connection_stats(ctx.server_id, id, None, Some(sent), None, Some(1))
311        .await;
312    response
313}
netget/server/rdap/mod.rs

netget/server/rdap/
mod.rs
