1use serde::{Deserialize, Serialize};
2
3use crate::blob::locator::{RemoteAudience, StoredBlobRef};
4use crate::circle::CircleId;
5use crate::circle_control::CircleControlCoord;
6use crate::store_commit::{
7 CandidateFamilyId, ObjectHash, StoreCommitCoord, StoreDeviceRegistrationRef,
8 STORE_PROTOCOL_VERSION,
9};
10use crate::write::WriteId;
11use coven_keys::encryption::KeyFingerprint;
12
13#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(rename_all = "snake_case", deny_unknown_fields)]
16pub enum PackageAudience {
17 Store,
18 Circle {
19 circle_id: CircleId,
20 control: CircleControlCoord,
21 key_fingerprint: KeyFingerprint,
22 },
23}
24
25impl PackageAudience {
26 pub fn remote_audience(&self) -> RemoteAudience {
27 match self {
28 Self::Store => RemoteAudience::Store,
29 Self::Circle { circle_id, .. } => RemoteAudience::Circle(*circle_id),
30 }
31 }
32}
33
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
36#[serde(deny_unknown_fields)]
37pub struct RowBlobLocatorBinding {
38 table: String,
39 row_id: String,
40 row_stamp: String,
41 column: String,
42 blob: StoredBlobRef,
43}
44
45impl RowBlobLocatorBinding {
46 pub fn new(
47 table: impl Into<String>,
48 row_id: impl Into<String>,
49 row_stamp: impl Into<String>,
50 column: impl Into<String>,
51 blob: StoredBlobRef,
52 ) -> Result<Self, AudiencePackageError> {
53 let binding = Self {
54 table: table.into(),
55 row_id: row_id.into(),
56 row_stamp: row_stamp.into(),
57 column: column.into(),
58 blob,
59 };
60 binding.validate()?;
61 Ok(binding)
62 }
63
64 pub fn table(&self) -> &str {
65 &self.table
66 }
67
68 pub fn row_id(&self) -> &str {
69 &self.row_id
70 }
71
72 pub fn row_stamp(&self) -> &str {
73 &self.row_stamp
74 }
75
76 pub fn column(&self) -> &str {
77 &self.column
78 }
79
80 pub fn blob(&self) -> &StoredBlobRef {
81 &self.blob
82 }
83
84 fn sort_key(&self) -> (&str, &str, &str, &str) {
85 (&self.table, &self.row_id, &self.column, &self.row_stamp)
86 }
87
88 fn identity(&self) -> (&str, &str, &str) {
89 (&self.table, &self.row_id, &self.column)
90 }
91
92 fn validate(&self) -> Result<(), AudiencePackageError> {
93 for (field, value) in [
94 ("table", self.table.as_str()),
95 ("row_id", self.row_id.as_str()),
96 ("row_stamp", self.row_stamp.as_str()),
97 ("column", self.column.as_str()),
98 ] {
99 if value.is_empty() {
100 return Err(AudiencePackageError::EmptyBindingField(field));
101 }
102 }
103 Ok(())
104 }
105}
106
107#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
109#[serde(deny_unknown_fields)]
110pub struct AudiencePackage {
111 version: u32,
112 store_root_hash: ObjectHash,
113 candidate_family: CandidateFamilyId,
114 write_id: WriteId,
115 commit_coord: StoreCommitCoord,
116 schema_version: u32,
117 audience: PackageAudience,
118 changeset: Vec<u8>,
119 blob_bindings: Vec<RowBlobLocatorBinding>,
120}
121
122impl AudiencePackage {
123 #[allow(clippy::too_many_arguments)]
124 pub fn store(
125 store_root_hash: ObjectHash,
126 candidate_family: CandidateFamilyId,
127 write_id: WriteId,
128 commit_coord: StoreCommitCoord,
129 schema_version: u32,
130 changeset: Vec<u8>,
131 blob_bindings: Vec<RowBlobLocatorBinding>,
132 ) -> Result<Self, AudiencePackageError> {
133 Self::new(
134 store_root_hash,
135 candidate_family,
136 write_id,
137 commit_coord,
138 schema_version,
139 PackageAudience::Store,
140 changeset,
141 blob_bindings,
142 )
143 }
144
145 #[allow(clippy::too_many_arguments)]
146 pub fn circle(
147 store_root_hash: ObjectHash,
148 candidate_family: CandidateFamilyId,
149 write_id: WriteId,
150 commit_coord: StoreCommitCoord,
151 schema_version: u32,
152 circle_id: CircleId,
153 control: CircleControlCoord,
154 key_fingerprint: KeyFingerprint,
155 changeset: Vec<u8>,
156 blob_bindings: Vec<RowBlobLocatorBinding>,
157 ) -> Result<Self, AudiencePackageError> {
158 Self::new(
159 store_root_hash,
160 candidate_family,
161 write_id,
162 commit_coord,
163 schema_version,
164 PackageAudience::Circle {
165 circle_id,
166 control,
167 key_fingerprint,
168 },
169 changeset,
170 blob_bindings,
171 )
172 }
173
174 #[allow(clippy::too_many_arguments)]
175 fn new(
176 store_root_hash: ObjectHash,
177 candidate_family: CandidateFamilyId,
178 write_id: WriteId,
179 commit_coord: StoreCommitCoord,
180 schema_version: u32,
181 audience: PackageAudience,
182 changeset: Vec<u8>,
183 mut blob_bindings: Vec<RowBlobLocatorBinding>,
184 ) -> Result<Self, AudiencePackageError> {
185 blob_bindings.sort_by(|left, right| left.sort_key().cmp(&right.sort_key()));
186 let package = Self {
187 version: STORE_PROTOCOL_VERSION,
188 store_root_hash,
189 candidate_family,
190 write_id,
191 commit_coord,
192 schema_version,
193 audience,
194 changeset,
195 blob_bindings,
196 };
197 package.validate()?;
198 Ok(package)
199 }
200
201 pub fn parse(bytes: &[u8]) -> Result<Self, AudiencePackageError> {
202 let package: Self = serde_json::from_slice(bytes).map_err(AudiencePackageError::Json)?;
203 package.validate()?;
204 if package.to_bytes() != bytes {
205 return Err(AudiencePackageError::NonCanonicalEncoding);
206 }
207 Ok(package)
208 }
209
210 pub fn to_bytes(&self) -> Vec<u8> {
211 serde_json::to_vec(self).expect("AudiencePackage serialization cannot fail")
212 }
213
214 pub fn store_root_hash(&self) -> ObjectHash {
215 self.store_root_hash
216 }
217
218 pub fn write_id(&self) -> &WriteId {
219 &self.write_id
220 }
221
222 pub fn candidate_family(&self) -> CandidateFamilyId {
223 self.candidate_family
224 }
225
226 pub fn commit_coord(&self) -> &StoreCommitCoord {
227 &self.commit_coord
228 }
229
230 pub fn schema_version(&self) -> u32 {
231 self.schema_version
232 }
233
234 pub fn audience(&self) -> &PackageAudience {
235 &self.audience
236 }
237
238 pub fn changeset(&self) -> &[u8] {
239 &self.changeset
240 }
241
242 pub fn blob_bindings(&self) -> &[RowBlobLocatorBinding] {
243 &self.blob_bindings
244 }
245
246 pub fn validate_blob_uploader(
249 &self,
250 author: &StoreDeviceRegistrationRef,
251 ) -> Result<(), AudiencePackageError> {
252 for binding in &self.blob_bindings {
253 let actual = binding.blob().locator().uploader();
254 if actual != author {
255 return Err(AudiencePackageError::LocatorUploaderMismatch {
256 table: binding.table.clone(),
257 row_id: binding.row_id.clone(),
258 expected: Box::new(author.clone()),
259 actual: Box::new(actual.clone()),
260 });
261 }
262 }
263 Ok(())
264 }
265
266 fn validate(&self) -> Result<(), AudiencePackageError> {
267 if self.version != STORE_PROTOCOL_VERSION {
268 return Err(AudiencePackageError::UnsupportedVersion(self.version));
269 }
270 self.commit_coord.validate()?;
271 if let PackageAudience::Circle { control, .. } = &self.audience {
272 control.validate()?;
273 }
274 let expected_audience = self.audience.remote_audience();
275 let mut previous_sort_key = None;
276 let mut identities = std::collections::BTreeSet::new();
277 for binding in &self.blob_bindings {
278 binding.validate()?;
279 if binding.blob.locator().audience() != expected_audience {
280 return Err(AudiencePackageError::LocatorAudienceMismatch {
281 table: binding.table.clone(),
282 row_id: binding.row_id.clone(),
283 expected: expected_audience.clone(),
284 actual: binding.blob.locator().audience(),
285 });
286 }
287 if let PackageAudience::Circle {
288 key_fingerprint, ..
289 } = &self.audience
290 {
291 if let Some(actual) = binding
292 .blob
293 .locator()
294 .key_fingerprint()
295 .filter(|actual| *actual != *key_fingerprint)
296 {
297 return Err(AudiencePackageError::LocatorKeyFingerprintMismatch {
298 table: binding.table.clone(),
299 row_id: binding.row_id.clone(),
300 expected: *key_fingerprint,
301 actual,
302 });
303 }
304 }
305 if !identities.insert(binding.identity()) {
306 return Err(AudiencePackageError::DuplicateBinding {
307 table: binding.table.clone(),
308 row_id: binding.row_id.clone(),
309 row_stamp: binding.row_stamp.clone(),
310 column: binding.column.clone(),
311 });
312 }
313 let sort_key = binding.sort_key();
314 if let Some(previous) = previous_sort_key {
315 if previous > sort_key {
316 return Err(AudiencePackageError::UnsortedBindings);
317 }
318 }
319 previous_sort_key = Some(sort_key);
320 }
321 Ok(())
322 }
323}
324
325#[derive(Debug, thiserror::Error)]
326pub enum AudiencePackageError {
327 #[error("unsupported audience package version {0}")]
328 UnsupportedVersion(u32),
329 #[error("invalid audience package Store commit coordinate: {0}")]
330 InvalidCommitCoord(#[from] crate::store_commit::StoreProtocolError),
331 #[error("invalid Circle control coordinate: {0}")]
332 InvalidCircleControl(#[from] crate::circle_control::CircleControlCoordError),
333 #[error("row blob binding has empty {0}")]
334 EmptyBindingField(&'static str),
335 #[error(
336 "row blob locator audience mismatch for {table:?}/{row_id:?}: expected {expected:?}, found {actual:?}"
337 )]
338 LocatorAudienceMismatch {
339 table: String,
340 row_id: String,
341 expected: RemoteAudience,
342 actual: RemoteAudience,
343 },
344 #[error(
345 "row blob locator key fingerprint mismatch for {table:?}/{row_id:?}: expected {expected}, found {actual:?}"
346 )]
347 LocatorKeyFingerprintMismatch {
348 table: String,
349 row_id: String,
350 expected: KeyFingerprint,
351 actual: KeyFingerprint,
352 },
353 #[error(
354 "row blob locator uploader mismatch for {table:?}/{row_id:?}: expected {expected:?}, found {actual:?}"
355 )]
356 LocatorUploaderMismatch {
357 table: String,
358 row_id: String,
359 expected: Box<StoreDeviceRegistrationRef>,
360 actual: Box<StoreDeviceRegistrationRef>,
361 },
362 #[error(
363 "duplicate row blob locator binding for {table:?}/{row_id:?}/{column:?} at {row_stamp:?}"
364 )]
365 DuplicateBinding {
366 table: String,
367 row_id: String,
368 row_stamp: String,
369 column: String,
370 },
371 #[error("row blob locator bindings are not canonically sorted")]
372 UnsortedBindings,
373 #[error("malformed audience package: {0}")]
374 Json(#[source] serde_json::Error),
375 #[error("audience package bytes are not canonical")]
376 NonCanonicalEncoding,
377}
378
379#[cfg(test)]
380mod tests {
381 use super::*;
382 use crate::blob::locator::{BlobLocator, RemoteAudience, StoredBlobRef};
383 use crate::blob::BlobScope;
384 use crate::circle::CircleId;
385 use crate::circle_control::CircleControlCoord;
386 use crate::membership::AuthorStreamId;
387 use crate::objects::ExactObjectRef;
388 use crate::objects::ObjectSlot;
389 use crate::store_commit::{
390 CandidateFamilyId, ObjectHash, StoreCommitCoord, StoreDeviceRegistrationRef,
391 };
392 use crate::write::WriteId;
393 use coven_keys::encryption::KeyFingerprint;
394
395 fn uploader() -> StoreDeviceRegistrationRef {
396 let bytes = b"audience-package uploader registration";
397 StoreDeviceRegistrationRef {
398 device_id: "11".repeat(32).parse().unwrap(),
399 registration_hash: ObjectHash::digest(bytes),
400 object: ExactObjectRef::new(
401 ObjectSlot::logical("store-v1/devices/audience-package-uploader.json".to_string())
402 .unwrap(),
403 bytes.len() as u64,
404 ObjectHash::digest(bytes),
405 ),
406 }
407 }
408
409 fn candidate_family() -> CandidateFamilyId {
410 CandidateFamilyId::from_hash(ObjectHash::digest(b"audience-package candidate family"))
411 }
412
413 fn merge_coord() -> StoreCommitCoord {
414 StoreCommitCoord {
415 stream_id: AuthorStreamId::from_bytes([3; 32]),
416 sequence: 3,
417 }
418 }
419
420 fn circle_control_coord() -> CircleControlCoord {
421 CircleControlCoord {
422 device_id: "device-a".to_string(),
423 stream_id: AuthorStreamId::from_bytes([4; 32]),
424 author_pubkey: "22".repeat(32),
425 author_owner_grant: crate::causal_grants::MembershipGrantId::from_test_label(
426 "audience-package owner",
427 ),
428 seq: 4,
429 control_hash: ObjectHash::digest(b"control"),
430 }
431 }
432
433 fn locator(id: &str, audience: RemoteAudience) -> BlobLocator {
434 BlobLocator::opaque(
435 "covers",
436 id,
437 uploader(),
438 audience,
439 BlobScope::Master,
440 KeyFingerprint::from_bytes([4; 32]),
441 7,
442 ObjectHash::digest(id.as_bytes()),
443 )
444 .unwrap()
445 }
446
447 fn stored(locator: BlobLocator) -> StoredBlobRef {
448 let key = locator.semantic_key();
449 StoredBlobRef::new(
450 locator,
451 ExactObjectRef::new(
452 ObjectSlot::logical(key).unwrap(),
453 6,
454 ObjectHash::digest(b"stored"),
455 ),
456 )
457 .unwrap()
458 }
459
460 fn binding(row: &str, locator: BlobLocator) -> RowBlobLocatorBinding {
461 RowBlobLocatorBinding::new(
462 "covers",
463 row,
464 "0000000001000-0000-device",
465 "blob_id",
466 stored(locator),
467 )
468 .unwrap()
469 }
470
471 #[test]
472 fn store_package_sorts_bindings_and_round_trips_canonical_bytes() {
473 let package = AudiencePackage::store(
474 ObjectHash::digest(b"root"),
475 candidate_family(),
476 WriteId::from_generated("write-a".to_string()),
477 merge_coord(),
478 8,
479 b"changeset".to_vec(),
480 vec![
481 binding("row-b", locator("b2c3-blob", RemoteAudience::Store)),
482 binding("row-a", locator("a1b2-blob", RemoteAudience::Store)),
483 ],
484 )
485 .unwrap();
486
487 assert_eq!(package.blob_bindings()[0].row_id(), "row-a");
488 let bytes = package.to_bytes();
489 assert_eq!(AudiencePackage::parse(&bytes).unwrap(), package);
490 assert_eq!(bytes, package.to_bytes());
491 }
492
493 #[test]
494 fn store_package_has_literal_canonical_bytes() {
495 let root = ObjectHash::digest(b"root");
496 let package = AudiencePackage::store(
497 root,
498 candidate_family(),
499 WriteId::from_generated("write-a".to_string()),
500 merge_coord(),
501 8,
502 b"cs".to_vec(),
503 Vec::new(),
504 )
505 .unwrap();
506
507 assert_eq!(
508 String::from_utf8(package.to_bytes()).unwrap(),
509 format!(
510 "{{\"version\":1,\"store_root_hash\":\"{root}\",\"candidate_family\":\"{}\",\"write_id\":\"write-a\",\"commit_coord\":{{\"stream_id\":\"{}\",\"sequence\":3}},\"schema_version\":8,\"audience\":\"store\",\"changeset\":[99,115],\"blob_bindings\":[]}}",
511 candidate_family().as_hash(),
512 AuthorStreamId::from_bytes([3; 32]),
513 )
514 );
515 }
516
517 #[test]
518 fn package_refuses_duplicate_row_binding() {
519 let one = binding("row-a", locator("a1b2-blob", RemoteAudience::Store));
520 let two = binding("row-a", locator("b2c3-blob", RemoteAudience::Store));
521 assert!(matches!(
522 AudiencePackage::store(
523 ObjectHash::digest(b"root"),
524 candidate_family(),
525 WriteId::from_generated("write-a".to_string()),
526 merge_coord(),
527 8,
528 Vec::new(),
529 vec![one, two],
530 ),
531 Err(AudiencePackageError::DuplicateBinding { .. })
532 ));
533 }
534
535 #[test]
536 fn package_refuses_locator_from_another_audience() {
537 let circle = CircleId::from_bytes([8; 16]);
538 assert!(matches!(
539 AudiencePackage::store(
540 ObjectHash::digest(b"root"),
541 candidate_family(),
542 WriteId::from_generated("write-a".to_string()),
543 merge_coord(),
544 8,
545 Vec::new(),
546 vec![binding(
547 "row-a",
548 locator("a1b2-blob", RemoteAudience::Circle(circle))
549 )],
550 ),
551 Err(AudiencePackageError::LocatorAudienceMismatch { .. })
552 ));
553 }
554
555 #[test]
556 fn circle_package_refuses_browsable_locator() {
557 let circle = CircleId::from_bytes([8; 16]);
558 let browsable = BlobLocator::browsable(
559 "audio",
560 "abcd-track",
561 uploader(),
562 "Artist/Album/track.flac",
563 7,
564 ObjectHash::digest(b"track"),
565 )
566 .unwrap();
567
568 assert!(matches!(
569 AudiencePackage::circle(
570 ObjectHash::digest(b"root"),
571 candidate_family(),
572 WriteId::from_generated("write-a".to_string()),
573 merge_coord(),
574 8,
575 circle,
576 circle_control_coord(),
577 KeyFingerprint::from_bytes([4; 32]),
578 Vec::new(),
579 vec![binding("row-a", browsable)],
580 ),
581 Err(AudiencePackageError::LocatorAudienceMismatch { .. })
582 ));
583 }
584
585 #[test]
586 fn circle_package_refuses_locator_from_another_key() {
587 let circle = CircleId::from_bytes([8; 16]);
588 let wrong_key_locator = BlobLocator::opaque(
589 "covers",
590 "a1b2-blob",
591 uploader(),
592 RemoteAudience::Circle(circle),
593 BlobScope::Master,
594 KeyFingerprint::from_bytes([5; 32]),
595 7,
596 ObjectHash::digest(b"cover"),
597 )
598 .unwrap();
599
600 assert!(matches!(
601 AudiencePackage::circle(
602 ObjectHash::digest(b"root"),
603 candidate_family(),
604 WriteId::from_generated("write-a".to_string()),
605 merge_coord(),
606 8,
607 circle,
608 circle_control_coord(),
609 KeyFingerprint::from_bytes([4; 32]),
610 Vec::new(),
611 vec![binding("row-a", wrong_key_locator)],
612 ),
613 Err(AudiencePackageError::LocatorKeyFingerprintMismatch { .. })
614 ));
615 }
616
617 #[test]
618 fn package_rejects_unknown_shape_and_noncanonical_bytes() {
619 let package = AudiencePackage::store(
620 ObjectHash::digest(b"root"),
621 candidate_family(),
622 WriteId::from_generated("write-a".to_string()),
623 merge_coord(),
624 8,
625 b"changeset".to_vec(),
626 Vec::new(),
627 )
628 .unwrap();
629 let bytes = package.to_bytes();
630
631 let mut unknown_field: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
632 unknown_field["unknown"] = serde_json::json!(true);
633 assert!(matches!(
634 AudiencePackage::parse(&serde_json::to_vec(&unknown_field).unwrap()),
635 Err(AudiencePackageError::Json(_))
636 ));
637
638 let mut unknown_variant: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
639 unknown_variant["audience"] = serde_json::json!({ "unknown": {} });
640 assert!(matches!(
641 AudiencePackage::parse(&serde_json::to_vec(&unknown_variant).unwrap()),
642 Err(AudiencePackageError::Json(_))
643 ));
644
645 let mut noncanonical = bytes;
646 noncanonical.push(b'\n');
647 assert!(matches!(
648 AudiencePackage::parse(&noncanonical),
649 Err(AudiencePackageError::NonCanonicalEncoding)
650 ));
651 }
652
653 #[test]
654 fn circle_package_binds_control_fingerprint_and_locator_audience() {
655 let circle = CircleId::from_bytes([8; 16]);
656 let package = AudiencePackage::circle(
657 ObjectHash::digest(b"root"),
658 candidate_family(),
659 WriteId::from_generated("write-a".to_string()),
660 merge_coord(),
661 8,
662 circle,
663 circle_control_coord(),
664 KeyFingerprint::from_bytes([4; 32]),
665 b"circle changeset".to_vec(),
666 vec![binding(
667 "row-a",
668 locator("a1b2-blob", RemoteAudience::Circle(circle)),
669 )],
670 )
671 .unwrap();
672
673 assert_eq!(
674 AudiencePackage::parse(&package.to_bytes()).unwrap(),
675 package
676 );
677 }
678}