twitchyliquid64 · Jun 1, 2026 · May 29, 2026 · May 31, 2026
@@ -80,6 +80,7 @@ prost-build = "0.14"
 prost-types = "0.14"
 resolv-conf = "0.7"
 russh = "0.61"
+russh-sftp = "2.1"
 serde_json = { version = "1.0" }
 serial_test = "3"
 sha2 = "0.11"

@@ -5,12 +5,15 @@ edition.workspace = true
 publish.workspace = true
 
 [dependencies]
+bytes.workspace = true
 camino.workspace = true
 clap.workspace = true
 clap_complete.workspace = true
 constcat.workspace = true
 dirs.workspace = true
 russh.workspace = true
+russh-sftp.workspace = true
+path-absolutize.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 uuid.workspace = true

@@ -16,6 +16,7 @@ use crate::{
     ChannelConfig, RequestedPty,
     rpc::{self},
     server::ServerStateHandle,
+    sftp,
 };
 
 static PROTOCOL_TRACE_ENABLED: LazyLock<bool> =
@@ -366,6 +367,10 @@ impl russh::server::Handler for ConnectionHandler {
             let c = self.0.clone();
             let s = c.0.lock().await.serv.clone();
             rpc::handle_ssh_rpc(s, c, name, id, session).await?;
+        } else if name == sftp::SUBSYSTEM_NAME {
+            let c = self.0.clone();
+            let s = c.0.lock().await.serv.clone();
+            sftp::handle_sftp_subsystem(s, c, id, session).await?;
         } else {
             session.channel_failure(id)?;
         }

@@ -5,6 +5,7 @@ pub mod rpc;
 pub mod server;
 mod session;
 mod sessions;
+mod sftp;
 #[cfg(test)]
 mod test_harness;
 

@@ -99,7 +99,7 @@ impl GetVersion {
 pub struct ListSessions;
 
 /// An entry in the ListSessions response.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct ListSessionsEntry {
     pub id: Uuid,
     pub name: Option<String>,
@@ -193,6 +193,46 @@ impl GetSessionRecord {
     }
 }
 
+/// An RPC to create a new session based on the given record.
+pub struct CreateSession;
+
+/// The request for a [`CreateSession`] RPC.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CreateSessionRequest {
+    pub record: sessions::Record,
+}
+
+/// The response for a [`CreateSession`] RPC.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct CreateSessionResponse {
+    pub id: Uuid,
+}
+
+impl OneshotSshRpc for CreateSession {
+    const NAME: &'static str = constcat::concat!(RPC_SUBSYSTEM_PREFIX, "CreateSession");
+    type Request<'a> = CreateSessionRequest;
+    type Response = CreateSessionResponse;
+}
+
+impl CreateSession {
+    pub async fn handle(self, s: ServerStateHandle, c: RuChannel<Msg>) {
+        let res = self
+            .handle_channel(c, async |req| {
+                let mngr = s.sessions_manager().await;
+                Ok(CreateSessionResponse {
+                    id: mngr
+                        .create_session(req.record)
+                        .await
+                        .map_err(|e| ConnectionError::Internal(e.to_string()))?,
+                })
+            })
+            .await;
+        if let Err(e) = res {
+            tracing::warn!("RPC handler for {} failed: {}", Self::NAME, e);
+        }
+    }
+}
+
 /// Handles an RPC going over an SSH subsystem channel.
 ///
 /// This method takes ownership of the ssh channel, including
@@ -211,7 +251,7 @@ pub async fn handle_ssh_rpc(
 ) -> Result<(), ConnectionError> {
     // Take the channel from connection state if its a known RPC.
     let channel = match name {
-        GetVersion::NAME | ListSessions::NAME | GetSessionRecord::NAME => {
+        GetVersion::NAME | ListSessions::NAME | GetSessionRecord::NAME | CreateSession::NAME => {
             let mut conn_lock = c.lock().await;
             let c_hnd = match conn_lock.take(id) {
                 None => {
@@ -239,6 +279,7 @@ pub async fn handle_ssh_rpc(
         GetVersion::NAME => spawn(GetVersion.handle(channel)),
         ListSessions::NAME => spawn(ListSessions.handle(s, channel)),
         GetSessionRecord::NAME => spawn(GetSessionRecord.handle(s, channel)),
+        CreateSession::NAME => spawn(CreateSession.handle(s, channel)),
         _ => unreachable!(),
     };
 
@@ -247,6 +288,8 @@ pub async fn handle_ssh_rpc(
 
 #[cfg(test)]
 mod tests {
+    use sessions::paths::HostAbsPath;
+
     use super::*;
     use crate::test_harness::TestServer;
 
@@ -312,4 +355,46 @@ mod tests {
             assert_eq!(resp.version, env!("CARGO_PKG_VERSION"));
         }
     }
+
+    #[tokio::test]
+    async fn create_session_shows_in_get_and_list() {
+        let server = TestServer::new().await;
+        let mut client = server.connect().await;
+
+        let create_session = client
+            .call::<CreateSession>(&CreateSessionRequest {
+                record: sessions::Record {
+                    id: Uuid::nil(),
+                    name: Some("my session".to_string()),
+                    username: None,
+                    project_path: HostAbsPath::try_new("/uwu").unwrap(),
+                    attrs: Default::default(),
+                },
+            })
+            .await;
+
+        assert!(create_session.id != Uuid::nil());
+
+        let get_session = client
+            .call::<GetSessionRecord>(&GetSessionRecordRequest::Id(create_session.id))
+            .await;
+        assert_eq!(get_session.record.as_ref().unwrap().id, create_session.id);
+        assert_eq!(
+            get_session.record.as_ref().unwrap().name,
+            Some("my session".to_string())
+        );
+        assert_eq!(
+            get_session.record.as_ref().unwrap().project_path,
+            HostAbsPath::try_new("/uwu").unwrap()
+            );
+
+        let list_sessions = client.call::<ListSessions>(&()).await;
+        assert_eq!(
+            list_sessions.sessions,
+            vec![ListSessionsEntry {
+                id: create_session.id,
+                name: Some("my session".to_string()),
+            }]
+        );
+    }
 }
@@ -3,7 +3,7 @@ use std::collections::BTreeMap;
 use crate::session::{Session, SessionHandle};
 use sessions::{
     paths::DaemonAbsPath,
-    store::{DiskLoader, Loader, SessionObject},
+    store::{DiskLoader, Loader, SessionKey, SessionObject},
 };
 use tokio::sync::{mpsc, oneshot};
 use uuid::Uuid;
@@ -48,6 +48,7 @@ enum ManagerMessage {
     List(Responder<Vec<SessionInfo>>),
     GetRecord(SessionKeyPredicate, Responder<Option<sessions::Record>>),
     GetSession(SessionKeyPredicate, Responder<Option<SessionHandle>>),
+    CreateSession(sessions::Record, Responder<Uuid>),
 }
 
 /// Manages session instances, and session state on disk.
@@ -155,6 +156,14 @@ impl<L: Loader> Manager<L> {
                 })
                 .await;
             }
+            // Creates a session using the given record.
+            ManagerMessage::CreateSession(record, r) => {
+                r.handle(async {
+                    let k = self.store.create(record)?;
+                    Ok(*k.uuid())
+                })
+                .await;
+            }
         }
     }
 }
@@ -183,6 +192,17 @@ impl ManagerHandle {
         recv.await.expect("corresponding sessions manager is dead")
     }
 
+    /// Creates a session based on the given record.
+    pub async fn create_session(&self, record: sessions::Record) -> Result<Uuid, ResponseError> {
+        let (send, recv) = Responder::channel();
+        // Ignore send errors - the recv will also fail.
+        let _ = self
+            .0
+            .send(ManagerMessage::CreateSession(record, send))
+            .await;
+        recv.await.expect("corresponding sessions manager is dead")
+    }
+
     /// Gets a handle to the session corresponding with the given predicate.
     pub async fn get_session(
         &self,