netget/server/sunrpc/
mod.rs

1//! ONC RPC portmapper / rpcbind server on TCP and UDP (the same port). Rust decodes every
2//! call and answers the RPC-level errors, NULL and GETTIME; the model answers lookups and
3//! dumps with mappings and decides registrations.
4pub mod actions;
5pub mod wire;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, EventType, SpawnContext};
11use crate::server::{
12    accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS},
13    connection::ConnectionId,
14};
15use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
16use anyhow::Result;
17use serde_json::{json, Value};
18use std::net::{IpAddr, SocketAddr};
19use std::sync::Arc;
20use std::time::Duration;
21use tokio::io::AsyncWriteExt;
22use tokio::net::{TcpStream, UdpSocket};
23use wire::{Accepted, Mapping, Reader, Writer};
24
25/// A TCP connection that sends no complete call for this long is closed.
26pub const IDLE_TIMEOUT: Duration = Duration::from_secs(120);
27/// UDP calls answered at once; past it, datagrams wait in the socket buffer.
28pub const MAX_UDP_IN_FLIGHT: usize = 64;
29
30pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
31    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
32    let local = listener.local_addr()?;
33    // UDP on the same port as TCP, as every portmapper does.
34    let udp = Arc::new(UdpSocket::bind(local).await?);
35    Log::new(Some(&ctx.status_tx)).info(format!(
36        "Portmapper (PMAP 2, RPCBIND 3-4) listening on {local} tcp and udp"
37    ));
38    let server_id = ctx.server_id;
39    let state = ctx.state.clone();
40
41    let udp_ctx = ctx.clone();
42    let udp_task = tokio::spawn(async move {
43        let permits = Arc::new(tokio::sync::Semaphore::new(MAX_UDP_IN_FLIGHT));
44        let mut buf = vec![0u8; 65_536];
45        loop {
46            let Ok((n, peer)) = udp.recv_from(&mut buf).await else {
47                break;
48            };
49            let Ok(permit) = permits.clone().acquire_owned().await else {
50                break;
51            };
52            let msg = buf[..n].to_vec();
53            let ctx = udp_ctx.clone();
54            let udp = udp.clone();
55            udp_ctx
56                .state
57                .spawn_server_task(server_id, async move {
58                    let _permit = permit;
59                    if let Some(reply) = handle(&ctx, None, &msg, "udp", local.ip()).await {
60                        let _ = udp.send_to(&reply, peer).await;
61                    }
62                })
63                .await;
64        }
65    });
66    state.register_server_task(server_id, udp_task).await;
67
68    let accept =
69        tokio::spawn(async move {
70            let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
71            loop {
72                let (stream, peer, permit) =
73                    match accept_bounded(&listener, &limiter, b"", "SunRPC", Some(&ctx.status_tx))
74                        .await
75                    {
76                        Ok(v) => v,
77                        Err(_) => break,
78                    };
79                let child = ctx.clone();
80                ctx.state
81                    .spawn_server_task(server_id, async move {
82                        let _permit = permit;
83                        connection(child, stream, peer, local).await;
84                    })
85                    .await;
86            }
87        });
88    state.register_server_task(server_id, accept).await;
89    Ok(local)
90}
91
92async fn connection(ctx: SpawnContext, stream: TcpStream, remote: SocketAddr, local: SocketAddr) {
93    let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
94    let now = crate::utils::clock::Instant::now();
95    ctx.state
96        .add_connection_to_server(
97            ctx.server_id,
98            ConnectionState {
99                id,
100                remote_addr: remote,
101                local_addr: local,
102                bytes_sent: 0,
103                bytes_received: 0,
104                packets_sent: 0,
105                packets_received: 0,
106                last_activity: now,
107                status: ConnectionStatus::Active,
108                status_changed_at: now,
109                protocol_info: ProtocolConnectionInfo::empty(),
110            },
111        )
112        .await;
113    let local_ip = stream.local_addr().map(|a| a.ip()).unwrap_or(local.ip());
114    let (mut r, mut w) = stream.into_split();
115    let reason = loop {
116        let msg = match tokio::time::timeout(IDLE_TIMEOUT, wire::read_record(&mut r)).await {
117            Err(_) => break "idle".to_string(),
118            Ok(Ok(None)) => break "closed by the peer".to_string(),
119            Ok(Err(e)) => break format!("refused: {e:#}"),
120            Ok(Ok(Some(m))) => m,
121        };
122        ctx.state
123            .update_connection_stats(
124                ctx.server_id,
125                id,
126                Some(msg.len() as u64),
127                None,
128                Some(1),
129                None,
130            )
131            .await;
132        if let Some(reply) = handle(&ctx, Some(id), &msg, "tcp", local_ip).await {
133            if w.write_all(&wire::record(&reply)).await.is_err() {
134                break "write failed".into();
135            }
136            ctx.state
137                .update_connection_stats(
138                    ctx.server_id,
139                    id,
140                    None,
141                    Some(reply.len() as u64),
142                    None,
143                    Some(1),
144                )
145                .await;
146        }
147    };
148    let _ = w.shutdown().await;
149    ctx.state
150        .update_connection_status(ctx.server_id, id, ConnectionStatus::Closed)
151        .await;
152    Log::new(Some(&ctx.status_tx)).debug(format!("SunRPC connection {id} ended: {reason}"));
153    let _ = ctx.status_tx.send("__UPDATE_UI__".into());
154}
155
156fn outcome(ctx: &SpawnContext, op: &str, decision: &str) {
157    let line = format!("SunRPC operation={op} decision={decision}");
158    let log = Log::new(Some(&ctx.status_tx));
159    if decision.starts_with("fail_closed") || decision == "model_silent" {
160        log.error(line)
161    } else {
162        log.info(line)
163    }
164}
165
166/// Ask the model; its one portmapper answer, or None when the call must fail (SYSTEM_ERR).
167async fn ask(
168    ctx: &SpawnContext,
169    conn: Option<ConnectionId>,
170    event_type: &'static EventType,
171    data: Value,
172    op: &str,
173) -> Option<(String, Value)> {
174    let event = Event::new(event_type, data);
175    let result = match call_llm(
176        &ctx.llm_client,
177        &ctx.state,
178        ctx.server_id,
179        conn,
180        &event,
181        &actions::SunRpcProtocol,
182    )
183    .await
184    {
185        Ok(r) if r.failures.is_empty() => r,
186        Ok(_) => {
187            outcome(ctx, op, "fail_closed_invalid_reply");
188            return None;
189        }
190        Err(_) => {
191            outcome(ctx, op, "fail_closed_llm_error");
192            return None;
193        }
194    };
195    let mut answers = Vec::new();
196    let mut stack = result.protocol_results;
197    while let Some(r) = stack.pop() {
198        match r {
199            ActionResult::Custom { name, data } if name.starts_with("sunrpc_") => {
200                answers.push((name, data))
201            }
202            ActionResult::Multiple(items) => stack.extend(items),
203            _ => {}
204        }
205    }
206    match answers.len() {
207        1 => {
208            outcome(ctx, op, "model_answer");
209            answers.pop()
210        }
211        0 => {
212            outcome(ctx, op, "model_silent");
213            None
214        }
215        _ => {
216            outcome(ctx, op, "fail_closed_invalid_reply");
217            None
218        }
219    }
220}
221
222/// The model's mappings for a query; None on failure.
223async fn mappings(
224    ctx: &SpawnContext,
225    conn: Option<ConnectionId>,
226    data: Value,
227    op: &str,
228) -> Option<Vec<Mapping>> {
229    let (name, answer) = ask(ctx, conn, &actions::QUERY_EVENT, data, op).await?;
230    if name != actions::MAPPINGS {
231        outcome(ctx, op, "fail_closed_invalid_reply");
232        return None;
233    }
234    answer["mappings"]
235        .as_array()?
236        .iter()
237        .map(wire::mapping_from_json)
238        .collect::<Result<Vec<_>>>()
239        .ok()
240}
241
242/// The mappings a lookup returns: exact version first; unless `exact`, any version of the
243/// program when that version is not registered (what rpcbind does for GETPORT/GETADDR).
244fn pick(ms: &[Mapping], program: u32, version: u32, netid: &str, exact: bool) -> Vec<Mapping> {
245    let of = |v: Option<u32>| -> Vec<Mapping> {
246        ms.iter()
247            .filter(|m| {
248                m.program == program
249                    && v.is_none_or(|v| m.version == v)
250                    && (netid.is_empty() || m.netid == netid)
251            })
252            .cloned()
253            .collect()
254    };
255    let hit = of(Some(version));
256    if hit.is_empty() && !exact {
257        of(None)
258    } else {
259        hit
260    }
261}
262
263async fn register(
264    ctx: &SpawnContext,
265    conn: Option<ConnectionId>,
266    data: Value,
267    op: &str,
268) -> Accepted {
269    match ask(ctx, conn, &actions::REGISTER_EVENT, data, op).await {
270        Some((name, _)) if name == actions::ACCEPT => {
271            Accepted::Success(Writer::default().u32(1).0.clone())
272        }
273        Some((name, _)) if name == actions::REJECT => {
274            Accepted::Success(Writer::default().u32(0).0.clone())
275        }
276        Some(_) => {
277            outcome(ctx, op, "fail_closed_invalid_reply");
278            Accepted::SystemErr
279        }
280        None => Accepted::SystemErr,
281    }
282}
283
284/// Answer one RPC message; None when there is nothing to answer (a reply, or no xid).
285pub async fn handle(
286    ctx: &SpawnContext,
287    conn: Option<ConnectionId>,
288    msg: &[u8],
289    transport: &str,
290    local: IpAddr,
291) -> Option<Vec<u8>> {
292    let call = match wire::parse_call(msg) {
293        Ok(Some(c)) => c,
294        Ok(None) => return None,
295        Err(_) => {
296            return wire::xid_of(msg).map(|x| wire::accepted_reply(x, &Accepted::GarbageArgs))
297        }
298    };
299    if call.rpc_version != 2 {
300        return Some(wire::rpc_mismatch_reply(call.xid));
301    }
302    let result = if call.program != wire::PROGRAM {
303        Accepted::ProgUnavail
304    } else if !(wire::LOW_VERSION..=wire::HIGH_VERSION).contains(&call.version) {
305        Accepted::ProgMismatch(wire::LOW_VERSION, wire::HIGH_VERSION)
306    } else {
307        let credentials = wire::credential_json(call.cred_flavor, call.cred);
308        match procedure(ctx, conn, &call, transport, local, credentials).await {
309            Ok(a) => a,
310            Err(_) => Accepted::GarbageArgs,
311        }
312    };
313    Some(wire::accepted_reply(call.xid, &result))
314}
315
316async fn procedure(
317    ctx: &SpawnContext,
318    conn: Option<ConnectionId>,
319    call: &wire::Call<'_>,
320    transport: &str,
321    local: IpAddr,
322    credentials: Value,
323) -> Result<Accepted> {
324    let v = call.version;
325    let mut r = Reader::new(call.args);
326    let base =
327        |procedure: &str| json!({"transport": transport, "rpc_version": v, "procedure": procedure});
328    Ok(match (v, call.procedure) {
329        (_, 0) => Accepted::Success(Vec::new()),
330        // PMAP v2
331        (2, 1 | 2) => {
332            let (program, version, prot, port) = wire::read_pmap(&mut r)?;
333            let set = call.procedure == 1;
334            let Some(netid) = wire::netid_of(prot) else {
335                // Not TCP or UDP: nothing to register.
336                return Ok(Accepted::Success(Writer::default().u32(0).0.clone()));
337            };
338            let mut data = json!({"transport": transport, "rpc_version": v,
339                "operation": if set { "set" } else { "unset" }, "program": program,
340                "program_name": wire::program_name(program), "program_version": version,
341                "protocol": netid, "credentials": credentials});
342            if set {
343                data["port"] = json!(port);
344            }
345            register(ctx, conn, data, if set { "set" } else { "unset" }).await
346        }
347        (2, 3) => {
348            let (program, version, prot, _) = wire::read_pmap(&mut r)?;
349            let netid = wire::netid_of(prot).unwrap_or("");
350            let mut data = base("getport");
351            data["program"] = json!(program);
352            data["program_name"] = json!(wire::program_name(program));
353            data["program_version"] = json!(version);
354            data["protocol"] = json!(netid);
355            match mappings(ctx, conn, data, "getport").await {
356                None => Accepted::SystemErr,
357                Some(ms) => {
358                    let port = if netid.is_empty() {
359                        0
360                    } else {
361                        pick(&ms, program, version, netid, false)
362                            .first()
363                            .map(|m| u32::from(m.port))
364                            .unwrap_or(0)
365                    };
366                    Accepted::Success(Writer::default().u32(port).0.clone())
367                }
368            }
369        }
370        (2, 4) => match mappings(ctx, conn, base("dump"), "dump").await {
371            None => Accepted::SystemErr,
372            Some(ms) => Accepted::Success(wire::pmaplist(&ms)),
373        },
374        // RPCBIND v3 and v4
375        (3 | 4, 1 | 2) => {
376            let b = wire::read_rpcb(&mut r)?;
377            let set = call.procedure == 1;
378            let mut data = json!({"transport": transport, "rpc_version": v,
379                "operation": if set { "set" } else { "unset" }, "program": b.program,
380                "program_name": wire::program_name(b.program), "program_version": b.version,
381                "protocol": b.netid, "owner": b.owner, "credentials": credentials});
382            if set {
383                if wire::protocol_of(&b.netid).is_none() {
384                    return Ok(Accepted::Success(Writer::default().u32(0).0.clone()));
385                }
386                let addr = wire::parse_uaddr(&b.addr)?;
387                data["port"] = json!(addr.port());
388                data["address"] = json!(b.addr);
389            }
390            register(ctx, conn, data, if set { "set" } else { "unset" }).await
391        }
392        (3 | 4, 3) | (4, 9) => {
393            let b = wire::read_rpcb(&mut r)?;
394            let exact = call.procedure == 9;
395            let procedure = if exact { "getversaddr" } else { "getaddr" };
396            let mut data = base(procedure);
397            data["program"] = json!(b.program);
398            data["program_name"] = json!(wire::program_name(b.program));
399            data["program_version"] = json!(b.version);
400            data["protocol"] = json!(b.netid);
401            match mappings(ctx, conn, data, procedure).await {
402                None => Accepted::SystemErr,
403                Some(ms) => {
404                    let netid = if b.netid.is_empty() {
405                        transport
406                    } else {
407                        b.netid.as_str()
408                    };
409                    let addr = pick(&ms, b.program, b.version, netid, exact)
410                        .first()
411                        .map(|m| wire::uaddr(wire::host_for(&m.netid, local), m.port))
412                        .unwrap_or_default();
413                    Accepted::Success(Writer::default().string(&addr).0.clone())
414                }
415            }
416        }
417        (3 | 4, 4) => match mappings(ctx, conn, base("dump"), "dump").await {
418            None => Accepted::SystemErr,
419            Some(ms) => Accepted::Success(wire::rpcblist(&ms, local)),
420        },
421        (3 | 4, 6) => {
422            let now = crate::utils::clock::SystemTime::now()
423                .duration_since(crate::utils::clock::UNIX_EPOCH)
424                .map(|d| d.as_secs() as u32)
425                .unwrap_or(0);
426            Accepted::Success(Writer::default().u32(now).0.clone())
427        }
428        (4, 11) => {
429            let b = wire::read_rpcb(&mut r)?;
430            let mut data = base("getaddrlist");
431            data["program"] = json!(b.program);
432            data["program_name"] = json!(wire::program_name(b.program));
433            data["program_version"] = json!(b.version);
434            data["protocol"] = json!(b.netid);
435            match mappings(ctx, conn, data, "getaddrlist").await {
436                None => Accepted::SystemErr,
437                Some(ms) => Accepted::Success(wire::entry_list(
438                    &pick(&ms, b.program, b.version, "", true),
439                    local,
440                )),
441            }
442        }
443        _ => Accepted::ProcUnavail,
444    })
445}
netget/server/sunrpc/mod.rs

netget/server/sunrpc/
mod.rs
