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