feat: Run Datum Connect as a headless tunnel endpoint

.dockerignore

@@ -0,0 +1,13 @@
+.git
+.github
+.claude
+.vscode
+target
+.env
+n0des-local
+ui/node_modules
+ui/target
+ui/dist
+ui/.dioxus
+**/*.log
+**/*.tmp

.github/workflows/cli-image.yml

@@ -0,0 +1,58 @@
+name: cli-image
+
+on:
+  push:
+    tags:
+      - "v*"
+  pull_request:
+    branches: [main]
+    paths:
+      - "**/Cargo.toml"
+      - "Cargo.lock"
+      - "cli/**"
+      - "lib/**"
+      - "Dockerfile"
+      - ".dockerignore"
+      - ".github/workflows/cli-image.yml"
+  workflow_call:
+    inputs:
+      version:
+        description: "Release tag to publish the image under"
+        required: true
+        type: string
+
+jobs:
+  image:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Log in to GHCR
+        if: github.event_name != 'pull_request'
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Build and push datum-connect image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: Dockerfile
+          platforms: linux/amd64,linux/arm64
+          push: ${{ github.event_name != 'pull_request' }}
+          build-args: |
+            VERSION=${{ github.event_name != 'pull_request' && (inputs.version || github.ref_name) || '' }}
+          tags: ${{ github.event_name != 'pull_request' && format('ghcr.io/datum-cloud/datum-connect:{0}', inputs.version || github.ref_name) || '' }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max

.github/workflows/manual-release.yml

@@ -421,3 +421,14 @@ jobs:
             artifacts/**/ui/dist/*.dmg
             artifacts/**/ui/dist/*.AppImage
             artifacts/**/ui/dist/*.exe
+
+  # The release tag is created with GITHUB_TOKEN, which does not trigger the
+  # tag push that publishes the image in cli-image.yml.
+  image:
+    needs: [release]
+    permissions:
+      contents: read
+      packages: write
+    uses: ./.github/workflows/cli-image.yml
+    with:
+      version: ${{ inputs.version }}

Dockerfile

@@ -0,0 +1,60 @@
+FROM --platform=$BUILDPLATFORM rust:1.98-bookworm AS builder
+
+ARG BUILDARCH
+ARG TARGETARCH
+
+RUN case "$TARGETARCH" in \
+      amd64) echo x86_64-unknown-linux-gnu > /rust-target ;; \
+      arm64) echo aarch64-unknown-linux-gnu > /rust-target ;; \
+      *) echo "unsupported architecture: $TARGETARCH" >&2; exit 1 ;; \
+    esac \
+  && rustup target add "$(cat /rust-target)" \
+  && if [ "$TARGETARCH" != "$BUILDARCH" ]; then \
+      dpkg --add-architecture "$TARGETARCH" \
+      && apt-get update \
+      && apt-get install -y --no-install-recommends \
+        "crossbuild-essential-$TARGETARCH" "libssl-dev:$TARGETARCH" \
+      && rm -rf /var/lib/apt/lists/*; \
+    fi
+
+ENV CARGO_TARGET_AARCH64_UNKNOWN_LINUX_GNU_LINKER=aarch64-linux-gnu-gcc \
+    CC_aarch64_unknown_linux_gnu=aarch64-linux-gnu-gcc \
+    AARCH64_UNKNOWN_LINUX_GNU_OPENSSL_LIB_DIR=/usr/lib/aarch64-linux-gnu \
+    AARCH64_UNKNOWN_LINUX_GNU_OPENSSL_INCLUDE_DIR=/usr/include \
+    CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_LINKER=x86_64-linux-gnu-gcc \
+    CC_x86_64_unknown_linux_gnu=x86_64-linux-gnu-gcc \
+    X86_64_UNKNOWN_LINUX_GNU_OPENSSL_LIB_DIR=/usr/lib/x86_64-linux-gnu \
+    X86_64_UNKNOWN_LINUX_GNU_OPENSSL_INCLUDE_DIR=/usr/include
+
+WORKDIR /src
+
+COPY . .
+
+ARG VERSION
+
+RUN --mount=type=cache,target=/usr/local/cargo/registry \
+    --mount=type=cache,target=/src/target,id=datum-connect-target-$TARGETARCH \
+    if [ -n "$VERSION" ]; then \
+      sed -i "s/^version = \".*\"/version = \"${VERSION#v}\"/" cli/Cargo.toml lib/Cargo.toml \
+      && cargo update --workspace; \
+    fi \
+    && target="$(cat /rust-target)" \
+    && cargo build --locked --release -p datum-connect --target "$target" \
+    && cp "target/$target/release/datum-connect" /usr/local/bin/datum-connect
+
+FROM debian:bookworm-slim
+
+RUN apt-get update \
+  && apt-get install -y --no-install-recommends ca-certificates libssl3 \
+  && rm -rf /var/lib/apt/lists/* \
+  && useradd -u 65532 -r -s /usr/sbin/nologin datum \
+  && install -d -o 65532 -g 65532 /var/lib/datum-connect
+
+COPY --from=builder /usr/local/bin/datum-connect /usr/local/bin/datum-connect
+
+ENV DATUM_CONNECT_REPO=/var/lib/datum-connect
+
+USER 65532:65532
+
+ENTRYPOINT ["/usr/local/bin/datum-connect"]
+CMD ["serve"]

cli/src/main.rs

@@ -9,7 +9,7 @@ use lib::{
 };
 use std::{net::SocketAddr, path::PathBuf};
 use tracing::info;
-use tracing_subscriber::prelude::*;
+use tracing_subscriber::{EnvFilter, prelude::*};
 
 /// Datum Connect Agent
 #[derive(Parser, Debug)]
@@ -23,7 +23,7 @@ struct Args {
 #[derive(Subcommand, Debug)]
 enum Commands {
     /// Start a tunnel server that exposes configured local services through the Datum gateway.
-    Serve,
+    Serve(ServeArgs),
 
     /// Join a proxy, i.e. connect to the proxy and expose the service locally.
     Connect(ConnectArgs),
@@ -117,6 +117,24 @@ pub struct TunnelDevArgs {
     pub target_protocol: String,
 }
 
+#[derive(Parser, Debug)]
+pub struct ServeArgs {
+    /// Read the endpoint identity from this file (a raw 32-byte ed25519 secret key)
+    /// instead of the repo's listen key. The file is only read, never created.
+    #[clap(long, env = "DATUM_CONNECT_LISTEN_KEY_FILE")]
+    pub listen_key_file: Option<PathBuf>,
+
+    /// Expose a TCP service at HOST:PORT before serving, adding it if not configured yet.
+    ///
+    /// Repeatable; the environment variable takes a comma-separated list.
+    #[clap(
+        long = "tcp-proxy",
+        env = "DATUM_CONNECT_TCP_PROXIES",
+        value_delimiter = ','
+    )]
+    pub tcp_proxies: Vec<String>,
+}
+
 #[derive(Parser, Debug)]
 pub struct ConnectArgs {
     /// The addresses to listen on for incoming tcp connections.
@@ -147,7 +165,7 @@ async fn main() -> n0_error::Result<()> {
             .filter(|s| !s.is_empty())
             .and_then(|s| s.parse().ok()),
         release: sentry::release_name!(),
-        send_default_pii: true,
+        send_default_pii: false,
         traces_sample_rate: 0.1,
         ..Default::default()
     });
@@ -164,8 +182,9 @@ async fn main() -> n0_error::Result<()> {
         }
     });
 
+    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
     tracing_subscriber::registry()
-        .with(tracing_subscriber::fmt::layer())
+        .with(tracing_subscriber::fmt::layer().with_filter(filter))
         .with(sentry_layer)
         .init();
 
@@ -218,8 +237,29 @@ async fn main() -> n0_error::Result<()> {
                 .await?;
             println!("OK.");
         }
-        Commands::Serve => {
-            let node = ListenNode::new(repo).await?;
+        Commands::Serve(args) => {
+            if !args.tcp_proxies.is_empty() {
+                let services = args
+                    .tcp_proxies
+                    .iter()
+                    .map(|host| TcpProxyData::from_host_port_str(host))
+                    .collect::<Result<Vec<_>, _>>()?;
+                let state = repo.load_state().await?;
+                state
+                    .update(&repo, |state| {
+                        for service in services {
+                            state.ensure_tcp_proxy(service, None);
+                        }
+                    })
+                    .await?;
+            }
+            let node = match args.listen_key_file {
+                Some(path) => {
+                    let secret_key = Repo::read_secret_key(&path).await?;
+                    ListenNode::with_secret_key(repo, secret_key).await?
+                }
+                None => ListenNode::new(repo).await?,
+            };
             let endpoint_id = node.endpoint_id();
             println!("listening as {}", endpoint_id);
             let bound_addrs = node.endpoint().bound_sockets();
@@ -253,7 +293,8 @@ async fn main() -> n0_error::Result<()> {
                     p.info.resource_id, p.info.data.host, p.info.data.port
                 )
             }
-            tokio::signal::ctrl_c().await?;
+            shutdown_signal().await?;
+            node.shutdown().await;
             println!()
         }
         Commands::Connect(args) => {
@@ -270,7 +311,7 @@ async fn main() -> n0_error::Result<()> {
                 handle.advertisment().host,
                 handle.advertisment().port,
             );
-            tokio::signal::ctrl_c().await?;
+            shutdown_signal().await?;
             handle.abort();
         }
         Commands::DnsDev(args) => match args {
@@ -299,3 +340,17 @@ async fn main() -> n0_error::Result<()> {
     }
     Ok(())
 }
+
+async fn shutdown_signal() -> std::io::Result<()> {
+    #[cfg(unix)]
+    {
+        let mut terminate =
+            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
+        tokio::select! {
+            res = tokio::signal::ctrl_c() => res,
+            _ = terminate.recv() => Ok(()),
+        }
+    }
+    #[cfg(not(unix))]
+    tokio::signal::ctrl_c().await
+}

lib/src/node.rs

@@ -59,10 +59,14 @@ pub struct ListenNode {
 }
 
 impl ListenNode {
-    #[instrument("listen-node", skip_all)]
     pub async fn new(repo: Repo) -> Result<Self> {
-        let config = repo.config().await?;
         let secret_key = repo.listen_key().await?;
+        Self::with_secret_key(repo, secret_key).await
+    }
+
+    #[instrument("listen-node", skip_all)]
+    pub async fn with_secret_key(repo: Repo, secret_key: SecretKey) -> Result<Self> {
+        let config = repo.config().await?;
         let endpoint = build_endpoint(secret_key, &config).await?;
         let state = repo.load_state().await?;
 
@@ -171,6 +175,12 @@ impl ListenNode {
     pub fn endpoint_id(&self) -> EndpointId {
         self.router.endpoint().id()
     }
+
+    pub async fn shutdown(&self) {
+        if let Err(err) = self.router.shutdown().await {
+            warn!("Listen node shutdown did not finish cleanly: {err:#}");
+        }
+    }
 }
 
 impl StateWrapper {

lib/src/repo.rs

@@ -1,4 +1,4 @@
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 
 use iroh::SecretKey;
 use log::{info, warn};
@@ -123,8 +123,18 @@ impl Repo {
             return self.create_key(&key_file_path).await;
         };
 
-        let key = tokio::fs::read(key_file_path).await?;
-        let key = key.as_slice().try_into().anyerr()?;
+        Self::read_secret_key(key_file_path).await
+    }
+
+    /// Reads a raw 32-byte ed25519 secret key from a file.
+    pub async fn read_secret_key(path: impl AsRef<Path>) -> Result<SecretKey> {
+        let path = path.as_ref();
+        let key = tokio::fs::read(path)
+            .await
+            .with_context(|_| format!("failed to read secret key {}", path.display()))?;
+        let key = key.as_slice().try_into().with_std_context(|_| {
+            format!("secret key {} must be exactly 32 bytes", path.display())
+        })?;
         Ok(SecretKey::from_bytes(key))
     }
 

lib/src/state.rs

@@ -29,6 +29,16 @@ impl State {
         }
     }
 
+    /// Enables the proxy for `data`, adding it if no proxy targets that host and port yet.
+    pub fn ensure_tcp_proxy(&mut self, data: TcpProxyData, label: Option<String>) {
+        if let Some(existing) = self.proxies.iter_mut().find(|p| p.info.data == data) {
+            existing.enabled = true;
+        } else {
+            self.proxies
+                .push(ProxyState::new(Advertisment::new(data, label)));
+        }
+    }
+
     pub fn remove_proxy(&mut self, resouce_id: &str) -> Option<ProxyState> {
         if let Some(idx) = self
             .proxies
@@ -296,6 +306,19 @@ mod tests {
         assert!(err.to_string().contains("invalid port"));
     }
 
+    #[test]
+    fn ensure_tcp_proxy_is_idempotent() {
+        let data = TcpProxyData::from_host_port_str("agent.test:7777").unwrap();
+        let mut state = State::default();
+        state.ensure_tcp_proxy(data.clone(), Some("exec".to_string()));
+        state.proxies[0].enabled = false;
+        state.ensure_tcp_proxy(data.clone(), None);
+        assert_eq!(state.proxies.len(), 1);
+        assert!(state.proxies[0].enabled);
+        assert_eq!(state.proxies[0].info.data, data);
+        assert_eq!(state.proxies[0].info.label.as_deref(), Some("exec"));
+    }
+
     // #[test]
     // fn test_tcp_proxy_has_codename() {
     //     let proxy = TcpProxy::new("127.0.0.1".to_string(), 8080);