netget/server/vxlan/
mod.rs

1//! VXLAN / Geneve / GRE (gretap, in UDP) tunnel endpoint: decapsulates what other tunnel
2//! endpoints send, and the model plays the hosts on the overlay — ARP, ping and UDP — with
3//! replies built in Rust and sent back through the tunnel. **Deliberately silent**: an overlay
4//! IP with no host never answers, and a backend failure reads the same way.
5pub mod actions;
6pub mod frame;
7
8use crate::llm::action_helper::call_llm;
9use crate::llm::actions::protocol_trait::ActionResult;
10use crate::logging::emit::Log;
11use crate::protocol::{Event, SpawnContext};
12use crate::server::connection::ConnectionId;
13use actions::{FrameContext, VxlanProtocol};
14use anyhow::{Context, Result};
15use frame::{Encap, Frame, Header, Mac, Payload};
16use serde_json::json;
17use std::collections::HashMap;
18use std::net::{Ipv4Addr, SocketAddr};
19use std::sync::{Arc, Mutex};
20use tokio::net::UdpSocket;
21
22#[derive(Clone)]
23struct Config {
24    encap: Encap,
25    default_mac: Mac,
26    macs: Arc<Mutex<HashMap<Ipv4Addr, Mac>>>,
27    /// GRE: the next sequence number for replies to peers that send them. Given as each reply
28    /// leaves, under this lock, so concurrent answers cannot go out of order.
29    sequence: Arc<Mutex<u32>>,
30}
31
32pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
33    let params = ctx.startup_params.as_ref();
34    let encap = Encap::parse(
35        &params
36            .map(|p| p.get_optional_string("encapsulation"))
37            .transpose()?
38            .flatten()
39            .unwrap_or_else(|| actions::DEFAULT_ENCAPSULATION.into()),
40    )?;
41    let only_vni = params
42        .map(|p| p.get_optional_u64("vni"))
43        .transpose()?
44        .flatten();
45    if let Some(v) = only_vni {
46        anyhow::ensure!(
47            v <= encap.max_vni() as u64,
48            "vni must be at most {} for {}",
49            encap.max_vni(),
50            encap.name()
51        );
52    }
53    let default_mac = frame::parse_mac(
54        &params
55            .map(|p| p.get_optional_string("overlay_mac"))
56            .transpose()?
57            .flatten()
58            .unwrap_or_else(|| frame::DEFAULT_MAC.into()),
59    )?;
60    let port = ctx.port.unwrap_or_else(|| ctx.legacy_listen_addr().port());
61    let host = ctx.host.clone().unwrap_or_else(|| "127.0.0.1".into());
62    let bind = tokio::net::lookup_host((host.as_str(), port))
63        .await
64        .with_context(|| format!("cannot resolve host {host}"))?
65        .next()
66        .with_context(|| format!("host {host} resolved to no address"))?;
67    let socket = Arc::new(
68        UdpSocket::bind(bind)
69            .await
70            .with_context(|| format!("{} failed to bind {bind}", encap.name()))?,
71    );
72    let local = socket.local_addr()?;
73    Log::new(Some(&ctx.status_tx)).info(format!(
74        "{} endpoint listening on {local}{}",
75        encap.name().to_uppercase(),
76        only_vni
77            .map(|v| format!(", {} only", frame::vni_str(encap, Some(v as u32))))
78            .unwrap_or_default()
79    ));
80    let config = Config {
81        encap,
82        default_mac,
83        macs: Default::default(),
84        sequence: Default::default(),
85    };
86    let state = ctx.state.clone();
87    let server_id = ctx.server_id;
88    let task = tokio::spawn(async move {
89        let mut buf = vec![0u8; 65_536];
90        loop {
91            let (n, from) = match socket.recv_from(&mut buf).await {
92                Ok(r) => r,
93                Err(e) => {
94                    Log::new(Some(&ctx.status_tx))
95                        .error(format!("{} receive error: {e}", encap.name()));
96                    break;
97                }
98            };
99            let parsed = frame::decap(encap, &buf[..n])
100                .and_then(|(header, inner)| Ok((header, frame::parse_frame(inner)?)));
101            let (header, frame) = match parsed {
102                Ok(p) => p,
103                Err(e) => {
104                    tracing::warn!("{} dropped a datagram from {from}: {e:#}", encap.name());
105                    continue;
106                }
107            };
108            let vni = frame::vni_str(encap, header.vni);
109            // A GRE frame with no key carries no VNI, so it is not on a filtered overlay.
110            if only_vni.is_some_and(|v| header.vni.map(u64::from) != Some(v)) {
111                tracing::debug!("{} dropped a frame for {vni} from {from}", encap.name());
112                continue;
113            }
114            let Some(event) = event_for(&frame, header.vni, from) else {
115                if let Payload::Other { what } = &frame.payload {
116                    tracing::debug!("{} from {from} {vni}: {what}, not answered", encap.name());
117                }
118                continue;
119            };
120            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
121            record(&ctx, id, local, from, n).await;
122            let (child, socket, config) = (ctx.clone(), socket.clone(), config.clone());
123            // Replies go to the sending VTEP at the tunnel port: VTEPs listen on the port
124            // they send to, not on the source port they send from.
125            let reply_to = SocketAddr::new(from.ip(), local.port());
126            let handle = tokio::spawn(async move {
127                answer(child, socket, config, header, frame, event, reply_to, id).await
128            });
129            ctx.state.register_server_task(ctx.server_id, handle).await;
130        }
131    });
132    state.register_server_task(server_id, task).await;
133    Ok(local)
134}
135
136fn event_for(f: &Frame, vni: Option<u32>, from: SocketAddr) -> Option<Event> {
137    let mut data = json!({
138        "vni": vni,
139        "vtep": from.to_string(),
140        "src_mac": frame::mac_str(&f.src_mac),
141    });
142    let ip = |a: Option<Ipv4Addr>| a.map(|a| a.to_string());
143    let t = match &f.payload {
144        Payload::Arp(arp) if arp.request => {
145            data["sender_ip"] = json!(arp.sender_ip.to_string());
146            data["target_ip"] = json!(arp.target_ip.to_string());
147            &*actions::ARP_EVENT
148        }
149        Payload::Echo {
150            request: true,
151            identifier,
152            sequence,
153            data: payload,
154        } => {
155            let (text, encoding) = frame::data_json(payload);
156            data["src_ip"] = json!(ip(f.src_ip));
157            data["dst_ip"] = json!(ip(f.dst_ip));
158            data["identifier"] = json!(identifier);
159            data["sequence"] = json!(sequence);
160            data["data"] = json!(text);
161            data["encoding"] = json!(encoding);
162            &*actions::ECHO_EVENT
163        }
164        Payload::Udp {
165            src_port,
166            dst_port,
167            data: payload,
168        } => {
169            let (text, encoding) = frame::data_json(payload);
170            data["src_ip"] = json!(ip(f.src_ip));
171            data["src_port"] = json!(src_port);
172            data["dst_ip"] = json!(ip(f.dst_ip));
173            data["dst_port"] = json!(dst_port);
174            data["data"] = json!(text);
175            data["encoding"] = json!(encoding);
176            &*actions::UDP_EVENT
177        }
178        _ => return None,
179    };
180    Some(Event::new(t, data))
181}
182
183async fn record(
184    ctx: &SpawnContext,
185    id: ConnectionId,
186    local: SocketAddr,
187    from: SocketAddr,
188    n: usize,
189) {
190    use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
191    let now = crate::utils::clock::Instant::now();
192    ctx.state
193        .add_connection_to_server(
194            ctx.server_id,
195            ConnectionState {
196                id,
197                remote_addr: from,
198                local_addr: local,
199                bytes_sent: 0,
200                bytes_received: n as u64,
201                packets_sent: 0,
202                packets_received: 1,
203                last_activity: now,
204                status: ConnectionStatus::Active,
205                status_changed_at: now,
206                protocol_info: ProtocolConnectionInfo::empty(),
207            },
208        )
209        .await;
210}
211
212#[allow(clippy::too_many_arguments)]
213async fn answer(
214    ctx: SpawnContext,
215    socket: Arc<UdpSocket>,
216    config: Config,
217    header: Header,
218    frame: Frame,
219    event: Event,
220    reply_to: SocketAddr,
221    id: ConnectionId,
222) {
223    let summary = format!(
224        "{} {} from {reply_to} {}",
225        config.encap.name(),
226        event.id(),
227        frame::vni_str(config.encap, header.vni)
228    );
229    let protocol = VxlanProtocol::for_frame(FrameContext {
230        encap: config.encap,
231        header,
232        frame,
233        default_mac: config.default_mac,
234        macs: config.macs.clone(),
235    });
236    let log = Log::new(Some(&ctx.status_tx));
237    let execution = match call_llm(
238        &ctx.llm_client,
239        &ctx.state,
240        ctx.server_id,
241        Some(id),
242        &event,
243        &protocol,
244    )
245    .await
246    {
247        Ok(e) => e,
248        Err(e) => {
249            log.error(format!(
250                "{summary} decision=fail_closed_llm_error (nothing sent: no host answers): {e}"
251            ));
252            return;
253        }
254    };
255    let mut sent = 0;
256    for result in &execution.protocol_results {
257        let items = match result {
258            ActionResult::Multiple(all) => all.iter().collect::<Vec<_>>(),
259            other => vec![other],
260        };
261        for item in items {
262            if let ActionResult::Output(bytes) = item {
263                match send(&socket, &config, bytes.clone(), reply_to).await {
264                    Ok(n) => {
265                        sent += 1;
266                        ctx.state
267                            .update_connection_stats(
268                                ctx.server_id,
269                                id,
270                                None,
271                                Some(n as u64),
272                                None,
273                                Some(1),
274                            )
275                            .await;
276                    }
277                    Err(e) => log.warn(format!("{summary}: sending to {reply_to} failed: {e}")),
278                }
279            }
280        }
281    }
282    let decision = match (sent, execution.failures.is_empty()) {
283        (0, true) => "model_silent",
284        (0, false) => "fail_closed_action_error",
285        _ => "model_answered",
286    };
287    log.info(format!("{summary} decision={decision}"));
288}
289
290/// Send one reply. A GRE reply that carries a sequence number gets the next one as it leaves:
291/// the number is written and the datagram handed to the socket under one lock, which is never
292/// held across an await (a full send buffer releases it, waits, and takes the next number).
293async fn send(
294    socket: &UdpSocket,
295    config: &Config,
296    mut datagram: Vec<u8>,
297    to: SocketAddr,
298) -> std::io::Result<usize> {
299    loop {
300        {
301            let mut next = config.sequence.lock().unwrap_or_else(|e| e.into_inner());
302            let numbered = frame::stamp_sequence(config.encap, &mut datagram, *next);
303            match socket.try_send_to(&datagram, to) {
304                Ok(n) => {
305                    if numbered {
306                        *next = next.wrapping_add(1);
307                    }
308                    return Ok(n);
309                }
310                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
311                Err(e) => return Err(e),
312            }
313        }
314        socket.writable().await?;
315    }
316}
netget/server/vxlan/mod.rs

netget/server/vxlan/
mod.rs
