1pub mod actions;
34pub mod codec;
35
36use anyhow::{anyhow, bail, Context, Result};
37use std::net::SocketAddr;
38use std::sync::Arc;
39use tokio::net::UdpSocket;
40use tokio::sync::mpsc;
41use tracing::{debug, error, info, trace, warn};
42
43use crate::llm::action_helper::call_llm;
44use crate::llm::ollama_client::OllamaClient;
45use crate::protocol::Event;
46use crate::state::app_state::AppState;
47use crate::{console_debug, console_error, console_info, console_trace};
48use actions::{
49 StpBridgeConfig, StpProtocol, StpTransport, STP_BPDU_RECEIVED_EVENT, STP_TOPOLOGY_CHANGE_EVENT,
50};
51
52#[derive(Clone)]
59enum Responder {
60 Udp {
61 socket: Arc<UdpSocket>,
62 peer: SocketAddr,
63 },
64 Pcap(mpsc::UnboundedSender<Vec<u8>>),
65}
66
67impl Responder {
68 async fn send(&self, frame: Vec<u8>) -> Result<()> {
69 match self {
70 Responder::Udp { socket, peer } => {
71 socket
72 .send_to(&frame, peer)
73 .await
74 .with_context(|| format!("failed to send STP frame to {peer}"))?;
75 Ok(())
76 }
77 Responder::Pcap(tx) => tx
78 .send(frame)
79 .map_err(|_| anyhow!("STP injection thread has exited")),
80 }
81 }
82
83 fn describe(&self) -> String {
84 match self {
85 Responder::Udp { peer, .. } => peer.to_string(),
86 Responder::Pcap(_) => "the segment".to_string(),
87 }
88 }
89}
90
91pub struct StpServer;
93
94impl StpServer {
95 pub async fn spawn_with_llm_actions(
103 interface: Option<String>,
104 listen_addr: SocketAddr,
105 config: StpBridgeConfig,
106 llm_client: OllamaClient,
107 app_state: Arc<AppState>,
108 status_tx: mpsc::UnboundedSender<String>,
109 server_id: crate::state::ServerId,
110 ) -> Result<SocketAddr> {
111 config.bridge_id()?;
113 config.port_id()?;
114
115 match config.transport {
116 StpTransport::Udp => {
117 Self::spawn_udp(
118 listen_addr,
119 config,
120 llm_client,
121 app_state,
122 status_tx,
123 server_id,
124 )
125 .await
126 }
127 StpTransport::Raw => {
128 let interface = interface
129 .context("STP raw transport requires a network interface (e.g. eth0, en0)")?;
130 Self::spawn_raw(
131 interface,
132 listen_addr,
133 config,
134 llm_client,
135 app_state,
136 status_tx,
137 server_id,
138 )
139 .await
140 }
141 }
142 }
143
144 async fn spawn_udp(
149 listen_addr: SocketAddr,
150 config: StpBridgeConfig,
151 llm_client: OllamaClient,
152 app_state: Arc<AppState>,
153 status_tx: mpsc::UnboundedSender<String>,
154 server_id: crate::state::ServerId,
155 ) -> Result<SocketAddr> {
156 let socket = Arc::new(
157 UdpSocket::bind(listen_addr)
158 .await
159 .with_context(|| format!("failed to bind STP UDP transport to {listen_addr}"))?,
160 );
161 let local_addr = socket.local_addr()?;
162 console_info!(
163 status_tx,
164 "STP server listening on {} (UDP transport: one complete 802.3 BPDU frame per datagram)",
165 local_addr
166 );
167
168 let protocol = Arc::new(StpProtocol::new());
169 let config = Arc::new(config);
170 let task_registrar = app_state.clone();
171
172 let recv_socket = socket.clone();
173 let accept_handle = tokio::spawn(async move {
174 let mut buffer = vec![0u8; 2048];
178 loop {
179 match recv_socket.recv_from(&mut buffer).await {
180 Ok((n, peer)) => {
181 let frame = buffer[..n].to_vec();
182 console_debug!(status_tx, "STP received {} byte frame from {}", n, peer);
183 console_trace!(status_tx, "STP frame (hex): {}", hex::encode(&frame));
184
185 let responder = Responder::Udp {
186 socket: socket.clone(),
187 peer,
188 };
189 let llm = llm_client.clone();
190 let state = app_state.clone();
191 let status = status_tx.clone();
192 let proto = protocol.clone();
193 let cfg = config.clone();
194 let task_owner = app_state.clone();
196 task_owner
197 .spawn_server_task(server_id, async move {
198 Self::handle_frame(
199 frame, llm, state, status, proto, cfg, responder, server_id,
200 )
201 .await;
202 })
203 .await;
204 }
205 Err(e) => {
206 console_error!(status_tx, "STP UDP receive error: {}", e);
207 break;
208 }
209 }
210 }
211 warn!("STP UDP receive loop terminated");
212 });
213
214 task_registrar
215 .register_server_task(server_id, accept_handle)
216 .await;
217
218 Ok(local_addr)
219 }
220
221 #[allow(clippy::too_many_arguments)]
228 async fn spawn_raw(
229 interface: String,
230 listen_addr: SocketAddr,
231 config: StpBridgeConfig,
232 llm_client: OllamaClient,
233 app_state: Arc<AppState>,
234 status_tx: mpsc::UnboundedSender<String>,
235 server_id: crate::state::ServerId,
236 ) -> Result<SocketAddr> {
237 use pcap::{Capture, Device};
238
239 console_info!(
240 status_tx,
241 "Starting STP capture on interface: {}",
242 interface
243 );
244
245 let status_tx_ready = status_tx.clone();
246 let protocol = Arc::new(StpProtocol::new());
247 let config = Arc::new(config);
248
249 let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<Result<()>>();
253
254 let stop = crate::utils::StopSignal::new();
257 let stop_in_loop = stop.clone();
258 let app_state_reg = app_state.clone();
259
260 let interface_clone = interface.clone();
261 tokio::task::spawn_blocking(move || {
262 let open_captures = || -> Result<(Capture<pcap::Active>, Capture<pcap::Active>)> {
263 let device = Device::list()
264 .context("failed to list capture devices")?
265 .into_iter()
266 .find(|d| d.name == interface_clone)
267 .with_context(|| format!("no such capture device '{interface_clone}'"))?;
268
269 let mut cap_rx = Capture::from_device(device.clone())
270 .map(|c| c.promisc(true).snaplen(65535).timeout(1000))
271 .and_then(|c| c.open())
272 .with_context(|| {
273 format!(
274 "failed to open pcap capture on '{interface_clone}' (needs root, or \
275 read access to /dev/bpf* on macOS/BSD, or CAP_NET_RAW on Linux)"
276 )
277 })?;
278
279 cap_rx
288 .filter("ether dst 01:80:c2:00:00:00", true)
289 .with_context(|| {
290 format!(
291 "failed to apply the BPDU BPF filter on '{interface_clone}'. BPDUs \
292 are 802.3 frames addressed to the Bridge Group Address, so this \
293 fails on any interface with no Ethernet link layer — loopback \
294 (lo/lo0), tunnels and raw-IP devices carry no spanning tree. Point \
295 this server at a real Ethernet interface, or use \
296 startup_params.transport = \"udp\"."
297 )
298 })?;
299
300 let cap_tx = Capture::from_device(device)
301 .map(|c| c.promisc(true).snaplen(65535).timeout(1000))
302 .and_then(|c| c.open())
303 .with_context(|| {
304 format!("failed to open pcap injection handle on '{interface_clone}'")
305 })?;
306
307 Ok((cap_rx, cap_tx))
308 };
309
310 let (mut cap_rx, mut cap_tx) = match open_captures() {
311 Ok(handles) => {
312 let _ = ready_tx.send(Ok(()));
313 handles
314 }
315 Err(e) => {
316 console_error!(status_tx, "STP capture startup failed: {:#}", e);
317 let _ = ready_tx.send(Err(e));
318 return;
319 }
320 };
321
322 let runtime = tokio::runtime::Handle::current();
323 let (packet_tx, mut packet_rx) = mpsc::unbounded_channel::<Vec<u8>>();
324
325 std::thread::spawn(move || {
327 while let Some(frame) = packet_rx.blocking_recv() {
328 if let Err(e) = cap_tx.sendpacket(frame) {
329 error!("STP failed to inject frame: {}", e);
330 }
331 }
332 });
333
334 loop {
335 if stop_in_loop.is_stopped() {
336 console_info!(status_tx, "STP capture on {} stopping", interface_clone);
337 break;
338 }
339 match cap_rx.next_packet() {
340 Ok(packet) => {
341 let frame = packet.data.to_vec();
342 console_debug!(
343 status_tx,
344 "STP captured {} byte frame on {}",
345 frame.len(),
346 interface_clone
347 );
348 console_trace!(status_tx, "STP frame (hex): {}", hex::encode(&frame));
349
350 let responder = Responder::Pcap(packet_tx.clone());
351 let llm = llm_client.clone();
352 let state = app_state.clone();
353 let status = status_tx.clone();
354 let proto = protocol.clone();
355 let cfg = config.clone();
356 runtime.spawn(async move {
357 Self::handle_frame(
358 frame, llm, state, status, proto, cfg, responder, server_id,
359 )
360 .await;
361 });
362 }
363 Err(pcap::Error::TimeoutExpired) => continue,
364 Err(e) => {
365 console_error!(status_tx, "STP packet capture error: {}", e);
366 break;
367 }
368 }
369 }
370
371 drop(packet_tx);
373 drop(cap_rx);
374 });
375
376 match ready_rx.await {
377 Ok(Ok(())) => {}
378 Ok(Err(e)) => return Err(e),
379 Err(_) => {
380 return Err(anyhow!(
381 "STP capture task on '{}' exited before signalling readiness",
382 interface
383 ))
384 }
385 }
386
387 app_state_reg
388 .register_server_task(server_id, stop.park_task())
389 .await;
390
391 console_info!(status_tx_ready, "STP capture active on {}", interface);
392
393 Ok(listen_addr)
396 }
397
398 #[allow(clippy::too_many_arguments)]
403 async fn handle_frame(
404 frame: Vec<u8>,
405 llm_client: OllamaClient,
406 app_state: Arc<AppState>,
407 status_tx: mpsc::UnboundedSender<String>,
408 protocol: Arc<StpProtocol>,
409 config: Arc<StpBridgeConfig>,
410 responder: Responder,
411 server_id: crate::state::ServerId,
412 ) {
413 let decoded = match codec::decode_frame(&frame) {
414 Ok(decoded) => decoded,
415 Err(e) => {
416 console_debug!(status_tx, "STP ignoring frame: {:#}", e);
417 return;
418 }
419 };
420
421 let bpdu = match codec::Bpdu::decode(&decoded.payload) {
422 Ok(bpdu) => bpdu,
423 Err(e) => {
424 console_debug!(status_tx, "STP ignoring malformed BPDU: {:#}", e);
425 return;
426 }
427 };
428
429 let source_mac = codec::format_mac(&decoded.source);
430 let destination_mac = codec::format_mac(&decoded.destination);
431
432 let (event_type, mut data) = match &bpdu {
433 codec::Bpdu::TopologyChangeNotification => {
434 let data = serde_json::json!({
435 "bpdu_type": "topology_change_notification",
436 "protocol_version": "stp",
437 "is_rstp": false,
438 "is_tcn": true,
439 "source_mac": source_mac,
440 "destination_mac": destination_mac,
441 "change_reason": "tcn_bpdu",
442 });
443 (&*STP_TOPOLOGY_CHANGE_EVENT, data)
444 }
445 codec::Bpdu::Config(config_bpdu) => {
446 let mut data = serde_json::json!({
447 "bpdu_type": if config_bpdu.is_rstp() { "rst" } else { "configuration" },
448 "protocol_version": version_name(config_bpdu.version),
449 "is_rstp": config_bpdu.is_rstp(),
450 "is_tcn": false,
451 "source_mac": source_mac,
452 "destination_mac": destination_mac,
453 "root_bridge_mac": config_bpdu.root.mac_string(),
454 "root_priority": config_bpdu.root.priority,
455 "root_system_id_extension": config_bpdu.root.system_id_extension,
456 "root_path_cost": config_bpdu.root_path_cost,
457 "bridge_mac": config_bpdu.bridge.mac_string(),
458 "bridge_priority": config_bpdu.bridge.priority,
459 "bridge_system_id_extension": config_bpdu.bridge.system_id_extension,
460 "port_priority": config_bpdu.port.priority,
461 "port_number": config_bpdu.port.number,
462 "flags": {
463 "topology_change": config_bpdu.flags.topology_change,
464 "topology_change_ack": config_bpdu.flags.topology_change_ack,
465 "proposal": config_bpdu.flags.proposal,
466 "agreement": config_bpdu.flags.agreement,
467 "learning": config_bpdu.flags.learning,
468 "forwarding": config_bpdu.flags.forwarding,
469 "port_role": config_bpdu.flags.port_role.as_str(),
470 },
471 "message_age": config_bpdu.message_age_seconds,
472 "max_age": config_bpdu.max_age_seconds,
473 "hello_time": config_bpdu.hello_time_seconds,
474 "forward_delay": config_bpdu.forward_delay_seconds,
475 });
476 if config_bpdu.flags.topology_change {
477 if let Some(obj) = data.as_object_mut() {
478 obj.insert(
479 "change_reason".to_string(),
480 serde_json::json!("topology_change_flag"),
481 );
482 }
483 (&*STP_TOPOLOGY_CHANGE_EVENT, data)
484 } else {
485 (&*STP_BPDU_RECEIVED_EVENT, data)
486 }
487 }
488 };
489
490 if let (Some(target), Some(local)) = (
493 data.as_object_mut(),
494 config.local_summary().as_object().cloned(),
495 ) {
496 for (key, value) in local {
497 target.insert(key, value);
498 }
499 }
500
501 info!(
502 "STP {} from {} (root {}, cost {})",
503 data.get("bpdu_type")
504 .and_then(|v| v.as_str())
505 .unwrap_or("bpdu"),
506 data.get("source_mac")
507 .and_then(|v| v.as_str())
508 .unwrap_or(""),
509 data.get("root_bridge_mac")
510 .and_then(|v| v.as_str())
511 .unwrap_or("n/a"),
512 data.get("root_path_cost")
513 .and_then(|v| v.as_u64())
514 .unwrap_or(0),
515 );
516
517 let event = Event::new(event_type, data);
518 Self::dispatch_event(
519 event, llm_client, app_state, status_tx, protocol, config, responder, server_id,
520 )
521 .await;
522 }
523
524 #[allow(clippy::too_many_arguments)]
527 async fn dispatch_event(
528 event: Event,
529 llm_client: OllamaClient,
530 app_state: Arc<AppState>,
531 status_tx: mpsc::UnboundedSender<String>,
532 protocol: Arc<StpProtocol>,
533 config: Arc<StpBridgeConfig>,
534 responder: Responder,
535 server_id: crate::state::ServerId,
536 ) {
537 let event_id = event.event_type.id.clone();
538
539 if !operator_wants_dynamic(&app_state, server_id, &event_id).await {
545 debug!(
546 "STP decision=no_policy: {} observed, no operator policy configured (no \
547 instruction and no handler), nothing transmitted and no LLM call",
548 event_id
549 );
550 let _ = status_tx.send(format!(
551 "STP decision=no_policy: {} observed passively (no policy configured, no LLM)",
552 event_id
553 ));
554 return;
555 }
556
557 match call_llm(
558 &llm_client,
559 &app_state,
560 server_id,
561 None,
562 &event,
563 protocol.as_ref(),
564 )
565 .await
566 {
567 Ok(execution_result) => {
568 for message in &execution_result.messages {
569 info!("{}", message);
570 let _ = status_tx.send(format!("[INFO] {}", message));
571 }
572
573 let mut frames_sent = 0usize;
574 for result in execution_result.protocol_results {
575 let crate::llm::actions::protocol_trait::ActionResult::Custom { name, data } =
576 &result
577 else {
578 continue;
579 };
580 if name.as_str() != "stp_action" {
581 continue;
582 }
583
584 match Self::frame_from_action(&config, data) {
585 Ok(frame) => match responder.send(frame.clone()).await {
586 Ok(()) => {
587 frames_sent += 1;
588 debug!(
589 "STP sent {} byte frame to {}",
590 frame.len(),
591 responder.describe()
592 );
593 trace!("STP sent (hex): {}", hex::encode(&frame));
594 let _ = status_tx.send(format!(
595 "[DEBUG] STP sent {} byte frame to {}",
596 frame.len(),
597 responder.describe()
598 ));
599 }
600 Err(e) => {
601 error!("STP transmit failed: {:#}", e);
602 let _ = status_tx.send(format!("✗ STP transmit failed: {:#}", e));
603 }
604 },
605 Err(e) => {
606 error!("STP could not build the requested BPDU: {:#}", e);
608 let _ = status_tx.send(format!(
609 "✗ STP decision=model_invalid: could not build the requested \
610 BPDU, nothing transmitted: {:#}",
611 e
612 ));
613 }
614 }
615 }
616
617 if frames_sent == 0 {
618 let model_rejected = execution_result
622 .raw_actions
623 .iter()
624 .any(|a| a.get("type").and_then(|t| t.as_str()) == Some("no_bpdu"));
625 let decision = if model_rejected {
626 "model_reject"
627 } else {
628 "model_silent"
629 };
630 info!(
631 "STP {} answered with no BPDU: decision={} (STP has no error BPDU; \
632 staying silent is the protocol-correct response)",
633 event_id, decision
634 );
635 let _ = status_tx.send(format!(
636 "STP decision={}: {} answered with no BPDU",
637 decision, event_id
638 ));
639 }
640 }
641 Err(e) => {
642 let category = crate::utils::WireFailure::classify(&e);
651 let decision = if category.is_overloaded() {
652 "fail_closed_overloaded"
653 } else {
654 "fail_closed_llm_error"
655 };
656 error!(
657 "STP decision={}: LLM call failed for {}, no BPDU transmitted (STP has no \
658 failure BPDU and a fabricated one can re-converge the segment): {}",
659 decision, event_id, e
660 );
661 let _ = status_tx.send(format!(
662 "✗ STP decision={}: {} unanswered ({}), nothing transmitted: {}",
663 decision,
664 event_id,
665 category.text(),
666 e
667 ));
668 }
669 }
670 }
671
672 fn frame_from_action(config: &StpBridgeConfig, data: &serde_json::Value) -> Result<Vec<u8>> {
677 let action_type = data
678 .get("type")
679 .and_then(|v| v.as_str())
680 .context("STP action has no 'type'")?;
681
682 match action_type {
683 "send_stp_bpdu" => {
684 let mut data = data.clone();
685 config.apply_defaults(&mut data);
686 let destination = StpProtocol::destination_from_action(&data)?;
687 let source = StpProtocol::source_from_action(&data, config.bridge_mac)?;
688 let body = StpProtocol::config_bpdu_from_action(&data)?.encode()?;
689 codec::encode_frame(destination, source, &body)
690 }
691 "send_stp_tcn" => {
692 let destination = StpProtocol::destination_from_action(data)?;
693 let source = StpProtocol::source_from_action(data, config.bridge_mac)?;
694 codec::encode_frame(destination, source, &codec::encode_tcn_bpdu())
695 }
696 other => bail!("unknown STP action '{other}'"),
697 }
698 }
699}
700
701fn version_name(version: u8) -> String {
702 match version {
703 codec::VERSION_STP => "stp".to_string(),
704 codec::VERSION_RSTP => "rstp".to_string(),
705 other => other.to_string(),
706 }
707}
708
709async fn operator_wants_dynamic(
715 state: &AppState,
716 server_id: crate::state::ServerId,
717 event_id: &str,
718) -> bool {
719 state
720 .with_server_mut(server_id, |server| {
721 let has_instruction = !server.instruction.trim().is_empty();
722 let has_handler = server
723 .event_handler_config
724 .as_ref()
725 .map(|c| c.find_handler(event_id).is_some())
726 .unwrap_or(false);
727 has_instruction || has_handler
728 })
729 .await
730 .unwrap_or(false)
731}