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.v6";
36
37#[derive(Clone, Debug, PartialEq, Eq, Hash)]
60pub struct ContentHash([u8; DIGEST_LEN]);
61
62impl ContentHash {
63 #[must_use]
65 pub const fn from_bytes(bytes: [u8; DIGEST_LEN]) -> Self {
66 Self(bytes)
67 }
68
69 #[must_use]
71 pub const fn as_bytes(&self) -> &[u8; DIGEST_LEN] {
72 &self.0
73 }
74}
75
76#[derive(thiserror::Error, Clone, Debug, PartialEq, Eq)]
78pub enum ContentHashParseError {
79 #[error("content hash text must be 64 lowercase hexadecimal characters, found {found} bytes")]
81 InvalidLength {
82 found: usize,
84 },
85
86 #[error("content hash text contains non-lowercase-hex byte 0x{byte:02x} at byte index {index}")]
88 InvalidCharacter {
89 index: usize,
91 byte: u8,
93 },
94}
95
96#[must_use]
104pub fn content_hash(beams: &BeamSet) -> ContentHash {
105 let mut digest = Sha256::new();
106 update_beams(&mut digest, beams);
107 ContentHash(digest.finalize().into())
108}
109
110#[must_use]
119pub fn content_hash_with_timeout(beams: &BeamSet, timeout: Duration) -> ContentHash {
120 let mut digest = Sha256::new();
121 update_beams(&mut digest, beams);
122 update_framed(&mut digest, WORKFLOW_TIMEOUT_DOMAIN);
123 digest.update(timeout.as_secs().to_be_bytes());
124 digest.update(timeout.subsec_nanos().to_be_bytes());
125 ContentHash(digest.finalize().into())
126}
127
128#[must_use]
156pub fn content_hash_with_timeouts(beams: &BeamSet, manifest: &Manifest) -> ContentHash {
157 let mut digest = Sha256::new();
158 update_beams(&mut digest, beams);
159 update_framed(&mut digest, WORKFLOW_TIMEOUTS_DOMAIN);
160 update_timeouts(&mut digest, manifest);
161 ContentHash(digest.finalize().into())
162}
163
164#[must_use]
171pub fn content_hash_with_contract(
172 beams: &BeamSet,
173 manifest: &Manifest,
174 contract: &PackageContract,
175) -> ContentHash {
176 let mut digest = Sha256::new();
177 update_beams(&mut digest, beams);
178 update_framed(&mut digest, WORKER_CONTRACT_DOMAIN);
179 update_timeouts(&mut digest, manifest);
180 update_framed(&mut digest, &contract.canonical_bytes());
181 ContentHash(digest.finalize().into())
182}
183
184fn update_timeouts(digest: &mut Sha256, manifest: &Manifest) {
185 let entry_count = 1 + manifest.additional_workflows.len() as u64;
186 digest.update(entry_count.to_be_bytes());
187 update_framed(digest, manifest.entry_module.as_bytes());
188 update_timeout_field(digest, manifest.timeout);
189 for entry in &manifest.additional_workflows {
190 update_framed(digest, entry.workflow_type.as_bytes());
191 update_timeout_field(digest, entry.timeout);
192 }
193}
194
195#[cfg(test)]
209pub(crate) fn has_explicit_timeout_identity(
210 beams: &BeamSet,
211 manifest: &Manifest,
212 hash: &ContentHash,
213) -> bool {
214 hash != &content_hash(beams) && hash == &content_hash_with_timeouts(beams, manifest)
215}
216
217pub(crate) fn has_contract_identity(
219 beams: &BeamSet,
220 manifest: &Manifest,
221 contract: Option<&PackageContract>,
222 hash: &ContentHash,
223) -> bool {
224 contract.is_some_and(|contract| hash == &content_hash_with_contract(beams, manifest, contract))
225}
226
227pub(crate) fn verified_content_hash_with_contract(
229 beams: &BeamSet,
230 manifest: &Manifest,
231 contract: Option<&PackageContract>,
232) -> Result<ContentHash, PackageError> {
233 if let Some(contract) = contract {
234 let contract_hash = content_hash_with_contract(beams, manifest, contract);
235 if manifest.version.as_str() == contract_hash.to_string() {
236 return Ok(contract_hash);
237 }
238 return verified_content_hash(beams, manifest).map_err(|error| match error {
246 PackageError::IntegrityMismatch { expected, .. } => PackageError::IntegrityMismatch {
247 expected,
248 computed: contract_hash.to_string(),
249 },
250 other => other,
251 });
252 }
253 verified_content_hash(beams, manifest)
254}
255
256pub(crate) fn verified_content_hash(
275 beams: &BeamSet,
276 manifest: &Manifest,
277) -> Result<ContentHash, PackageError> {
278 let legacy_hash = content_hash(beams);
279 let stored = manifest.version.as_str();
280 if stored == legacy_hash.to_string() {
281 return Ok(legacy_hash);
282 }
283 let timeouts_hash = content_hash_with_timeouts(beams, manifest);
284 if stored == timeouts_hash.to_string() {
285 return Ok(timeouts_hash);
286 }
287 if let Some(primary) = manifest.timeout {
292 let v1_hash = content_hash_with_timeout(beams, primary);
293 if stored == v1_hash.to_string() {
294 return Ok(v1_hash);
295 }
296 }
297 Err(PackageError::IntegrityMismatch {
298 expected: stored.to_owned(),
299 computed: legacy_hash.to_string(),
300 })
301}
302
303fn update_timeout_field(digest: &mut Sha256, timeout: Option<Duration>) {
308 match timeout {
309 Some(timeout) => {
310 digest.update([1_u8]);
311 digest.update(timeout.as_secs().to_be_bytes());
312 digest.update(timeout.subsec_nanos().to_be_bytes());
313 }
314 None => digest.update([0_u8]),
315 }
316}
317
318fn update_beams(digest: &mut Sha256, beams: &BeamSet) {
319 for module in beams.iter() {
320 update_framed(digest, module.name().as_bytes());
321 update_framed(digest, module.bytes());
322 }
323}
324
325fn update_framed(digest: &mut Sha256, bytes: &[u8]) {
326 let length = bytes.len() as u64;
327 digest.update(length.to_be_bytes().as_slice());
328 digest.update(bytes);
329}
330
331impl fmt::Display for ContentHash {
332 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
333 for byte in &self.0 {
334 write!(formatter, "{byte:02x}")?;
335 }
336
337 Ok(())
338 }
339}
340
341impl FromStr for ContentHash {
342 type Err = ContentHashParseError;
343
344 fn from_str(text: &str) -> Result<Self, Self::Err> {
345 let bytes = text.as_bytes();
346 if bytes.len() != TEXT_LEN {
347 return Err(ContentHashParseError::InvalidLength { found: bytes.len() });
348 }
349
350 let mut digest = [0_u8; DIGEST_LEN];
351 for (index, pair) in bytes.chunks_exact(2).enumerate() {
352 let high_index = index * 2;
353 let low_index = high_index + 1;
354 digest[index] = (hex_value(pair[0], high_index)? << 4) | hex_value(pair[1], low_index)?;
355 }
356
357 Ok(Self(digest))
358 }
359}
360
361fn hex_value(byte: u8, index: usize) -> Result<u8, ContentHashParseError> {
362 match byte {
363 b'0'..=b'9' => Ok(byte - b'0'),
364 b'a'..=b'f' => Ok(byte - b'a' + 10),
365 _ => Err(ContentHashParseError::InvalidCharacter { index, byte }),
366 }
367}
368
369impl Serialize for ContentHash {
370 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
371 where
372 S: Serializer,
373 {
374 serializer.serialize_str(&self.to_string())
375 }
376}
377
378impl<'de> Deserialize<'de> for ContentHash {
379 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
380 where
381 D: Deserializer<'de>,
382 {
383 deserializer.deserialize_str(ContentHashVisitor)
384 }
385}
386
387struct ContentHashVisitor;
388
389impl de::Visitor<'_> for ContentHashVisitor {
390 type Value = ContentHash;
391
392 fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
393 formatter.write_str("a 64-character lowercase hexadecimal SHA-256 content hash")
394 }
395
396 fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
397 where
398 E: de::Error,
399 {
400 ContentHash::from_str(value).map_err(E::custom)
401 }
402}
403
404#[cfg(test)]
405mod tests {
406 use std::time::Duration;
407
408 use serde_json::json;
409
410 use super::{
411 ContentHash, content_hash, content_hash_with_timeout, content_hash_with_timeouts,
412 has_explicit_timeout_identity, verified_content_hash,
413 };
414 use crate::{
415 BeamModule, BeamSet, CURRENT_FORMAT_VERSION, Manifest, ManifestVersion, PackageError,
416 WorkflowEntry,
417 };
418
419 fn manifest_with(primary: Option<Duration>, additional: Vec<WorkflowEntry>) -> Manifest {
420 Manifest {
421 entry_module: "workflow/a".to_owned(),
422 entry_function: "run".to_owned(),
423 input_schema: json!({ "type": "object" }),
424 output_schema: json!({ "type": "object" }),
425 timeout: primary,
426 activities: Vec::new(),
427 version: ManifestVersion::new("unstamped"),
428 format_version: CURRENT_FORMAT_VERSION,
429 additional_workflows: additional,
430 }
431 }
432
433 fn additional_entry(workflow_type: &str, timeout: Option<Duration>) -> WorkflowEntry {
434 WorkflowEntry {
435 workflow_type: workflow_type.to_owned(),
436 entry_module: "workflow/a".to_owned(),
437 entry_function: format!("{workflow_type}_run"),
438 input_schema: json!({ "type": "object" }),
439 output_schema: json!({ "type": "object" }),
440 timeout,
441 internal: true,
442 }
443 }
444
445 #[test]
446 fn content_hash_is_independent_of_insertion_order() -> Result<(), PackageError> {
447 let first = BeamSet::new(vec![
448 BeamModule::new("workflow/c", vec![3]),
449 BeamModule::new("workflow/a", vec![1]),
450 BeamModule::new("workflow/b", vec![2]),
451 ])?;
452 let second = BeamSet::new(vec![
453 BeamModule::new("workflow/b", vec![2]),
454 BeamModule::new("workflow/c", vec![3]),
455 BeamModule::new("workflow/a", vec![1]),
456 ])?;
457
458 assert_eq!(content_hash(&first), content_hash(&second));
459
460 Ok(())
461 }
462
463 #[test]
464 fn legacy_identity_remains_exactly_the_beams_only_hash() -> Result<(), PackageError> {
465 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
466 let pre_change_rule = content_hash(&beams);
467 assert_eq!(content_hash(&beams), pre_change_rule);
468 Ok(())
469 }
470
471 #[test]
472 fn explicit_timeout_identity_is_deterministic_and_value_sensitive() -> Result<(), PackageError>
473 {
474 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
475 let two_hours = content_hash_with_timeout(&beams, Duration::from_secs(7_200));
476 assert_eq!(
477 two_hours,
478 content_hash_with_timeout(&beams, Duration::from_secs(7_200))
479 );
480 assert_ne!(
481 two_hours,
482 content_hash_with_timeout(&beams, Duration::from_secs(21_600))
483 );
484 assert_ne!(
485 two_hours,
486 content_hash_with_timeout(&beams, Duration::new(7_200, 500_000_000))
487 );
488 assert_ne!(two_hours, content_hash(&beams));
489 Ok(())
490 }
491
492 #[test]
493 fn per_entry_identity_binds_every_additional_entry_timeout() -> Result<(), PackageError> {
494 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
495 let base = manifest_with(
496 Some(Duration::from_secs(60)),
497 vec![additional_entry("child", Some(Duration::from_secs(30)))],
498 );
499
500 let changed_value = manifest_with(
502 Some(Duration::from_secs(60)),
503 vec![additional_entry("child", Some(Duration::from_secs(31)))],
504 );
505 assert_ne!(
506 content_hash_with_timeouts(&beams, &base),
507 content_hash_with_timeouts(&beams, &changed_value),
508 );
509
510 let injected = manifest_with(
513 Some(Duration::from_secs(60)),
514 vec![additional_entry("child", Some(Duration::from_secs(3_600)))],
515 );
516 assert_ne!(
517 content_hash_with_timeouts(&beams, &base),
518 content_hash_with_timeouts(&beams, &injected),
519 );
520
521 let absent = manifest_with(
523 Some(Duration::from_secs(60)),
524 vec![additional_entry("child", None)],
525 );
526 assert_ne!(
527 content_hash_with_timeouts(&beams, &base),
528 content_hash_with_timeouts(&beams, &absent),
529 );
530 Ok(())
531 }
532
533 #[test]
534 fn v3_identity_binds_the_primary_routing_identity() -> Result<(), PackageError> {
535 let beams = BeamSet::new(vec![
540 BeamModule::new("workflow/a", vec![1, 2, 3]),
541 BeamModule::new("workflow/b", vec![4, 5, 6]),
542 ])?;
543 let on_a = manifest_with(Some(Duration::from_secs(60)), Vec::new());
544 let mut on_b = on_a.clone();
545 on_b.entry_module = "workflow/b".to_owned();
546 assert_ne!(
547 content_hash_with_timeouts(&beams, &on_a),
548 content_hash_with_timeouts(&beams, &on_b),
549 "re-routing the primary entry_module changes identity",
550 );
551 let stored_for_a = content_hash_with_timeouts(&beams, &on_a);
554 assert!(!has_explicit_timeout_identity(&beams, &on_b, &stored_for_a));
555 assert!(has_explicit_timeout_identity(&beams, &on_a, &stored_for_a));
556 Ok(())
557 }
558
559 #[test]
560 fn v1_single_value_archive_loads_but_reads_undeclared() -> Result<(), PackageError> {
561 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
567 let mut manifest = manifest_with(Some(Duration::from_secs(60)), Vec::new());
568 let v1 = content_hash_with_timeout(&beams, Duration::from_secs(60));
569 manifest.version = ManifestVersion::new(v1.to_string());
570 assert_eq!(verified_content_hash(&beams, &manifest)?, v1);
571 assert!(!has_explicit_timeout_identity(&beams, &manifest, &v1));
572 Ok(())
573 }
574
575 #[test]
576 fn non_v1_non_v3_timeout_identity_is_rejected() -> Result<(), PackageError> {
577 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
581 let mut manifest = manifest_with(Some(Duration::from_secs(60)), Vec::new());
582 manifest.version = ManifestVersion::new("f".repeat(64));
583 assert!(matches!(
584 verified_content_hash(&beams, &manifest),
585 Err(PackageError::IntegrityMismatch { .. })
586 ));
587 Ok(())
588 }
589
590 #[test]
591 fn mixed_archive_with_injected_additional_timeout_reads_as_undeclared()
592 -> Result<(), PackageError> {
593 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
599 let manifest = manifest_with(
600 Some(Duration::from_secs(60)),
601 vec![additional_entry("child", Some(Duration::from_secs(3_600)))],
602 );
603 let primary_only = content_hash_with_timeout(&beams, Duration::from_secs(60));
605 assert!(
606 !has_explicit_timeout_identity(&beams, &manifest, &primary_only),
607 "an injected additional timeout cannot ride the primary-only identity"
608 );
609 assert!(!has_explicit_timeout_identity(
611 &beams,
612 &manifest,
613 &content_hash(&beams)
614 ));
615 assert!(has_explicit_timeout_identity(
617 &beams,
618 &manifest,
619 &content_hash_with_timeouts(&beams, &manifest)
620 ));
621 Ok(())
622 }
623
624 #[test]
625 fn content_hash_changes_when_a_module_byte_changes() -> Result<(), PackageError> {
626 let original = BeamSet::new(vec![
627 BeamModule::new("workflow/a", vec![1, 2, 3]),
628 BeamModule::new("workflow/b", vec![4, 5, 6]),
629 ])?;
630 let changed = BeamSet::new(vec![
631 BeamModule::new("workflow/a", vec![1, 2, 3]),
632 BeamModule::new("workflow/b", vec![4, 5, 7]),
633 ])?;
634
635 assert_ne!(content_hash(&original), content_hash(&changed));
636
637 Ok(())
638 }
639
640 #[test]
641 fn content_hash_changes_when_a_module_name_changes() -> Result<(), PackageError> {
642 let original = BeamSet::new(vec![BeamModule::new("workflow/a", vec![1, 2, 3])])?;
643 let renamed = BeamSet::new(vec![BeamModule::new("workflow/renamed", vec![1, 2, 3])])?;
644
645 assert_ne!(content_hash(&original), content_hash(&renamed));
646
647 Ok(())
648 }
649
650 #[test]
651 fn content_hash_framing_prevents_name_bytes_boundary_ambiguity() -> Result<(), PackageError> {
652 let first = BeamSet::new(vec![BeamModule::new("ab", b"c".to_vec())])?;
653 let second = BeamSet::new(vec![BeamModule::new("a", b"bc".to_vec())])?;
654
655 assert_ne!(content_hash(&first), content_hash(&second));
656
657 Ok(())
658 }
659
660 #[test]
661 fn content_hash_text_round_trips() -> Result<(), PackageError> {
662 let beams = BeamSet::new(vec![BeamModule::new("workflow/a", vec![0, 1, 2, 255])])?;
663 let hash = content_hash(&beams);
664 let text = hash.to_string();
665 let parsed = text.parse::<ContentHash>();
666
667 assert_eq!(text.len(), 64);
668 assert!(
669 text.bytes()
670 .all(|byte| matches!(byte, b'0'..=b'9' | b'a'..=b'f'))
671 );
672 assert_eq!(parsed, Ok(hash));
673
674 Ok(())
675 }
676
677 #[test]
678 fn content_hash_rejects_uppercase_text() {
679 let text = "A000000000000000000000000000000000000000000000000000000000000000";
680
681 assert!(text.parse::<ContentHash>().is_err());
682 }
683}