1use std::{fmt, str::FromStr, time::Duration};
4
5use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
6use sha2::{Digest, Sha256};
7
8use crate::{BeamSet, Manifest, PackageContract, PackageError};
9
10const DIGEST_LEN: usize = 32;
11const TEXT_LEN: usize = DIGEST_LEN * 2;
12const WORKFLOW_TIMEOUT_DOMAIN: &[u8] = b"aion.package.version.workflow-timeout.v1";
13const WORKFLOW_TIMEOUTS_DOMAIN: &[u8] = b"aion.package.version.workflow-timeouts.v3";
14const WORKER_CONTRACT_DOMAIN: &[u8] = b"aion.package.version.worker-contract.v5";
21
22#[derive(Clone, Debug, PartialEq, Eq, Hash)]
45pub struct ContentHash([u8; DIGEST_LEN]);
46
47impl ContentHash {
48 #[must_use]
50 pub const fn from_bytes(bytes: [u8; DIGEST_LEN]) -> Self {
51 Self(bytes)
52 }
53
54 #[must_use]
56 pub const fn as_bytes(&self) -> &[u8; DIGEST_LEN] {
57 &self.0
58 }
59}
60
61#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq)]
63pub enum ContentHashParseError {
64 #[error("content hash text must be 64 lowercase hexadecimal characters, found {found} bytes")]
66 InvalidLength {
67 found: usize,
69 },
70
71 #[error("content hash text contains non-lowercase-hex byte 0x{byte:02x} at byte index {index}")]
73 InvalidCharacter {
74 index: usize,
76 byte: u8,
78 },
79}
80
81#[must_use]
89pub fn content_hash(beams: &BeamSet) -> ContentHash {
90 let mut digest = Sha256::new();
91 update_beams(&mut digest, beams);
92 ContentHash(digest.finalize().into())
93}
94
95#[must_use]
104pub fn content_hash_with_timeout(beams: &BeamSet, timeout: Duration) -> ContentHash {
105 let mut digest = Sha256::new();
106 update_beams(&mut digest, beams);
107 update_framed(&mut digest, WORKFLOW_TIMEOUT_DOMAIN);
108 digest.update(timeout.as_secs().to_be_bytes());
109 digest.update(timeout.subsec_nanos().to_be_bytes());
110 ContentHash(digest.finalize().into())
111}
112
113#[must_use]
141pub fn content_hash_with_timeouts(beams: &BeamSet, manifest: &Manifest) -> ContentHash {
142 let mut digest = Sha256::new();
143 update_beams(&mut digest, beams);
144 update_framed(&mut digest, WORKFLOW_TIMEOUTS_DOMAIN);
145 update_timeouts(&mut digest, manifest);
146 ContentHash(digest.finalize().into())
147}
148
149#[must_use]
156pub fn content_hash_with_contract(
157 beams: &BeamSet,
158 manifest: &Manifest,
159 contract: &PackageContract,
160) -> ContentHash {
161 let mut digest = Sha256::new();
162 update_beams(&mut digest, beams);
163 update_framed(&mut digest, WORKER_CONTRACT_DOMAIN);
164 update_timeouts(&mut digest, manifest);
165 update_framed(&mut digest, &contract.canonical_bytes());
166 ContentHash(digest.finalize().into())
167}
168
169fn update_timeouts(digest: &mut Sha256, manifest: &Manifest) {
170 let entry_count = 1 + manifest.additional_workflows.len() as u64;
171 digest.update(entry_count.to_be_bytes());
172 update_framed(digest, manifest.entry_module.as_bytes());
173 update_timeout_field(digest, manifest.timeout);
174 for entry in &manifest.additional_workflows {
175 update_framed(digest, entry.workflow_type.as_bytes());
176 update_timeout_field(digest, entry.timeout);
177 }
178}
179
180#[cfg(test)]
194pub(crate) fn has_explicit_timeout_identity(
195 beams: &BeamSet,
196 manifest: &Manifest,
197 hash: &ContentHash,
198) -> bool {
199 hash != &content_hash(beams) && hash == &content_hash_with_timeouts(beams, manifest)
200}
201
202pub(crate) fn has_contract_identity(
204 beams: &BeamSet,
205 manifest: &Manifest,
206 contract: Option<&PackageContract>,
207 hash: &ContentHash,
208) -> bool {
209 contract.is_some_and(|contract| hash == &content_hash_with_contract(beams, manifest, contract))
210}
211
212pub(crate) fn verified_content_hash_with_contract(
214 beams: &BeamSet,
215 manifest: &Manifest,
216 contract: Option<&PackageContract>,
217) -> Result<ContentHash, PackageError> {
218 if let Some(contract) = contract {
219 let contract_hash = content_hash_with_contract(beams, manifest, contract);
220 if manifest.version.as_str() == contract_hash.to_string() {
221 return Ok(contract_hash);
222 }
223 return verified_content_hash(beams, manifest).map_err(|error| match error {
231 PackageError::IntegrityMismatch { expected, .. } => PackageError::IntegrityMismatch {
232 expected,
233 computed: contract_hash.to_string(),
234 },
235 other => other,
236 });
237 }
238 verified_content_hash(beams, manifest)
239}
240
241pub(crate) fn verified_content_hash(
260 beams: &BeamSet,
261 manifest: &Manifest,
262) -> Result<ContentHash, PackageError> {
263 let legacy_hash = content_hash(beams);
264 let stored = manifest.version.as_str();
265 if stored == legacy_hash.to_string() {
266 return Ok(legacy_hash);
267 }
268 let timeouts_hash = content_hash_with_timeouts(beams, manifest);
269 if stored == timeouts_hash.to_string() {
270 return Ok(timeouts_hash);
271 }
272 if let Some(primary) = manifest.timeout {
277 let v1_hash = content_hash_with_timeout(beams, primary);
278 if stored == v1_hash.to_string() {
279 return Ok(v1_hash);
280 }
281 }
282 Err(PackageError::IntegrityMismatch {
283 expected: stored.to_owned(),
284 computed: legacy_hash.to_string(),
285 })
286}
287
288fn update_timeout_field(digest: &mut Sha256, timeout: Option<Duration>) {
293 match timeout {
294 Some(timeout) => {
295 digest.update([1_u8]);
296 digest.update(timeout.as_secs().to_be_bytes());
297 digest.update(timeout.subsec_nanos().to_be_bytes());
298 }
299 None => digest.update([0_u8]),
300 }
301}
302
303fn update_beams(digest: &mut Sha256, beams: &BeamSet) {
304 for module in beams.iter() {
305 update_framed(digest, module.name().as_bytes());
306 update_framed(digest, module.bytes());
307 }
308}
309
310fn update_framed(digest: &mut Sha256, bytes: &[u8]) {
311 let length = bytes.len() as u64;
312 digest.update(length.to_be_bytes().as_slice());
313 digest.update(bytes);
314}
315
316impl fmt::Display for ContentHash {
317 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
318 for byte in &self.0 {
319 write!(formatter, "{byte:02x}")?;
320 }
321
322 Ok(())
323 }
324}
325
326impl FromStr for ContentHash {
327 type Err = ContentHashParseError;
328
329 fn from_str(text: &str) -> Result<Self, Self::Err> {
330 let bytes = text.as_bytes();
331 if bytes.len() != TEXT_LEN {
332 return Err(ContentHashParseError::InvalidLength { found: bytes.len() });
333 }
334
335 let mut digest = [0_u8; DIGEST_LEN];
336 for (index, pair) in bytes.chunks_exact(2).enumerate() {
337 let high_index = index * 2;
338 let low_index = high_index + 1;
339 digest[index] = (hex_value(pair[0], high_index)? << 4) | hex_value(pair[1], low_index)?;
340 }
341
342 Ok(Self(digest))
343 }
344}
345
346fn hex_value(byte: u8, index: usize) -> Result<u8, ContentHashParseError> {
347 match byte {
348 b'0'..=b'9' => Ok(byte - b'0'),
349 b'a'..=b'f' => Ok(byte - b'a' + 10),
350 _ => Err(ContentHashParseError::InvalidCharacter { index, byte }),
351 }
352}
353
354impl Serialize for ContentHash {
355 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
356 where
357 S: Serializer,
358 {
359 serializer.serialize_str(&self.to_string())
360 }
361}
362
363impl<'de> Deserialize<'de> for ContentHash {
364 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
365 where
366 D: Deserializer<'de>,
367 {
368 deserializer.deserialize_str(ContentHashVisitor)
369 }
370}
371
372struct ContentHashVisitor;
373
374impl de::Visitor<'_> for ContentHashVisitor {
375 type Value = ContentHash;
376
377 fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
378 formatter.write_str("a 64-character lowercase hexadecimal SHA-256 content hash")
379 }
380
381 fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
382 where
383 E: de::Error,
384 {
385 ContentHash::from_str(value).map_err(E::custom)
386 }
387}
388
389#[cfg(test)]
390mod tests {
391 use std::time::Duration;
392
393 use serde_json::json;
394
395 use super::{
396 ContentHash, content_hash, content_hash_with_timeout, content_hash_with_timeouts,
397 has_explicit_timeout_identity, verified_content_hash,
398 };
399 use crate::{
400 BeamModule, BeamSet, CURRENT_FORMAT_VERSION, Manifest, ManifestVersion, PackageError,
401 WorkflowEntry,
402 };
403
404 fn manifest_with(primary: Option<Duration>, additional: Vec<WorkflowEntry>) -> Manifest {
405 Manifest {
406 entry_module: "workflow/a".to_owned(),
407 entry_function: "run".to_owned(),
408 input_schema: json!({ "type": "object" }),
409 output_schema: json!({ "type": "object" }),
410 timeout: primary,
411 activities: Vec::new(),
412 version: ManifestVersion::new("unstamped"),
413 format_version: CURRENT_FORMAT_VERSION,
414 additional_workflows: additional,
415 }
416 }
417
418 fn additional_entry(workflow_type: &str, timeout: Option<Duration>) -> WorkflowEntry {
419 WorkflowEntry {
420 workflow_type: workflow_type.to_owned(),
421 entry_module: "workflow/a".to_owned(),
422 entry_function: format!("{workflow_type}_run"),
423 input_schema: json!({ "type": "object" }),
424 output_schema: json!({ "type": "object" }),
425 timeout,
426 internal: true,
427 }
428 }
429
430 #[test]
431 fn content_hash_is_independent_of_insertion_order() -> Result<(), PackageError> {
432 let first = BeamSet::new(vec![
433 BeamModule::new("workflow/c", vec![3]),
434 BeamModule::new("workflow/a", vec![1]),
435 BeamModule::new("workflow/b", vec![2]),
436 ])?;
437 let second = BeamSet::new(vec![
438 BeamModule::new("workflow/b", vec![2]),
439 BeamModule::new("workflow/c", vec![3]),
440 BeamModule::new("workflow/a", vec![1]),
441 ])?;
442
443 assert_eq!(content_hash(&first), content_hash(&second));
444
445 Ok(())
446 }
447
448 #[test]
449 fn legacy_identity_remains_exactly_the_beams_only_hash() -> Result<(), PackageError> {
450 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
451 let pre_change_rule = content_hash(&beams);
452 assert_eq!(content_hash(&beams), pre_change_rule);
453 Ok(())
454 }
455
456 #[test]
457 fn explicit_timeout_identity_is_deterministic_and_value_sensitive() -> Result<(), PackageError>
458 {
459 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
460 let two_hours = content_hash_with_timeout(&beams, Duration::from_secs(7_200));
461 assert_eq!(
462 two_hours,
463 content_hash_with_timeout(&beams, Duration::from_secs(7_200))
464 );
465 assert_ne!(
466 two_hours,
467 content_hash_with_timeout(&beams, Duration::from_secs(21_600))
468 );
469 assert_ne!(
470 two_hours,
471 content_hash_with_timeout(&beams, Duration::new(7_200, 500_000_000))
472 );
473 assert_ne!(two_hours, content_hash(&beams));
474 Ok(())
475 }
476
477 #[test]
478 fn per_entry_identity_binds_every_additional_entry_timeout() -> Result<(), PackageError> {
479 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
480 let base = manifest_with(
481 Some(Duration::from_secs(60)),
482 vec![additional_entry("child", Some(Duration::from_secs(30)))],
483 );
484
485 let changed_value = manifest_with(
487 Some(Duration::from_secs(60)),
488 vec![additional_entry("child", Some(Duration::from_secs(31)))],
489 );
490 assert_ne!(
491 content_hash_with_timeouts(&beams, &base),
492 content_hash_with_timeouts(&beams, &changed_value),
493 );
494
495 let injected = manifest_with(
498 Some(Duration::from_secs(60)),
499 vec![additional_entry("child", Some(Duration::from_secs(3_600)))],
500 );
501 assert_ne!(
502 content_hash_with_timeouts(&beams, &base),
503 content_hash_with_timeouts(&beams, &injected),
504 );
505
506 let absent = manifest_with(
508 Some(Duration::from_secs(60)),
509 vec![additional_entry("child", None)],
510 );
511 assert_ne!(
512 content_hash_with_timeouts(&beams, &base),
513 content_hash_with_timeouts(&beams, &absent),
514 );
515 Ok(())
516 }
517
518 #[test]
519 fn v3_identity_binds_the_primary_routing_identity() -> Result<(), PackageError> {
520 let beams = BeamSet::new(vec![
525 BeamModule::new("workflow/a", vec![1, 2, 3]),
526 BeamModule::new("workflow/b", vec![4, 5, 6]),
527 ])?;
528 let on_a = manifest_with(Some(Duration::from_secs(60)), Vec::new());
529 let mut on_b = on_a.clone();
530 on_b.entry_module = "workflow/b".to_owned();
531 assert_ne!(
532 content_hash_with_timeouts(&beams, &on_a),
533 content_hash_with_timeouts(&beams, &on_b),
534 "re-routing the primary entry_module changes identity",
535 );
536 let stored_for_a = content_hash_with_timeouts(&beams, &on_a);
539 assert!(!has_explicit_timeout_identity(&beams, &on_b, &stored_for_a));
540 assert!(has_explicit_timeout_identity(&beams, &on_a, &stored_for_a));
541 Ok(())
542 }
543
544 #[test]
545 fn v1_single_value_archive_loads_but_reads_undeclared() -> Result<(), PackageError> {
546 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
552 let mut manifest = manifest_with(Some(Duration::from_secs(60)), Vec::new());
553 let v1 = content_hash_with_timeout(&beams, Duration::from_secs(60));
554 manifest.version = ManifestVersion::new(v1.to_string());
555 assert_eq!(verified_content_hash(&beams, &manifest)?, v1);
556 assert!(!has_explicit_timeout_identity(&beams, &manifest, &v1));
557 Ok(())
558 }
559
560 #[test]
561 fn non_v1_non_v3_timeout_identity_is_rejected() -> Result<(), PackageError> {
562 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
566 let mut manifest = manifest_with(Some(Duration::from_secs(60)), Vec::new());
567 manifest.version = ManifestVersion::new("f".repeat(64));
568 assert!(matches!(
569 verified_content_hash(&beams, &manifest),
570 Err(PackageError::IntegrityMismatch { .. })
571 ));
572 Ok(())
573 }
574
575 #[test]
576 fn mixed_archive_with_injected_additional_timeout_reads_as_undeclared()
577 -> Result<(), PackageError> {
578 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
584 let manifest = manifest_with(
585 Some(Duration::from_secs(60)),
586 vec![additional_entry("child", Some(Duration::from_secs(3_600)))],
587 );
588 let primary_only = content_hash_with_timeout(&beams, Duration::from_secs(60));
590 assert!(
591 !has_explicit_timeout_identity(&beams, &manifest, &primary_only),
592 "an injected additional timeout cannot ride the primary-only identity"
593 );
594 assert!(!has_explicit_timeout_identity(
596 &beams,
597 &manifest,
598 &content_hash(&beams)
599 ));
600 assert!(has_explicit_timeout_identity(
602 &beams,
603 &manifest,
604 &content_hash_with_timeouts(&beams, &manifest)
605 ));
606 Ok(())
607 }
608
609 #[test]
610 fn content_hash_changes_when_a_module_byte_changes() -> Result<(), PackageError> {
611 let original = BeamSet::new(vec![
612 BeamModule::new("workflow/a", vec![1, 2, 3]),
613 BeamModule::new("workflow/b", vec![4, 5, 6]),
614 ])?;
615 let changed = BeamSet::new(vec![
616 BeamModule::new("workflow/a", vec![1, 2, 3]),
617 BeamModule::new("workflow/b", vec![4, 5, 7]),
618 ])?;
619
620 assert_ne!(content_hash(&original), content_hash(&changed));
621
622 Ok(())
623 }
624
625 #[test]
626 fn content_hash_changes_when_a_module_name_changes() -> Result<(), PackageError> {
627 let original = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
628 let renamed = BeamSet::new(vec![BeamModule::new("workflow/renamed", vec![1, 2, 3])])?;
629
630 assert_ne!(content_hash(&original), content_hash(&renamed));
631
632 Ok(())
633 }
634
635 #[test]
636 fn content_hash_framing_prevents_name_bytes_boundary_ambiguity() -> Result<(), PackageError> {
637 let first = BeamSet::new(vec![BeamModule::new("ab", b"c".to_vec())])?;
638 let second = BeamSet::new(vec![BeamModule::new("a", b"bc".to_vec())])?;
639
640 assert_ne!(content_hash(&first), content_hash(&second));
641
642 Ok(())
643 }
644
645 #[test]
646 fn content_hash_text_round_trips() -> Result<(), PackageError> {
647 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![0, 1, 2, 255])])?;
648 let hash = content_hash(&beams);
649 let text = hash.to_string();
650 let parsed = text.parse::<ContentHash>();
651
652 assert_eq!(text.len(), 64);
653 assert!(
654 text.bytes()
655 .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
656 );
657 assert_eq!(parsed, Ok(hash));
658
659 Ok(())
660 }
661
662 #[test]
663 fn content_hash_rejects_uppercase_text() {
664 let text = "A000000000000000000000000000000000000000000000000000000000000000";
665
666 assert!(text.parse::<ContentHash>().is_err());
667 }
668}