reddb_server/runtime/
signed_writes_kind.rs

1//! Issue #522 — runtime wiring for `CREATE COLLECTION ... SIGNED_BY (...)`
2//! collections.
3//!
4//! The pure logic — registry + verify_insert + error taxonomy — lives in
5//! [`crate::storage::signed_writes`]. This module is the thin adapter
6//! that:
7//!
8//! 1. Persists the per-collection signer registry on the existing
9//!    `red_config` config tree under
10//!    `red.collection.{name}.signed_writes.*` so it survives restarts.
11//! 2. Loads the registry on demand for the INSERT-time verification
12//!    path and the `ALTER COLLECTION ... ADD|REVOKE SIGNER` executor.
13//! 3. Builds the canonical bytes the client must have signed by
14//!    reusing the engine's existing canonical-payload encoding
15//!    ([`super::blockchain_kind::canonical_payload`]) with the two
16//!    signed-writes reserved columns stripped — same encoding the
17//!    blockchain hash binds, so no new on-the-wire spec is introduced.
18
19use crate::storage::signed_writes::{
20    verify_insert, InsertSignatureFields, SignedWriteError, SignerHistoryAction,
21    SignerHistoryEntry, SignerRegistry, RESERVED_SIGNATURE_COL, RESERVED_SIGNER_PUBKEY_COL,
22    SIGNATURE_LEN, SIGNER_PUBKEY_LEN,
23};
24use crate::storage::unified::UnifiedStore;
25use crate::RedDBResult;
26use reddb_types::Value;
27
28use std::time::{SystemTime, UNIX_EPOCH};
29
30/// Marker stored at `red.collection.{name}.signed_writes.enabled = true`
31/// when the collection was created with a non-empty `SIGNED_BY` list.
32const ENABLED_SUFFIX: &str = "signed_writes.enabled";
33
34/// Single JSON-encoded array of currently-allowed Ed25519 pubkeys
35/// (lowercase hex). Stored as one text value to keep the read path one
36/// `get_config` hop instead of a tree scan.
37const ALLOWED_SUFFIX: &str = "signed_writes.allowed_json";
38
39/// Single JSON-encoded array of [`SignerHistoryEntry`] records.
40/// Append-only; revoke pushes a `Revoke` entry rather than rewriting the
41/// `Add` row, so the audit trail is preserved across registry
42/// mutations.
43const HISTORY_SUFFIX: &str = "signed_writes.history_json";
44
45fn key(name: &str, suffix: &str) -> String {
46    format!("red.collection.{name}.{suffix}")
47}
48
49fn now_ms() -> u128 {
50    SystemTime::now()
51        .duration_since(UNIX_EPOCH)
52        .map(|d| d.as_millis())
53        .unwrap_or(0)
54}
55
56fn hex_encode(bytes: &[u8]) -> String {
57    let mut s = String::with_capacity(bytes.len() * 2);
58    for b in bytes {
59        s.push_str(&format!("{b:02x}"));
60    }
61    s
62}
63
64fn hex_decode_32(s: &str) -> Option<[u8; SIGNER_PUBKEY_LEN]> {
65    if s.len() != SIGNER_PUBKEY_LEN * 2 {
66        return None;
67    }
68    let mut out = [0u8; SIGNER_PUBKEY_LEN];
69    for i in 0..SIGNER_PUBKEY_LEN {
70        out[i] = u8::from_str_radix(&s[i * 2..i * 2 + 2], 16).ok()?;
71    }
72    Some(out)
73}
74
75fn action_str(a: SignerHistoryAction) -> &'static str {
76    match a {
77        SignerHistoryAction::Add => "add",
78        SignerHistoryAction::Revoke => "revoke",
79    }
80}
81
82fn action_from_str(s: &str) -> Option<SignerHistoryAction> {
83    match s {
84        "add" => Some(SignerHistoryAction::Add),
85        "revoke" => Some(SignerHistoryAction::Revoke),
86        _ => None,
87    }
88}
89
90fn entry_to_json(e: &SignerHistoryEntry) -> crate::serde_json::Value {
91    let mut obj = crate::serde_json::Map::new();
92    obj.insert(
93        "action".to_string(),
94        crate::serde_json::Value::String(action_str(e.action).to_string()),
95    );
96    obj.insert(
97        "pubkey".to_string(),
98        crate::serde_json::Value::String(hex_encode(&e.pubkey)),
99    );
100    obj.insert(
101        "actor".to_string(),
102        crate::serde_json::Value::String(e.actor.clone()),
103    );
104    obj.insert(
105        "ts_unix_ms".to_string(),
106        crate::serde_json::Value::Number(e.ts_unix_ms as f64),
107    );
108    crate::serde_json::Value::Object(obj)
109}
110
111fn entry_from_json(v: &crate::serde_json::Value) -> Option<SignerHistoryEntry> {
112    let obj = v.as_object()?;
113    let action = action_from_str(obj.get("action")?.as_str()?)?;
114    let pubkey = hex_decode_32(obj.get("pubkey")?.as_str()?)?;
115    let actor = obj.get("actor")?.as_str()?.to_string();
116    let ts_unix_ms = obj.get("ts_unix_ms")?.as_u64()? as u128;
117    Some(SignerHistoryEntry {
118        action,
119        pubkey,
120        actor,
121        ts_unix_ms,
122    })
123}
124
125/// Returns true if `CREATE COLLECTION ... SIGNED_BY (...)` was issued
126/// (or `ALTER COLLECTION ... ADD SIGNER` has been used to enable the
127/// registry) and there is at least a marker in `red_config`.
128pub fn is_signed(store: &UnifiedStore, collection: &str) -> bool {
129    // Compatibility probe: failed policy reads require verification. Runtime
130    // decisions use the fallible API and report the actual storage failure.
131    try_is_signed(store, collection).unwrap_or(true)
132}
133
134pub fn try_is_signed(store: &UnifiedStore, collection: &str) -> RedDBResult<bool> {
135    Ok(matches!(
136        store.try_get_config(&key(collection, ENABLED_SUFFIX))?,
137        Some(Value::Boolean(true)) | Some(Value::Text(_))
138    ))
139}
140
141/// Compatibility wrapper. Authoritative DDL uses `try_install` so failed
142/// persistence is returned to the caller.
143pub fn install(
144    store: &UnifiedStore,
145    collection: &str,
146    initial: &[[u8; SIGNER_PUBKEY_LEN]],
147    actor: &str,
148) {
149    if let Err(error) = try_install(store, collection, initial, actor) {
150        tracing::error!(collection, %error, "failed to install signer registry");
151    }
152}
153
154/// Persist allowed signers and history before publishing their enable marker.
155pub fn try_install(
156    store: &UnifiedStore,
157    collection: &str,
158    initial: &[[u8; SIGNER_PUBKEY_LEN]],
159    actor: &str,
160) -> RedDBResult<()> {
161    let _publication = store.vault_publication_lock.lock();
162    if try_is_signed(store, collection)? {
163        return Ok(());
164    }
165    let reg = SignerRegistry::from_initial(initial, actor.to_string(), now_ms());
166    write_registry(store, collection, &reg)?;
167    store.try_set_config_tree(
168        &key(collection, ENABLED_SUFFIX),
169        &crate::serde_json::Value::Bool(true),
170    )?;
171    Ok(())
172}
173
174/// Serialise registry state into the config tree. Overwrites any prior
175/// value at the same key — the store treats `red_config` as
176/// insert-only, but the read path returns the most recent matching row.
177fn write_registry(store: &UnifiedStore, collection: &str, reg: &SignerRegistry) -> RedDBResult<()> {
178    let allowed: Vec<crate::serde_json::Value> = reg
179        .allowed()
180        .map(|pk| crate::serde_json::Value::String(hex_encode(pk)))
181        .collect();
182    let history: Vec<crate::serde_json::Value> = reg.history().iter().map(entry_to_json).collect();
183    store.try_set_config_tree(
184        &key(collection, ALLOWED_SUFFIX),
185        &crate::serde_json::Value::String(crate::serde_json::Value::Array(allowed).to_string()),
186    )?;
187    store.try_set_config_tree(
188        &key(collection, HISTORY_SUFFIX),
189        &crate::serde_json::Value::String(crate::serde_json::Value::Array(history).to_string()),
190    )?;
191    Ok(())
192}
193
194fn read_registry(store: &UnifiedStore, collection: &str) -> RedDBResult<SignerRegistry> {
195    let _publication = store.vault_publication_lock.lock();
196    let allowed_json = match store.try_get_config(&key(collection, ALLOWED_SUFFIX))? {
197        Some(Value::Text(s)) => s.to_string(),
198        _ => "[]".to_string(),
199    };
200    let history_json = match store.try_get_config(&key(collection, HISTORY_SUFFIX))? {
201        Some(Value::Text(s)) => s.to_string(),
202        _ => "[]".to_string(),
203    };
204    let parsed_allowed: Vec<[u8; SIGNER_PUBKEY_LEN]> =
205        match crate::utils::json::parse_json(&allowed_json) {
206            Ok(v) => match crate::serde_json::Value::from(v) {
207                crate::serde_json::Value::Array(arr) => arr
208                    .iter()
209                    .filter_map(|v| v.as_str().and_then(hex_decode_32))
210                    .collect(),
211                _ => Vec::new(),
212            },
213            Err(_) => Vec::new(),
214        };
215    let parsed_history: Vec<SignerHistoryEntry> =
216        match crate::utils::json::parse_json(&history_json) {
217            Ok(v) => match crate::serde_json::Value::from(v) {
218                crate::serde_json::Value::Array(arr) => {
219                    arr.iter().filter_map(entry_from_json).collect()
220                }
221                _ => Vec::new(),
222            },
223            Err(_) => Vec::new(),
224        };
225    Ok(SignerRegistry::from_persisted_parts(
226        parsed_allowed,
227        parsed_history,
228    ))
229}
230
231/// Load the current registry. Cheap enough for the INSERT hot path:
232/// two `red_config` reads + two JSON parses, no scan of the parent
233/// collection.
234pub fn registry(store: &UnifiedStore, collection: &str) -> SignerRegistry {
235    // An empty compatibility result denies every signer; it never authorizes
236    // an unsigned write. Runtime verification propagates failures via try_registry.
237    try_registry(store, collection)
238        .unwrap_or_else(|_| SignerRegistry::from_persisted_parts(Vec::new(), Vec::new()))
239}
240
241pub fn try_registry(store: &UnifiedStore, collection: &str) -> RedDBResult<SignerRegistry> {
242    read_registry(store, collection)
243}
244
245pub fn add_signer(
246    store: &UnifiedStore,
247    collection: &str,
248    pubkey: [u8; SIGNER_PUBKEY_LEN],
249    actor: &str,
250) -> bool {
251    try_add_signer(store, collection, pubkey, actor).unwrap_or_else(|error| {
252        tracing::error!(collection, %error, "failed to add signer");
253        false
254    })
255}
256
257pub fn try_add_signer(
258    store: &UnifiedStore,
259    collection: &str,
260    pubkey: [u8; SIGNER_PUBKEY_LEN],
261    actor: &str,
262) -> RedDBResult<bool> {
263    let _publication = store.vault_publication_lock.lock();
264    let mut reg = read_registry(store, collection)?;
265    let changed = reg.add_signer(pubkey, actor.to_string(), now_ms());
266    if changed {
267        write_registry(store, collection, &reg)?;
268    }
269    Ok(changed)
270}
271
272pub fn revoke_signer(
273    store: &UnifiedStore,
274    collection: &str,
275    pubkey: &[u8; SIGNER_PUBKEY_LEN],
276    actor: &str,
277) -> bool {
278    try_revoke_signer(store, collection, pubkey, actor).unwrap_or_else(|error| {
279        tracing::error!(collection, %error, "failed to revoke signer");
280        false
281    })
282}
283
284pub fn try_revoke_signer(
285    store: &UnifiedStore,
286    collection: &str,
287    pubkey: &[u8; SIGNER_PUBKEY_LEN],
288    actor: &str,
289) -> RedDBResult<bool> {
290    let _publication = store.vault_publication_lock.lock();
291    let mut reg = read_registry(store, collection)?;
292    let changed = reg.revoke_signer(pubkey, actor.to_string(), now_ms());
293    if changed {
294        write_registry(store, collection, &reg)?;
295    }
296    Ok(changed)
297}
298
299/// Reserved column set automatically present on every signed-writes
300/// collection. Filtered out of the canonical-payload bytes the client
301/// signs.
302pub const RESERVED_COLUMNS: &[&str] = &[RESERVED_SIGNER_PUBKEY_COL, RESERVED_SIGNATURE_COL];
303
304/// Pulled-apart signer / signature reserved columns. Carries:
305///
306/// * The user's original `Value` (for round-trip storage so SELECT and
307///   `WHERE signer_pubkey = '<hex>'` predicates compare against the
308///   same encoding the caller supplied — Text-typed hex on the JSON /
309///   SQL path, Blob on the binary protobuf path).
310/// * The decoded raw bytes used to verify the Ed25519 signature.
311pub struct SignerColumn {
312    pub raw_value: Value,
313    pub bytes: Vec<u8>,
314}
315
316/// Pull the `signer_pubkey` and `signature` values out of the row's
317/// fields. Returns the parsed reserved columns + the residual field
318/// list (fields stripped of the two reserved columns) — the residual
319/// goes into the canonical payload.
320pub fn split_signature_fields(
321    fields: Vec<(String, Value)>,
322) -> (
323    Option<SignerColumn>,
324    Option<SignerColumn>,
325    Vec<(String, Value)>,
326) {
327    let mut pubkey: Option<SignerColumn> = None;
328    let mut signature: Option<SignerColumn> = None;
329    let mut residual: Vec<(String, Value)> = Vec::with_capacity(fields.len());
330    for (k, v) in fields {
331        if k == RESERVED_SIGNER_PUBKEY_COL {
332            let bytes = match &v {
333                Value::Blob(b) => Some(b.clone()),
334                // Accept hex-encoded pubkey from JSON / SQL callers
335                // that can't easily express literal blobs.
336                Value::Text(s) => decode_hex(s.as_ref()),
337                _ => None,
338            };
339            if let Some(bytes) = bytes {
340                pubkey = Some(SignerColumn {
341                    raw_value: v,
342                    bytes,
343                });
344            }
345            continue;
346        }
347        if k == RESERVED_SIGNATURE_COL {
348            let bytes = match &v {
349                Value::Blob(b) => Some(b.clone()),
350                Value::Text(s) => decode_hex(s.as_ref()),
351                _ => None,
352            };
353            if let Some(bytes) = bytes {
354                signature = Some(SignerColumn {
355                    raw_value: v,
356                    bytes,
357                });
358            }
359            continue;
360        }
361        residual.push((k, v));
362    }
363    (pubkey, signature, residual)
364}
365
366fn decode_hex(s: &str) -> Option<Vec<u8>> {
367    if !s.len().is_multiple_of(2) {
368        return None;
369    }
370    let mut out = Vec::with_capacity(s.len() / 2);
371    for i in (0..s.len()).step_by(2) {
372        out.push(u8::from_str_radix(&s[i..i + 2], 16).ok()?);
373    }
374    Some(out)
375}
376
377/// Top-level INSERT-time check used by the engine. Computes the
378/// canonical payload from the (already reserved-column-stripped)
379/// residual fields and dispatches to [`verify_insert`].
380pub fn verify_row(
381    registry: &SignerRegistry,
382    signer_pubkey: Option<&[u8]>,
383    signature: Option<&[u8]>,
384    canonical_payload: &[u8],
385) -> Result<(), SignedWriteError> {
386    verify_insert(
387        registry,
388        &InsertSignatureFields {
389            signer_pubkey,
390            signature,
391        },
392        canonical_payload,
393    )
394}
395
396/// Map a [`SignedWriteError`] onto a [`RedDBError`] whose marker prefix
397/// is picked up by the transport-layer status mapper.
398///
399/// | Variant                  | Prefix                                     | HTTP |
400/// |--------------------------|--------------------------------------------|------|
401/// | `MissingSignatureFields` | `SignedWriteError:MissingSignatureFields:` | 400  |
402/// | `MalformedSignerPubkey`  | `SignedWriteError:MalformedSignerPubkey`   | 400  |
403/// | `MalformedSignature`     | `SignedWriteError:MalformedSignature`      | 400  |
404/// | `UnknownSigner`          | `SignedWriteError:UnknownSigner`           | 401  |
405/// | `RevokedSigner`          | `SignedWriteError:RevokedSigner`           | 401  |
406/// | `InvalidSignature`       | `SignedWriteError:InvalidSignature`        | 401  |
407pub fn map_error(err: SignedWriteError) -> crate::api::RedDBError {
408    let body = match &err {
409        SignedWriteError::MissingSignatureFields { fields } => {
410            format!(
411                "SignedWriteError:MissingSignatureFields:{}",
412                fields.join(",")
413            )
414        }
415        SignedWriteError::UnknownSigner { pubkey } => {
416            format!("SignedWriteError:UnknownSigner:{}", hex_encode(pubkey))
417        }
418        SignedWriteError::RevokedSigner { pubkey } => {
419            format!("SignedWriteError:RevokedSigner:{}", hex_encode(pubkey))
420        }
421        SignedWriteError::InvalidSignature => "SignedWriteError:InvalidSignature".to_string(),
422        SignedWriteError::MalformedSignerPubkey => {
423            "SignedWriteError:MalformedSignerPubkey".to_string()
424        }
425        SignedWriteError::MalformedSignature => "SignedWriteError:MalformedSignature".to_string(),
426    };
427    crate::api::RedDBError::InvalidOperation(body)
428}
429
430/// Length sanity: a signature blob must be exactly 64 bytes. Surfaced
431/// to the caller so it can return `MalformedSignature` before computing
432/// the canonical payload.
433pub const SIGNATURE_BYTES: usize = SIGNATURE_LEN;
434
435#[cfg(test)]
436mod tests {
437    use super::*;
438    use ed25519_dalek::{Signer, SigningKey};
439
440    fn signing_key(seed: u8) -> SigningKey {
441        SigningKey::from_bytes(&[seed; 32])
442    }
443
444    fn pubkey_of(sk: &SigningKey) -> [u8; SIGNER_PUBKEY_LEN] {
445        sk.verifying_key().to_bytes()
446    }
447
448    fn make_store() -> UnifiedStore {
449        UnifiedStore::new()
450    }
451
452    #[test]
453    fn install_and_read_roundtrip_preserves_registry() {
454        let store = make_store();
455        let pk1 = pubkey_of(&signing_key(1));
456        let pk2 = pubkey_of(&signing_key(2));
457        install(&store, "sc", &[pk1, pk2], "@system/create");
458        assert!(is_signed(&store, "sc"));
459        let reg = registry(&store, "sc");
460        assert_eq!(reg.allowed_len(), 2);
461        assert!(reg.is_allowed(&pk1));
462        assert!(reg.is_allowed(&pk2));
463        assert_eq!(reg.history().len(), 2);
464    }
465
466    #[test]
467    fn add_signer_persists_and_records_history() {
468        let store = make_store();
469        let pk1 = pubkey_of(&signing_key(1));
470        install(&store, "sc", &[pk1], "@system/create");
471        let pk2 = pubkey_of(&signing_key(2));
472        assert!(add_signer(&store, "sc", pk2, "admin:alice"));
473        // Idempotent re-add returns false.
474        assert!(!add_signer(&store, "sc", pk2, "admin:alice"));
475        let reg = registry(&store, "sc");
476        assert!(reg.is_allowed(&pk2));
477        assert_eq!(reg.history().len(), 2);
478        let last = reg.history().last().unwrap();
479        assert_eq!(last.action, SignerHistoryAction::Add);
480        assert_eq!(last.actor, "admin:alice");
481    }
482
483    #[test]
484    fn revoke_signer_blocks_future_inserts_but_history_preserved() {
485        let store = make_store();
486        let sk = signing_key(7);
487        let pk = pubkey_of(&sk);
488        install(&store, "sc", &[pk], "@system/create");
489        assert!(revoke_signer(&store, "sc", &pk, "admin:bob"));
490        let reg = registry(&store, "sc");
491        assert!(!reg.is_allowed(&pk));
492        assert!(reg.ever_added(&pk));
493        let last = reg.history().last().unwrap();
494        assert_eq!(last.action, SignerHistoryAction::Revoke);
495        assert_eq!(last.actor, "admin:bob");
496    }
497
498    #[test]
499    fn split_signature_fields_extracts_blob_columns() {
500        let fields = vec![
501            ("name".to_string(), Value::text("alice".to_string())),
502            (
503                RESERVED_SIGNER_PUBKEY_COL.to_string(),
504                Value::Blob(vec![0x11; 32]),
505            ),
506            (
507                RESERVED_SIGNATURE_COL.to_string(),
508                Value::Blob(vec![0x22; 64]),
509            ),
510        ];
511        let (pk, sig, residual) = split_signature_fields(fields);
512        assert_eq!(pk.as_ref().unwrap().bytes.len(), 32);
513        assert!(matches!(pk.unwrap().raw_value, Value::Blob(_)));
514        assert_eq!(sig.as_ref().unwrap().bytes.len(), 64);
515        assert!(matches!(sig.unwrap().raw_value, Value::Blob(_)));
516        assert_eq!(residual.len(), 1);
517        assert_eq!(residual[0].0, "name");
518    }
519
520    #[test]
521    fn split_signature_fields_accepts_hex_text() {
522        let pk_hex = "11".repeat(32);
523        let sig_hex = "22".repeat(64);
524        let fields = vec![
525            (RESERVED_SIGNER_PUBKEY_COL.to_string(), Value::text(pk_hex)),
526            (RESERVED_SIGNATURE_COL.to_string(), Value::text(sig_hex)),
527        ];
528        let (pk, sig, residual) = split_signature_fields(fields);
529        assert_eq!(pk.as_ref().unwrap().bytes, vec![0x11; 32]);
530        assert!(matches!(pk.unwrap().raw_value, Value::Text(_)));
531        assert_eq!(sig.as_ref().unwrap().bytes, vec![0x22; 64]);
532        assert!(matches!(sig.unwrap().raw_value, Value::Text(_)));
533        assert!(residual.is_empty());
534    }
535
536    #[test]
537    fn map_error_carries_variant_prefix() {
538        let pk = [0u8; SIGNER_PUBKEY_LEN];
539        match map_error(SignedWriteError::UnknownSigner { pubkey: pk }) {
540            crate::api::RedDBError::InvalidOperation(s) => {
541                assert!(s.starts_with("SignedWriteError:UnknownSigner"));
542            }
543            other => panic!("unexpected mapping: {other:?}"),
544        }
545        match map_error(SignedWriteError::InvalidSignature) {
546            crate::api::RedDBError::InvalidOperation(s) => {
547                assert_eq!(s, "SignedWriteError:InvalidSignature");
548            }
549            other => panic!("unexpected mapping: {other:?}"),
550        }
551    }
552
553    #[test]
554    fn verify_row_accepts_valid_signature_over_canonical_payload() {
555        let sk = signing_key(3);
556        let pk = pubkey_of(&sk);
557        let store = make_store();
558        install(&store, "sc", &[pk], "@system/create");
559        let payload = b"hello-world";
560        let sig = sk.sign(payload).to_bytes();
561        let reg = registry(&store, "sc");
562        verify_row(&reg, Some(&pk), Some(&sig), payload).unwrap();
563    }
564
565    #[test]
566    fn verify_row_rejects_tampered_payload() {
567        let sk = signing_key(4);
568        let pk = pubkey_of(&sk);
569        let store = make_store();
570        install(&store, "sc", &[pk], "@system/create");
571        let payload = b"hello-world";
572        let sig = sk.sign(payload).to_bytes();
573        let reg = registry(&store, "sc");
574        let err = verify_row(&reg, Some(&pk), Some(&sig), b"tampered").unwrap_err();
575        assert_eq!(err, SignedWriteError::InvalidSignature);
576    }
577}
reddb_server/runtime/signed_writes_kind.rs

reddb_server/runtime/
signed_writes_kind.rs
