diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 928eb54be..f6d8729c9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1270,6 +1270,65 @@ jobs: psexec -accepteula -s pwsh.exe $scriptPath Get-Content -Path ./crates/pedm-simulator/pedm-simulator_run-expect-elevation.out + agent-policy-e2e: + name: Agent policy end-to-end test + runs-on: windows-2022 + needs: [preflight] + + steps: + - name: Checkout ${{ github.repository }} + uses: actions/checkout@v6 + with: + ref: ${{ needs.preflight.outputs.ref }} + + - name: Setup Rust cache + uses: ./.github/actions/setup-rust-cache + with: + sccache-enabled: ${{ needs.preflight.outputs.sccache }} + + # Keep this installation aligned with the PEDM simulator job. + - name: Install PsExec + shell: pwsh + run: | + $expectedHash = '4F49964CC9CBAC2B5D87BDC8F9526012E9C4B243D8B7D0C0BB51F254A721CA2E' + $zipPath = Join-Path $env:RUNNER_TEMP 'PSTools.zip' + $toolsDir = Join-Path $env:RUNNER_TEMP 'PSTools' + Invoke-WebRequest -Uri 'https://download.sysinternals.com/files/PSTools.zip' -OutFile $zipPath + $actualHash = (Get-FileHash -Path $zipPath -Algorithm SHA256).Hash + if ($actualHash -ne $expectedHash) { + throw "PSTools.zip checksum mismatch: expected $expectedHash, got $actualHash" + } + Expand-Archive -Path $zipPath -DestinationPath $toolsDir + Add-Content -Path $env:GITHUB_PATH -Value $toolsDir + + - name: Build Agent policy test executables + shell: pwsh + run: | + cargo build --locked -p devolutions-agent --features dev-skip-broker-signature + if ($LASTEXITCODE -ne 0) { + exit $LASTEXITCODE + } + cargo build --locked -p agent-policy-tester + if ($LASTEXITCODE -ne 0) { + exit $LASTEXITCODE + } + + - name: Run Agent policy tester as LocalSystem + shell: pwsh + run: | + $scriptPath = Resolve-Path -Path "./crates/agent-policy-tester/run-as-system.ps1" + psexec -accepteula -s pwsh.exe -NoProfile -File $scriptPath + $exitCode = $LASTEXITCODE + Get-Content -Path ./crates/agent-policy-tester/agent-policy-tester.out + if ($exitCode -ne 0) { + exit $exitCode + } + + - name: Show sccache stats + if: ${{ needs.preflight.outputs.sccache == 'true' && !cancelled() }} + shell: pwsh + run: sccache --show-stats + secure-memory-verifier: name: secure-memory-verifier runs-on: windows-2022 @@ -1298,7 +1357,7 @@ jobs: success: name: Success if: ${{ always() }} - needs: [tests, agent-tunnel-e2e, lints, check-dependencies, jetsocat-lipo, devolutions-gateway-powershell, devolutions-gateway, devolutions-gateway-merge, devolutions-pedm-desktop, devolutions-agent, devolutions-agent-merge, devolutions-pedm-client, dotnet-utils-tests, winapi-sanitizer-tests, winapi-miri, pedm-simulator, secure-memory-verifier] + needs: [tests, agent-tunnel-e2e, agent-policy-e2e, lints, check-dependencies, jetsocat-lipo, devolutions-gateway-powershell, devolutions-gateway, devolutions-gateway-merge, devolutions-pedm-desktop, devolutions-agent, devolutions-agent-merge, devolutions-pedm-client, dotnet-utils-tests, winapi-sanitizer-tests, winapi-miri, pedm-simulator, secure-memory-verifier] runs-on: ubuntu-latest steps: diff --git a/Cargo.lock b/Cargo.lock index a6df6c2b6..0aedd5f92 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -85,6 +85,17 @@ dependencies = [ "const-oid 0.10.2", ] +[[package]] +name = "agent-policy-tester" +version = "0.0.0" +dependencies = [ + "anyhow", + "fastrand", + "serde_json", + "tempfile", + "tokio 1.52.3", +] + [[package]] name = "agent-tunnel" version = "0.0.0" @@ -4804,7 +4815,6 @@ checksum = "fa0817fd85c0a6b0173e2b837fa2b369be684c93c11fed1b5182284021411839" dependencies = [ "chrono", "derive_more", - "now-policy", "schemars", "semver", "serde", @@ -4822,7 +4832,6 @@ dependencies = [ "aide", "async-trait", "axum 0.8.9", - "now-policy", "now-policy-api", "schemars", "serde", diff --git a/crates/agent-policy-tester/.gitignore b/crates/agent-policy-tester/.gitignore new file mode 100644 index 000000000..27f5781ff --- /dev/null +++ b/crates/agent-policy-tester/.gitignore @@ -0,0 +1 @@ +/agent-policy-tester.out diff --git a/crates/agent-policy-tester/Cargo.toml b/crates/agent-policy-tester/Cargo.toml new file mode 100644 index 000000000..ba2f20f77 --- /dev/null +++ b/crates/agent-policy-tester/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "agent-policy-tester" +version = "0.0.0" +edition = "2024" +publish = false + +[dependencies] +anyhow = "1" + +[target.'cfg(windows)'.dependencies] +fastrand = "2" +serde_json = "1" +tempfile = "3" +tokio = { version = "1", features = ["io-util", "macros", "net", "process", "rt-multi-thread", "time"] } + +[lints] +workspace = true diff --git a/crates/agent-policy-tester/run-as-system.ps1 b/crates/agent-policy-tester/run-as-system.ps1 new file mode 100644 index 000000000..10b988fb4 --- /dev/null +++ b/crates/agent-policy-tester/run-as-system.ps1 @@ -0,0 +1,16 @@ +$ErrorActionPreference = "Stop" + +$workspacePath = (Resolve-Path (Join-Path $PSScriptRoot "../..")).Path +$testerPath = Join-Path $workspacePath "target/debug/agent-policy-tester.exe" +$agentPath = Join-Path $workspacePath "target/debug/devolutions-agent.exe" +$outputPath = Join-Path $PSScriptRoot "agent-policy-tester.out" + +try { + & $testerPath $agentPath 2>&1 | Out-File $outputPath + $exitCode = $LASTEXITCODE +} catch { + $_ | Out-File $outputPath -Append + exit 1 +} + +exit $exitCode diff --git a/crates/agent-policy-tester/src/main.rs b/crates/agent-policy-tester/src/main.rs new file mode 100644 index 000000000..17b1d2160 --- /dev/null +++ b/crates/agent-policy-tester/src/main.rs @@ -0,0 +1,13 @@ +#[cfg(windows)] +mod windows; + +#[cfg(windows)] +#[tokio::main] +async fn main() -> anyhow::Result<()> { + windows::run().await +} + +#[cfg(not(windows))] +fn main() -> anyhow::Result<()> { + anyhow::bail!("agent policy tester only supports Windows") +} diff --git a/crates/agent-policy-tester/src/windows.rs b/crates/agent-policy-tester/src/windows.rs new file mode 100644 index 000000000..3ff4250a5 --- /dev/null +++ b/crates/agent-policy-tester/src/windows.rs @@ -0,0 +1,318 @@ +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::time::{Duration, Instant}; + +use anyhow::{Context as _, bail, ensure}; +use serde_json::{Value, json}; +use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; +use tokio::net::windows::named_pipe::ClientOptions; + +const FULL_POLICY: &str = include_str!("../../now-package-broker/src/assets/samples/corporate-allowlist.policy.json"); + +struct AgentHarness { + child: tokio::process::Child, + _data_dir: tempfile::TempDir, + pipe_name: String, + policy_path: PathBuf, +} + +impl AgentHarness { + async fn start(agent_path: &Path, policy: Option<&Value>) -> anyhow::Result { + let data_dir = tempfile::tempdir().context("create Agent data directory")?; + let pipe_name = format!( + r"\\.\pipe\Devolutions.Now.PackageBroker.tests.{}.{}", + std::process::id(), + fastrand::u64(..) + ); + let policy_path = data_dir.path().join("policy.json"); + + if let Some(policy) = policy { + std::fs::write(&policy_path, serde_json::to_vec_pretty(policy)?).context("write policy")?; + secure_policy_file(&policy_path)?; + } + + let config = json!({ + "PackageBroker": { + "Enabled": true, + "PipeName": pipe_name, + "PolicyPath": policy_path, + }, + "__debug__": { + "skip_broker_signature_validation": true, + }, + }); + std::fs::write(data_dir.path().join("agent.json"), serde_json::to_vec_pretty(&config)?) + .context("write Agent configuration")?; + + let child = tokio::process::Command::new(agent_path) + .env("DAGENT_CONFIG_PATH", data_dir.path()) + .arg("run") + .kill_on_drop(true) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .context("start Devolutions Agent")?; + + let mut harness = Self { + child, + _data_dir: data_dir, + pipe_name, + policy_path, + }; + harness.wait_until_ready().await?; + + Ok(harness) + } + + async fn wait_until_ready(&mut self) -> anyhow::Result<()> { + let deadline = Instant::now() + Duration::from_secs(20); + + loop { + if let Some(status) = self.child.try_wait().context("query Agent status")? { + bail!("agent exited before package broker startup with {status}"); + } + + match request(&self.pipe_name, "GET", "/v1/health").await { + Ok(response) if response.status == 200 => return Ok(()), + Ok(_) | Err(_) if Instant::now() < deadline => tokio::time::sleep(Duration::from_millis(50)).await, + Ok(response) => bail!("agent package broker returned HTTP {}", response.status), + Err(error) => return Err(error).context("agent package broker did not become ready"), + } + } + } +} + +impl Drop for AgentHarness { + fn drop(&mut self) { + let _ = self.child.start_kill(); + } +} + +struct HttpResponse { + status: u16, + body: Vec, +} + +impl HttpResponse { + fn json(&self) -> anyhow::Result { + serde_json::from_slice(&self.body).context("response body is not valid JSON") + } +} + +pub(crate) async fn run() -> anyhow::Result<()> { + let agent_path = std::env::args_os() + .nth(1) + .map(PathBuf::from) + .context("usage: agent-policy-tester ")?; + ensure!( + agent_path.is_file(), + "agent executable does not exist: {}", + agent_path.display() + ); + + unavailable_policy_and_method_restrictions(&agent_path).await?; + complete_snapshots_across_reload(&agent_path).await?; + + Ok(()) +} + +async fn request(pipe_name: &str, method: &str, path: &str) -> anyhow::Result { + let deadline = Instant::now() + Duration::from_secs(10); + let mut pipe = loop { + match ClientOptions::new().open(pipe_name) { + Ok(pipe) => break pipe, + Err(_) if Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(25)).await; + } + Err(error) => return Err(error).with_context(|| format!("open named pipe {pipe_name}")), + } + }; + + let request = format!("{method} {path} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n"); + pipe.write_all(request.as_bytes()).await.context("write HTTP request")?; + pipe.flush().await.context("flush HTTP request")?; + + let mut raw_response = Vec::new(); + tokio::time::timeout(Duration::from_secs(10), pipe.read_to_end(&mut raw_response)) + .await + .context("timed out reading HTTP response")? + .context("read HTTP response")?; + + parse_response(raw_response) +} + +fn parse_response(raw_response: Vec) -> anyhow::Result { + let header_end = raw_response + .windows(4) + .position(|window| window == b"\r\n\r\n") + .context("HTTP response has no header terminator")?; + let headers = std::str::from_utf8(&raw_response[..header_end]).context("HTTP response headers are not UTF-8")?; + let status = headers + .lines() + .next() + .and_then(|line| line.split_ascii_whitespace().nth(1)) + .context("HTTP response has no status")? + .parse() + .context("HTTP response status is invalid")?; + + Ok(HttpResponse { + status, + body: raw_response[header_end + 4..].to_vec(), + }) +} + +fn full_policy() -> Value { + serde_json::from_str(FULL_POLICY).expect("sample policy is valid JSON") +} + +fn empty_policy() -> Value { + let mut policy = full_policy(); + policy["Metadata"]["Id"] = json!("tests.empty-policy"); + policy["Metadata"]["Revision"] = json!(1); + policy["Rules"] = json!([]); + policy +} + +fn secure_policy_file(path: &Path) -> anyhow::Result<()> { + let owner_status = std::process::Command::new("icacls.exe") + .arg(path) + .args(["/setowner", "*S-1-5-18"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .context("set policy owner")?; + ensure!( + owner_status.success(), + "setting the policy owner to LocalSystem failed; run the tester as LocalSystem" + ); + + let dacl_status = std::process::Command::new("icacls.exe") + .arg(path) + .args(["/inheritance:r", "/grant:r", "*S-1-5-18:(F)", "*S-1-5-32-544:(F)"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .context("set policy DACL")?; + ensure!( + dacl_status.success(), + "failed to set a system-and-administrators-only policy DACL" + ); + + Ok(()) +} + +async fn unavailable_policy_and_method_restrictions(agent_path: &Path) -> anyhow::Result<()> { + let agent = AgentHarness::start(agent_path, None).await?; + + for path in ["/v1/health", "/v1/capabilities"] { + let response = request(&agent.pipe_name, "GET", path).await?; + ensure!(response.status == 200, "{path} returned HTTP {}", response.status); + } + + let response = request(&agent.pipe_name, "GET", "/v1/policy").await?; + ensure!( + response.status == 503, + "unavailable policy returned HTTP {}", + response.status + ); + let error = response.json()?; + ensure!( + error["Code"] == "BrokerPaused", + "unexpected unavailable-policy error code" + ); + ensure!( + error["Message"] == "active policy is unavailable", + "unexpected unavailable-policy error message" + ); + ensure!( + error["Details"].is_null(), + "unavailable-policy error details are not null" + ); + ensure!( + error.get("Policy").is_none(), + "unavailable-policy response exposed a policy" + ); + + for method in ["POST", "PUT", "PATCH", "DELETE", "OPTIONS", "TRACE", "CONNECT"] { + let response = request(&agent.pipe_name, method, "/v1/policy").await?; + ensure!( + response.status == 405, + "{method} /v1/policy returned HTTP {}", + response.status + ); + } + + let response = request(&agent.pipe_name, "GET", "/v1/not-a-route").await?; + ensure!( + response.status == 404, + "unknown route returned HTTP {}", + response.status + ); + + Ok(()) +} + +async fn complete_snapshots_across_reload(agent_path: &Path) -> anyhow::Result<()> { + let empty = empty_policy(); + let agent = AgentHarness::start(agent_path, Some(&empty)).await?; + + let initial = request(&agent.pipe_name, "GET", "/v1/policy").await?; + ensure!(initial.status == 200, "active policy returned HTTP {}", initial.status); + let initial = initial.json()?; + ensure!( + initial["ResponseKind"] == "PolicyResponse", + "unexpected policy response kind" + ); + ensure!( + initial["ResponseVersion"] == "1.0", + "unexpected policy response version" + ); + ensure!( + initial["Server"]["Transport"] == "HttpNamedPipe", + "unexpected policy response transport" + ); + ensure!( + initial["Policy"] == empty, + "initial policy response does not match the empty policy" + ); + + let head = request(&agent.pipe_name, "HEAD", "/v1/policy").await?; + ensure!(head.status == 200, "HEAD /v1/policy returned HTTP {}", head.status); + ensure!(head.body.is_empty(), "HEAD /v1/policy returned a body"); + + let full = full_policy(); + let replacement_path = agent.policy_path.clone(); + let replacement = serde_json::to_vec_pretty(&full)?; + let replace = tokio::task::spawn_blocking(move || { + std::thread::sleep(Duration::from_millis(25)); + std::fs::write(replacement_path, replacement) + }); + + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let response = request(&agent.pipe_name, "GET", "/v1/policy").await?; + ensure!( + response.status == 200, + "policy reload returned HTTP {}", + response.status + ); + let response = response.json()?; + let policy = &response["Policy"]; + ensure!( + policy == &empty || policy == &full, + "response contained a partial policy snapshot" + ); + if policy == &full { + break; + } + ensure!(Instant::now() < deadline, "agent did not reload the policy"); + tokio::task::yield_now().await; + } + + replace + .await + .context("join policy replacement task")? + .context("replace policy")?; + + Ok(()) +} diff --git a/crates/now-package-broker/Cargo.toml b/crates/now-package-broker/Cargo.toml index 379c5024a..273b11d97 100644 --- a/crates/now-package-broker/Cargo.toml +++ b/crates/now-package-broker/Cargo.toml @@ -31,8 +31,8 @@ hyper = { version = "1", features = ["http1", "server"] } hyper-util = { version = "0.1", features = ["tokio", "server", "server-auto", "service"] } notify = { version = "7", default-features = false } now-policy = "0.2" -now-policy-api = { version = "0.3", features = ["policy-compat"] } -now-policy-server-template = { version = "0.3", features = ["policy-compat"] } +now-policy-api = "0.3" +now-policy-server-template = "0.3" parking_lot = "0.12" regex = "1" semver = "1" diff --git a/crates/now-package-broker/src/auth.rs b/crates/now-package-broker/src/auth.rs index bc770d149..65b5515d4 100644 --- a/crates/now-package-broker/src/auth.rs +++ b/crates/now-package-broker/src/auth.rs @@ -31,6 +31,10 @@ impl PipeClient { /// unauthenticated work a connection flood can trigger. pub(crate) fn from_connected_pipe(server: &NamedPipeServer) -> anyhow::Result { let process_id = connected_pipe_client_process_id(server).context("failed to query pipe client process id")?; + Self::from_process_id(process_id) + } + + fn from_process_id(process_id: u32) -> anyhow::Result { 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 @@ -50,6 +54,11 @@ impl PipeClient { }) } + #[cfg(test)] + pub(crate) fn from_current_process() -> anyhow::Result { + Self::from_process_id(std::process::id()) + } + /// Security identifier of the authenticated pipe client user, captured at connect. pub(crate) fn user_sid(&self) -> &Sid { &self.user_sid @@ -61,7 +70,7 @@ impl PipeClient { skip_signature_validation: bool, ) -> anyhow::Result<()> { self.validate_client_context(&request.client)?; - self.validate_signature(skip_signature_validation) + self.validate_connection(skip_signature_validation) } pub(crate) fn validate_status_request( @@ -70,7 +79,7 @@ impl PipeClient { skip_signature_validation: bool, ) -> anyhow::Result<()> { self.validate_client_context(&request.client)?; - self.validate_signature(skip_signature_validation) + self.validate_connection(skip_signature_validation) } pub(crate) fn validate_cancel_request( @@ -79,7 +88,7 @@ impl PipeClient { skip_signature_validation: bool, ) -> anyhow::Result<()> { self.validate_client_context(&request.client)?; - self.validate_signature(skip_signature_validation) + self.validate_connection(skip_signature_validation) } fn validate_client_context(&self, client: &ClientContext) -> anyhow::Result<()> { @@ -87,7 +96,7 @@ impl PipeClient { self.validate_executable_path(&client.client_executable_path) } - fn validate_signature(&self, skip_signature_validation: bool) -> anyhow::Result<()> { + pub(crate) fn validate_connection(&self, skip_signature_validation: bool) -> anyhow::Result<()> { if signature_validation_skipped(skip_signature_validation) { warn!("DEBUG MODE: Skipping package broker client signature validation"); return Ok(()); @@ -364,7 +373,7 @@ mod tests { user_sid: client_user_sid(), }; - assert!(client.validate_signature(true).is_err()); + assert!(client.validate_connection(true).is_err()); } } diff --git a/crates/now-package-broker/src/evaluator/matching.rs b/crates/now-package-broker/src/evaluator/matching.rs index 213ae8d08..b76566aa3 100644 --- a/crates/now-package-broker/src/evaluator/matching.rs +++ b/crates/now-package-broker/src/evaluator/matching.rs @@ -18,16 +18,16 @@ pub(super) fn rule_matches( ) -> bool { let m = &rule.match_criteria; - operations_match(request.operation.into(), &m.operations) - && managers_match(request.manager.into(), &m.managers) + operations_match(request.operation, &m.operations) + && managers_match(request.manager, &m.managers) && wildcard_any(&request.source.name, &m.sources) && wildcard_any(&request.package.id, &m.package_identifiers) && m.package_names.is_empty() && string_in_set(effective_version, &m.versions) && version_range_matches(effective_version, &m.version_range) - && scopes_match(request.options.scope.map(Into::into), &m.scopes) - && architectures_match(request.package.architecture.map(Into::into), &m.architectures) - && elevation_match(request.client.requested_elevation.into(), &m.elevation) + && scopes_match(request.options.scope, &m.scopes) + && architectures_match(request.package.architecture, &m.architectures) + && elevation_match(request.client.requested_elevation, &m.elevation) && bool_in_set(request.options.interactive, &m.interactive) && bool_in_set(request.options.skip_hash_check, &m.skip_hash_check) && bool_in_set(request.options.pre_release, &m.pre_release) @@ -39,30 +39,95 @@ pub(super) fn rule_matches( && constraints_pass(&rule.constraints, request, flags) } -fn operations_match(op: Operation, allowed: &BTreeSet) -> bool { - allowed.is_empty() || allowed.contains(&op) +fn policy_operation(operation: now_policy_api::Operation) -> Operation { + use now_policy_api as api; + + match operation { + api::Operation::Install => Operation::Install, + api::Operation::Update => Operation::Update, + api::Operation::Uninstall => Operation::Uninstall, + } +} + +fn policy_manager(manager: now_policy_api::ManagerName) -> ManagerName { + use now_policy_api as api; + + match manager { + api::ManagerName::Winget => ManagerName::Winget, + api::ManagerName::PowerShell => ManagerName::PowerShell, + api::ManagerName::PowerShell7 => ManagerName::PowerShell7, + api::ManagerName::Apt => ManagerName::Apt, + api::ManagerName::Bun => ManagerName::Bun, + api::ManagerName::Cargo => ManagerName::Cargo, + api::ManagerName::Chocolatey => ManagerName::Chocolatey, + api::ManagerName::Dnf => ManagerName::Dnf, + api::ManagerName::Dotnet => ManagerName::Dotnet, + api::ManagerName::Flatpak => ManagerName::Flatpak, + api::ManagerName::Homebrew => ManagerName::Homebrew, + api::ManagerName::Npm => ManagerName::Npm, + api::ManagerName::Pacman => ManagerName::Pacman, + api::ManagerName::Pip => ManagerName::Pip, + api::ManagerName::Scoop => ManagerName::Scoop, + api::ManagerName::Snap => ManagerName::Snap, + api::ManagerName::Vcpkg => ManagerName::Vcpkg, + } +} + +fn policy_scope(scope: now_policy_api::Scope) -> Scope { + use now_policy_api as api; + + match scope { + api::Scope::User => Scope::User, + api::Scope::Machine => Scope::Machine, + } +} + +fn policy_architecture(architecture: now_policy_api::Architecture) -> Architecture { + use now_policy_api as api; + + match architecture { + api::Architecture::X86 => Architecture::X86, + api::Architecture::X64 => Architecture::X64, + api::Architecture::Arm64 => Architecture::Arm64, + api::Architecture::Neutral => Architecture::Neutral, + } +} + +fn policy_elevation(elevation: now_policy_api::Elevation) -> Elevation { + use now_policy_api as api; + + match elevation { + api::Elevation::Standard => Elevation::Standard, + api::Elevation::Elevated => Elevation::Elevated, + } +} + +fn operations_match(operation: now_policy_api::Operation, allowed: &BTreeSet) -> bool { + allowed.is_empty() || allowed.contains(&policy_operation(operation)) } -fn managers_match(name: ManagerName, allowed: &BTreeSet) -> bool { - allowed.is_empty() || allowed.contains(&name) +fn managers_match(manager: now_policy_api::ManagerName, allowed: &BTreeSet) -> bool { + allowed.is_empty() || allowed.contains(&policy_manager(manager)) } -fn scopes_match(scope: Option, allowed: &BTreeSet) -> bool { +fn scopes_match(scope: Option, allowed: &BTreeSet) -> bool { if allowed.is_empty() { return true; } - scope.is_some_and(|s| allowed.contains(&s)) + scope.map(policy_scope).is_some_and(|scope| allowed.contains(&scope)) } -fn architectures_match(arch: Option, allowed: &BTreeSet) -> bool { +fn architectures_match(architecture: Option, allowed: &BTreeSet) -> bool { if allowed.is_empty() { return true; } - arch.is_some_and(|a| allowed.contains(&a)) + architecture + .map(policy_architecture) + .is_some_and(|architecture| allowed.contains(&architecture)) } -fn elevation_match(elev: Elevation, allowed: &BTreeSet) -> bool { - allowed.is_empty() || allowed.contains(&elev) +fn elevation_match(elevation: now_policy_api::Elevation, allowed: &BTreeSet) -> bool { + allowed.is_empty() || allowed.contains(&policy_elevation(elevation)) } fn bool_in_set(value: bool, set: &BTreeSet) -> bool { diff --git a/crates/now-package-broker/src/server/mod.rs b/crates/now-package-broker/src/server/mod.rs index 6f1e65b20..27d9b0c46 100644 --- a/crates/now-package-broker/src/server/mod.rs +++ b/crates/now-package-broker/src/server/mod.rs @@ -11,8 +11,8 @@ use now_policy_api::{ CancelRequest, CancelResponse, CancelResponseKind, CapabilitiesResponse, CapabilitiesResponseKind, Decision, DecisionInfo, Elevation, ErrorCode, ErrorResponse, EvaluationResponse, EvaluationResponseKind, ExecutionResponse, ExecutionResponseKind, HealthResponse, HealthResponseKind, HealthStatus, ManagerCapability, ManagerName, - OperationStatus, OperationSubmission, PackageRequest, Scope, StatusRequest, StatusResponse, StatusResponseKind, - Transport, + OperationStatus, OperationSubmission, PackageRequest, PolicyResponse, PolicyResponseKind, Scope, StatusRequest, + StatusResponse, StatusResponseKind, Transport, }; use now_policy_server_template::{MAX_REQUEST_BODY_BYTES, PackageBrokerServer, SharedPackageBrokerServer}; use tracing::{info, trace, warn}; @@ -115,6 +115,17 @@ impl PackageBrokerServer for BrokerConnection { self.state.capabilities(self.client.user_sid()).await } + async fn active_policy(&self) -> Result { + self.client + .validate_connection(self.state.skip_signature_validation) + .map_err(|error| { + warn!(error = format!("{error:#}"), "Rejected package broker policy request"); + error_response(ErrorCode::Unauthorized, "pipe client authentication failed") + })?; + + self.state.policy_response() + } + async fn evaluate(&self, request: PackageRequest) -> Result { self.client .validate_request(&request, self.state.skip_signature_validation) @@ -163,6 +174,25 @@ impl PackageBrokerServer for BrokerConnection { } impl BrokerState { + fn active_policy(&self) -> Result, ErrorResponse> { + let guard = self.policy.read().expect("policy lock poisoned"); + guard + .as_ref() + .map(Arc::clone) + .ok_or_else(|| error_response(ErrorCode::BrokerPaused, "active policy is unavailable")) + } + + fn policy_response(&self) -> Result { + let policy = self.active_policy()?; + + Ok(PolicyResponse { + response_kind: PolicyResponseKind, + response_version: api_version(), + server: server_context(), + policy: (*policy).clone(), + }) + } + async fn health(&self) -> HealthResponse { let policy_guard = self.policy.read().expect("policy lock poisoned"); let (status, policy_id) = match policy_guard.as_ref() { @@ -419,18 +449,7 @@ impl BrokerState { } let received_at = Utc::now(); - let policy = { - let guard = self.policy.read().expect("policy lock poisoned"); - match guard.as_ref() { - Some(policy) => Arc::clone(policy), - None => { - return Err(error_response( - ErrorCode::BrokerPaused, - "policy file is unavailable or corrupted", - )); - } - } - }; + let policy = self.active_policy()?; if let Some(reason) = policy_validity_failure(&policy, received_at) { warn!(%reason, "Rejecting request: policy outside validity window"); @@ -458,7 +477,10 @@ impl BrokerState { let effective_decision = if audit_mode { Decision::Allow } else { - decision.decision.into() + match decision.decision { + now_policy::Decision::Allow => Decision::Allow, + now_policy::Decision::Deny => Decision::Deny, + } }; let reason = if audit_mode && decision.decision != now_policy::Decision::Allow { @@ -521,12 +543,15 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; + use axum::body::{Body, to_bytes}; + use axum::http::{Method, Request, StatusCode}; use chrono::Utc; use now_policy::{ PackageBrokerPolicy, PolicyEnforcement, PolicyMetadata, PolicySchemaUri, ResourceId, RulePrecedence, SemanticVersion, }; use now_policy_api as api; + use tower_service::Service as _; use super::*; use crate::executor::{ExecutionOutput, OperationCanceled, ProcessStartedCallback}; @@ -601,6 +626,102 @@ mod tests { } } + fn shared_state(policy: Option) -> Arc { + let mut state = state(); + state.policy = RwLock::new(policy.map(Arc::new)); + Arc::new(state) + } + + async fn route_request(state: Arc, method: Method, uri: &str) -> axum::response::Response { + let client = PipeClient::from_current_process().expect("capture current test process"); + let mut router = build_router_for_client(state, client); + router + .call( + Request::builder() + .method(method) + .uri(uri) + .body(Body::empty()) + .expect("valid test request"), + ) + .await + .expect("router is infallible") + } + + async fn response_json(response: axum::response::Response) -> serde_json::Value { + let body = to_bytes(response.into_body(), usize::MAX) + .await + .expect("read response body"); + serde_json::from_slice(&body).expect("response is valid JSON") + } + + #[tokio::test] + async fn policy_route_rejects_unsigned_client() { + let mut state = state(); + state.skip_signature_validation = false; + let response = route_request(Arc::new(state), Method::GET, "/v1/policy").await; + + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + + let body = response_json(response).await; + let error: ErrorResponse = serde_json::from_value(body.clone()).expect("deserialize error response"); + assert_eq!(error.code, ErrorCode::Unauthorized); + assert_eq!(error.message, "pipe client authentication failed"); + assert!(body.get("Policy").is_none()); + } + + #[test] + fn concurrent_policy_replacement_returns_only_complete_snapshots() { + let policy_a = permissive_policy(); + let mut policy_b = + now_policy::schema::parse_policy_json(include_str!("../assets/samples/corporate-allowlist.policy.json")) + .expect("sample policy is valid"); + policy_b.metadata.id = ResourceId::from("replacement-policy"); + policy_b.metadata.revision = 42; + + let current_policy_json = serde_json::to_value(&policy_a).unwrap(); + let replacement_policy_json = serde_json::to_value(&policy_b).unwrap(); + let policy_a = Arc::new(policy_a); + let policy_b = Arc::new(policy_b); + let state = shared_state(None); + *state.policy.write().expect("policy lock") = Some(Arc::clone(&policy_a)); + + const READER_COUNT: usize = 4; + const ITERATIONS: usize = 1_000; + let barrier = Arc::new(std::sync::Barrier::new(READER_COUNT + 1)); + + std::thread::scope(|scope| { + for _ in 0..READER_COUNT { + let state = Arc::clone(&state); + let barrier = Arc::clone(&barrier); + let current_policy_json = ¤t_policy_json; + let replacement_policy_json = &replacement_policy_json; + scope.spawn(move || { + barrier.wait(); + for _ in 0..ITERATIONS { + let response = state.policy_response().expect("active policy response"); + let actual = serde_json::to_value(response.policy).unwrap(); + assert!( + actual == *current_policy_json || actual == *replacement_policy_json, + "response mixed two policy snapshots" + ); + std::thread::yield_now(); + } + }); + } + + barrier.wait(); + for index in 0..ITERATIONS { + let replacement = if index % 2 == 0 { + Arc::clone(&policy_b) + } else { + Arc::clone(&policy_a) + }; + *state.policy.write().expect("policy lock") = Some(replacement); + std::thread::yield_now(); + } + }); + } + fn request() -> PackageRequest { PackageRequest { request_kind: api::PackageRequestKind, diff --git a/crates/now-package-broker/src/server/responses.rs b/crates/now-package-broker/src/server/responses.rs index 3899f8759..02758756f 100644 --- a/crates/now-package-broker/src/server/responses.rs +++ b/crates/now-package-broker/src/server/responses.rs @@ -5,7 +5,7 @@ use now_policy::PolicyDocument; use now_policy_api::{ API_VERSION_STR, ApiVersion, Architecture, ErrorCode, ErrorResponse, ErrorResponseKind, ManagerCapability, ManagerName, Operation, OperationDiagnostics, PackageRequest, RequestSummary, ResourceId, ResponsePolicyInfo, - RuleId, Scope, ServerContext, Transport, + RuleId, Scope, SemanticVersion, ServerContext, Transport, }; use crate::operation_tracker::OperationTracker; @@ -182,9 +182,9 @@ pub(super) fn request_summary(request: &PackageRequest) -> RequestSummary { pub(super) fn policy_info(policy: &PolicyDocument) -> ResponsePolicyInfo { ResponsePolicyInfo { - id: policy.metadata.id.clone().into(), + id: ResourceId(policy.metadata.id.0.clone()), revision: policy.metadata.revision, - policy_version: policy.policy_version.clone().into(), + policy_version: SemanticVersion(policy.policy_version.0.clone()), } }