pg_core/client/web/
stream.rs

1//! Streaming mode.
2
3use super::aesgcm::{decrypt, encrypt, get_key};
4
5use crate::artifacts::{PublicKey, SigningKeyExt, UserSecretKey, VerifyingKey};
6use crate::client::*;
7use crate::error::Error;
8use crate::identity::{EncryptionPolicy, Policy};
9use crate::util::preamble_checked;
10use ibs::gg::{Identity, Signature, Signer, Verifier, SIG_BYTES};
11
12use futures::{Sink, SinkExt, Stream, StreamExt};
13use ibe::kem::cgw_kv::CGWKV;
14use js_sys::Uint8Array;
15use rand::{CryptoRng, RngCore};
16use wasm_bindgen::{JsCast, JsValue};
17
18use alloc::string::ToString;
19use alloc::vec::Vec;
20
21/// Configures an [`Sealer`] to process a payload stream.
22#[derive(Debug)]
23pub struct StreamSealerConfig {
24    segment_size: u32,
25    key: [u8; KEY_SIZE],
26    nonce: [u8; STREAM_NONCE_SIZE],
27}
28
29/// Configures an [`Unsealer`] to process a payload stream.
30#[derive(Debug)]
31pub struct StreamUnsealerConfig {
32    segment_size: u32,
33    spill: Vec<u8>,
34}
35
36impl SealerConfig for StreamSealerConfig {}
37impl UnsealerConfig for StreamUnsealerConfig {}
38impl crate::client::sealed::SealerConfig for StreamSealerConfig {}
39impl crate::client::sealed::UnsealerConfig for StreamUnsealerConfig {}
40
41impl<'r, Rng: RngCore + CryptoRng> Sealer<'r, Rng, StreamSealerConfig> {
42    /// Construct a new [`Sealer`] that can process payloads streamingly.
43    pub fn new(
44        pk: &PublicKey<CGWKV>,
45        policies: &EncryptionPolicy,
46        pub_sign_key: &SigningKeyExt,
47        rng: &'r mut Rng,
48    ) -> Result<Self, Error> {
49        let (header, ss) = Header::new(pk, policies, rng)?;
50
51        let (segment_size, _) = stream_mode_checked(&header)?;
52        let Algorithm::Aes128Gcm(iv) = header.algo;
53
54        let mut key = [0u8; KEY_SIZE];
55        let mut nonce = [0u8; STREAM_NONCE_SIZE];
56
57        key.copy_from_slice(&ss.0[..KEY_SIZE]);
58        nonce.copy_from_slice(&iv.0[..STREAM_NONCE_SIZE]);
59
60        Ok(Sealer {
61            rng,
62            header,
63            pub_sign_key: pub_sign_key.clone(),
64            priv_sign_key: None,
65            config: StreamSealerConfig {
66                segment_size,
67                key,
68                nonce,
69            },
70        })
71    }
72
73    /// Seals payload data from a [`Stream`] of [`JsValue`] to a Sink of [`JsValue`].
74    ///
75    /// # Errors
76    ///
77    /// Make sure the [`JsValue`]s *can* dynamically be cast to [`Uint8Array`],
78    /// otherwise this operation *will* error.
79    pub async fn seal<R, W>(mut self, mut r: R, mut w: W) -> Result<(), Error>
80    where
81        R: Stream<Item = Result<JsValue, JsValue>> + Unpin,
82        W: Sink<JsValue, Error = JsValue> + Unpin,
83    {
84        let size_hint = r.size_hint();
85        let new_hint = (size_hint.0 as u64, size_hint.1.map(|x| x as u64));
86
87        self.header = self.header.with_mode(Mode::Streaming {
88            segment_size: self.config.segment_size,
89            size_hint: new_hint,
90        });
91
92        w.feed(Uint8Array::from(&PRELUDE[..]).into()).await?;
93        w.feed(Uint8Array::from(&VERSION_V3.to_be_bytes()[..]).into())
94            .await?;
95
96        let header_vec = crate::bincode_compat::serialize(&self.header)?;
97
98        w.feed(Uint8Array::from(&(header_vec.len() as u32).to_be_bytes()[..]).into())
99            .await?;
100
101        w.feed(Uint8Array::from(&header_vec[..]).into()).await?;
102
103        let mut signer = Signer::default().chain(&header_vec);
104        let header_sig = signer.clone().sign(&self.pub_sign_key.key.0, self.rng);
105        let header_sig_ext = SignatureExt {
106            sig: header_sig,
107            pol: self.pub_sign_key.policy.clone(),
108        };
109        let header_sig_bytes = crate::bincode_compat::serialize(&header_sig_ext)?;
110
111        w.feed(Uint8Array::from(&(header_sig_bytes.len() as u32).to_be_bytes()[..]).into())
112            .await?;
113        w.feed(Uint8Array::from(&header_sig_bytes[..]).into())
114            .await?;
115
116        let key = get_key(&self.config.key).await?;
117
118        // Check for a private signing key, otherwise fall back to the public one.
119        let signing_key = self.priv_sign_key.unwrap_or(self.pub_sign_key);
120
121        let pol_bytes = crate::bincode_compat::serialize(&signing_key.policy)?;
122        let pol_len: u32 = pol_bytes.len() as u32;
123
124        if pol_len + POL_SIZE_SIZE as u32 > self.config.segment_size {
125            return Err(Error::ConstraintViolation.into());
126        }
127
128        let buf = Uint8Array::new_with_length(self.config.segment_size + SIG_BYTES as u32);
129
130        buf.set(
131            &Uint8Array::from(&(pol_len as u32).to_be_bytes()[..]).into(),
132            0,
133        );
134        buf.set(
135            &Uint8Array::from(&pol_bytes[..]).into(),
136            POL_SIZE_SIZE as u32,
137        );
138
139        let mut counter = 0u32;
140        let mut buf_tail: u32 = POL_SIZE_SIZE as u32 + pol_len;
141        let mut start: u32 = buf_tail;
142
143        while let Some(Ok(data)) = r.next().await {
144            let mut array: Uint8Array = data.dyn_into()?;
145
146            while array.byte_length() != 0 {
147                let len = array.byte_length();
148                let rem = self.config.segment_size - buf_tail;
149
150                if len < rem {
151                    buf.set(&array, buf_tail);
152                    array = Uint8Array::new_with_length(0);
153                    buf_tail += len;
154                } else {
155                    buf.set(&array.slice(0, rem), buf_tail);
156                    array = array.slice(rem, len);
157                    buf_tail += rem;
158
159                    signer.update(&buf.slice(start, buf_tail).to_vec());
160                    let sig = signer
161                        .clone()
162                        .chain(&counter.to_be_bytes())
163                        .chain(&[0x00])
164                        .sign(&signing_key.key.0, self.rng);
165                    let sig_bytes = crate::bincode_compat::serialize(&sig)?;
166
167                    buf.set(&Uint8Array::from(&sig_bytes[..]).into(), buf_tail);
168
169                    let ct = encrypt(
170                        &key,
171                        &aead_nonce(&self.config.nonce, counter, false),
172                        &Uint8Array::new_with_length(0),
173                        &buf,
174                    )
175                    .await?;
176
177                    w.feed(ct.into()).await?;
178
179                    counter = counter.checked_add(1).ok_or(Error::Symmetric)?;
180                    buf_tail = 0;
181                    start = 0;
182                }
183            }
184        }
185
186        signer.update(&buf.slice(start, buf_tail).to_vec());
187        let sig = signer
188            .chain(&counter.to_be_bytes())
189            .chain(&[0x01])
190            .sign(&signing_key.key.0, self.rng);
191        let sig_bytes = crate::bincode_compat::serialize(&sig)?;
192
193        buf.set(&Uint8Array::from(&sig_bytes[..]).into(), buf_tail);
194        buf_tail += SIG_BYTES as u32;
195
196        let final_ct = encrypt(
197            &key,
198            &aead_nonce(&self.config.nonce, counter, true),
199            &Uint8Array::new_with_length(0),
200            &buf.slice(0, buf_tail),
201        )
202        .await?;
203
204        w.feed(final_ct.into()).await?;
205
206        w.flush().await?;
207        w.close().await?;
208
209        Ok(())
210    }
211}
212
213// Nonce generation as defined in the STREAM construction.
214fn aead_nonce(nonce: &[u8], counter: u32, last_block: bool) -> [u8; IV_SIZE] {
215    let mut iv = [0u8; IV_SIZE];
216
217    iv[..STREAM_NONCE_SIZE].copy_from_slice(nonce);
218    iv[STREAM_NONCE_SIZE..IV_SIZE - 1].copy_from_slice(&counter.to_be_bytes());
219    iv[IV_SIZE - 1] = last_block as u8;
220
221    iv
222}
223
224async fn read_atleast<R>(mut r: R, buf: &mut [u8], spill: &mut Vec<u8>) -> Result<(), Error>
225where
226    R: Stream<Item = Result<JsValue, JsValue>> + Unpin,
227{
228    let buf_len = buf.len();
229    let spill_len = spill.len();
230
231    if buf_len <= spill_len {
232        buf.copy_from_slice(&spill[..buf_len]);
233        spill.drain(..buf_len);
234
235        Ok(())
236    } else {
237        buf[..spill_len].copy_from_slice(&spill);
238        let mut rem = buf_len - spill_len;
239        spill.clear();
240
241        while let Some(Ok(data)) = r.next().await {
242            let arr: Uint8Array = data.dyn_into()?;
243            let len = arr.byte_length();
244
245            if len as usize >= rem {
246                buf[buf_len - rem..].copy_from_slice(&arr.slice(0, rem as u32).to_vec()[..]);
247                spill.extend_from_slice(&arr.slice(rem as u32, len).to_vec()[..]);
248                rem = 0;
249                break;
250            } else {
251                buf[buf_len - rem..buf_len - rem + len as usize].copy_from_slice(&arr.to_vec()[..]);
252                rem -= len as usize;
253            }
254        }
255
256        if rem == 0 {
257            Ok(())
258        } else {
259            Err(Error::FormatViolation("unexpected EOF".to_string()).into())
260        }
261    }
262}
263
264// Note: It might be easier to work with R: ReadableStream.
265
266impl<R> Unsealer<R, StreamUnsealerConfig>
267where
268    R: Stream<Item = Result<JsValue, JsValue>> + Unpin,
269{
270    /// Create a new [`Unsealer`] that starts reading from a [`Stream<Item = Result<Uint8Array, JsValue>>`][Stream].
271    ///
272    /// # Errors
273    ///
274    /// Errors if the bytestream is not a legitimate PostGuard bytestream.
275    /// Also errors if the items (of type [`JsValue`]) cannot be cast into [`Uint8Array`].
276    pub async fn new(mut r: R, vk: &VerifyingKey) -> Result<Self, Error> {
277        let mut spill = Vec::new();
278
279        let mut preamble = [0u8; PREAMBLE_SIZE];
280        read_atleast(&mut r, &mut preamble, &mut spill).await?;
281        let (version, header_len) = preamble_checked(&preamble)?;
282
283        let mut header_raw = vec![0u8; header_len];
284        read_atleast(&mut r, &mut header_raw, &mut spill).await?;
285
286        let mut h_sig_len_bytes = [0u8; SIG_SIZE_SIZE];
287        read_atleast(&mut r, &mut h_sig_len_bytes, &mut spill).await?;
288        let header_sig_len = u32::from_be_bytes(h_sig_len_bytes);
289
290        let mut header_sig_raw = vec![0u8; header_sig_len as usize];
291        read_atleast(&mut r, &mut header_sig_raw, &mut spill).await?;
292        let h_sig_ext: SignatureExt = crate::bincode_compat::deserialize(&header_sig_raw)?;
293
294        let verifier = Verifier::default().chain(&header_raw);
295        let pub_id = h_sig_ext.pol.derive_ibs()?;
296
297        if !verifier.clone().verify(&vk.0, &h_sig_ext.sig, &pub_id) {
298            return Err(Error::IncorrectSignature.into());
299        }
300
301        let header: Header = crate::bincode_compat::deserialize(&header_raw)?;
302        let (segment_size, _) = stream_mode_checked(&header)?;
303
304        Ok(Unsealer {
305            version,
306            header,
307            pub_id: h_sig_ext.pol,
308            verifier,
309            vk: vk.clone(),
310            r,
311            config: StreamUnsealerConfig {
312                spill,
313                segment_size,
314            },
315        })
316    }
317
318    /// Unseal into an [`Sink<Uint8Array, Error = JsValue>`][Sink].
319    pub async fn unseal<W>(
320        &mut self,
321        ident: &str,
322        usk: &UserSecretKey<CGWKV>,
323        mut w: W,
324    ) -> Result<VerificationResult, Error>
325    where
326        W: Sink<JsValue, Error = JsValue> + Unpin,
327    {
328        let rec_info = self
329            .header
330            .recipients
331            .get(ident)
332            .ok_or_else(|| Error::UnknownIdentifier(ident.to_string()))?;
333
334        let ss = rec_info.decaps(usk)?;
335        let key = get_key(&ss.0[..KEY_SIZE]).await?;
336
337        let Algorithm::Aes128Gcm(iv) = self.header.algo;
338        let nonce = &iv.0[..STREAM_NONCE_SIZE];
339
340        let segment_size: u32 = self.config.segment_size + (SIG_BYTES + TAG_SIZE) as u32;
341
342        let buf = Uint8Array::new_with_length(segment_size);
343        let mut counter = 0u32;
344        let mut buf_tail = 0;
345        let mut pol_id: Option<(Policy, Identity)> = None;
346
347        fn extract_policy(
348            plain: Uint8Array,
349        ) -> Result<(Option<(Policy, Identity)>, Uint8Array), Error> {
350            if plain.byte_length() < POL_SIZE_SIZE as u32 {
351                return Err(Error::FormatViolation(alloc::string::String::from(
352                    "policy length",
353                )));
354            }
355            let pol_len =
356                u32::from_be_bytes(plain.slice(0, POL_SIZE_SIZE as u32).to_vec()[..].try_into()?);
357            let pol_end = (POL_SIZE_SIZE as u32).checked_add(pol_len).ok_or_else(|| {
358                Error::FormatViolation(alloc::string::String::from("policy length overflow"))
359            })?;
360            if plain.byte_length() < pol_end {
361                return Err(Error::FormatViolation(alloc::string::String::from(
362                    "policy truncated",
363                )));
364            }
365            let pol_bytes = plain.slice(POL_SIZE_SIZE as u32, pol_end);
366            let pol: Policy = crate::bincode_compat::deserialize(&pol_bytes.to_vec())?;
367            let id = pol.derive_ibs()?;
368            let new_plain = plain.slice(pol_end, plain.byte_length());
369
370            Ok((Some((pol, id)), new_plain))
371        }
372
373        loop {
374            // First exhaust the spillage, then the rest of the stream.
375            let mut array: Uint8Array = if !self.config.spill.is_empty() {
376                let arr = Uint8Array::from(&self.config.spill[..]);
377                self.config.spill.clear();
378                arr
379            } else if let Some(Ok(data)) = self.r.next().await {
380                data.dyn_into()?
381            } else {
382                break;
383            };
384
385            while array.byte_length() != 0 {
386                let len = array.byte_length();
387                let rem = buf.byte_length() - buf_tail;
388
389                if len < rem {
390                    buf.set(&array, buf_tail);
391                    array = Uint8Array::new_with_length(0);
392                    buf_tail += len;
393                } else {
394                    buf.set(&array.slice(0, rem), buf_tail);
395                    array = array.slice(rem, len);
396
397                    let mut plain = decrypt(
398                        &key,
399                        &aead_nonce(nonce, counter, false),
400                        &Uint8Array::new_with_length(0),
401                        &buf,
402                    )
403                    .await?;
404
405                    if counter == 0 {
406                        (pol_id, plain) = extract_policy(plain)?;
407                    }
408
409                    if plain.byte_length() < SIG_BYTES as u32 {
410                        return Err(Error::FormatViolation(alloc::string::String::from(
411                            "segment too short for signature",
412                        ))
413                        .into());
414                    }
415
416                    let m = plain.slice(0, plain.byte_length() - SIG_BYTES as u32);
417                    let sig =
418                        plain.slice(plain.byte_length() - SIG_BYTES as u32, plain.byte_length());
419                    let sig: Signature = crate::bincode_compat::deserialize(&sig.to_vec())?;
420
421                    self.verifier.update(&m.to_vec());
422
423                    if !self
424                        .verifier
425                        .clone()
426                        .chain(&counter.to_be_bytes())
427                        .chain(&[0x00])
428                        .verify(&self.vk.0, &sig, &pol_id.as_ref().unwrap().1)
429                    {
430                        return Err(Error::IncorrectSignature.into());
431                    }
432
433                    w.feed(m.into()).await?;
434
435                    counter = counter.checked_add(1).ok_or(Error::Symmetric)?;
436                    buf_tail = 0;
437                }
438            }
439        }
440
441        let mut final_plain = decrypt(
442            &key,
443            &aead_nonce(nonce, counter, true),
444            &Uint8Array::new_with_length(0),
445            &buf.slice(0, buf_tail),
446        )
447        .await?;
448
449        if counter == 0 {
450            (pol_id, final_plain) = extract_policy(final_plain)?;
451        }
452
453        if final_plain.byte_length() < SIG_BYTES as u32 {
454            return Err(Error::FormatViolation(alloc::string::String::from(
455                "final segment too short for signature",
456            ))
457            .into());
458        }
459        let m = final_plain.slice(0, final_plain.byte_length() - SIG_BYTES as u32);
460        let sig = final_plain.slice(
461            final_plain.byte_length() - SIG_BYTES as u32,
462            final_plain.byte_length(),
463        );
464
465        let sig: Signature = crate::bincode_compat::deserialize(&sig.to_vec())?;
466        self.verifier.update(&m.to_vec());
467        if !self
468            .verifier
469            .clone()
470            .chain(&counter.to_be_bytes())
471            .chain(&[0x01])
472            .verify(&self.vk.0, &sig, &pol_id.as_ref().unwrap().1)
473        {
474            return Err(Error::IncorrectSignature.into());
475        }
476
477        w.feed(m.into()).await?;
478
479        w.flush().await?;
480        w.close().await?;
481
482        let private_id = pol_id.unwrap().0;
483        let private = if self.pub_id == private_id {
484            None
485        } else {
486            Some(private_id)
487        };
488
489        Ok(VerificationResult {
490            public: self.pub_id.clone(),
491            private,
492        })
493    }
494}
pg_core/client/web/stream.rs

pg_core/client/web/
stream.rs
