pub mod actions;
pub use actions::ArpClientProtocol;
use anyhow::{Context, Result};
use pcap::{Capture, Device};
use pnet::packet::arp::{ArpHardwareTypes, ArpOperations, ArpPacket, MutableArpPacket};
use pnet::packet::ethernet::{EtherTypes, EthernetPacket, MutableEthernetPacket};
use pnet::packet::Packet;
use pnet::util::MacAddr;
use std::net::{Ipv4Addr, SocketAddr};
use std::str::FromStr;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};
use tracing::{debug, error, info, trace};
use crate::client::arp::actions::{ARP_CLIENT_RESPONSE_RECEIVED_EVENT, ARP_CLIENT_STARTED_EVENT};
use crate::client::llm_budget::call_llm_for_client;
use crate::llm::actions::client_trait::{Client, ClientActionResult};
use crate::llm::ollama_client::OllamaClient;
use crate::llm::ClientLlmResult;
use crate::logging::emit::Log;
use crate::protocol::Event;
use crate::state::app_state::AppState;
use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
use crate::state::{AccessLogOwner, ClientId, ClientStatus};
#[derive(Debug, Clone, PartialEq)]
enum ConnectionState {
Idle,
Processing,
}
pub struct InjectedPacket {
pub frame: Vec<u8>,
pub ack: Option<tokio::sync::oneshot::Sender<std::result::Result<usize, String>>>,
}
const INJECTION_ACK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
struct ClientData {
state: ConnectionState,
memory: String,
}
pub struct ArpClient;
impl ArpClient {
pub fn list_devices() -> Result<Vec<Device>> {
Device::list().context("Failed to list network devices")
}
pub fn find_device(name: &str) -> Result<Device> {
let devices = Self::list_devices()?;
devices
.into_iter()
.find(|d| d.name == name)
.ok_or_else(|| anyhow::anyhow!("Device '{}' not found", name))
}
pub async fn start_with_llm_actions(
interface: String,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
client_id: ClientId,
) -> Result<SocketAddr> {
info!(
"ARP client {} starting on interface: {}",
client_id, interface
);
let _ = status_tx.send(format!(
"[CLIENT] ARP client {} starting on interface: {}",
client_id, interface
));
app_state
.update_client_status(client_id, ClientStatus::Connected)
.await;
let _ = status_tx.send("__UPDATE_UI__".to_string());
let client_data = Arc::new(Mutex::new(ClientData {
state: ConnectionState::Idle,
memory: String::new(),
}));
let instruction = app_state
.get_instruction_for_client(client_id)
.await
.unwrap_or_else(|| "Monitor ARP traffic".to_string());
let protocol = Arc::new(ArpClientProtocol::new());
let event = Event::new(
&ARP_CLIENT_STARTED_EVENT,
serde_json::json!({
"interface": interface,
}),
);
let (packet_tx, packet_rx) = mpsc::unbounded_channel::<InjectedPacket>();
let command_rx =
crate::client::command_support::register_command_channel(&app_state, client_id).await;
let cmd_task = tokio::spawn(Self::command_loop(
command_rx,
protocol.clone(),
packet_tx.clone(),
client_id,
app_state.clone(),
status_tx.clone(),
));
app_state.register_client_task(client_id, cmd_task).await;
let memory_snapshot = client_data.lock().await.memory.clone();
match call_llm_for_client(
&llm_client,
&app_state,
client_id.to_string(),
&instruction,
&memory_snapshot,
Some(&event),
protocol.as_ref(),
&status_tx,
)
.await
{
Ok(ClientLlmResult {
actions,
memory_updates,
}) => {
if let Some(mem) = memory_updates {
client_data.lock().await.memory = mem;
}
Self::execute_client_actions(
actions,
protocol.as_ref(),
&packet_tx,
client_id,
&app_state,
&status_tx,
)
.await;
}
Err(e) => {
error!("ARP client {} initial LLM call failed: {}", client_id, e);
}
}
let interface_clone = interface.clone();
let protocol_clone = protocol.clone();
let app_state_clone = app_state.clone();
let status_tx_clone = status_tx.clone();
let llm_client_clone = llm_client.clone();
let client_data_clone = client_data.clone();
let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<Result<()>>();
let stop = crate::utils::StopSignal::new();
let stop_in_loop = stop.clone();
tokio::task::spawn_blocking(move || {
let open_captures = || -> Result<(Capture<pcap::Active>, Capture<pcap::Active>)> {
let device = Self::find_device(&interface_clone)
.with_context(|| format!("no such capture device '{}'", interface_clone))?;
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)",
interface_clone
)
})?;
cap_rx
.filter("arp", true)
.context("failed to apply the 'arp' BPF filter")?;
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 '{}'",
interface_clone
)
})?;
Ok((cap_rx, cap_tx))
};
let (mut cap_rx, mut cap_tx) = match open_captures() {
Ok(handles) => {
let _ = ready_tx.send(Ok(()));
handles
}
Err(e) => {
Log::new(Some(&status_tx_clone))
.error(format!("ARP client capture startup failed: {:#}", e));
let _ = ready_tx.send(Err(e));
return;
}
};
let runtime = tokio::runtime::Handle::current();
let mut packet_rx = packet_rx;
std::thread::spawn(move || {
while let Some(packet) = packet_rx.blocking_recv() {
let len = packet.frame.len();
let result = match cap_tx.sendpacket(&packet.frame[..]) {
Ok(()) => Ok(len),
Err(e) => {
error!("Failed to send ARP packet: {}", e);
Err(e.to_string())
}
};
if let Some(ack) = packet.ack {
let _ = ack.send(result);
}
}
});
loop {
if stop_in_loop.is_stopped() {
info!(
"ARP client {} capture on {} stopping",
client_id, interface_clone
);
break;
}
match cap_rx.next_packet() {
Ok(packet) => {
let data = packet.data.to_vec();
let eth_packet = match EthernetPacket::new(&data) {
Some(p) => p,
None => {
debug!("Failed to parse Ethernet packet");
continue;
}
};
if eth_packet.get_ethertype() != EtherTypes::Arp {
continue;
}
let arp_packet = match ArpPacket::new(eth_packet.payload()) {
Some(p) => p,
None => {
debug!("Failed to parse ARP packet");
continue;
}
};
let operation = arp_packet.get_operation();
let sender_mac = arp_packet.get_sender_hw_addr();
let sender_ip = arp_packet.get_sender_proto_addr();
let target_mac = arp_packet.get_target_hw_addr();
let target_ip = arp_packet.get_target_proto_addr();
Log::new(Some(&status_tx_clone)).debug(format!(
"ARP client {} received {} from {} ({}) for {} ({})",
client_id,
operation_to_string(operation),
sender_mac,
sender_ip,
target_mac,
target_ip
));
let hex_str = hex::encode(&data);
trace!("ARP packet (hex): {}", hex_str);
let llm_clone = llm_client_clone.clone();
let state_clone = app_state_clone.clone();
let status_clone = status_tx_clone.clone();
let protocol_task_clone = protocol_clone.clone();
let packet_tx_clone = packet_tx.clone();
let client_data_task_clone = client_data_clone.clone();
runtime.spawn(async move {
let mut client_data_lock = client_data_task_clone.lock().await;
match client_data_lock.state {
ConnectionState::Idle => {
client_data_lock.state = ConnectionState::Processing;
let current_memory = client_data_lock.memory.clone();
drop(client_data_lock);
let event = Event::new(
&ARP_CLIENT_RESPONSE_RECEIVED_EVENT,
serde_json::json!({
"operation": operation_to_string(operation),
"sender_mac": sender_mac.to_string(),
"sender_ip": sender_ip.to_string(),
"target_mac": target_mac.to_string(),
"target_ip": target_ip.to_string(),
}),
);
debug!(
"ARP client {} calling LLM for {} packet",
client_id,
operation_to_string(operation)
);
let instruction = state_clone
.get_instruction_for_client(client_id)
.await
.unwrap_or_else(|| "Monitor ARP traffic".to_string());
match call_llm_for_client(
&llm_clone,
&state_clone,
client_id.to_string(),
&instruction,
¤t_memory,
Some(&event),
protocol_task_clone.as_ref(),
&status_clone,
)
.await
{
Ok(ClientLlmResult {
actions,
memory_updates,
}) => {
if let Some(mem) = memory_updates {
client_data_task_clone.lock().await.memory = mem;
}
if Self::execute_client_actions(
actions,
protocol_task_clone.as_ref(),
&packet_tx_clone,
client_id,
&state_clone,
&status_clone,
)
.await
{
return;
}
}
Err(e) => {
error!("LLM error for ARP client {}: {}", client_id, e);
}
}
client_data_task_clone.lock().await.state =
ConnectionState::Idle;
}
ConnectionState::Processing => {
debug!(
"ARP client {} is processing, skipping packet",
client_id
);
}
}
});
}
Err(pcap::Error::TimeoutExpired) => {
continue;
}
Err(e) => {
Log::new(Some(&status_tx_clone)).error(format!(
"ARP client {} packet capture error: {}",
client_id, e
));
break;
}
}
}
runtime.block_on(app_state_clone.remove_client_handle(client_id));
let _ = status_tx_clone.send("__UPDATE_UI__".to_string());
});
match ready_rx.await {
Ok(Ok(())) => {}
Ok(Err(e)) => {
app_state.remove_client_handle(client_id).await;
return Err(e);
}
Err(_) => {
app_state.remove_client_handle(client_id).await;
return Err(anyhow::anyhow!(
"ARP capture task on '{}' exited before signalling readiness",
interface
));
}
}
app_state
.register_client_task(client_id, stop.park_task())
.await;
Ok(SocketAddr::from(([0, 0, 0, 0], 0)))
}
async fn execute_client_actions(
actions: Vec<serde_json::Value>,
protocol: &ArpClientProtocol,
packet_tx: &mpsc::UnboundedSender<InjectedPacket>,
client_id: ClientId,
app_state: &Arc<AppState>,
status_tx: &mpsc::UnboundedSender<String>,
) -> bool {
for action in actions {
match protocol.execute_action(action) {
Ok(ClientActionResult::Custom { name, data }) => {
match build_packet_for_custom_result(&name, &data) {
Some(packet) => {
if packet_tx
.send(InjectedPacket {
frame: packet,
ack: None,
})
.is_ok()
{
debug!("ARP client {} queued {}", client_id, name);
} else {
error!(
"ARP client {} could not queue '{}': the pcap injection \
thread is not running",
client_id, name
);
}
}
None => error!(
"ARP client {} could not build a frame for '{}': check the MAC and \
IP fields",
client_id, name
),
}
}
Ok(ClientActionResult::Disconnect) => {
info!("ARP client {} stopping capture", client_id);
app_state
.update_client_status(client_id, ClientStatus::Disconnected)
.await;
let _ = status_tx.send("__UPDATE_UI__".to_string());
return true;
}
Ok(_) => {}
Err(e) => {
error!("ARP client {} action failed: {}", client_id, e);
let _ = status_tx.send(format!(
"[WARN] ARP client {} action failed: {}",
client_id, e
));
}
}
}
false
}
pub async fn command_loop(
mut command_rx: mpsc::Receiver<ClientCommand>,
protocol: Arc<ArpClientProtocol>,
packet_tx: mpsc::UnboundedSender<InjectedPacket>,
client_id: ClientId,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
) {
use crate::llm::actions::protocol_trait::Protocol;
while let Some(command) = command_rx.recv().await {
let action = command.action.clone();
let outcome = Self::execute_injected_action(&protocol, &packet_tx, &action).await;
let outcome_json = match &outcome {
Ok(outcome) => serde_json::to_value(outcome).unwrap_or(serde_json::Value::Null),
Err(e) => serde_json::json!({"error": e.to_string()}),
};
app_state
.record_access_log(
AccessLogOwner::Client(client_id.as_u32()),
protocol.protocol_name(),
None,
"injected_action",
action,
vec![outcome_json],
)
.await;
let disconnect = matches!(outcome, Ok(ClientSendOutcome::Disconnected));
if let Err(e) = &outcome {
error!("ARP client {} injected action failed: {}", client_id, e);
let _ = status_tx.send(format!(
"[WARN] Client {} injected action failed: {}",
client_id, e
));
}
let _ = status_tx.send("__UPDATE_UI__".to_string());
crate::client::command_support::reply(command, outcome);
if disconnect {
app_state.remove_client_handle(client_id).await;
app_state
.update_client_status(client_id, ClientStatus::Disconnected)
.await;
let _ = status_tx.send("__UPDATE_UI__".to_string());
break;
}
}
}
async fn execute_injected_action(
protocol: &Arc<ArpClientProtocol>,
packet_tx: &mpsc::UnboundedSender<InjectedPacket>,
action: &serde_json::Value,
) -> Result<ClientSendOutcome> {
let result = match protocol.as_ref().execute_action(action.clone()) {
Ok(result) => result,
Err(e) => {
return Ok(ClientSendOutcome::Rejected {
error: e.to_string(),
})
}
};
let (name, data) = match result {
ClientActionResult::Custom { name, data } => (name, data),
ClientActionResult::Disconnect => return Ok(ClientSendOutcome::Disconnected),
ClientActionResult::WaitForMore => {
return Ok(ClientSendOutcome::Executed {
detail: "wait_for_more".to_string(),
})
}
other => {
return Ok(ClientSendOutcome::Executed {
detail: format!("{other:?} builds no ARP frame"),
})
}
};
let frame = match build_packet_for_custom_result(&name, &data) {
Some(frame) => frame,
None => {
return Ok(ClientSendOutcome::Rejected {
error: format!(
"could not build an ARP frame for '{name}': check the MAC and IP fields"
),
})
}
};
let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
if packet_tx
.send(InjectedPacket {
frame,
ack: Some(ack_tx),
})
.is_err()
{
return Ok(ClientSendOutcome::Executed {
detail: format!(
"'{name}' frame built but not injected: the pcap injection thread is not \
running (the capture failed to open - ARP capture and injection need \
root / BPF access)"
),
});
}
match tokio::time::timeout(INJECTION_ACK_TIMEOUT, ack_rx).await {
Ok(Ok(Ok(bytes_sent))) => Ok(ClientSendOutcome::Sent { bytes_sent }),
Ok(Ok(Err(e))) => Err(anyhow::anyhow!("pcap sendpacket failed: {e}")),
Ok(Err(_)) => Ok(ClientSendOutcome::Executed {
detail: format!(
"'{name}' frame handed to the pcap injection thread, which exited \
without reporting the result"
),
}),
Err(_) => Ok(ClientSendOutcome::Executed {
detail: format!(
"'{name}' frame queued, but the pcap injection thread did not \
acknowledge it within {}s",
INJECTION_ACK_TIMEOUT.as_secs()
),
}),
}
}
}
pub fn build_packet_for_custom_result(name: &str, data: &serde_json::Value) -> Option<Vec<u8>> {
match name {
"send_arp_request" => build_arp_request_from_action(data),
"send_arp_reply" => build_arp_reply_from_action(data),
_ => None,
}
}
fn operation_to_string(op: pnet::packet::arp::ArpOperation) -> &'static str {
match op {
ArpOperations::Request => "REQUEST",
ArpOperations::Reply => "REPLY",
_ => "UNKNOWN",
}
}
fn build_arp_request_from_action(data: &serde_json::Value) -> Option<Vec<u8>> {
let sender_mac = data["sender_mac"].as_str()?;
let sender_ip = data["sender_ip"].as_str()?;
let target_ip = data["target_ip"].as_str()?;
let sender_mac = MacAddr::from_str(sender_mac).ok()?;
let sender_ip = Ipv4Addr::from_str(sender_ip).ok()?;
let target_ip = Ipv4Addr::from_str(target_ip).ok()?;
Some(build_arp_request(sender_mac, sender_ip, target_ip))
}
fn build_arp_reply_from_action(data: &serde_json::Value) -> Option<Vec<u8>> {
let sender_mac = data["sender_mac"].as_str()?;
let sender_ip = data["sender_ip"].as_str()?;
let target_mac = data["target_mac"].as_str()?;
let target_ip = data["target_ip"].as_str()?;
let sender_mac = MacAddr::from_str(sender_mac).ok()?;
let sender_ip = Ipv4Addr::from_str(sender_ip).ok()?;
let target_mac = MacAddr::from_str(target_mac).ok()?;
let target_ip = Ipv4Addr::from_str(target_ip).ok()?;
Some(build_arp_reply(
sender_mac, sender_ip, target_mac, target_ip,
))
}
fn build_arp_request(sender_mac: MacAddr, sender_ip: Ipv4Addr, target_ip: Ipv4Addr) -> Vec<u8> {
let mut eth_buffer = vec![0u8; 42];
{
let mut eth_packet = MutableEthernetPacket::new(&mut eth_buffer).unwrap();
eth_packet.set_destination(MacAddr::broadcast()); eth_packet.set_source(sender_mac);
eth_packet.set_ethertype(EtherTypes::Arp);
let mut arp_buffer = vec![0u8; 28];
{
let mut arp_packet = MutableArpPacket::new(&mut arp_buffer).unwrap();
arp_packet.set_hardware_type(ArpHardwareTypes::Ethernet);
arp_packet.set_protocol_type(EtherTypes::Ipv4);
arp_packet.set_hw_addr_len(6);
arp_packet.set_proto_addr_len(4);
arp_packet.set_operation(ArpOperations::Request);
arp_packet.set_sender_hw_addr(sender_mac);
arp_packet.set_sender_proto_addr(sender_ip);
arp_packet.set_target_hw_addr(MacAddr::zero()); arp_packet.set_target_proto_addr(target_ip);
}
eth_packet.set_payload(&arp_buffer);
}
eth_buffer
}
fn build_arp_reply(
sender_mac: MacAddr,
sender_ip: Ipv4Addr,
target_mac: MacAddr,
target_ip: Ipv4Addr,
) -> Vec<u8> {
let mut eth_buffer = vec![0u8; 42];
{
let mut eth_packet = MutableEthernetPacket::new(&mut eth_buffer).unwrap();
eth_packet.set_destination(target_mac);
eth_packet.set_source(sender_mac);
eth_packet.set_ethertype(EtherTypes::Arp);
let mut arp_buffer = vec![0u8; 28];
{
let mut arp_packet = MutableArpPacket::new(&mut arp_buffer).unwrap();
arp_packet.set_hardware_type(ArpHardwareTypes::Ethernet);
arp_packet.set_protocol_type(EtherTypes::Ipv4);
arp_packet.set_hw_addr_len(6);
arp_packet.set_proto_addr_len(4);
arp_packet.set_operation(ArpOperations::Reply);
arp_packet.set_sender_hw_addr(sender_mac);
arp_packet.set_sender_proto_addr(sender_ip);
arp_packet.set_target_hw_addr(target_mac);
arp_packet.set_target_proto_addr(target_ip);
}
eth_packet.set_payload(&arp_buffer);
}
eth_buffer
}