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