pub mod actions;
pub mod codec;
use anyhow::{anyhow, bail, Context, Result};
use std::net::{Ipv6Addr, SocketAddr, SocketAddrV6};
use std::sync::{Arc, Mutex};
use tokio::net::UdpSocket;
use tokio::sync::mpsc;
use tracing::{debug, error, info, trace, warn};
use crate::llm::action_helper::call_llm;
use crate::llm::ollama_client::OllamaClient;
use crate::protocol::Event;
use crate::server::connection::ConnectionId;
use crate::state::app_state::AppState;
use crate::{console_debug, console_error, console_info, console_trace};
use actions::{
NdpProtocol, DEFAULT_LINK_LAYER_ADDRESS, DEFAULT_LINK_LOCAL, NDP_MESSAGE_RESULT,
NDP_NEIGHBOR_ADVERTISEMENT_EVENT, NDP_NEIGHBOR_SOLICITATION_EVENT, NDP_REDIRECT_RECEIVED_EVENT,
NDP_ROUTER_ADVERTISEMENT_RECEIVED_EVENT, NDP_ROUTER_SOLICITATION_EVENT, NO_RESPONSE_ACTION,
};
use codec::NdpMessage;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TransportKind {
Raw,
Udp,
}
#[derive(Debug, Clone)]
struct NdpConfig {
transport: TransportKind,
udp_peer: Option<SocketAddr>,
link_local_address: Ipv6Addr,
link_layer_address: Option<[u8; 6]>,
}
impl NdpConfig {
fn from_params(params: Option<&crate::protocol::StartupParams>) -> Result<Self> {
let default_link_local = DEFAULT_LINK_LOCAL
.parse::<Ipv6Addr>()
.expect("DEFAULT_LINK_LOCAL is a literal IPv6 address");
let Some(params) = params else {
return Ok(Self {
transport: TransportKind::Raw,
udp_peer: None,
link_local_address: default_link_local,
link_layer_address: None,
});
};
let transport = match params.get_optional_string("transport")?.as_deref() {
None | Some("raw") => TransportKind::Raw,
Some("udp") => TransportKind::Udp,
Some(other) => bail!(
"transport must be \"raw\" (a real ICMPv6 raw socket) or \"udp\" (testing \
transport carrying addressed ICMPv6 messages in datagrams), got '{other}'"
),
};
let udp_peer = match params.get_optional_string("udp_peer")? {
None => None,
Some(text) => {
if transport != TransportKind::Udp {
bail!(
"udp_peer was given but transport is \"raw\", where it would do nothing. \
Set transport to \"udp\" or drop udp_peer."
);
}
Some(text.parse::<SocketAddr>().with_context(|| {
format!("udp_peer must be HOST:PORT (e.g. 127.0.0.1:34567), got '{text}'")
})?)
}
};
let link_local_address = match params.get_optional_string("link_local_address")? {
None => default_link_local,
Some(text) => text.trim().parse::<Ipv6Addr>().with_context(|| {
format!("link_local_address must be an IPv6 address, got '{text}'")
})?,
};
let link_layer_address = match params.get_optional_string("link_layer_address")? {
None => None,
Some(text) => Some(codec::parse_mac(&text)?),
};
Ok(Self {
transport,
udp_peer,
link_local_address,
link_layer_address,
})
}
}
#[derive(Clone)]
enum MessageSink {
Raw {
socket: Arc<socket2::Socket>,
scope_id: u32,
},
Udp {
socket: Arc<UdpSocket>,
configured_peer: Option<SocketAddr>,
last_peer: Arc<Mutex<Option<SocketAddr>>>,
},
}
impl MessageSink {
async fn send(&self, source: Ipv6Addr, destination: Ipv6Addr, message: &[u8]) -> Result<usize> {
match self {
MessageSink::Raw { socket, scope_id } => {
let target = SocketAddrV6::new(destination, 0, 0, *scope_id);
Ok(socket.send_to(message, &socket2::SockAddr::from(target))?)
}
MessageSink::Udp {
socket,
configured_peer,
last_peer,
} => {
let peer = match configured_peer {
Some(p) => Some(*p),
None => *last_peer.lock().expect("ndp peer mutex poisoned"),
};
let peer = peer.ok_or_else(|| {
anyhow!(
"no UDP peer to transmit to: nothing has been received yet and no \
udp_peer startup parameter was given"
)
})?;
let datagram = codec::encode_addressed(source, destination, message);
Ok(socket.send_to(&datagram, peer).await?)
}
}
}
}
struct NdpContext {
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
protocol: Arc<NdpProtocol>,
server_id: crate::state::ServerId,
link_local_address: Ipv6Addr,
link_layer_address: [u8; 6],
}
pub struct NdpServer;
impl NdpServer {
pub async fn spawn_with_llm_actions(
interface: Option<String>,
listen_addr: SocketAddr,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
startup_params: Option<crate::protocol::StartupParams>,
) -> Result<SocketAddr> {
let config = NdpConfig::from_params(startup_params.as_ref())?;
match config.transport {
TransportKind::Raw => {
let interface = interface.ok_or_else(|| {
anyhow!(
"NDP transport \"raw\" needs an interface to send link-local and \
multicast messages on (pass `interface`), or use transport \"udp\" for \
an unprivileged test run"
)
})?;
Self::spawn_raw(
interface, config, llm_client, app_state, status_tx, server_id,
)
.await
}
TransportKind::Udp => {
Self::spawn_udp(
listen_addr,
config,
llm_client,
app_state,
status_tx,
server_id,
)
.await
}
}
}
fn build_context(
config: &NdpConfig,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
interface_mac: Option<[u8; 6]>,
) -> Arc<NdpContext> {
let link_layer_address = config
.link_layer_address
.or(interface_mac)
.unwrap_or_else(|| {
codec::parse_mac(DEFAULT_LINK_LAYER_ADDRESS)
.expect("DEFAULT_LINK_LAYER_ADDRESS is a literal link-layer address")
});
Arc::new(NdpContext {
llm_client,
app_state,
status_tx,
protocol: Arc::new(NdpProtocol::new()),
server_id,
link_local_address: config.link_local_address,
link_layer_address,
})
}
async fn spawn_udp(
listen_addr: SocketAddr,
config: NdpConfig,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
) -> Result<SocketAddr> {
let socket = Arc::new(UdpSocket::bind(listen_addr).await.with_context(|| {
format!("failed to bind the NDP UDP test transport to {listen_addr}")
})?);
let local_addr = socket.local_addr()?;
console_info!(
status_tx,
"NDP listening on {} as {} (transport=udp — a TEST transport carrying addressed \
ICMPv6 messages in datagrams; no real IPv6 stack speaks it)",
local_addr,
config.link_local_address
);
let ctx = Self::build_context(
&config,
llm_client,
app_state.clone(),
status_tx.clone(),
server_id,
None,
);
let sink = MessageSink::Udp {
socket: socket.clone(),
configured_peer: config.udp_peer,
last_peer: Arc::new(Mutex::new(None)),
};
let receive_ctx = ctx.clone();
let receive_sink = sink.clone();
let loop_state = app_state.clone();
let receive_handle = tokio::spawn(async move {
let mut buffer = vec![0u8; 65535];
loop {
match socket.recv_from(&mut buffer).await {
Ok((n, peer)) => {
if let MessageSink::Udp { last_peer, .. } = &receive_sink {
*last_peer.lock().expect("ndp peer mutex poisoned") = Some(peer);
}
let datagram = buffer[..n].to_vec();
let ctx = receive_ctx.clone();
let sink = receive_sink.clone();
let task_owner = loop_state.clone();
task_owner
.spawn_server_task(server_id, async move {
Self::handle_datagram(&datagram, peer, local_addr, ctx, sink).await;
})
.await;
}
Err(e) => {
console_error!(receive_ctx.status_tx, "NDP UDP receive error: {}", e);
break;
}
}
}
});
app_state
.register_server_task(server_id, receive_handle)
.await;
Ok(local_addr)
}
async fn spawn_raw(
interface: String,
config: NdpConfig,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
) -> Result<SocketAddr> {
use socket2::{Domain, Protocol, Socket, Type};
let (scope_id, interface_mac) = interface_details(&interface).ok_or_else(|| {
anyhow!(
"no interface named '{interface}' on this host. NDP is a link-local protocol: \
link-local and multicast destinations need the interface's scope id, so there \
is no useful default. Use transport \"udp\" for an unprivileged test run."
)
})?;
console_info!(
status_tx,
"NDP opening a raw ICMPv6 socket on interface {} (index {}) as {}",
interface,
scope_id,
config.link_local_address
);
let ctx = Self::build_context(
&config,
llm_client,
app_state.clone(),
status_tx.clone(),
server_id,
interface_mac,
);
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<Result<()>>();
let stop = crate::utils::StopSignal::new();
let stop_in_loop = stop.clone();
let loop_interface = interface.clone();
let loop_ctx = ctx.clone();
let (sink_tx, sink_rx) = tokio::sync::oneshot::channel::<MessageSink>();
tokio::task::spawn_blocking(move || {
let open = || -> Result<Socket> {
let socket = Socket::new(Domain::IPV6, Type::RAW, Some(Protocol::ICMPV6)).context(
"failed to create a raw ICMPv6 socket (needs root, or CAP_NET_RAW on \
Linux). Use transport \"udp\" for an unprivileged test run.",
)?;
socket
.set_nonblocking(true)
.context("failed to put the raw ICMPv6 socket in non-blocking mode")?;
socket
.set_unicast_hops_v6(codec::NDP_HOP_LIMIT)
.context("failed to set the IPv6 unicast hop limit to 255")?;
socket
.set_multicast_hops_v6(codec::NDP_HOP_LIMIT)
.context("failed to set the IPv6 multicast hop limit to 255")?;
socket.set_multicast_if_v6(scope_id).with_context(|| {
format!("failed to send IPv6 multicast out of interface '{loop_interface}'")
})?;
Ok(socket)
};
let socket = match open() {
Ok(socket) => {
let _ = ready_tx.send(Ok(()));
Arc::new(socket)
}
Err(e) => {
console_error!(loop_ctx.status_tx, "NDP raw socket startup failed: {:#}", e);
let _ = ready_tx.send(Err(e));
return;
}
};
let loop_sink = MessageSink::Raw {
socket: socket.clone(),
scope_id,
};
let _ = sink_tx.send(loop_sink.clone());
let runtime = tokio::runtime::Handle::current();
let mut buffer = [std::mem::MaybeUninit::<u8>::uninit(); 65535];
loop {
if stop_in_loop.is_stopped() {
console_info!(
loop_ctx.status_tx,
"NDP receive loop on {} stopping",
loop_interface
);
break;
}
match socket.recv_from(&mut buffer) {
Ok((n, from)) => {
let message =
unsafe { std::slice::from_raw_parts(buffer.as_ptr() as *const u8, n) }
.to_vec();
let Some(source) = from.as_socket_ipv6().map(|a| *a.ip()) else {
debug!("NDP ignoring a packet from a non-IPv6 source address");
continue;
};
let ctx = loop_ctx.clone();
let sink = loop_sink.clone();
runtime.spawn(async move {
let nowhere: SocketAddr =
"[::]:0".parse().expect("[::]:0 is a valid socket address");
Self::handle_message(&message, source, None, None, nowhere, ctx, sink)
.await;
});
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(std::time::Duration::from_millis(10));
}
Err(e) => {
console_error!(loop_ctx.status_tx, "NDP raw receive error: {}", e);
break;
}
}
}
});
match ready_rx.await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e),
Err(_) => {
return Err(anyhow!(
"NDP raw socket task on '{interface}' exited before signalling readiness"
))
}
}
let _sink = sink_rx.await.map_err(|_| {
anyhow!("NDP raw socket task on '{interface}' never produced a transmit handle")
})?;
app_state
.register_server_task(server_id, stop.park_task())
.await;
console_info!(status_tx, "NDP raw ICMPv6 active on {}", interface);
Ok("[::]:0".parse().expect("[::]:0 is a valid socket address"))
}
async fn handle_datagram(
datagram: &[u8],
peer: SocketAddr,
local_addr: SocketAddr,
ctx: Arc<NdpContext>,
sink: MessageSink,
) {
let (source, destination, message) = match codec::decode_addressed(datagram) {
Ok(parts) => parts,
Err(e) => {
debug!(
"NDP ignoring an undecodable datagram ({} octets) from {}: {}",
datagram.len(),
peer,
e
);
return;
}
};
if let Err(e) = codec::verify_checksum(message, source, destination) {
console_debug!(ctx.status_tx, "NDP dropping a message: {}", e);
return;
}
Self::handle_message(
message,
source,
Some(destination),
Some(peer),
local_addr,
ctx,
sink,
)
.await;
}
#[allow(clippy::too_many_arguments)]
async fn handle_message(
message: &[u8],
source: Ipv6Addr,
destination: Option<Ipv6Addr>,
peer: Option<SocketAddr>,
local_addr: SocketAddr,
ctx: Arc<NdpContext>,
sink: MessageSink,
) {
let decoded = match NdpMessage::decode(message) {
Ok(decoded) => decoded,
Err(e) => {
trace!(
"NDP ignoring a {}-octet ICMPv6 message from {}: {}",
message.len(),
source,
e
);
return;
}
};
console_debug!(
ctx.status_tx,
"NDP {} from {} ({} octets)",
decoded.type_name(),
source,
message.len()
);
console_trace!(
ctx.status_tx,
"NDP {} from {}: {} option(s)",
decoded.type_name(),
source,
decoded.options().len()
);
let connection_id = ConnectionId::new(ctx.app_state.get_next_unified_id().await);
{
use crate::state::server::{
ConnectionState as ServerConnectionState, ConnectionStatus, ProtocolConnectionInfo,
};
let now = crate::utils::clock::Instant::now();
let nowhere: SocketAddr = "[::]:0".parse().expect("[::]:0 is a valid socket address");
let conn = ServerConnectionState {
id: connection_id,
remote_addr: peer.unwrap_or(SocketAddr::V6(SocketAddrV6::new(source, 0, 0, 0))),
local_addr: if local_addr.port() == 0 {
nowhere
} else {
local_addr
},
bytes_sent: 0,
bytes_received: message.len() as u64,
packets_sent: 0,
packets_received: 1,
last_activity: now,
status: ConnectionStatus::Active,
status_changed_at: now,
protocol_info: ProtocolConnectionInfo::empty(),
};
ctx.app_state
.add_connection_to_server(ctx.server_id, conn)
.await;
let _ = ctx.status_tx.send("__UPDATE_UI__".to_string());
}
let mut data = decoded.to_event_data();
data.insert(
"source_address".into(),
serde_json::json!(source.to_string()),
);
if let Some(destination) = destination {
data.insert(
"destination_address".into(),
serde_json::json!(destination.to_string()),
);
}
data.insert(
"connection_id".into(),
serde_json::json!(connection_id.to_string()),
);
let event_type = match &decoded {
NdpMessage::RouterSolicitation { .. } => &*NDP_ROUTER_SOLICITATION_EVENT,
NdpMessage::RouterAdvertisement(_) => &*NDP_ROUTER_ADVERTISEMENT_RECEIVED_EVENT,
NdpMessage::NeighborSolicitation { .. } => &*NDP_NEIGHBOR_SOLICITATION_EVENT,
NdpMessage::NeighborAdvertisement { .. } => &*NDP_NEIGHBOR_ADVERTISEMENT_EVENT,
NdpMessage::Redirect { .. } => &*NDP_REDIRECT_RECEIVED_EVENT,
};
let event = Event::new(event_type, serde_json::Value::Object(data));
Self::dispatch_event(event, Some(connection_id), Some(source), ctx, sink).await;
}
async fn dispatch_event(
event: Event,
connection_id: Option<ConnectionId>,
peer_address: Option<Ipv6Addr>,
ctx: Arc<NdpContext>,
sink: MessageSink,
) {
let event_id = event.event_type.id.clone();
if !operator_wants_dynamic(&ctx.app_state, ctx.server_id, &event_id).await {
debug!(
"NDP {} decision=no_policy: no instruction and no handler, nothing sent and no \
LLM call",
event_id
);
let _ = ctx.status_tx.send(format!(
"NDP {} decision=no_policy: observing only (no instruction or handler)",
event_id
));
return;
}
match call_llm(
&ctx.llm_client,
&ctx.app_state,
ctx.server_id,
connection_id,
&event,
ctx.protocol.as_ref(),
)
.await
{
Ok(result) => {
for message in &result.messages {
info!("{}", message);
let _ = ctx.status_tx.send(format!("[INFO] {}", message));
}
let mut sent = 0usize;
let mut netget_failures = 0usize;
for protocol_result in &result.protocol_results {
let crate::llm::actions::protocol_trait::ActionResult::Custom { name, data } =
protocol_result
else {
continue;
};
if name != NDP_MESSAGE_RESULT {
continue;
}
let Some(action) = data.get("action") else {
warn!("NDP message result carried no action payload");
continue;
};
let request = match codec::SendRequest::from_action(action) {
Ok(request) => request.with_default_link_layer(ctx.link_layer_address),
Err(e) => {
netget_failures += 1;
console_error!(ctx.status_tx, "NDP cannot build message: {:#}", e);
continue;
}
};
let source = request.source.unwrap_or(ctx.link_local_address);
let destination = request
.destination
.unwrap_or_else(|| request.default_destination(peer_address));
let bytes = match request.message.encode(source, destination) {
Ok(bytes) => bytes,
Err(e) => {
netget_failures += 1;
console_error!(ctx.status_tx, "NDP cannot encode message: {:#}", e);
continue;
}
};
match sink.send(source, destination, &bytes).await {
Ok(n) => {
sent += 1;
if let Some(connection_id) = connection_id {
ctx.app_state
.update_connection_stats(
ctx.server_id,
connection_id,
None,
Some(n as u64),
None,
Some(1),
)
.await;
}
console_debug!(
ctx.status_tx,
"NDP sent {} to {} ({} octets on the wire)",
request.message.type_name(),
destination,
n
);
}
Err(e) => {
netget_failures += 1;
console_error!(ctx.status_tx, "NDP transmit failed: {:#}", e);
}
}
}
if sent == 0 {
let rejected = result.raw_actions.iter().any(|a| {
a.get("type").and_then(|t| t.as_str()) == Some(NO_RESPONSE_ACTION)
});
let decision = if rejected {
"model_reject"
} else if netget_failures > 0 {
"fail_closed_send_error"
} else if !result.failures.is_empty() {
"fail_closed_action_error"
} else {
"model_silent"
};
info!(
"NDP {} decision={}: nothing sent (NDP has no error message; silence is \
the protocol-correct answer)",
event_id, decision
);
let _ = ctx.status_tx.send(format!(
"NDP {} decision={}: nothing sent",
event_id, decision
));
}
}
Err(e) => {
let category = crate::utils::WireFailure::classify(&e);
let decision = if category.is_overloaded() {
"fail_closed_overloaded"
} else {
"fail_closed_llm_error"
};
error!(
"NDP {} decision={}: nothing sent, NDP has no failure message ({}): {}",
event_id,
decision,
category.text(),
e
);
let _ = ctx.status_tx.send(format!(
"✗ NDP {} decision={}: nothing sent ({}): {}",
event_id,
decision,
category.text(),
e
));
}
}
}
}
fn interface_details(interface: &str) -> Option<(u32, Option<[u8; 6]>)> {
pnet::datalink::interfaces()
.into_iter()
.find(|i| i.name == interface)
.map(|i| {
(
i.index,
i.mac.map(|mac| [mac.0, mac.1, mac.2, mac.3, mac.4, mac.5]),
)
})
}
async fn operator_wants_dynamic(
state: &AppState,
server_id: crate::state::ServerId,
event_id: &str,
) -> bool {
state
.with_server_mut(server_id, |server| {
let has_instruction = !server.instruction.trim().is_empty();
let has_handler = server
.event_handler_config
.as_ref()
.map(|c| c.find_handler(event_id).is_some())
.unwrap_or(false);
has_instruction || has_handler
})
.await
.unwrap_or(false)
}