diff --git a/Cargo.lock b/Cargo.lock index db09cb7ba59..4fcc2c4cee6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12969,6 +12969,7 @@ dependencies = [ "model-entity", "model_notifications", "notification", + "reqwest", "secretsmanager_client", "serde", "serde_json", diff --git a/apps/web/src/features/block-automation/component/automationUtils.ts b/apps/web/src/features/block-automation/component/automationUtils.ts index 811fd83a2b1..d6cc8e5935d 100644 --- a/apps/web/src/features/block-automation/component/automationUtils.ts +++ b/apps/web/src/features/block-automation/component/automationUtils.ts @@ -255,9 +255,11 @@ export function createEmptyDraft(): ScheduleDraft { }; } -function getAgentTask(schedule: ScheduledAction): AgentTask { - // Backend stores task as a JSON object; for kind === "Agent" it is shaped - // like AgentTask. Cast through unknown to satisfy the open-ended type. +function getAgentTask(schedule: ScheduledAction): AgentTask | null { + // Backend stores task as a JSON object and `kind` is its discriminant: only + // kind === "Agent" is shaped like AgentTask. Other kinds (RemoteAgent) carry + // a different payload, so the cast is only sound behind this check. + if (schedule.kind !== 'Agent') return null; return schedule.task as unknown as AgentTask; } @@ -268,12 +270,12 @@ export function draftFromSchedule(schedule: ScheduledAction): ScheduleDraft { return { id: schedule.id ?? undefined, name: schedule.name, - prompt: task.user_prompt ?? '', + prompt: task?.user_prompt ?? '', frequency: parsed.frequency, time: parsed.time, daysOfWeek: parsed.daysOfWeek, dayOfMonth: parsed.dayOfMonth, - model: (task.model as Model) ?? undefined, + model: (task?.model as Model) ?? undefined, enabled: schedule.enabled, }; } @@ -304,9 +306,16 @@ export function draftToUpdateBody( return { name: draft.name.trim() || deriveScheduleName(draft.prompt), schedule: buildCron(draft), - kind: 'Agent', + // Preserve the stored kind and, for kinds this editor cannot build, the + // stored task: an update is a full overwrite, so writing an Agent-shaped + // task over a RemoteAgent action would drop its endpoint_url. Name, + // schedule and enabled still apply to every kind. + kind: previous.kind, timezone: previous.timezone || getDefaultTimezone(), - task: buildAgentTask(draft) as unknown as UpdateScheduledAction['task'], + task: + previous.kind === 'Agent' + ? (buildAgentTask(draft) as unknown as UpdateScheduledAction['task']) + : previous.task, enabled: draft.enabled, }; } diff --git a/services/scheduled_action/Cargo.toml b/services/scheduled_action/Cargo.toml index 1b871f2a942..24b4d0f9108 100644 --- a/services/scheduled_action/Cargo.toml +++ b/services/scheduled_action/Cargo.toml @@ -54,6 +54,7 @@ model = { path = "../../crates/model" } model-entity = { path = "../../crates/model-entity" } model_notifications = { path = "../../crates/model_notifications" } notification = { path = "../../crates/notification" } +reqwest = { workspace = true, features = ["stream"] } utoipa = { workspace = true, features = ["axum_extras", "chrono"] } utoipa-swagger-ui = { workspace = true } diff --git a/services/scheduled_action/src/bins/service.rs b/services/scheduled_action/src/bins/service.rs index be9fbe93018..11fb0195602 100644 --- a/services/scheduled_action/src/bins/service.rs +++ b/services/scheduled_action/src/bins/service.rs @@ -26,6 +26,7 @@ use scheduled_action::outbound::pg_polling_dispatcher::{ PgPollingDispatcher, PgPollingDispatcherLifecycle, }; use scheduled_action::outbound::pg_scheduled_action_repo::PgScheduledActionRepo; +use scheduled_action::outbound::remote_agent_http::ReqwestRemoteAgentClient; use scheduled_action::swagger::ApiDoc; use sqlx::postgres::PgPoolOptions; use tokio_util::{sync::CancellationToken, task::TaskTracker}; @@ -76,6 +77,10 @@ async fn main() -> Result<()> { let repo = Arc::new(PgScheduledActionRepo::new(db.clone())); + let remote_agent_client = Arc::new( + ReqwestRemoteAgentClient::new().context("failed to build remote agent http client")?, + ); + // The dispatcher consumes its executor, so build a second executor for the // service to use when handling execute-now requests. Both executors share // the underlying repo/pool/tool-context via cheap Arc/PgPool clones. @@ -85,6 +90,7 @@ async fn main() -> Result<()> { tool_context.clone(), Arc::clone(¬ification_ingress), Arc::clone(&live_updates), + Arc::clone(&remote_agent_client), ); let service_executor = Arc::new(InProcessExecutor::new( Arc::clone(&repo), @@ -92,6 +98,7 @@ async fn main() -> Result<()> { tool_context, notification_ingress, live_updates, + remote_agent_client, )); let dispatcher_cancellation_token = CancellationToken::new(); diff --git a/services/scheduled_action/src/domain/models.rs b/services/scheduled_action/src/domain/models.rs index 22851d829e0..3d2560dd8bd 100644 --- a/services/scheduled_action/src/domain/models.rs +++ b/services/scheduled_action/src/domain/models.rs @@ -51,7 +51,11 @@ impl<'de> Deserialize<'de> for Schedule { #[derive(Debug, Serialize, Deserialize, Clone, ToSchema)] pub enum ActionKind { + /// Executed by Macro's own agent loop against a Macro model. Agent, + /// Executed by an agent running outside Macro, reached over HTTPS. Macro + /// remains the scheduler, the system of record and the UI. + RemoteAgent, } #[derive(Serialize, Deserialize, Debug, Clone, ToSchema)] @@ -61,6 +65,56 @@ pub struct AgentTask { pub user_prompt: String, } +/// Label attributed to run messages when a [`RemoteAgentTask`] does not name +/// the agent. Stored on the chat message in place of a Macro model name. +pub const DEFAULT_REMOTE_AGENT_LABEL: &str = "remote-agent"; + +/// Task for an agent executing outside Macro, stored in the untyped `task` +/// column of a [`ScheduledAction`] whose kind is [`ActionKind::RemoteAgent`]. +#[derive(Serialize, Deserialize, Debug, Clone, ToSchema)] +pub struct RemoteAgentTask { + /// HTTPS endpoint invoked on each run. + pub endpoint_url: String, + /// Opaque prompt forwarded to the remote agent. + pub user_prompt: String, + /// Optional label shown in place of a model name. + #[serde(default)] + pub agent_label: Option, +} + +impl RemoteAgentTask { + /// Label to attribute this task's chat messages to. + pub fn label(&self) -> &str { + match self.agent_label.as_deref() { + Some(label) if !label.trim().is_empty() => label, + _ => DEFAULT_REMOTE_AGENT_LABEL, + } + } +} + +/// Payload POSTed to a remote agent endpoint on each run. The remote agent is +/// given the identifiers of the run so it can correlate its own logs; it is not +/// expected to write to Macro itself. +#[derive(Serialize, Deserialize, Debug, Clone, ToSchema)] +pub struct RemoteAgentRunRequest { + /// Scheduled action being run. + #[schema(value_type = String, format = Uuid)] + pub action_id: Uuid, + /// Chat the run transcript is written to. + pub chat_id: String, + /// Name of the scheduled action, as shown in the UI. + pub action_name: String, + /// Prompt configured on the action. + pub user_prompt: String, +} + +/// Response a remote agent endpoint is expected to return. +#[derive(Serialize, Deserialize, Debug, Clone, ToSchema)] +pub struct RemoteAgentRunResponse { + /// Text written back into the run chat as the assistant reply. + pub output: String, +} + /// Client-supplied payload for creating a scheduled action. The server fills /// in `id`, `owner` (from the authenticated user), timestamps, `claimed`, and /// `next_run_at` (derived from the cron). diff --git a/services/scheduled_action/src/domain/ports.rs b/services/scheduled_action/src/domain/ports.rs index 57c79e88e1f..ab40a2d4433 100644 --- a/services/scheduled_action/src/domain/ports.rs +++ b/services/scheduled_action/src/domain/ports.rs @@ -1,6 +1,6 @@ use super::models::{ - ActionExecutionRecord, DispatchEvent, InProgressExecution, ScheduledAction, - ScheduledActionUpdate, + ActionExecutionRecord, DispatchEvent, InProgressExecution, RemoteAgentRunRequest, + RemoteAgentRunResponse, RemoteAgentTask, ScheduledAction, ScheduledActionUpdate, }; use anyhow::Result; use chrono::{DateTime, Utc}; @@ -111,3 +111,14 @@ pub trait ScheduledActionExecutor { pub trait ScheduledActionLiveUpdate: Send + Sync + 'static { fn publish_update(&self, update: ScheduledActionUpdate) -> impl Future + Send; } + +/// Invokes an agent that runs outside Macro. Implementations own transport +/// concerns — endpoint validation, timeouts, redirect policy and any future +/// request signing — so the executor stays transport-agnostic. +pub trait RemoteAgentClient: Send + Sync + 'static { + fn run( + &self, + task: &RemoteAgentTask, + request: &RemoteAgentRunRequest, + ) -> impl Future> + Send; +} diff --git a/services/scheduled_action/src/outbound.rs b/services/scheduled_action/src/outbound.rs index 8ca29def91a..38e71408d48 100644 --- a/services/scheduled_action/src/outbound.rs +++ b/services/scheduled_action/src/outbound.rs @@ -2,4 +2,5 @@ pub mod conn_gateway_live_updates; pub mod inprocess_executor; pub mod pg_polling_dispatcher; pub mod pg_scheduled_action_repo; +pub mod remote_agent_http; pub mod tokio_dispatcher; diff --git a/services/scheduled_action/src/outbound/inprocess_executor/mod.rs b/services/scheduled_action/src/outbound/inprocess_executor/mod.rs index 43603caeafe..329428be0bf 100644 --- a/services/scheduled_action/src/outbound/inprocess_executor/mod.rs +++ b/services/scheduled_action/src/outbound/inprocess_executor/mod.rs @@ -1,5 +1,6 @@ mod agent_task; mod notify; +mod remote_agent_task; use std::sync::Arc; @@ -16,24 +17,32 @@ use crate::domain::models::{ ScheduledAction, ScheduledActionUpdate, }; use crate::domain::ports::{ - ScheduledActionExecutor, ScheduledActionLiveUpdate, ScheduledActionRepo, + RemoteAgentClient, ScheduledActionExecutor, ScheduledActionLiveUpdate, ScheduledActionRepo, }; -pub struct InProcessExecutor { +pub struct InProcessExecutor< + Rpo: ScheduledActionRepo, + Live: ScheduledActionLiveUpdate, + Remote: RemoteAgentClient, +> { repo: Arc, db: PgPool, tool_context: ToolServiceContext, notification_ingress: Arc>, live_updates: Arc, + remote_agent_client: Arc, } -impl InProcessExecutor { +impl + InProcessExecutor +{ pub fn new( repo: Arc, db: PgPool, tool_context: ToolServiceContext, notification_ingress: Arc>, live_updates: Arc, + remote_agent_client: Arc, ) -> Self { Self { repo, @@ -41,6 +50,7 @@ impl InProcessExecuto tool_context, notification_ingress, live_updates, + remote_agent_client, } } } @@ -57,10 +67,11 @@ fn try_claim(action: &ScheduledAction) -> Result<()> { Ok(()) } -impl ScheduledActionExecutor for InProcessExecutor +impl ScheduledActionExecutor for InProcessExecutor where Rpo: ScheduledActionRepo + Send + Sync + 'static, Live: ScheduledActionLiveUpdate, + Remote: RemoteAgentClient, { async fn execute_action(&self, action: ScheduledAction) -> Result { try_claim(&action)?; @@ -70,8 +81,29 @@ where // Create the chat up front so the caller gets a chat_id synchronously // and the eventual execution record can link back to it. - let chat_id = match action.kind { - ActionKind::Agent => agent_task::create_run_chat(&self.db, &action).await?, + // Both kinds write their run transcript to a chat, so the UI (which + // navigates by chat_id) is identical for a remote agent. + let created_chat = match action.kind { + ActionKind::Agent | ActionKind::RemoteAgent => { + agent_task::create_run_chat(&self.db, &action).await + } + }; + + // The claim is taken before setup, and only the spawned run releases + // it. Failing here without releasing would leave the action unrunnable + // until the claim goes stale (MAX_ACTION_TIME). + let chat_id = match created_chat { + Ok(chat_id) => chat_id, + Err(e) => { + if let Err(release_error) = self.repo.release_action(&id).await { + tracing::error!( + error=?release_error, + action_id=?id, + "failed to release action claim after run setup failed" + ); + } + return Err(e); + } }; self.live_updates @@ -92,11 +124,19 @@ where let tool_context = self.tool_context.clone(); let notification_ingress = Arc::clone(&self.notification_ingress); let live_updates = Arc::clone(&self.live_updates); + let remote_agent_client = Arc::clone(&self.remote_agent_client); let start_time = Utc::now(); let record_resource_id = chat_id.clone(); tokio::spawn(async move { - let result = - run_job(&db, &tool_context, ¬ification_ingress, &action, &chat_id).await; + let result = run_job( + &db, + &tool_context, + ¬ification_ingress, + remote_agent_client.as_ref(), + &action, + &chat_id, + ) + .await; let end_time = Utc::now(); let is_success = result.is_ok(); @@ -154,10 +194,11 @@ where } } -async fn run_job( +async fn run_job( db: &PgPool, tool_context: &ToolServiceContext, notification_ingress: &Arc>, + remote_agent_client: &Remote, action: &ScheduledAction, chat_id: &str, ) -> Result<()> { @@ -167,5 +208,16 @@ async fn run_job( .await?; Ok(()) } + ActionKind::RemoteAgent => { + remote_agent_task::run_remote_agent_task( + db, + remote_agent_client, + notification_ingress, + action, + chat_id, + ) + .await?; + Ok(()) + } } } diff --git a/services/scheduled_action/src/outbound/inprocess_executor/remote_agent_task.rs b/services/scheduled_action/src/outbound/inprocess_executor/remote_agent_task.rs new file mode 100644 index 00000000000..1630fc7b48d --- /dev/null +++ b/services/scheduled_action/src/outbound/inprocess_executor/remote_agent_task.rs @@ -0,0 +1,97 @@ +//! Run path for [`ActionKind::RemoteAgent`]. +//! +//! Mirrors [`super::agent_task`] so the whole run surface — the chat, the live +//! update, the execution record, the completion notification — is inherited by +//! remote agents. The only difference is who produces the assistant message: +//! here it comes back over HTTP instead of from Macro's own agent loop. + +use std::sync::Arc; + +use agent::types::{AssistantMessagePart, ChatMessageContent, Role}; +use anyhow::{Context, Result, bail}; +use macro_db_client::dcs::create_chat_message::create_chat_message; +use model::chat::NewChatMessage; +use notification::domain::service::SqsNotificationIngress; +use notification::outbound::queue::SqsQueue; +use sqlx::PgPool; + +use super::notify::notify_completion; +use crate::domain::models::{RemoteAgentRunRequest, RemoteAgentTask, ScheduledAction}; +use crate::domain::ports::RemoteAgentClient; + +pub async fn run_remote_agent_task( + db: &PgPool, + remote_client: &Remote, + notification_ingress: &Arc>, + action: &ScheduledAction, + chat_id: &str, +) -> Result<()> { + let task: RemoteAgentTask = serde_json::from_value(action.task.clone()) + .context("invalid remote agent task definition")?; + + let action_id = *action + .id + .as_ref() + .context("scheduled action is missing an id")?; + + store_user_message(db, chat_id, &task).await?; + + let request = RemoteAgentRunRequest { + action_id, + chat_id: chat_id.to_string(), + action_name: action.name.clone(), + user_prompt: task.user_prompt.clone(), + }; + + let response = remote_client.run(&task, &request).await?; + + // A 200 with nothing in it is a misconfigured endpoint, not a run that did + // nothing: recording it as successful would leave an empty chat with no + // explanation. Failing puts the reason on the execution record instead. + if response.output.trim().is_empty() { + bail!("remote agent returned an empty response"); + } + + store_assistant_message(db, chat_id, &task, &response.output).await?; + notify_completion(notification_ingress, chat_id, action, &response.output); + + Ok(()) +} + +async fn store_user_message(db: &PgPool, chat_id: &str, task: &RemoteAgentTask) -> Result { + let now = chrono::Utc::now(); + let message = NewChatMessage { + id: None, + content: ChatMessageContent::Text(task.user_prompt.clone()), + role: Role::User, + attachments: None, + created_at: now, + updated_at: now, + model: task.label().to_string(), + }; + create_chat_message(db.clone(), chat_id, message).await +} + +async fn store_assistant_message( + db: &PgPool, + chat_id: &str, + task: &RemoteAgentTask, + output: &str, +) -> Result<()> { + let now = chrono::Utc::now(); + let message = NewChatMessage { + id: None, + content: ChatMessageContent::AssistantMessageParts(vec![AssistantMessagePart::Text { + text: output.to_string(), + }]), + role: Role::Assistant, + attachments: None, + created_at: now, + updated_at: now, + model: task.label().to_string(), + }; + create_chat_message(db.clone(), chat_id, message) + .await + .context("failed to store remote agent message")?; + Ok(()) +} diff --git a/services/scheduled_action/src/outbound/pg_scheduled_action_repo.rs b/services/scheduled_action/src/outbound/pg_scheduled_action_repo.rs index b6c10436599..71967547144 100644 --- a/services/scheduled_action/src/outbound/pg_scheduled_action_repo.rs +++ b/services/scheduled_action/src/outbound/pg_scheduled_action_repo.rs @@ -1,3 +1,6 @@ +#[cfg(test)] +mod test; + use anyhow::{Result, bail}; use chrono::{DateTime, Utc}; use chrono_tz::Tz; @@ -27,9 +30,13 @@ fn parse_timezone(s: &str) -> Result { Tz::from_str(s).map_err(|e| anyhow::anyhow!("invalid timezone: {e}")) } +/// Map the `kind` TEXT column to the domain enum. Kept in lockstep with +/// [`kind_to_str`]: a variant missing here turns every row of that kind into a +/// read error, since `kind` is plain TEXT rather than a Postgres enum. fn parse_kind(s: &str) -> Result { match s { "Agent" => Ok(ActionKind::Agent), + "RemoteAgent" => Ok(ActionKind::RemoteAgent), other => bail!("unknown action kind: {other}"), } } @@ -37,6 +44,7 @@ fn parse_kind(s: &str) -> Result { fn kind_to_str(kind: &ActionKind) -> &'static str { match kind { ActionKind::Agent => "Agent", + ActionKind::RemoteAgent => "RemoteAgent", } } diff --git a/services/scheduled_action/src/outbound/pg_scheduled_action_repo/test.rs b/services/scheduled_action/src/outbound/pg_scheduled_action_repo/test.rs new file mode 100644 index 00000000000..9ef6a9e220a --- /dev/null +++ b/services/scheduled_action/src/outbound/pg_scheduled_action_repo/test.rs @@ -0,0 +1,56 @@ +use super::{kind_to_str, parse_kind}; +use crate::domain::models::{ActionKind, RemoteAgentTask}; + +/// `kind` is a plain TEXT column, so the two hand-rolled mappings are the only +/// thing keeping stored rows readable. A variant added to one and not the other +/// turns every row of that kind into a read error at runtime rather than a +/// compile error, so pin the round trip. +#[test] +fn every_kind_round_trips_through_the_text_column() { + for kind in [ActionKind::Agent, ActionKind::RemoteAgent] { + let stored = kind_to_str(&kind); + let parsed = parse_kind(stored).expect("stored kind parses back"); + + assert_eq!(kind_to_str(&parsed), stored); + } +} + +#[test] +fn kinds_use_their_serde_names() { + assert_eq!(kind_to_str(&ActionKind::Agent), "Agent"); + assert_eq!(kind_to_str(&ActionKind::RemoteAgent), "RemoteAgent"); +} + +#[test] +fn unknown_kinds_are_rejected() { + assert!(parse_kind("NotAKind").is_err()); + assert!(parse_kind("remoteagent").is_err()); +} + +/// The `task` column is untyped JSONB and is only interpreted by the executor, +/// so the shape a remote agent action stores is part of this service's contract. +#[test] +fn remote_agent_tasks_deserialize_from_stored_json() { + let stored = serde_json::json!({ + "endpoint_url": "https://agent.example.com/run", + "user_prompt": "summarise yesterday's incidents", + "agent_label": "hermes" + }); + + let task: RemoteAgentTask = serde_json::from_value(stored).expect("valid remote agent task"); + + assert_eq!(task.endpoint_url, "https://agent.example.com/run"); + assert_eq!(task.label(), "hermes"); +} + +#[test] +fn remote_agent_tasks_fall_back_to_a_default_label() { + let stored = serde_json::json!({ + "endpoint_url": "https://agent.example.com/run", + "user_prompt": "summarise yesterday's incidents" + }); + + let task: RemoteAgentTask = serde_json::from_value(stored).expect("valid remote agent task"); + + assert_eq!(task.label(), "remote-agent"); +} diff --git a/services/scheduled_action/src/outbound/remote_agent_http.rs b/services/scheduled_action/src/outbound/remote_agent_http.rs new file mode 100644 index 00000000000..d576a3fb3c4 --- /dev/null +++ b/services/scheduled_action/src/outbound/remote_agent_http.rs @@ -0,0 +1,300 @@ +//! HTTP adapter for agents that execute outside Macro. +//! +//! The endpoint URL is user-supplied, so this adapter applies the same +//! guardrails the webhook delivery path uses: HTTPS only, no redirects, a +//! request timeout, and rejection of hosts that resolve to private, loopback, +//! link-local or cloud-metadata addresses. +//! +//! The address rules are mirrored from +//! `crates/webhook/src/outbound/http_validator.rs`, whose validator is +//! `pub(super)` and therefore not reachable from this service. If a third +//! caller appears, that validator is worth promoting into a shared crate rather +//! than copied again. +//! +//! Known residual risk, shared with the webhook path it mirrors: the host is +//! resolved for validation and then resolved again by the request itself, so a +//! DNS record that changes between the two can still point the request at a +//! blocked address. Closing that means pinning the request to the address that +//! was validated; it is deliberately left as-is here so both paths keep the +//! same behaviour and can be fixed together. + +#[cfg(test)] +mod test; + +use std::net::{IpAddr, Ipv4Addr}; +use std::time::Duration; + +use anyhow::{Result, bail}; +use futures::StreamExt; +use reqwest::{Client, Response, Url, redirect::Policy}; +use tokio::net::lookup_host; + +use crate::domain::models::{RemoteAgentRunRequest, RemoteAgentRunResponse, RemoteAgentTask}; +use crate::domain::ports::RemoteAgentClient; + +/// Wall-clock budget for a single remote run. Deliberately shorter than +/// [`crate::domain::models::MAX_ACTION_TIME`] so a hung endpoint releases the +/// action's claim well before the staleness window expires. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(120); +/// Bytes of a failed response echoed into the execution record. +const RESPONSE_PREVIEW_MAX_BYTES: usize = 4096; +/// Upper bound on a response body this service will buffer. +const MAX_RESPONSE_BYTES: usize = 1024 * 1024; +/// Upper bound on the assistant text stored from a successful response. +const MAX_OUTPUT_CHARS: usize = 100_000; + +/// Whether non-public endpoints may be used. Self-hosted deployments own both +/// ends of the connection, so they can opt into local addresses; the default +/// matches the webhook default and refuses them. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum RemoteAgentEndpointPolicy { + /// Require HTTPS endpoints that resolve to public addresses. + #[default] + PublicHttpsOnly, + /// Allow HTTP and non-public addresses. Intended for local development and + /// self-hosted deployments where the operator owns the remote agent. + AllowLocal, +} + +impl RemoteAgentEndpointPolicy { + /// Whether a URL scheme is permitted by this policy. + fn allows_scheme(self, scheme: &str) -> bool { + scheme == "https" || (self == Self::AllowLocal && scheme == "http") + } + + /// Whether loopback, private and link-local addresses are permitted. + fn allows_local_addresses(self) -> bool { + self == Self::AllowLocal + } +} + +/// `reqwest`-backed [`RemoteAgentClient`]. +#[derive(Clone)] +pub struct ReqwestRemoteAgentClient { + client: Client, + endpoint_policy: RemoteAgentEndpointPolicy, +} + +impl ReqwestRemoteAgentClient { + /// Create a client that requires public HTTPS endpoints. + pub fn new() -> Result { + Self::new_with_endpoint_policy(RemoteAgentEndpointPolicy::default()) + } + + /// Create a client with an explicit endpoint policy. + pub fn new_with_endpoint_policy( + endpoint_policy: RemoteAgentEndpointPolicy, + ) -> Result { + let client = Client::builder() + .timeout(REQUEST_TIMEOUT) + .redirect(Policy::none()) + .build()?; + + Ok(Self { + client, + endpoint_policy, + }) + } +} + +impl RemoteAgentClient for ReqwestRemoteAgentClient { + #[tracing::instrument(skip(self, task, request), fields(action_id = %request.action_id), err)] + async fn run( + &self, + task: &RemoteAgentTask, + request: &RemoteAgentRunRequest, + ) -> Result { + let url = validate_endpoint_url(&task.endpoint_url, self.endpoint_policy)?; + validate_resolved_addresses(&url, self.endpoint_policy).await?; + + // `reqwest::Error`'s Display embeds the request URL, and this error is + // persisted on the execution record and shown in the UI, so strip it. + let response = self + .client + .post(url) + .json(request) + .send() + .await + .map_err(|e| anyhow::anyhow!("remote agent request failed: {}", e.without_url()))?; + + let status = response.status(); + let body = read_capped_body(response).await?; + + if !status.is_success() { + bail!( + "remote agent returned {}: {}", + status.as_u16(), + preview(&body) + ); + } + + Ok(parse_response(&body)) + } +} + +/// Read a response body, refusing to buffer more than +/// [`MAX_RESPONSE_BYTES`]. A remote agent is operator-run but not trusted to +/// bound its own output, and this runs inside the service process. +async fn read_capped_body(response: Response) -> Result { + let mut stream = response.bytes_stream(); + let mut buffer: Vec = Vec::new(); + + while let Some(chunk) = stream.next().await { + let chunk = chunk + .map_err(|e| anyhow::anyhow!("remote agent response could not be read: {}", e.without_url()))?; + + if buffer.len() + chunk.len() > MAX_RESPONSE_BYTES { + bail!("remote agent response exceeded {MAX_RESPONSE_BYTES} bytes"); + } + + buffer.extend_from_slice(&chunk); + } + + Ok(String::from_utf8_lossy(&buffer).into_owned()) +} + +/// Interpret a successful response body. +/// +/// The documented contract is a JSON [`RemoteAgentRunResponse`]. Self-hosted +/// agents that answer with plain text are still usable: their body becomes the +/// output verbatim, which keeps the integration approachable without loosening +/// the documented shape. +fn parse_response(body: &str) -> RemoteAgentRunResponse { + let output = match serde_json::from_str::(body) { + Ok(parsed) => parsed.output, + Err(_) => body.trim().to_string(), + }; + + RemoteAgentRunResponse { + output: truncate_chars(&output, MAX_OUTPUT_CHARS), + } +} + +/// Parse and scheme-check the endpoint, rejecting obviously local hosts before +/// any DNS work happens. +fn validate_endpoint_url(value: &str, policy: RemoteAgentEndpointPolicy) -> Result { + let Ok(url) = Url::parse(value) else { + bail!("remote agent endpoint URL is invalid"); + }; + + if !policy.allows_scheme(url.scheme()) { + bail!("remote agent endpoint URL must use HTTPS"); + } + + let Some(host) = url.host_str() else { + bail!("remote agent endpoint URL host is invalid"); + }; + + if !policy.allows_local_addresses() && is_blocked_host(host) { + bail!("remote agent endpoint host is not allowed"); + } + + Ok(url) +} + +/// Resolve the endpoint host and reject it if any address it resolves to is +/// disallowed. Checking after resolution is what stops a public hostname that +/// points at an internal address. +async fn validate_resolved_addresses(url: &Url, policy: RemoteAgentEndpointPolicy) -> Result<()> { + if policy.allows_local_addresses() { + return Ok(()); + } + + let Some(host) = url.host_str().map(str::to_owned) else { + bail!("remote agent endpoint URL host is invalid"); + }; + let Some(port) = url.port_or_known_default() else { + bail!("remote agent endpoint URL port is invalid"); + }; + + let resolved = match tokio::time::timeout(REQUEST_TIMEOUT, lookup_host((host.as_str(), port))) + .await + { + Ok(Ok(resolved)) => resolved, + Ok(Err(_)) => bail!("remote agent endpoint host could not be resolved"), + Err(_) => bail!("remote agent endpoint host resolution timed out"), + }; + + let mut saw_address = false; + for address in resolved { + saw_address = true; + if is_blocked_ip(address.ip()) { + bail!("remote agent endpoint host resolves to a disallowed address"); + } + } + + if !saw_address { + bail!("remote agent endpoint host could not be resolved"); + } + + Ok(()) +} + +/// Hosts that are rejected without resolving them first. +fn is_blocked_host(host: &str) -> bool { + let host = host.trim_matches(['[', ']']).to_ascii_lowercase(); + if host == "localhost" || host.ends_with(".localhost") { + return true; + } + + host.parse::().is_ok_and(is_blocked_ip) +} + +/// Addresses a scheduled action must never reach. +fn is_blocked_ip(ip: IpAddr) -> bool { + match ip { + IpAddr::V4(ip) => { + ip.is_private() + || ip.is_loopback() + || ip.is_link_local() + || ip.is_broadcast() + || ip.is_unspecified() + || ip.octets() == [169, 254, 169, 254] + || is_shared_v4(ip) + } + IpAddr::V6(ip) => { + // `::ffff:169.254.169.254` reaches the same host as the IPv4 + // address it embeds, so fold mapped addresses back onto the IPv4 + // rules before applying the v6 ones. + if let Some(mapped) = ip.to_ipv4_mapped() { + return is_blocked_ip(IpAddr::V4(mapped)); + } + + ip.is_loopback() + || ip.is_unspecified() + || ip.is_unique_local() + || (ip.segments()[0] & 0xffc0) == 0xfe80 + } + } +} + +/// RFC 6598 shared address space (100.64.0.0/10), used for carrier-grade NAT +/// and inside some cloud networks. `Ipv4Addr::is_shared` is still unstable, so +/// check the prefix directly. +fn is_shared_v4(ip: Ipv4Addr) -> bool { + let [first, second, ..] = ip.octets(); + first == 100 && (64..=127).contains(&second) +} + +/// First [`RESPONSE_PREVIEW_MAX_BYTES`] of a body, on a char boundary. +fn preview(body: &str) -> &str { + if body.len() <= RESPONSE_PREVIEW_MAX_BYTES { + return body; + } + + let mut end = RESPONSE_PREVIEW_MAX_BYTES; + while end > 0 && !body.is_char_boundary(end) { + end -= 1; + } + &body[..end] +} + +/// Keep at most `max` characters, counting characters rather than bytes so the +/// result is always valid UTF-8. +fn truncate_chars(value: &str, max: usize) -> String { + if value.chars().count() <= max { + return value.to_string(); + } + + value.chars().take(max).collect() +} diff --git a/services/scheduled_action/src/outbound/remote_agent_http/test.rs b/services/scheduled_action/src/outbound/remote_agent_http/test.rs new file mode 100644 index 00000000000..1a0dbf98ee2 --- /dev/null +++ b/services/scheduled_action/src/outbound/remote_agent_http/test.rs @@ -0,0 +1,97 @@ +use super::{ + MAX_OUTPUT_CHARS, RemoteAgentEndpointPolicy, is_blocked_host, is_blocked_ip, parse_response, + validate_endpoint_url, +}; +use std::net::IpAddr; +use std::str::FromStr; + +fn blocked(ip: &str) -> bool { + is_blocked_ip(IpAddr::from_str(ip).expect("valid ip")) +} + +#[test] +fn rejects_private_loopback_and_metadata_addresses() { + assert!(blocked("127.0.0.1")); + assert!(blocked("10.1.2.3")); + assert!(blocked("192.168.0.4")); + assert!(blocked("172.16.9.9")); + assert!(blocked("169.254.169.254")); + assert!(blocked("0.0.0.0")); + assert!(blocked("255.255.255.255")); + assert!(blocked("::1")); + assert!(blocked("fd00::1")); + assert!(blocked("fe80::1")); + // IPv4-mapped forms reach the same hosts as the addresses they embed. + assert!(blocked("::ffff:169.254.169.254")); + assert!(blocked("::ffff:127.0.0.1")); + assert!(blocked("::ffff:10.0.0.1")); + // RFC 6598 shared address space (carrier-grade NAT, some cloud networks). + assert!(blocked("100.64.0.1")); + assert!(blocked("100.127.255.255")); +} + +#[test] +fn allows_public_addresses() { + assert!(!blocked("93.184.216.34")); + // Neighbours of the RFC 6598 block that are ordinary public addresses. + assert!(!blocked("100.63.255.255")); + assert!(!blocked("100.128.0.1")); + assert!(!blocked("2606:2800:220:1:248:1893:25c8:1946")); +} + +#[test] +fn rejects_localhost_hostnames_without_resolving() { + assert!(is_blocked_host("localhost")); + assert!(is_blocked_host("LOCALHOST")); + assert!(is_blocked_host("agent.localhost")); + assert!(is_blocked_host("127.0.0.1")); + assert!(is_blocked_host("[::1]")); + assert!(!is_blocked_host("agent.example.com")); +} + +#[test] +fn requires_https_under_the_default_policy() { + let policy = RemoteAgentEndpointPolicy::default(); + + assert!(validate_endpoint_url("https://agent.example.com/run", policy).is_ok()); + assert!(validate_endpoint_url("http://agent.example.com/run", policy).is_err()); + assert!(validate_endpoint_url("ftp://agent.example.com/run", policy).is_err()); + assert!(validate_endpoint_url("not a url", policy).is_err()); +} + +#[test] +fn rejects_local_endpoints_under_the_default_policy() { + let policy = RemoteAgentEndpointPolicy::default(); + + assert!(validate_endpoint_url("https://localhost:8443/run", policy).is_err()); + assert!(validate_endpoint_url("https://127.0.0.1/run", policy).is_err()); + assert!(validate_endpoint_url("https://[::1]/run", policy).is_err()); +} + +#[test] +fn allows_local_endpoints_only_when_configured() { + let policy = RemoteAgentEndpointPolicy::AllowLocal; + + assert!(validate_endpoint_url("http://localhost:8443/run", policy).is_ok()); + assert!(validate_endpoint_url("https://10.0.0.5/run", policy).is_ok()); +} + +#[test] +fn reads_the_documented_json_response() { + let parsed = parse_response(r#"{"output":"the daily digest"}"#); + assert_eq!(parsed.output, "the daily digest"); +} + +#[test] +fn falls_back_to_the_raw_body_for_plain_text_agents() { + let parsed = parse_response(" the daily digest\n"); + assert_eq!(parsed.output, "the daily digest"); +} + +#[test] +fn truncates_oversized_output_on_a_char_boundary() { + let body = "é".repeat(MAX_OUTPUT_CHARS + 10); + let parsed = parse_response(&body); + + assert_eq!(parsed.output.chars().count(), MAX_OUTPUT_CHARS); +} diff --git a/services/scheduled_action/src/swagger.rs b/services/scheduled_action/src/swagger.rs index 51178b8aaf1..692f7ac456d 100644 --- a/services/scheduled_action/src/swagger.rs +++ b/services/scheduled_action/src/swagger.rs @@ -11,7 +11,8 @@ use crate::inbound::axum_router::{ use crate::domain::models::{ ActionExecutionRecord, ActionKind, AgentTask, CreateScheduledAction, InProgressExecution, - Schedule, ScheduledAction, ScheduledActionUpdate, UpdateScheduledAction, + RemoteAgentRunRequest, RemoteAgentRunResponse, RemoteAgentTask, Schedule, ScheduledAction, + ScheduledActionUpdate, UpdateScheduledAction, }; use model::response::EmptyResponse; @@ -39,6 +40,9 @@ use model::response::EmptyResponse; Schedule, ActionKind, AgentTask, + RemoteAgentTask, + RemoteAgentRunRequest, + RemoteAgentRunResponse, InProgressExecution, ActionExecutionRecord, ScheduledActionUpdate,