veloren_query_server/
server.rs

1#[expect(deprecated)] use std::hash::SipHasher;
2use std::{
3    hash::{Hash, Hasher},
4    io::{self, ErrorKind},
5    net::SocketAddr,
6    sync::{Arc, Mutex},
7    time::{Duration, Instant},
8};
9
10use protocol::Parcel;
11use rand::{RngExt, rng};
12use tokio::{net::UdpSocket, sync::watch};
13use tracing::{debug, error, trace};
14
15use crate::{
16    proto::{
17        Init, MAX_REQUEST_SIZE, MAX_RESPONSE_SIZE, QueryServerRequest, QueryServerResponse,
18        RawQueryServerRequest, RawQueryServerResponse, ServerInfo, VELOREN_HEADER, VERSION,
19    },
20    ratelimit::{RateLimiter, ReducedIpAddr},
21};
22
23const SECRET_REGEN_INTERNVAL: Duration = Duration::from_secs(300);
24
25pub struct QueryServer {
26    addr: SocketAddr,
27    server_info: watch::Receiver<ServerInfo>,
28    settings: protocol::Settings,
29    ratelimit: RateLimiter,
30}
31
32#[derive(Default, Clone, Copy, Debug)]
33pub struct Metrics {
34    pub received_packets: u32,
35    /// Note, these are not counted in `received_packets`.
36    pub dropped_packets: u32,
37    pub invalid_packets: u32,
38    pub proccessing_errors: u32,
39    pub info_requests: u32,
40    pub init_requests: u32,
41    pub sent_responses: u32,
42    pub failed_responses: u32,
43    pub timed_out_responses: u32,
44    pub ratelimited: u32,
45}
46
47impl QueryServer {
48    pub fn new(addr: SocketAddr, server_info: watch::Receiver<ServerInfo>, ratelimit: u16) -> Self {
49        Self {
50            addr,
51            server_info,
52            ratelimit: RateLimiter::new(ratelimit),
53            settings: Default::default(),
54        }
55    }
56
57    /// This produces TRACE level logs for any packet received on the assigned
58    /// port. To prevent potentially unfettered log spam, disable the TRACE
59    /// level for this crate (when outside of debugging contexts).
60    ///
61    /// NOTE: TRACE and DEBUG levels are disabled by default for this crate when
62    /// using `veloren-common-frontend`.
63    pub async fn run(&mut self, metrics: Arc<Mutex<Metrics>>) -> Result<(), tokio::io::Error> {
64        let mut socket = UdpSocket::bind(self.addr).await?;
65
66        let gen_secret = || {
67            let mut rng = rng();
68            (rng.random::<u64>(), rng.random::<u64>())
69        };
70        let mut secrets = gen_secret();
71        let mut last_secret_refresh = Instant::now();
72
73        let mut buf = Box::new([0; MAX_REQUEST_SIZE]);
74        loop {
75            let (len, remote_addr) = match socket.recv_from(&mut *buf).await {
76                Ok(v) => v,
77                Err(e) if e.kind() == ErrorKind::NotConnected => {
78                    error!(
79                        ?e,
80                        "Query server connection was closed, re-binding to socket..."
81                    );
82                    socket = UdpSocket::bind(self.addr).await?;
83                    continue;
84                },
85                err => {
86                    debug!(?err, "Error while receiving from query server socket");
87                    continue;
88                },
89            };
90
91            let mut new_metrics = Metrics {
92                received_packets: 1,
93                ..Default::default()
94            };
95
96            let raw_msg_buf = &buf[..len];
97            let msg_buf = if Self::validate_datagram(raw_msg_buf) {
98                // Require 2 extra bytes for version (currently unused)
99                &raw_msg_buf[2..(raw_msg_buf.len() - VELOREN_HEADER.len())]
100            } else {
101                if let Ok(mut metrics) = metrics.lock() {
102                    metrics.dropped_packets += 1;
103                }
104                continue;
105            };
106
107            self.process_datagram(msg_buf, remote_addr, secrets, &mut new_metrics, &socket)
108                .await;
109
110            // Update metrics at the end of eath packet
111            if let Ok(mut metrics) = metrics.lock() {
112                *metrics += new_metrics;
113            }
114
115            {
116                let now = Instant::now();
117                if now.duration_since(last_secret_refresh) > SECRET_REGEN_INTERNVAL {
118                    last_secret_refresh = now;
119                    secrets = gen_secret();
120                }
121
122                self.ratelimit.maintain(now);
123            }
124        }
125    }
126
127    // Header must be discarded after this validation passes
128    fn validate_datagram(data: &[u8]) -> bool {
129        let len = data.len();
130        // Require 2 extra bytes for version (currently unused)
131        if len < MAX_RESPONSE_SIZE.max(VELOREN_HEADER.len() + 2) {
132            trace!(?len, "Datagram too short");
133            false
134        } else if len > MAX_REQUEST_SIZE {
135            trace!(?len, "Datagram too large");
136            false
137        } else if data[(len - VELOREN_HEADER.len())..] != VELOREN_HEADER {
138            trace!(?len, "Datagram header invalid");
139            false
140        // TODO: Allow lower versions once proper versioning is added.
141        } else if u16::from_ne_bytes(data[..2].try_into().unwrap()) != VERSION {
142            trace!(
143                "Datagram has invalid version {:?}, current {VERSION:?}",
144                &data[..2]
145            );
146            false
147        } else {
148            true
149        }
150    }
151
152    async fn process_datagram(
153        &mut self,
154        datagram: &[u8],
155        remote: SocketAddr,
156        secrets: (u64, u64),
157        metrics: &mut Metrics,
158        socket: &UdpSocket,
159    ) {
160        let Ok(RawQueryServerRequest {
161            p: client_p,
162            request,
163        }) =
164            <RawQueryServerRequest as Parcel>::read(&mut io::Cursor::new(datagram), &self.settings)
165        else {
166            metrics.invalid_packets += 1;
167            return;
168        };
169
170        trace!(?request, "Received packet");
171
172        #[expect(deprecated)]
173        let real_p = {
174            // Use SipHash-2-4 to compute the `p` value from a server specific
175            // secret and the client's address.
176            //
177            // This is used to verify that packets are from an entity that can
178            // receive packets at the given address.
179            //
180            // Only use the first 64 bits from Ipv6 addresses since the latter
181            // 64 bits can change very frequently (as much as for every
182            // request).
183            let mut hasher = SipHasher::new_with_keys(secrets.0, secrets.1);
184            ReducedIpAddr::from(remote.ip()).hash(&mut hasher);
185            hasher.finish()
186        };
187
188        if real_p != client_p {
189            Self::send_response(
190                RawQueryServerResponse::Init(Init {
191                    p: real_p,
192                    max_supported_version: VERSION,
193                }),
194                remote,
195                socket,
196                metrics,
197            )
198            .await;
199
200            return;
201        }
202
203        if !self.ratelimit.can_request(remote.ip().into()) {
204            trace!("Ratelimited request");
205            metrics.ratelimited += 1;
206            return;
207        }
208
209        match request {
210            QueryServerRequest::Init => {
211                metrics.init_requests += 1;
212                Self::send_response(
213                    RawQueryServerResponse::Init(Init {
214                        p: real_p,
215                        max_supported_version: VERSION,
216                    }),
217                    remote,
218                    socket,
219                    metrics,
220                )
221                .await;
222            },
223            QueryServerRequest::ServerInfo => {
224                metrics.info_requests += 1;
225                let server_info = *self.server_info.borrow();
226                Self::send_response(
227                    RawQueryServerResponse::Response(QueryServerResponse::ServerInfo(server_info)),
228                    remote,
229                    socket,
230                    metrics,
231                )
232                .await;
233            },
234        }
235    }
236
237    async fn send_response(
238        response: RawQueryServerResponse,
239        addr: SocketAddr,
240        socket: &UdpSocket,
241        metrics: &mut Metrics,
242    ) {
243        // TODO: Once more versions are added, send the packet in the same version as
244        // the request here.
245        match <RawQueryServerResponse as Parcel>::raw_bytes(&response, &Default::default()) {
246            Ok(data) => {
247                if data.len() > MAX_RESPONSE_SIZE {
248                    error!(
249                        ?MAX_RESPONSE_SIZE,
250                        "Attempted to send a response larger than the maximum allowed size (size: \
251                         {}, response: {response:?})",
252                        data.len()
253                    );
254                    #[cfg(debug_assertions)]
255                    panic!(
256                        "Attempted to send a response larger than the maximum allowed size (size: \
257                         {}, max: {}, response: {response:?})",
258                        data.len(),
259                        MAX_RESPONSE_SIZE
260                    );
261                }
262
263                match socket.send_to(&data, addr).await {
264                    Ok(_) => {
265                        metrics.sent_responses += 1;
266                    },
267                    Err(err) => {
268                        metrics.failed_responses += 1;
269                        debug!(?err, "Failed to send query server response");
270                    },
271                }
272            },
273            Err(error) => {
274                trace!(?error, "Failed to serialize response");
275                #[cfg(debug_assertions)]
276                panic!("Serializing response failed: {error:?} ({response:?})");
277            },
278        }
279    }
280}
281
282impl std::ops::AddAssign for Metrics {
283    fn add_assign(
284        &mut self,
285        Self {
286            received_packets,
287            dropped_packets,
288            invalid_packets,
289            proccessing_errors,
290            info_requests,
291            init_requests,
292            sent_responses,
293            failed_responses,
294            timed_out_responses,
295            ratelimited,
296        }: Self,
297    ) {
298        self.received_packets += received_packets;
299        self.dropped_packets += dropped_packets;
300        self.invalid_packets += invalid_packets;
301        self.proccessing_errors += proccessing_errors;
302        self.info_requests += info_requests;
303        self.init_requests += init_requests;
304        self.sent_responses += sent_responses;
305        self.failed_responses += failed_responses;
306        self.timed_out_responses += timed_out_responses;
307        self.ratelimited += ratelimited;
308    }
309}
310
311impl Metrics {
312    /// Resets all metrics to 0 and returns previous ones
313    ///
314    /// Used by the consumer of the metrics.
315    pub fn reset(&mut self) -> Self { std::mem::take(self) }
316}
veloren_query_server/server.rs

veloren_query_server/
server.rs
