Skip to content

fix(agent): harden package broker pipe against connection floods #1902

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Vladyslav Nikonov (vnikonov-devolutions) merged 2 commits into from
Aug 3, 2026
Merged

fix(agent): harden package broker pipe against connection floods #1902

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b5da1a0
fix(agent): bound unauthenticated work in package broker pipe accept …
pacmancoder Aug 3, 2026
e119538
fix(agent): enforce deadline on package broker pipe connections
pacmancoder Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 31 additions & 33 deletions devolutions-agent/src/broker/auth.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,50 +20,40 @@ use crate::code_signing::validate_devolutions_authenticode_signature;
pub(crate) struct PipeClient {
process_id: u32,
executable_path: PathBuf,
user: ClientUser,
}

#[derive(Clone, Debug)]
struct ClientUser {
/// Security identifier of the pipe client process token user, captured at connect.
sid: Sid,
domain: String,
name: String,
user_sid: Sid,
}

impl PipeClient {
/// Captures the identity of the process on the other end of a connected pipe instance.
///
/// Deliberately limited to fast, local syscalls (no account-name resolution, which may
/// hit a domain controller), because it runs before any signature gate and is therefore
/// unauthenticated work a connection flood can trigger.
pub(crate) fn from_connected_pipe(server: &NamedPipeServer) -> anyhow::Result<Self> {
let process_id = connected_pipe_client_process_id(server).context("failed to query pipe client process id")?;
let process = Process::get_by_pid(process_id, PROCESS_QUERY_LIMITED_INFORMATION)
.with_context(|| format!("failed to open pipe client process {process_id}"))?;
let executable_path = process
.exe_path()
.with_context(|| format!("failed to query pipe client process {process_id} executable path"))?;
let sid = process
let user_sid = process
.token(TOKEN_QUERY)
.with_context(|| format!("failed to open pipe client process {process_id} token"))?
.sid_and_attributes()
.with_context(|| format!("failed to query pipe client process {process_id} token user"))?
.sid;
let account = sid
.lookup_account(None)
.with_context(|| format!("failed to resolve pipe client process {process_id} user"))?;
let user = ClientUser {
sid,
domain: account.domain_name.to_string_lossy(),
name: account.name.to_string_lossy(),
};

Ok(Self {
process_id,
executable_path,
user,
user_sid,
})
}

/// Security identifier of the authenticated pipe client user, captured at connect.
pub(crate) fn user_sid(&self) -> &Sid {
&self.user.sid
&self.user_sid
}

pub(crate) fn validate_request(
Expand Down Expand Up @@ -116,15 +106,28 @@ impl PipeClient {
let requested_sid = resolve_account_sid(effective_user)
.with_context(|| format!("failed to resolve request effective_user '{effective_user}'"))?;

if requested_sid == self.user.sid {
if requested_sid == self.user_sid {
return Ok(());
}

// Resolve the client account name lazily, only on this rare error path; doing it at
// connect time would be unauthenticated work triggerable by a connection flood.
let client_account = self
.user_sid
.lookup_account(None)
.map(|account| {
format!(
"{}\\{}",
account.domain_name.to_string_lossy(),
account.name.to_string_lossy()
)
})
.unwrap_or_else(|_| String::from("<unresolved>"));

bail!(
"pipe client user '{}\\{}' ({}) does not match request effective_user '{}' ({})",
self.user.domain,
self.user.name,
self.user.sid,
"pipe client user '{}' ({}) does not match request effective_user '{}' ({})",
client_account,
self.user_sid,
effective_user,
requested_sid
)
Expand Down Expand Up @@ -256,21 +259,16 @@ mod tests {
}

fn system_client() -> PipeClient {
let (domain, name) = system_account_names();
PipeClient {
process_id: 0,
executable_path: PathBuf::new(),
user: ClientUser {
sid: system_sid(),
domain,
name,
},
user_sid: system_sid(),
}
}

#[cfg(not(feature = "dev-skip-broker-signature"))]
fn client_user() -> ClientUser {
system_client().user
fn client_user_sid() -> Sid {
system_client().user_sid
}

#[test]
Expand Down Expand Up @@ -355,7 +353,7 @@ mod tests {
let client = PipeClient {
process_id: std::process::id(),
executable_path: std::env::current_exe().expect("current test executable path"),
user: client_user(),
user_sid: client_user_sid(),
};

assert!(client.validate_signature(true).is_err());
Expand Down
68 changes: 57 additions & 11 deletions devolutions-agent/src/broker/pipe.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ mod windows_pipe {

use anyhow::Context as _;
use tokio::net::windows::named_pipe::{NamedPipeServer, ServerOptions};
use tokio::sync::Semaphore;
use tokio_util::sync::CancellationToken;
use tracing::{error, info, warn};
use win_api_wrappers::identity::sid::Sid;
Expand All @@ -25,13 +26,44 @@ mod windows_pipe {
/// Default pipe name for the package broker.
pub const DEFAULT_PIPE_NAME: &str = r"\\.\pipe\Devolutions.Now.PackageBroker.v1";

/// Maximum number of concurrently served pipe connections.
///
/// Connection setup performs unauthenticated work (client process identity lookups)
/// before any signature gate, so a connection flood could otherwise trigger unbounded
/// work and task spawning. While all slots are taken, no pipe instance is listening and
/// further clients fail to connect until a slot frees up.
const MAX_CONCURRENT_CONNECTIONS: usize = 16;

/// Deadline for serving a single pipe connection, from accept to response completion.
///
/// Each connection serves exactly one HTTP request (`keep_alive` is disabled) and all
/// endpoints respond without blocking on package operations (execution is asynchronous,
/// tracked via the operation tracker), so a healthy exchange completes well within this
/// deadline. Without it, idle clients holding their connection open without sending a
/// request would each pin a connection slot indefinitely and could exhaust the pool.
const CONNECTION_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30);

/// Start the named pipe server and accept connections until shutdown.
pub async fn run_pipe_server(state: Arc<BrokerState>, shutdown: CancellationToken) -> anyhow::Result<()> {
let pipe_name = state.pipe_name.clone();
info!(%pipe_name, "Starting named pipe server");

let connection_permits = Arc::new(Semaphore::new(MAX_CONCURRENT_CONNECTIONS));

let mut first_instance = true;
loop {
// Wait for a free connection slot before exposing a new pipe instance,
// bounding the number of concurrently served connections.
let permit = tokio::select! {
permit = Arc::clone(&connection_permits).acquire_owned() => {
permit.expect("the semaphore is never closed")
}
_ = shutdown.cancelled() => {
info!("Pipe server shutting down");
return Ok(());
}
};

// Create a new pipe instance for each connection.
let server = create_pipe_instance(&pipe_name, first_instance)?;
first_instance = false;
Expand All @@ -40,18 +72,32 @@ mod windows_pipe {
result = server.connect() => {
match result {
Ok(()) => {
let client = match PipeClient::from_connected_pipe(&server) {
Ok(client) => client,
Err(error) => {
warn!(%error, "Rejected named pipe client");
continue;
}
};
info!("Client connected to named pipe");
let router = build_router_for_client(Arc::clone(&state), client);
let state = Arc::clone(&state);
tokio::spawn(async move {
serve_connection(server, router).await;
info!("Client disconnected from named pipe");
// The permit is held for the lifetime of the connection task.
let _permit = permit;
Comment thread
vnikonov-devolutions marked this conversation as resolved.

let serve = async move {
// Capture the client identity off the accept loop so a slow
// lookup cannot stall accepting other connections.
let client = match PipeClient::from_connected_pipe(&server) {
Ok(client) => client,
Err(error) => {
warn!(%error, "Rejected named pipe client");
return;
}
};
info!("Client connected to named pipe");
let router = build_router_for_client(state, client);
serve_connection(server, router).await;
info!("Client disconnected from named pipe");
};

// Enforce a deadline so idle or slow clients cannot pin
// a connection slot indefinitely.
if tokio::time::timeout(CONNECTION_DEADLINE, serve).await.is_err() {
warn!("Closed named pipe connection: deadline exceeded");
}
});
}
Err(error) => {
Expand Down
Loading