feat(remote): forward ports of a dev container (#2072)

Refs #1083

A dev container's ports can be forwarded: the Ports tool window shows in a dev container window, `forwardPorts` of devcontainer.json are forwarded on open (keeping the local number when free, falling back otherwise), and ports forwarded by hand are restored per folder and configuration. Each forwarded connection is a loopback listener whose connections run `docker exec … openidea-server --connect localhost <port>` through the container's own network, so no published ports or extra tools in the image are needed. The container tunnel lives in `ide-remote::container_tunnel`; tunnels own their connections and Docker children, so stopping a forward, closing the window or losing the connection leaves no listeners or processes behind (the same ownership fix applies to SSH forwards).

Co-authored-by: andreabasile <[email protected]>
Цей коміт міститься в:
Andrea Basileтаandreabasile authored and GitHub committed 2026-11-08 06:57:19 +08:00
1 джерело 826e5837b1
коміт 7ecb452350
15 files changed
+870 -44

No files matched your search

+15 -1
Переглянути файл
@@ -15,8 +15,22 @@ fn main() {
println!("openidea-server {} (protocol {})", env!("CARGO_PKG_VERSION"), rpc::PROTOCOL_VERSION);
return;
}
if args.first().is_some_and(|a| a == "--connect") {
// One forwarded connection (see `ide_host::relay`): stdout carries its bytes.
let code = match ide_host::relay::parse_args(&args[1..]).and_then(|(host, port)| {
ide_host::relay::relay(&host, port, std::io::stdin(), std::io::stdout().lock())
.map_err(|e| format!("cannot connect to {host}:{port}: {e}"))
}) {
Ok(()) => 0,
Err(e) => {
eprintln!("{e}");
1
}
};
std::process::exit(code);
}
if !args.iter().any(|a| a == "--stdio") {
eprintln!("usage: openidea-server --stdio | --version");
eprintln!("usage: openidea-server --stdio | --connect <host> <port> | --version");
std::process::exit(2);
}
// stdout carries the protocol: logs go to stderr.
+1
Переглянути файл
@@ -14,6 +14,7 @@ mod phpstan;
pub mod plugins;
mod project;
mod proxy;
pub mod relay;
mod requests;
pub mod rpc;
pub mod shell_env;
+191
Переглянути файл
@@ -0,0 +1,191 @@
//! `openidea-server --connect <host> <port>`: one forwarded connection. The desktop starts it
//! in a dev container (`docker exec -i`) for each connection to a forwarded port and pipes the
//! connection through stdin / stdout, so a port of the container is reachable wherever the
//! container's network is not.
use std::io::{self, Read, Write};
use std::net::{Shutdown, TcpStream};
use std::sync::{Arc, Mutex, mpsc};
use std::time::{Duration, Instant};
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const EOF_GRACE: Duration = Duration::from_secs(5);
/// `<host> <port>` from the arguments after `--connect`.
pub fn parse_args(args: &[String]) -> Result<(String, u16), String> {
let [host, port] = args else { return Err("usage: openidea-server --connect <host> <port>".into()) };
if host.is_empty() || host.starts_with('-') || host.chars().any(|c| c.is_control() || c.is_whitespace()) {
return Err(format!("invalid host {host:?}"));
}
match port.parse::<u16>() {
Ok(p) if p != 0 => Ok((host.clone(), p)),
_ => Err(format!("invalid port {port:?}")),
}
}
/// Connects to `host:port`, copies `input` to it and its replies to `output`, and returns when
/// the target is done (the copy of `input` may still be blocked: the caller exits). Fails only
/// when the connection can't be opened. An input EOF allows five idle seconds for a final reply.
pub fn relay(host: &str, port: u16, mut input: impl Read + Send + 'static, mut output: impl Write) -> io::Result<()> {
let mut from_target = connect(host, port)?;
from_target.set_read_timeout(Some(Duration::from_millis(100)))?;
let mut to_target = from_target.try_clone()?;
let writing = Arc::new(Mutex::new(None::<Instant>));
let stalled = writing.clone();
let (finished, eof) = mpsc::channel();
std::thread::spawn(move || {
let mut buf = [0; 16 * 1024];
loop {
let n = match input.read(&mut buf) {
Ok(0) => break,
Ok(n) => n,
Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
Err(_) => break,
};
*writing.lock().unwrap() = Some(Instant::now());
let result = to_target.write_all(&buf[..n]);
*writing.lock().unwrap() = None;
if result.is_err() {
break;
}
}
// The client closed its side: tell the target, which may still answer.
let _ = to_target.shutdown(Shutdown::Write);
let _ = finished.send(Instant::now());
});
let mut buf = [0u8; 16 * 1024];
let mut input_closed = None;
loop {
// A blocked write must not hide Docker closing stdin indefinitely.
if stalled.lock().unwrap().is_some_and(|at| at.elapsed() >= CONNECT_TIMEOUT) {
break;
}
if let Ok(at) = eof.try_recv() {
input_closed = Some(at);
}
// Docker closing stdin must end an idle relay too. Keep a bounded grace for
// a target replying to a half-closed request before tearing down the socket.
if input_closed.is_some_and(|at| at.elapsed() >= EOF_GRACE) {
break;
}
match from_target.read(&mut buf) {
Err(e) if matches!(e.kind(), io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut) => continue,
Ok(0) | Err(_) => break,
Ok(n) => {
if output.write_all(&buf[..n]).and_then(|()| output.flush()).is_err() {
break;
}
if input_closed.is_some() {
input_closed = Some(Instant::now());
}
}
}
}
let _ = from_target.shutdown(Shutdown::Both);
Ok(())
}
fn connect(host: &str, port: u16) -> io::Result<TcpStream> {
use std::net::ToSocketAddrs;
let mut last = io::Error::new(io::ErrorKind::NotFound, format!("{host} has no address"));
for addr in (host, port).to_socket_addrs()? {
match TcpStream::connect_timeout(&addr, CONNECT_TIMEOUT) {
Ok(s) => return Ok(s),
Err(e) => last = e,
}
}
Err(last)
}
#[cfg(test)]
mod tests {
use super::*;
use std::net::TcpListener;
fn args(a: &[&str]) -> Vec<String> {
a.iter().map(|s| (*s).to_owned()).collect()
}
#[test]
fn reads_host_and_port() {
assert_eq!(parse_args(&args(&["localhost", "8000"])).unwrap(), ("localhost".into(), 8000));
for bad in [&["localhost"][..], &["localhost", "0"], &["localhost", "70000"], &["-x", "80"], &["a b", "80"]] {
assert!(parse_args(&args(bad)).is_err(), "{bad:?}");
}
}
#[test]
fn pipes_a_request_and_its_reply() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let server = std::thread::spawn(move || {
let (mut s, _) = listener.accept().unwrap();
let mut request = String::new();
s.read_to_string(&mut request).unwrap();
s.write_all(format!("got {request}").as_bytes()).unwrap();
});
let mut out = Vec::new();
relay("127.0.0.1", port, &b"ping"[..], &mut out).unwrap();
server.join().unwrap();
assert_eq!(out, b"got ping");
}
#[test]
fn input_eof_ends_a_relay_even_when_the_target_ignores_it() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let (done, wait) = mpsc::channel();
let server = std::thread::spawn(move || {
let (mut socket, _) = listener.accept().unwrap();
let mut request = Vec::new();
socket.read_to_end(&mut request).unwrap();
// Remain open after seeing EOF, like an idle keep-alive server.
wait.recv_timeout(Duration::from_secs(10)).unwrap();
});
let start = Instant::now();
relay("127.0.0.1", port, io::empty(), io::sink()).unwrap();
assert!(start.elapsed() < Duration::from_secs(8));
done.send(()).unwrap();
server.join().unwrap();
}
#[test]
fn final_responses_keep_flowing_after_input_eof() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let server = std::thread::spawn(move || {
let (mut socket, _) = listener.accept().unwrap();
socket.read_to_end(&mut Vec::new()).unwrap();
for _ in 0..7 {
socket.write_all(b"x").unwrap();
std::thread::sleep(Duration::from_secs(1));
}
});
let mut out = Vec::new();
relay("127.0.0.1", port, io::empty(), &mut out).unwrap();
assert_eq!(out, b"xxxxxxx");
server.join().unwrap();
}
#[test]
fn a_target_that_never_reads_cannot_keep_a_relay_alive() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let port = listener.local_addr().unwrap().port();
let (done, wait) = mpsc::channel();
let server = std::thread::spawn(move || {
let (_socket, _) = listener.accept().unwrap();
wait.recv_timeout(Duration::from_secs(30)).unwrap();
});
let start = Instant::now();
relay("127.0.0.1", port, io::repeat(0).take(32 * 1024 * 1024), io::sink()).unwrap();
assert!(start.elapsed() < Duration::from_secs(25));
done.send(()).unwrap();
server.join().unwrap();
}
#[test]
fn a_closed_port_is_an_error() {
let port = TcpListener::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port();
assert!(relay("127.0.0.1", port, io::empty(), io::sink()).is_err());
}
}
+313
Переглянути файл
@@ -0,0 +1,313 @@
//! Loopback port forwards carried through a container's stdin/stdout relay.
use crate::devcontainer::Up;
use ide_core::{CancellationToken, IdeResult};
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
/// How to run the installed server in a started container (forwarded connections).
pub struct Exec {
docker: PathBuf,
up: Up,
bin: String,
}
impl Exec {
pub fn new(docker: PathBuf, up: Up, bin: String) -> Self {
Self { docker, up, bin }
}
/// The arguments of `docker` for one connection to `localhost:<port>` in the container.
fn connect_args(&self, port: u16) -> Vec<String> {
let port = port.to_string();
["exec", "-i", "-u", &self.up.user, &self.up.container, &self.bin, "--connect", "localhost", &port]
.map(str::to_owned)
.into()
}
}
/// A forwarded port of the container; dropping it closes its listener and connections.
pub struct Tunnel {
port: u16,
cancel: CancellationToken,
closed: CancellationToken,
}
impl Tunnel {
/// Listens on local port `preferred` when it is free, else on any free one, and forwards
/// each connection to `localhost:<port>` in the container. Nothing needs to listen there yet.
pub async fn listen(exec: Arc<Exec>, preferred: u16, port: u16) -> IdeResult<Tunnel> {
let listener = match tokio::net::TcpListener::bind(("127.0.0.1", preferred)).await {
Ok(l) => l,
Err(_) => tokio::net::TcpListener::bind(("127.0.0.1", 0)).await?,
};
let local = listener.local_addr()?.port();
let cancel = CancellationToken::new();
let stop = cancel.clone();
let closed = CancellationToken::new();
let done = closed.clone();
tokio::spawn(async move {
let mut connections = tokio::task::JoinSet::new();
loop {
tokio::select! {
_ = stop.cancelled() => break,
Some(_) = connections.join_next(), if !connections.is_empty() => {},
accepted = listener.accept() => {
let Ok((socket, _)) = accepted else { break };
connections.spawn(forward(exec.clone(), socket, port, stop.clone()));
}
}
}
stop.cancel();
drop(listener);
while connections.join_next().await.is_some() {}
done.cancel();
});
Ok(Tunnel { port: local, cancel, closed })
}
/// Stops listening and waits for all Docker children to be reaped.
pub async fn close(self) {
self.cancel.cancel();
self.closed.cancelled().await;
}
/// The local port to connect to.
pub fn port(&self) -> u16 {
self.port
}
}
impl Drop for Tunnel {
fn drop(&mut self) {
self.cancel.cancel();
}
}
async fn forward(exec: Arc<Exec>, mut socket: tokio::net::TcpStream, port: u16, cancel: CancellationToken) {
let mut cmd = tokio::process::Command::new(&exec.docker);
ide_core::process::prepare(cmd.as_std_mut());
cmd.args(exec.connect_args(port))
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
let mut child = match cmd.spawn() {
Ok(c) => c,
Err(e) => return tracing::warn!("forwarding port {port}: cannot start docker: {e}"),
};
let (Some(mut stdin), Some(mut stdout), Some(mut stderr)) =
(child.stdin.take(), child.stdout.take(), child.stderr.take())
else {
return;
};
let (mut from_client, mut to_client) = socket.split();
let input_done = CancellationToken::new();
let eof = input_done.clone();
let upstream = async move {
let _ = tokio::io::copy(&mut from_client, &mut stdin).await;
drop(stdin);
eof.cancel();
};
let downstream = async {
let mut buf = [0; 16 * 1024];
loop {
let read = stdout.read(&mut buf);
tokio::pin!(read);
let result = tokio::select! {
result = &mut read => result,
_ = input_done.cancelled() => {
match tokio::time::timeout(Duration::from_secs(5), &mut read).await {
Ok(result) => result,
Err(_) => break,
}
}
};
let Ok(n) = result else { break };
if n == 0 || to_client.write_all(&buf[..n]).await.is_err() {
break;
}
}
let _ = to_client.shutdown().await;
};
let diagnostics = async {
let mut text = Vec::new();
let _ = (&mut stderr).take(4096).read_to_end(&mut text).await;
let _ = tokio::io::copy(&mut stderr, &mut tokio::io::sink()).await;
text
};
let transfer = async {
tokio::pin!(downstream);
tokio::select! {
_ = &mut downstream => {},
_ = upstream => {
// The downstream inactivity timer still permits a streaming final reply.
downstream.await;
}
}
};
let exchange = async {
tokio::select! {
_ = cancel.cancelled() => {},
_ = transfer => {},
}
let _ = child.kill().await;
};
let (_, diagnostics) = tokio::join!(exchange, diagnostics);
if !diagnostics.is_empty() {
tracing::warn!("forwarding port {port}: {}", String::from_utf8_lossy(&diagnostics).trim());
}
}
#[cfg(test)]
mod tests {
use super::*;
const ID: &str = "3f1c2b9a7d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8";
#[test]
fn a_forwarded_connection_runs_the_server_as_a_relay() {
let up = Up { container: ID.into(), user: "vscode".into(), folder: "/w".into() };
let exec = Exec { docker: "docker".into(), up, bin: "/s/openidea-server".into() };
assert_eq!(
exec.connect_args(8000),
["exec", "-i", "-u", "vscode", ID, "/s/openidea-server", "--connect", "localhost", "8000"]
);
}
/// A stand-in for `docker` that answers with its arguments, then echoes the connection.
#[cfg(unix)]
#[tokio::test]
async fn a_tunnel_pipes_each_connection_through_docker_exec() {
use std::os::unix::fs::PermissionsExt;
use tokio::io::AsyncReadExt;
let dir = tempfile::tempdir().unwrap();
let docker = dir.path().join("docker");
std::fs::write(&docker, "#!/bin/sh\necho \"$*\"\nexec cat\n").unwrap();
std::fs::set_permissions(&docker, std::fs::Permissions::from_mode(0o755)).unwrap();
let up = Up { container: ID.into(), user: "root".into(), folder: "/w".into() };
let exec = Arc::new(Exec { docker, up, bin: "/s/openidea-server".into() });
let tunnel = Tunnel::listen(exec, 0, 8000).await.unwrap();
for _ in 0..2 {
let mut socket = tokio::net::TcpStream::connect(("127.0.0.1", tunnel.port())).await.unwrap();
socket.write_all(b"ping").await.unwrap();
socket.shutdown().await.unwrap();
let mut reply = String::new();
tokio::time::timeout(Duration::from_secs(10), socket.read_to_string(&mut reply)).await.unwrap().unwrap();
assert_eq!(reply, format!("exec -i -u root {ID} /s/openidea-server --connect localhost 8000\nping"));
}
let port = tunnel.port();
tunnel.close().await;
assert!(tokio::net::TcpStream::connect(("127.0.0.1", port)).await.is_err(), "a dropped tunnel stops listening");
}
#[cfg(unix)]
fn docker_script(script: &str) -> (tempfile::TempDir, Arc<Exec>) {
use std::os::unix::fs::PermissionsExt;
let dir = tempfile::tempdir().unwrap();
let docker = dir.path().join("docker");
std::fs::write(&docker, format!("#!/bin/sh\n{script}\n")).unwrap();
std::fs::set_permissions(&docker, std::fs::Permissions::from_mode(0o755)).unwrap();
let up = Up { container: ID.into(), user: "root".into(), folder: "/w".into() };
let exec = Arc::new(Exec::new(docker, up, "/s/openidea-server".into()));
(dir, exec)
}
#[cfg(unix)]
#[tokio::test]
async fn dropping_a_tunnel_closes_active_clients_and_reaps_children() {
let (_dir, exec) = docker_script("echo $$; exec cat");
let tunnel = Tunnel::listen(exec, 0, 8000).await.unwrap();
let port = tunnel.port();
let mut clients = Vec::new();
let mut pids = Vec::new();
for _ in 0..2 {
let socket = tokio::net::TcpStream::connect(("127.0.0.1", port)).await.unwrap();
let mut socket = tokio::io::BufReader::new(socket);
let mut pid = String::new();
use tokio::io::AsyncBufReadExt;
tokio::time::timeout(Duration::from_secs(5), socket.read_line(&mut pid)).await.unwrap().unwrap();
pids.push(pid.trim().parse::<u32>().unwrap());
clients.push(socket);
}
drop(tunnel);
for mut socket in clients {
let mut reply = Vec::new();
tokio::time::timeout(Duration::from_secs(5), socket.read_to_end(&mut reply)).await.unwrap().unwrap();
}
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let mut alive = false;
for pid in &pids {
let status = tokio::process::Command::new("ps")
.args(["-p", &pid.to_string(), "-o", "stat="])
.output()
.await
.unwrap();
alive |= !status.stdout.is_empty();
}
if !alive {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.unwrap();
assert!(tokio::net::TcpStream::connect(("127.0.0.1", port)).await.is_err());
}
#[cfg(unix)]
#[tokio::test]
async fn target_eof_closes_a_client_that_has_not_sent_eof() {
let (_dir, exec) = docker_script("echo $$; printf done");
let tunnel = Tunnel::listen(exec, 0, 8000).await.unwrap();
let mut socket = tokio::net::TcpStream::connect(("127.0.0.1", tunnel.port())).await.unwrap();
let mut reply = String::new();
tokio::time::timeout(Duration::from_secs(5), socket.read_to_string(&mut reply)).await.unwrap().unwrap();
let (pid, body) = reply.split_once('\n').unwrap();
let pid = pid.parse::<u32>().unwrap();
assert_eq!(body, "done");
tokio::time::timeout(Duration::from_secs(5), async {
loop {
let status = tokio::process::Command::new("ps")
.args(["-p", &pid.to_string(), "-o", "stat="])
.output()
.await
.unwrap();
if status.stdout.is_empty() {
break;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn stderr_is_drained_while_the_connection_is_open() {
let (_dir, exec) = docker_script("dd if=/dev/zero bs=65536 count=4 >&2 2>/dev/null; printf done");
let tunnel = Tunnel::listen(exec, 0, 8000).await.unwrap();
let mut socket = tokio::net::TcpStream::connect(("127.0.0.1", tunnel.port())).await.unwrap();
let mut reply = String::new();
tokio::time::timeout(Duration::from_secs(5), socket.read_to_string(&mut reply)).await.unwrap().unwrap();
assert_eq!(reply, "done");
tunnel.close().await;
}
#[cfg(unix)]
#[tokio::test]
async fn occupied_preferred_port_falls_back_to_a_free_loopback_port() {
let (_dir, exec) = docker_script("exec cat");
let busy = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
let preferred = busy.local_addr().unwrap().port();
let tunnel = Tunnel::listen(exec, preferred, 8000).await.unwrap();
assert_ne!(tunnel.port(), preferred);
let socket = tokio::net::TcpStream::connect(("127.0.0.1", tunnel.port())).await.unwrap();
assert!(socket.peer_addr().unwrap().ip().is_loopback());
tunnel.close().await;
}
}
+1
Переглянути файл
@@ -1,6 +1,7 @@
//! Remote work over SSH: SFTP deployment ([`deploy`], [`mapping`]) and run
//! configurations on run targets ([`target`]; Docker and Compose ones in [`docker`]).
pub mod container_tunnel;
pub mod deploy;
pub mod devcontainer;
pub mod docker;
+7 -2
Переглянути файл
@@ -61,11 +61,16 @@ impl Drop for Tunnel {
}
async fn forward(listener: TcpListener, session: Arc<Session>, host: String, port: u16) {
let mut connections = tokio::task::JoinSet::new();
loop {
let Ok((mut socket, _)) = listener.accept().await else { return };
let accepted = tokio::select! {
Some(_) = connections.join_next(), if !connections.is_empty() => continue,
accepted = listener.accept() => accepted,
};
let Ok((mut socket, _)) = accepted else { return };
let session = session.clone();
let host = host.clone();
tokio::spawn(async move {
connections.spawn(async move {
match session.direct_tcpip(&host, port).await {
Ok(mut stream) => {
let _ = tokio::io::copy_bidirectional(&mut socket, &mut stream).await;
+19
Переглянути файл
@@ -236,6 +236,25 @@ async fn forwarding_keeps_the_port_number_when_it_is_free() {
round_trip(&other).await;
}
#[tokio::test]
async fn dropping_a_tunnel_closes_active_connections() {
let env = env().await;
let session = Arc::new(env.connect_trusted(&env.target(SshAuth::Password), Some("secret")).await.ok().unwrap());
let tunnel = Tunnel::listen(session.clone(), "127.0.0.1", env.echo_port).await.unwrap();
let mut socket = TcpStream::connect(("127.0.0.1", tunnel.port())).await.unwrap();
socket.write_all(b"ping").await.unwrap();
let mut reply = [0; 4];
socket.read_exact(&mut reply).await.unwrap();
assert_eq!(&reply, b"ping");
drop(tunnel);
let mut byte = [0];
assert_eq!(
tokio::time::timeout(std::time::Duration::from_secs(5), socket.read(&mut byte)).await.unwrap().unwrap(),
0
);
assert!(!session.is_closed());
}
#[tokio::test]
async fn key_pair_authentication_with_and_without_passphrase() {
let env = env().await;
+62 -4
Переглянути файл
@@ -1,6 +1,7 @@
//! The project's `.devcontainer/devcontainer.json` (or `.devcontainer.json`) as a
//! remote host reads it: `remoteEnv` for every child process of the IDE backend and
//! `postStartCommand` once per backend start. No container is built.
//! `postStartCommand` once per backend start. No container is built. The desktop reads
//! `forwardPorts` of a dev container it opens.
use serde_json::Value;
use std::collections::BTreeMap;
@@ -15,6 +16,10 @@ pub struct DevContainer {
/// Values as written (`None`: unset the variable), substitutions not applied.
pub remote_env: BTreeMap<String, Option<String>>,
pub post_start: Option<LifecycleCommand>,
/// `forwardPorts` of the container itself (a number, `localhost:<port>` or
/// `127.0.0.1:<port>`), in order, without duplicates. Other entries, such as the port of
/// another Compose service, are left out.
pub forward_ports: Vec<u16>,
}
/// A lifecycle command in its three forms.
@@ -83,10 +88,15 @@ pub fn find(root: &Path) -> Option<PathBuf> {
/// it can't be read or parsed.
pub fn load(root: &Path) -> Result<Option<DevContainer>, String> {
let Some(path) = find(root) else { return Ok(None) };
let text = std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display()))?;
load_file(&path).map(Some)
}
/// Reads the configuration at `path`.
pub fn load_file(path: &Path) -> Result<DevContainer, String> {
let text = std::fs::read_to_string(path).map_err(|e| format!("{}: {e}", path.display()))?;
parse(&text).map(|mut c| {
c.path = path;
Some(c)
c.path = path.to_owned();
c
})
}
@@ -111,9 +121,31 @@ pub fn parse(text: &str) -> Result<DevContainer, String> {
if let Some(cmd) = top.get("postStartCommand") {
config.post_start = LifecycleCommand::parse(cmd).map_err(|e| format!("postStartCommand: {e}"))?;
}
// Only the desktop uses these: an entry it can't forward never makes the file unreadable.
if let Some(Value::Array(ports)) = top.get("forwardPorts") {
for port in ports.iter().filter_map(container_port) {
if !config.forward_ports.contains(&port) {
config.forward_ports.push(port);
}
}
}
Ok(config)
}
/// A `forwardPorts` entry that is a port of the container itself.
fn container_port(v: &Value) -> Option<u16> {
let port = match v {
Value::Number(n) => n.as_u64()?,
Value::String(s) => match s.trim().split_once(':') {
Some(("localhost" | "127.0.0.1", port)) => port.parse().ok()?,
Some(_) => return None,
None => s.trim().parse().ok()?,
},
_ => return None,
};
u16::try_from(port).ok().filter(|p| *p != 0)
}
/// Expands `${localEnv:NAME}`, `${containerEnv:NAME}` and `${env:NAME}` (all the remote
/// host's environment, through `env`; `${…:NAME:default}` when unset), and
/// `${localWorkspaceFolder}` / `${containerWorkspaceFolder}` (and their `…Basename`) to
@@ -200,6 +232,32 @@ mod tests {
assert_eq!(c.post_start, Some(LifecycleCommand::Shell("make dev".into())));
}
#[test]
fn forward_ports_of_the_container_only() {
let c = parse(
r#"{ "forwardPorts": [8000, "5173", "localhost:3000", "127.0.0.1:8000", "db:5432", 0, 70000, "x", null] }"#,
)
.unwrap();
assert_eq!(c.forward_ports, [8000, 5173, 3000]);
assert!(parse(r#"{ "forwardPorts": "8000" }"#).unwrap().forward_ports.is_empty());
}
#[test]
fn forward_ports_skip_malformed_entries_and_keep_valid_bounds() {
let config = parse(
r#"{
"remoteEnv": {"PORT": "8000"},
"forwardPorts": [1, 65535, " 8000 ", "localhost:8000", "127.0.0.1:8000",
-1, 1.5, 65536, 18446744073709551616, true, false, {}, [], null,
"", " ", ":80", "db:5432", "localhost:", "localhost:0", "localhost:65536",
"localhost:80:90", "localhost: 80", "http://localhost:80", "::1:80"]
}"#,
)
.unwrap();
assert_eq!(config.forward_ports, [1, 65535, 8000]);
assert_eq!(config.remote_env["PORT"].as_deref(), Some("8000"));
}
#[test]
fn post_start_command_forms() {
let parse_cmd = |json: &str| parse(&format!(r#"{{ "postStartCommand": {json} }}"#)).map(|c| c.post_start);
+38 -3
Переглянути файл
@@ -4,11 +4,16 @@
//!
//! The container id, user and folder come from the CLI's output, are checked, and only ever
//! reach `docker exec` as separate arguments; only fixed scripts reach `sh -c`.
//!
//! A forwarded port is a local listener whose every connection runs
//! `openidea-server --connect localhost <port>` in the container and pipes through it: the
//! container's own network is reached the same way on every Docker setup.
use crate::remote_dev::{Progress, RemoteStage, SERVER_BIN};
use crate::remote_wsl::{Probe, Tail, install_script, parse_probe, probe_script, upload, version_dir};
use ide_core::{IdeError, IdeResult};
use ide_jsonrpc::{Handler, Peer};
use ide_remote::container_tunnel::Exec;
use ide_remote::devcontainer::{Up, parse_up};
use std::path::{Path, PathBuf};
use std::process::Stdio;
@@ -172,7 +177,8 @@ fn server_args(up: &Up, bin: &str) -> Vec<String> {
}
/// Opens the dev container of local `folder`: starts it with the CLI, installs and starts the
/// server in it. Returns the connection and the project folder in the container.
/// server in it. Returns the connection, the project folder in the container and how to run the
/// server there.
pub(crate) async fn connect(
state: &crate::state::AppState,
folder: &str,
@@ -180,7 +186,7 @@ pub(crate) async fn connect(
handler: Arc<dyn Handler>,
tail: &Tail,
progress: &Arc<Progress>,
) -> IdeResult<(Peer, String)> {
) -> IdeResult<(Peer, String, Exec)> {
let local = PathBuf::from(folder);
// `devcontainer up` runs the project's code on this machine (`initializeCommand`, the build).
if !state.host.trust.trusted(&local).unwrap_or(false) {
@@ -210,7 +216,8 @@ pub(crate) async fn connect(
let bin = ensure_server(state, &docker, &up, progress).await?;
progress(RemoteStage::Starting, None);
let peer = spawn(&docker, &up, &bin, handler, tail)?;
Ok((peer, up.folder))
let folder = up.folder.clone();
Ok((peer, folder, Exec::new(docker, up, bin)))
}
fn spawn(docker: &Path, up: &Up, bin: &str, handler: Arc<dyn Handler>, tail: &Tail) -> IdeResult<Peer> {
@@ -223,6 +230,7 @@ fn spawn(docker: &Path, up: &Up, bin: &str, handler: Arc<dyn Handler>, tail: &Ta
#[cfg(test)]
mod tests {
use super::*;
use ide_remote::container_tunnel::Tunnel;
const ID: &str = "3f1c2b9a7d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8";
@@ -375,6 +383,33 @@ echo '{{"outcome":"success","containerId":"{ID}","remoteUser":"vscode","remoteWo
assert!(files.to_string().contains("main.py"), "the project is the local folder: {files}");
tokio::time::sleep(Duration::from_secs(2)).await;
assert_eq!(docker_sh(&docker, &up.container, "wc -l < /tmp/post-start"), "1", "postStartCommand ran once");
// A port only the container's own network reaches (perl is in every Debian image).
let listen = "perl -MIO::Socket::INET -e '$s = IO::Socket::INET->new(LocalAddr => \"127.0.0.1:8000\", \
Listen => 5, ReuseAddr => 1) or die; while ($c = $s->accept) { print $c \"from the container\\n\"; close $c }'";
let started = std::process::Command::new(&docker)
.args(["exec", "-d", &up.container, "sh", "-c", listen])
.status()
.unwrap();
assert!(started.success());
let exec = Arc::new(Exec::new(
docker.clone(),
Up { container: up.container.clone(), user: up.user.clone(), folder: up.folder.clone() },
bin.clone(),
));
let tunnel = Tunnel::listen(exec, 0, 8000).await.unwrap();
let mut reply = String::new();
for _ in 0..50 {
use tokio::io::AsyncReadExt;
let mut socket = tokio::net::TcpStream::connect(("127.0.0.1", tunnel.port())).await.unwrap();
reply.clear();
let _ = socket.read_to_string(&mut reply).await;
if !reply.is_empty() {
break;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
assert_eq!(reply, "from the container\n");
drop(tunnel);
drop(peer);
let removed = std::process::Command::new(&docker).args(["rm", "-f", &up.container]).output().unwrap();
assert!(removed.status.success());
+89 -9
Переглянути файл
@@ -71,8 +71,25 @@ pub struct Connection {
next_channel: std::sync::atomic::AtomicU64,
/// Keeps the SSH connection alive; `None` for WSL and dev containers.
session: Option<Arc<ide_ssh::Session>>,
/// How to reach a dev container's ports; `None` for SSH and WSL.
exec: Option<Arc<ide_remote::container_tunnel::Exec>>,
/// Remote ports forwarded to this machine (Ports tool window), by remote port.
forwards: Mutex<Vec<(u16, ide_ssh::Tunnel)>>,
forwards: Mutex<Vec<(u16, Forward)>>,
}
/// A forwarded port: through the SSH connection, or `docker exec` in a dev container.
enum Forward {
Ssh(ide_ssh::Tunnel),
Container(ide_remote::container_tunnel::Tunnel),
}
impl Forward {
fn port(&self) -> u16 {
match self {
Self::Ssh(t) => t.port(),
Self::Container(t) => t.port(),
}
}
}
type Channels = Arc<Mutex<HashMap<u64, (Channel<ChannelFrame>, u64)>>>;
@@ -129,6 +146,7 @@ impl Connections {
pub fn disconnect(&self, window: &str) {
if let Some(c) = self.0.lock().remove(window) {
c.peer.kill();
c.forwards.lock().clear();
}
}
}
@@ -433,6 +451,7 @@ pub async fn remote_connect(
let handler =
Arc::new(ClientHandler { app: window.app_handle().clone(), window: label.clone(), channels: channels.clone() });
let tail = crate::remote_wsl::Tail::default();
let mut exec = None;
let (peer, session, host, title_tag) = match &target {
RemoteTarget::Ssh { config: id } => {
let config = state.ssh.config(id)?;
@@ -453,9 +472,10 @@ pub async fn remote_connect(
(peer, None, distro.clone(), format!("WSL: {distro}"))
}
RemoteTarget::DevContainer { folder, config } => {
let (peer, root) =
let (peer, root, run) =
crate::remote_container::connect(&state, folder, config, handler, &tail, &progress).await?;
path = root;
exec = Some(Arc::new(run));
let name = std::path::Path::new(folder)
.file_name()
.map_or_else(|| folder.clone(), |n| n.to_string_lossy().into_owned());
@@ -506,10 +526,22 @@ pub async fn remote_connect(
next_channel: std::sync::atomic::AtomicU64::new(1),
session,
forwards: Mutex::new(Vec::new()),
exec,
});
if let RemoteTarget::DevContainer { folder, config } = &target {
forward_configured_ports(&connection, folder, config).await;
}
state.remote_dev.disconnect(&label);
state.host.close_project(&label);
state.remote_dev.0.lock().insert(label.clone(), connection.clone());
let weak = Arc::downgrade(&connection);
let peer = connection.peer.clone();
tauri::async_runtime::spawn(async move {
peer.closed().await;
if let Some(connection) = weak.upgrade() {
connection.forwards.lock().clear();
}
});
let plugins = state.plugins.clone();
tauri::async_runtime::spawn(async move {
sync_plugin_languages(&plugins, &connection, &label).await;
@@ -659,13 +691,12 @@ pub async fn remote_forward_port(
if let Some(p) = forwarded(&c).into_iter().find(|p| p.remote_port == remote_port) {
return Ok(p);
}
let session = c
.session
.clone()
.ok_or_else(|| IdeError::unavailable("Forwarding ports is not supported in a WSL project yet"))?;
let tunnel = ide_ssh::Tunnel::listen_on(session, Some(remote_port), "localhost", remote_port).await?;
let tunnel = open_forward(&c, remote_port).await?;
let port = ForwardedPort { remote_port, local_port: tunnel.port() };
let mut forwards = c.forwards.lock();
if c.peer.is_closed() {
return Err(IdeError::cancelled());
}
if let Some(existing) = forwards.iter().find(|(remote, _)| *remote == remote_port) {
return Ok(ForwardedPort { remote_port, local_port: existing.1.port() });
}
@@ -673,11 +704,60 @@ pub async fn remote_forward_port(
Ok(port)
}
/// Forwards `remote_port` to the same local port when it is free, else to a free one.
async fn open_forward(c: &Connection, remote_port: u16) -> IdeResult<Forward> {
if let Some(session) = c.session.clone() {
let tunnel = ide_ssh::Tunnel::listen_on(session, Some(remote_port), "localhost", remote_port).await?;
return Ok(Forward::Ssh(tunnel));
}
if let Some(exec) = c.exec.clone() {
return Ok(Forward::Container(
ide_remote::container_tunnel::Tunnel::listen(exec, remote_port, remote_port).await?,
));
}
Err(IdeError::unavailable("Forwarding ports is not supported in a WSL project yet"))
}
/// Forwards the `forwardPorts` of the dev container's configuration (the local folder's).
async fn forward_configured_ports(c: &Connection, folder: &str, config: &str) {
let local = std::path::PathBuf::from(folder);
let config = config.to_owned();
let ports = tokio::task::spawn_blocking(move || {
let path = match config.trim() {
"" => ide_workspace::devcontainer::find(&local),
config => Some(local.join(config)),
};
path.map(|path| ide_workspace::devcontainer::load_file(&path).map(|c| c.forward_ports)).transpose()
})
.await;
let ports = match ports {
Ok(Ok(Some(ports))) => ports,
Ok(Ok(None)) => return,
result => return tracing::warn!("{}: forwardPorts: {result:?}", c.host),
};
for remote_port in ports {
if c.forwards.lock().iter().any(|(remote, _)| *remote == remote_port) {
continue;
}
match open_forward(c, remote_port).await {
Ok(tunnel) => c.forwards.lock().push((remote_port, tunnel)),
Err(e) => tracing::warn!("{}: cannot forward port {remote_port}: {e}", c.host),
}
}
}
/// Stops forwarding `remote_port`.
#[tauri::command]
#[specta::specta]
pub fn remote_stop_forwarding(window: Window, state: State<'_, AppState>, remote_port: u16) -> IdeResult<()> {
connection(&window, &state)?.forwards.lock().retain(|(remote, _)| *remote != remote_port);
pub async fn remote_stop_forwarding(window: Window, state: State<'_, AppState>, remote_port: u16) -> IdeResult<()> {
let c = connection(&window, &state)?;
let tunnel = {
let mut forwards = c.forwards.lock();
forwards.iter().position(|(remote, _)| *remote == remote_port).map(|index| forwards.remove(index).1)
};
if let Some(Forward::Container(tunnel)) = tunnel {
tunnel.close().await;
}
Ok(())
}
+12 -5
Переглянути файл
@@ -15,12 +15,19 @@ import { RemoteStatus } from "./RemoteStatus";
import { closeConnectDialog, openConnectDialog } from "./model";
import { restorePorts, usePorts } from "./ports";
/** SSH tunnels are available while this window is connected to an SSH host. */
/** Forwarded ports are available while this window is connected to an SSH host or a dev container. */
function trackPortsWindow(): () => void {
let off: (() => void) | null = null;
const sync = (connected: boolean) => {
if (connected && !off) {
let session = useRemote.getState().session;
const sync = () => {
const next = useRemote.getState().session;
const connected = next?.target.kind === "ssh" || next?.target.kind === "devContainer";
if (connected && (next !== session || !off)) {
usePorts.setState({ ports: [] });
void restorePorts();
}
session = next;
if (connected && !off) {
off = registerToolWindow({
id: "remote.ports",
title: "Ports",
@@ -36,8 +43,8 @@ function trackPortsWindow(): () => void {
usePorts.setState({ ports: [] });
}
};
sync(useRemote.getState().session?.target.kind === "ssh");
const unsubscribe = useRemote.subscribe((s) => sync(s.session?.target.kind === "ssh"));
sync();
const unsubscribe = useRemote.subscribe(sync);
return () => {
unsubscribe();
off?.();
+99 -8
Переглянути файл
@@ -34,6 +34,7 @@ ipc.remoteStopForwarding.mockImplementation(async (remotePort: number) => {
afterEach(() => {
cleanup();
forwarded.length = 0;
localStorage.clear();
usePorts.setState({ ports: [] });
});
@@ -87,19 +88,109 @@ describe("restoring forwarded ports", () => {
});
});
it("doesn't forward ports of a dev container yet", async () => {
it("forwards and remembers ports of a dev container, per folder and configuration", async () => {
const { useRemote } = await import("@/platform/remote");
const { savedPorts } = await import("./ports");
const container = (config: string) => ({
target: { kind: "devContainer" as const, folder: "/w/app", config },
root: "/workspaces/app",
host: "app",
os: "linux",
commands: [],
});
ipc.remoteForwardPort.mockClear();
useRemote.setState({ session: container("") });
expect(await forwardPort(8000)).toEqual({ remotePort: 8000, localPort: 8000 });
expect(ipc.remoteForwardPort).toHaveBeenCalledWith(8000);
expect(savedPorts()).toEqual([8000]);
useRemote.setState({ session: { ...container(""), root: "/another-workspace" } });
expect(savedPorts()).toEqual([8000]);
useRemote.setState({ session: container(".devcontainer/other/devcontainer.json") });
expect(savedPorts()).toEqual([]);
useRemote.setState({ session: null });
});
it("doesn't forward ports of a WSL project yet", async () => {
const { useRemote } = await import("@/platform/remote");
ipc.remoteForwardPort.mockClear();
useRemote.setState({
session: {
target: { kind: "devContainer", folder: "/w/app", config: "" },
root: "/workspaces/app",
host: "app",
os: "linux",
commands: [],
},
session: { target: { kind: "wsl", distro: "u" }, root: "/home/u", host: "u", os: "linux", commands: [] },
});
expect(await forwardPort(8080)).toBeNull();
expect(ipc.remoteForwardPort).not.toHaveBeenCalled();
useRemote.setState({ session: null });
});
it("restores manually forwarded container ports alongside configured ports", async () => {
const { useRemote } = await import("@/platform/remote");
const { restorePorts, savedPorts } = await import("./ports");
const session = {
target: { kind: "devContainer" as const, folder: "/w/app", config: "" },
root: "/workspaces/app",
host: "app",
os: "linux",
commands: [],
};
useRemote.setState({ session });
await forwardPort(9000);
expect(savedPorts()).toEqual([9000]);
forwarded.splice(0, forwarded.length, { remotePort: 8000, localPort: 8000 });
await restorePorts();
expect(usePorts.getState().ports).toEqual([
{ remotePort: 8000, localPort: 8000 },
{ remotePort: 9000, localPort: 9000 },
]);
useRemote.setState({ session: { ...session, target: { ...session.target, folder: "/w/other" } } });
expect(savedPorts()).toEqual([]);
useRemote.setState({ session: null });
});
it("does not display a previous session's delayed port response", async () => {
const { useRemote } = await import("@/platform/remote");
const { refreshPorts } = await import("./ports");
let resolve!: (ports: typeof forwarded) => void;
ipc.remoteForwardedPorts.mockImplementationOnce(
() =>
new Promise((done) => {
resolve = done;
}),
);
const pending = refreshPorts(true);
useRemote.setState({
session: {
target: { kind: "devContainer", folder: "/other", config: "" },
root: "/workspaces/other",
host: "other",
os: "linux",
commands: [],
},
});
resolve([{ remotePort: 8000, localPort: 8000 }]);
await pending;
expect(usePorts.getState().ports).toEqual([]);
useRemote.setState({ session: null });
});
it("reloads ports when switching directly between connected sessions", async () => {
const { useRemote } = await import("@/platform/remote");
const { activate } = await import("./index");
const session = (folder: string) => ({
target: { kind: "devContainer" as const, folder, config: "" },
root: "/workspaces/app",
host: "app",
os: "linux",
commands: [],
});
useRemote.setState({ session: session("/first") });
forwarded.push({ remotePort: 8000, localPort: 8000 });
const dispose = activate();
try {
await waitFor(() => expect(usePorts.getState().ports).toEqual([{ remotePort: 8000, localPort: 8000 }]));
forwarded.splice(0, forwarded.length, { remotePort: 9000, localPort: 9000 });
useRemote.setState({ session: session("/second") });
await waitFor(() => expect(usePorts.getState().ports).toEqual([{ remotePort: 9000, localPort: 9000 }]));
} finally {
dispose();
useRemote.setState({ session: null });
}
});
+23 -10
Переглянути файл
@@ -1,4 +1,7 @@
/** Remote ports forwarded to this machine, remembered per host and folder and forwarded again on reconnect. */
/**
* Remote ports forwarded to this machine, remembered per host (or dev container) and folder and
* forwarded again on reconnect. A dev container also forwards its `forwardPorts` on its own.
*/
import { create } from "zustand";
import type { ForwardedPort } from "@/bindings";
import { tx } from "@/platform/i18n";
@@ -10,7 +13,14 @@ const SAVED_KEY = "openidea.remote.ports:";
function savedKey(): string | null {
const session = useRemote.getState().session;
return session?.target.kind === "ssh" ? `${SAVED_KEY}${session.target.config}:${session.root}` : null;
switch (session?.target.kind) {
case "ssh":
return `${SAVED_KEY}${session.target.config}:${session.root}`;
case "devContainer":
return `${SAVED_KEY}devcontainer:${JSON.stringify([session.target.folder, session.target.config])}`;
default:
return null;
}
}
export function savedPorts(): number[] {
@@ -39,7 +49,9 @@ export const usePorts = create<{ ports: ForwardedPort[] }>(() => ({ ports: [] })
/** Reads the forwarded ports; `remember` keeps them for the next connection (after the user changed them). */
export async function refreshPorts(remember = false): Promise<void> {
const session = useRemote.getState().session;
const ports = await commands.remoteForwardedPorts().catch(() => []);
if (useRemote.getState().session !== session) return;
usePorts.setState({ ports });
if (remember) save(ports);
}
@@ -53,15 +65,11 @@ export function parsePort(text: string): number | null {
}
export async function forwardPort(port: number): Promise<ForwardedPort | null> {
const kind = useRemote.getState().session?.target.kind;
if (kind === "wsl" || kind === "devContainer") {
if (useRemote.getState().session?.target.kind === "wsl") {
notifications().notify({
group: "remote",
level: "error",
title:
kind === "wsl"
? tx("Forwarding ports is not supported in a WSL project yet")
: tx("Forwarding ports is not supported in a dev container yet"),
title: tx("Forwarding ports is not supported in a WSL project yet"),
});
return null;
}
@@ -87,7 +95,12 @@ export async function stopForwarding(port: number): Promise<void> {
/** After (re)connecting: forwards the ports this host and folder had forwarded before. */
export async function restorePorts(): Promise<void> {
const session = useRemote.getState().session;
const saved = savedPorts();
for (const port of saved) await call(commands.remoteForwardPort(port)).catch(() => null);
usePorts.setState({ ports: await commands.remoteForwardedPorts().catch(() => []) });
for (const port of saved) {
if (useRemote.getState().session !== session) return;
await call(commands.remoteForwardPort(port)).catch(() => null);
}
const ports = await commands.remoteForwardedPorts().catch(() => []);
if (useRemote.getState().session === session) usePorts.setState({ ports });
}
-1
Переглянути файл
@@ -322,7 +322,6 @@
"Read-only: the table schema could not be identified in the result database.": "Read-only: the table schema could not be identified in the result database.",
"Starting the dev container…": "Starting the dev container…",
"Command-line tool": "Command-line tool",
"Forwarding ports is not supported in a dev container yet": "Forwarding ports is not supported in a dev container yet",
"Could not reopen the project in a container": "Could not reopen the project in a container",
"Reopen in Container": "Reopen in Container",
"Builds or starts the project's dev container (devcontainer.json) with the Dev Containers CLI and opens the project in it.": "Builds or starts the project's dev container (devcontainer.json) with the Dev Containers CLI and opens the project in it.",
-1
Переглянути файл
@@ -6686,7 +6686,6 @@
"Read-only: the result has no primary key of {tableLabel}.": "只读:结果中没有 {tableLabel} 的主键。",
"Starting the dev container…": "正在启动开发容器…",
"Command-line tool": "命令行工具",
"Forwarding ports is not supported in a dev container yet": "开发容器暂不支持端口转发",
"Could not reopen the project in a container": "无法在容器中重新打开项目",
"Reopen in Container": "在容器中重新打开",
"Builds or starts the project's dev container (devcontainer.json) with the Dev Containers CLI and opens the project in it.": "使用 Dev Containers CLI 构建或启动项目的开发容器(devcontainer.json),并在其中打开项目。",