netget/server/rpki_rtr/
codec.rs

1//! Selected RFC8210/RFC6810 prefix records, with no RPKI repository or VRP store.
2use anyhow::{bail, ensure, Context, Result};
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5use std::{net::IpAddr, time::Duration};
6use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
7
8pub const MAX_PDU_BYTES: usize = 4096;
9/// Records one handler answer may carry (reset or delta).
10pub const MAX_RECORDS: usize = 4096;
11pub const MAX_JSON_BYTES: usize = 1024 * 1024;
12pub const DEFAULT_REFRESH_SECONDS: u32 = 3600;
13pub const DEFAULT_RETRY_SECONDS: u32 = 600;
14pub const DEFAULT_EXPIRE_SECONDS: u32 = 7200;
15pub const DEFAULT_PROTOCOL_VERSION: u8 = 1;
16pub const WRITE_TIMEOUT: Duration = Duration::from_secs(10);
17
18fn announced() -> bool {
19    true
20}
21#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
22#[serde(deny_unknown_fields)]
23pub struct Record {
24    pub prefix: String,
25    pub max_length: u8,
26    pub asn: u32,
27    #[serde(default = "announced")]
28    pub announcement: bool,
29}
30impl Record {
31    pub fn parsed(&self) -> Result<(IpAddr, u8)> {
32        ensure!(self.prefix.len() <= 49, "RPKI-RTR prefix text bound");
33        let (address, bits) = self
34            .prefix
35            .split_once('/')
36            .context("RPKI-RTR prefix requires CIDR length")?;
37        let ip: IpAddr = address.parse().context("RPKI-RTR prefix address")?;
38        let len: u8 = bits.parse().context("RPKI-RTR prefix length")?;
39        let canonical = match ip {
40            IpAddr::V4(v) => {
41                ensure!(
42                    len <= self.max_length && self.max_length <= 32,
43                    "RPKI-RTR IPv4 prefix lengths"
44                );
45                u32::from(v) & u32::MAX.checked_shr(u32::from(len)).unwrap_or(0) == 0
46            }
47            IpAddr::V6(v) => {
48                ensure!(
49                    len <= self.max_length && self.max_length <= 128,
50                    "RPKI-RTR IPv6 prefix lengths"
51                );
52                u128::from(v) & u128::MAX.checked_shr(u32::from(len)).unwrap_or(0) == 0
53            }
54        };
55        ensure!(canonical, "RPKI-RTR prefix has nonzero host bits");
56        Ok((ip, len))
57    }
58    pub fn canonical(&self) -> Result<Self> {
59        let (ip, len) = self.parsed()?;
60        Ok(Self {
61            prefix: format!("{ip}/{len}"),
62            ..self.clone()
63        })
64    }
65}
66#[derive(Clone, Debug, Deserialize, Serialize)]
67#[serde(deny_unknown_fields)]
68pub struct Batch {
69    pub serial: u32,
70    pub records: Vec<Record>,
71}
72impl Batch {
73    pub fn validate(&self, reset: bool) -> Result<()> {
74        ensure!(
75            self.records.len() <= MAX_RECORDS,
76            "RPKI-RTR transaction record bound"
77        );
78        let mut announcements = std::collections::HashSet::new();
79        for record in &self.records {
80            let (ip, len) = record.parsed()?;
81            ensure!(
82                !reset || record.announcement,
83                "RPKI-RTR reset cannot withdraw records"
84            );
85            if record.announcement {
86                ensure!(
87                    announcements.insert((ip, len, record.max_length, record.asn)),
88                    "RPKI-RTR duplicate announcement in transaction"
89                );
90            }
91        }
92        Ok(())
93    }
94}
95#[derive(Clone, Copy, Debug, PartialEq, Eq)]
96pub struct Intervals {
97    pub refresh: u32,
98    pub retry: u32,
99    pub expire: u32,
100}
101impl Default for Intervals {
102    fn default() -> Self {
103        Self {
104            refresh: DEFAULT_REFRESH_SECONDS,
105            retry: DEFAULT_RETRY_SECONDS,
106            expire: DEFAULT_EXPIRE_SECONDS,
107        }
108    }
109}
110impl Intervals {
111    pub fn validate(&self) -> Result<()> {
112        ensure!(
113            (1..=86400).contains(&self.refresh),
114            "RPKI-RTR refresh interval1..86400"
115        );
116        ensure!(
117            (1..=7200).contains(&self.retry),
118            "RPKI-RTR retry interval1..7200"
119        );
120        ensure!(
121            (600..=172800).contains(&self.expire)
122                && self.expire > self.refresh
123                && self.expire > self.retry,
124            "RPKI-RTR expire interval600..172800 and larger than refresh/retry"
125        );
126        Ok(())
127    }
128}
129/// RFC1982 strict order, including wrap-around. The half-space is ambiguous.
130pub fn serial_newer(next: u32, old: u32) -> bool {
131    let difference = next.wrapping_sub(old);
132    difference != 0 && difference < 0x8000_0000
133}
134#[derive(Clone, Debug, PartialEq, Eq)]
135pub enum Pdu {
136    SerialNotify {
137        session: u16,
138        serial: u32,
139    },
140    SerialQuery {
141        session: u16,
142        serial: u32,
143    },
144    ResetQuery,
145    CacheResponse {
146        session: u16,
147    },
148    Prefix(Record),
149    EndOfData {
150        session: u16,
151        serial: u32,
152        intervals: Intervals,
153    },
154    CacheReset,
155    ErrorReport {
156        code: u16,
157        erroneous: Vec<u8>,
158        diagnostic: String,
159    },
160}
161#[derive(Clone, Debug, PartialEq, Eq)]
162pub struct Packet {
163    pub version: u8,
164    pub pdu: Pdu,
165}
166fn number(bytes: &[u8], start: usize) -> u32 {
167    u32::from_be_bytes(
168        bytes[start..start + 4]
169            .try_into()
170            .expect("checked fixed PDU length"),
171    )
172}
173impl Packet {
174    pub fn encode(&self) -> Result<Vec<u8>> {
175        ensure!(self.version <= 1, "RPKI-RTR supported versions0/1");
176        let mut body = Vec::new();
177        let (kind, field) = match &self.pdu {
178            Pdu::SerialNotify { session, serial } => {
179                body.extend(serial.to_be_bytes());
180                (0, *session)
181            }
182            Pdu::SerialQuery { session, serial } => {
183                body.extend(serial.to_be_bytes());
184                (1, *session)
185            }
186            Pdu::ResetQuery => (2, 0),
187            Pdu::CacheResponse { session } => (3, *session),
188            Pdu::Prefix(record) => {
189                let (ip, len) = record.parsed()?;
190                body.extend([u8::from(record.announcement), len, record.max_length, 0]);
191                let kind = match ip {
192                    IpAddr::V4(v) => {
193                        body.extend(v.octets());
194                        4
195                    }
196                    IpAddr::V6(v) => {
197                        body.extend(v.octets());
198                        6
199                    }
200                };
201                body.extend(record.asn.to_be_bytes());
202                (kind, 0)
203            }
204            Pdu::EndOfData {
205                session,
206                serial,
207                intervals,
208            } => {
209                intervals.validate()?;
210                body.extend(serial.to_be_bytes());
211                if self.version == 1 {
212                    body.extend(intervals.refresh.to_be_bytes());
213                    body.extend(intervals.retry.to_be_bytes());
214                    body.extend(intervals.expire.to_be_bytes());
215                }
216                (7, *session)
217            }
218            Pdu::CacheReset => (8, 0),
219            Pdu::ErrorReport {
220                code,
221                erroneous,
222                diagnostic,
223            } => {
224                ensure!(
225                    *code <= 8 && erroneous.len() <= 64 && diagnostic.len() <= 1024,
226                    "RPKI-RTR error report bound"
227                );
228                body.extend((erroneous.len() as u32).to_be_bytes());
229                body.extend(erroneous);
230                body.extend((diagnostic.len() as u32).to_be_bytes());
231                body.extend(diagnostic.as_bytes());
232                (10, *code)
233            }
234        };
235        let len = body.len() + 8;
236        ensure!(len <= MAX_PDU_BYTES, "RPKI-RTR encoded PDU bound");
237        let mut frame = Vec::with_capacity(len);
238        frame.extend([self.version, kind]);
239        frame.extend(field.to_be_bytes());
240        frame.extend((len as u32).to_be_bytes());
241        frame.extend(body);
242        Ok(frame)
243    }
244    pub fn decode(bytes: &[u8]) -> Result<Self> {
245        ensure!(
246            (8..=MAX_PDU_BYTES).contains(&bytes.len()),
247            "RPKI-RTR PDU length bound"
248        );
249        let version = bytes[0];
250        ensure!(version <= 1, "RPKI-RTR unsupported protocol version");
251        ensure!(
252            number(bytes, 4) as usize == bytes.len(),
253            "RPKI-RTR exact PDU length"
254        );
255        let field = u16::from_be_bytes([bytes[2], bytes[3]]);
256        let pdu = match bytes[1] {
257            0 | 1 => {
258                ensure!(bytes.len() == 12, "RPKI-RTR serial PDU length");
259                let serial = number(bytes, 8);
260                if bytes[1] == 0 {
261                    Pdu::SerialNotify {
262                        session: field,
263                        serial,
264                    }
265                } else {
266                    Pdu::SerialQuery {
267                        session: field,
268                        serial,
269                    }
270                }
271            }
272            2 => {
273                ensure!(bytes.len() == 8, "RPKI-RTR reset query length");
274                Pdu::ResetQuery
275            }
276            3 => {
277                ensure!(bytes.len() == 8, "RPKI-RTR cache response length");
278                Pdu::CacheResponse { session: field }
279            }
280            4 | 6 => {
281                let v4 = bytes[1] == 4;
282                ensure!(
283                    bytes.len() == if v4 { 20 } else { 32 },
284                    "RPKI-RTR prefix PDU length"
285                );
286                let ip = if v4 {
287                    IpAddr::V4(std::net::Ipv4Addr::from(<[u8; 4]>::try_from(
288                        &bytes[12..16],
289                    )?))
290                } else {
291                    IpAddr::V6(std::net::Ipv6Addr::from(<[u8; 16]>::try_from(
292                        &bytes[12..28],
293                    )?))
294                };
295                let record = Record {
296                    prefix: format!("{ip}/{}", bytes[9]),
297                    max_length: bytes[10],
298                    asn: number(bytes, if v4 { 16 } else { 28 }),
299                    announcement: bytes[8] & 1 == 1,
300                };
301                record.parsed()?;
302                Pdu::Prefix(record)
303            }
304            7 => {
305                ensure!(
306                    bytes.len() == if version == 1 { 24 } else { 12 },
307                    "RPKI-RTR version-specific EndOfData length"
308                );
309                let intervals = if version == 1 {
310                    Intervals {
311                        refresh: number(bytes, 12),
312                        retry: number(bytes, 16),
313                        expire: number(bytes, 20),
314                    }
315                } else {
316                    Intervals::default()
317                };
318                intervals.validate()?;
319                Pdu::EndOfData {
320                    session: field,
321                    serial: number(bytes, 8),
322                    intervals,
323                }
324            }
325            8 => {
326                ensure!(bytes.len() == 8, "RPKI-RTR cache reset length");
327                Pdu::CacheReset
328            }
329            10 => {
330                ensure!(bytes.len() >= 16 && field <= 8, "RPKI-RTR error header");
331                let encapsulated = number(bytes, 8) as usize;
332                ensure!(
333                    encapsulated <= 64 && encapsulated + 16 <= bytes.len(),
334                    "RPKI-RTR encapsulated error bound"
335                );
336                let text_len = number(bytes, 12 + encapsulated) as usize;
337                ensure!(
338                    text_len <= 1024 && 16 + encapsulated + text_len == bytes.len(),
339                    "RPKI-RTR error diagnostic bound"
340                );
341                Pdu::ErrorReport {
342                    code: field,
343                    erroneous: bytes[12..12 + encapsulated].to_vec(),
344                    diagnostic: std::str::from_utf8(&bytes[16 + encapsulated..])?.to_owned(),
345                }
346            }
347            _ => bail!("RPKI-RTR unsupported selected PDU type"),
348        };
349        Ok(Self { version, pdu })
350    }
351}
352/// Read exactly one header/body before any allocation; TCP chunking is irrelevant.
353pub async fn read_frame<R: AsyncRead + Unpin>(
354    reader: &mut R,
355    deadline: Duration,
356) -> Result<Vec<u8>> {
357    tokio::time::timeout(deadline, async {
358        let mut header = [0u8; 8];
359        reader.read_exact(&mut header).await?;
360        let len = number(&header, 4) as usize;
361        ensure!(
362            (8..=MAX_PDU_BYTES).contains(&len),
363            "RPKI-RTR inbound PDU bound"
364        );
365        let mut bytes = vec![0u8; len];
366        bytes[..8].copy_from_slice(&header);
367        reader.read_exact(&mut bytes[8..]).await?;
368        Ok(bytes)
369    })
370    .await
371    .context("RPKI-RTR whole-frame deadline")?
372}
373pub async fn write_packet<W: AsyncWrite + Unpin>(writer: &mut W, packet: &Packet) -> Result<usize> {
374    let bytes = packet.encode()?;
375    tokio::time::timeout(WRITE_TIMEOUT, async {
376        writer.write_all(&bytes).await?;
377        writer.flush().await?;
378        Ok::<_, anyhow::Error>(())
379    })
380    .await
381    .context("RPKI-RTR write deadline")??;
382    Ok(bytes.len())
383}
384pub fn within_json_budget(value: &Value) -> bool {
385    crate::utils::json_budget::within_budget(value, MAX_JSON_BYTES, 6 * MAX_RECORDS + 64, 8)
386}
387pub fn owned_json(value: Value) -> Result<Value> {
388    if !within_json_budget(&value) {
389        crate::utils::json_budget::drop_iteratively(value);
390        bail!("RPKI-RTR action JSON bound");
391    }
392    Ok(value)
393}
394pub fn seconds(value: Option<u64>, default: u64, max: u64, name: &str) -> Result<Duration> {
395    let seconds = value.unwrap_or(default);
396    ensure!(
397        (1..=max).contains(&seconds),
398        "{name} must be between 1 and {max}"
399    );
400    Ok(Duration::from_secs(seconds))
401}
402
403/// RFC 8210 §12 error codes.
404pub fn error_name(code: u16) -> &'static str {
405    match code {
406        0 => "corrupt_data",
407        1 => "internal_error",
408        2 => "no_data_available",
409        3 => "invalid_request",
410        4 => "unsupported_protocol_version",
411        5 => "unsupported_pdu_type",
412        6 => "withdrawal_of_unknown_record",
413        7 => "duplicate_announcement_received",
414        8 => "unexpected_protocol_version",
415        _ => "unknown",
416    }
417}
418
419/// An Error Report carrying up to 64 bytes of the offending PDU.
420pub fn error_report(version: u8, code: u16, offending: &[u8], diagnostic: &str) -> Packet {
421    let erroneous = offending[..offending.len().min(64)].to_vec();
422    let diagnostic = crate::utils::truncate::truncate_for_log(diagnostic, 1000);
423    Packet {
424        version: version.min(1),
425        pdu: Pdu::ErrorReport {
426            code,
427            erroneous,
428            diagnostic,
429        },
430    }
431}
netget/server/rpki_rtr/codec.rs

netget/server/rpki_rtr/
codec.rs
