pub mod actions;
pub mod codec;
use anyhow::{anyhow, Context, Result};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::os::unix::io::AsRawFd;
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::protocol::Event;
use crate::server::connection::ConnectionId;
use crate::state::app_state::AppState;
use crate::state::ServerId;
use crate::{console_debug, console_error, console_info, console_trace};
use actions::{
VrrpGroupConfig, VrrpProtocol, VrrpTransport, VRRP_ADVERTISEMENT_RECEIVED_EVENT,
VRRP_MASTER_RESIGNED_EVENT,
};
use codec::{Advertisement, PseudoHeader, Variant, VRRP_MULTICAST_IPV4};
const IPV4_HEADER_MIN_LEN: usize = 20;
#[derive(Clone)]
enum Responder {
Udp {
socket: Arc<UdpSocket>,
peer: SocketAddr,
local: Ipv4Addr,
},
Raw {
fd: i32,
source: Ipv4Addr,
},
}
impl Responder {
fn source_address(&self) -> Ipv4Addr {
match self {
Responder::Udp { local, .. } => *local,
Responder::Raw { source, .. } => *source,
}
}
async fn send(&self, packet: &[u8], destination: Ipv4Addr) -> Result<()> {
match self {
Responder::Udp { socket, peer, .. } => {
socket
.send_to(packet, peer)
.await
.with_context(|| format!("failed to send VRRP packet to {peer}"))?;
Ok(())
}
Responder::Raw { fd, .. } => send_raw(*fd, destination, packet),
}
}
fn describe(&self, destination: Ipv4Addr) -> String {
match self {
Responder::Udp { peer, .. } => {
format!("{peer} (udp transport, addressed to {destination})")
}
Responder::Raw { .. } => destination.to_string(),
}
}
}
pub struct VrrpServer;
impl VrrpServer {
pub async fn spawn_with_llm_actions(
listen_addr: SocketAddr,
config: VrrpGroupConfig,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: ServerId,
) -> Result<SocketAddr> {
match config.transport {
VrrpTransport::Udp => {
Self::spawn_udp(
listen_addr,
config,
llm_client,
app_state,
status_tx,
server_id,
)
.await
}
VrrpTransport::Raw => {
Self::spawn_raw(
listen_addr,
config,
llm_client,
app_state,
status_tx,
server_id,
)
.await
}
}
}
async fn spawn_udp(
listen_addr: SocketAddr,
config: VrrpGroupConfig,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: ServerId,
) -> Result<SocketAddr> {
let socket = Arc::new(
UdpSocket::bind(listen_addr)
.await
.with_context(|| format!("failed to bind VRRP UDP transport to {listen_addr}"))?,
);
let local_addr = socket.local_addr()?;
let local_v4 = match local_addr.ip() {
IpAddr::V4(ip) => ip,
IpAddr::V6(_) => {
return Err(anyhow!(
"VRRP carries IPv4 virtual addresses only; the UDP transport must bind an \
IPv4 address, got {local_addr}"
))
}
};
console_info!(
status_tx,
"VRRP server listening on {} (UDP transport: one complete {} message per datagram, \
vrid {}, priority {})",
local_addr,
config.variant.as_str(),
config.vrid,
config.priority
);
let protocol = Arc::new(VrrpProtocol::new());
let config = Arc::new(config);
let task_registrar = app_state.clone();
let recv_socket = socket.clone();
let accept_handle = tokio::spawn(async move {
let mut buffer = vec![0u8; 2048];
loop {
match recv_socket.recv_from(&mut buffer).await {
Ok((n, peer)) => {
let packet = buffer[..n].to_vec();
console_debug!(status_tx, "VRRP received {} byte message from {}", n, peer);
console_trace!(status_tx, "VRRP message (hex): {}", hex::encode(&packet));
let peer_v4 = match peer.ip() {
IpAddr::V4(ip) => ip,
IpAddr::V6(_) => {
console_debug!(
status_tx,
"VRRP ignoring datagram from IPv6 peer {}: no IPv6 transport",
peer
);
continue;
}
};
let responder = Responder::Udp {
socket: socket.clone(),
peer,
local: local_v4,
};
let inbound_pseudo = PseudoHeader::new(peer_v4, VRRP_MULTICAST_IPV4);
Self::spawn_handler(
packet,
peer_v4,
inbound_pseudo,
n,
responder,
&llm_client,
&app_state,
&status_tx,
&protocol,
&config,
server_id,
local_addr,
peer,
)
.await;
}
Err(e) => {
console_error!(status_tx, "VRRP UDP receive error: {}", e);
break;
}
}
}
warn!("VRRP UDP receive loop terminated");
});
task_registrar
.register_server_task(server_id, accept_handle)
.await;
Ok(local_addr)
}
async fn spawn_raw(
listen_addr: SocketAddr,
config: VrrpGroupConfig,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: ServerId,
) -> Result<SocketAddr> {
let interface_ip = match listen_addr.ip() {
IpAddr::V4(ip) => ip,
IpAddr::V6(_) => {
return Err(anyhow!(
"VRRP over IPv6 (FF02::12) is not implemented; bind an IPv4 interface \
address instead of {listen_addr}"
))
}
};
let raw_socket = create_vrrp_raw_socket(interface_ip).with_context(|| {
format!(
"failed to open a raw IP-protocol-112 socket on {interface_ip}. VRRP and CARP \
ride directly on IP, so this needs CAP_NET_RAW on Linux or root elsewhere. Use \
startup_params.transport = \"udp\" to exercise the protocol without privilege."
)
})?;
let socket_fd = raw_socket.as_raw_fd();
console_info!(
status_tx,
"VRRP server on {} (raw IP protocol 112, joined {}, variant {}, vrid {})",
interface_ip,
VRRP_MULTICAST_IPV4,
config.variant.as_str(),
config.vrid
);
let async_socket = tokio::io::unix::AsyncFd::new(raw_socket)
.context("failed to register the VRRP raw socket with the tokio reactor")?;
let protocol = Arc::new(VrrpProtocol::new());
let config = Arc::new(config);
let task_registrar = app_state.clone();
let accept_handle = tokio::spawn(async move {
let mut buffer = vec![0u8; 65535];
loop {
let mut guard = match async_socket.readable().await {
Ok(guard) => guard,
Err(e) => {
console_error!(status_tx, "VRRP raw socket error: {}", e);
break;
}
};
let read = guard.try_io(|inner| {
let fd = inner.as_raw_fd();
let n = unsafe {
libc::recv(
fd,
buffer.as_mut_ptr() as *mut libc::c_void,
buffer.len(),
0,
)
};
if n < 0 {
Err(std::io::Error::last_os_error())
} else {
Ok(n as usize)
}
});
let n = match read {
Ok(Ok(n)) => n,
Ok(Err(e)) => {
console_error!(status_tx, "VRRP raw receive error: {}", e);
continue;
}
Err(_would_block) => continue,
};
if n < IPV4_HEADER_MIN_LEN {
continue;
}
let header_len = ((buffer[0] & 0x0f) as usize) * 4;
if header_len < IPV4_HEADER_MIN_LEN || n <= header_len {
continue;
}
let source = Ipv4Addr::new(buffer[12], buffer[13], buffer[14], buffer[15]);
let destination = Ipv4Addr::new(buffer[16], buffer[17], buffer[18], buffer[19]);
let packet = buffer[header_len..n].to_vec();
let payload_len = packet.len();
console_debug!(
status_tx,
"VRRP received {} byte message from {} to {}",
payload_len,
source,
destination
);
console_trace!(status_tx, "VRRP message (hex): {}", hex::encode(&packet));
let responder = Responder::Raw {
fd: socket_fd,
source: interface_ip,
};
let inbound_pseudo = PseudoHeader::new(source, destination);
Self::spawn_handler(
packet,
source,
inbound_pseudo,
payload_len,
responder,
&llm_client,
&app_state,
&status_tx,
&protocol,
&config,
server_id,
SocketAddr::new(IpAddr::V4(interface_ip), 0),
SocketAddr::new(IpAddr::V4(source), 0),
)
.await;
}
warn!("VRRP raw receive loop terminated");
});
task_registrar
.register_server_task(server_id, accept_handle)
.await;
Ok(SocketAddr::new(IpAddr::V4(interface_ip), 0))
}
#[allow(clippy::too_many_arguments)]
async fn spawn_handler(
packet: Vec<u8>,
source: Ipv4Addr,
inbound_pseudo: PseudoHeader,
bytes: usize,
responder: Responder,
llm_client: &OllamaClient,
app_state: &Arc<AppState>,
status_tx: &mpsc::UnboundedSender<String>,
protocol: &Arc<VrrpProtocol>,
config: &Arc<VrrpGroupConfig>,
server_id: ServerId,
local_addr: SocketAddr,
peer_addr: SocketAddr,
) {
let llm = llm_client.clone();
let state = app_state.clone();
let status = status_tx.clone();
let proto = protocol.clone();
let cfg = config.clone();
let task_owner = app_state.clone();
task_owner
.spawn_server_task(server_id, async move {
Self::record_connection(&state, server_id, local_addr, peer_addr, bytes).await;
Self::handle_packet(
packet,
source,
inbound_pseudo,
responder,
llm,
state,
status,
proto,
cfg,
server_id,
)
.await;
})
.await;
}
async fn record_connection(
state: &Arc<AppState>,
server_id: ServerId,
local_addr: SocketAddr,
peer_addr: SocketAddr,
bytes: usize,
) {
use crate::state::server::{
ConnectionState as ServerConnectionState, ConnectionStatus, ProtocolConnectionInfo,
};
let connection_id = ConnectionId::new(state.get_next_unified_id().await);
let now = crate::utils::clock::Instant::now();
state
.add_connection_to_server(
server_id,
ServerConnectionState {
id: connection_id,
remote_addr: peer_addr,
local_addr,
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::empty(),
},
)
.await;
}
#[allow(clippy::too_many_arguments)]
async fn handle_packet(
packet: Vec<u8>,
source: Ipv4Addr,
inbound_pseudo: PseudoHeader,
responder: Responder,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
protocol: Arc<VrrpProtocol>,
config: Arc<VrrpGroupConfig>,
server_id: ServerId,
) {
let decoded = match Advertisement::decode(config.variant, &packet) {
Ok(decoded) => decoded,
Err(e) => {
console_debug!(status_tx, "VRRP ignoring message from {}: {:#}", source, e);
return;
}
};
let (event_type, data) = match &decoded {
Advertisement::Vrrp(advertisement) => {
let scope = advertisement.checksum_scope();
let checksum_valid = match scope {
codec::ChecksumScope::Message => codec::checksum_is_valid(&packet, None),
codec::ChecksumScope::PseudoHeaderAndMessage => {
codec::checksum_is_valid(&packet, Some(&inbound_pseudo))
}
};
let mut data = serde_json::json!({
"variant": Variant::Vrrp.as_str(),
"version": advertisement.version,
"vrid": advertisement.vrid,
"priority": advertisement.priority,
"advert_interval": advertisement.advert_interval_seconds,
"addresses": advertisement.addresses.iter().map(|a| a.to_string())
.collect::<Vec<_>>(),
"address_count": advertisement.addresses.len(),
"source_address": source.to_string(),
"checksum_valid": checksum_valid,
"checksum_scope": scope.as_str(),
"auth_type": advertisement.auth_type,
"is_address_owner": advertisement.is_address_owner(),
"is_resignation": advertisement.is_resignation(),
});
merge(&mut data, config.local_summary());
if advertisement.is_resignation() {
(&*VRRP_MASTER_RESIGNED_EVENT, data)
} else {
(&*VRRP_ADVERTISEMENT_RECEIVED_EVENT, data)
}
}
Advertisement::Carp(advertisement) => {
let checksum_valid = codec::checksum_is_valid(&packet, None);
let hmac_valid = if config.carp_passphrase.is_empty() {
serde_json::Value::Null
} else {
serde_json::Value::Bool(
codec::carp_hmac(
config.carp_passphrase.as_bytes(),
advertisement.vhid,
&config.addresses,
advertisement.counter,
) == advertisement.hmac,
)
};
let mut data = serde_json::json!({
"variant": Variant::Carp.as_str(),
"version": advertisement.version,
"vrid": advertisement.vhid,
"advert_interval": advertisement.interval_seconds(),
"advskew": advertisement.advskew,
"advbase": advertisement.advbase,
"demote": advertisement.demote,
"counter": advertisement.counter,
"source_address": source.to_string(),
"checksum_valid": checksum_valid,
"checksum_scope": codec::ChecksumScope::Message.as_str(),
"hmac_valid": hmac_valid,
});
merge(&mut data, config.local_summary());
(&*VRRP_ADVERTISEMENT_RECEIVED_EVENT, data)
}
};
info!(
"VRRP {} from {}: vrid={} priority={:?}",
data.get("variant").and_then(|v| v.as_str()).unwrap_or("?"),
source,
data.get("vrid").and_then(|v| v.as_u64()).unwrap_or(0),
data.get("priority").and_then(|v| v.as_u64()),
);
let event = Event::new(event_type, data);
Self::dispatch_event(
event, responder, llm_client, app_state, status_tx, protocol, config, server_id,
)
.await;
}
#[allow(clippy::too_many_arguments)]
async fn dispatch_event(
event: Event,
responder: Responder,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
protocol: Arc<VrrpProtocol>,
config: Arc<VrrpGroupConfig>,
server_id: ServerId,
) {
let event_id = event.event_type.id.clone();
if !operator_wants_dynamic(&app_state, server_id, &event_id).await {
debug!(
"VRRP decision=no_policy: {} observed, no operator policy configured (no \
instruction and no handler), nothing transmitted and no LLM call",
event_id
);
let _ = status_tx.send(format!(
"VRRP decision=no_policy: {} observed passively (no policy configured, no LLM)",
event_id
));
return;
}
match call_llm(
&llm_client,
&app_state,
server_id,
None,
&event,
protocol.as_ref(),
)
.await
{
Ok(execution_result) => {
for message in &execution_result.messages {
info!("{}", message);
let _ = status_tx.send(format!("[INFO] {}", message));
}
let mut sent = 0usize;
for result in execution_result.protocol_results {
let crate::llm::actions::protocol_trait::ActionResult::Custom { name, data } =
&result
else {
continue;
};
if name.as_str() != "vrrp_action" {
continue;
}
match Self::packet_from_action(&config, data, responder.source_address()) {
Ok((packet, destination)) => {
match responder.send(&packet, destination).await {
Ok(()) => {
sent += 1;
debug!(
"VRRP sent {} byte advertisement to {}",
packet.len(),
responder.describe(destination)
);
trace!("VRRP sent (hex): {}", hex::encode(&packet));
let _ = status_tx.send(format!(
"[DEBUG] VRRP sent {} byte advertisement to {}",
packet.len(),
responder.describe(destination)
));
}
Err(e) => {
error!("VRRP transmit failed: {:#}", e);
let _ =
status_tx.send(format!("✗ VRRP transmit failed: {:#}", e));
}
}
}
Err(e) => {
error!("VRRP could not build the requested advertisement: {:#}", e);
let _ = status_tx.send(format!(
"✗ VRRP decision=model_invalid: could not build the requested \
advertisement, nothing transmitted: {:#}",
e
));
}
}
}
if sent == 0 {
let model_rejected = execution_result.raw_actions.iter().any(|a| {
a.get("type").and_then(|t| t.as_str()) == Some("no_advertisement")
});
let decision = if model_rejected {
"model_reject"
} else {
"model_silent"
};
info!(
"VRRP {} answered with no advertisement: decision={} (VRRP has no error \
message; staying silent is the protocol-correct response)",
event_id, decision
);
let _ = status_tx.send(format!(
"VRRP decision={}: {} answered with no advertisement",
decision, event_id
));
}
}
Err(e) => {
let category = crate::utils::WireFailure::classify(&e);
let decision = if category.is_overloaded() {
"fail_closed_overloaded"
} else {
"fail_closed_llm_error"
};
error!(
"VRRP decision={}: LLM call failed for {}, nothing transmitted (VRRP has no \
failure message and a fabricated advertisement can black-hole the \
segment): {}",
decision, event_id, e
);
let _ = status_tx.send(format!(
"✗ VRRP decision={}: {} unanswered ({}), nothing transmitted: {}",
decision,
event_id,
category.text(),
e
));
}
}
}
fn packet_from_action(
config: &VrrpGroupConfig,
data: &serde_json::Value,
source: Ipv4Addr,
) -> Result<(Vec<u8>, Ipv4Addr)> {
let action_type = data
.get("type")
.and_then(|v| v.as_str())
.context("VRRP action has no 'type'")?;
if action_type != "send_vrrp_advertisement" {
return Err(anyhow!("unknown VRRP action '{action_type}'"));
}
let mut data = data.clone();
config.apply_defaults(&mut data);
let destination = VrrpProtocol::destination_from_action(&data)?;
let advertisement = VrrpProtocol::advertisement_from_action(&data, config)?;
let pseudo = PseudoHeader::new(source, destination);
let packet = advertisement.encode(Some(&pseudo))?;
Ok((packet, destination))
}
}
fn merge(target: &mut serde_json::Value, extra: serde_json::Value) {
if let (Some(target), Some(extra)) = (target.as_object_mut(), extra.as_object()) {
for (key, value) in extra {
target.insert(key.clone(), value.clone());
}
}
}
async fn operator_wants_dynamic(state: &AppState, server_id: 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)
}
fn create_vrrp_raw_socket(interface_addr: Ipv4Addr) -> Result<socket2::Socket> {
use std::os::unix::io::FromRawFd;
let socket = unsafe {
let fd = libc::socket(
libc::AF_INET,
libc::SOCK_RAW,
codec::IP_PROTOCOL_VRRP as libc::c_int,
);
if fd < 0 {
return Err(std::io::Error::last_os_error().into());
}
socket2::Socket::from_raw_fd(fd)
};
socket.set_nonblocking(true)?;
socket.set_reuse_address(true)?;
socket.join_multicast_v4(&VRRP_MULTICAST_IPV4, &interface_addr)?;
socket.set_multicast_ttl_v4(255)?;
socket.set_multicast_if_v4(&interface_addr)?;
Ok(socket)
}
fn send_raw(fd: i32, destination: Ipv4Addr, packet: &[u8]) -> Result<()> {
unsafe {
let mut addr = std::mem::zeroed::<libc::sockaddr_in>();
#[cfg(target_os = "macos")]
{
addr.sin_family = libc::AF_INET as libc::sa_family_t;
addr.sin_len = std::mem::size_of::<libc::sockaddr_in>() as u8;
}
#[cfg(not(target_os = "macos"))]
{
addr.sin_family = libc::AF_INET as u16;
}
addr.sin_port = 0;
addr.sin_addr.s_addr = u32::from(destination).to_be();
let sent = libc::sendto(
fd,
packet.as_ptr() as *const libc::c_void,
packet.len(),
0,
&addr as *const _ as *const libc::sockaddr,
std::mem::size_of::<libc::sockaddr_in>() as u32,
);
if sent < 0 {
return Err(std::io::Error::last_os_error().into());
}
}
Ok(())
}