Initial commit

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "syndicate-pty-driver"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+lazy_static = "1.4.0"
+libc = "0.2.150"
+preserves-schema = "4.992.0"
+pty = "0.2.2"
+
+# syndicate = { path = "../syndicate-rs/syndicate", version = "0.30"}
+syndicate = "0.30"
+
+syndicate-macros = "0.25"
+
+tokio = { version = "1.34.0", features = ["io-std", "fs"] }
+tracing = "0.1.40"
+
+[build-dependencies]
+preserves-schema = "4.992.0"
+syndicate-schema-plugin = "0.2.0"

build.rs

@@ -0,0 +1,15 @@
+use preserves_schema::compiler::*;
+
+fn main() -> std::io::Result<()> {
+    let buildroot = std::path::PathBuf::from(std::env::var_os("OUT_DIR").unwrap());
+
+    let mut gen_dir = buildroot.clone();
+    gen_dir.push("src/schemas");
+
+    let mut c = CompilerConfig::new(gen_dir, "crate::schemas".to_owned());
+    c.plugins.push(Box::new(syndicate_schema_plugin::PatternPlugin));
+
+    let inputs = expand_inputs(&vec!["protocols/schema-bundle.bin".to_owned()])?;
+    c.load_schemas_and_bundles(&inputs, &vec![])?;
+    compile(&c)
+}

protocols/Makefile

@@ -0,0 +1,8 @@
+all: schema-bundle.bin
+
+clean:
+	rm -f schema-bundle.bin
+
+schema-bundle.bin: schemas/*.prs
+	preserves-schemac schemas > $@.tmp
+	mv $@.tmp $@

protocols/schema-bundle.bin

@@ -0,0 +1,4 @@
+´³bundle·µ³pty„´³schema·³version°

³definitions·³Chunk´³atom³
+ByteString„³PtyInput´³rec´³lit³	pty-input„´³tupleµ´³named³id³any„´³named³data´³refµ„³Chunk„„„„„³	PtyOutput´³rec´³lit³
+pty-output„´³tupleµ´³named³id³any„´³named³data´³refµ„³Chunk„„„„„³
+PtySession´³rec´³lit³pty-session„´³tupleµ´³named³id³any„´³named³commandLine´³refµ„³CommandLine„„„„„³CommandLine´³tuplePrefixµ´³named³command³any„„´³named³args´³seqof³any„„„³PtySessionRunning´³rec´³lit³pty-session-running„´³tupleµ´³named³id³any„„„„„³embeddedType€„„„„

protocols/schemas/pty.prs

@@ -0,0 +1,14 @@
+version 1 .
+
+PtySession = <pty-session @id any @commandLine CommandLine> .
+CommandLine = [@command any @args any ...] .
+
+PtySessionRunning = <pty-session-running @id any> .
+
+# To the subprocess
+PtyInput = <pty-input @id any @data Chunk> .
+
+# From the subprocess
+PtyOutput = <pty-output @id any @data Chunk> .
+
+Chunk = bytes .

rust-toolchain

@@ -0,0 +1 @@
+nightly

server-config.pr

@@ -0,0 +1,9 @@
+<require-service <daemon pty>>
+<daemon pty {
+  argv: "cargo run"
+  protocol: application/syndicate
+  env: { RUST_LOG: "syndicate::dataspace=trace,syndicate_pty_driver=debug,info" }
+}>
+? <service-object <daemon pty> ?cap> [
+  $cap += <pty-session 1 [bash -i]>
+]

src/main.rs

@@ -0,0 +1,136 @@
+use std::io::Write;
+use std::os::unix::io::AsRawFd;
+use std::os::unix::io::FromRawFd;
+use std::os::unix::prelude::CommandExt;
+use std::sync::Arc;
+
+use syndicate::actor::Account;
+use syndicate::actor::Activation;
+use syndicate::actor::Actor;
+use syndicate::actor::ActorError;
+use syndicate::actor::ActorResult;
+use syndicate::actor::AnyValue;
+use syndicate::actor::Cap;
+use syndicate::actor::LinkedTaskTermination;
+use syndicate::dataspace::Dataspace;
+use syndicate::enclose;
+use syndicate::relay;
+use syndicate::value::NestedValue;
+
+use syndicate::value::NoEmbeddedDomainCodec;
+use syndicate::value::TextWriter;
+use syndicate_macros::during;
+use syndicate_macros::on_message;
+
+use tokio::io::AsyncReadExt;
+
+mod schemas {
+    include!(concat!(env!("OUT_DIR"), "/src/schemas/mod.rs"));
+}
+
+preserves_schema::define_language!(language(): Language<AnyValue> {
+    syndicate: syndicate::schemas::Language,
+    main: crate::schemas::Language,
+});
+
+use schemas::pty::{PtySession, PtySessionRunning, Chunk, PtyOutput};
+
+fn stringify(v: &AnyValue) -> Result<String, ActorError> {
+    Ok(TextWriter::encode(&mut NoEmbeddedDomainCodec, v)?)
+}
+
+fn pty_session(t: &mut Activation, local_ds: Arc<Cap>, s: PtySession<AnyValue>) -> ActorResult {
+    let mut pieces: Vec<String> = Vec::new();
+    for piece_value in s.command_line.args.into_iter() {
+        pieces.push(stringify(&piece_value)?);
+    }
+    let mut cmd = std::process::Command::new(stringify(&s.command_line.command)?);
+    cmd.args(pieces);
+    tracing::info!(?cmd);
+
+    let fork = pty::fork::Fork::from_ptmx()?;
+    match &fork {
+        pty::prelude::Fork::Parent(child_pid, parent) => {
+            tracing::info!(?child_pid);
+            let mut parent_w: std::fs::File = unsafe { FromRawFd::from_raw_fd(libc::dup(parent.as_raw_fd())) };
+            let mut parent_r: tokio::fs::File = unsafe { FromRawFd::from_raw_fd(libc::dup(parent.as_raw_fd())) };
+            let child_pid = *child_pid;
+            let facet = t.facet.clone();
+
+            local_ds.assert(t, language(), &PtySessionRunning {
+                id: s.id.clone(),
+            });
+
+            on_message!(t, local_ds, language(), <pty-input #(&s.id) $body: Chunk>, |_t| {
+                tracing::trace!(?body, "sending to child");
+                parent_w.write_all(&body.0[..])?;
+                Ok(())
+            });
+
+            t.linked_task(Some(AnyValue::symbol("waiter")), async move {
+                let mut status: i32 = 0;
+                loop {
+                    match unsafe { libc::waitpid(child_pid, &mut status, libc::WNOHANG) } {
+                        0 => tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await,
+                        -1 => Err(std::io::Error::last_os_error())?,
+                        _ => break,
+                    }
+                }
+                tracing::info!(?status, "child exited");
+                Ok(LinkedTaskTermination::Normal)
+            });
+
+            t.linked_task(Some(AnyValue::symbol("reader")), async move {
+                let read_account = Account::new(Some(AnyValue::symbol("reader_account")), None);
+                let mut buf = [0; 1024];
+                loop {
+                    let n = parent_r.read(&mut buf[..]).await?;
+                    tracing::trace!(?n, buf=?&buf[0..n], "receiving from child");
+                    if n == 0 || !facet.activate(&read_account, None, |t| {
+                        local_ds.message(t, language(), &PtyOutput {
+                            id: s.id.clone(),
+                            data: Chunk(buf[0..n].to_vec()),
+                        });
+                        Ok(())
+                    }) {
+                        unsafe {
+                            libc::kill(child_pid, libc::SIGTERM);
+                        }
+                        return Ok(LinkedTaskTermination::Normal);
+                    }
+                }
+            });
+            Ok(())
+        }
+        pty::prelude::Fork::Child(_child) => {
+            Err(cmd.exec())?
+        }
+    }
+}
+
+#[tokio::main]
+async fn main() -> ActorResult {
+    syndicate::convenient_logging()?;
+
+    Actor::top(None, move |t| {
+        let local_ds = Cap::new(&t.create(Dataspace::new(Some(AnyValue::symbol("pty")))));
+
+        relay::TunnelRelay::run(t,
+                                relay::Input::Bytes(Box::pin(tokio::io::stdin())),
+                                relay::Output::Bytes(Box::pin(tokio::io::stdout())),
+                                Some(local_ds.clone()),
+                                None,
+                                false);
+
+        during!(t, local_ds, language(), $s: PtySession::<AnyValue>, |t: &mut Activation| {
+            t.spawn_link(Some(s.id.clone()), enclose!((local_ds) |t| {
+                pty_session(t, local_ds, s)
+            }));
+            Ok(())
+        });
+
+        Ok(())
+    }).await??;
+
+    Ok(())
+}