netget/server/bmp/
mod.rs

1//! BGP Monitoring Protocol collector (RFC 7854). Rust owns framing, the per-peer header, TLVs
2//! and the embedded BGP PDUs; the handler sees each message as JSON and decides whether to keep
3//! monitoring the router. A collector never writes to the router.
4pub mod actions;
5pub mod codec;
6
7use crate::llm::action_helper::call_llm;
8use crate::llm::actions::protocol_trait::ActionResult;
9use crate::logging::emit::Log;
10use crate::protocol::{Event, SpawnContext};
11use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
12use crate::server::connection::ConnectionId;
13use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
14use crate::utils::clock::Instant;
15use anyhow::{bail, Context, Result};
16use serde_json::{json, Value as Json};
17use std::net::SocketAddr;
18use std::sync::Arc;
19use std::time::Duration;
20use tokio::io::AsyncReadExt;
21use tokio::net::TcpStream;
22
23/// Once a message has started, the rest of it must arrive within this long. Between messages a
24/// router may stay silent indefinitely: BMP has no keep-alive.
25pub const MESSAGE_DEADLINE: Duration = Duration::from_secs(60);
26
27pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
28    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
29    let local = listener.local_addr()?;
30    Log::new(Some(&ctx.status_tx)).info(format!("BMP collector on {local}"));
31    let shared = Arc::new(ctx.clone());
32    let server_id = ctx.server_id;
33    let accept =
34        tokio::spawn(async move {
35            let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
36            loop {
37                let (stream, peer, permit) =
38                    match accept_bounded(&listener, &limiter, b"", "BMP", Some(&shared.status_tx))
39                        .await
40                    {
41                        Ok(v) => v,
42                        Err(_) => break,
43                    };
44                let id = ConnectionId::new(shared.state.get_next_unified_id().await);
45                let now = Instant::now();
46                shared
47                    .state
48                    .add_connection_to_server(
49                        server_id,
50                        ConnectionState {
51                            id,
52                            remote_addr: peer,
53                            local_addr: local,
54                            bytes_sent: 0,
55                            bytes_received: 0,
56                            packets_sent: 0,
57                            packets_received: 0,
58                            last_activity: now,
59                            status: ConnectionStatus::Active,
60                            status_changed_at: now,
61                            protocol_info: ProtocolConnectionInfo::empty(),
62                        },
63                    )
64                    .await;
65                let child = shared.clone();
66                shared
67                    .state
68                    .spawn_server_task(server_id, async move {
69                        let _permit = permit;
70                        if let Err(e) = connection(&child, id, peer, stream).await {
71                            Log::new(Some(&child.status_tx))
72                                .debug(format!("BMP connection {id}: {e:#}"));
73                        }
74                        child
75                            .state
76                            .update_connection_status(server_id, id, ConnectionStatus::Closed)
77                            .await;
78                        let _ = child.status_tx.send("__UPDATE_UI__".into());
79                    })
80                    .await;
81            }
82        });
83    ctx.state.register_server_task(server_id, accept).await;
84    Ok(local)
85}
86
87fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
88    let summary = format!("BMP connection {id} operation={operation} decision={decision}");
89    let log = Log::new(Some(&ctx.status_tx));
90    if decision.starts_with("fail_closed_") || decision == "model_silent" {
91        log.error(summary);
92    } else {
93        log.info(summary);
94    }
95}
96
97/// The handler's answer; None closes the session.
98async fn ask(ctx: &SpawnContext, id: ConnectionId, event: Event, operation: &str) -> Option<Json> {
99    let result = match call_llm(
100        &ctx.llm_client,
101        &ctx.state,
102        ctx.server_id,
103        Some(id),
104        &event,
105        &actions::BmpProtocol,
106    )
107    .await
108    {
109        Ok(r) => r,
110        Err(_) => {
111            outcome(ctx, id, operation, "fail_closed_llm_error");
112            return None;
113        }
114    };
115    let mut answers = Vec::new();
116    let mut pending = result.protocol_results;
117    while let Some(r) = pending.pop() {
118        match r {
119            ActionResult::Custom { data, .. } => answers.push(data),
120            ActionResult::Multiple(items) => pending.extend(items),
121            _ => {}
122        }
123    }
124    match (result.failures.is_empty(), answers.len()) {
125        (true, 1) => Some(answers.remove(0)),
126        (true, 0) => {
127            outcome(ctx, id, operation, "model_silent");
128            None
129        }
130        _ => {
131            outcome(ctx, id, operation, "fail_closed_invalid_reply");
132            None
133        }
134    }
135}
136
137/// Read one whole message: Ok(None) when the router closed between messages.
138async fn read_message(stream: &mut TcpStream, buf: &mut Vec<u8>) -> Result<Option<Vec<u8>>> {
139    let mut deadline: Option<tokio::time::Instant> = None;
140    loop {
141        if let Some(len) = codec::message_len(buf)? {
142            if buf.len() >= len {
143                return Ok(Some(buf.drain(..len).collect()));
144            }
145        }
146        if !buf.is_empty() && deadline.is_none() {
147            deadline = Some(tokio::time::Instant::now() + MESSAGE_DEADLINE);
148        }
149        let mut chunk = [0u8; 16 * 1024];
150        let n = match deadline {
151            Some(d) => tokio::time::timeout_at(d, stream.read(&mut chunk))
152                .await
153                .context("a BMP message did not finish within the deadline")??,
154            None => stream.read(&mut chunk).await?,
155        };
156        if n == 0 {
157            if buf.is_empty() {
158                return Ok(None);
159            }
160            bail!("the router closed the session mid-message");
161        }
162        buf.extend_from_slice(&chunk[..n]);
163    }
164}
165
166async fn connection(
167    ctx: &SpawnContext,
168    id: ConnectionId,
169    peer: SocketAddr,
170    mut stream: TcpStream,
171) -> Result<()> {
172    let mut buf = Vec::new();
173    let mut router = peer.ip().to_string();
174    let mut first = true;
175    loop {
176        let message = match read_message(&mut stream, &mut buf).await {
177            Ok(Some(m)) => m,
178            Ok(None) => return Ok(()),
179            Err(e) => {
180                outcome(ctx, id, "read", "protocol_refusal");
181                return Err(e);
182            }
183        };
184        ctx.state
185            .update_connection_stats(
186                ctx.server_id,
187                id,
188                Some(message.len() as u64),
189                None,
190                Some(1),
191                None,
192            )
193            .await;
194        let kind = message[5];
195        if first && kind != codec::INITIATION {
196            outcome(ctx, id, "read", "protocol_refusal");
197            bail!("the router's first message is type {kind}, not an Initiation");
198        }
199        first = false;
200        let (event_id, mut data) = match codec::describe(kind, &message[codec::HEADER_LEN..]) {
201            Ok(Some(d)) => d,
202            Ok(None) => {
203                Log::new(Some(&ctx.status_tx))
204                    .debug(format!("BMP connection {id}: skipped message type {kind}"));
205                continue;
206            }
207            Err(e) => {
208                outcome(ctx, id, "decode", "protocol_refusal");
209                return Err(e);
210            }
211        };
212        if kind == codec::INITIATION {
213            if let Some(name) = data["sys_name"].as_str().filter(|n| !n.is_empty()) {
214                router = name.to_owned();
215            }
216        } else {
217            data["router"] = json!(router);
218        }
219        let event_type: &'static crate::protocol::EventType = match kind {
220            codec::INITIATION => &actions::INITIATION_EVENT,
221            codec::PEER_UP => &actions::PEER_UP_EVENT,
222            codec::ROUTE_MONITORING => &actions::ROUTE_MONITORING_EVENT,
223            codec::STATISTICS => &actions::STATISTICS_EVENT,
224            codec::PEER_DOWN => &actions::PEER_DOWN_EVENT,
225            codec::TERMINATION => &actions::TERMINATION_EVENT,
226            _ => &actions::ROUTE_MIRRORING_EVENT,
227        };
228        let operation = event_id.trim_start_matches("bmp_");
229        let Some(answer) = ask(ctx, id, Event::new(event_type, data), operation).await else {
230            return Ok(());
231        };
232        if answer["type"] == "bmp_close" {
233            outcome(ctx, id, operation, "model_reject");
234            return Ok(());
235        }
236        outcome(ctx, id, operation, "model_answer");
237        if kind == codec::TERMINATION {
238            return Ok(());
239        }
240    }
241}
netget/server/bmp/mod.rs

netget/server/bmp/
mod.rs
