diff --git a/Cargo.toml b/Cargo.toml index f2ee7a8..d0b0a35 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,7 +21,7 @@ bytes = "1.12.1" chrono = { version = "0.4.45", features = ["serde"] } clap = { version = "4.6.3", features = ["derive"] } futures-util = "0.3.33" -nix = { version = "0.31.3", features = ["hostname", "net", "fs", "user"] } +nix = { version = "0.31.3", features = ["hostname", "net", "fs", "term", "user"] } serde = { version = "1.0.229", features = ["derive"] } tokio = { version = "1.53.0", features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] } toml = "1.1.3" diff --git a/src/bin/wall.rs b/src/bin/wall.rs index d2f63e9..5f0cebf 100644 --- a/src/bin/wall.rs +++ b/src/bin/wall.rs @@ -1,8 +1,12 @@ -use std::io::Read as _; +use std::{ + io::{self, Read as _}, + os::fd::AsFd, +}; use anyhow::Context; use clap::{CommandFactory, Parser}; use clap_complete::{Shell, generate}; +use nix::unistd; use roowho2_lib::server::varlink_api::{ DEFAULT_CLIENT_SERVER_SOCKET_PATH, VarlinkWalldClientProxy, VarlinkWalldClientResponse, }; @@ -52,8 +56,18 @@ async fn main() -> anyhow::Result<()> { format!("failed to connect to roowho2 at {DEFAULT_CLIENT_SERVER_SOCKET_PATH}") })?; + let stdin = io::stdin(); + let source_tty = if !unistd::isatty(stdin.as_fd()).unwrap_or(false) { + None + } else { + // TODO: should we send this as a CStr maybe? + unistd::ttyname(stdin.as_fd()) + .ok() + .map(|p| p.to_string_lossy().to_string()) + }; + let response = conn - .wall(message, args.group, args.nobanner, args.timeout) + .wall(source_tty, message, args.group, args.nobanner, args.timeout) .await .context("varlink call to walld failed")? .map_err(|err| anyhow::format_err!("{err}")) diff --git a/src/bin/write.rs b/src/bin/write.rs index 5459dbb..164af6e 100644 --- a/src/bin/write.rs +++ b/src/bin/write.rs @@ -1,9 +1,13 @@ -use std::io::Read as _; +use std::{ + io::{self, Read as _}, + os::fd::AsFd, +}; use anyhow::Context; use clap::{CommandFactory, Parser}; use clap_complete::{Shell, generate}; +use nix::unistd; use roowho2_lib::server::varlink_api::{ DEFAULT_CLIENT_SERVER_SOCKET_PATH, VarlinkWalldClientProxy, VarlinkWalldClientResponse, }; @@ -49,7 +53,17 @@ async fn main() -> anyhow::Result<()> { .user .expect("required_unless_present=completions guarantees this is set"); - VarlinkWalldClientProxy::write(&mut conn, user, args.ttyname, message) + let stdin = io::stdin(); + let source_tty = if !unistd::isatty(stdin.as_fd()).unwrap_or(false) { + None + } else { + // TODO: should we send this as a CStr maybe? + unistd::ttyname(stdin.as_fd()) + .ok() + .map(|p| p.to_string_lossy().to_string()) + }; + + VarlinkWalldClientProxy::write(&mut conn, source_tty, user, args.ttyname, message) .await .context("varlink call to walld failed")? .map_err(|err| anyhow::format_err!("{err}")) diff --git a/src/server.rs b/src/server.rs index 7b58c8f..65a71c6 100644 --- a/src/server.rs +++ b/src/server.rs @@ -3,3 +3,4 @@ pub mod fingerd; pub mod ignore_list; pub mod rwhod; pub mod varlink_api; +pub mod walld; diff --git a/src/server/varlink_api.rs b/src/server/varlink_api.rs index 7dd96f5..666fe31 100644 --- a/src/server/varlink_api.rs +++ b/src/server/varlink_api.rs @@ -34,6 +34,7 @@ pub enum VarlinkMethod { pub enum VarlinkReply { Rwhod(VarlinkRwhodClientResponse), Finger(VarlinkFingerClientResponse), + Walld(VarlinkWalldClientResponse), } #[derive(Debug, Clone, PartialEq, Serialize)] @@ -42,6 +43,7 @@ pub enum VarlinkReply { pub enum VarlinkReplyError { Rwhod(VarlinkRwhodClientError), Finger(VarlinkFingerClientError), + Walld(VarlinkWalldClientError), } #[derive(Debug, Clone)] @@ -84,13 +86,31 @@ impl zlink::Service for VarlinkRoowhoo2ClientServer { async fn handle<'service>( &'service mut self, call: &'service zlink::Call>, - _conn: &mut zlink::Connection, + conn: &mut zlink::Connection, _fds: Vec, ) -> zlink::service::HandleResult< Self::ReplyParams<'service>, Self::ReplyStream, Self::ReplyError<'service>, > { + let (peer_pid, peer_uid) = match conn.peer_credentials().await { + Ok(creds) => ( + creds.process_id().as_raw_pid() as u32, + creds.unix_user_id().as_raw(), + ), + Err(e) => { + tracing::error!("failed to read peer credentials: {e}"); + // TODO: peercreds are currently only used for walld, but this should be a more "toplevel" + // error at some point when we start using peercred for other things as well. + return ( + MethodReply::Error(VarlinkReplyError::Walld(VarlinkWalldClientError::Io { + message: "failed to identify caller".to_string(), + })), + Default::default(), + ); + } + }; + match call.method() { VarlinkMethod::Rwhod(VarlinkRwhodClientRequest::Rwho { max_idle_seconds }) => { if !self.rwhod_enabled { @@ -250,23 +270,99 @@ impl zlink::Service for VarlinkRoowhoo2ClientServer { Default::default(), ) } - VarlinkMethod::Walld(VarlinkWalldClientRequest::Wall { .. }) => { - tracing::error!("Received Wall request, but walld service is not implemented"); - ( - MethodReply::Error(VarlinkReplyError::Rwhod( - VarlinkRwhodClientError::InvalidRequest, - )), - Default::default(), + VarlinkMethod::Walld(VarlinkWalldClientRequest::Wall { + source_tty, + message, + group, + nobanner, + timeout_secs, + }) => { + // TODO: add walld toggle to config + + // TODO: ensure the outer duration is more than the inner duration. + let result = match timeout( + Duration::from_secs(2), + handle_wall_request( + peer_pid, + peer_uid, + source_tty.clone(), + message.clone(), + group.clone(), + *nobanner, + *timeout_secs, + ), ) + .await + { + Ok(response) => response, + Err(_) => { + tracing::error!("Wall request timed out after 2 seconds"); + return ( + MethodReply::Error(VarlinkReplyError::Walld( + VarlinkWalldClientError::TimedOut, + )), + Default::default(), + ); + } + }; + + match result { + Ok(response) => ( + MethodReply::Single(Some(VarlinkReply::Walld( + VarlinkWalldClientResponse::Wall(response), + ))), + Default::default(), + ), + Err(err) => ( + MethodReply::Error(VarlinkReplyError::Walld(err)), + Default::default(), + ), + } } - VarlinkMethod::Walld(VarlinkWalldClientRequest::Write { .. }) => { - tracing::error!("Received Write request, but walld service is not implemented"); - ( - MethodReply::Error(VarlinkReplyError::Rwhod( - VarlinkRwhodClientError::InvalidRequest, - )), - Default::default(), + VarlinkMethod::Walld(VarlinkWalldClientRequest::Write { + source_tty, + target_user, + target_tty, + message, + }) => { + // TODO: add walld toggle to config + let result = match timeout( + Duration::from_secs(2), + handle_write_request( + peer_pid, + peer_uid, + source_tty.clone(), + target_user.clone(), + target_tty.clone(), + message.clone(), + ), ) + .await + { + Ok(response) => response, + Err(_) => { + tracing::error!("Write request timed out after 2 seconds"); + return ( + MethodReply::Error(VarlinkReplyError::Walld( + VarlinkWalldClientError::TimedOut, + )), + Default::default(), + ); + } + }; + + match result { + Ok(response) => ( + MethodReply::Single(Some(VarlinkReply::Walld( + VarlinkWalldClientResponse::Write(response), + ))), + Default::default(), + ), + Err(err) => ( + MethodReply::Error(VarlinkReplyError::Walld(err)), + Default::default(), + ), + } } } } diff --git a/src/server/varlink_api/walld.rs b/src/server/varlink_api/walld.rs index 5d68ab1..3bf2121 100644 --- a/src/server/varlink_api/walld.rs +++ b/src/server/varlink_api/walld.rs @@ -1,11 +1,39 @@ +use std::time::Duration; + +use nix::unistd::gethostname; use serde::{Deserialize, Serialize}; use zlink::ReplyError; +use crate::server::walld::tty_utils::{self, TtyError}; + +#[zlink::proxy("no.ntnu.pvv.roowho2.walld")] +pub trait VarlinkWalldClientProxy { + /// Broadcast `message` to the tty of every logged in user, optionally restricted to `group`. + async fn wall( + &mut self, + source_tty: Option, + message: String, + group: Option, + nobanner: bool, + timeout_secs: u32, + ) -> zlink::Result>; + + /// Send `message` to a the tty of a single user, optionally targeting a specific tty. + async fn write( + &mut self, + source_tty: Option, + target_user: String, + target_tty: Option, + message: String, + ) -> zlink::Result>; +} + #[derive(Debug, Deserialize)] #[serde(tag = "method", content = "parameters")] pub enum VarlinkWalldClientRequest { #[serde(rename = "no.ntnu.pvv.roowho2.walld.Wall")] Wall { + source_tty: Option, message: String, group: Option, nobanner: bool, @@ -14,32 +42,13 @@ pub enum VarlinkWalldClientRequest { #[serde(rename = "no.ntnu.pvv.roowho2.walld.Write")] Write { - user: String, - tty: Option, + source_tty: Option, + target_user: String, + target_tty: Option, message: String, }, } -#[zlink::proxy("no.ntnu.pvv.roowho2.walld")] -pub trait VarlinkWalldClientProxy { - /// Broadcast `message` to the tty of every logged in user, optionally restricted to `group`. - async fn wall( - &mut self, - message: String, - group: Option, - nobanner: bool, - timeout_secs: u32, - ) -> zlink::Result>; - - /// Send `message` to a the tty of a single `user`, optionally targeting a specific `tty`. - async fn write( - &mut self, - user: String, - tty: Option, - message: String, - ) -> zlink::Result>; -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(untagged)] pub enum VarlinkWalldClientResponse { @@ -122,3 +131,146 @@ impl std::fmt::Display for VarlinkWalldClientError { } } } + +fn username_for_uid(uid: u32) -> String { + users::get_user_by_uid(uid) + .map(|u| u.name().to_string_lossy().into_owned()) + .unwrap_or_else(|| uid.to_string()) +} + +fn hostname() -> String { + gethostname() + .ok() + .and_then(|h| h.into_string().ok()) + .unwrap_or_else(|| "unknown".to_string()) +} + +fn tty_error_to_walld_error(user: &str, tty: &str, err: TtyError) -> VarlinkWalldClientError { + match err { + TtyError::NotFound(_) | TtyError::InvalidName(_) | TtyError::NotCharacterDevice(_) => { + VarlinkWalldClientError::TtyNotFound { + user: user.to_string(), + tty: tty.to_string(), + } + } + TtyError::Unavailable(_) => VarlinkWalldClientError::MessagesDisabled { + user: user.to_string(), + }, + other => VarlinkWalldClientError::Io { + message: other.to_string(), + }, + } +} + +// TODO: have the client send the name of the TTY +// verify that the TTY is owned by the user from peercred. + +pub async fn handle_wall_request( + _source_pid: u32, + source_uid: u32, + source_tty: Option, + message: String, + group: Option, + nobanner: bool, + timeout_secs: u32, +) -> Result { + if let Some(group) = &group + && users::get_group_by_name(group).is_none() + { + return Err(VarlinkWalldClientError::InvalidRequest { + message: format!("unknown group {group:?}"), + }); + } + + // TODO: authorize against polkit + + let body = tty_utils::format_wall_message( + &username_for_uid(source_uid), + &hostname(), + source_tty.as_deref(), + &message, + nobanner, + ); + + let timeout = Duration::from_secs(timeout_secs.max(1) as u64); + let sessions = tty_utils::all_sessions(group.as_deref()); + + let mut response = VarlinkWallResponse::default(); + for session in sessions { + match tty_utils::deliver_message(&session.tty, &body, timeout).await { + Ok(()) => response.delivered.push(VarlinkWallDelivery { + user: session.user, + tty: session.tty, + }), + Err(err) => response.failures.push(VarlinkWallDeliveryFailure { + user: session.user, + tty: session.tty, + reason: err.to_string(), + }), + } + } + + Ok(response) +} + +const DEFAULT_WRITE_TIMEOUT: Duration = Duration::from_secs(30); + +pub async fn handle_write_request( + _source_pid: u32, + source_uid: u32, + source_tty: Option, + user: String, + tty: Option, + message: String, +) -> Result { + // TODO: authorize against polkit + + let sessions = tty_utils::sessions_for_user(&user); + if sessions.is_empty() { + return Err(VarlinkWalldClientError::UserNotLoggedIn { user }); + } + + let target_tty = match &tty { + Some(tty) => { + if !sessions.iter().any(|s| &s.tty == tty) { + return Err(VarlinkWalldClientError::TtyNotFound { + user, + tty: tty.clone(), + }); + } + tty.clone() + } + // No tty requested, pick the one the user touched most recently + None => sessions + .iter() + .filter_map(|s| { + tty_utils::stat_tty(&s.tty) + .ok() + .map(|info| (s.tty.clone(), info)) + }) + .max_by_key(|(_, info)| info.atime) + .map(|(tty, _)| tty) + .ok_or_else(|| VarlinkWalldClientError::MessagesDisabled { user: user.clone() })?, + }; + + match tty_utils::stat_tty(&target_tty) { + Ok(info) if info.writable => {} + Ok(_) => return Err(VarlinkWalldClientError::MessagesDisabled { user }), + Err(_) => { + return Err(VarlinkWalldClientError::TtyNotFound { + user, + tty: target_tty, + }); + } + } + + let source_user = username_for_uid(source_uid); + let body = + tty_utils::format_write_message(&source_user, &hostname(), source_tty.as_deref(), &message); + + tty_utils::deliver_message(&target_tty, &body, DEFAULT_WRITE_TIMEOUT) + .await + .map_err(|err| tty_error_to_walld_error(&user, &target_tty, err))?; + + Ok(VarlinkWriteResponse { tty: target_tty }) +} diff --git a/src/server/walld.rs b/src/server/walld.rs new file mode 100644 index 0000000..f29129c --- /dev/null +++ b/src/server/walld.rs @@ -0,0 +1 @@ +pub mod tty_utils; diff --git a/src/server/walld/tty_utils.rs b/src/server/walld/tty_utils.rs new file mode 100644 index 0000000..0a5780f --- /dev/null +++ b/src/server/walld/tty_utils.rs @@ -0,0 +1,403 @@ +//! Low-level tty discovery, permission checks and message delivery. + +use std::{ + collections::HashSet, + io::Error, + path::{Path, PathBuf}, + time::{Duration, SystemTime}, +}; + +use chrono::Timelike; +use nix::{ + errno::Errno, + fcntl::{self, OFlag}, + sys::stat::{Mode, SFlag, stat}, + unistd, +}; +use tokio::{io::unix::AsyncFd, time}; +use uucore::utmpx::Utmpx; + +const TERM_WIDTH: usize = 79; + +#[derive(Debug, thiserror::Error)] +pub enum TtyError { + #[error("invalid tty name: {0:?}")] + InvalidName(String), + #[error("{0}: no such tty")] + NotFound(String), + #[error("{0}: not a character device")] + NotCharacterDevice(String), + #[error("{0}: messages are disabled on this tty")] + MessagesDisabled(String), + #[error("{0}: device is busy or not accessible")] + Unavailable(String), + #[error("timed out writing to {0}")] + Timeout(String), + #[error("{path}: {source}")] + Io { + path: String, + #[source] + source: Error, + }, +} + +fn validate_tty_name(name: &str) -> Result<(), TtyError> { + if name.is_empty() + || name.starts_with(':') + || name.starts_with('/') + || name.split('/').any(|part| part == ".." || part.is_empty()) + { + return Err(TtyError::InvalidName(name.to_string())); + } + Ok(()) +} + +pub fn tty_device_path(name: &str) -> Result { + validate_tty_name(name)?; + Ok(Path::new("/dev").join(name)) +} + +pub struct TtyInfo { + /// Whether the tty currently accepts unsolicited writes (`mesg y`), signalled by the + /// group-write bit on the device node. + pub writable: bool, + pub atime: SystemTime, +} + +pub fn stat_tty(name: &str) -> Result { + let path = tty_device_path(name)?; + + let st = stat(&path).map_err(|e| match e { + Errno::ENOENT => TtyError::NotFound(name.to_string()), + other => TtyError::Io { + path: path.display().to_string(), + source: other.into(), + }, + })?; + + let file_type = SFlag::from_bits_truncate(st.st_mode) & SFlag::S_IFMT; + if file_type != SFlag::S_IFCHR { + return Err(TtyError::NotCharacterDevice(name.to_string())); + } + + let mode = Mode::from_bits_truncate(st.st_mode); + let atime = SystemTime::UNIX_EPOCH + Duration::from_secs(st.st_atime.max(0) as u64); + + Ok(TtyInfo { + writable: mode.contains(Mode::S_IWGRP), + atime, + }) +} + +/// A single logged-in session, as reported by utmpx (systemd-logind) +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Session { + pub user: String, + pub tty: String, +} + +fn utmpx_sessions() -> impl Iterator { + Utmpx::iter_all_records().filter_map(|record| { + if !record.is_user_process() { + return None; + } + let tty = record.tty_device(); + // Skip empty ttys and X11/Wayland sessions (":0", ":1", etc.). + if tty.is_empty() || tty.starts_with(':') { + return None; + } + Some(Session { + user: record.user(), + tty, + }) + }) +} + +/// All distinct (user, tty) sessions, optionally restricted to members of `group`, +/// deduplicated by tty. +pub fn all_sessions(group: Option<&str>) -> Vec { + let mut seen = HashSet::new(); + utmpx_sessions() + .filter(|s| group.is_none_or(|g| is_group_member(&s.user, g))) + .filter(|s| seen.insert(s.tty.clone())) + .collect() +} + +/// Distinct ttys that `user` is currently logged in on. +pub fn sessions_for_user(user: &str) -> Vec { + let mut seen = HashSet::new(); + utmpx_sessions() + .filter(|s| s.user == user) + .filter(|s| seen.insert(s.tty.clone())) + .collect() +} + +/// Whether `username` is a member of `group`, by primary or supplementary group. +pub fn is_group_member(username: &str, group: &str) -> bool { + let Some(target_group) = users::get_group_by_name(group) else { + return false; + }; + let Some(user) = users::get_user_by_name(username) else { + return false; + }; + if user.primary_group_id() == target_group.gid() { + return true; + } + users::get_user_groups(username, user.primary_group_id()) + .is_some_and(|groups| groups.iter().any(|g| g.gid() == target_group.gid())) +} + +/// - Escape non-printable characters as `^X` +/// - Rewrite LF TO CRLF +/// - Wrap at `wrap_width` columns if given. +pub fn escape_and_wrap_content(input: &str, wrap_width: Option) -> String { + let mut out = String::with_capacity(input.len()); + let mut col = 0usize; + + for ch in input.chars() { + if ch == '\n' { + out.push_str("\r\n"); + col = 0; + continue; + } + + let rendered = if ch.is_control() { + format!("^{}", (ch as u8 ^ 0x40) as char) + } else { + ch.to_string() + }; + + if let Some(width) = wrap_width + && col > 0 + && col + rendered.len() > width + { + out.push_str("\r\n"); + col = 0; + } + + out.push_str(&rendered); + col += rendered.len(); + } + + out +} + +fn pad_or_truncate(s: &str, width: usize) -> String { + let truncated: String = s.chars().take(width).collect(); + format!("{truncated:, + message: &str, + nobanner: bool, +) -> Vec { + let mut out = String::new(); + + if !nobanner { + let location = from_tty.unwrap_or("somewhere"); + let now = chrono::Local::now().format("%a %b %e %H:%M:%S %Y"); + out.push('\r'); + out.push_str(&" ".repeat(TERM_WIDTH)); + out.push_str("\r\n"); + + let banner = + format!("Broadcast message from {from_user}@{from_host} ({location}) ({now}):"); + out.push_str(&pad_or_truncate(&banner, TERM_WIDTH)); + out.push_str("\x07\x07\r\n"); + } + + out.push_str(&" ".repeat(TERM_WIDTH)); + out.push_str("\r\n"); + + out.push_str(&escape_and_wrap_content(message, Some(TERM_WIDTH))); + if !out.ends_with("\r\n") { + out.push_str("\r\n"); + } + + out.push_str(&" ".repeat(TERM_WIDTH)); + out.push_str("\r\n"); + + out.into_bytes() +} + +pub fn format_write_message( + from_user: &str, + from_host: &str, + from_tty: Option<&str>, + message: &str, +) -> Vec { + let mut out = String::new(); + out.push_str("\r\n\x07\x07\x07"); + + let now = chrono::Local::now(); + let tty = from_tty.unwrap_or(""); + out.push_str(&format!( + "Message from {from_user}@{from_host} on {tty} at {:02}:{:02} ...\r\n", + now.hour(), + now.minute(), + )); + + out.push_str(&escape_and_wrap_content(message, None)); + if !out.ends_with("\r\n") { + out.push_str("\r\n"); + } + out.push_str("EOF\r\n"); + + out.into_bytes() +} + +/// Open `tty` and write `message` to it, giving up after `timeout`. +pub async fn deliver_message( + tty_name: &str, + message: &[u8], + timeout: Duration, +) -> Result<(), TtyError> { + let path = tty_device_path(tty_name)?; + + let fd = match fcntl::open( + &path, + OFlag::O_WRONLY | OFlag::O_NONBLOCK | OFlag::O_NOCTTY, + Mode::empty(), + ) { + Ok(fd) => fd, + Err(Errno::ENOENT | Errno::EACCES | Errno::EBUSY) => { + return Err(TtyError::Unavailable(tty_name.to_string())); + } + Err(e) => { + return Err(TtyError::Io { + path: path.display().to_string(), + source: e.into(), + }); + } + }; + + let async_fd = AsyncFd::new(fd).map_err(|source| TtyError::Io { + path: path.display().to_string(), + source, + })?; + + let write_all = async { + let mut written = 0usize; + while written < message.len() { + let mut guard = async_fd.writable().await?; + match guard.try_io(|inner| { + unistd::write(inner.get_ref(), &message[written..]).map_err(Error::from) + }) { + Ok(Ok(n)) => written += n, + Ok(Err(e)) => return Err(e), + Err(_would_block) => continue, + } + } + Ok::<(), Error>(()) + }; + + time::timeout(timeout, write_all) + .await + .map_err(|_| TtyError::Timeout(tty_name.to_string()))? + .map_err(|source| TtyError::Io { + path: path.display().to_string(), + source, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use nix::{ + pty::openpty, + sys::termios::{self, SetArg}, + }; + + #[test] + fn validate_tty_name_rejects_traversal() { + assert!(validate_tty_name("pts/3").is_ok()); + assert!(validate_tty_name("").is_err()); + assert!(validate_tty_name(":0").is_err()); + assert!(validate_tty_name("/etc/passwd").is_err()); + assert!(validate_tty_name("../etc/passwd").is_err()); + assert!(validate_tty_name("foo/../../bar").is_err()); + } + + #[test] + fn careful_escape_handles_control_chars_and_wrapping() { + assert_eq!(escape_and_wrap_content("hi\nthere", None), "hi\r\nthere"); + assert_eq!(escape_and_wrap_content("a\x01b", None), "a^Ab"); + + let wrapped = escape_and_wrap_content("aaaa bbbb", Some(4)); + assert!(wrapped.contains("\r\n")); + } + + #[test] + fn wall_message_contains_banner_and_body() { + let msg = format_wall_message("alice", "host", Some("pts/0"), "hello there", false); + let msg = String::from_utf8(msg).unwrap(); + assert!(msg.contains("Broadcast message from alice@host (pts/0)")); + assert!(msg.contains("hello there")); + } + + #[test] + fn wall_message_without_banner_omits_it() { + let msg = format_wall_message("alice", "host", Some("pts/0"), "hello there", true); + let msg = String::from_utf8(msg).unwrap(); + assert!(!msg.contains("Broadcast message")); + assert!(msg.contains("hello there")); + } + + #[test] + fn write_message_has_eof_marker() { + let msg = format_write_message("bob", "host", Some("pts/1"), "yo"); + let msg = String::from_utf8(msg).unwrap(); + assert!(msg.contains("Message from bob@host on pts/1")); + assert!(msg.ends_with("EOF\r\n")); + } + + fn open_fake_tty() -> (nix::pty::OpenptyResult, String) { + let pty = openpty(None, None).expect("openpty"); + let name = nix::unistd::ttyname(&pty.slave) + .expect("ttyname") + .to_str() + .unwrap() + .trim_start_matches("/dev/") + .to_string(); + + // Put the slave side in raw mode so the line discipline doesn't rewrite the + // `\r\n` we send before we get to assert on it. + let mut termios = termios::tcgetattr(&pty.slave).expect("tcgetattr"); + termios::cfmakeraw(&mut termios); + termios::tcsetattr(&pty.slave, SetArg::TCSANOW, &termios).expect("tcsetattr"); + + (pty, name) + } + + #[test] + fn stat_tty_reports_character_device() { + let (pty, name) = open_fake_tty(); + let info = stat_tty(&name).expect("stat_tty"); + let _ = info.writable; + drop(pty); + } + + #[tokio::test] + async fn deliver_message_writes_to_pty() { + let (pty, name) = open_fake_tty(); + + deliver_message(&name, b"hello pty\r\n", Duration::from_secs(2)) + .await + .expect("deliver_message"); + + let mut buf = [0u8; 64]; + let n = nix::unistd::read(&pty.master, &mut buf).expect("read"); + assert_eq!(&buf[..n], b"hello pty\r\n"); + } + + #[tokio::test] + async fn deliver_message_to_missing_tty_is_unavailable() { + let err = deliver_message("this-tty-does-not-exist", b"x", Duration::from_millis(200)) + .await + .unwrap_err(); + assert!(matches!(err, TtyError::Unavailable(_))); + } +}