krabka_remote_storage/diskless/capture/
authentication.rs

1//! Capture signing and external pinned-head authentication using WORM manifests.
2
3use {
4    super::{CAPTURE_HEAD_NAME, CAPTURE_STATE_KEY, DisklessWalCapture},
5    crate::worm::{
6        ChainHead, ChainStamp, EpochId, HexBytes, MANIFEST_FORMAT_VERSION, ManifestBody,
7        ManifestSeq, ManifestSignature, ObjectEntry, SegmentIdentity, SegmentManifest,
8        Sha256Digest, TrustedManifestKeys, manifest_head, manifest_signing_bytes,
9        verify_manifest_signature,
10    },
11    krabka_audit::FileEd25519Signer,
12    std::collections::{BTreeMap, BTreeSet},
13    uuid::Uuid,
14};
15
16impl DisklessWalCapture {
17    fn unsigned_bytes(&self) -> Result<Vec<u8>, String> {
18        let mut unsigned = self.clone();
19        unsigned.authentication = None;
20        serde_json::to_vec(&unsigned).map_err(|error| error.to_string())
21    }
22
23    /// Sign the exact capture state and referenced WAL object claims.
24    ///
25    /// # Errors
26    /// Returns an error when capture state is invalid, object claims do not exactly cover it,
27    /// or its deterministic encoding fails.
28    pub fn seal(
29        &mut self,
30        mut objects: Vec<ObjectEntry>,
31        signer: &FileEd25519Signer,
32    ) -> Result<ChainHead, String> {
33        self.validate()?;
34        let expected = self
35            .partitions
36            .iter()
37            .flat_map(|partition| &partition.ranges)
38            .map(|range| range.object_key.clone())
39            .collect::<BTreeSet<_>>();
40        objects.sort_by(|a, b| a.key.cmp(&b.key));
41        let actual = objects
42            .iter()
43            .map(|object| object.key.clone())
44            .collect::<BTreeSet<_>>();
45        if expected != actual || objects.len() != actual.len() {
46            return Err("signed WAL claims do not exactly cover capture references".to_owned());
47        }
48        let state = self.unsigned_bytes()?;
49        let state_digest = Sha256Digest::of(&state);
50        objects.push(ObjectEntry {
51            suffix: ".capture-state".to_owned(),
52            key: CAPTURE_STATE_KEY.to_owned(),
53            size_bytes: state.len() as u64,
54            sha256: state_digest,
55            e_tag: None,
56            version_id: None,
57            create_precondition: false,
58        });
59        let mut segment_id = [0; 16];
60        segment_id.copy_from_slice(&state_digest.0[..16]);
61        let identity = SegmentIdentity {
62            topic: CAPTURE_HEAD_NAME.to_owned(),
63            topic_id: Uuid::nil(),
64            partition: 0,
65            segment_id: Uuid::from_bytes(segment_id),
66            start_offset: 0,
67            end_offset: 0,
68            max_timestamp_ms: i64::try_from(self.captured_at_ms).unwrap_or(i64::MAX),
69            broker_id: -1,
70            event_timestamp_ms: i64::try_from(self.captured_at_ms).unwrap_or(i64::MAX),
71            segment_size_bytes: objects
72                .iter()
73                .map(|object| object.size_bytes)
74                .sum::<u64>()
75                .try_into()
76                .unwrap_or(i64::MAX),
77            leader_epochs: BTreeMap::new(),
78            txn_index_empty: true,
79        };
80        let chain = ChainStamp {
81            epoch_id: EpochId(Uuid::nil()),
82            seq: ManifestSeq(0),
83            prev_head: ChainHead::GENESIS,
84        };
85        let body = ManifestBody {
86            format_version: MANIFEST_FORMAT_VERSION,
87            segment: identity,
88            objects,
89            chain,
90        };
91        let head = manifest_head(&body);
92        let message = manifest_signing_bytes(signer.key_id(), chain.epoch_id, chain.seq, head);
93        self.authentication = Some(SegmentManifest {
94            body,
95            signature: Some(ManifestSignature {
96                key_id: signer.key_id().to_owned(),
97                public_key: HexBytes(signer.public_key()),
98                signature: HexBytes(signer.sign(&message)),
99            }),
100        });
101        Ok(head)
102    }
103
104    /// Verify the embedded signed boundary and return trusted WAL claims.
105    ///
106    /// # Errors
107    /// Returns an error for invalid capture state, missing or untrusted evidence, a bad
108    /// signature, a head mismatch, or incomplete object claims.
109    pub fn authenticate(
110        &self,
111        trusted: &TrustedManifestKeys,
112        expected_head: Option<&str>,
113    ) -> Result<BTreeMap<String, ObjectEntry>, String> {
114        self.validate()?;
115        let manifest = self
116            .authentication
117            .as_ref()
118            .ok_or_else(|| "diskless capture has no signed WORM boundary".to_owned())?;
119        if manifest.body.format_version != MANIFEST_FORMAT_VERSION {
120            return Err("unsupported diskless capture manifest version".to_owned());
121        }
122        let signature = manifest
123            .signature
124            .as_ref()
125            .ok_or_else(|| "diskless capture WORM boundary is unsigned".to_owned())?;
126        let key = trusted
127            .get(&signature.key_id)
128            .ok_or_else(|| format!("diskless capture uses untrusted key `{}`", signature.key_id))?;
129        if !verify_manifest_signature(manifest, key) {
130            return Err("diskless capture WORM signature is invalid".to_owned());
131        }
132        let head = manifest_head(&manifest.body).to_string();
133        let expected_head = expected_head.ok_or_else(|| {
134            format!("trusted diskless restore requires --worm-expect-head {CAPTURE_HEAD_NAME}=HEX")
135        })?;
136        if !head.eq_ignore_ascii_case(expected_head) {
137            return Err(format!(
138                "pinned diskless capture head is {expected_head}, authenticated head is {head}"
139            ));
140        }
141        let state = self.unsigned_bytes()?;
142        let mut claims = BTreeMap::new();
143        let mut state_claim = None;
144        for object in &manifest.body.objects {
145            if object.key == CAPTURE_STATE_KEY {
146                if state_claim.replace(object).is_some() {
147                    return Err("duplicate signed diskless capture-state claim".to_owned());
148                }
149            } else if claims.insert(object.key.clone(), object.clone()).is_some() {
150                return Err(format!("duplicate signed WAL claim `{}`", object.key));
151            }
152        }
153        let state_claim = state_claim
154            .ok_or_else(|| "signed diskless capture-state claim is missing".to_owned())?;
155        if state_claim.size_bytes != state.len() as u64
156            || state_claim.sha256 != Sha256Digest::of(&state)
157        {
158            return Err("diskless capture state changed after signing".to_owned());
159        }
160        let expected = self
161            .partitions
162            .iter()
163            .flat_map(|partition| &partition.ranges)
164            .map(|range| range.object_key.as_str())
165            .collect::<BTreeSet<_>>();
166        if expected != claims.keys().map(String::as_str).collect() {
167            return Err("signed WAL claims do not exactly cover capture references".to_owned());
168        }
169        Ok(claims)
170    }
171}
krabka_remote_storage/diskless/capture/authentication.rs

krabka_remote_storage/diskless/capture/
authentication.rs
