netget/server/knx/
mod.rs

1//! KNXnet/IP tunnelling gateway. Rust owns the frames, the tunnels (channels, sequence numbers,
2//! heartbeats), acknowledgements, confirmations and the routing of group telegrams between
3//! tunnels; the handler is every device on the bus, answering reads and reacting to writes.
4pub mod actions;
5pub mod wire;
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::connection::ConnectionId;
12use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
13use anyhow::{ensure, Context, Result};
14use serde_json::{json, Value};
15use std::collections::{BTreeMap, HashMap};
16use std::net::SocketAddr;
17use std::sync::Arc;
18use std::time::Duration;
19use tokio::net::UdpSocket;
20use tokio::sync::{mpsc, Mutex};
21use wire::{Apci, Data, Telegram};
22
23pub const DEFAULT_ADDRESS: &str = "1.1.250";
24pub const DEFAULT_MAX_TUNNELS: u64 = 8;
25pub const MAX_TUNNELS_CEILING: u64 = 64;
26/// A tunnel that sends nothing (not even a CONNECTIONSTATE heartbeat) for this long is gone.
27pub const TUNNEL_TIMEOUT: Duration = Duration::from_secs(120);
28/// Telegrams waiting for the handler at once; past this, telegrams are dropped (logged).
29pub const TELEGRAM_QUEUE: usize = 64;
30
31struct Tunnel {
32    control: SocketAddr,
33    data: SocketAddr,
34    address: u16,
35    /// The sequence number expected next from the client.
36    seq_in: u8,
37    /// The sequence number of the next request NetGet sends.
38    seq_out: u8,
39    last_seen: crate::utils::clock::Instant,
40    connection: ConnectionId,
41}
42
43struct Gateway {
44    socket: UdpSocket,
45    tunnels: Mutex<BTreeMap<u8, Tunnel>>,
46    address: u16,
47    types: HashMap<u16, String>,
48    max_tunnels: usize,
49    local: SocketAddr,
50}
51
52impl Gateway {
53    /// Send a telegram as L_Data.ind to every tunnel except `except`.
54    async fn broadcast(&self, telegram: &Telegram, except: Option<u8>) {
55        let cemi = wire::cemi(&Telegram {
56            message_code: wire::L_DATA_IND,
57            ..telegram.clone()
58        });
59        let mut sends = Vec::new();
60        {
61            let mut tunnels = self.tunnels.lock().await;
62            for (channel, t) in tunnels.iter_mut() {
63                if Some(*channel) == except {
64                    continue;
65                }
66                sends.push((t.data, wire::tunnelling(*channel, t.seq_out, &cemi)));
67                t.seq_out = t.seq_out.wrapping_add(1);
68            }
69        }
70        for (to, bytes) in sends {
71            let _ = self.socket.send_to(&bytes, to).await;
72        }
73    }
74
75    fn describe(&self, d: &Data, destination: u16) -> (Option<String>, Value, Value) {
76        let dpt = self.types.get(&destination).cloned();
77        let value = dpt
78            .as_deref()
79            .and_then(|t| wire::decode(t, d).ok())
80            .unwrap_or(Value::Null);
81        (dpt, value, wire::interpretations(d))
82    }
83}
84
85fn config(ctx: &SpawnContext) -> Result<(u16, HashMap<u16, String>, usize)> {
86    let params = ctx.startup_params.as_ref();
87    let address = params
88        .map(|p| p.get_optional_string("individual_address"))
89        .transpose()?
90        .flatten()
91        .unwrap_or_else(|| DEFAULT_ADDRESS.into());
92    let address = wire::parse_individual(&address)?;
93    let max = params
94        .map(|p| p.get_optional_u64("max_tunnels"))
95        .transpose()?
96        .flatten()
97        .unwrap_or(DEFAULT_MAX_TUNNELS);
98    ensure!(
99        (1..=MAX_TUNNELS_CEILING).contains(&max),
100        "max_tunnels must be 1..={MAX_TUNNELS_CEILING}"
101    );
102    let mut types = HashMap::new();
103    if let Some(map) = params
104        .map(|p| p.get_optional_object("group_types"))
105        .transpose()?
106        .flatten()
107    {
108        for (ga, dpt) in map {
109            let dpt = dpt.as_str().context("group_types values are DPT names")?;
110            wire::dpt_main(dpt)?;
111            types.insert(wire::parse_group(ga)?, dpt.to_string());
112        }
113    }
114    Ok((address, types, max as usize))
115}
116
117pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
118    let (address, types, max_tunnels) = config(&ctx)?;
119    let socket = UdpSocket::bind(ctx.legacy_listen_addr()).await?;
120    let local = socket.local_addr()?;
121    Log::new(Some(&ctx.status_tx)).info(format!(
122        "KNXnet/IP gateway {} listening on {local}",
123        wire::format_individual(address)
124    ));
125    let gw = Arc::new(Gateway {
126        socket,
127        tunnels: Mutex::new(BTreeMap::new()),
128        address,
129        types,
130        max_tunnels,
131        local,
132    });
133    let (tx, rx) = mpsc::channel::<(u8, Telegram)>(TELEGRAM_QUEUE);
134    let worker_ctx = ctx.clone();
135    let worker_gw = gw.clone();
136    ctx.state
137        .register_server_task(
138            ctx.server_id,
139            tokio::spawn(worker(worker_ctx, worker_gw, rx)),
140        )
141        .await;
142    let reaper_ctx = ctx.clone();
143    let reaper_gw = gw.clone();
144    ctx.state
145        .register_server_task(
146            ctx.server_id,
147            tokio::spawn(async move {
148                loop {
149                    tokio::time::sleep(Duration::from_secs(10)).await;
150                    let gone: Vec<(u8, ConnectionId, SocketAddr)> = {
151                        let mut tunnels = reaper_gw.tunnels.lock().await;
152                        let stale: Vec<u8> = tunnels
153                            .iter()
154                            .filter(|(_, t)| t.last_seen.elapsed() > TUNNEL_TIMEOUT)
155                            .map(|(c, _)| *c)
156                            .collect();
157                        stale
158                            .into_iter()
159                            .filter_map(|c| {
160                                tunnels.remove(&c).map(|t| (c, t.connection, t.control))
161                            })
162                            .collect()
163                    };
164                    for (channel, id, control) in gone {
165                        // The gateway closes a tunnel it has given up on, so the client knows.
166                        let mut body = vec![channel, 0];
167                        body.extend(wire::hpai(reaper_gw.local));
168                        let _ = reaper_gw
169                            .socket
170                            .send_to(&wire::frame(wire::DISCONNECT_REQUEST, &body), control)
171                            .await;
172                        Log::new(Some(&reaper_ctx.status_tx))
173                            .info(format!("KNX tunnel {channel} timed out"));
174                        reaper_ctx
175                            .state
176                            .update_connection_status(
177                                reaper_ctx.server_id,
178                                id,
179                                ConnectionStatus::Closed,
180                            )
181                            .await;
182                    }
183                }
184            }),
185        )
186        .await;
187    let recv_ctx = ctx.clone();
188    ctx.state
189        .register_server_task(
190            ctx.server_id,
191            tokio::spawn(async move {
192                let mut buf = vec![0u8; wire::MAX_FRAME + 1];
193                loop {
194                    let (n, from) = match gw.socket.recv_from(&mut buf).await {
195                        Ok(v) => v,
196                        Err(_) => continue,
197                    };
198                    if let Err(e) = handle(&recv_ctx, &gw, &tx, &buf[..n], from).await {
199                        Log::new(Some(&recv_ctx.status_tx))
200                            .debug(format!("KNX frame from {from} ignored: {e:#}"));
201                    }
202                }
203            }),
204        )
205        .await;
206    Ok(local)
207}
208
209fn device_info(gw: &Gateway) -> Vec<u8> {
210    // Device information DIB (54 bytes) then supported service families DIB.
211    let mut d = vec![54, 0x01, 0x02, 0x00];
212    d.extend(gw.address.to_be_bytes());
213    d.extend([0, 0]); // project installation id
214    d.extend([0x00, 0x00, 0x4e, 0x47, 0x00, 0x01]); // serial number
215    d.extend([0, 0, 0, 0]); // routing multicast address (unused)
216    d.extend([0x00, 0x00, 0x4e, 0x47, 0x00, 0x01]); // MAC
217    let mut name = b"NetGet KNX/IP".to_vec();
218    name.resize(30, 0);
219    d.extend(name);
220    d.extend([8, 0x02, 0x02, 0x01, 0x03, 0x01, 0x04, 0x01]); // core, device mgmt, tunnelling v1
221    d
222}
223
224async fn handle(
225    ctx: &SpawnContext,
226    gw: &Gateway,
227    tx: &mpsc::Sender<(u8, Telegram)>,
228    frame: &[u8],
229    from: SocketAddr,
230) -> Result<()> {
231    let (service, body) = wire::parse_frame(frame)?;
232    match service {
233        wire::SEARCH_REQUEST | wire::DESCRIPTION_REQUEST => {
234            let (hpai, _) = wire::parse_hpai(body)?;
235            let to = wire::endpoint(hpai, from);
236            let mut reply = Vec::new();
237            if service == wire::SEARCH_REQUEST {
238                reply.extend(wire::hpai(gw.local));
239            }
240            reply.extend(device_info(gw));
241            let kind = if service == wire::SEARCH_REQUEST {
242                wire::SEARCH_RESPONSE
243            } else {
244                wire::DESCRIPTION_RESPONSE
245            };
246            gw.socket.send_to(&wire::frame(kind, &reply), to).await?;
247        }
248        wire::CONNECT_REQUEST => {
249            let (control, n) = wire::parse_hpai(body)?;
250            let (data, m) = wire::parse_hpai(&body[n..])?;
251            let cri = body.get(n + m..).context("missing CRI")?;
252            let control = wire::endpoint(control, from);
253            let data = wire::endpoint(data, from);
254            let refuse = |status: u8| wire::frame(wire::CONNECT_RESPONSE, &[0, status]);
255            if cri.len() < 4 || cri[1] != 0x04 {
256                gw.socket
257                    .send_to(&refuse(wire::E_CONNECTION_TYPE), control)
258                    .await?;
259                return Ok(());
260            }
261            if cri[2] != 0x02 {
262                gw.socket
263                    .send_to(&refuse(wire::E_CONNECTION_OPTION), control)
264                    .await?;
265                return Ok(());
266            }
267            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
268            let mut tunnels = gw.tunnels.lock().await;
269            let channel = (1..=255u8).find(|c| !tunnels.contains_key(c));
270            let address = (1..=255u16)
271                .map(|d| (gw.address & 0xff00) | d)
272                .find(|a| *a != gw.address && tunnels.values().all(|t| t.address != *a));
273            let (Some(channel), Some(address), true) =
274                (channel, address, tunnels.len() < gw.max_tunnels)
275            else {
276                drop(tunnels);
277                gw.socket
278                    .send_to(&refuse(wire::E_NO_MORE_CONNECTIONS), control)
279                    .await?;
280                return Ok(());
281            };
282            let now = crate::utils::clock::Instant::now();
283            tunnels.insert(
284                channel,
285                Tunnel {
286                    control,
287                    data,
288                    address,
289                    seq_in: 0,
290                    seq_out: 0,
291                    last_seen: now,
292                    connection: id,
293                },
294            );
295            drop(tunnels);
296            ctx.state
297                .add_connection_to_server(
298                    ctx.server_id,
299                    ConnectionState {
300                        id,
301                        remote_addr: control,
302                        local_addr: gw.local,
303                        bytes_sent: 0,
304                        bytes_received: 0,
305                        packets_sent: 0,
306                        packets_received: 0,
307                        last_activity: now,
308                        status: ConnectionStatus::Active,
309                        status_changed_at: now,
310                        protocol_info: ProtocolConnectionInfo::new(json!({
311                            "channel": channel, "individual_address": wire::format_individual(address)
312                        })),
313                    },
314                )
315                .await;
316            let mut reply = vec![channel, wire::E_NO_ERROR];
317            reply.extend(wire::hpai(gw.local));
318            reply.extend([4, 0x04]);
319            reply.extend(address.to_be_bytes());
320            gw.socket
321                .send_to(&wire::frame(wire::CONNECT_RESPONSE, &reply), control)
322                .await?;
323            Log::new(Some(&ctx.status_tx)).info(format!(
324                "KNX tunnel {channel} opened for {control} as {}",
325                wire::format_individual(address)
326            ));
327        }
328        wire::CONNECTIONSTATE_REQUEST | wire::DISCONNECT_REQUEST => {
329            ensure!(body.len() >= 10, "short connection request");
330            let channel = body[0];
331            let (control, _) = wire::parse_hpai(&body[2..])?;
332            let control = wire::endpoint(control, from);
333            let (status, closed) = {
334                let mut tunnels = gw.tunnels.lock().await;
335                let known = tunnels.contains_key(&channel);
336                let status = if known {
337                    wire::E_NO_ERROR
338                } else {
339                    wire::E_CONNECTION_ID
340                };
341                if service == wire::CONNECTIONSTATE_REQUEST {
342                    if let Some(t) = tunnels.get_mut(&channel) {
343                        t.last_seen = crate::utils::clock::Instant::now();
344                    }
345                    (status, None)
346                } else {
347                    (status, tunnels.remove(&channel).map(|t| t.connection))
348                }
349            };
350            let reply = if service == wire::CONNECTIONSTATE_REQUEST {
351                wire::frame(wire::CONNECTIONSTATE_RESPONSE, &[channel, status])
352            } else {
353                wire::frame(wire::DISCONNECT_RESPONSE, &[channel, status])
354            };
355            gw.socket.send_to(&reply, control).await?;
356            if let Some(id) = closed {
357                ctx.state
358                    .update_connection_status(ctx.server_id, id, ConnectionStatus::Closed)
359                    .await;
360                Log::new(Some(&ctx.status_tx)).info(format!("KNX tunnel {channel} closed"));
361            }
362        }
363        wire::DISCONNECT_RESPONSE | wire::TUNNELLING_ACK => {
364            // Acks for NetGet's own requests: nothing is retransmitted, so nothing to clear.
365            if let Some(channel) = body.get(1) {
366                if let Some(t) = gw.tunnels.lock().await.get_mut(channel) {
367                    t.last_seen = crate::utils::clock::Instant::now();
368                }
369            }
370        }
371        wire::TUNNELLING_REQUEST => {
372            ensure!(
373                body.len() >= 4 && body[0] == 4,
374                "malformed connection header"
375            );
376            let (channel, seq) = (body[1], body[2]);
377            let (data_endpoint, fresh, address) = {
378                let mut tunnels = gw.tunnels.lock().await;
379                let Some(t) = tunnels.get_mut(&channel) else {
380                    return Ok(());
381                };
382                t.last_seen = crate::utils::clock::Instant::now();
383                let fresh = seq == t.seq_in;
384                // A repeat of the last request is acknowledged again and not processed; any
385                // other sequence number is dropped unanswered, as the specification says.
386                if !fresh && seq != t.seq_in.wrapping_sub(1) {
387                    return Ok(());
388                }
389                if fresh {
390                    t.seq_in = t.seq_in.wrapping_add(1);
391                }
392                (t.data, fresh, t.address)
393            };
394            let parsed = wire::parse_cemi(&body[4..]);
395            let status = if parsed.is_ok() {
396                wire::E_NO_ERROR
397            } else {
398                0x29
399            };
400            gw.socket
401                .send_to(&wire::tunnelling_ack(channel, seq, status), data_endpoint)
402                .await?;
403            let Ok(Some(mut telegram)) = parsed else {
404                return Ok(());
405            };
406            if !fresh || telegram.message_code != wire::L_DATA_REQ {
407                return Ok(());
408            }
409            // The tunnel's own address is the source of what it sends.
410            telegram.source = address;
411            let con = Telegram {
412                message_code: wire::L_DATA_CON,
413                ..telegram.clone()
414            };
415            let bytes = {
416                let mut tunnels = gw.tunnels.lock().await;
417                let Some(t) = tunnels.get_mut(&channel) else {
418                    return Ok(());
419                };
420                let b = wire::tunnelling(channel, t.seq_out, &wire::cemi(&con));
421                t.seq_out = t.seq_out.wrapping_add(1);
422                b
423            };
424            gw.socket.send_to(&bytes, data_endpoint).await?;
425            // On the bus, every other tunnel hears it.
426            gw.broadcast(&telegram, Some(channel)).await;
427            if tx.try_send((channel, telegram)).is_err() {
428                Log::new(Some(&ctx.status_tx))
429                    .warn("KNX telegram queue full: telegram not handed to the handler");
430            }
431        }
432        other => {
433            Log::new(Some(&ctx.status_tx))
434                .debug(format!("KNX service {other:#06x} from {from} not handled"));
435        }
436    }
437    Ok(())
438}
439
440fn outcome(ctx: &SpawnContext, op: &str, decision: &str) {
441    let line = format!("KNX operation={op} decision={decision}");
442    let log = Log::new(Some(&ctx.status_tx));
443    if decision.starts_with("fail_closed") {
444        log.error(line)
445    } else {
446        log.info(line)
447    }
448}
449
450/// The handler, one telegram at a time.
451async fn worker(ctx: SpawnContext, gw: Arc<Gateway>, mut rx: mpsc::Receiver<(u8, Telegram)>) {
452    while let Some((channel, telegram)) = rx.recv().await {
453        let connection = gw.tunnels.lock().await.get(&channel).map(|t| t.connection);
454        let source = wire::format_individual(telegram.source);
455        let destination = wire::format_group(telegram.destination);
456        let (dpt, value, interpretations) = gw.describe(&telegram.data, telegram.destination);
457        let (event, op) = match telegram.apci {
458            Apci::Read => (
459                Event::new(
460                    &actions::READ_EVENT,
461                    json!({"source": source, "destination": destination, "dpt": dpt}),
462                ),
463                "group_read",
464            ),
465            kind => (
466                Event::new(
467                    &actions::TELEGRAM_EVENT,
468                    json!({"kind": kind.name(), "source": source, "destination": destination,
469                           "dpt": dpt, "value": value, "interpretations": interpretations}),
470                ),
471                "group_telegram",
472            ),
473        };
474        let result = match call_llm(
475            &ctx.llm_client,
476            &ctx.state,
477            ctx.server_id,
478            connection,
479            &event,
480            &actions::KnxProtocol,
481        )
482        .await
483        {
484            Ok(r) if r.failures.is_empty() => r,
485            Ok(_) => {
486                outcome(&ctx, op, "fail_closed_invalid_reply");
487                continue;
488            }
489            Err(_) => {
490                outcome(&ctx, op, "fail_closed_llm_error");
491                continue;
492            }
493        };
494        let mut answers = Vec::new();
495        let mut stack = result.protocol_results;
496        while let Some(r) = stack.pop() {
497            match r {
498                ActionResult::Custom { name, data } if name.starts_with("knx_") => {
499                    answers.push((name, data))
500                }
501                ActionResult::Multiple(items) => stack.extend(items),
502                _ => {}
503            }
504        }
505        answers.reverse();
506        if answers.is_empty() {
507            outcome(&ctx, op, "model_silent");
508        }
509        for (name, data) in answers {
510            let sent = match name.as_str() {
511                "knx_group_response" if telegram.apci == Apci::Read => {
512                    actions::value_of(&data, dpt.as_deref()).map(|d| Telegram {
513                        message_code: wire::L_DATA_IND,
514                        source: gw.address,
515                        destination: telegram.destination,
516                        apci: Apci::Response,
517                        data: d,
518                    })
519                }
520                "knx_group_write" => (|| {
521                    let ga = wire::parse_group(data["group_address"].as_str().unwrap_or_default())?;
522                    let d = actions::value_of(&data, gw.types.get(&ga).map(String::as_str))?;
523                    Ok(Telegram {
524                        message_code: wire::L_DATA_IND,
525                        source: gw.address,
526                        destination: ga,
527                        apci: Apci::Write,
528                        data: d,
529                    })
530                })(),
531                "knx_ignore" => {
532                    outcome(&ctx, op, "model_ignore");
533                    continue;
534                }
535                _ => continue,
536            };
537            match sent {
538                Ok(t) => {
539                    outcome(&ctx, op, "model_answer");
540                    gw.broadcast(&t, None).await;
541                }
542                Err(e) => {
543                    Log::new(Some(&ctx.status_tx)).error(format!("KNX answer refused: {e:#}"));
544                    outcome(&ctx, op, "fail_closed_invalid_reply");
545                }
546            }
547        }
548    }
549}
netget/server/knx/mod.rs

netget/server/knx/
mod.rs
