1use std::collections::{HashMap, HashSet};
39use std::io::{BufRead, BufReader, Read};
40use std::sync::{mpsc, Arc, Condvar, Mutex, PoisonError};
41use std::time::Duration;
42
43use base64::Engine as _;
44use ring::signature::{Ed25519KeyPair, KeyPair, UnparsedPublicKey, ED25519};
45use serde_json::Value;
46
47use crate::errors::RuntimeError;
48use crate::formation::{peer_url_format_error, FormationId, FormationPeer, FormationPeers};
49use crate::mac_token::{split_token, zeroize, B64};
50
51pub const FORMATION_TOKEN_TAG: &str = "mft1";
53
54pub const FORMATION_TOKEN_DOMAIN: &[u8] = b"murmur-formation-token-v1";
57
58const FIELD_SEPARATOR: u8 = 0x1f;
60
61pub const FORMATION_CHANNEL_ENV: &str = "MURMUR_FORMATION_CHANNEL";
64
65pub const FORMATION_CHANNEL_FIRST_LINE_DEADLINE: Duration = Duration::from_secs(10);
67
68pub const FORMATION_ADDRESS_WAIT: Duration = Duration::from_secs(10);
71
72pub const FORMATION_PEER_DOMAIN: &str = "formation.invalid";
75
76pub const FORMATION_CALL_SCOPES: [&str; 3] = ["message/send", "tasks/get", "tasks/cancel"];
82
83pub const FORMATION_CREDENTIAL_PREFIX: &str = "member:";
86
87const MAX_CHANNEL_LINE_BYTES: u64 = 1 << 20;
89
90const KEY_BYTES: usize = 32;
92
93const SIGNATURE_BYTES: usize = 64;
95
96#[derive(Clone, PartialEq, Eq)]
103pub struct FormationToken(String);
104
105impl FormationToken {
106 pub fn expose(&self) -> &str {
109 &self.0
110 }
111}
112
113impl std::fmt::Debug for FormationToken {
114 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
115 f.write_str("FormationToken(<redacted>)")
116 }
117}
118
119fn signed_input(payload_b64: &str) -> Vec<u8> {
121 let mut input = Vec::with_capacity(FORMATION_TOKEN_DOMAIN.len() + payload_b64.len() + 6);
122 input.extend_from_slice(FORMATION_TOKEN_DOMAIN);
123 input.push(FIELD_SEPARATOR);
124 input.extend_from_slice(FORMATION_TOKEN_TAG.as_bytes());
125 input.push(FIELD_SEPARATOR);
126 input.extend_from_slice(payload_b64.as_bytes());
127 input
128}
129
130struct Claims {
132 formation: String,
133 from: String,
134 to: String,
135}
136
137fn open(key: &FormationVerifyKey, token: &str) -> Option<Claims> {
139 let (payload_b64, payload, signature) = split_token(FORMATION_TOKEN_TAG, token).ok()?;
140 if signature.len() != SIGNATURE_BYTES {
141 return None;
142 }
143 UnparsedPublicKey::new(&ED25519, &key.0)
144 .verify(&signed_input(payload_b64), &signature)
145 .ok()?;
146 let payload: Value = serde_json::from_slice(&payload).ok()?;
147 let object = payload.as_object()?;
148 if object.len() != 3 {
149 return None;
150 }
151 let field = |name: &str| object.get(name)?.as_str().map(str::to_string);
152 let claims = Claims {
153 formation: field("formation")?,
154 from: field("from")?,
155 to: field("to")?,
156 };
157 let names_are_members = murmur_artifact::member_name_format_error(&claims.from).is_none()
158 && murmur_artifact::member_name_format_error(&claims.to).is_none();
159 names_are_members.then_some(claims)
160}
161
162pub struct FormationAuthority {
170 formation_id: FormationId,
171 key: Ed25519KeyPair,
172}
173
174impl std::fmt::Debug for FormationAuthority {
175 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
177 f.debug_struct("FormationAuthority")
178 .field("formation_id", &self.formation_id.as_str())
179 .finish_non_exhaustive()
180 }
181}
182
183impl FormationAuthority {
184 pub fn generate(formation_id: &FormationId) -> Result<Self, String> {
188 let mut seed = [0u8; KEY_BYTES];
189 getrandom::fill(&mut seed)
190 .map_err(|error| format!("failed to generate a formation key: {error}"))?;
191 let key = Ed25519KeyPair::from_seed_unchecked(&seed);
192 zeroize(&mut seed);
193 let key = key.map_err(|_| "failed to build a formation key".to_string())?;
194 Ok(Self {
195 formation_id: formation_id.clone(),
196 key,
197 })
198 }
199
200 pub fn formation_id(&self) -> &FormationId {
201 &self.formation_id
202 }
203
204 pub fn mint(&self, from: &str, to: &str) -> FormationToken {
206 let payload = serde_json::json!({
207 "formation": self.formation_id.as_str(),
208 "from": from,
209 "to": to,
210 });
211 let payload_b64 = B64.encode(payload.to_string());
212 let signature = self.key.sign(&signed_input(&payload_b64));
213 FormationToken(format!(
214 "{FORMATION_TOKEN_TAG}.{payload_b64}.{}",
215 B64.encode(signature.as_ref())
216 ))
217 }
218
219 pub fn verify_key(&self) -> FormationVerifyKey {
221 let mut bytes = [0u8; KEY_BYTES];
222 bytes.copy_from_slice(self.key.public_key().as_ref());
223 FormationVerifyKey(bytes)
224 }
225
226 pub fn member_bundle(&self, member: &str, callees: &[&str]) -> FormationBundle {
229 FormationBundle {
230 formation_id: self.formation_id.clone(),
231 member: member.to_string(),
232 launcher: None,
233 verify_key: self.verify_key(),
234 calls: callees
235 .iter()
236 .map(|callee| (callee.to_string(), self.mint(member, callee)))
237 .collect(),
238 }
239 }
240}
241
242#[cfg(test)]
243impl FormationAuthority {
244 pub(crate) fn for_test() -> Self {
246 Self::generate(&FormationId::mint()).expect("the OS hands out randomness")
247 }
248
249 pub(crate) fn verifier_for(&self, member: &str) -> FormationVerifier {
251 FormationVerifier::new(
252 self.formation_id.clone(),
253 member.to_string(),
254 self.verify_key(),
255 )
256 }
257}
258
259#[derive(Clone, PartialEq, Eq)]
261pub struct FormationVerifyKey([u8; KEY_BYTES]);
262
263impl FormationVerifyKey {
264 pub fn render(&self) -> String {
266 B64.encode(self.0)
267 }
268
269 pub fn parse(value: &str) -> Result<Self, String> {
271 let bytes = B64
272 .decode(value)
273 .map_err(|_| "it is not base64url".to_string())?;
274 let bytes: [u8; KEY_BYTES] = bytes
275 .try_into()
276 .map_err(|_| format!("it is not {KEY_BYTES} bytes"))?;
277 Ok(Self(bytes))
278 }
279}
280
281impl std::fmt::Debug for FormationVerifyKey {
282 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
283 f.write_str("FormationVerifyKey(..)")
284 }
285}
286
287#[derive(Clone)]
292pub struct FormationVerifier {
293 formation_id: FormationId,
294 member: String,
295 key: FormationVerifyKey,
296}
297
298impl std::fmt::Debug for FormationVerifier {
299 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
300 f.debug_struct("FormationVerifier")
301 .field("formation_id", &self.formation_id.as_str())
302 .field("member", &self.member)
303 .finish_non_exhaustive()
304 }
305}
306
307#[derive(Debug, Clone, PartialEq, Eq)]
309pub struct FormationCaller {
310 from: String,
311}
312
313impl FormationCaller {
314 pub fn from(&self) -> &str {
316 &self.from
317 }
318}
319
320#[derive(Debug, Clone, PartialEq, Eq)]
322pub enum FormationRefusal {
323 Invalid,
326 NotPermitted { caller: String },
329}
330
331impl FormationVerifier {
332 pub fn new(formation_id: FormationId, member: String, key: FormationVerifyKey) -> Self {
333 Self {
334 formation_id,
335 member,
336 key,
337 }
338 }
339
340 pub fn verify(&self, token: &str) -> Result<FormationCaller, FormationRefusal> {
343 let claims = open(&self.key, token).ok_or(FormationRefusal::Invalid)?;
344 if claims.formation != self.formation_id.as_str() {
345 return Err(FormationRefusal::Invalid);
346 }
347 if claims.to != self.member {
348 return Err(FormationRefusal::NotPermitted {
349 caller: claims.from,
350 });
351 }
352 Ok(FormationCaller { from: claims.from })
353 }
354}
355
356#[derive(Clone)]
366pub struct FormationBundle {
367 pub formation_id: FormationId,
368 pub member: String,
369 pub launcher: Option<crate::running::ProcessIdentity>,
373 pub verify_key: FormationVerifyKey,
374 pub calls: Vec<(String, FormationToken)>,
376}
377
378impl std::fmt::Debug for FormationBundle {
379 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
380 f.debug_struct("FormationBundle")
381 .field("formation_id", &self.formation_id.as_str())
382 .field("member", &self.member)
383 .field("launcher", &self.launcher)
384 .field(
385 "calls",
386 &self
387 .calls
388 .iter()
389 .map(|(name, _)| name.as_str())
390 .collect::<Vec<_>>(),
391 )
392 .finish_non_exhaustive()
393 }
394}
395
396fn channel_unreadable(reason: impl Into<String>) -> RuntimeError {
397 RuntimeError::FormationChannelUnreadable {
398 reason: reason.into(),
399 }
400}
401
402fn json_string(value: &str) -> String {
404 Value::String(value.to_string()).to_string()
405}
406
407impl FormationBundle {
408 pub fn render_line(&self) -> String {
411 let calls: Vec<String> = self
412 .calls
413 .iter()
414 .map(|(name, token)| {
415 format!(
416 "{{\"name\":{},\"token\":{}}}",
417 json_string(name),
418 json_string(token.expose())
419 )
420 })
421 .collect();
422 let launcher = self
423 .launcher
424 .as_ref()
425 .map(|launcher| {
426 format!(
427 ",\"launcher\":{{\"pid\":{},\"process_start\":{}}}",
428 launcher.pid,
429 json_string(&launcher.process_start)
430 )
431 })
432 .unwrap_or_default();
433 format!(
434 "{{\"formation_id\":{},\"member\":{}{launcher},\"verify_key\":{},\"calls\":[{}]}}",
435 json_string(self.formation_id.as_str()),
436 json_string(&self.member),
437 json_string(&self.verify_key.render()),
438 calls.join(",")
439 )
440 }
441
442 pub fn parse_line(line: &str) -> Result<Self, RuntimeError> {
447 let value: Value = serde_json::from_str(line)
448 .map_err(|_| channel_unreadable("its first line is not JSON"))?;
449 let object = value
450 .as_object()
451 .ok_or_else(|| channel_unreadable("its first line is not a JSON object"))?;
452 let string = |name: &str| {
453 object
454 .get(name)
455 .and_then(Value::as_str)
456 .ok_or_else(|| channel_unreadable(format!("its first line has no string {name}")))
457 };
458 let formation_id = FormationId::parse(string("formation_id")?).map_err(|error| {
459 channel_unreadable(match error {
460 RuntimeError::FormationIdUnreadable { reason } => {
461 format!("its first line's formation_id is not a formation id: {reason}")
462 }
463 _ => "its first line's formation_id is not a formation id".to_string(),
464 })
465 })?;
466 let member = string("member")?.to_string();
467 if let Some(problem) = murmur_artifact::member_name_format_error(&member) {
468 return Err(channel_unreadable(format!(
469 "its first line's member is not a member name: it {problem}"
470 )));
471 }
472 let launcher = object.get("launcher").map(parse_launcher).transpose()?;
473 let verify_key = FormationVerifyKey::parse(string("verify_key")?).map_err(|why| {
474 channel_unreadable(format!(
475 "its first line's verify_key is not a formation key: {why}"
476 ))
477 })?;
478 let entries = object
479 .get("calls")
480 .and_then(Value::as_array)
481 .ok_or_else(|| channel_unreadable("its first line has no calls array"))?;
482 let mut calls = Vec::with_capacity(entries.len());
483 let mut seen = HashSet::new();
484 for (index, entry) in entries.iter().enumerate() {
485 let field = |name: &str| {
486 entry.get(name).and_then(Value::as_str).ok_or_else(|| {
487 channel_unreadable(format!(
488 "its first line's calls[{index}] has no string {name}"
489 ))
490 })
491 };
492 let name = field("name")?;
493 if let Some(problem) = murmur_artifact::member_name_format_error(name) {
494 return Err(channel_unreadable(format!(
495 "its first line's calls[{index}].name is not a member name: it {problem}"
496 )));
497 }
498 if !seen.insert(name.to_string()) {
499 return Err(channel_unreadable(format!(
500 "its first line's calls[{index}] names a member a second time"
501 )));
502 }
503 let token = field("token")?;
504 let issued = open(&verify_key, token).is_some_and(|claims| {
505 claims.formation == formation_id.as_str()
506 && claims.from == member
507 && claims.to == name
508 });
509 if !issued {
510 return Err(channel_unreadable(format!(
511 "its first line's calls[{index}].token was not issued to this member for that \
512 callee under the line's own key"
513 )));
514 }
515 calls.push((name.to_string(), FormationToken(token.to_string())));
516 }
517 Ok(Self {
518 formation_id,
519 member,
520 launcher,
521 verify_key,
522 calls,
523 })
524 }
525}
526
527fn parse_launcher(value: &Value) -> Result<crate::running::ProcessIdentity, RuntimeError> {
531 let object = value
532 .as_object()
533 .ok_or_else(|| channel_unreadable("its first line's launcher is not a JSON object"))?;
534 let pid = object
535 .get("pid")
536 .and_then(Value::as_u64)
537 .filter(|pid| (1..=i32::MAX as u64).contains(pid))
538 .and_then(|pid| u32::try_from(pid).ok())
539 .ok_or_else(|| {
540 channel_unreadable("its first line's launcher has no positive integer pid")
541 })?;
542 let process_start = object
543 .get("process_start")
544 .and_then(Value::as_str)
545 .filter(|start| !start.is_empty())
546 .ok_or_else(|| {
547 channel_unreadable("its first line's launcher has no non-empty string process_start")
548 })?;
549 Ok(crate::running::ProcessIdentity {
550 pid,
551 process_start: process_start.to_string(),
552 })
553}
554
555pub fn render_address_line(addresses: &[FormationPeer]) -> String {
557 let entries: Vec<String> = addresses
558 .iter()
559 .map(|peer| {
560 format!(
561 "{{\"name\":{},\"url\":{}}}",
562 json_string(&peer.name),
563 json_string(&peer.url)
564 )
565 })
566 .collect();
567 format!("{{\"addresses\":[{}]}}", entries.join(","))
568}
569
570pub fn parse_address_line(line: &str) -> Result<Vec<FormationPeer>, String> {
572 let value: Value =
573 serde_json::from_str(line).map_err(|_| "an address line is not JSON".to_string())?;
574 let entries = value
575 .get("addresses")
576 .and_then(Value::as_array)
577 .ok_or_else(|| "an address line has no addresses array".to_string())?;
578 entries
579 .iter()
580 .enumerate()
581 .map(|(index, entry)| {
582 let field = |name: &str| {
583 entry
584 .get(name)
585 .and_then(Value::as_str)
586 .map(str::to_string)
587 .ok_or_else(|| format!("addresses[{index}] has no string {name}"))
588 };
589 let peer = FormationPeer {
590 name: field("name")?,
591 url: field("url")?,
592 };
593 if let Some(problem) = murmur_artifact::member_name_format_error(&peer.name) {
594 return Err(format!("addresses[{index}].name {problem}"));
595 }
596 if let Some(problem) = peer_url_format_error(&peer.url) {
597 return Err(format!("addresses[{index}] {problem}"));
598 }
599 Ok(peer)
600 })
601 .collect()
602}
603
604pub fn virtual_url(name: &str) -> String {
606 format!("http://{name}.{FORMATION_PEER_DOMAIN}")
607}
608
609pub fn is_formation_host(host: &str) -> bool {
612 let host = host.to_ascii_lowercase();
613 host == FORMATION_PEER_DOMAIN || host.ends_with(&format!(".{FORMATION_PEER_DOMAIN}"))
614}
615
616pub fn virtual_member(authority: &str) -> Option<&str> {
619 let suffix_len = FORMATION_PEER_DOMAIN.len() + 1;
620 if authority.len() <= suffix_len || !authority.is_char_boundary(authority.len() - suffix_len) {
621 return None;
622 }
623 let (name, suffix) = authority.split_at(authority.len() - suffix_len);
624 let suffix_matches = suffix
625 .strip_prefix('.')
626 .is_some_and(|domain| domain.eq_ignore_ascii_case(FORMATION_PEER_DOMAIN));
627 (suffix_matches && murmur_artifact::member_name_format_error(name).is_none()).then_some(name)
628}
629
630#[derive(Default)]
632struct AddressBook {
633 urls: HashMap<String, String>,
634 closed: bool,
636}
637
638pub struct FormationMember {
641 verifier: FormationVerifier,
642 launcher: Option<crate::running::ProcessIdentity>,
644 calls: Vec<(String, FormationToken)>,
646 book: Mutex<AddressBook>,
647 arrived: Condvar,
648 address_wait: Duration,
650}
651
652impl std::fmt::Debug for FormationMember {
653 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
654 f.debug_struct("FormationMember")
655 .field("formation_id", &self.verifier.formation_id.as_str())
656 .field("member", &self.verifier.member)
657 .field("callees", &self.callees().collect::<Vec<_>>())
658 .finish_non_exhaustive()
659 }
660}
661
662impl FormationMember {
663 pub fn from_bundle(bundle: FormationBundle) -> Self {
665 Self {
666 verifier: FormationVerifier::new(bundle.formation_id, bundle.member, bundle.verify_key),
667 launcher: bundle.launcher,
668 calls: bundle.calls,
669 book: Mutex::new(AddressBook::default()),
670 arrived: Condvar::new(),
671 address_wait: FORMATION_ADDRESS_WAIT,
672 }
673 }
674
675 #[cfg(test)]
677 pub(crate) fn with_address_wait(mut self, wait: Duration) -> Self {
678 self.address_wait = wait;
679 self
680 }
681
682 pub fn address_wait(&self) -> Duration {
684 self.address_wait
685 }
686
687 pub fn from_env(formation_id: Option<&FormationId>) -> Result<Option<Arc<Self>>, RuntimeError> {
695 let Some(raw) = std::env::var_os(FORMATION_CHANNEL_ENV) else {
696 return Ok(None);
697 };
698 let Some(formation_id) = formation_id else {
699 return Err(channel_unreadable(
700 "it is set, but MURMUR_FORMATION_ID is not, and a formation launcher always sets \
701 both",
702 ));
703 };
704 let fd: i32 = raw
705 .to_str()
706 .filter(|digits| !digits.is_empty() && digits.bytes().all(|b| b.is_ascii_digit()))
707 .and_then(|digits| digits.parse().ok())
708 .ok_or_else(|| channel_unreadable("it does not name a file descriptor by number"))?;
709 let channel = adopt_channel_fd(fd)?;
710 Self::from_channel(channel, formation_id, FORMATION_CHANNEL_FIRST_LINE_DEADLINE).map(Some)
711 }
712
713 pub fn from_channel<R: Read + Send + 'static>(
716 channel: R,
717 formation_id: &FormationId,
718 deadline: Duration,
719 ) -> Result<Arc<Self>, RuntimeError> {
720 type FirstLine = Result<Option<String>, &'static str>;
721 let (first_tx, first_rx) = mpsc::sync_channel::<FirstLine>(1);
722 let (member_tx, member_rx) = mpsc::sync_channel::<Arc<Self>>(1);
723 std::thread::Builder::new()
724 .name("formation-channel".to_string())
725 .spawn(move || {
726 let mut reader = BufReader::new(channel);
727 let first = read_channel_line(&mut reader);
728 let read_one = matches!(first, Ok(Some(_)));
729 if first_tx.send(first).is_err() || !read_one {
730 return;
731 }
732 let Ok(member) = member_rx.recv() else {
733 return;
734 };
735 while let Ok(Some(line)) = read_channel_line(&mut reader) {
736 member.apply_address_line(&line);
737 }
738 member.close();
739 })
740 .map_err(|error| {
741 channel_unreadable(format!(
742 "the thread that reads it could not be started: {error}"
743 ))
744 })?;
745 let line = match first_rx.recv_timeout(deadline) {
746 Ok(Ok(Some(line))) => line,
747 Ok(Ok(None)) | Err(mpsc::RecvTimeoutError::Disconnected) => {
748 return Err(channel_unreadable(
749 "the channel ended before its first line arrived",
750 ))
751 }
752 Ok(Err(why)) => return Err(channel_unreadable(why)),
753 Err(mpsc::RecvTimeoutError::Timeout) => {
754 return Err(channel_unreadable(format!(
755 "no first line arrived within {}s",
756 deadline.as_secs_f32()
757 )))
758 }
759 };
760 let bundle = FormationBundle::parse_line(&line)?;
761 if bundle.formation_id != *formation_id {
762 return Err(channel_unreadable(
763 "its first line names another formation than MURMUR_FORMATION_ID",
764 ));
765 }
766 let member = Arc::new(Self::from_bundle(bundle));
767 let _ = member_tx.send(Arc::clone(&member));
768 Ok(member)
769 }
770
771 pub fn name(&self) -> &str {
773 &self.verifier.member
774 }
775
776 pub fn formation_id(&self) -> &FormationId {
777 &self.verifier.formation_id
778 }
779
780 pub fn launcher(&self) -> Option<&crate::running::ProcessIdentity> {
783 self.launcher.as_ref()
784 }
785
786 pub fn verifier(&self) -> &FormationVerifier {
788 &self.verifier
789 }
790
791 pub fn callees(&self) -> impl Iterator<Item = &str> {
793 self.calls.iter().map(|(name, _)| name.as_str())
794 }
795
796 pub fn guest_peers(&self) -> Option<FormationPeers> {
799 let peers: Vec<FormationPeer> = self
800 .callees()
801 .map(|name| FormationPeer {
802 name: name.to_string(),
803 url: virtual_url(name),
804 })
805 .collect();
806 if peers.is_empty() {
807 return None;
808 }
809 FormationPeers::new(peers).ok()
810 }
811
812 pub fn resolve(&self, name: &str, wait: Duration) -> Option<(String, FormationToken)> {
816 let token = self
817 .calls
818 .iter()
819 .find(|(callee, _)| callee == name)
820 .map(|(_, token)| token.clone())?;
821 let book = self.book.lock().unwrap_or_else(PoisonError::into_inner);
822 let (book, _) = self
823 .arrived
824 .wait_timeout_while(book, wait, |book| {
825 !book.closed && !book.urls.contains_key(name)
826 })
827 .unwrap_or_else(PoisonError::into_inner);
828 book.urls.get(name).map(|url| (url.clone(), token))
829 }
830
831 pub fn install_addresses(&self, addresses: Vec<FormationPeer>) {
834 let mut book = self.book.lock().unwrap_or_else(PoisonError::into_inner);
835 for peer in addresses {
836 if self.calls.iter().any(|(callee, _)| *callee == peer.name) {
837 book.urls.insert(peer.name, peer.url);
838 }
839 }
840 drop(book);
841 self.arrived.notify_all();
842 }
843
844 fn apply_address_line(&self, line: &str) {
845 match parse_address_line(line) {
846 Ok(addresses) => self.install_addresses(addresses),
847 Err(why) => crate::runtime_err!(
848 "[capsule-runtime] the formation channel sent an address line that could not be \
849 read ({why}); it was ignored"
850 ),
851 }
852 }
853
854 fn close(&self) {
856 self.book
857 .lock()
858 .unwrap_or_else(PoisonError::into_inner)
859 .closed = true;
860 self.arrived.notify_all();
861 }
862}
863
864fn read_channel_line(reader: &mut impl BufRead) -> Result<Option<String>, &'static str> {
866 let mut line = Vec::new();
867 let read = reader
868 .by_ref()
869 .take(MAX_CHANNEL_LINE_BYTES)
870 .read_until(b'\n', &mut line)
871 .map_err(|_| "the channel could not be read")?;
872 if read == 0 {
873 return Ok(None);
874 }
875 if line.last() == Some(&b'\n') {
876 line.pop();
877 if line.last() == Some(&b'\r') {
878 line.pop();
879 }
880 } else if read as u64 >= MAX_CHANNEL_LINE_BYTES {
881 return Err("a line is longer than the channel accepts");
882 }
883 String::from_utf8(line)
884 .map(Some)
885 .map_err(|_| "a line is not UTF-8")
886}
887
888#[cfg(unix)]
892#[allow(unsafe_code)]
893fn adopt_channel_fd(fd: i32) -> Result<std::fs::File, RuntimeError> {
894 use std::os::fd::FromRawFd;
895 use std::os::unix::fs::FileTypeExt;
896
897 if fd <= 2 {
898 return Err(channel_unreadable(format!(
899 "descriptor {fd} is standard input, output or error, never a channel"
900 )));
901 }
902 let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
904 if flags == -1 {
905 return Err(channel_unreadable(format!("descriptor {fd} is not open")));
906 }
907 if unsafe { libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC) } == -1 {
909 return Err(channel_unreadable(format!(
910 "descriptor {fd} could not be marked close-on-exec"
911 )));
912 }
913 let channel = unsafe { std::fs::File::from_raw_fd(fd) };
917 let kind = channel
918 .metadata()
919 .map_err(|_| channel_unreadable(format!("descriptor {fd} could not be examined")))?
920 .file_type();
921 if !kind.is_fifo() && !kind.is_file() {
922 return Err(channel_unreadable(format!(
923 "descriptor {fd} is neither a pipe nor a file"
924 )));
925 }
926 Ok(channel)
927}
928
929#[cfg(not(unix))]
930fn adopt_channel_fd(_fd: i32) -> Result<std::fs::File, RuntimeError> {
931 Err(channel_unreadable(
932 "an inherited descriptor channel is supported on Unix hosts only",
933 ))
934}
935
936#[cfg(test)]
937mod tests {
938 use super::*;
939 use std::time::Instant;
940
941 fn chain_roster() -> (FormationAuthority, Vec<(&'static str, Vec<&'static str>)>) {
943 let authority = FormationAuthority::generate(&FormationId::mint()).unwrap();
944 (
945 authority,
946 vec![
947 ("planner", vec!["coder"]),
948 ("coder", vec!["reviewer"]),
949 ("reviewer", vec![]),
950 ],
951 )
952 }
953
954 fn payload_of(token: &FormationToken) -> Value {
955 let payload = token.expose().split('.').nth(1).unwrap();
956 serde_json::from_slice(&B64.decode(payload).unwrap()).unwrap()
957 }
958
959 #[test]
960 fn a_token_is_mft1_over_exactly_the_three_claims() {
961 let (authority, _) = chain_roster();
962 let token = authority.mint("planner", "coder");
963 assert!(token.expose().starts_with("mft1."));
964 assert_eq!(token.expose().split('.').count(), 3);
965 let payload = token.expose().split('.').nth(1).unwrap();
966 assert_eq!(
967 String::from_utf8(B64.decode(payload).unwrap()).unwrap(),
968 format!(
969 r#"{{"formation":"{}","from":"planner","to":"coder"}}"#,
970 authority.formation_id()
971 )
972 );
973 let signature = token.expose().split('.').nth(2).unwrap();
974 assert_eq!(B64.decode(signature).unwrap().len(), SIGNATURE_BYTES);
975 }
976
977 #[test]
978 fn a_token_verifies_at_its_audience_and_is_not_permitted_anywhere_else() {
979 let (authority, _) = chain_roster();
980 let token = authority.mint("planner", "coder");
981 assert_eq!(
982 authority
983 .verifier_for("coder")
984 .verify(token.expose())
985 .unwrap()
986 .from(),
987 "planner"
988 );
989 for door in ["reviewer", "planner"] {
990 assert_eq!(
991 authority.verifier_for(door).verify(token.expose()),
992 Err(FormationRefusal::NotPermitted {
993 caller: "planner".to_string()
994 }),
995 "{door}"
996 );
997 }
998 }
999
1000 #[test]
1001 fn every_forgery_is_the_one_invalid() {
1002 let (authority, _) = chain_roster();
1003 let coder = authority.verifier_for("coder");
1004 let token = authority.mint("planner", "coder");
1005 let (_, payload, signature) = {
1006 let mut parts = token.expose().split('.');
1007 (
1008 parts.next().unwrap(),
1009 parts.next().unwrap(),
1010 parts.next().unwrap(),
1011 )
1012 };
1013
1014 let other = FormationAuthority::generate(&FormationId::mint()).unwrap();
1016 let foreign = other.mint("planner", "coder");
1017 let readdressed = format!(
1019 "mft1.{}.{signature}",
1020 B64.encode(
1021 format!(
1022 r#"{{"formation":"{}","from":"planner","to":"reviewer"}}"#,
1023 authority.formation_id()
1024 )
1025 .as_bytes()
1026 )
1027 );
1028 let mut bytes = B64.decode(signature).unwrap();
1030 bytes[7] ^= 0x01;
1031 let flipped = format!("mft1.{payload}.{}", B64.encode(&bytes));
1032 let mut same_key_other_formation = authority.mint("planner", "coder");
1034 same_key_other_formation.0 = {
1035 let other_id = FormationId::mint();
1036 let forged_payload = B64.encode(format!(
1037 r#"{{"formation":"{other_id}","from":"planner","to":"coder"}}"#
1038 ));
1039 let signature = authority.key.sign(&signed_input(&forged_payload));
1040 format!("mft1.{forged_payload}.{}", B64.encode(signature.as_ref()))
1041 };
1042 let extra = {
1044 let forged_payload = B64.encode(format!(
1045 r#"{{"formation":"{}","from":"planner","to":"coder","admin":"yes"}}"#,
1046 authority.formation_id()
1047 ));
1048 let signature = authority.key.sign(&signed_input(&forged_payload));
1049 format!("mft1.{forged_payload}.{}", B64.encode(signature.as_ref()))
1050 };
1051 for forged in [
1052 foreign.expose().to_string(),
1053 readdressed,
1054 flipped,
1055 same_key_other_formation.expose().to_string(),
1056 extra,
1057 format!("mdt1.{payload}.{signature}"),
1058 format!("mft1.{payload}"),
1059 format!("mft1.{payload}.{signature}.x"),
1060 "mft1.x.y".to_string(),
1061 "garbage".to_string(),
1062 String::new(),
1063 ] {
1064 assert_eq!(
1065 coder.verify(&forged),
1066 Err(FormationRefusal::Invalid),
1067 "{forged}"
1068 );
1069 }
1070 assert_eq!(
1072 FormationVerifier::new(
1073 other.formation_id().clone(),
1074 "coder".to_string(),
1075 other.verify_key()
1076 )
1077 .verify(token.expose()),
1078 Err(FormationRefusal::Invalid)
1079 );
1080 }
1081
1082 #[test]
1085 fn a_bundle_holds_exactly_its_callees_tokens_in_roster_order() {
1086 let (authority, roster) = chain_roster();
1087 for (member, callees) in &roster {
1088 let bundle = authority.member_bundle(member, callees);
1089 let names: Vec<&str> = bundle.calls.iter().map(|(name, _)| name.as_str()).collect();
1090 assert_eq!(&names, callees, "{member}");
1091 for (callee, token) in &bundle.calls {
1092 assert_eq!(
1093 authority
1094 .verifier_for(callee)
1095 .verify(token.expose())
1096 .unwrap()
1097 .from(),
1098 *member
1099 );
1100 assert_eq!(payload_of(token)["to"], callee.as_str());
1101 }
1102 }
1103 let planner = authority.member_bundle("planner", &["coder"]).render_line();
1104 assert!(!planner.contains("reviewer"), "{planner}");
1105
1106 let all = FormationAuthority::generate(&FormationId::mint()).unwrap();
1108 let bundle = all.member_bundle("coder", &["reviewer", "tester"]);
1109 let names: Vec<&str> = bundle.calls.iter().map(|(n, _)| n.as_str()).collect();
1110 assert_eq!(names, ["reviewer", "tester"]);
1111 }
1112
1113 #[test]
1114 fn a_first_line_round_trips_and_its_field_order_is_fixed() {
1115 let (authority, _) = chain_roster();
1116 let bundle = authority.member_bundle("coder", &["reviewer"]);
1117 let line = bundle.render_line();
1118 assert!(
1119 line.starts_with(&format!(
1120 "{{\"formation_id\":\"{}\",\"member\":\"coder\",\"verify_key\":\"",
1121 authority.formation_id()
1122 )),
1123 "{line}"
1124 );
1125 assert!(line.contains(",\"calls\":[{\"name\":\"reviewer\",\"token\":\"mft1."));
1126 assert!(!line.contains('\n'));
1127 let back = FormationBundle::parse_line(&line).unwrap();
1128 assert_eq!(back.formation_id, bundle.formation_id);
1129 assert_eq!(back.member, "coder");
1130 assert_eq!(back.verify_key, authority.verify_key());
1131 assert_eq!(back.calls, bundle.calls);
1132 let empty = authority.member_bundle("reviewer", &[]).render_line();
1133 assert!(empty.ends_with("\"calls\":[]}"), "{empty}");
1134 assert!(FormationBundle::parse_line(&empty)
1135 .unwrap()
1136 .calls
1137 .is_empty());
1138 }
1139
1140 #[test]
1143 fn a_first_line_carries_its_launcher_when_one_is_named() {
1144 let (authority, _) = chain_roster();
1145 let launcher = crate::running::ProcessIdentity {
1146 pid: 4242,
1147 process_start: "987654".to_string(),
1148 };
1149 let mut bundle = authority.member_bundle("coder", &["reviewer"]);
1150 assert_eq!(bundle.launcher, None);
1151 bundle.launcher = Some(launcher.clone());
1152 let line = bundle.render_line();
1153 assert!(
1154 line.starts_with(&format!(
1155 "{{\"formation_id\":\"{}\",\"member\":\"coder\",\
1156 \"launcher\":{{\"pid\":4242,\"process_start\":\"987654\"}},\"verify_key\":\"",
1157 authority.formation_id()
1158 )),
1159 "{line}"
1160 );
1161 let back = FormationBundle::parse_line(&line).unwrap();
1162 assert_eq!(back.launcher, Some(launcher.clone()));
1163 assert_eq!(back.calls, bundle.calls);
1164 assert_eq!(
1165 FormationMember::from_bundle(back).launcher(),
1166 Some(&launcher)
1167 );
1168
1169 let without = authority
1170 .member_bundle("coder", &["reviewer"])
1171 .render_line();
1172 assert!(!without.contains("launcher"), "{without}");
1173 let back = FormationBundle::parse_line(&without).unwrap();
1174 assert_eq!(back.launcher, None);
1175 assert_eq!(FormationMember::from_bundle(back).launcher(), None);
1176 }
1177
1178 #[test]
1179 fn a_malformed_launcher_is_refused_naming_launcher() {
1180 let (authority, _) = chain_roster();
1181 let good: Value = serde_json::from_str(
1182 &authority
1183 .member_bundle("coder", &["reviewer"])
1184 .render_line(),
1185 )
1186 .unwrap();
1187 let cases = [
1188 (serde_json::json!(4242), "launcher is not a JSON object"),
1189 (Value::Null, "launcher is not a JSON object"),
1190 (
1191 serde_json::json!({"process_start": "1"}),
1192 "launcher has no positive integer pid",
1193 ),
1194 (
1195 serde_json::json!({"pid": 0, "process_start": "1"}),
1196 "launcher has no positive integer pid",
1197 ),
1198 (
1199 serde_json::json!({"pid": -5, "process_start": "1"}),
1200 "launcher has no positive integer pid",
1201 ),
1202 (
1203 serde_json::json!({"pid": 2_147_483_648_u64, "process_start": "1"}),
1204 "launcher has no positive integer pid",
1205 ),
1206 (
1207 serde_json::json!({"pid": 1.5, "process_start": "1"}),
1208 "launcher has no positive integer pid",
1209 ),
1210 (
1211 serde_json::json!({"pid": "42", "process_start": "1"}),
1212 "launcher has no positive integer pid",
1213 ),
1214 (
1215 serde_json::json!({"pid": 42}),
1216 "launcher has no non-empty string process_start",
1217 ),
1218 (
1219 serde_json::json!({"pid": 42, "process_start": ""}),
1220 "launcher has no non-empty string process_start",
1221 ),
1222 (
1223 serde_json::json!({"pid": 42, "process_start": 7}),
1224 "launcher has no non-empty string process_start",
1225 ),
1226 ];
1227 for (launcher, wording) in cases {
1228 let mut line = good.clone();
1229 line["launcher"] = launcher.clone();
1230 let message = line_refusal(&line.to_string());
1231 assert!(message.contains(wording), "{launcher}: {message}");
1232 }
1233 let mut largest = good.clone();
1234 largest["launcher"] = serde_json::json!({"pid": i32::MAX, "process_start": "1"});
1235 assert_eq!(
1236 FormationBundle::parse_line(&largest.to_string())
1237 .unwrap()
1238 .launcher
1239 .map(|launcher| launcher.pid),
1240 Some(i32::MAX as u32)
1241 );
1242 }
1243
1244 fn line_refusal(line: &str) -> String {
1245 match FormationBundle::parse_line(line) {
1246 Err(error @ RuntimeError::FormationChannelUnreadable { .. }) => error.to_string(),
1247 other => panic!("expected FormationChannelUnreadable, got {other:?}"),
1248 }
1249 }
1250
1251 #[test]
1253 fn a_first_line_refusal_names_the_problem_and_never_quotes_the_line() {
1254 let (authority, _) = chain_roster();
1255 let good: Value = serde_json::from_str(
1256 &authority
1257 .member_bundle("coder", &["reviewer"])
1258 .render_line(),
1259 )
1260 .unwrap();
1261 let token = good["calls"][0]["token"].as_str().unwrap().to_string();
1262 let with = |field: &str, value: Value| {
1263 let mut line = good.clone();
1264 line[field] = value;
1265 line.to_string()
1266 };
1267 let other = FormationAuthority::generate(&FormationId::mint()).unwrap();
1268 let cases = [
1269 ("not json".to_string(), "is not JSON"),
1270 ("[1]".to_string(), "is not a JSON object"),
1271 (with("formation_id", json_str("frm_XYZ")), "formation_id"),
1272 (
1273 with("member", json_str("Coder")),
1274 "member is not a member name",
1275 ),
1276 (with("verify_key", json_str("abc")), "verify_key is not"),
1277 (
1278 with("verify_key", json_str(&other.verify_key().render())),
1279 "calls[0].token was not issued",
1280 ),
1281 (with("calls", Value::Null), "no calls array"),
1282 (
1283 with(
1284 "calls",
1285 serde_json::json!([{"name": "Reviewer", "token": token}]),
1286 ),
1287 "calls[0].name is not a member name",
1288 ),
1289 (
1290 with(
1291 "calls",
1292 serde_json::json!([{"name": "tester", "token": token}]),
1293 ),
1294 "calls[0].token was not issued",
1295 ),
1296 (
1297 with(
1298 "calls",
1299 serde_json::json!([
1300 {"name": "reviewer", "token": token},
1301 {"name": "reviewer", "token": token}
1302 ]),
1303 ),
1304 "calls[1] names a member a second time",
1305 ),
1306 ];
1307 for (line, wording) in cases {
1308 let message = line_refusal(&line);
1309 assert!(message.contains(wording), "{wording}: {message}");
1310 assert!(!message.contains("mft1."), "{message}");
1311 assert!(!message.contains(&token), "{message}");
1312 assert!(
1313 !message.contains(&authority.verify_key().render()),
1314 "{message}"
1315 );
1316 assert!(message.contains(FORMATION_CHANNEL_ENV), "{message}");
1317 }
1318 }
1319
1320 fn json_str(value: &str) -> Value {
1321 Value::String(value.to_string())
1322 }
1323
1324 #[test]
1325 fn an_address_line_round_trips_and_refuses_what_is_not_a_door() {
1326 let peers = vec![FormationPeer {
1327 name: "reviewer".to_string(),
1328 url: "http://127.0.0.1:41873".to_string(),
1329 }];
1330 let line = render_address_line(&peers);
1331 assert_eq!(
1332 line,
1333 r#"{"addresses":[{"name":"reviewer","url":"http://127.0.0.1:41873"}]}"#
1334 );
1335 assert_eq!(parse_address_line(&line).unwrap(), peers);
1336 for bad in [
1337 r#"{"addresses":[{"name":"reviewer","url":"https://127.0.0.1:1"}]}"#,
1338 r#"{"addresses":[{"name":"reviewer","url":"http://127.0.0.1"}]}"#,
1339 r#"{"addresses":[{"name":"Reviewer","url":"http://127.0.0.1:1"}]}"#,
1340 r#"{"addresses":{}}"#,
1341 "nope",
1342 ] {
1343 assert!(parse_address_line(bad).is_err(), "{bad}");
1344 }
1345 }
1346
1347 #[test]
1348 fn a_virtual_address_names_one_member_and_nothing_else() {
1349 assert_eq!(virtual_url("coder"), "http://coder.formation.invalid");
1350 assert_eq!(virtual_member("coder.formation.invalid"), Some("coder"));
1351 assert_eq!(virtual_member("coder.FORMATION.invalid"), Some("coder"));
1352 for authority in [
1353 "coder.formation.invalid:80",
1354 "u@coder.formation.invalid",
1355 "a.coder.formation.invalid",
1356 ".formation.invalid",
1357 "formation.invalid",
1358 "Coder.formation.invalid",
1359 "coder.formation.invalid.",
1360 "coder.example.com",
1361 "",
1362 ] {
1363 assert_eq!(virtual_member(authority), None, "{authority}");
1364 }
1365 assert!(is_formation_host("coder.formation.invalid"));
1366 assert!(is_formation_host("x.y.FORMATION.INVALID"));
1367 assert!(is_formation_host("formation.invalid"));
1368 assert!(!is_formation_host("formation.invalid.example"));
1369 assert!(!is_formation_host("notformation.invalid"));
1370 }
1371
1372 #[test]
1374 fn a_members_guests_see_its_callees_at_virtual_addresses_only() {
1375 let (authority, _) = chain_roster();
1376 let planner = FormationMember::from_bundle(authority.member_bundle("planner", &["coder"]));
1377 let peers = planner.guest_peers().unwrap();
1378 assert_eq!(peers.render(), "coder=http://coder.formation.invalid");
1379 let reviewer = FormationMember::from_bundle(authority.member_bundle("reviewer", &[]));
1380 assert!(reviewer.guest_peers().is_none());
1381 }
1382
1383 #[test]
1384 fn an_address_resolves_only_for_a_callee_and_waits_for_a_late_one() {
1385 let (authority, _) = chain_roster();
1386 let member = Arc::new(FormationMember::from_bundle(
1387 authority.member_bundle("coder", &["reviewer"]),
1388 ));
1389 assert!(member
1390 .resolve("reviewer", Duration::from_millis(50))
1391 .is_none());
1392 member.install_addresses(vec![FormationPeer {
1393 name: "planner".to_string(),
1394 url: "http://127.0.0.1:9".to_string(),
1395 }]);
1396 assert!(
1397 member.resolve("planner", Duration::ZERO).is_none(),
1398 "no route without a token"
1399 );
1400 let late = Arc::clone(&member);
1401 let writer = std::thread::spawn(move || {
1402 std::thread::sleep(Duration::from_millis(300));
1403 late.install_addresses(vec![FormationPeer {
1404 name: "reviewer".to_string(),
1405 url: "http://127.0.0.1:41873".to_string(),
1406 }]);
1407 });
1408 let (url, token) = member.resolve("reviewer", Duration::from_secs(10)).unwrap();
1409 writer.join().unwrap();
1410 assert_eq!(url, "http://127.0.0.1:41873");
1411 assert_eq!(
1412 authority
1413 .verifier_for("reviewer")
1414 .verify(token.expose())
1415 .unwrap()
1416 .from(),
1417 "coder"
1418 );
1419 member.close();
1420 }
1421
1422 #[test]
1423 fn a_channel_delivers_the_first_line_then_addresses_until_it_ends() {
1424 let (authority, _) = chain_roster();
1425 let (reader, mut writer) = std::io::pipe().unwrap();
1426 use std::io::Write;
1427 writeln!(
1428 writer,
1429 "{}",
1430 authority
1431 .member_bundle("coder", &["reviewer"])
1432 .render_line()
1433 )
1434 .unwrap();
1435 let member =
1436 FormationMember::from_channel(reader, authority.formation_id(), Duration::from_secs(5))
1437 .unwrap();
1438 assert_eq!(member.name(), "coder");
1439 writeln!(writer, "not an address line").unwrap();
1440 writeln!(
1441 writer,
1442 "{}",
1443 render_address_line(&[FormationPeer {
1444 name: "reviewer".to_string(),
1445 url: "http://127.0.0.1:41873".to_string(),
1446 }])
1447 )
1448 .unwrap();
1449 let (url, _) = member.resolve("reviewer", Duration::from_secs(5)).unwrap();
1450 assert_eq!(url, "http://127.0.0.1:41873");
1451 drop(writer);
1452 let started = Instant::now();
1454 let bare = FormationMember::from_channel(
1455 {
1456 let (reader, mut writer) = std::io::pipe().unwrap();
1457 writeln!(
1458 writer,
1459 "{}",
1460 authority.member_bundle("planner", &["coder"]).render_line()
1461 )
1462 .unwrap();
1463 reader
1464 },
1465 authority.formation_id(),
1466 Duration::from_secs(5),
1467 )
1468 .unwrap();
1469 assert!(bare.resolve("coder", Duration::from_secs(10)).is_none());
1470 assert!(started.elapsed() < Duration::from_secs(5));
1471 }
1472
1473 #[test]
1474 fn a_channel_refuses_a_late_empty_or_foreign_first_line() {
1475 let (authority, _) = chain_roster();
1476 let (reader, _writer) = std::io::pipe().unwrap();
1477 let late = FormationMember::from_channel(
1478 reader,
1479 authority.formation_id(),
1480 Duration::from_millis(200),
1481 )
1482 .unwrap_err()
1483 .to_string();
1484 assert!(late.contains("no first line arrived within 0.2s"), "{late}");
1485
1486 let (reader, writer) = std::io::pipe().unwrap();
1487 drop(writer);
1488 let ended =
1489 FormationMember::from_channel(reader, authority.formation_id(), Duration::from_secs(5))
1490 .unwrap_err()
1491 .to_string();
1492 assert!(ended.contains("ended before its first line"), "{ended}");
1493
1494 let foreign = FormationMember::from_channel(
1495 std::io::Cursor::new(format!(
1496 "{}\n",
1497 authority.member_bundle("coder", &[]).render_line()
1498 )),
1499 &FormationId::mint(),
1500 Duration::from_secs(5),
1501 )
1502 .unwrap_err()
1503 .to_string();
1504 assert!(foreign.contains("another formation"), "{foreign}");
1505 }
1506
1507 #[cfg(unix)]
1509 #[test]
1510 #[allow(unsafe_code)]
1511 fn an_adopted_channel_descriptor_is_close_on_exec() {
1512 use std::os::fd::{AsRawFd, IntoRawFd};
1513 let (reader, _writer) = std::io::pipe().unwrap();
1514 let fd = reader.into_raw_fd();
1515 unsafe { libc::fcntl(fd, libc::F_SETFD, 0) };
1517 let adopted = adopt_channel_fd(fd).unwrap();
1518 let flags = unsafe { libc::fcntl(adopted.as_raw_fd(), libc::F_GETFD) };
1520 assert_ne!(flags & libc::FD_CLOEXEC, 0);
1521
1522 assert!(adopt_channel_fd(987_654)
1523 .unwrap_err()
1524 .to_string()
1525 .contains("is not open"));
1526 assert!(adopt_channel_fd(1)
1527 .unwrap_err()
1528 .to_string()
1529 .contains("never a channel"));
1530 }
1531
1532 #[test]
1534 fn no_debug_output_carries_a_token_or_key() {
1535 let (authority, _) = chain_roster();
1536 let mut bundle = authority.member_bundle("coder", &["reviewer"]);
1537 bundle.launcher = crate::running::ProcessIdentity::of_this_process();
1538 let token = bundle.calls[0].1.clone();
1539 let signature = token.expose().split('.').nth(2).unwrap().to_string();
1540 let key = authority.verify_key().render();
1541 let member = FormationMember::from_bundle(bundle.clone());
1542 let outputs = [
1543 format!("{token:?}"),
1544 format!("{bundle:?}"),
1545 format!("{member:?}"),
1546 format!("{:?}", member.verifier()),
1547 format!("{authority:?}"),
1548 format!("{:?}", authority.verify_key()),
1549 ];
1550 assert_eq!(outputs[0], "FormationToken(<redacted>)");
1551 for output in &outputs {
1552 assert!(!output.contains("mft1."), "{output}");
1553 assert!(!output.contains(&signature), "{output}");
1554 assert!(!output.contains(&key), "{output}");
1555 }
1556 assert!(outputs[2].contains("reviewer"), "{}", outputs[2]);
1557 }
1558
1559 #[test]
1561 fn the_channel_is_read_from_the_process_environment_in_one_place() {
1562 let mut sources = crate::source_scan::sources_under(
1563 &crate::source_scan::sibling_crate_src("capsule-runtime"),
1564 );
1565 sources.extend(crate::source_scan::sources_under(
1566 &crate::source_scan::sibling_crate_src("murmur-cli"),
1567 ));
1568 let mut readers = 0;
1569 for (_, text) in sources {
1570 let production = crate::source_scan::production_part(&text);
1571 readers += production.matches("var_os(FORMATION_CHANNEL_ENV)").count()
1572 + production.matches("var(FORMATION_CHANNEL_ENV)").count()
1573 + production
1574 .matches("var_os(\"MURMUR_FORMATION_CHANNEL\")")
1575 .count()
1576 + production
1577 .matches("var(\"MURMUR_FORMATION_CHANNEL\")")
1578 .count();
1579 }
1580 assert_eq!(readers, 1, "FormationMember::from_env is the one reader");
1581 }
1582}