pub mod actions;
pub mod codec;
use anyhow::{anyhow, Context, Result};
use std::net::SocketAddr;
use std::sync::Arc;
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::logging::emit::Log;
use crate::protocol::Event;
use crate::server::connection::ConnectionId;
use crate::state::app_state::AppState;
use actions::{CdpProtocol, CdpTransport, CDP_ACTION_RESULT, CDP_NEIGHBOR_ADVERTISEMENT_EVENT};
use codec::{CdpAdvertisement, FrameHeader};
const DEFAULT_UDP_SOURCE_MAC: [u8; 6] = [0x02, 0x00, 0x0c, 0xcc, 0xcc, 0x01];
const CDP_BPF_FILTER: &str = "ether dst 01:00:0c:cc:cc:cc and ether[14:2] = 0xaaaa \
and ether[16:1] = 0x03 and ether[17:2] = 0x0000 \
and ether[19:1] = 0x0c and ether[20:2] = 0x2000";
#[derive(Clone)]
enum CdpSink {
#[cfg(all(unix, feature = "libpcap"))]
Raw(Arc<std::sync::Mutex<pcap::Capture<pcap::Active>>>),
Udp(Arc<UdpSocket>, SocketAddr),
}
impl CdpSink {
async fn send(&self, frame: &[u8]) -> Result<()> {
match self {
#[cfg(all(unix, feature = "libpcap"))]
CdpSink::Raw(cap) => {
let mut guard = cap.lock().unwrap_or_else(|e| e.into_inner());
guard
.sendpacket(frame)
.context("pcap injection of a CDP frame failed")
}
CdpSink::Udp(socket, peer) => {
socket
.send_to(frame, peer)
.await
.with_context(|| format!("sending a CDP frame to {} failed", peer))?;
Ok(())
}
}
}
fn describe(&self) -> String {
match self {
#[cfg(all(unix, feature = "libpcap"))]
CdpSink::Raw(_) => "pcap".to_string(),
CdpSink::Udp(_, peer) => format!("udp:{}", peer),
}
}
}
struct CdpContext {
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
protocol: Arc<CdpProtocol>,
server_id: crate::state::ServerId,
source_mac: [u8; 6],
}
pub struct CdpServer;
impl CdpServer {
pub async fn spawn_with_llm_actions(ctx: crate::protocol::SpawnContext) -> Result<SocketAddr> {
let crate::protocol::SpawnContext {
interface,
host,
port,
mac_address,
llm_client,
state: app_state,
status_tx,
server_id,
startup_params,
..
} = ctx;
let mut transport = CdpTransport::Raw;
let mut configured_mac: Option<[u8; 6]> = None;
if let Some(ref params) = startup_params {
if let Some(value) = params.get_optional_string("transport")? {
transport = CdpTransport::parse(&value)?;
}
if let Some(value) = params.get_optional_string("source_mac")? {
configured_mac = Some(codec::parse_mac(&value)?);
}
}
if configured_mac.is_none() {
if let Some(ref value) = mac_address {
configured_mac = Some(codec::parse_mac(value)?);
}
}
let protocol = Arc::new(CdpProtocol::new());
match transport {
CdpTransport::Udp => {
Self::spawn_udp(
host.unwrap_or_else(|| "127.0.0.1".to_string()),
port.unwrap_or(0),
configured_mac.unwrap_or(DEFAULT_UDP_SOURCE_MAC),
llm_client,
app_state,
status_tx,
server_id,
protocol,
)
.await
}
CdpTransport::Raw => {
let interface = interface.context(
"CDP requires a network interface: pass `interface` (e.g. \"en0\"), or use \
the `transport: \"udp\"` startup parameter for the unprivileged test \
transport",
)?;
Self::spawn_raw(
interface,
configured_mac,
llm_client,
app_state,
status_tx,
server_id,
protocol,
)
.await
}
}
}
#[allow(clippy::too_many_arguments)]
async fn spawn_udp(
host: String,
port: u16,
source_mac: [u8; 6],
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
protocol: Arc<CdpProtocol>,
) -> Result<SocketAddr> {
let bind_addr = format!("{}:{}", host, port);
let socket = UdpSocket::bind(&bind_addr)
.await
.with_context(|| format!("CDP (udp transport) could not bind {}", bind_addr))?;
let local_addr = socket
.local_addr()
.context("CDP (udp transport) bound socket has no local address")?;
let socket = Arc::new(socket);
Log::new(Some(&status_tx)).info(format!(
"CDP listening on {} (udp test transport, source MAC {})",
local_addr,
codec::mac_to_string(&source_mac)
));
let cdp_ctx = Arc::new(CdpContext {
llm_client,
app_state: app_state.clone(),
status_tx: status_tx.clone(),
protocol,
server_id,
source_mac,
});
let recv_socket = socket.clone();
let loop_state = app_state.clone();
let handle = tokio::spawn(async move {
let mut buffer = vec![0u8; 65535];
loop {
match recv_socket.recv_from(&mut buffer).await {
Ok((n, peer)) => {
let frame = buffer[..n].to_vec();
let sink = CdpSink::Udp(recv_socket.clone(), peer);
let cdp_ctx = cdp_ctx.clone();
let task_owner = loop_state.clone();
task_owner
.spawn_server_task(server_id, async move {
if let Err(e) = Self::handle_frame(&frame, sink, cdp_ctx).await {
debug!("CDP frame from {} ignored: {:#}", peer, e);
}
})
.await;
}
Err(e) => {
error!("CDP (udp transport) receive error: {}", e);
break;
}
}
}
warn!("CDP (udp transport) receive loop terminated");
});
app_state.register_server_task(server_id, handle).await;
Ok(local_addr)
}
#[allow(clippy::too_many_arguments)]
async fn spawn_raw(
interface: String,
configured_mac: Option<[u8; 6]>,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
protocol: Arc<CdpProtocol>,
) -> Result<SocketAddr> {
Log::new(Some(&status_tx)).info(format!("CDP capture starting on interface {}", interface));
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<Result<()>>();
let stop = crate::utils::StopSignal::new();
let stop_in_loop = stop.clone();
let status_tx_ready = status_tx.clone();
let app_state_reg = app_state.clone();
let interface_for_thread = interface.clone();
tokio::task::spawn_blocking(move || {
let open = || -> Result<(pcap::Capture<pcap::Active>, [u8; 6])> {
let device = pcap::Device::list()
.context("failed to list capture devices")?
.into_iter()
.find(|d| d.name == interface_for_thread)
.ok_or_else(|| anyhow!("no such capture device '{}'", interface_for_thread))?;
let local_mac = match configured_mac {
Some(mac) => mac,
None => Self::interface_mac(&interface_for_thread).unwrap_or_else(|| {
warn!(
"CDP could not determine the MAC of '{}'; using {}",
interface_for_thread,
codec::mac_to_string(&DEFAULT_UDP_SOURCE_MAC)
);
DEFAULT_UDP_SOURCE_MAC
}),
};
let mut cap = pcap::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 capture on '{}' (needs root, or read access to \
/dev/bpf* on macOS/BSD, or CAP_NET_RAW on Linux)",
interface_for_thread
)
})?;
cap.filter(CDP_BPF_FILTER, true).with_context(|| {
format!(
"failed to apply the CDP BPF filter on '{}'. CDP rides 802.3 with an \
LLC/SNAP header, so this filter cannot compile on an interface with no \
Ethernet link layer — loopback (lo/lo0), tunnels and raw-IP devices \
carry no CDP. Point this server at a real Ethernet or Wi-Fi interface.",
interface_for_thread
)
})?;
Ok((cap, local_mac))
};
let (cap, local_mac) = match open() {
Ok(v) => {
let _ = ready_tx.send(Ok(()));
v
}
Err(e) => {
Log::new(Some(&status_tx))
.error(format!("CDP capture startup failed: {:#}", e));
let _ = ready_tx.send(Err(e));
return;
}
};
info!(
"CDP capture live on {} as {}",
interface_for_thread,
codec::mac_to_string(&local_mac)
);
let cdp_ctx = Arc::new(CdpContext {
llm_client,
app_state,
status_tx: status_tx.clone(),
protocol,
server_id,
source_mac: local_mac,
});
let runtime = tokio::runtime::Handle::current();
let cap = Arc::new(std::sync::Mutex::new(cap));
loop {
if stop_in_loop.is_stopped() {
Log::new(Some(&status_tx))
.info(format!("CDP capture on {} stopping", interface_for_thread));
break;
}
let mut guard = cap.lock().unwrap_or_else(|e| e.into_inner());
match guard.next_packet() {
Ok(packet) => {
let frame = packet.data.to_vec();
drop(guard);
let sink = CdpSink::Raw(cap.clone());
let cdp_ctx = cdp_ctx.clone();
runtime.spawn(async move {
if let Err(e) = Self::handle_frame(&frame, sink, cdp_ctx).await {
debug!("CDP frame ignored: {:#}", e);
}
});
}
Err(pcap::Error::TimeoutExpired) => {
drop(guard);
continue;
}
Err(e) => {
drop(guard);
Log::new(Some(&status_tx)).error(format!("CDP capture error: {}", e));
break;
}
}
}
drop(cap);
});
match ready_rx.await {
Ok(Ok(())) => {}
Ok(Err(e)) => return Err(e),
Err(_) => {
return Err(anyhow!(
"CDP capture task on '{}' exited before signalling readiness",
interface
))
}
}
app_state_reg
.register_server_task(server_id, stop.park_task())
.await;
Log::new(Some(&status_tx_ready)).info(format!("CDP capture active on {}", interface));
Ok("0.0.0.0:0".parse().expect("literal address parses"))
}
fn interface_mac(name: &str) -> Option<[u8; 6]> {
pnet::datalink::interfaces()
.into_iter()
.find(|i| i.name == name)
.and_then(|i| i.mac)
.map(|m| [m.0, m.1, m.2, m.3, m.4, m.5])
}
async fn handle_frame(frame: &[u8], sink: CdpSink, ctx: Arc<CdpContext>) -> Result<()> {
let (header, payload) = codec::decode_frame(frame)?;
let decoded = codec::decode_payload(payload)?;
if header.source_mac == ctx.source_mac {
trace!("CDP ignoring a frame from our own source MAC");
return Ok(());
}
let connection_id = ConnectionId::new(ctx.app_state.get_next_unified_id().await);
Self::register_connection(&ctx, connection_id, &header, frame.len()).await;
let ad = &decoded.advertisement;
Log::new(Some(&ctx.status_tx)).info(format!(
"CDP advertisement from {} ({}) via {}",
ad.device_id.as_deref().unwrap_or("<no device id>"),
ad.platform.as_deref().unwrap_or("unknown platform"),
codec::mac_to_string(&header.source_mac),
));
if !decoded.checksum_valid() {
Log::new(Some(&ctx.status_tx)).warn(format!(
"CDP advertisement from {} carries checksum 0x{:04x} but computes 0x{:04x}",
codec::mac_to_string(&header.source_mac),
decoded.declared_checksum,
decoded.computed_checksum
));
}
let event = Event::new(
&CDP_NEIGHBOR_ADVERTISEMENT_EVENT,
decoded.to_event_data(&header, &connection_id.to_string()),
);
if !operator_wants_dynamic(&ctx.app_state, ctx.server_id, &event.event_type.id).await {
debug!(
"CDP {} from {} decision=passive_no_policy (no instruction and no handler; no \
advertisement and no LLM call)",
event.event_type.id,
codec::mac_to_string(&header.source_mac)
);
Log::new(Some(&ctx.status_tx)).debug(
"CDP advertisement observed passively: no operator policy configured, no reply"
.to_string(),
);
return Ok(());
}
Self::dispatch(event, connection_id, sink, header, ctx).await
}
async fn register_connection(
ctx: &CdpContext,
connection_id: ConnectionId,
header: &FrameHeader,
bytes: usize,
) {
use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
let now = crate::utils::clock::Instant::now();
let conn = ConnectionState {
id: connection_id,
remote_addr: "0.0.0.0:0".parse().expect("literal address parses"),
local_addr: "0.0.0.0:0".parse().expect("literal address parses"),
bytes_sent: 0,
bytes_received: bytes as u64,
packets_sent: 0,
packets_received: 1,
last_activity: now,
status: ConnectionStatus::Active,
status_changed_at: now,
protocol_info: ProtocolConnectionInfo::new(serde_json::json!({
"source_mac": codec::mac_to_string(&header.source_mac),
"destination_mac": codec::mac_to_string(&header.destination_mac),
})),
};
ctx.app_state
.add_connection_to_server(ctx.server_id, conn)
.await;
let _ = ctx.status_tx.send("__UPDATE_UI__".to_string());
}
async fn dispatch(
event: Event,
connection_id: ConnectionId,
sink: CdpSink,
header: FrameHeader,
ctx: Arc<CdpContext>,
) -> Result<()> {
let event_id = event.event_type.id.clone();
let peer = codec::mac_to_string(&header.source_mac);
let result = call_llm(
&ctx.llm_client,
&ctx.app_state,
ctx.server_id,
Some(connection_id),
&event,
ctx.protocol.as_ref(),
)
.await;
let execution = match result {
Ok(execution) => execution,
Err(e) => {
let category = crate::utils::WireFailure::classify(&e);
let decision = if category.is_overloaded() {
"fail_closed_llm_error_overloaded"
} else {
"fail_closed_llm_error_unavailable"
};
error!(
"CDP {} from {} decision={} ({}) — no advertisement emitted; CDP has no \
error frame and a fabricated one would poison the neighbour's table. \
LLM error: {}",
event_id,
peer,
decision,
category.text(),
e
);
Log::new(Some(&ctx.status_tx)).error(format!(
"✗ CDP LLM error (decision={}, no advertisement emitted): {}",
decision, e
));
return Ok(());
}
};
for message in &execution.messages {
Log::new(Some(&ctx.status_tx)).info(message.to_string());
}
let mut frames_sent = 0usize;
let mut explicit_refusal: Option<String> = None;
for protocol_result in &execution.protocol_results {
let crate::llm::actions::protocol_trait::ActionResult::Custom { name, data } =
protocol_result
else {
continue;
};
if name != CDP_ACTION_RESULT {
continue;
}
match data.get("type").and_then(|v| v.as_str()) {
Some("no_advertisement") => {
explicit_refusal = Some(
data.get("reason")
.and_then(|v| v.as_str())
.unwrap_or("no reason given")
.to_string(),
);
}
Some("send_cdp_advertisement") => {
let advertisement = match CdpAdvertisement::from_action(data) {
Ok(ad) => ad,
Err(e) => {
error!(
"CDP {} from {} decision=model_invalid_action — no \
advertisement emitted: {:#}",
event_id, peer, e
);
Log::new(Some(&ctx.status_tx))
.error(format!("✗ CDP advertisement rejected: {:#}", e));
continue;
}
};
let payload = match codec::encode_payload(&advertisement) {
Ok(p) => p,
Err(e) => {
error!("CDP payload encode failed: {:#}", e);
continue;
}
};
let frame = match codec::encode_frame(ctx.source_mac, &payload) {
Ok(f) => f,
Err(e) => {
error!("CDP frame encode failed: {:#}", e);
continue;
}
};
match sink.send(&frame).await {
Ok(()) => {
frames_sent += 1;
ctx.app_state
.update_connection_stats(
ctx.server_id,
connection_id,
None,
Some(frame.len() as u64),
None,
Some(1),
)
.await;
Log::new(Some(&ctx.status_tx)).info(format!(
"→ CDP advertisement sent via {} ({} bytes): {}",
sink.describe(),
frame.len(),
advertisement.summary()
));
trace!("CDP sent (hex): {}", hex::encode(&frame));
}
Err(e) => {
Log::new(Some(&ctx.status_tx))
.error(format!("CDP send failed: {:#}", e));
}
}
}
other => {
warn!("CDP ignoring unexpected action type {:?}", other);
}
}
}
if frames_sent == 0 {
let decision = match &explicit_refusal {
Some(_) => "model_reject",
None => "model_silent",
};
info!(
"CDP {} from {} decision={} — no advertisement emitted{}",
event_id,
peer,
decision,
explicit_refusal
.as_ref()
.map(|r| format!(" (reason: {})", r))
.unwrap_or_default()
);
Log::new(Some(&ctx.status_tx)).info(format!(
"CDP answered with no advertisement: decision={}",
decision
));
}
Ok(())
}
}
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)
}