Supply data_decryption_key and aad for StorableBuilder

Previously, we the `vss-client` didn't allow to set `ChaCha20Poly1305RFC`'s `aad` field,
which had the `tag` not commit to any particular key. This would allow a
malicious VSS provider to substitute blobs stored under a different key
without the client noticing.

Here, we now set the `aad` field to the key under which the `Storable`
will be stored, ensuring that the retrieved data was originally stored
under the key we expected.

We also account for `StorableBuilder` now taking `data_decryption_key`
by reference on `build`/`deconstruct`.

Author: tnull

src/io/vss_store.rs

@@ -316,7 +316,7 @@ impl Drop for VssStore {
 struct VssStoreInner {
 	client: VssClient<CustomRetryPolicy>,
 	store_id: String,
-	storable_builder: StorableBuilder<RandEntropySource>,
+	data_encryption_key: [u8; 32],
 	key_obfuscator: KeyObfuscator,
 	// Per-key locks that ensures that we don't have concurrent writes to the same namespace/key.
 	// The lock also encapsulates the latest written version per key.
@@ -331,7 +331,6 @@ impl VssStoreInner {
 		let (data_encryption_key, obfuscation_master_key) =
 			derive_data_encryption_and_obfuscation_keys(&vss_seed);
 		let key_obfuscator = KeyObfuscator::new(obfuscation_master_key);
-		let storable_builder = StorableBuilder::new(data_encryption_key, RandEntropySource);
 		let retry_policy = ExponentialBackoffRetryPolicy::new(Duration::from_millis(10))
 			.with_max_attempts(10)
 			.with_max_total_delay(Duration::from_secs(15))
@@ -347,7 +346,7 @@ impl VssStoreInner {
 
 		let client = VssClient::new_with_headers(base_url, retry_policy, header_provider);
 		let locks = Mutex::new(HashMap::new());
-		Self { client, store_id, storable_builder, key_obfuscator, locks }
+		Self { client, store_id, data_encryption_key, key_obfuscator, locks }
 	}
 
 	fn get_inner_lock_ref(&self, locking_key: String) -> Arc<tokio::sync::Mutex<u64>> {
@@ -413,9 +412,8 @@ impl VssStoreInner {
 	) -> io::Result<Vec<u8>> {
 		check_namespace_key_validity(&primary_namespace, &secondary_namespace, Some(&key), "read")?;
 
-		let obfuscated_key =
-			self.build_obfuscated_key(&primary_namespace, &secondary_namespace, &key);
-		let request = GetObjectRequest { store_id: self.store_id.clone(), key: obfuscated_key };
+		let store_key = self.build_obfuscated_key(&primary_namespace, &secondary_namespace, &key);
+		let request = GetObjectRequest { store_id: self.store_id.clone(), key: store_key.clone() };
 		let resp = self.client.get_object(&request).await.map_err(|e| {
 			let msg = format!(
 				"Failed to read from key {}/{}/{}: {}",
@@ -437,7 +435,11 @@ impl VssStoreInner {
 			Error::new(ErrorKind::Other, msg)
 		})?;
 
-		Ok(self.storable_builder.deconstruct(storable)?.0)
+		let storable_builder = StorableBuilder::new(RandEntropySource);
+		let decrypted = storable_builder
+			.deconstruct(storable, &self.data_encryption_key, store_key.as_bytes())?
+			.0;
+		Ok(decrypted)
 	}
 
 	async fn write_internal(
@@ -451,22 +453,27 @@ impl VssStoreInner {
 			"write",
 		)?;
 
-		self.execute_locked_write(inner_lock_ref, locking_key, version, async move || {
-			let obfuscated_key =
-				self.build_obfuscated_key(&primary_namespace, &secondary_namespace, &key);
-			let vss_version = -1;
-			let storable = self.storable_builder.build(buf, vss_version);
-			let request = PutObjectRequest {
-				store_id: self.store_id.clone(),
-				global_version: None,
-				transaction_items: vec![KeyValue {
-					key: obfuscated_key,
-					version: vss_version,
-					value: storable.encode_to_vec(),
-				}],
-				delete_items: vec![],
-			};
+		let store_key = self.build_obfuscated_key(&primary_namespace, &secondary_namespace, &key);
+		let vss_version = -1;
+		let storable_builder = StorableBuilder::new(RandEntropySource);
+		let storable = storable_builder.build(
+			buf.to_vec(),
+			vss_version,
+			&&self.data_encryption_key,
+			store_key.as_bytes(),
+		);
+		let request = PutObjectRequest {
+			store_id: self.store_id.clone(),
+			global_version: None,
+			transaction_items: vec![KeyValue {
+				key: store_key,
+				version: vss_version,
+				value: storable.encode_to_vec(),
+			}],
+			delete_items: vec![],
+		};
 
+		self.execute_locked_write(inner_lock_ref, locking_key, version, async move || {
 			self.client.put_object(&request).await.map_err(|e| {
 				let msg = format!(
 					"Failed to write to key {}/{}/{}: {}",