bacnet-network 0.12.0

use std::collections::HashSet;
use std::sync::Arc;
use std::time::{Duration, Instant};

use bacnet_encoding::npdu::{encode_npdu, Npdu, NpduAddress};
use bacnet_types::enums::{NetworkMessageType, RejectMessageReason};
use bacnet_types::MacAddr;
use bytes::{BufMut, Bytes, BytesMut};
use tokio::sync::{mpsc, Mutex};
use tracing::{debug, warn};

use crate::router_table::RouterTable;

use super::control_policy::{ControlClass, ControlGate};
use super::local_control::LocalControl;
use super::reject::{route_received_reject, send_reject, Refused};
use super::{IngressContext, SendRequest};

/// One validated Initialize-Routing-Table / Initialize-Routing-Table-Ack
/// entry: DNET(2) + Port ID(1) + Port Info Length(1) + Port Info(N).
/// Port Info octets are envelope-checked but not retained: this router holds
/// no PTP/modem dial information (135-2020 6.4.7 gives the optional Port Info
/// no structure beyond an octet string; it serves dial-up PTP links, which
/// this router does not implement).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct RoutingTableEntry {
    network: u16,
    port_id: u8,
}

/// Validate an Initialize-Routing-Table / Initialize-Routing-Table-Ack data
/// portion without touching the table (RB-03).
///
/// Returns the entries in order when the payload is exactly `Number of Ports`
/// entries — each `DNET(2) + Port ID(1) + Port Info Length(1) + Port Info(N)`
/// — with no truncation and no trailing bytes. Returns `None` for any
/// incomplete tail, including a missing `Number of Ports` octet. Per-entry
/// policy (reserved networks, existing routes, route cap) stays in the apply
/// phase; this only proves the envelope is complete.
fn parse_routing_table_entries(data: &[u8]) -> Option<Vec<RoutingTableEntry>> {
    let count = usize::from(*data.first()?);
    let mut offset = 1usize;
    let mut entries = Vec::with_capacity(count);
    for _ in 0..count {
        if offset + 4 > data.len() {
            return None;
        }
        let network = u16::from_be_bytes([data[offset], data[offset + 1]]);
        let port_id = data[offset + 2];
        let info_len = usize::from(data[offset + 3]);
        if offset + 4 + info_len > data.len() {
            return None;
        }
        entries.push(RoutingTableEntry { network, port_id });
        offset += 4 + info_len;
    }
    if offset != data.len() {
        return None;
    }
    Some(entries)
}

/// Bound for an encoded Initialize-Routing-Table-Ack reply NPDU (RB-05).
///
/// Query replies must fit the smallest standard link so the answer is
/// receivable anywhere: an MS/TP frame sent without COBS encoding may carry
/// at most 501 Data octets (135-2020 9.3), and the NPDU envelope rides inside
/// that Data field. This octet bound and the 1-octet entry-count field (6.4.7,
/// Fig. 6-11) apply independently when chunking a complete table.
const MAX_ROUTER_CONTROL_REPLY_NPDU: usize = 501;
/// Encoded NPDU overhead of a link-local Init-Ack: version + control +
/// message type (no DNET/SNET, hence no hop count; type < 0x80, no vendor ID).
const INIT_ACK_NPDU_OVERHEAD: usize = 3;
/// Encoded octets per query entry: DNET(2) + Port ID(1) + Port Info Length(1)
/// with empty Port Info (no PTP/modem info held — see [`RoutingTableEntry`]).
const INIT_ACK_ENTRY_OCTETS: usize = 4;
/// Entries per query acknowledgment, from the octet bound above (124); the
/// `assert` pins both independent limits — never more than the 255 the count
/// field can name, never an encoded reply over the egress bound.
const MAX_INIT_ACK_ENTRIES: usize =
    (MAX_ROUTER_CONTROL_REPLY_NPDU - INIT_ACK_NPDU_OVERHEAD - 1) / INIT_ACK_ENTRY_OCTETS;
const _: () = assert!(
    MAX_INIT_ACK_ENTRIES >= 1
        && MAX_INIT_ACK_ENTRIES <= 255
        && INIT_ACK_NPDU_OVERHEAD + 1 + MAX_INIT_ACK_ENTRIES * INIT_ACK_ENTRY_OCTETS
            <= MAX_ROUTER_CONTROL_REPLY_NPDU
);

/// Handle a network-layer message.
///
/// The caller ([`super::dispatch::dispatch_network_message`]) has already
/// applied the routing-first rule: only link-local messages (no DNET or global
/// broadcast), messages for our own ingress network, never-routed controls
/// (What-Is / Network-Number-Is), and rejects reach this admission point.
///
/// Every learning path below validates its complete payload BEFORE acquiring
/// the table lock (RB-03, Clauses 6.2 / 6.4 / 6.5.4 / 6.6.3.2). A truncated
/// tail rejects the whole message: no table change, no partial forward, no
/// rebroadcast, and — for Initialize-Routing-Table — no ACK.
///
/// RB-09: protected controls authorize post-validation, pre-lock, lock-free
/// via `control`. Deny is a silent drop: no mutation, relay, ACK, or Reject.
///
/// `local` is the router's own network-control consumer, which hears about
/// a reject addressed to this router (#1175).
pub(super) async fn handle_network_message(
    table: &Arc<Mutex<RouterTable>>,
    send_txs: &[mpsc::Sender<SendRequest>],
    ctx: &IngressContext,
    control: &ControlGate,
    local: &LocalControl,
) {
    const MAX_LEARNED_ROUTES: usize = 256;

    let port_idx = ctx.port_idx;
    let port_network = ctx.port_network;
    let source_mac = ctx.source_mac.as_slice();
    let npdu: &Npdu = &ctx.npdu;
    let ingress_attributes = ctx.data_attributes.as_slice();

    let msg_type = match npdu.message_type {
        Some(t) => t,
        None => return,
    };

    if msg_type == NetworkMessageType::WHO_IS_ROUTER_TO_NETWORK.to_raw() {
        // Clause 6.4.1: the payload is empty or a single 16-bit network.
        // Anything else (a 1-octet tail, trailing bytes) is malformed and
        // must not promote a scoped query into a global scan.
        let requested_network = match npdu.payload.len() {
            0 => None,
            2 => {
                let net = u16::from_be_bytes([npdu.payload[0], npdu.payload[1]]);
                if net == 0 {
                    return;
                }
                Some(net)
            }
            _ => return,
        };

        let table = table.lock().await;

        let networks: Vec<u16> = if let Some(net) = requested_network {
            // Only respond if the network is reachable via a different port.
            match table.lookup(net) {
                Some(entry) if entry.port_index != port_idx => vec![net],
                _ => {
                    // Unknown: forward Who-Is-Router to all other ports to discover the path.
                    drop(table);
                    // Clause 6.6.3.2 discovery relay: retain an already-routed
                    // SNET/SADR, else add the ingress network and the
                    // immediate source MAC (the originator is local). The
                    // immediate next hop stays implicit (our broadcast SA),
                    // never merged into the origin fields. Discovery relays
                    // are local broadcasts: no DNET travels with them.
                    let source = npdu.source.clone().unwrap_or(NpduAddress {
                        network: port_network,
                        mac_address: MacAddr::from_slice(source_mac),
                    });
                    let forward = Npdu {
                        is_network_message: true,
                        expecting_reply: npdu.expecting_reply,
                        priority: npdu.priority,
                        source: Some(source),
                        message_type: Some(NetworkMessageType::WHO_IS_ROUTER_TO_NETWORK.to_raw()),
                        payload: npdu.payload.clone(),
                        ..Npdu::default()
                    };
                    let mut fwd_buf = BytesMut::with_capacity(8);
                    if let Ok(()) = encode_npdu(&mut fwd_buf, &forward) {
                        let frozen = fwd_buf.freeze();
                        for (i, tx) in send_txs.iter().enumerate() {
                            if i != port_idx {
                                let _ = tx.try_send(SendRequest::broadcast_with_attributes(
                                    frozen.clone(),
                                    ingress_attributes,
                                ));
                            }
                        }
                    }
                    return;
                }
            }
        } else {
            table.networks_not_on_port(port_idx)
        };

        if networks.is_empty() {
            return;
        }

        let mut payload = BytesMut::with_capacity(networks.len() * 2);
        for net in &networks {
            payload.put_u16(*net);
        }

        let payload_len = payload.len();
        let response = Npdu {
            is_network_message: true,
            message_type: Some(NetworkMessageType::I_AM_ROUTER_TO_NETWORK.to_raw()),
            payload: payload.freeze(),
            ..Npdu::default()
        };

        let mut buf = BytesMut::with_capacity(4 + payload_len);
        if let Err(e) = encode_npdu(&mut buf, &response) {
            warn!("Failed to encode I-Am-Router response NPDU: {e}");
            return;
        }

        // I-Am-Router-To-Network is always broadcast.
        if let Err(e) = send_txs[port_idx].try_send(SendRequest::broadcast_with_attributes(
            buf.freeze(),
            ingress_attributes,
        )) {
            warn!(%e, "Router dropped I-Am-Router response: output channel full");
        }
    } else if msg_type == NetworkMessageType::I_AM_ROUTER_TO_NETWORK.to_raw() {
        // Clause 6.4.2: a non-empty list of 16-bit networks. A truncated
        // tail rejects the whole message — no prefix learning, no rebroadcast.
        let data = &npdu.payload;
        if data.is_empty() || !data.len().is_multiple_of(2) {
            return;
        }
        // RB-09 protected: authorize lock-free; deny drops silently.
        {
            let targets: Vec<u16> = data
                .as_chunks::<2>()
                .0
                .iter()
                .map(|c| u16::from_be_bytes(*c))
                .collect();
            if !control.authorize(ctx, ControlClass::IAm, &targets, false) {
                return;
            }
        }

        let mut table = table.lock().await;

        let mut replacement_networks = HashSet::new();
        let mut offset = 0;
        while offset + 2 <= data.len() {
            let net = u16::from_be_bytes([data[offset], data[offset + 1]]);
            offset += 2;

            if table.len() >= MAX_LEARNED_ROUTES && table.lookup(net).is_none() {
                table.record_learning_cap();
                warn!("Router table learned routes cap ({MAX_LEARNED_ROUTES}) reached, ignoring further networks");
                break;
            }

            // Repeated entries in one packet are not independent arrivals.
            // Leave absent learning and same-port refresh entries unchanged.
            if table
                .lookup(net)
                .is_some_and(|entry| !entry.directly_connected && entry.port_index != port_idx)
                && !replacement_networks.insert(net)
            {
                continue;
            }
            if table.apply_learning_claim(
                net,
                port_idx,
                MacAddr::from_slice(source_mac),
                Instant::now(),
            ) {
                table.record_learned();
                debug!(
                    network = net,
                    port = port_idx,
                    "Learned route from I-Am-Router-To-Network"
                );
            }
        }
        drop(table);

        // Re-broadcast to all other ports unconditionally (spec 6.6.3.3).
        // Only reached for complete payloads; the payload is re-emitted
        // verbatim with the ingress attributes carried (Clause 6.5.4).
        let rebroadcast = Npdu {
            is_network_message: true,
            message_type: Some(NetworkMessageType::I_AM_ROUTER_TO_NETWORK.to_raw()),
            payload: npdu.payload.clone(),
            ..Npdu::default()
        };
        let mut buf = BytesMut::with_capacity(4 + npdu.payload.len());
        if let Ok(()) = encode_npdu(&mut buf, &rebroadcast) {
            let frozen = buf.freeze();
            for (i, tx) in send_txs.iter().enumerate() {
                if i != port_idx {
                    let _ = tx.try_send(SendRequest::broadcast_with_attributes(
                        frozen.clone(),
                        ingress_attributes,
                    ));
                }
            }
        }
    } else if msg_type == NetworkMessageType::REJECT_MESSAGE_TO_NETWORK.to_raw() {
        // Clause 6.4.4: exactly one reason octet plus one 2-octet network.
        // Anything else is malformed: no table change, no relay.
        if npdu.payload.len() != 3 {
            return;
        }
        let reason = npdu.payload[0];
        let rejected_net = u16::from_be_bytes([npdu.payload[1], npdu.payload[2]]);
        if !control.authorize(ctx, ControlClass::Reject, &[rejected_net], false) {
            return;
        }
        warn!(
            network = rejected_net,
            reason = reason,
            "Received Reject-Message-To-Network"
        );
        {
            let mut tbl = table.lock().await;
            tbl.apply_reject(rejected_net, port_idx, reason, Instant::now());
        }

        // Hand it to this router's own consumer when it is addressed here
        // (#1175), otherwise pass it on toward the node its DNET/DADR names
        // (#1158).
        route_received_reject(table, send_txs, local, ctx).await;
    } else if msg_type == NetworkMessageType::ROUTER_BUSY_TO_NETWORK.to_raw() {
        // Clauses 6.4.5/6.6.3.6: an optional list of 2-octet networks. When
        // the list is absent, the router is asking that traffic be held back
        // for every network it ordinarily serves. We read that omitted scope
        // as the via-peer set (learned routes egressing
        // this ingress port toward the immediate source MAC), never every
        // route and never none. Explicit lists intersect with that set, and
        // directly-attached paths are never overridden (see
        // `RouterTable::is_served_via_peer`). A truncated tail rejects the
        // whole message: no marks, no rebroadcast. Marks apply immediately
        // with a 30s deadline (no reject-style damping); learning freshness
        // (`last_seen`) is untouched. Propagation rebroadcasts the payload
        // verbatim out every other port; full queues drop locally (bounded,
        // no retry, no admission path) while per-route marks stay correct.
        let data = &npdu.payload;
        if !data.len().is_multiple_of(2) {
            return;
        }
        let mut listed = Vec::with_capacity(data.len() / 2);
        let mut offset = 0;
        while offset + 2 <= data.len() {
            listed.push(u16::from_be_bytes([data[offset], data[offset + 1]]));
            offset += 2;
        }
        if !control.authorize(ctx, ControlClass::Busy, &listed, listed.is_empty()) {
            return;
        }
        let peer = MacAddr::from_slice(source_mac);
        let deadline = Instant::now() + Duration::from_secs(30);
        {
            let mut tbl = table.lock().await;
            if listed.is_empty() {
                for net in tbl.routes_served_via_peer(port_idx, &peer) {
                    tbl.mark_busy(net, deadline);
                    debug!(
                        network = net,
                        "Router busy — marked network as congested (30s timer)"
                    );
                }
            } else {
                for net in listed {
                    if tbl.is_served_via_peer(net, port_idx, &peer) {
                        tbl.mark_busy(net, deadline);
                        debug!(
                            network = net,
                            "Router busy — marked network as congested (30s timer)"
                        );
                    }
                }
            }
        }
        // Re-broadcast to all other ports (spec 6.6.3.6)
        let rebroadcast = Npdu {
            is_network_message: true,
            message_type: Some(NetworkMessageType::ROUTER_BUSY_TO_NETWORK.to_raw()),
            payload: npdu.payload.clone(),
            ..Npdu::default()
        };
        let mut buf = BytesMut::with_capacity(4 + npdu.payload.len());
        if let Ok(()) = encode_npdu(&mut buf, &rebroadcast) {
            let frozen = buf.freeze();
            for (i, tx) in send_txs.iter().enumerate() {
                if i != port_idx {
                    let _ = tx.try_send(SendRequest::broadcast_with_attributes(
                        frozen.clone(),
                        ingress_attributes,
                    ));
                }
            }
        }
    } else if msg_type == NetworkMessageType::ROUTER_AVAILABLE_TO_NETWORK.to_raw() {
        // Clauses 6.4.6/6.6.3.7: same scope rule as Router-Busy. A missing
        // list means the router is lifting its hold on traffic toward every
        // network it serves. As with Busy, the omitted scope is the via-peer
        // set, and explicit lists intersect with it.
        let data = &npdu.payload;
        if !data.len().is_multiple_of(2) {
            return;
        }
        let mut listed = Vec::with_capacity(data.len() / 2);
        let mut offset = 0;
        while offset + 2 <= data.len() {
            listed.push(u16::from_be_bytes([data[offset], data[offset + 1]]));
            offset += 2;
        }
        if !control.authorize(ctx, ControlClass::Available, &listed, listed.is_empty()) {
            return;
        }
        let peer = MacAddr::from_slice(source_mac);
        {
            let mut tbl = table.lock().await;
            if listed.is_empty() {
                for net in tbl.routes_served_via_peer(port_idx, &peer) {
                    tbl.mark_available(net);
                    debug!(network = net, "Router available — cleared congestion");
                }
            } else {
                for net in listed {
                    if tbl.is_served_via_peer(net, port_idx, &peer) {
                        tbl.mark_available(net);
                        debug!(network = net, "Router available — cleared congestion");
                    }
                }
            }
        }
        // Re-broadcast to all other ports (spec 6.6.3.7)
        let rebroadcast = Npdu {
            is_network_message: true,
            message_type: Some(NetworkMessageType::ROUTER_AVAILABLE_TO_NETWORK.to_raw()),
            payload: npdu.payload.clone(),
            ..Npdu::default()
        };
        let mut buf = BytesMut::with_capacity(4 + npdu.payload.len());
        if let Ok(()) = encode_npdu(&mut buf, &rebroadcast) {
            let frozen = buf.freeze();
            for (i, tx) in send_txs.iter().enumerate() {
                if i != port_idx {
                    let _ = tx.try_send(SendRequest::broadcast_with_attributes(
                        frozen.clone(),
                        ingress_attributes,
                    ));
                }
            }
        }
    } else if msg_type == NetworkMessageType::INITIALIZE_ROUTING_TABLE.to_raw() {
        // Clauses 6.4.7/6.6.3.8: Number of Ports plus exactly that many
        // entries, fully validated before the table lock. A truncated entry
        // or trailing bytes rejects the whole message: no table change and
        // no ACK. A zero count with no trailing bytes is a table query.
        let data = &npdu.payload;
        let Some(entries) = parse_routing_table_entries(data) else {
            return;
        };

        if data[0] == 0 {
            // Query: the receiving router answers with its entire routing
            // table in Initialize-Routing-Table-Ack form and leaves that table
            // as it was (6.4.7). A table too large for one ACK is split across
            // several ACKs, each carrying the next slice, until all entries
            // have gone out (6.6.3.9). Entries ascend by DNET for determinism; every
            // ACK is bounded by the egress-octet limit and the count field
            // independently. A full queue drops the remaining ACKs (bounded,
            // no retry); the table itself is never mutated here. This arm
            // never consults the RB-04 via-peer selector.
            let snapshot: Vec<(u16, u8)> = {
                let tbl = table.lock().await;
                let mut snapshot = Vec::with_capacity(tbl.len());
                for net in tbl.sorted_networks() {
                    let port_id = tbl
                        .lookup(net)
                        .and_then(|entry| RouterTable::wire_port_id(entry.port_index));
                    match port_id {
                        Some(port_id) => snapshot.push((net, port_id)),
                        None => warn!(
                            network = net,
                            "Init-Routing-Table query: route has no wire Port ID, skipped"
                        ),
                    }
                }
                snapshot
            };
            // An empty table is still complete in one count-zero ACK; a
            // count-zero ACK is not the same wire value as an update ACK
            // (which carries no data at all — see below and 6.4.8).
            let chunk_len = MAX_INIT_ACK_ENTRIES.clamp(1, 255);
            let mut chunks: Vec<&[(u16, u8)]> = snapshot.chunks(chunk_len).collect();
            if chunks.is_empty() {
                chunks.push(&[]);
            }
            for chunk in chunks {
                let mut payload = BytesMut::with_capacity(1 + chunk.len() * INIT_ACK_ENTRY_OCTETS);
                payload.put_u8(chunk.len() as u8);
                for (net, port_id) in chunk {
                    payload.put_u16(*net);
                    payload.put_u8(*port_id);
                    payload.put_u8(0); // Port Info Length: none held.
                }
                let response = Npdu {
                    is_network_message: true,
                    message_type: Some(NetworkMessageType::INITIALIZE_ROUTING_TABLE_ACK.to_raw()),
                    payload: payload.freeze(),
                    ..Npdu::default()
                };
                let mut buf = BytesMut::with_capacity(
                    INIT_ACK_NPDU_OVERHEAD + 1 + chunk.len() * INIT_ACK_ENTRY_OCTETS,
                );
                if let Err(e) = encode_npdu(&mut buf, &response) {
                    warn!("Failed to encode Init-Routing-Table-ACK NPDU: {e}");
                    return;
                }
                debug_assert!(buf.len() <= MAX_ROUTER_CONTROL_REPLY_NPDU);
                if let Err(e) = send_txs[port_idx].try_send(SendRequest::unicast_with_attributes(
                    buf.freeze(),
                    MacAddr::from_slice(source_mac),
                    ingress_attributes,
                )) {
                    warn!(%e, "Router dropped Init-Routing-Table-ACK: output channel full");
                    return;
                }
            }
            return;
        }

        // Update: each network listed in the NPDU has its port mapping rewritten
        // from the entry supplied for it, then the sender is answered with an
        // Initialize-Routing-Table-Ack whose data portion is empty, not even a
        // count octet (6.6.3.8). Entries
        // apply in wire order (last wins); an unknown wire Port ID names no
        // local port, so that entry is skipped while the rest still apply.
        // RB-09 protected: deny drops silently with no empty ACK. Direct
        // immunity stays safety scoping, not auth. No RB-04 via-peer check.
        {
            let targets: Vec<u16> = entries.iter().map(|e| e.network).collect();
            if !control.authorize(ctx, ControlClass::InitMgmt, &targets, false) {
                return;
            }
        }
        let port_count = send_txs.len();
        {
            let mut tbl = table.lock().await;
            for entry in &entries {
                // Reserved DNETs name no routable network: skip before the
                // route-cap check so they consume neither cap nor suffix.
                if entry.network == 0 || entry.network == 0xFFFF {
                    continue;
                }
                // Port ID 0 purges: every row held for that DNET is deleted
                // (6.4.7), here scoped to learned entries by direct-route
                // safety (see the apply helper).
                if entry.port_id == 0 {
                    if tbl.apply_management_removal(entry.network) {
                        debug!(
                            network = entry.network,
                            "Init-Routing-Table purged learned route"
                        );
                    }
                    continue;
                }
                let Some(mapped) = RouterTable::port_index_for_wire_id(entry.port_id, port_count)
                else {
                    warn!(
                        network = entry.network,
                        port_id = entry.port_id,
                        "Init-Routing-Table: unknown Port ID, entry skipped"
                    );
                    continue;
                };
                if tbl.len() >= MAX_LEARNED_ROUTES && tbl.lookup(entry.network).is_none() {
                    tbl.record_learning_cap();
                    warn!("Init-Routing-Table: route cap reached, ignoring further entries");
                    break;
                }
                // Non-zero Port ID replaces or appends: the supplied route
                // overwrites whatever row already exists for that DNET, or is
                // added as a new row when there is none (6.4.7).
                if tbl.apply_management_update(
                    entry.network,
                    mapped,
                    MacAddr::from_slice(source_mac),
                ) {
                    debug!(
                        network = entry.network,
                        port = mapped,
                        "Applied route from Init-Routing-Table management update"
                    );
                }
            }
        }

        // Update ACK carries no data at all: 6.6.3.9 answers an update with a
        // data-less Initialize-Routing-Table-Ack, and 6.4.8 reserves the data
        // portion for replies to a table query. An empty payload is not
        // interchangeable with a zero count octet.
        let response = Npdu {
            is_network_message: true,
            message_type: Some(NetworkMessageType::INITIALIZE_ROUTING_TABLE_ACK.to_raw()),
            payload: Bytes::new(),
            ..Npdu::default()
        };

        let mut buf = BytesMut::with_capacity(8);
        if let Err(e) = encode_npdu(&mut buf, &response) {
            warn!("Failed to encode Init-Routing-Table-ACK NPDU: {e}");
            return;
        }

        if let Err(e) = send_txs[port_idx].try_send(SendRequest::unicast_with_attributes(
            buf.freeze(),
            MacAddr::from_slice(source_mac),
            ingress_attributes,
        )) {
            warn!(%e, "Router dropped Init-Routing-Table-ACK: output channel full");
        }
    } else if msg_type == NetworkMessageType::I_COULD_BE_ROUTER_TO_NETWORK.to_raw() {
        // Clause 6.4.3 / Figure 6-10: exactly DNET(2) + Performance Index(1).
        if npdu.payload.len() != 3 {
            return;
        }
        let net = u16::from_be_bytes([npdu.payload[0], npdu.payload[1]]);
        if !control.authorize(ctx, ControlClass::ICouldBe, &[net], false) {
            return;
        }
        let performance_index = npdu.payload[2];
        debug!(
            network = net,
            performance_index = performance_index,
            port = port_idx,
            "Received I-Could-Be-Router-To-Network"
        );
        // Store only if no existing route (lower priority than direct/learned).
        let mut tbl = table.lock().await;
        if tbl.lookup(net).is_none() {
            tbl.add_learned(net, port_idx, MacAddr::from_slice(source_mac));
            debug!(
                network = net,
                port = port_idx,
                "Stored potential route from I-Could-Be-Router-To-Network"
            );
        }
    } else if msg_type == NetworkMessageType::ESTABLISH_CONNECTION_TO_NETWORK.to_raw() {
        // Clause 6.4.9 / Figure 6-6: exactly DNET(2) + Termination Time(1).
        if npdu.payload.len() != 3 {
            return;
        }
        let net = u16::from_be_bytes([npdu.payload[0], npdu.payload[1]]);
        let termination_time_min = npdu.payload[2];
        tracing::info!(
            network = net,
            termination_time_minutes = termination_time_min,
            "Received Establish-Connection-To-Network (PTP not implemented)"
        );
    } else if msg_type == NetworkMessageType::DISCONNECT_CONNECTION_TO_NETWORK.to_raw() {
        // Clause 6.4.10: exactly one 2-octet network number.
        if npdu.payload.len() != 2 {
            return;
        }
        let net = u16::from_be_bytes([npdu.payload[0], npdu.payload[1]]);
        debug!(
            network = net,
            "Received Disconnect-Connection-To-Network (PTP not implemented; route removal ignored)"
        );
        let mut tbl = table.lock().await;
        tbl.record_disconnect_removal_ignored();
    } else if msg_type == NetworkMessageType::WHAT_IS_NETWORK_NUMBER.to_raw() {
        // Clause 6.4.19: never routed (enforced by dispatch); ignore when
        // SNET/SADR or DNET/DADR is present. X'12 defines no data octets, so
        // a non-empty payload is malformed. A link unicast or broadcast
        // What-Is may both be answered.
        if npdu.source.is_some() || npdu.destination.is_some() {
            return;
        }
        if !npdu.payload.is_empty() {
            return;
        }
        let mut payload = BytesMut::with_capacity(3);
        payload.put_u16(port_network);
        payload.put_u8(1); // configured

        let response = Npdu {
            is_network_message: true,
            message_type: Some(NetworkMessageType::NETWORK_NUMBER_IS.to_raw()),
            payload: payload.freeze(),
            ..Npdu::default()
        };

        let mut buf = BytesMut::with_capacity(8);
        if let Err(e) = encode_npdu(&mut buf, &response) {
            warn!("Failed to encode Network-Number-Is NPDU: {e}");
            return;
        }

        if let Err(e) = send_txs[port_idx].try_send(SendRequest::broadcast_with_attributes(
            buf.freeze(),
            ingress_attributes,
        )) {
            warn!(%e, "Router dropped Network-Number-Is: output channel full");
        }
    } else if msg_type == NetworkMessageType::NETWORK_NUMBER_IS.to_raw() {
        // Clause 6.4.20: never routed (enforced by dispatch); ignore messages
        // that carry SNET/SADR or DNET/DADR, that arrive via link unicast, or
        // whose payload is not exactly network(2) + flag(1).
        if npdu.source.is_some() || npdu.destination.is_some() {
            return;
        }
        if !ctx.link_layer_group {
            return;
        }
        if npdu.payload.len() != 3 {
            return;
        }
        let net = u16::from_be_bytes([npdu.payload[0], npdu.payload[1]]);
        let configured = npdu.payload[2];
        if net != port_network {
            // Clause 6.4.20 recommends reporting a peer's configured number
            // that conflicts with this port's configured one.
            if configured == 1 {
                warn!(
                    local_network = port_network,
                    peer_network = net,
                    "Network number conflict: port configured as {} but peer reports {} (configured)",
                    port_network, net
                );
            } else {
                debug!(
                    local_network = port_network,
                    peer_network = net,
                    "Network-Number-Is from peer (learned, differs from local)"
                );
            }
        }
    } else if msg_type == NetworkMessageType::INITIALIZE_ROUTING_TABLE_ACK.to_raw() {
        // Clauses 6.4.8/6.6.3.9: same data format as Initialize-Routing-Table.
        // The listed Port IDs are the SENDER's local numbering, so only the
        // DNETs are learned — via the ingress port and immediate source MAC,
        // through the corroboration gate. Management reconfiguration never
        // happens here (see the Init arm): ACK claims stay learned-only and
        // direct routes stay safe. An empty payload carries no entries and is
        // a no-op; any other malformed envelope is dropped without learning.
        let data = &npdu.payload;
        if data.is_empty() {
            return;
        }
        let Some(entries) = parse_routing_table_entries(data) else {
            return;
        };
        {
            let targets: Vec<u16> = entries.iter().map(|e| e.network).collect();
            if !control.authorize(ctx, ControlClass::InitAck, &targets, false) {
                return;
            }
        }
        let mut table = table.lock().await;
        let mut replacement_networks = HashSet::new();
        for entry in &entries {
            let net = entry.network;
            if net == 0 || net == 0xFFFF {
                continue;
            }
            if table.len() >= MAX_LEARNED_ROUTES {
                table.record_learning_cap();
                break;
            }
            if table
                .lookup(net)
                .is_some_and(|entry| !entry.directly_connected && entry.port_index != port_idx)
                && !replacement_networks.insert(net)
            {
                continue;
            }
            if table.apply_learning_claim(
                net,
                port_idx,
                MacAddr::from_slice(source_mac),
                Instant::now(),
            ) {
                table.record_learned();
                debug!(
                    network = net,
                    port = port_idx,
                    "Learned route from Init-Routing-Table-Ack"
                );
            }
        }
    } else if (0x0A..=0x11).contains(&msg_type) {
        // Security messages — acknowledge but do not reject.
        debug!(
            message_type = msg_type,
            "Router received security network message (not implemented)"
        );
    } else {
        // Unknown message type — reject with reason 3 (spec 6.6.3.5).
        // Proprietary types (0x80+) arriving without a routable destination
        // take this explicit-reject path; they are never accepted as known
        // controls. Directed proprietary messages are routed opaquely by the
        // dispatcher and never reach here.
        debug!(
            message_type = msg_type,
            "Router rejecting unknown network message type"
        );
        send_reject(
            &Refused::control(send_txs, local.addresses(), ctx),
            0,
            RejectMessageReason::UNKNOWN_MESSAGE_TYPE,
        );
    }
}

#[cfg(test)]
#[path = "corroboration_tests.rs"]
mod corroboration_tests;