macula_rust/station_link/
dht.rs

1//! The DHT, as macula 12's facade reaches it: station procedures on the zero
2//! realm, targeting the connected station, carrying record wire bytes. Every
3//! record found is verified here before it is handed on, and one that does
4//! not verify is dropped and counted.
5
6use crate::cbor::Value;
7use crate::record::{self, RecordType, Verified};
8
9use crate::profile::Profile;
10
11use super::{now_ms, Call, Link, LinkError};
12
13/// The version of every `_dht.*` argument and reply form (WIRE_1_0.md
14/// section 6).
15const DHT_V: i128 = 1;
16
17/// The longest refusal reason a refused put names.
18const MAX_REASON_BYTES: usize = 64;
19
20/// A `_dht.*` reply as a caller reads it.
21#[derive(Debug, Clone, PartialEq)]
22enum DhtReply {
23    /// `_dht.put_record` stored the record.
24    Stored,
25    /// `_dht.put_record` refused it, by the custodian's refusal name.
26    Refused(String),
27    /// `_dht.find_record` found a record, as its wire bytes.
28    Record(Vec<u8>),
29    /// `_dht.find_record` found none.
30    NotFound,
31    /// `_dht.find_records` or `_dht.find_records_by_type`: the records, as
32    /// their wire bytes, perhaps none.
33    Records(Vec<Vec<u8>>),
34}
35
36/// Why a caller refused a `_dht.*` reply.
37#[derive(Debug, Clone, Copy, PartialEq, Eq)]
38enum ReplyRefusal {
39    /// A `v` other than 1 (`reply_version_unsupported`).
40    VersionUnsupported,
41    /// A missing or extra key, a field of the wrong type or length, or a
42    /// result `procedure` never answers with (`malformed_reply`).
43    Malformed,
44}
45
46impl From<ReplyRefusal> for LinkError {
47    fn from(refusal: ReplyRefusal) -> LinkError {
48        match refusal {
49            ReplyRefusal::VersionUnsupported => LinkError::ReplyVersionUnsupported,
50            ReplyRefusal::Malformed => LinkError::UnexpectedReply("malformed_reply".into()),
51        }
52    }
53}
54
55/// Reads `procedure`'s reply: a map with text keys, `v` 1, and exactly the
56/// keys of one of the procedure's results, each of its type.
57fn read_reply(procedure: &str, reply: &Value) -> Result<DhtReply, ReplyRefusal> {
58    let Value::Map(pairs) = reply else {
59        return Err(ReplyRefusal::Malformed);
60    };
61    match reply.get("v") {
62        Some(Value::Int(DHT_V)) => {}
63        Some(Value::Int(_)) => return Err(ReplyRefusal::VersionUnsupported),
64        _ => return Err(ReplyRefusal::Malformed),
65    }
66    let result = match reply.get("result") {
67        Some(Value::Text(r)) => r.as_str(),
68        _ => return Err(ReplyRefusal::Malformed),
69    };
70    let read = match (procedure, result) {
71        ("_dht.put_record", "ok") => Some((2, DhtReply::Stored)),
72        ("_dht.put_record", "refused") => match reply.get("reason") {
73            Some(Value::Text(r)) if r.len() <= MAX_REASON_BYTES => {
74                Some((3, DhtReply::Refused(r.clone())))
75            }
76            _ => None,
77        },
78        ("_dht.find_record", "record") => match reply.get("record") {
79            Some(Value::Bytes(wire)) => Some((3, DhtReply::Record(wire.clone()))),
80            _ => None,
81        },
82        ("_dht.find_record", "not_found") => Some((2, DhtReply::NotFound)),
83        ("_dht.find_records" | "_dht.find_records_by_type", "records") => {
84            record_list(reply.get("records")).map(|records| (3, DhtReply::Records(records)))
85        }
86        _ => None,
87    };
88    match read {
89        Some((size, reply)) if pairs.len() == size => Ok(reply),
90        _ => Err(ReplyRefusal::Malformed),
91    }
92}
93
94/// A `records` field: a list of byte strings, perhaps empty.
95fn record_list(v: Option<&Value>) -> Option<Vec<Vec<u8>>> {
96    let Some(Value::List(items)) = v else {
97        return None;
98    };
99    items
100        .iter()
101        .map(|item| match item {
102            Value::Bytes(wire) => Some(wire.clone()),
103            _ => None,
104        })
105        .collect()
106}
107
108/// A `_dht.*` argument form: `v` 1 and one field.
109fn dht_args(name: &str, value: Value) -> Value {
110    Value::Map(vec![
111        (Value::text("v"), Value::Int(DHT_V)),
112        (Value::text(name), value),
113    ])
114}
115
116impl Link {
117    /// Stores a signed record, as its wire bytes, in the station's DHT. A
118    /// station that refuses it names the custodian's refusal
119    /// ([`LinkError::RecordRefused`]).
120    pub async fn put_record(&self, wire: &[u8]) -> Result<(), LinkError> {
121        let args = dht_args("record", Value::Bytes(wire.to_vec()));
122        match self.dht_call("_dht.put_record", args).await? {
123            DhtReply::Stored => Ok(()),
124            DhtReply::Refused(reason) => Err(LinkError::RecordRefused(reason)),
125            _ => Err(ReplyRefusal::Malformed.into()),
126        }
127    }
128
129    /// The record stored under `key`, verified as a reader verifies it.
130    pub async fn find_record(&self, key: &[u8; 32]) -> Result<Verified, LinkError> {
131        let args = dht_args("key", Value::Bytes(key.to_vec()));
132        match self.dht_call("_dht.find_record", args).await? {
133            DhtReply::NotFound => Err(LinkError::RecordNotFound),
134            DhtReply::Record(wire) => Ok(record::verify(&wire, self.profile(), now_ms())?),
135            _ => Err(ReplyRefusal::Malformed.into()),
136        }
137    }
138
139    /// Every record stored under `key` that verifies, and how many the
140    /// station returned that did not.
141    pub async fn find_records(&self, key: &[u8; 32]) -> Result<(Vec<Verified>, usize), LinkError> {
142        let args = dht_args("key", Value::Bytes(key.to_vec()));
143        self.verified_list("_dht.find_records", args).await
144    }
145
146    /// Every record of type `t` the station holds that verifies, and how many
147    /// it returned that did not.
148    pub async fn find_records_by_type(
149        &self,
150        t: RecordType,
151    ) -> Result<(Vec<Verified>, usize), LinkError> {
152        let args = dht_args("type", Value::Int(i128::from(t.0)));
153        self.verified_list("_dht.find_records_by_type", args).await
154    }
155
156    async fn verified_list(
157        &self,
158        procedure: &str,
159        args: Value,
160    ) -> Result<(Vec<Verified>, usize), LinkError> {
161        let DhtReply::Records(records) = self.dht_call(procedure, args).await? else {
162            return Err(ReplyRefusal::Malformed.into());
163        };
164        Ok(verified_records(&records, self.profile(), now_ms()))
165    }
166
167    async fn dht_call(&self, procedure: &str, args: Value) -> Result<DhtReply, LinkError> {
168        let reply = self
169            .call(Call {
170                procedure: procedure.to_string(),
171                payload: args,
172                ..Call::default()
173            })
174            .await?;
175        Ok(read_reply(procedure, &reply)?)
176    }
177}
178
179/// The records that verify as a reader verifies them, and how many did not.
180fn verified_records(records: &[Vec<u8>], profile: Profile, now: i64) -> (Vec<Verified>, usize) {
181    let verified: Vec<Verified> = records
182        .iter()
183        .filter_map(|wire| record::verify(wire, profile, now).ok())
184        .collect();
185    let dropped = records.len() - verified.len();
186    (verified, dropped)
187}
188
189#[cfg(test)]
190mod tests {
191    //! The wire 1.0 `_dht.*` vectors (macula#102): the argument forms this
192    //! crate sends are the accepted ones, byte for byte, and each reply reads
193    //! as a caller reads it.
194
195    use super::*;
196    use crate::cbor;
197    use crate::wire_vectors::{self, Case};
198
199    /// The argument form this crate sends for an accepted case's call.
200    fn sent_args(c: &Case, args: &Value) -> Value {
201        match c.text("procedure") {
202            "_dht.put_record" => dht_args("record", args.get("record").unwrap().clone()),
203            "_dht.find_records_by_type" => dht_args("type", args.get("type").unwrap().clone()),
204            _ => dht_args("key", args.get("key").unwrap().clone()),
205        }
206    }
207
208    fn reply_verdict(c: &Case) -> (String, String, Option<usize>) {
209        let reply = cbor::decode(&c.hex("reply")).unwrap();
210        let profile = Profile::parse("pq_pure").unwrap();
211        match read_reply(c.text("procedure"), &reply) {
212            Err(ReplyRefusal::VersionUnsupported) => {
213                ("refuse".into(), "reply_version_unsupported".into(), None)
214            }
215            Err(ReplyRefusal::Malformed) => ("refuse".into(), "malformed_reply".into(), None),
216            Ok(DhtReply::Stored) => ("accept".into(), "ok".into(), None),
217            Ok(DhtReply::Refused(_)) => ("accept".into(), "refused".into(), None),
218            Ok(DhtReply::Record(_)) => ("accept".into(), "record".into(), None),
219            Ok(DhtReply::NotFound) => ("accept".into(), "not_found".into(), None),
220            Ok(DhtReply::Records(records)) => {
221                let kept = verified_records(&records, profile, c.int("now_ms").unwrap())
222                    .0
223                    .len();
224                ("accept".into(), "records".into(), Some(kept))
225            }
226        }
227    }
228
229    #[test]
230    fn the_wire_1_0_dht_vectors() {
231        let file = wire_vectors::load("dht_procedures");
232        let accepted_args = file
233            .cases
234            .iter()
235            .filter(|c| c.has("args") && c.outcome() == "accept");
236        for c in accepted_args {
237            let args = cbor::decode(&c.hex("args")).unwrap();
238            assert_eq!(
239                cbor::encode(&sent_args(c, &args)).unwrap(),
240                c.hex("args"),
241                "{}",
242                c.id
243            );
244        }
245        for c in file.cases.iter().filter(|c| c.has("reply")) {
246            let (outcome, said, kept) = reply_verdict(c);
247            let want = match c.outcome() {
248                "accept" => c.expect_text("result"),
249                _ => c.reason(),
250            };
251            assert_eq!(
252                (outcome.as_str(), said.as_str()),
253                (c.outcome(), want),
254                "{}",
255                c.id
256            );
257            let want_kept = c.expect_int("records_kept").map(|n| n as usize);
258            assert!(
259                want_kept.is_none() || kept == want_kept,
260                "{}: records kept",
261                c.id
262            );
263        }
264    }
265}
macula_rust/station_link/dht.rs

macula_rust/station_link/
dht.rs
