haematite_cli/serve/query/
binding.rs

1use std::collections::BTreeMap;
2
3use haematite::tree::Hash;
4use haematite_query::page::Allowance;
5use serde::{Deserialize, Serialize};
6use serde_json::value::RawValue;
7
8use super::Rows;
9use super::canonical;
10use crate::serve::contracts::Caller;
11use crate::serve::storage::Failure;
12
13const DOMAIN: &[u8] = b"haem.query.continuation.binding\0";
14const FORMAT: u32 = 1;
15
16#[derive(Clone, Debug, Serialize)]
17pub struct Engine {
18    pub protocol: u32,
19    pub worker: String,
20    pub engine: String,
21    pub jaq_core: String,
22    pub jaq_std: String,
23    pub jaq_json: String,
24}
25
26#[derive(Clone, Debug, Deserialize, Serialize)]
27#[serde(deny_unknown_fields)]
28struct Components {
29    query: String,
30    parameters: String,
31    engine: String,
32    caller: String,
33    allowances: String,
34}
35
36#[derive(Clone, Debug, Deserialize, Serialize)]
37#[serde(deny_unknown_fields)]
38pub struct Binding {
39    components: Components,
40    aggregate: String,
41}
42
43#[derive(Serialize)]
44struct PageLimits {
45    rows: u64,
46    bytes: u64,
47    scan: u64,
48    /// What each row carries: a token of one form is not one of the other.
49    form: Rows,
50}
51
52fn invalid() -> Failure {
53    Failure::new(
54        "query_continuation_invalid",
55        "continuation binding is not a canonical digest",
56    )
57}
58
59fn digest(kind: &str, canonical: &[u8]) -> String {
60    let mut bytes = DOMAIN.to_vec();
61    bytes.extend_from_slice(&FORMAT.to_be_bytes());
62    bytes.extend_from_slice(kind.as_bytes());
63    bytes.push(0);
64    bytes.extend_from_slice(canonical);
65    Hash::of(&bytes).to_string()
66}
67
68fn serialized(kind: &str, value: &impl Serialize) -> Result<String, Failure> {
69    let bytes = serde_json::to_vec(value).map_err(|_| invalid())?;
70    Ok(digest(kind, &bytes))
71}
72
73fn canonical_digest(value: &str) -> bool {
74    value.len() == 64
75        && value
76            .bytes()
77            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
78}
79
80impl Binding {
81    /// Bind the caller's parameters; the fixed clock value travels in the token.
82    ///
83    /// # Errors
84    /// Refuses parameters that cannot be canonicalized or components that cannot be serialized for their digests.
85    pub fn new(
86        query: &str,
87        parameters: &BTreeMap<String, Box<RawValue>>,
88        engine: &Engine,
89        caller: &Caller,
90        allowance: &Allowance,
91        form: Rows,
92    ) -> Result<Self, Failure> {
93        let components = Components {
94            query: serialized("query", &query)?,
95            parameters: digest("parameters", &canonical::parameters(parameters)?),
96            engine: serialized("engine", engine)?,
97            caller: serialized("caller", caller)?,
98            allowances: serialized(
99                "allowances",
100                &PageLimits {
101                    rows: allowance.rows.get(),
102                    bytes: allowance.bytes.get(),
103                    scan: allowance.scan_records,
104                    form,
105                },
106            )?,
107        };
108        let aggregate = serialized("aggregate", &components)?;
109        Ok(Self {
110            components,
111            aggregate,
112        })
113    }
114
115    /// Validate all component digests and their aggregate.
116    ///
117    /// # Errors
118    /// Returns `query_continuation_invalid` for a non-canonical digest or an inconsistent aggregate.
119    pub fn validate(&self) -> Result<(), Failure> {
120        let components = &self.components;
121        if [
122            &components.query,
123            &components.parameters,
124            &components.engine,
125            &components.caller,
126            &components.allowances,
127            &self.aggregate,
128        ]
129        .into_iter()
130        .any(|value| !canonical_digest(value))
131            || serialized("aggregate", components)? != self.aggregate
132        {
133            return Err(invalid());
134        }
135        Ok(())
136    }
137
138    /// Check that this binding belongs to the expected request context.
139    ///
140    /// # Errors
141    /// Refuses invalid bindings or names the query, parameters, engine, caller or allowances that changed.
142    pub fn check(&self, expected: &Self) -> Result<(), Failure> {
143        self.validate()?;
144        expected.validate()?;
145        let actual = &self.components;
146        let expected = &expected.components;
147        for (kind, actual, expected) in [
148            ("query", &actual.query, &expected.query),
149            ("parameters", &actual.parameters, &expected.parameters),
150            ("engine", &actual.engine, &expected.engine),
151            ("caller", &actual.caller, &expected.caller),
152            ("allowances", &actual.allowances, &expected.allowances),
153        ] {
154            if actual != expected {
155                return Err(Failure::new(
156                    &format!("query_continuation_{kind}"),
157                    format!("continuation is bound to different {kind}"),
158                ));
159            }
160        }
161        Ok(())
162    }
163}
haematite_cli/serve/query/binding.rs

haematite_cli/serve/query/
binding.rs
