pub mod actions;
pub mod codec;
use anyhow::{anyhow, bail, Context, Result};
use std::net::SocketAddr;
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::{
LldpProtocol, LLDP_ADVERTISEMENT_RESULT, LLDP_ADVERTISE_DUE_EVENT,
LLDP_NEIGHBOR_ADVERTISEMENT_EVENT, NO_ADVERTISEMENT_ACTION,
};
const LLDP_BPF_FILTER: &str = "ether proto 0x88cc";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum TransportKind {
Raw,
Udp,
}
#[derive(Debug, Clone)]
struct LldpConfig {
transport: TransportKind,
udp_peer: Option<SocketAddr>,
advertise_interval_secs: u64,
source_mac: Option<[u8; 6]>,
}
impl LldpConfig {
fn from_params(params: Option<&crate::protocol::StartupParams>) -> Result<Self> {
let Some(params) = params else {
return Ok(Self {
transport: TransportKind::Raw,
udp_peer: None,
advertise_interval_secs: 0,
source_mac: 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\" (real Ethernet via libpcap) or \"udp\" (testing \
transport carrying Ethernet frames 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 advertise_interval_secs = match params.get_optional_i64("advertise_interval_secs")? {
None => 0,
Some(v) if (0..=86_400).contains(&v) => v as u64,
Some(v) => bail!(
"advertise_interval_secs must be between 0 (timer disabled) and 86400, got {v}"
),
};
let source_mac = match params.get_optional_string("source_mac")? {
None => None,
Some(text) => Some(codec::parse_mac(&text)?),
};
Ok(Self {
transport,
udp_peer,
advertise_interval_secs,
source_mac,
})
}
}
#[derive(Clone)]
enum FrameSink {
Pcap(std::sync::mpsc::Sender<Vec<u8>>),
Udp {
socket: Arc<UdpSocket>,
configured_peer: Option<SocketAddr>,
last_peer: Arc<Mutex<Option<SocketAddr>>>,
},
}
impl FrameSink {
async fn send(&self, frame: Vec<u8>) -> Result<usize> {
match self {
FrameSink::Pcap(tx) => {
let len = frame.len();
tx.send(frame)
.map_err(|_| anyhow!("pcap injection thread has gone away"))?;
Ok(len)
}
FrameSink::Udp {
socket,
configured_peer,
last_peer,
} => {
let peer = match configured_peer {
Some(p) => Some(*p),
None => *last_peer.lock().expect("lldp 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"
)
})?;
Ok(socket.send_to(&frame, peer).await?)
}
}
}
}
struct LldpContext {
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
protocol: Arc<LldpProtocol>,
server_id: crate::state::ServerId,
source_mac: [u8; 6],
interface: Option<String>,
}
pub struct LldpServer;
impl LldpServer {
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 = LldpConfig::from_params(startup_params.as_ref())?;
match config.transport {
TransportKind::Raw => {
let interface = interface.ok_or_else(|| {
anyhow!(
"LLDP transport \"raw\" needs an interface to capture on (pass \
`interface`), or use transport \"udp\" for an unprivileged test run"
)
})?;
let source_mac = config
.source_mac
.or_else(|| interface_mac(&interface))
.unwrap_or_else(|| {
codec::parse_mac(actions::DEFAULT_SOURCE_MAC)
.expect("DEFAULT_SOURCE_MAC is a literal MAC address")
});
Self::spawn_raw(
interface, config, source_mac, llm_client, app_state, status_tx, server_id,
)
.await
}
TransportKind::Udp => {
let source_mac = config.source_mac.unwrap_or_else(|| {
codec::parse_mac(actions::DEFAULT_SOURCE_MAC)
.expect("DEFAULT_SOURCE_MAC is a literal MAC address")
});
Self::spawn_udp(
listen_addr,
interface,
config,
source_mac,
llm_client,
app_state,
status_tx,
server_id,
)
.await
}
}
}
#[allow(clippy::too_many_arguments)]
async fn spawn_udp(
listen_addr: SocketAddr,
interface: Option<String>,
config: LldpConfig,
source_mac: [u8; 6],
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 LLDP UDP test transport to {listen_addr}")
})?);
let local_addr = socket.local_addr()?;
console_info!(
status_tx,
"LLDP listening on {} (transport=udp — a TEST transport carrying Ethernet frames in \
datagrams; no real LLDP neighbour speaks it)",
local_addr
);
let ctx = Arc::new(LldpContext {
llm_client,
app_state: app_state.clone(),
status_tx: status_tx.clone(),
protocol: Arc::new(LldpProtocol::new()),
server_id,
source_mac,
interface,
});
let sink = FrameSink::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 FrameSink::Udp { last_peer, .. } = &receive_sink {
*last_peer.lock().expect("lldp peer mutex poisoned") = Some(peer);
}
let frame = 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_frame(&frame, Some(peer), local_addr, ctx, sink).await;
})
.await;
}
Err(e) => {
console_error!(receive_ctx.status_tx, "LLDP UDP receive error: {}", e);
break;
}
}
}
});
app_state
.register_server_task(server_id, receive_handle)
.await;
Self::spawn_advertise_timer(&config, ctx, sink, app_state, server_id).await;
Ok(local_addr)
}
#[allow(clippy::too_many_arguments)]
async fn spawn_raw(
interface: String,
config: LldpConfig,
source_mac: [u8; 6],
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
) -> Result<SocketAddr> {
use pcap::{Capture, Device};
console_info!(
status_tx,
"LLDP starting capture on interface {} (source MAC {})",
interface,
codec::format_mac(&source_mac)
);
let ctx = Arc::new(LldpContext {
llm_client,
app_state: app_state.clone(),
status_tx: status_tx.clone(),
protocol: Arc::new(LldpProtocol::new()),
server_id,
source_mac,
interface: Some(interface.clone()),
});
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<Result<()>>();
let stop = crate::utils::StopSignal::new();
let stop_in_loop = stop.clone();
let (packet_tx, packet_rx) = std::sync::mpsc::channel::<Vec<u8>>();
let sink = FrameSink::Pcap(packet_tx.clone());
let loop_interface = interface.clone();
let loop_ctx = ctx.clone();
let loop_sink = sink.clone();
tokio::task::spawn_blocking(move || {
let open = || -> Result<(Capture<pcap::Active>, Capture<pcap::Active>)> {
let device = Device::list()
.context("failed to list network devices")?
.into_iter()
.find(|d| d.name == loop_interface)
.ok_or_else(|| anyhow!("no such capture device '{}'", loop_interface))?;
let mut cap_rx = Capture::from_device(device.clone())
.map(|c| c.promisc(true).snaplen(65535).timeout(1000))
.and_then(|c| c.open())
.with_context(|| {
format!(
"failed to open pcap capture on '{}' (needs root, or read access to \
/dev/bpf* on macOS/BSD, or CAP_NET_RAW on Linux)",
loop_interface
)
})?;
cap_rx.filter(LLDP_BPF_FILTER, true).with_context(|| {
format!(
"failed to apply the '{LLDP_BPF_FILTER}' BPF filter on '{}'. LLDP is an \
Ethernet-only protocol, so this fails on any interface with no Ethernet \
link layer — loopback (lo/lo0), tunnels and raw-IP devices carry no LLDP \
and libpcap rejects the filter outright. Point this server at a real \
Ethernet interface, or use transport \"udp\" for an unprivileged test.",
loop_interface
)
})?;
let cap_tx = Capture::from_device(device)
.map(|c| c.promisc(true).snaplen(65535).timeout(1000))
.and_then(|c| c.open())
.with_context(|| {
format!(
"failed to open pcap injection handle on '{}'",
loop_interface
)
})?;
Ok((cap_rx, cap_tx))
};
let (mut cap_rx, mut cap_tx) = match open() {
Ok(handles) => {
let _ = ready_tx.send(Ok(()));
handles
}
Err(e) => {
console_error!(loop_ctx.status_tx, "LLDP capture startup failed: {:#}", e);
let _ = ready_tx.send(Err(e));
return;
}
};
std::thread::spawn(move || {
while let Ok(frame) = packet_rx.recv() {
if let Err(e) = cap_tx.sendpacket(frame) {
error!("LLDP failed to inject frame: {}", e);
}
}
});
let runtime = tokio::runtime::Handle::current();
loop {
if stop_in_loop.is_stopped() {
console_info!(
loop_ctx.status_tx,
"LLDP capture on {} stopping",
loop_interface
);
break;
}
match cap_rx.next_packet() {
Ok(packet) => {
let frame = packet.data.to_vec();
let ctx = loop_ctx.clone();
let sink = loop_sink.clone();
runtime.spawn(async move {
let nowhere: SocketAddr = "0.0.0.0:0"
.parse()
.expect("0.0.0.0:0 is a valid socket address");
Self::handle_frame(&frame, None, nowhere, ctx, sink).await;
});
}
Err(pcap::Error::TimeoutExpired) => continue,
Err(e) => {
console_error!(loop_ctx.status_tx, "LLDP capture error: {}", e);
break;
}
}
}
drop(packet_tx);
drop(cap_rx);
});
match ready_rx.await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e),
Err(_) => {
return Err(anyhow!(
"LLDP capture task on '{}' exited before signalling readiness",
interface
))
}
}
app_state
.register_server_task(server_id, stop.park_task())
.await;
console_info!(status_tx, "LLDP capture active on {}", interface);
Self::spawn_advertise_timer(&config, ctx, sink, app_state, server_id).await;
Ok("0.0.0.0:0"
.parse()
.expect("0.0.0.0:0 is a valid socket address"))
}
async fn spawn_advertise_timer(
config: &LldpConfig,
ctx: Arc<LldpContext>,
sink: FrameSink,
app_state: Arc<AppState>,
server_id: crate::state::ServerId,
) {
if config.advertise_interval_secs == 0 {
debug!("LLDP advertise timer disabled (advertise_interval_secs = 0)");
return;
}
let interval_secs = config.advertise_interval_secs;
let handle = tokio::spawn(async move {
let mut sequence: u64 = 0;
let mut ticker = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
ticker.tick().await;
loop {
ticker.tick().await;
sequence += 1;
let n = sequence;
let event = Event::new(
&LLDP_ADVERTISE_DUE_EVENT,
serde_json::json!({
"interval_secs": interval_secs,
"sequence": n,
"interface": ctx.interface,
}),
);
Self::dispatch_event(event, None, ctx.clone(), sink.clone()).await;
}
});
app_state.register_server_task(server_id, handle).await;
}
async fn handle_frame(
frame: &[u8],
peer: Option<SocketAddr>,
local_addr: SocketAddr,
ctx: Arc<LldpContext>,
sink: FrameSink,
) {
let decoded = match codec::decode_frame(frame) {
Ok(decoded) => decoded,
Err(e) => {
debug!(
"LLDP ignoring an undecodable frame ({} bytes): {}",
frame.len(),
e
);
return;
}
};
if decoded.source_mac == ctx.source_mac {
trace!(
"LLDP ignoring our own frame from {}",
codec::format_mac(&decoded.source_mac)
);
return;
}
let source_mac = codec::format_mac(&decoded.source_mac);
let destination_mac = codec::format_mac(&decoded.destination_mac);
console_debug!(
ctx.status_tx,
"LLDP advertisement from {} ({} bytes)",
source_mac,
frame.len()
);
console_trace!(
ctx.status_tx,
"LLDP neighbour {}: chassis={} port={} ttl={}",
source_mac,
decoded.lldpdu.chassis_id,
decoded.lldpdu.port_id,
decoded.lldpdu.ttl
);
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.0.0.0:0"
.parse()
.expect("0.0.0.0:0 is a valid socket address");
let conn = ServerConnectionState {
id: connection_id,
remote_addr: peer.unwrap_or(nowhere),
local_addr,
bytes_sent: 0,
bytes_received: frame.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.lldpdu.to_event_data();
data.insert("source_mac".into(), serde_json::json!(source_mac));
data.insert("destination_mac".into(), serde_json::json!(destination_mac));
data.insert(
"connection_id".into(),
serde_json::json!(connection_id.to_string()),
);
let event = Event::new(
&LLDP_NEIGHBOR_ADVERTISEMENT_EVENT,
serde_json::Value::Object(data),
);
Self::dispatch_event(event, Some(connection_id), ctx, sink).await;
}
async fn dispatch_event(
event: Event,
connection_id: Option<ConnectionId>,
ctx: Arc<LldpContext>,
sink: FrameSink,
) {
let event_id = event.event_type.id.clone();
if !operator_wants_dynamic(&ctx.app_state, ctx.server_id, &event_id).await {
debug!(
"LLDP {} decision=no_policy: no instruction and no handler, nothing advertised \
and no LLM call",
event_id
);
let _ = ctx.status_tx.send(format!(
"LLDP {} decision=no_policy: listening 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;
for protocol_result in &result.protocol_results {
let crate::llm::actions::protocol_trait::ActionResult::Custom { name, data } =
protocol_result
else {
continue;
};
if name != LLDP_ADVERTISEMENT_RESULT {
continue;
}
let Some(action) = data.get("action") else {
warn!("LLDP advertisement result carried no action payload");
continue;
};
let request = match codec::AdvertisementRequest::from_action(action) {
Ok(request) => request,
Err(e) => {
console_error!(
ctx.status_tx,
"LLDP cannot build advertisement: {:#}",
e
);
continue;
}
};
let frame = match request.to_frame(ctx.source_mac) {
Ok(frame) => frame,
Err(e) => {
console_error!(
ctx.status_tx,
"LLDP cannot encode advertisement: {:#}",
e
);
continue;
}
};
match sink.send(frame).await {
Ok(n) => {
sent += 1;
console_debug!(
ctx.status_tx,
"LLDP advertised {} ({} on port {}, {} bytes)",
request
.lldpdu
.system_name
.clone()
.unwrap_or_else(|| "unnamed".to_string()),
request.lldpdu.chassis_id,
request.lldpdu.port_id,
n
);
}
Err(e) => {
console_error!(ctx.status_tx, "LLDP 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_ADVERTISEMENT_ACTION)
});
let decision = if rejected {
"model_reject"
} else {
"model_silent"
};
info!(
"LLDP {} decision={}: nothing advertised (LLDP has no error frame; \
silence is the protocol-correct answer)",
event_id, decision
);
let _ = ctx.status_tx.send(format!(
"LLDP {} decision={}: nothing advertised",
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!(
"LLDP {} decision={}: nothing advertised, LLDP has no failure frame ({}): {}",
event_id,
decision,
category.text(),
e
);
let _ = ctx.status_tx.send(format!(
"✗ LLDP {} decision={}: nothing advertised ({}): {}",
event_id,
decision,
category.text(),
e
));
}
}
}
}
fn interface_mac(interface: &str) -> Option<[u8; 6]> {
pnet::datalink::interfaces()
.into_iter()
.find(|i| i.name == interface)
.and_then(|i| 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)
}