From 97f9898fca288edaeb355cb11a88953bef61b94d Mon Sep 17 00:00:00 2001 From: Adam Driscoll Date: Wed, 8 Jul 2026 12:49:13 -0500 Subject: [PATCH 1/5] feat(agent): support PSU gRPC app token secrets Resolve configured PSU gRPC AppToken values before opening the agent stream so literal tokens are passed through and secret references reuse the existing PowerShell secret resolver. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- devolutions-agent/src/psu_event_hub/mod.rs | 2 +- .../src/psu_event_hub/powershell_worker.rs | 16 ++--- devolutions-agent/src/psu_grpc_agent/mod.rs | 63 +++++++++++++++++-- 3 files changed, 68 insertions(+), 13 deletions(-) diff --git a/devolutions-agent/src/psu_event_hub/mod.rs b/devolutions-agent/src/psu_event_hub/mod.rs index 1a2416209..7722bc8ac 100644 --- a/devolutions-agent/src/psu_event_hub/mod.rs +++ b/devolutions-agent/src/psu_event_hub/mod.rs @@ -1,7 +1,7 @@ pub(crate) mod compat; mod executor; mod models; -mod powershell_worker; +pub(crate) mod powershell_worker; mod result_store; mod signalr; diff --git a/devolutions-agent/src/psu_event_hub/powershell_worker.rs b/devolutions-agent/src/psu_event_hub/powershell_worker.rs index d036c5d9b..4dcdfa89f 100644 --- a/devolutions-agent/src/psu_event_hub/powershell_worker.rs +++ b/devolutions-agent/src/psu_event_hub/powershell_worker.rs @@ -169,7 +169,7 @@ $response | ConvertTo-Json -Compress -Depth 16 const POWERSHELL_EXECUTION_TIMEOUT: Duration = Duration::from_secs(30 * 60); #[derive(Debug, Clone)] -pub(super) struct PowerShellWorker { +pub(crate) struct PowerShellWorker { conf: PsuPowerShellConf, permits: Arc, worker_script: Arc, @@ -177,7 +177,7 @@ pub(super) struct PowerShellWorker { } impl PowerShellWorker { - pub(super) fn new(conf: PsuPowerShellConf) -> anyhow::Result { + pub(crate) fn new(conf: PsuPowerShellConf) -> anyhow::Result { Self::with_execution_timeout(conf, POWERSHELL_EXECUTION_TIMEOUT) } @@ -191,8 +191,8 @@ impl PowerShellWorker { }) } - pub(super) async fn resolve_app_token(&self, app_token: &str) -> anyhow::Result { - let Some(secret_name) = secret_reference_name(app_token) else { + pub(crate) async fn resolve_app_token(&self, app_token: &str) -> anyhow::Result { + let Some(secret_name) = app_token_secret_reference_name(app_token) else { return Ok(app_token.to_owned()); }; @@ -399,7 +399,7 @@ impl WorkerRequest { } } -fn secret_reference_name(app_token: &str) -> Option<&str> { +pub(crate) fn app_token_secret_reference_name(app_token: &str) -> Option<&str> { let prefix = "$secret:"; app_token .get(..prefix.len()) @@ -563,9 +563,9 @@ mod tests { #[test] fn secret_reference_name_is_case_insensitive() { - assert_eq!(secret_reference_name("$secret:AppToken"), Some("AppToken")); - assert_eq!(secret_reference_name("$SECRET:AppToken"), Some("AppToken")); - assert_eq!(secret_reference_name("literal-token"), None); + assert_eq!(app_token_secret_reference_name("$secret:AppToken"), Some("AppToken")); + assert_eq!(app_token_secret_reference_name("$SECRET:AppToken"), Some("AppToken")); + assert_eq!(app_token_secret_reference_name("literal-token"), None); } #[test] diff --git a/devolutions-agent/src/psu_grpc_agent/mod.rs b/devolutions-agent/src/psu_grpc_agent/mod.rs index aa1819355..582376dfa 100644 --- a/devolutions-agent/src/psu_grpc_agent/mod.rs +++ b/devolutions-agent/src/psu_grpc_agent/mod.rs @@ -16,6 +16,7 @@ use tonic::transport::Endpoint; use uuid::Uuid; use crate::config::{ConfHandle, dto}; +use crate::psu_event_hub::powershell_worker::{PowerShellWorker, app_token_secret_reference_name}; use crate::psu_grpc_agent::process::{ProcessControl, ProcessRegistry}; #[allow(unused_qualifications, clippy::clone_on_ref_ptr, clippy::similar_names)] @@ -140,6 +141,7 @@ impl PsuGrpcAgent { } async fn run_single_connection(&self, shutdown_signal: &mut ShutdownSignal) -> anyhow::Result<()> { + let app_token = self.resolve_app_token().await?; let endpoint = Endpoint::from_shared(self.server_url.clone())?; let channel = endpoint .connect() @@ -155,10 +157,7 @@ impl PsuGrpcAgent { .context("failed to queue PSU gRPC agent registration")?; let mut response_stream = client - .connect(connect_request( - ReceiverStream::new(outgoing_rx), - self.app_token.as_deref(), - )?) + .connect(connect_request(ReceiverStream::new(outgoing_rx), app_token.as_deref())?) .await .context("failed to start PSU gRPC agent stream")? .into_inner(); @@ -259,6 +258,25 @@ impl PsuGrpcAgent { Ok(()) } + async fn resolve_app_token(&self) -> anyhow::Result> { + let Some(app_token) = self.app_token.as_deref() else { + return Ok(None); + }; + + if app_token_secret_reference_name(app_token).is_none() { + return Ok(Some(app_token.to_owned())); + } + + let worker = PowerShellWorker::new(self.conf.powershell.clone()) + .context("failed to initialize PSU PowerShell worker for gRPC AppToken secret resolution")?; + + worker + .resolve_app_token(app_token) + .await + .map(Some) + .context("failed to resolve PSU gRPC AppToken secret") + } + fn create_registration_message(&self, powershell_version: String) -> AgentMessage { AgentMessage { request_id: Uuid::new_v4().simple().to_string(), @@ -437,4 +455,41 @@ mod tests { "Bearer token" ); } + + #[tokio::test] + async fn literal_app_token_does_not_require_secret_resolution() { + let agent = PsuGrpcAgent::new(dto::PsuGrpcAgentConf { + enabled: true, + server_url: Some("http://localhost:5000".parse().expect("server URL")), + agent_id: Some("agent-01".to_owned()), + display_name: None, + app_token: Some("literal-token".to_owned()), + hubs: Vec::new(), + powershell: dto::PsuPowerShellConf { + executable_path: Some("missing-pwsh".into()), + ..dto::PsuPowerShellConf::default() + }, + }) + .expect("create agent"); + + let app_token = agent.resolve_app_token().await.expect("resolve AppToken"); + + assert_eq!(app_token.as_deref(), Some("literal-token")); + } + + #[test] + fn empty_app_token_is_ignored() { + let agent = PsuGrpcAgent::new(dto::PsuGrpcAgentConf { + enabled: true, + server_url: Some("http://localhost:5000".parse().expect("server URL")), + agent_id: Some("agent-01".to_owned()), + display_name: None, + app_token: Some(" ".to_owned()), + hubs: Vec::new(), + powershell: dto::PsuPowerShellConf::default(), + }) + .expect("create agent"); + + assert!(agent.app_token.is_none()); + } } From 341fd1d2dd59ac812172050ceed1c92a03f4828b Mon Sep 17 00:00:00 2001 From: Adam Driscoll Date: Thu, 9 Jul 2026 17:50:55 -0500 Subject: [PATCH 2/5] Address issue with secret resolving and update C# namespace. --- devolutions-agent/proto/psu_agent.proto | 2 ++ devolutions-agent/src/psu_grpc_agent/mod.rs | 15 +++++++++++---- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/devolutions-agent/proto/psu_agent.proto b/devolutions-agent/proto/psu_agent.proto index 77e23bd51..1f19f0a7b 100644 --- a/devolutions-agent/proto/psu_agent.proto +++ b/devolutions-agent/proto/psu_agent.proto @@ -2,6 +2,8 @@ syntax = "proto3"; package devolutions.psu.agent.poc.v1; +option csharp_namespace = "Devolutions.Psu.Agent.Protocol"; + import "google/protobuf/timestamp.proto"; service AgentControl { diff --git a/devolutions-agent/src/psu_grpc_agent/mod.rs b/devolutions-agent/src/psu_grpc_agent/mod.rs index 582376dfa..b51b0b249 100644 --- a/devolutions-agent/src/psu_grpc_agent/mod.rs +++ b/devolutions-agent/src/psu_grpc_agent/mod.rs @@ -107,11 +107,15 @@ impl PsuGrpcAgent { .with_multiplier(RETRY_MULTIPLIER) .with_max_elapsed_time(None) .build(); + let app_token = self.resolve_app_token().await?; loop { let start = Instant::now(); - match self.run_single_connection(&mut shutdown_signal).await { + match self + .run_single_connection(&mut shutdown_signal, app_token.as_deref()) + .await + { Ok(()) => return Ok(()), Err(error) => { warn!(url = %self.server_url, error = format!("{error:#}"), "PSU gRPC agent connection failed") @@ -140,8 +144,11 @@ impl PsuGrpcAgent { } } - async fn run_single_connection(&self, shutdown_signal: &mut ShutdownSignal) -> anyhow::Result<()> { - let app_token = self.resolve_app_token().await?; + async fn run_single_connection( + &self, + shutdown_signal: &mut ShutdownSignal, + app_token: Option<&str>, + ) -> anyhow::Result<()> { let endpoint = Endpoint::from_shared(self.server_url.clone())?; let channel = endpoint .connect() @@ -157,7 +164,7 @@ impl PsuGrpcAgent { .context("failed to queue PSU gRPC agent registration")?; let mut response_stream = client - .connect(connect_request(ReceiverStream::new(outgoing_rx), app_token.as_deref())?) + .connect(connect_request(ReceiverStream::new(outgoing_rx), app_token)?) .await .context("failed to start PSU gRPC agent stream")? .into_inner(); From ed2673d3401b0d173a4697f6a1d5015b5a8b4ea4 Mon Sep 17 00:00:00 2001 From: Adam Driscoll Date: Thu, 9 Jul 2026 17:58:55 -0500 Subject: [PATCH 3/5] Lift psu_powershell out of SignalR transport so it can be used across that and gRPC. --- devolutions-agent/src/lib.rs | 1 + .../src/psu_event_hub/executor.rs | 9 +- devolutions-agent/src/psu_event_hub/mod.rs | 4 +- devolutions-agent/src/psu_event_hub/models.rs | 170 ------------------ .../src/psu_event_hub/result_store.rs | 26 +-- devolutions-agent/src/psu_grpc_agent/mod.rs | 2 +- ...powershell_worker.rs => psu_powershell.rs} | 169 +++++++++++++++-- 7 files changed, 179 insertions(+), 202 deletions(-) delete mode 100644 devolutions-agent/src/psu_event_hub/models.rs rename devolutions-agent/src/{psu_event_hub/powershell_worker.rs => psu_powershell.rs} (80%) diff --git a/devolutions-agent/src/lib.rs b/devolutions-agent/src/lib.rs index 5fb601919..b1647a7ab 100644 --- a/devolutions-agent/src/lib.rs +++ b/devolutions-agent/src/lib.rs @@ -11,6 +11,7 @@ pub mod enrollment; pub mod log; pub mod psu_event_hub; pub mod psu_grpc_agent; +pub(crate) mod psu_powershell; pub mod remote_desktop; pub mod tunnel; mod tunnel_helpers; diff --git a/devolutions-agent/src/psu_event_hub/executor.rs b/devolutions-agent/src/psu_event_hub/executor.rs index 20865ead5..16dc731cb 100644 --- a/devolutions-agent/src/psu_event_hub/executor.rs +++ b/devolutions-agent/src/psu_event_hub/executor.rs @@ -7,9 +7,8 @@ use tokio::task::JoinSet; use uuid::Uuid; use crate::config::dto::PsuEventHubConnectionConf; -use crate::psu_event_hub::models::WebsocketEventResponse; -use crate::psu_event_hub::powershell_worker::PowerShellWorker; use crate::psu_event_hub::result_store::ResultStore; +use crate::psu_powershell::{PowerShellWorker, PowerShellWorkerResponse}; #[derive(Debug, Clone)] pub(super) struct EventHubExecutor { @@ -92,7 +91,7 @@ impl EventHubExecutor { Err(error) if return_result => { result_store.insert( stored_execution_id, - WebsocketEventResponse::terminating_error(error.to_string()), + PowerShellWorkerResponse::terminating_error(error.to_string()), ); } Err(error) => warn!(error = format!("{error:#}"), "PSU command execution failed"), @@ -108,7 +107,7 @@ impl EventHubExecutor { if return_result { self.result_store.insert( execution_id.clone(), - WebsocketEventResponse::terminating_error("No script block found."), + PowerShellWorkerResponse::terminating_error("No script block found."), ); } return execution_id; @@ -125,7 +124,7 @@ impl EventHubExecutor { Err(error) if return_result => { result_store.insert( stored_execution_id, - WebsocketEventResponse::terminating_error(error.to_string()), + PowerShellWorkerResponse::terminating_error(error.to_string()), ); } Err(error) => warn!(error = format!("{error:#}"), "PSU script execution failed"), diff --git a/devolutions-agent/src/psu_event_hub/mod.rs b/devolutions-agent/src/psu_event_hub/mod.rs index 7722bc8ac..e151dc5a2 100644 --- a/devolutions-agent/src/psu_event_hub/mod.rs +++ b/devolutions-agent/src/psu_event_hub/mod.rs @@ -1,7 +1,5 @@ pub(crate) mod compat; mod executor; -mod models; -pub(crate) mod powershell_worker; mod result_store; mod signalr; @@ -14,7 +12,7 @@ use tokio::task::JoinSet; use crate::config::{ConfHandle, dto}; use crate::psu_event_hub::executor::EventHubExecutor; -use crate::psu_event_hub::powershell_worker::PowerShellWorker; +use crate::psu_powershell::PowerShellWorker; pub struct PsuEventHubTask { conf_handle: ConfHandle, diff --git a/devolutions-agent/src/psu_event_hub/models.rs b/devolutions-agent/src/psu_event_hub/models.rs deleted file mode 100644 index 9d2989bb4..000000000 --- a/devolutions-agent/src/psu_event_hub/models.rs +++ /dev/null @@ -1,170 +0,0 @@ -use std::fmt; - -use serde::{Deserialize, Deserializer, Serialize, Serializer}; - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub(super) struct WebsocketEventResponse { - #[serde(skip_serializing_if = "Option::is_none")] - pub data: Option, - #[serde(default)] - pub job_outputs: Vec, - #[serde(default)] - pub complete: bool, - #[serde(default)] - pub timeout: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub terminating_error: Option, -} - -impl WebsocketEventResponse { - pub(super) fn pending() -> Self { - Self { - data: None, - job_outputs: Vec::new(), - complete: false, - timeout: false, - terminating_error: None, - } - } - - pub(super) fn terminating_error(message: impl Into) -> Self { - Self { - data: None, - job_outputs: Vec::new(), - complete: true, - timeout: false, - terminating_error: Some(message.into()), - } - } - - pub(super) fn timeout(message: impl Into) -> Self { - Self { - data: None, - job_outputs: Vec::new(), - complete: true, - timeout: true, - terminating_error: Some(message.into()), - } - } -} - -impl Default for WebsocketEventResponse { - fn default() -> Self { - Self::pending() - } -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub(super) struct JobOutput { - #[serde(default)] - pub id: i64, - #[serde(skip_serializing_if = "Option::is_none")] - pub message: Option, - #[serde(rename = "type")] - pub output_type: JobOutputType, - #[serde(skip_serializing_if = "Option::is_none")] - pub data: Option, - #[serde(default)] - pub timestamp: String, - #[serde(default)] - pub job_id: i64, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -#[repr(u8)] -pub(super) enum JobOutputType { - Information = 0, - Verbose = 1, - Debug = 2, - Warning = 3, - Error = 4, - Progress = 5, -} - -impl JobOutputType { - fn as_u8(self) -> u8 { - self as u8 - } - - fn from_u8(value: u8) -> Option { - match value { - 0 => Some(Self::Information), - 1 => Some(Self::Verbose), - 2 => Some(Self::Debug), - 3 => Some(Self::Warning), - 4 => Some(Self::Error), - 5 => Some(Self::Progress), - _ => None, - } - } -} - -impl Serialize for JobOutputType { - fn serialize(&self, serializer: S) -> Result - where - S: Serializer, - { - serializer.serialize_u8(self.as_u8()) - } -} - -impl<'de> Deserialize<'de> for JobOutputType { - fn deserialize(deserializer: D) -> Result - where - D: Deserializer<'de>, - { - struct Visitor; - - impl serde::de::Visitor<'_> for Visitor { - type Value = JobOutputType; - - fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str("a PSU JobOutputType numeric value or name") - } - - fn visit_u64(self, value: u64) -> Result - where - E: serde::de::Error, - { - let value = u8::try_from(value).map_err(|_| E::custom("JobOutputType value is out of range"))?; - JobOutputType::from_u8(value).ok_or_else(|| E::custom("unknown JobOutputType value")) - } - - fn visit_str(self, value: &str) -> Result - where - E: serde::de::Error, - { - match value { - "Information" => Ok(JobOutputType::Information), - "Verbose" => Ok(JobOutputType::Verbose), - "Debug" => Ok(JobOutputType::Debug), - "Warning" => Ok(JobOutputType::Warning), - "Error" => Ok(JobOutputType::Error), - "Progress" => Ok(JobOutputType::Progress), - _ => Err(E::custom("unknown JobOutputType name")), - } - } - } - - deserializer.deserialize_any(Visitor) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn job_output_type_serializes_as_psu_numeric_value() { - let json = serde_json::to_string(&JobOutputType::Error).expect("serialize output type"); - assert_eq!(json, "4"); - } - - #[test] - fn job_output_type_accepts_worker_names() { - let output_type: JobOutputType = serde_json::from_str("\"Warning\"").expect("deserialize output type"); - assert_eq!(output_type, JobOutputType::Warning); - } -} diff --git a/devolutions-agent/src/psu_event_hub/result_store.rs b/devolutions-agent/src/psu_event_hub/result_store.rs index d567a3b80..d68a9831a 100644 --- a/devolutions-agent/src/psu_event_hub/result_store.rs +++ b/devolutions-agent/src/psu_event_hub/result_store.rs @@ -4,7 +4,7 @@ use std::time::{Duration, Instant}; use parking_lot::Mutex; -use crate::psu_event_hub::models::WebsocketEventResponse; +use crate::psu_powershell::PowerShellWorkerResponse; const RESULT_TTL: Duration = Duration::from_secs(15 * 60); const MAX_RESULTS: usize = 1024; @@ -25,7 +25,7 @@ impl ResultStore { } } - pub(super) fn insert(&self, execution_id: String, response: WebsocketEventResponse) { + pub(super) fn insert(&self, execution_id: String, response: PowerShellWorkerResponse) { let mut inner = self.inner.lock(); let now = Instant::now(); @@ -41,14 +41,14 @@ impl ResultStore { inner.enforce_limit(self.max_results); } - pub(super) fn take(&self, execution_id: &str) -> WebsocketEventResponse { + pub(super) fn take(&self, execution_id: &str) -> PowerShellWorkerResponse { let mut inner = self.inner.lock(); inner.remove_expired(Instant::now(), self.ttl); inner .results .remove(execution_id) .map(|stored| stored.response) - .unwrap_or_else(WebsocketEventResponse::pending) + .unwrap_or_else(PowerShellWorkerResponse::pending) } } @@ -95,7 +95,7 @@ impl ResultStoreInner { #[derive(Debug)] struct StoredResult { inserted_at: Instant, - response: WebsocketEventResponse, + response: PowerShellWorkerResponse, } #[cfg(test)] @@ -113,9 +113,9 @@ mod tests { let store = ResultStore::default(); store.insert( "execution-id".to_owned(), - WebsocketEventResponse { + PowerShellWorkerResponse { complete: true, - ..WebsocketEventResponse::default() + ..PowerShellWorkerResponse::default() }, ); @@ -128,16 +128,16 @@ mod tests { let store = ResultStore::test_with_limits(Duration::from_secs(60), 1); store.insert( "first".to_owned(), - WebsocketEventResponse { + PowerShellWorkerResponse { complete: true, - ..WebsocketEventResponse::default() + ..PowerShellWorkerResponse::default() }, ); store.insert( "second".to_owned(), - WebsocketEventResponse { + PowerShellWorkerResponse { complete: true, - ..WebsocketEventResponse::default() + ..PowerShellWorkerResponse::default() }, ); @@ -150,9 +150,9 @@ mod tests { let store = ResultStore::test_with_limits(Duration::ZERO, 10); store.insert( "execution-id".to_owned(), - WebsocketEventResponse { + PowerShellWorkerResponse { complete: true, - ..WebsocketEventResponse::default() + ..PowerShellWorkerResponse::default() }, ); diff --git a/devolutions-agent/src/psu_grpc_agent/mod.rs b/devolutions-agent/src/psu_grpc_agent/mod.rs index b51b0b249..851e8385c 100644 --- a/devolutions-agent/src/psu_grpc_agent/mod.rs +++ b/devolutions-agent/src/psu_grpc_agent/mod.rs @@ -16,8 +16,8 @@ use tonic::transport::Endpoint; use uuid::Uuid; use crate::config::{ConfHandle, dto}; -use crate::psu_event_hub::powershell_worker::{PowerShellWorker, app_token_secret_reference_name}; use crate::psu_grpc_agent::process::{ProcessControl, ProcessRegistry}; +use crate::psu_powershell::{PowerShellWorker, app_token_secret_reference_name}; #[allow(unused_qualifications, clippy::clone_on_ref_ptr, clippy::similar_names)] pub mod protocol { diff --git a/devolutions-agent/src/psu_event_hub/powershell_worker.rs b/devolutions-agent/src/psu_powershell.rs similarity index 80% rename from devolutions-agent/src/psu_event_hub/powershell_worker.rs rename to devolutions-agent/src/psu_powershell.rs index 4dcdfa89f..22a40ab11 100644 --- a/devolutions-agent/src/psu_event_hub/powershell_worker.rs +++ b/devolutions-agent/src/psu_powershell.rs @@ -1,16 +1,166 @@ use std::ffi::OsString; +use std::fmt; use std::process::Stdio; use std::sync::Arc; use std::time::Duration; use anyhow::{Context as _, bail}; use camino::{Utf8Path, Utf8PathBuf}; -use serde::Serialize; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; use tokio::process::Command; use tokio::sync::Semaphore; use crate::config::dto::PsuPowerShellConf; -use crate::psu_event_hub::models::WebsocketEventResponse; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct PowerShellWorkerResponse { + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, + #[serde(default)] + pub job_outputs: Vec, + #[serde(default)] + pub complete: bool, + #[serde(default)] + pub timeout: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub terminating_error: Option, +} + +impl PowerShellWorkerResponse { + pub(crate) fn pending() -> Self { + Self { + data: None, + job_outputs: Vec::new(), + complete: false, + timeout: false, + terminating_error: None, + } + } + + pub(crate) fn terminating_error(message: impl Into) -> Self { + Self { + data: None, + job_outputs: Vec::new(), + complete: true, + timeout: false, + terminating_error: Some(message.into()), + } + } + + fn timeout(message: impl Into) -> Self { + Self { + data: None, + job_outputs: Vec::new(), + complete: true, + timeout: true, + terminating_error: Some(message.into()), + } + } +} + +impl Default for PowerShellWorkerResponse { + fn default() -> Self { + Self::pending() + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct JobOutput { + #[serde(default)] + pub id: i64, + #[serde(skip_serializing_if = "Option::is_none")] + pub message: Option, + #[serde(rename = "type")] + pub output_type: JobOutputType, + #[serde(skip_serializing_if = "Option::is_none")] + pub data: Option, + #[serde(default)] + pub timestamp: String, + #[serde(default)] + pub job_id: i64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] +pub(crate) enum JobOutputType { + Information = 0, + Verbose = 1, + Debug = 2, + Warning = 3, + Error = 4, + Progress = 5, +} + +impl JobOutputType { + fn as_u8(self) -> u8 { + self as u8 + } + + fn from_u8(value: u8) -> Option { + match value { + 0 => Some(Self::Information), + 1 => Some(Self::Verbose), + 2 => Some(Self::Debug), + 3 => Some(Self::Warning), + 4 => Some(Self::Error), + 5 => Some(Self::Progress), + _ => None, + } + } +} + +impl Serialize for JobOutputType { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_u8(self.as_u8()) + } +} + +impl<'de> Deserialize<'de> for JobOutputType { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + struct Visitor; + + impl serde::de::Visitor<'_> for Visitor { + type Value = JobOutputType; + + fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("a PSU JobOutputType numeric value or name") + } + + fn visit_u64(self, value: u64) -> Result + where + E: serde::de::Error, + { + let value = u8::try_from(value).map_err(|_| E::custom("JobOutputType value is out of range"))?; + JobOutputType::from_u8(value).ok_or_else(|| E::custom("unknown JobOutputType value")) + } + + fn visit_str(self, value: &str) -> Result + where + E: serde::de::Error, + { + match value { + "Information" => Ok(JobOutputType::Information), + "Verbose" => Ok(JobOutputType::Verbose), + "Debug" => Ok(JobOutputType::Debug), + "Warning" => Ok(JobOutputType::Warning), + "Error" => Ok(JobOutputType::Error), + "Progress" => Ok(JobOutputType::Progress), + _ => Err(E::custom("unknown JobOutputType name")), + } + } + } + + deserializer.deserialize_any(Visitor) + } +} const WORKER_SCRIPT: &str = r#" param([string] $RequestPath) @@ -207,27 +357,27 @@ impl PowerShellWorker { .with_context(|| format!("PSU AppToken secret {secret_name} resolved to an empty value")) } - pub(super) async fn execute_command( + pub(crate) async fn execute_command( &self, command: String, data: String, return_result: bool, - ) -> anyhow::Result { + ) -> anyhow::Result { self.run_request(WorkerRequest::command(command, data, return_result)) .await } - pub(super) async fn execute_script( + pub(crate) async fn execute_script( &self, script_path: Utf8PathBuf, data: String, return_result: bool, - ) -> anyhow::Result { + ) -> anyhow::Result { self.run_request(WorkerRequest::script(script_path, data, return_result)) .await } - async fn run_request(&self, request: WorkerRequest) -> anyhow::Result { + async fn run_request(&self, request: WorkerRequest) -> anyhow::Result { let _permit = self .permits .acquire() @@ -242,7 +392,7 @@ impl PowerShellWorker { &self, script_path: &Utf8Path, request_path: &Utf8Path, - ) -> anyhow::Result { + ) -> anyhow::Result { let executable = resolve_powershell_executable(&self.conf); let mut command = Command::new(&executable); command @@ -275,7 +425,7 @@ impl PowerShellWorker { timeout_secs = self.execution_timeout.as_secs(), "PowerShell worker timed out" ); - return Ok(WebsocketEventResponse::timeout("PowerShell worker timed out.")); + return Ok(PowerShellWorkerResponse::timeout("PowerShell worker timed out.")); } }; @@ -451,7 +601,6 @@ fn temp_dir() -> anyhow::Result { #[cfg(test)] mod tests { use super::*; - use crate::psu_event_hub::models::JobOutputType; const HASHTABLE_PS_VERSION_TABLE: &str = r#" From 6bdda93914fbe1da8d61049a29234290bd5d8876 Mon Sep 17 00:00:00 2001 From: Adam Driscoll Date: Thu, 9 Jul 2026 18:01:54 -0500 Subject: [PATCH 4/5] Add comment due to intent. --- devolutions-agent/src/psu_grpc_agent/mod.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/devolutions-agent/src/psu_grpc_agent/mod.rs b/devolutions-agent/src/psu_grpc_agent/mod.rs index 851e8385c..c712d8b5a 100644 --- a/devolutions-agent/src/psu_grpc_agent/mod.rs +++ b/devolutions-agent/src/psu_grpc_agent/mod.rs @@ -270,6 +270,7 @@ impl PsuGrpcAgent { return Ok(None); }; + // Avoid constructing a PowerShell worker unless the token is a secret reference. if app_token_secret_reference_name(app_token).is_none() { return Ok(Some(app_token.to_owned())); } From 061d38f71f58f173508e24e4aa695272fc78c373 Mon Sep 17 00:00:00 2001 From: Adam Driscoll Date: Thu, 9 Jul 2026 18:03:47 -0500 Subject: [PATCH 5/5] Update test name. --- devolutions-agent/src/psu_powershell.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/devolutions-agent/src/psu_powershell.rs b/devolutions-agent/src/psu_powershell.rs index 22a40ab11..dc06fab87 100644 --- a/devolutions-agent/src/psu_powershell.rs +++ b/devolutions-agent/src/psu_powershell.rs @@ -711,7 +711,7 @@ mod tests { } #[test] - fn secret_reference_name_is_case_insensitive() { + fn app_token_secret_reference_name_is_case_insensitive() { assert_eq!(app_token_secret_reference_name("$secret:AppToken"), Some("AppToken")); assert_eq!(app_token_secret_reference_name("$SECRET:AppToken"), Some("AppToken")); assert_eq!(app_token_secret_reference_name("literal-token"), None);