netget/server/coap/
blockwise.rs

1//! Block-wise transfer (RFC 7959) for the CoAP server: the two tables it needs and nothing
2//! else. Pure and synchronous — `mod.rs` holds each behind a `std::sync::Mutex` that is never
3//! held across an `.await`, and passes `now` in so the expiry rules can be tested without a
4//! clock.
5//!
6//! * [`Transfers`] — **Block2**, the response direction. The model answers a request once; a
7//!   representation over one block is cached and every later block is served from the cache,
8//!   so a 64-block transfer costs one model call rather than sixty-four, and every block comes
9//!   from the same answer (the ETag says so). A miss is not an error: the request goes to the
10//!   model again, as RFC 7959 §2.4's stateless server would do, and its answer is sliced.
11//!   The source address a cached entry is keyed by is spoofable, so a hit is a reply the
12//!   server sends to whoever that address is, without the model: each entry therefore lives
13//!   [`TRANSFER_TTL`] from when it was cached (a hit does not renew it) and serves at most
14//!   [`TRANSFER_SERVE_FACTOR`] times its own length before it is evicted, which bounds what
15//!   one model answer can be made to reflect.
16//! * [`Uploads`] — **Block1**, the request direction. Blocks are appended in order and
17//!   answered 2.31 Continue without the model; the body reaches the model once, whole, with
18//!   the last block. Out of order is 4.08, over [`codec::MAX_REPRESENTATION_LEN`] is 4.13 with
19//!   Size1, and a full table is 5.03. One source IP holds at most [`MAX_UPLOADS_PER_IP`]
20//!   uploads; a new one past that evicts that IP's least recently advanced upload.
21//!
22//! Both are keyed by the peer's address and the target resource (RFC 7959 §2.4: block-wise
23//! transfers are matched by endpoint and request URI, not by token — a client may use a new
24//! token for every block, and libcoap does).
25
26use std::collections::HashMap;
27use std::net::SocketAddr;
28use std::sync::Arc;
29use std::time::Duration;
30
31use crate::utils::clock::Instant;
32
33use super::codec::{self, BlockValue};
34
35/// Cached Block2 representations a server holds at once. Inserting past this evicts the
36/// oldest one — a cache, so eviction costs a model call, never a refusal.
37pub const MAX_TRANSFERS: usize = 32;
38
39/// Bytes one cached representation may serve, as a multiple of its payload length, counting
40/// the block that was sent when it was cached. A whole transfer is 1×; 2× leaves room for a
41/// client that retransmits every block once. A block that would pass it evicts the entry and
42/// goes to the model instead.
43pub const TRANSFER_SERVE_FACTOR: usize = 2;
44
45/// Block1 uploads in progress at once. A new upload past this is refused 5.03: an upload
46/// cannot be evicted to make room for a different peer without losing what was already
47/// acknowledged with 2.31.
48pub const MAX_UPLOADS: usize = 16;
49
50/// Block1 uploads one source IP holds at once. A new upload from an IP already holding this
51/// many evicts that IP's least recently advanced one, so one address cannot hold the table.
52pub const MAX_UPLOADS_PER_IP: usize = 2;
53
54/// How long a cached representation is kept after it was cached, and a half-finished upload
55/// after its last block. RFC 7252's EXCHANGE_LIFETIME is 247 s; a client that needs longer
56/// than a minute for one transfer is not doing block-wise transfer, and a miss only costs a
57/// model call.
58pub const TRANSFER_TTL: Duration = Duration::from_secs(60);
59
60/// Which resource, at which endpoint.
61#[derive(Debug, Clone, PartialEq, Eq, Hash)]
62pub struct ResourceKey {
63    pub peer: SocketAddr,
64    pub path: String,
65    pub query: Option<String>,
66}
67
68/// One answer the model gave, kept so every block of it is served from the same bytes.
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct Representation {
71    pub code: u8,
72    pub payload: Vec<u8>,
73    pub content_format: Option<u16>,
74    /// Identifies this representation across its blocks (RFC 7252 §5.10.6): a client that sees
75    /// it change mid-transfer knows it is stitching two answers together.
76    pub etag: Vec<u8>,
77}
78
79impl Representation {
80    pub fn new(code: u8, payload: Vec<u8>, content_format: Option<u16>) -> Representation {
81        let etag = etag_for(&payload, content_format);
82        Representation {
83            code,
84            payload,
85            content_format,
86            etag,
87        }
88    }
89
90    /// Block `block.num` at `block.size()`, and whether more follow. `None` when the block
91    /// starts past the end — except block 0 of an empty representation, which is itself.
92    pub fn block(&self, block: BlockValue) -> Option<(&[u8], bool)> {
93        let offset = block.offset();
94        if offset >= self.payload.len() && !(offset == 0 && self.payload.is_empty()) {
95            return None;
96        }
97        let end = (offset + block.size()).min(self.payload.len());
98        Some((&self.payload[offset..end], end < self.payload.len()))
99    }
100}
101
102/// An 8-byte ETag: FNV-1a over the content format and the payload. Not a security property,
103/// only an identity — two different answers to the same request must not share one.
104pub fn etag_for(payload: &[u8], content_format: Option<u16>) -> Vec<u8> {
105    let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
106    let cf = content_format.map_or([0xFF, 0xFF, 0xFF], |c| {
107        let b = c.to_be_bytes();
108        [1, b[0], b[1]]
109    });
110    for byte in cf.iter().chain(payload.iter()) {
111        hash ^= u64::from(*byte);
112        hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
113    }
114    hash.to_be_bytes().to_vec()
115}
116
117#[derive(Debug)]
118struct Cached {
119    /// When the representation was cached. Never renewed by a hit.
120    cached: Instant,
121    rep: Arc<Representation>,
122    /// Payload bytes served from this entry, including the block sent when it was cached.
123    served: usize,
124}
125
126/// The Block2 cache.
127#[derive(Debug)]
128pub struct Transfers {
129    entries: HashMap<ResourceKey, Cached>,
130    cap: usize,
131    ttl: Duration,
132}
133
134impl Default for Transfers {
135    fn default() -> Self {
136        Transfers::new(MAX_TRANSFERS, TRANSFER_TTL)
137    }
138}
139
140impl Transfers {
141    pub fn new(cap: usize, ttl: Duration) -> Transfers {
142        Transfers {
143            entries: HashMap::new(),
144            cap: cap.max(1),
145            ttl,
146        }
147    }
148
149    pub fn len(&self) -> usize {
150        self.entries.len()
151    }
152
153    pub fn is_empty(&self) -> bool {
154        self.entries.is_empty()
155    }
156
157    fn expire(&mut self, now: Instant) {
158        let ttl = self.ttl;
159        self.entries
160            .retain(|_, c| now.saturating_duration_since(c.cached) < ttl);
161    }
162
163    /// The cached representation for `key`, to serve `block` of it. The block's bytes are
164    /// charged to the entry; a block that would take it past [`TRANSFER_SERVE_FACTOR`] times
165    /// the payload length evicts the entry and is a miss. A hit does not renew the entry's
166    /// age. A block past the end is charged nothing and still returns the representation, so
167    /// the caller can answer 4.02.
168    pub fn get(
169        &mut self,
170        key: &ResourceKey,
171        block: BlockValue,
172        now: Instant,
173    ) -> Option<Arc<Representation>> {
174        self.expire(now);
175        let entry = self.entries.get_mut(key)?;
176        let len = entry.rep.block(block).map_or(0, |(slice, _)| slice.len());
177        let budget = entry
178            .rep
179            .payload
180            .len()
181            .saturating_mul(TRANSFER_SERVE_FACTOR);
182        if entry.served.saturating_add(len) > budget {
183            self.entries.remove(key);
184            return None;
185        }
186        entry.served += len;
187        Some(entry.rep.clone())
188    }
189
190    /// Cache `rep` under `key`, charged `served` bytes for the block just sent from it. If
191    /// `key` already holds this same representation (a block served from the cache), the
192    /// entry is left as it is: its age and its charge stand. Returns the key evicted to make
193    /// room, if any.
194    pub fn insert(
195        &mut self,
196        key: ResourceKey,
197        rep: Arc<Representation>,
198        served: usize,
199        now: Instant,
200    ) -> Option<ResourceKey> {
201        self.expire(now);
202        if self
203            .entries
204            .get(&key)
205            .is_some_and(|c| Arc::ptr_eq(&c.rep, &rep))
206        {
207            return None;
208        }
209        let mut evicted = None;
210        if !self.entries.contains_key(&key) && self.entries.len() >= self.cap {
211            evicted = self
212                .entries
213                .iter()
214                .min_by_key(|(_, c)| c.cached)
215                .map(|(k, _)| k.clone());
216            if let Some(old) = &evicted {
217                self.entries.remove(old);
218            }
219        }
220        self.entries.insert(
221            key,
222            Cached {
223                cached: now,
224                rep,
225                served,
226            },
227        );
228        evicted
229    }
230
231    /// Forget `key` — the last block was served.
232    pub fn remove(&mut self, key: &ResourceKey) {
233        self.entries.remove(key);
234    }
235}
236
237/// An upload is also keyed by method: a PUT and a POST to the same resource are two bodies.
238#[derive(Debug, Clone, PartialEq, Eq, Hash)]
239pub struct UploadKey {
240    pub resource: ResourceKey,
241    pub method: u8,
242}
243
244#[derive(Debug)]
245struct Upload {
246    body: Vec<u8>,
247    touched: Instant,
248}
249
250/// What to do with one Block1 block.
251#[derive(Debug, Clone, PartialEq, Eq)]
252pub enum UploadStep {
253    /// Acknowledge with 2.31 Continue, echoing this Block1 value. The model is not asked.
254    Continue(BlockValue),
255    /// The last block arrived: this is the whole body, in `blocks` blocks.
256    Complete {
257        body: Vec<u8>,
258        blocks: u32,
259        last: BlockValue,
260    },
261    /// Refuse with `code`. `size1` is the largest body this server takes, for a 4.13.
262    Refuse {
263        code: u8,
264        size1: Option<usize>,
265        /// `decision=` token for the log.
266        decision: &'static str,
267        reason: String,
268    },
269}
270
271/// The Block1 reassembly table.
272#[derive(Debug)]
273pub struct Uploads {
274    entries: HashMap<UploadKey, Upload>,
275    cap: usize,
276    per_ip: usize,
277    ttl: Duration,
278    max_body: usize,
279}
280
281impl Default for Uploads {
282    fn default() -> Self {
283        Uploads::new(
284            MAX_UPLOADS,
285            MAX_UPLOADS_PER_IP,
286            TRANSFER_TTL,
287            codec::MAX_REPRESENTATION_LEN,
288        )
289    }
290}
291
292impl Uploads {
293    pub fn new(cap: usize, per_ip: usize, ttl: Duration, max_body: usize) -> Uploads {
294        Uploads {
295            entries: HashMap::new(),
296            cap: cap.max(1),
297            per_ip: per_ip.max(1),
298            ttl,
299            max_body,
300        }
301    }
302
303    pub fn len(&self) -> usize {
304        self.entries.len()
305    }
306
307    pub fn is_empty(&self) -> bool {
308        self.entries.is_empty()
309    }
310
311    fn too_large(&self, what: String) -> UploadStep {
312        UploadStep::Refuse {
313            code: codec::CODE_REQUEST_ENTITY_TOO_LARGE,
314            size1: Some(self.max_body),
315            decision: "refused_upload_too_large",
316            reason: format!("{what}; this server takes at most {} bytes", self.max_body),
317        }
318    }
319
320    /// Take one block. `size1` is the request's Size1 option, the size the client announced.
321    pub fn accept(
322        &mut self,
323        key: UploadKey,
324        block: BlockValue,
325        payload: &[u8],
326        size1: Option<u32>,
327        now: Instant,
328    ) -> UploadStep {
329        let ttl = self.ttl;
330        self.entries
331            .retain(|_, u| now.saturating_duration_since(u.touched) < ttl);
332
333        // Refused on the announcement, before a byte of it is held (RFC 7959 §4).
334        if let Some(announced) = size1 {
335            if announced as usize > self.max_body {
336                self.entries.remove(&key);
337                return self.too_large(format!("the client announced {announced} bytes"));
338            }
339        }
340        if block.more && payload.len() != block.size() {
341            self.entries.remove(&key);
342            return UploadStep::Refuse {
343                code: codec::CODE_BAD_REQUEST,
344                size1: None,
345                decision: "refused_upload_bad_block",
346                reason: format!(
347                    "block {} says more follow but carries {} bytes, not its size {}",
348                    block.num,
349                    payload.len(),
350                    block.size()
351                ),
352            };
353        }
354        if payload.len() > block.size() {
355            self.entries.remove(&key);
356            return UploadStep::Refuse {
357                code: codec::CODE_BAD_REQUEST,
358                size1: None,
359                decision: "refused_upload_bad_block",
360                reason: format!(
361                    "block {} carries {} bytes, over its size {}",
362                    block.num,
363                    payload.len(),
364                    block.size()
365                ),
366            };
367        }
368
369        if block.num == 0 {
370            self.entries.remove(&key);
371            if !block.more {
372                return UploadStep::Complete {
373                    body: payload.to_vec(),
374                    blocks: 1,
375                    last: block,
376                };
377            }
378            // One IP at its own cap gives up its least recently advanced upload; only then is
379            // the shared cap checked, so one address cannot be the reason another is refused.
380            let ip = key.resource.peer.ip();
381            loop {
382                let held = self
383                    .entries
384                    .iter()
385                    .filter(|(k, _)| k.resource.peer.ip() == ip);
386                if held.clone().count() < self.per_ip {
387                    break;
388                }
389                let Some(oldest) = held.min_by_key(|(_, u)| u.touched).map(|(k, _)| k.clone())
390                else {
391                    break;
392                };
393                self.entries.remove(&oldest);
394                tracing::info!(
395                    "CoAP Block1 upload {} {}{} from {} evicted: {} uploads from that IP are \
396                     in progress decision=upload_evicted_per_ip",
397                    codec::method_name(oldest.method).unwrap_or("?"),
398                    oldest.resource.path,
399                    oldest
400                        .resource
401                        .query
402                        .as_ref()
403                        .map(|q| format!("?{q}"))
404                        .unwrap_or_default(),
405                    oldest.resource.peer,
406                    self.per_ip
407                );
408            }
409            if self.entries.len() >= self.cap {
410                return UploadStep::Refuse {
411                    code: codec::CODE_SERVICE_UNAVAILABLE,
412                    size1: None,
413                    decision: "refused_upload_table_full",
414                    reason: format!("{} uploads are already in progress", self.cap),
415                };
416            }
417            self.entries.insert(
418                key,
419                Upload {
420                    body: payload.to_vec(),
421                    touched: now,
422                },
423            );
424            return UploadStep::Continue(block);
425        }
426
427        // A later block must continue exactly where the body ends. Comparing offsets rather
428        // than numbers is what lets a client lower the block size mid-transfer (§2.3).
429        let expected = self.entries.get(&key).map(|u| u.body.len());
430        if expected != Some(block.offset()) {
431            self.entries.remove(&key);
432            return UploadStep::Refuse {
433                code: codec::CODE_REQUEST_ENTITY_INCOMPLETE,
434                size1: None,
435                decision: "refused_upload_out_of_order",
436                reason: match expected {
437                    Some(have) => format!(
438                        "block {} starts at byte {}, but {have} bytes have arrived",
439                        block.num,
440                        block.offset()
441                    ),
442                    None => format!("block {} arrived with no upload in progress", block.num),
443                },
444            };
445        }
446        let upload = self.entries.get_mut(&key).expect("checked above");
447        if upload.body.len() + payload.len() > self.max_body {
448            let have = upload.body.len() + payload.len();
449            self.entries.remove(&key);
450            return self.too_large(format!("the upload reached {have} bytes"));
451        }
452        upload.body.extend_from_slice(payload);
453        upload.touched = now;
454        if block.more {
455            return UploadStep::Continue(block);
456        }
457        let upload = self.entries.remove(&key).expect("present above");
458        UploadStep::Complete {
459            body: upload.body,
460            blocks: block.num + 1,
461            last: block,
462        }
463    }
464}
netget/server/coap/blockwise.rs

netget/server/coap/
blockwise.rs
