netget/server/radsec/
mod.rs

1//! RadSec server (RFC 6614): RADIUS over TLS. TLS and framing live here; every request goes
2//! through the RADIUS server's own decision path, fail-closed rule included.
3pub mod actions;
4pub mod tls;
5
6use crate::logging::emit::Log;
7use crate::protocol::SpawnContext;
8use crate::server::{
9    accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS},
10    connection::ConnectionId,
11    radius::{packet, packet::RadiusPacket, RadiusServer},
12};
13use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
14use anyhow::{ensure, Context, Result};
15use std::{net::SocketAddr, sync::Arc, time::Duration};
16use tokio::{io::AsyncWriteExt, net::TcpListener, sync::mpsc};
17
18/// How long a connection may stay silent between packets by default.
19pub const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
20pub const MAX_IDLE_TIMEOUT_SECS: u64 = 86_400;
21/// Requests on one connection being answered at once; the next waits.
22pub const MAX_IN_FLIGHT: usize = 32;
23
24pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
25    let params = ctx.startup_params.as_ref();
26    let get = |k: &str| -> Result<Option<String>> {
27        Ok(params
28            .map(|p| p.get_optional_string(k))
29            .transpose()?
30            .flatten())
31    };
32    let config = tls::server_config(
33        get("certificate_file")?.as_deref(),
34        get("private_key_file")?.as_deref(),
35        get("ca_file")?.as_deref(),
36    )?;
37    let mutual = get("ca_file")?.is_some();
38    let secret = get("shared_secret")?.unwrap_or_else(|| tls::DEFAULT_SECRET.to_string());
39    ensure!(!secret.is_empty(), "shared_secret must not be empty");
40    let secret = Arc::new(secret.into_bytes());
41    let idle_secs = params
42        .map(|p| p.get_optional_u64("idle_timeout_secs"))
43        .transpose()?
44        .flatten()
45        .unwrap_or(IDLE_TIMEOUT.as_secs());
46    ensure!(
47        (1..=MAX_IDLE_TIMEOUT_SECS).contains(&idle_secs),
48        "idle_timeout_secs must be between 1 and {MAX_IDLE_TIMEOUT_SECS}"
49    );
50    let idle = Duration::from_secs(idle_secs);
51    let listener = TcpListener::bind(ctx.legacy_listen_addr())
52        .await
53        .context("RadSec failed to bind")?;
54    let local = listener.local_addr()?;
55    Log::new(Some(&ctx.status_tx)).info(format!(
56        "RadSec listening on {local} ({})",
57        if mutual {
58            "client certificates required"
59        } else {
60            "no client certificate required"
61        }
62    ));
63    let acceptor = tokio_rustls::TlsAcceptor::from(config);
64    let state = ctx.state.clone();
65    let server_id = ctx.server_id;
66    let accept = tokio::spawn(async move {
67        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
68        loop {
69            let Ok((stream, peer, permit)) =
70                accept_bounded(&listener, &limiter, b"", "RadSec", Some(&ctx.status_tx)).await
71            else {
72                break;
73            };
74            let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
75            let now = crate::utils::clock::Instant::now();
76            ctx.state
77                .add_connection_to_server(
78                    server_id,
79                    ConnectionState {
80                        id,
81                        remote_addr: peer,
82                        local_addr: local,
83                        bytes_sent: 0,
84                        bytes_received: 0,
85                        packets_sent: 0,
86                        packets_received: 0,
87                        last_activity: now,
88                        status: ConnectionStatus::Active,
89                        status_changed_at: now,
90                        protocol_info: ProtocolConnectionInfo::empty(),
91                    },
92                )
93                .await;
94            let child = ctx.clone();
95            let acceptor = acceptor.clone();
96            let secret = secret.clone();
97            ctx.state
98                .spawn_server_task(server_id, async move {
99                    let _permit = permit;
100                    let log = Log::new(Some(&child.status_tx));
101                    match tokio::time::timeout(tls::HANDSHAKE_TIMEOUT, acceptor.accept(stream))
102                        .await
103                    {
104                        Ok(Ok(tls)) => {
105                            if let Err(e) = session(&child, id, peer, tls, secret, idle).await {
106                                log.warn(format!(
107                                    "RadSec connection {id} from {peer} ended: {e:#}"
108                                ));
109                            }
110                        }
111                        Ok(Err(e)) => log.warn(format!("RadSec handshake with {peer} failed: {e}")),
112                        Err(_) => log.warn(format!("RadSec handshake with {peer} timed out")),
113                    }
114                    child
115                        .state
116                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
117                        .await;
118                    let _ = child.status_tx.send("__UPDATE_UI__".into());
119                })
120                .await;
121        }
122    });
123    state.register_server_task(server_id, accept).await;
124    Ok(local)
125}
126
127/// Serve one TLS connection: read packets, answer each through RADIUS's decision path, write
128/// the replies in the order they are ready.
129async fn session(
130    ctx: &SpawnContext,
131    id: ConnectionId,
132    peer: SocketAddr,
133    tls: tokio_rustls::server::TlsStream<tokio::net::TcpStream>,
134    secret: Arc<Vec<u8>>,
135    idle: Duration,
136) -> Result<()> {
137    let (reader, mut writer) = tokio::io::split(tls);
138    let (frames_tx, mut frames) = mpsc::channel(8);
139    let reader_task = tokio::spawn(packet::read_frames(reader, frames_tx));
140    let (replies_tx, mut replies) = mpsc::channel::<Vec<u8>>(MAX_IN_FLIGHT);
141    let slots = Arc::new(tokio::sync::Semaphore::new(MAX_IN_FLIGHT));
142    let mut answering = tokio::task::JoinSet::new();
143    let result = loop {
144        tokio::select! {
145            frame = tokio::time::timeout(idle, frames.recv()) => {
146                let frame = match frame {
147                    Err(_) => break Ok(()),
148                    Ok(None) => break Ok(()),
149                    Ok(Some(Err(e))) if e.kind() == std::io::ErrorKind::UnexpectedEof => break Ok(()),
150                    Ok(Some(Err(e))) => break Err(anyhow::anyhow!(e)),
151                    Ok(Some(Ok(f))) => f,
152                };
153                ctx.state
154                    .update_connection_stats(ctx.server_id, id, Some(frame.len() as u64), None, Some(1), None)
155                    .await;
156                let request = match RadiusPacket::decode(&frame) {
157                    Ok(p) => p,
158                    Err(e) => {
159                        Log::new(Some(&ctx.status_tx)).warn(format!("RadSec dropped a packet from {peer}: {e}"));
160                        continue;
161                    }
162                };
163                if let Err(e) = packet::verify_request_message_authenticator(&frame, &secret) {
164                    Log::new(Some(&ctx.status_tx)).warn(format!(
165                        "RadSec dropped {} id={} from {peer}: {e}",
166                        packet::code_name(request.code),
167                        request.identifier
168                    ));
169                    continue;
170                }
171                if request.code == packet::CODE_ACCOUNTING_REQUEST {
172                    if let Err(e) = packet::verify_accounting_request(&request, &secret) {
173                        Log::new(Some(&ctx.status_tx)).warn(format!(
174                            "RadSec dropped Accounting-Request id={} from {peer}: {e}", request.identifier
175                        ));
176                        continue;
177                    }
178                }
179                let Ok(slot) = slots.clone().acquire_owned().await else { break Ok(()) };
180                let (llm, state, status, secret, tx) =
181                    (ctx.llm_client.clone(), ctx.state.clone(), ctx.status_tx.clone(), secret.clone(), replies_tx.clone());
182                let server_id = ctx.server_id;
183                answering.spawn(async move {
184                    let _slot = slot;
185                    if let Some(reply) =
186                        RadiusServer::answer(request, peer, Some(id), llm, state, status, server_id, secret).await
187                    {
188                        let _ = tx.send(reply).await;
189                    }
190                });
191            }
192            Some(reply) = replies.recv() => {
193                writer.write_all(&reply).await?;
194                writer.flush().await?;
195                ctx.state
196                    .update_connection_stats(ctx.server_id, id, None, Some(reply.len() as u64), None, Some(1))
197                    .await;
198            }
199            Some(_) = answering.join_next(), if !answering.is_empty() => {}
200        }
201    };
202    // Requests already being answered still get their replies before the connection closes.
203    drop(replies_tx);
204    while answering.join_next().await.is_some() {}
205    while let Ok(reply) = replies.try_recv() {
206        let _ = writer.write_all(&reply).await;
207    }
208    let _ = writer.shutdown().await;
209    reader_task.abort();
210    result
211}
netget/server/radsec/mod.rs

netget/server/radsec/
mod.rs
