Lines
32.48 %
Functions
44.19 %
Branches
100 %
//! russh `Handler` + `Server` glue.
//!
//! Accepts `publickey` and `password` (rate-limited) auth, accepts
//! `pty-req` + `shell`, and refuses everything else the threat model
//! excludes (`exec`, `subsystem`, `direct-tcpip`, `x11-req`,
//! `auth-agent-req`).
//! The `shell_request` launch site runs the live TUI: it builds an
//! `SshTransport`, drives `tui::run_loop`, and attaches a per-session
//! DB-backed `ConsoleEval` so the Console tab evaluates nomiscript
//! against the authenticated user's `rpc::Session`.
use crate::auth::{Outcome, authenticate_password, authenticate_publickey};
use crate::eval_channel::{EvalChannelState, RusshSink};
use crate::key_decoder::KeyDecoder;
use crate::rate_limit::{InstantClock, RateLimiter};
use crate::tui_transport::SshTransport;
use cli_core::ssh_keys::parse_authorized_keys_line;
use rpc::ScriptCtx;
use russh::keys::{HashAlg, PublicKey};
use russh::server::{Auth, ChannelOpenHandle, Handler, Server, Session};
use russh::{Channel, ChannelId, ChannelOpenFailure, Pty};
use server::command::CmdResult;
use server::command::ssh_key::AddSshKey;
use server::command::user::Refusal;
use sqlx::types::Uuid;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::sync::mpsc::{UnboundedSender, unbounded_channel};
use tui::transport::{RawEvent, Transport};
use tui::widgets::EditMode;
/// Banner template used after a successful ssh-copy-id upload. The
/// daemon substitutes `{hostname}`, `{port}`, `{user}`, `{fingerprint}`
/// at runtime.
#[derive(Clone)]
pub struct SshConnectInfo {
pub hostname: String,
pub port: u16,
pub host_fingerprint: String,
}
/// Authentication method used by the active session. Phase C only
/// permits the `exec` channel that ssh-copy-id opens when the
/// caller authenticated via password.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AuthMethod {
Password,
Publickey,
/// Shared state across every accepted session.
pub struct SshdState {
pub limiter: Arc<Mutex<RateLimiter<InstantClock>>>,
pub connect_info: SshConnectInfo,
impl Server for SshdState {
type Handler = Session_;
fn new_client(&mut self, peer_addr: Option<SocketAddr>) -> Self::Handler {
Session_ {
peer: peer_addr,
limiter: Arc::clone(&self.limiter),
connect_info: self.connect_info.clone(),
user_id: None,
user_name: None,
auth_method: None,
pubkey_uploads: HashMap::new(),
tui_channels: HashMap::new(),
eval_channels: HashMap::new(),
fn handle_session_error(&mut self, error: <Self::Handler as Handler>::Error) {
// The Kubernetes TCP probes connect and hang up every few seconds, and a
// client that walks away looks the same; logging those at warn buried
// real session failures under dozens of lines an hour.
if is_peer_hangup(&error) {
log::debug!("session ended by peer: {error:?}");
} else {
log::warn!("session error: {error:?}");
/// Per-channel state for an active TUI session — the keystroke
/// decoder, the input mpsc sender, and the PTY geometry the client
/// announced. Created in `pty_request`, completed in `shell_request`,
/// and dropped in `channel_close`.
struct TuiChannelState {
decoder: KeyDecoder,
input_tx: Option<UnboundedSender<RawEvent>>,
term: Option<String>,
cols: u16,
rows: u16,
/// Per-connection handler. Odd name because `Session` is taken by
/// russh.
pub struct Session_ {
peer: Option<SocketAddr>,
limiter: Arc<Mutex<RateLimiter<InstantClock>>>,
connect_info: SshConnectInfo,
user_id: Option<Uuid>,
user_name: Option<String>,
auth_method: Option<AuthMethod>,
/// Buffers for `exec` channels that ssh-copy-id is feeding pubkey
/// lines through. Drained on `channel_eof`.
pubkey_uploads: HashMap<ChannelId, Vec<u8>>,
/// Per-channel TUI bridge state. Disjoint from `pubkey_uploads`:
/// a channel is either an ssh-copy-id upload or a TUI session,
/// never both.
tui_channels: HashMap<ChannelId, TuiChannelState>,
/// Per-channel state for the `nomisync-eval` subsystem. Disjoint
/// from `pubkey_uploads` and `tui_channels`: a channel commits
/// to one channel-type on its first request. Created in
/// `subsystem_request`, drained in `data`, dropped in
/// `channel_eof` / `channel_close`.
eval_channels: HashMap<ChannelId, EvalChannelState>,
impl Session_ {
fn peer_ip(&self) -> std::net::IpAddr {
self.peer
.map_or(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), |s| {
s.ip()
})
/// Refuses a channel whose tenant database is withheld by the fleet
/// migrator, telling the user why rather than failing opaquely inside the
/// session.
///
/// Fails **closed**: a directory lookup that errors refuses too. Treating an
/// unreadable directory as "not quarantined" would hand a session to
/// exactly the tenant whose state could not be confirmed.
/// Every channel that can reach the tenant's database goes through here.
/// Guarding only the shell left `nomisync-eval` open, and that subsystem
/// evaluates scripts against the same database.
async fn refuse_if_withheld(
&self,
user_id: Uuid,
channel: ChannelId,
session: &mut Session,
kind: &str,
) -> Result<bool, russh::Error> {
let refusal = match server::command::user::account_refusal(user_id).await {
Ok(refusal) => refusal,
Err(e) => {
log::error!("{kind}_request refusing user_id={user_id}: {e}");
Some(Refusal::Quarantined)
};
let Some(refusal) = refusal else {
return Ok(false);
let notice = match refusal {
Refusal::Quarantined => {
log::warn!("{kind}_request refused user_id={user_id} reason=quarantined");
"\r\nThis account is undergoing maintenance and is temporarily \
unavailable. Please try again shortly.\r\n"
Refusal::Disabled => {
log::warn!("{kind}_request refused user_id={user_id} reason=disabled");
"\r\nThis account is disabled.\r\n"
.to_owned();
session.data(channel, notice.into_bytes())?;
session.close(channel)?;
Ok(true)
fn refuse_request(&self, kind: &str) {
log::warn!(
"refused request kind={kind} peer={:?} user_id={:?}",
self.peer,
self.user_id
);
/// True only for the ssh-copy-id bootstrap window: the session
/// authenticated with a password AND the user does not yet have
/// any registered key.
async fn is_bootstrap_upload_allowed(&self) -> bool {
if self.auth_method != Some(AuthMethod::Password) {
return false;
let Some(uid) = self.user_id else {
matches!(
server::command::ssh_key::UserHasSshKey::new()
.user_id(uid)
.run()
.await,
Ok(Some(CmdResult::Bool(false)))
)
/// Parse the bytes ssh-copy-id wrote to stdin (one or more
/// `authorized_keys` lines) and register each line via
/// [`AddSshKey`]. Returns the operator-facing banner that the
/// channel writes back before closing.
async fn process_pubkey_upload(&self, buf: &[u8]) -> String {
return "ssh-copy-id: no authenticated user; aborting.\n".to_string();
let raw = match std::str::from_utf8(buf) {
Ok(s) => s,
Err(_) => {
return "ssh-copy-id: pubkey payload is not UTF-8.\n".to_string();
let mut accepted = 0usize;
let mut errors: Vec<String> = Vec::new();
for line in raw.lines() {
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
let parsed = match parse_authorized_keys_line(trimmed) {
Ok(p) => p,
errors.push(format!("parse: {e}"));
let annotation = if parsed.comment.is_empty() {
"ssh-copy-id".to_string()
format!("ssh-copy-id ({})", parsed.comment)
match AddSshKey::new()
.key_type(parsed.key_type)
.key_blob(parsed.key_blob)
.fingerprint(parsed.fingerprint)
.annotation(annotation)
.await
{
Ok(Some(CmdResult::Uuid(_))) => accepted += 1,
Ok(_) => errors.push("server returned no key id".to_string()),
Err(e) => errors.push(e.to_string()),
if accepted == 0 {
let detail = if errors.is_empty() {
"no usable pubkey lines".to_string()
errors.join("; ")
return format!("ssh-copy-id: no key accepted ({detail}).\n");
let user_label = self.user_name.as_deref().unwrap_or("YOUR_LOGIN");
format!(
"Identity added; password auth disabled.\r\n\
\r\n\
Suggested ~/.ssh/config entry:\r\n\
Host nomisync\r\n\
\x20 HostName {hostname}\r\n\
\x20 Port {port}\r\n\
\x20 User {user}\r\n\
\x20 IdentitiesOnly yes\r\n\
\x20 # IdentityAgent ${{SSH_AUTH_SOCK}} # uncomment if using gpg-agent\r\n\
\x20 # IdentityFile ~/.ssh/id_ed25519 # or point at your private key\r\n\
Host key fingerprint (verify on first connect):\r\n\
\x20 {fp}\r\n",
hostname = self.connect_info.hostname,
port = self.connect_info.port,
user = user_label,
fp = self.connect_info.host_fingerprint,
impl Handler for Session_ {
type Error = russh::Error;
async fn auth_publickey(
&mut self,
user: &str,
public_key: &PublicKey,
) -> Result<Auth, Self::Error> {
let fingerprint = public_key.fingerprint(HashAlg::Sha256).to_string();
let outcome = authenticate_publickey(user, &fingerprint, self.peer_ip(), &self.limiter)
.map_err(|_| russh::Error::Disconnect)?;
match outcome {
Outcome::Accept(uid) => {
self.user_id = Some(uid);
self.user_name = Some(user.to_string());
self.auth_method = Some(AuthMethod::Publickey);
Ok(Auth::Accept)
Outcome::Reject => Ok(Auth::reject()),
async fn auth_password(&mut self, user: &str, password: &str) -> Result<Auth, Self::Error> {
let outcome = authenticate_password(user, password, self.peer_ip(), &self.limiter)
self.auth_method = Some(AuthMethod::Password);
async fn channel_open_session(
_channel: Channel<russh::server::Msg>,
reply: ChannelOpenHandle,
_session: &mut Session,
) -> Result<(), Self::Error> {
reply.accept().await;
Ok(())
async fn pty_request(
term: &str,
col_width: u32,
row_height: u32,
_pix_width: u32,
_pix_height: u32,
_modes: &[(Pty, u32)],
let cols = u16::try_from(col_width).unwrap_or(80);
let rows = u16::try_from(row_height).unwrap_or(24);
self.tui_channels.insert(
channel,
TuiChannelState {
decoder: KeyDecoder::new(),
input_tx: None,
term: Some(term.to_string()),
cols,
rows,
},
async fn shell_request(
log::warn!("shell_request without authenticated user; rejecting");
return Err(russh::Error::Disconnect);
if self
.refuse_if_withheld(uid, channel, session, "shell")
.await?
return Ok(());
let Some(state) = self.tui_channels.get_mut(&channel) else {
log::warn!("shell_request without prior pty-req on channel {channel:?}; rejecting");
log::info!(
"shell_request accepted user_id={uid} peer={:?} term={:?} cols={} rows={}",
state.term,
state.cols,
state.rows,
let (input_tx, input_rx) = unbounded_channel::<RawEvent>();
state.input_tx = Some(input_tx);
let term = state.term.clone();
let cols = state.cols;
let rows = state.rows;
let handle = session.handle();
let runtime = tokio::runtime::Handle::current();
let runtime_for_spawn = runtime.clone();
let runtime_for_console = runtime.clone();
tokio::task::spawn_blocking(move || {
match SshTransport::new(
handle,
runtime_for_spawn,
input_rx,
term,
) {
Ok(mut transport) => {
let mut app = tui::App::new(uid, EditMode::Emacs);
match rpc::Session::guarded(rpc::ScriptCtx::new(uid))
.map(|s| tui::ConsoleEval::spawn_session(&runtime_for_console, s))
Ok(eval) => app.attach_console(eval),
Err(err) => {
log::warn!("console eval unavailable for user_id={uid}: {err}");
app.set_status(format!("console unavailable: {err}"));
if let Err(err) = tui::run_loop(&mut transport, &mut app) {
log::info!("tui session ended: {err}");
if let Err(err) = transport.finish() {
log::warn!("tui transport finish failed: {err}");
log::error!("tui transport build failed: {err}");
});
async fn window_change_request(
if let Some(state) = self.tui_channels.get_mut(&channel) {
let cols = u16::try_from(col_width).unwrap_or(state.cols);
let rows = u16::try_from(row_height).unwrap_or(state.rows);
state.cols = cols;
state.rows = rows;
if let Some(tx) = &state.input_tx {
let _ = tx.send(RawEvent::Resize(cols, rows));
async fn exec_request(
_data: &[u8],
// The only `exec` we accept is the one ssh-copy-id opens to
// append to authorized_keys. Gate is "session authenticated
// via password AND user has zero keys on file" — which is
// exactly the bootstrap window. The requested command string
// is ignored: stdin is consumed verbatim as authorized_keys
// content, deferred to channel_eof.
if !self.is_bootstrap_upload_allowed().await {
self.refuse_request("exec");
"exec_request accepted as ssh-copy-id upload user_id={:?} peer={:?}",
self.user_id,
self.pubkey_uploads.entry(channel).or_default();
async fn data(
data: &[u8],
if let Some(buf) = self.pubkey_uploads.get_mut(&channel) {
buf.extend_from_slice(data);
let events = state.decoder.feed(data);
for ev in events {
if tx.send(ev).is_err() {
// run_loop has exited — drop subsequent input
// until channel_close cleans up.
break;
if let Some(state) = self.eval_channels.get_mut(&channel)
&& let Err(err) = state.feed(data)
// Non-UTF8 on an s-expr channel — surface as a parse
// error envelope and drop the connection.
let msg = format!("(:id 0 :error (:code parse :message \"{err}\"))\n");
session.data(channel, msg.into_bytes())?;
self.eval_channels.remove(&channel);
// Responses are delivered out-of-band by the channel's
// worker via the russh server Handle (see `RusshSink`).
// data() returns immediately so future bytes — including
// ETX (C-g) — get processed without waiting on the
// current eval.
async fn channel_eof(
if let Some(buf) = self.pubkey_uploads.remove(&channel) {
let banner = self.process_pubkey_upload(&buf).await;
session.data(channel, banner.into_bytes())?;
self.tui_channels.remove(&channel);
async fn channel_close(
// Either path is harmless if the entry is already gone.
self.pubkey_uploads.remove(&channel);
async fn subsystem_request(
name: &str,
if name != "nomisync-eval" {
self.refuse_request("subsystem");
let Some(user_id) = self.user_id else {
log::warn!("nomisync-eval requested before authentication; refusing");
// The eval subsystem runs scripts against the tenant's own database, so
// it is exactly as unsafe as a shell while that database is withheld.
.refuse_if_withheld(user_id, channel, session, "subsystem")
let sink = Box::new(RusshSink::new(session.handle(), channel));
match rpc::Session::guarded(ScriptCtx::new(user_id))
.map(|session| EvalChannelState::with_session(session, sink))
Ok(state) => {
self.eval_channels.insert(channel, state);
log::error!("nomisync-eval session init failed: {err}");
Err(russh::Error::Disconnect)
async fn channel_open_direct_tcpip(
_host_to_connect: &str,
_port_to_connect: u32,
_originator_address: &str,
_originator_port: u32,
self.refuse_request("direct-tcpip");
reply
.reject(ChannelOpenFailure::AdministrativelyProhibited)
.await;
async fn tcpip_forward(
_address: &str,
_port: &mut u32,
) -> Result<bool, Self::Error> {
self.refuse_request("tcpip-forward");
Ok(false)
/// Whether a session error only means the peer went away.
fn is_peer_hangup(error: &russh::Error) -> bool {
match error {
russh::Error::Disconnect | russh::Error::HUP => true,
russh::Error::IO(e) => matches!(
e.kind(),
std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::UnexpectedEof
| std::io::ErrorKind::BrokenPipe
),
_ => false,
#[cfg(test)]
mod tests {
use super::is_peer_hangup;
use std::io::{Error, ErrorKind};
#[test]
fn a_peer_walking_away_is_not_a_session_error() {
for hangup in [
russh::Error::Disconnect,
russh::Error::HUP,
russh::Error::IO(Error::from(ErrorKind::ConnectionReset)),
russh::Error::IO(Error::from(ErrorKind::UnexpectedEof)),
russh::Error::IO(Error::from(ErrorKind::BrokenPipe)),
] {
assert!(is_peer_hangup(&hangup), "{hangup:?}");
fn a_real_failure_is_still_reported() {
for failure in [
russh::Error::InactivityTimeout,
russh::Error::IO(Error::from(ErrorKind::PermissionDenied)),
assert!(!is_peer_hangup(&failure), "{failure:?}");