diff --git a/Cargo.lock b/Cargo.lock index 5207fce6f..acc234f64 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2909,7 +2909,16 @@ version = "0.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4aaa26c720c68b866f2c96ef5c1264b3e6f473fe5d4ce61cd44bbe913e553018" dependencies = [ - "hmac", + "hmac 0.13.0", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest 0.10.7", ] [[package]] @@ -4802,6 +4811,7 @@ dependencies = [ "devolutions-agent-shared", "devolutions-gateway-task", "hex", + "hmac 0.12.1", "hyper 1.10.1", "hyper-util", "notify 7.0.0", @@ -5189,7 +5199,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112d82ceb8c5bf524d9af484d4e4970c9fd5a0cc15ba14ad93dccd28873b0629" dependencies = [ "digest 0.11.3", - "hmac", + "hmac 0.13.0", ] [[package]] @@ -5339,7 +5349,7 @@ dependencies = [ "ecdsa", "ed25519-dalek", "hex", - "hmac", + "hmac 0.13.0", "http 1.4.2", "inout 0.2.2", "md-5 0.11.0", @@ -5423,7 +5433,7 @@ dependencies = [ "cipher 0.5.2", "crypto-bigint", "des", - "hmac", + "hmac 0.13.0", "inout 0.2.2", "oid", "pbkdf2", @@ -5637,7 +5647,7 @@ dependencies = [ "byteorder", "bytes 1.12.1", "fallible-iterator 0.2.0", - "hmac", + "hmac 0.13.0", "md-5 0.11.0", "memchr", "rand 0.10.2", @@ -6418,7 +6428,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9935425142ac6e252364413291d96c8bc9898d0876a801824c7af4eae397b689" dependencies = [ "ctutils", - "hmac", + "hmac 0.13.0", ] [[package]] @@ -7325,7 +7335,7 @@ dependencies = [ "ed25519-dalek", "futures", "getrandom 0.3.4", - "hmac", + "hmac 0.13.0", "md-5 0.11.0", "md4", "num-derive", diff --git a/crates/agent-policy-tester/src/windows.rs b/crates/agent-policy-tester/src/windows.rs index eb0bb00c4..82eb6b4bf 100644 --- a/crates/agent-policy-tester/src/windows.rs +++ b/crates/agent-policy-tester/src/windows.rs @@ -18,7 +18,7 @@ struct AgentHarness { impl AgentHarness { async fn start(agent_path: &Path, policy: Option<&Value>) -> anyhow::Result { - let data_dir = tempfile::tempdir().context("create Agent data directory")?; + let data_dir = create_data_dir()?; let pipe_name = format!( r"\\.\pipe\Devolutions.Now.PackageBroker.tests.{}.{}", std::process::id(), @@ -28,9 +28,18 @@ impl AgentHarness { if let Some(policy) = policy { std::fs::write(&policy_path, serde_json::to_vec_pretty(policy)?).context("write policy")?; - secure_policy_file(&policy_path)?; + secure_policy_path(&policy_path, false)?; } + Self::start_with_path(agent_path, data_dir, pipe_name, policy_path).await + } + + async fn start_with_path( + agent_path: &Path, + data_dir: tempfile::TempDir, + pipe_name: String, + policy_path: PathBuf, + ) -> anyhow::Result { let config = json!({ "PackageBroker": { "Enabled": true, @@ -112,11 +121,23 @@ pub(crate) async fn run() -> anyhow::Result<()> { unavailable_policy_and_method_restrictions(&agent_path).await?; complete_snapshots_across_reload(&agent_path).await?; + redirected_policy_paths_fail_closed(&agent_path).await?; + management_write_tokens_survive_watcher_reload(&agent_path).await?; Ok(()) } async fn request(pipe_name: &str, method: &str, path: &str) -> anyhow::Result { + request_with_body(pipe_name, method, path, None, &[]).await +} + +async fn request_with_body( + pipe_name: &str, + method: &str, + path: &str, + content_type: Option<&str>, + body: &[u8], +) -> anyhow::Result { let deadline = Instant::now() + Duration::from_secs(10); let mut pipe = loop { match ClientOptions::new().open(pipe_name) { @@ -128,8 +149,14 @@ async fn request(pipe_name: &str, method: &str, path: &str) -> anyhow::Result Value { policy } -fn secure_policy_file(path: &Path) -> anyhow::Result<()> { +fn policy_draft(id: &str, publisher: &str) -> Value { + json!({ + "$schema": "https://devolutions.net/schemas/now-policy-draft.schema.1.0.json", + "PolicyVersion": "1.0.0", + "PolicyType": "PackageBrokerPolicy", + "Metadata": { "Id": id, "Publisher": publisher }, + "Enforcement": { "DefaultDecision": "Deny", "RulePrecedence": "PriorityThenDeny" }, + "Rules": [] + }) +} + +fn create_data_dir() -> anyhow::Result { + let program_data = std::env::var_os("ProgramData").context("ProgramData is not defined")?; + let data_dir = tempfile::Builder::new() + .prefix("dgw-agent-policy-") + .tempdir_in(program_data) + .context("create Agent data directory")?; + secure_policy_path(data_dir.path(), true)?; + Ok(data_dir) +} + +fn secure_policy_path(path: &Path, directory: bool) -> anyhow::Result<()> { let owner_status = std::process::Command::new("icacls.exe") .arg(path) .args(["/setowner", "*S-1-5-18"]) @@ -183,24 +231,238 @@ fn secure_policy_file(path: &Path) -> anyhow::Result<()> { .context("set policy owner")?; ensure!( owner_status.success(), - "setting the policy owner to LocalSystem failed; run the tester as LocalSystem" + "setting the policy path owner to LocalSystem failed; run the tester as LocalSystem" ); + let system_grant = if directory { + "*S-1-5-18:(OI)(CI)(F)" + } else { + "*S-1-5-18:(F)" + }; + let administrators_grant = if directory { + "*S-1-5-32-544:(OI)(CI)(F)" + } else { + "*S-1-5-32-544:(F)" + }; 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)"]) + .args(["/inheritance:r", "/grant:r", system_grant, administrators_grant]) .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" + "failed to set a system-and-administrators-only policy path DACL" ); Ok(()) } +fn grant_users_full_control(path: &Path) -> anyhow::Result<()> { + let status = std::process::Command::new("icacls.exe") + .arg(path) + .args(["/grant:r", "*S-1-5-32-545:(OI)(CI)(F)"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .context("grant Users control of test directory")?; + ensure!(status.success(), "failed to make test directory user-controlled"); + Ok(()) +} + +fn create_junction(link: &Path, target: &Path) -> anyhow::Result<()> { + let status = std::process::Command::new("cmd.exe") + .args(["/d", "/c", "mklink", "/J"]) + .arg(link) + .arg(target) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .context("create test junction")?; + ensure!(status.success(), "failed to create test junction"); + Ok(()) +} + +async fn assert_redirected_policy_rejected( + agent_path: &Path, + data_dir: tempfile::TempDir, + policy_path: PathBuf, +) -> anyhow::Result<()> { + let pipe_name = format!( + r"\\.\pipe\Devolutions.Now.PackageBroker.tests.{}.{}", + std::process::id(), + fastrand::u64(..) + ); + let agent = AgentHarness::start_with_path(agent_path, data_dir, pipe_name, policy_path).await?; + let management = request(&agent.pipe_name, "GET", "/v1/policy/management") + .await? + .json()?; + ensure!(management["Management"]["State"] == "Invalid"); + ensure!(management["Management"]["WriteCapability"] == "ReadOnly"); + ensure!(management["Management"]["ReadOnlyReason"] == "UnsafePath"); + ensure!(request(&agent.pipe_name, "GET", "/v1/policy").await?.status == 404); + + let replacement = json!({ + "RequestKind": "PolicyReplacementRequest", + "RequestVersion": "1.0", + "ExpectedStoreToken": management["Management"]["StoreToken"], + "Operation": "Repair", + "ConflictHandling": "Reject", + "WarningsAcknowledged": false, + "Draft": full_policy(), + "ValidationReceipt": "invalid" + }); + let response = request_with_body( + &agent.pipe_name, + "PUT", + "/v1/policy", + Some("application/json"), + &serde_json::to_vec(&replacement)?, + ) + .await?; + ensure!(response.json()?["Code"] == "UnsafePolicyPath"); + Ok(()) +} + +async fn redirected_policy_paths_fail_closed(agent_path: &Path) -> anyhow::Result<()> { + let data_dir = create_data_dir()?; + let final_dir = data_dir.path().join("trusted-final"); + let unsafe_dir = data_dir.path().join("unsafe-hop"); + std::fs::create_dir(&final_dir)?; + std::fs::create_dir(&unsafe_dir)?; + grant_users_full_control(&unsafe_dir)?; + let policy = final_dir.join("policy.json"); + std::fs::write(&policy, serde_json::to_vec_pretty(&empty_policy())?)?; + secure_policy_path(&policy, false)?; + create_junction(&unsafe_dir.join("hop"), &final_dir)?; + let outer = data_dir.path().join("PolicyLink"); + create_junction(&outer, &unsafe_dir)?; + let redirected = outer.join("hop").join("policy.json"); + assert_redirected_policy_rejected(agent_path, data_dir, redirected).await?; + + let data_dir = create_data_dir()?; + let unsafe_dir = data_dir.path().join("unsafe-hop"); + std::fs::create_dir(&unsafe_dir)?; + grant_users_full_control(&unsafe_dir)?; + let target = unsafe_dir.join("policy.json"); + std::fs::write(&target, serde_json::to_vec_pretty(&empty_policy())?)?; + secure_policy_path(&target, false)?; + let redirected = data_dir.path().join("policy.json"); + std::os::windows::fs::symlink_file(&target, &redirected).context("create test policy symlink")?; + assert_redirected_policy_rejected(agent_path, data_dir, redirected).await +} + +async fn policy_management(agent: &AgentHarness) -> anyhow::Result { + let response = request(&agent.pipe_name, "GET", "/v1/policy/management").await?; + ensure!( + response.status == 200, + "GET /v1/policy/management returned HTTP {}", + response.status + ); + Ok(response.json()?["Management"].clone()) +} + +async fn replace_policy( + agent: &AgentHarness, + operation: &str, + expected_store_token: Value, + draft: Value, +) -> anyhow::Result { + let validation_request = json!({ + "RequestKind": "PolicyValidationRequest", + "RequestVersion": "1.0", + "Draft": draft + }); + let validation_response = request_with_body( + &agent.pipe_name, + "POST", + "/v1/policy/validate", + Some("application/json"), + &serde_json::to_vec(&validation_request)?, + ) + .await?; + ensure!( + validation_response.status == 200, + "POST /v1/policy/validate returned HTTP {}", + validation_response.status + ); + let validation = validation_response.json()?["Validation"].clone(); + ensure!(validation["IsValid"] == true, "policy validation failed"); + let replacement_request = json!({ + "RequestKind": "PolicyReplacementRequest", + "RequestVersion": "1.0", + "ExpectedStoreToken": expected_store_token, + "Operation": operation, + "ConflictHandling": "Reject", + "WarningsAcknowledged": true, + "Draft": validation["CanonicalDraft"], + "ValidationReceipt": validation["ValidationReceipt"] + }); + let response = request_with_body( + &agent.pipe_name, + "PUT", + "/v1/policy", + Some("application/json"), + &serde_json::to_vec(&replacement_request)?, + ) + .await?; + ensure!(response.status == 200, "{operation} returned HTTP {}", response.status); + response.json() +} + +async fn management_write_tokens_survive_watcher_reload(agent_path: &Path) -> anyhow::Result<()> { + for verbatim in [false, true] { + let data_dir = create_data_dir()?; + let policy_path = data_dir.path().join("policy.json"); + let policy_path = if verbatim { + PathBuf::from(format!(r"\\?\{}", policy_path.display())) + } else { + policy_path + }; + let pipe_name = format!( + r"\\.\pipe\Devolutions.Now.PackageBroker.tests.{}.{}", + std::process::id(), + fastrand::u64(..) + ); + let agent = AgentHarness::start_with_path(agent_path, data_dir, pipe_name, policy_path).await?; + let initial = policy_management(&agent).await?; + ensure!(initial["State"] == "Missing"); + ensure!(initial["WriteCapability"] == "Writable"); + + let created = replace_policy( + &agent, + "Create", + initial["StoreToken"].clone(), + policy_draft("tests.managed-write", "Test"), + ) + .await?; + ensure!(created["Policy"]["Metadata"]["Revision"] == 1); + let created_token = created["Management"]["StoreToken"].clone(); + tokio::time::sleep(Duration::from_secs(2)).await; + ensure!( + policy_management(&agent).await?["StoreToken"] == created_token, + "watcher reload rotated the Create token (verbatim={verbatim})" + ); + + let updated = replace_policy( + &agent, + "Update", + created_token, + policy_draft("tests.managed-write", "Updated Test"), + ) + .await?; + ensure!(updated["Policy"]["Metadata"]["Revision"] == 2); + let updated_token = updated["Management"]["StoreToken"].clone(); + tokio::time::sleep(Duration::from_secs(2)).await; + ensure!( + policy_management(&agent).await?["StoreToken"] == updated_token, + "watcher reload rotated the Update token (verbatim={verbatim})" + ); + } + Ok(()) +} + async fn unavailable_policy_and_method_restrictions(agent_path: &Path) -> anyhow::Result<()> { let agent = AgentHarness::start(agent_path, None).await?; @@ -230,7 +492,7 @@ async fn unavailable_policy_and_method_restrictions(agent_path: &Path) -> anyhow "unavailable-policy response exposed a policy" ); - for method in ["POST", "PUT", "PATCH", "DELETE", "OPTIONS", "TRACE", "CONNECT"] { + for method in ["POST", "PATCH", "DELETE", "OPTIONS", "TRACE", "CONNECT"] { let response = request(&agent.pipe_name, method, "/v1/policy").await?; ensure!( response.status == 405, @@ -239,10 +501,36 @@ async fn unavailable_policy_and_method_restrictions(agent_path: &Path) -> anyhow ); } - for (method, path) in [("GET", "/v1/policy/management"), ("POST", "/v1/policy/validate")] { + let management = request(&agent.pipe_name, "GET", "/v1/policy/management").await?; + ensure!( + management.status == 200, + "GET /v1/policy/management returned HTTP {}", + management.status + ); + ensure!(management.json()?["Management"]["State"] == "Missing"); + + for (method, path) in [("POST", "/v1/policy/validate"), ("PUT", "/v1/policy")] { let response = request(&agent.pipe_name, method, path).await?; ensure!( - response.status == 404, + response.status == 415, + "{method} {path} returned HTTP {}", + response.status + ); + ensure!(response.json()?["Code"] == "UnsupportedMediaType"); + + let response = request_with_body(&agent.pipe_name, method, path, Some("application/json"), b"{}").await?; + ensure!( + response.status == 400, + "malformed {method} {path} returned HTTP {}", + response.status + ); + ensure!(response.json()?["Code"] == "MalformedDraft"); + } + + for (method, path) in [("POST", "/v1/policy/management"), ("GET", "/v1/policy/validate")] { + let response = request(&agent.pipe_name, method, path).await?; + ensure!( + response.status == 405, "{method} {path} returned HTTP {}", response.status ); @@ -261,6 +549,7 @@ async fn unavailable_policy_and_method_restrictions(agent_path: &Path) -> anyhow 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_token = policy_management(&agent).await?["StoreToken"].clone(); let initial = request(&agent.pipe_name, "GET", "/v1/policy").await?; ensure!(initial.status == 200, "active policy returned HTTP {}", initial.status); @@ -314,6 +603,10 @@ async fn complete_snapshots_across_reload(agent_path: &Path) -> anyhow::Result<( ensure!(Instant::now() < deadline, "agent did not reload the policy"); tokio::task::yield_now().await; } + ensure!( + policy_management(&agent).await?["StoreToken"] != initial_token, + "external policy replacement did not rotate the store token" + ); replace .await diff --git a/crates/now-package-broker/Cargo.toml b/crates/now-package-broker/Cargo.toml index 134e15649..3c31102f8 100644 --- a/crates/now-package-broker/Cargo.toml +++ b/crates/now-package-broker/Cargo.toml @@ -27,6 +27,7 @@ chrono = { version = "0.4", features = ["serde"] } devolutions-agent-shared = { path = "../devolutions-agent-shared" } devolutions-gateway-task = { path = "../devolutions-gateway-task" } hex = "0.4" +hmac = "0.12" hyper = { version = "1", features = ["http1", "server"] } hyper-util = { version = "0.1", features = ["tokio", "server", "server-auto", "service"] } notify = { version = "7", default-features = false } @@ -50,6 +51,7 @@ win-api-wrappers = { path = "../win-api-wrappers" } version = "0.61" features = [ "Win32_Foundation", + "Win32_Globalization", "Win32_Security", "Win32_Security_Authorization", "Win32_Storage_FileSystem", diff --git a/crates/now-package-broker/src/auth.rs b/crates/now-package-broker/src/auth.rs index 65b5515d4..c5126dcb0 100644 --- a/crates/now-package-broker/src/auth.rs +++ b/crates/now-package-broker/src/auth.rs @@ -11,7 +11,7 @@ use widestring::U16CString; use win_api_wrappers::identity::account::lookup_account_by_name; use win_api_wrappers::identity::sid::Sid; use win_api_wrappers::process::Process; -use windows::Win32::Security::TOKEN_QUERY; +use windows::Win32::Security::{TOKEN_DUPLICATE, TOKEN_QUERY, WinBuiltinAdministratorsSid}; use windows::Win32::Storage::FileSystem::FILE_ID_INFO; use windows::Win32::System::Threading::PROCESS_QUERY_LIMITED_INFORMATION; @@ -21,6 +21,10 @@ pub(crate) struct PipeClient { executable_path: PathBuf, /// Security identifier of the pipe client process token user, captured at connect. user_sid: Sid, + /// Actual connected process token elevation, captured at connect. + is_elevated: bool, + /// Enabled built-in Administrators membership, captured at connect. + is_administrator: bool, } impl PipeClient { @@ -40,17 +44,28 @@ impl PipeClient { let executable_path = process .exe_path() .with_context(|| format!("failed to query pipe client process {process_id} executable path"))?; - let user_sid = process - .token(TOKEN_QUERY) - .with_context(|| format!("failed to open pipe client process {process_id} token"))? + let token = process + .token(TOKEN_QUERY | TOKEN_DUPLICATE) + .with_context(|| format!("failed to open pipe client process {process_id} token"))?; + let user_sid = token .sid_and_attributes() .with_context(|| format!("failed to query pipe client process {process_id} token user"))? .sid; + let is_elevated = token + .is_elevated() + .with_context(|| format!("failed to query pipe client process {process_id} token elevation"))?; + let administrators_sid = + Sid::from_well_known(WinBuiltinAdministratorsSid, None).context("resolve Administrators SID")?; + let is_administrator = token + .is_member(&administrators_sid) + .with_context(|| format!("failed to query pipe client process {process_id} Administrators membership"))?; Ok(Self { process_id, executable_path, user_sid, + is_elevated, + is_administrator, }) } @@ -59,11 +74,23 @@ impl PipeClient { Self::from_process_id(std::process::id()) } + #[cfg(all(test, feature = "dev-skip-broker-signature"))] + pub(crate) fn test_with_authority(is_elevated: bool, is_administrator: bool) -> anyhow::Result { + let mut client = Self::from_current_process()?; + client.is_elevated = is_elevated; + client.is_administrator = is_administrator; + Ok(client) + } + /// Security identifier of the authenticated pipe client user, captured at connect. pub(crate) fn user_sid(&self) -> &Sid { &self.user_sid } + pub(crate) fn is_elevated_administrator(&self) -> bool { + self.is_elevated && self.is_administrator + } + pub(crate) fn validate_request( &self, request: &PackageRequest, @@ -280,6 +307,8 @@ mod tests { process_id: 0, executable_path: PathBuf::new(), user_sid: system_sid(), + is_elevated: true, + is_administrator: true, } } @@ -371,6 +400,8 @@ mod tests { process_id: std::process::id(), executable_path: std::env::current_exe().expect("current test executable path"), user_sid: client_user_sid(), + is_elevated: false, + is_administrator: false, }; assert!(client.validate_connection(true).is_err()); diff --git a/crates/now-package-broker/src/lib.rs b/crates/now-package-broker/src/lib.rs index b55c994db..e1542dda4 100644 --- a/crates/now-package-broker/src/lib.rs +++ b/crates/now-package-broker/src/lib.rs @@ -24,6 +24,8 @@ pub mod policy_loader; #[cfg(windows)] mod policy_security; #[cfg(windows)] +pub mod policy_store; +#[cfg(windows)] pub mod policy_watcher; #[cfg(windows)] pub mod server; diff --git a/crates/now-package-broker/src/policy_security.rs b/crates/now-package-broker/src/policy_security.rs index 4ebb290d6..a9f38e915 100644 --- a/crates/now-package-broker/src/policy_security.rs +++ b/crates/now-package-broker/src/policy_security.rs @@ -30,13 +30,16 @@ use std::ffi::OsString; use std::fs::{File, OpenOptions}; -use std::os::windows::ffi::OsStringExt as _; +use std::mem::size_of; +use std::os::windows::ffi::{OsStrExt as _, OsStringExt as _}; use std::os::windows::fs::OpenOptionsExt as _; use std::os::windows::io::AsRawHandle as _; use std::path::{Path, PathBuf}; use anyhow::{Context as _, bail}; +use sha2::{Digest as _, Sha256}; use windows::Win32::Foundation::{ERROR_SUCCESS, GENERIC_ALL, GENERIC_WRITE, HANDLE, HLOCAL, LocalFree}; +use windows::Win32::Globalization::{CSTR_EQUAL, CompareStringOrdinal}; use windows::Win32::Security::Authorization::{ConvertSidToStringSidW, GetSecurityInfo, SE_FILE_OBJECT}; use windows::Win32::Security::{ ACCESS_ALLOWED_ACE, ACE_HEADER, ACL, DACL_SECURITY_INFORMATION, GetAce, INHERIT_ONLY_ACE, IsWellKnownSid, @@ -44,9 +47,10 @@ use windows::Win32::Security::{ WinLocalSystemSid, }; use windows::Win32::Storage::FileSystem::{ - DELETE, FILE_APPEND_DATA, FILE_DELETE_CHILD, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, - FILE_NAME_NORMALIZED, FILE_READ_ATTRIBUTES, FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, - FILE_WRITE_ATTRIBUTES, FILE_WRITE_DATA, FILE_WRITE_EA, GetFinalPathNameByHandleW, READ_CONTROL, WRITE_DAC, + DELETE, FILE_APPEND_DATA, FILE_ATTRIBUTE_REPARSE_POINT, FILE_ATTRIBUTE_TAG_INFO, FILE_DELETE_CHILD, + FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_NAME_NORMALIZED, FILE_READ_ATTRIBUTES, + FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, FILE_WRITE_ATTRIBUTES, FILE_WRITE_DATA, FILE_WRITE_EA, + FileAttributeTagInfo, GetFileInformationByHandleEx, GetFinalPathNameByHandleW, READ_CONTROL, WRITE_DAC, WRITE_OWNER, }; use windows::core::PWSTR; @@ -73,6 +77,9 @@ const DIRECTORY_TAMPER_MASK: u32 = FILE_DELETE_CHILD.0 /* delete or rename child | WRITE_OWNER.0 /* take ownership */ | GENERIC_ALL.0; /* full control */ +/// Additional rights that allow retargeting an ancestor that is itself a reparse point. +const REPARSE_POINT_TAMPER_MASK: u32 = FILE_WRITE_DATA.0 | GENERIC_WRITE.0; + /// Access rights on the directory hosting a verified executable that allow tampering with /// its execution. On top of [`DIRECTORY_TAMPER_MASK`], create rights are rejected: a /// principal able to add entries beside the executable can plant a DLL or another @@ -102,6 +109,8 @@ enum TrustedWriters { /// it is a low-privilege shared service identity, and accepting it for elevated /// executables would open a privilege-escalation path. AdminOrTrustedInstaller, + /// Policy-path ancestors may be controlled by the policy writers or TrustedInstaller. + PolicyAncestor, } // ACE type constants from winnt.h (the Win32_System_SystemServices feature is not enabled). @@ -166,6 +175,102 @@ pub(crate) fn verify_policy_file_security(file: &File) -> anyhow::Result<()> { verify_handle_security(file, "policy file", TrustedWriters::AdminOnly, WRITE_ACCESS_MASK) } +/// Verify that the directory hosting a managed policy is not writable by untrusted principals. +pub(crate) fn verify_policy_directory_security(directory: &File) -> anyhow::Result<()> { + verify_handle_security( + directory, + "policy directory", + TrustedWriters::AdminOnly, + PARENT_DIRECTORY_TAMPER_MASK, + ) +} + +/// Verify every lexical ancestor of a managed policy path. +pub(crate) fn verify_policy_path_ancestors(path: &Path) -> anyhow::Result<()> { + let subject = format!("policy file '{}'", path.display()); + verify_directory_chain( + path.parent(), + &subject, + TrustedWriters::AdminOnly, + TrustedWriters::PolicyAncestor, + true, + ) +} + +/// Verify that an opened policy file is not a reparse point and resolves to the validated path. +pub(crate) fn verify_policy_file_path(file: &File, path: &Path) -> anyhow::Result<()> { + if is_reparse_point(file)? { + bail!("policy file '{}' is a reparse point", path.display()); + } + let final_path = final_path_from_handle(file)?; + if !windows_paths_equal( + &final_path, + &dos_path_from_wide(&path.as_os_str().encode_wide().collect::>()), + ) { + bail!("policy file '{}' resolved outside the validated path", path.display()); + } + Ok(()) +} + +/// Compare Windows paths using the operating system's ordinal case folding. +pub(crate) fn windows_paths_equal(left: &Path, right: &Path) -> bool { + let left: Vec = left.as_os_str().encode_wide().collect(); + let right: Vec = right.as_os_str().encode_wide().collect(); + // SAFETY: Both slices contain valid, initialized UTF-16 code units. + unsafe { CompareStringOrdinal(&left, &right, true) == CSTR_EQUAL } +} + +/// Digest verified owner and DACL state for opaque policy-store fingerprints. +pub(crate) fn security_state_digest(file: &File) -> anyhow::Result<[u8; 32]> { + let handle = HANDLE(file.as_raw_handle()); + let mut owner = PSID::default(); + let mut dacl: *mut ACL = std::ptr::null_mut(); + let mut descriptor = OwnedSecurityDescriptor(PSECURITY_DESCRIPTOR::default()); + + // SAFETY: The handle and all security-information output pointers are valid. + let status = unsafe { + GetSecurityInfo( + handle, + SE_FILE_OBJECT, + OWNER_SECURITY_INFORMATION | DACL_SECURITY_INFORMATION, + Some(&mut owner), + None, + Some(&mut dacl), + None, + Some(&mut descriptor.0), + ) + }; + if status != ERROR_SUCCESS { + bail!("failed to read policy security state: error {}", status.0); + } + + let mut hasher = Sha256::new(); + if owner.0.is_null() { + hasher.update(b"no-owner"); + } else { + // SAFETY: The owner SID points into the live security descriptor. + let owner = unsafe { sid_to_string(owner) }; + hasher.update(owner.as_bytes()); + } + if dacl.is_null() { + hasher.update(b"null-dacl"); + } else { + // SAFETY: The DACL points into the live security descriptor. + let ace_count = u32::from(unsafe { (*dacl).AceCount }); + hasher.update(ace_count.to_le_bytes()); + for index in 0..ace_count { + let mut ace: *mut core::ffi::c_void = std::ptr::null_mut(); + // SAFETY: The index is within the DACL's reported ACE count. + unsafe { GetAce(dacl, index, &mut ace) }.context("failed to read policy DACL entry")?; + // SAFETY: GetAce returned a complete ACE beginning with ACE_HEADER. + let size = usize::from(unsafe { (*ace.cast::()).AceSize }); + // SAFETY: AceSize bounds the complete ACE within the validated ACL. + hasher.update(unsafe { std::slice::from_raw_parts(ace.cast::(), size) }); + } + } + Ok(hasher.finalize().into()) +} + /// A package-manager executable that was verified for elevated execution. /// /// The held file handle was opened without write or delete sharing, so the verified file @@ -440,8 +545,24 @@ fn parse_app_exec_alias(buffer: &[u8]) -> Option { /// granted to unprivileged users on stock drive roots), since they cannot redirect an /// existing path component. fn verify_ancestor_directories(path: &Path, subject: &str) -> anyhow::Result<()> { - let mut current = path.parent(); + verify_directory_chain( + path.parent(), + subject, + TrustedWriters::AdminOrTrustedInstaller, + TrustedWriters::AdminOrTrustedInstaller, + false, + ) +} + +fn verify_directory_chain( + mut current: Option<&Path>, + subject: &str, + first_writers: TrustedWriters, + ancestor_writers: TrustedWriters, + reject_reparse: bool, +) -> anyhow::Result<()> { let mut tamper_mask = PARENT_DIRECTORY_TAMPER_MASK; + let mut trusted_writers = first_writers; while let Some(dir) = current { let dir_subject = format!("{subject} ancestor directory '{}'", dir.display()); @@ -449,24 +570,40 @@ fn verify_ancestor_directories(path: &Path, subject: &str) -> anyhow::Result<()> let handle = OpenOptions::new() .access_mode(FILE_READ_ATTRIBUTES.0 | READ_CONTROL.0) .share_mode(FILE_SHARE_READ.0 | FILE_SHARE_WRITE.0 | FILE_SHARE_DELETE.0) - .custom_flags(FILE_FLAG_BACKUP_SEMANTICS.0) + .custom_flags(FILE_FLAG_BACKUP_SEMANTICS.0 | FILE_FLAG_OPEN_REPARSE_POINT.0) .open(dir) .with_context(|| format!("failed to open {dir_subject}"))?; - verify_handle_security( - &handle, - &dir_subject, - TrustedWriters::AdminOrTrustedInstaller, - tamper_mask, - )?; + let is_reparse = is_reparse_point(&handle).with_context(|| format!("failed to inspect {dir_subject}"))?; + if reject_reparse && is_reparse { + bail!("{dir_subject} is a reparse point"); + } + let reparse_mask = if is_reparse { REPARSE_POINT_TAMPER_MASK } else { 0 }; + verify_handle_security(&handle, &dir_subject, trusted_writers, tamper_mask | reparse_mask)?; tamper_mask = DIRECTORY_TAMPER_MASK; + trusted_writers = ancestor_writers; current = dir.parent(); } Ok(()) } +fn is_reparse_point(file: &File) -> anyhow::Result { + let mut info = FILE_ATTRIBUTE_TAG_INFO::default(); + let size = u32::try_from(size_of::()).expect("attribute info size fits u32"); + // SAFETY: The file handle and correctly sized output buffer are valid for the call. + unsafe { + GetFileInformationByHandleEx( + HANDLE(file.as_raw_handle()), + FileAttributeTagInfo, + (&raw mut info).cast(), + size, + ) + }?; + Ok(info.FileAttributes & FILE_ATTRIBUTE_REPARSE_POINT.0 != 0) +} + /// Resolve the normalized final path of an open file from its handle. fn final_path_from_handle(file: &File) -> anyhow::Result { let handle = HANDLE(file.as_raw_handle()); @@ -655,8 +792,10 @@ unsafe fn is_trusted_sid(sid: PSID, trusted_writers: TrustedWriters) -> bool { // write access for `LOCAL SERVICE`, so it must be trusted for the policy file. // It is a low-privilege shared service identity, however, so it is not trusted for // elevated executables, where accepting it would open a privilege-escalation path. - // SAFETY: Per function contract, `sid` points to a valid SID. - if trusted_writers == TrustedWriters::AdminOnly && unsafe { IsWellKnownSid(sid, WinLocalServiceSid) }.as_bool() { + if trusted_writers != TrustedWriters::AdminOrTrustedInstaller + // SAFETY: Per function contract, `sid` points to a valid SID. + && unsafe { IsWellKnownSid(sid, WinLocalServiceSid) }.as_bool() + { return true; } @@ -665,7 +804,7 @@ unsafe fn is_trusted_sid(sid: PSID, trusted_writers: TrustedWriters) -> bool { return true; } - if trusted_writers != TrustedWriters::AdminOrTrustedInstaller { + if trusted_writers == TrustedWriters::AdminOnly { return false; } @@ -1126,11 +1265,29 @@ mod tests { // FILE_DELETE_CHILD (0x40) lets a principal swap path components underneath the // verified executable. let sd = SddlDescriptor::parse("O:SYD:(A;;FA;;;SY)(A;;FA;;;BA)(A;;0x40;;;BU)"); - let error = sd.verify_with_mask(DIRECTORY_TAMPER_MASK).unwrap_err(); + // SAFETY: The owner and DACL point into the live SDDL-backed descriptor. + let error = unsafe { + verify_owner_and_dacl( + "policy ancestor", + sd.owner, + sd.dacl, + TrustedWriters::PolicyAncestor, + DIRECTORY_TAMPER_MASK, + ) + } + .unwrap_err(); assert!( error.to_string().contains("grants write access"), "unexpected error: {error}" ); + let sd = SddlDescriptor::parse("O:SYD:(A;;FA;;;SY)(A;;FA;;;BA)(A;;0x100;;;BU)"); + sd.verify_with_mask(DIRECTORY_TAMPER_MASK) + .expect("write-attributes rights on a regular higher ancestor must be tolerated"); + let sd = SddlDescriptor::parse("O:SYD:(A;;FA;;;SY)(A;;FA;;;BA)(A;;GW;;;BU)"); + let error = sd + .verify_with_mask(DIRECTORY_TAMPER_MASK | REPARSE_POINT_TAMPER_MASK) + .unwrap_err(); + assert!(error.to_string().contains("grants write access")); } #[test] @@ -1208,19 +1365,20 @@ mod tests { #[test] fn system_owned_admin_only_policy_file_is_accepted() { let temp = tempfile::NamedTempFile::new().unwrap(); + let path = temp.path().canonicalize().unwrap(); let admins = Sid::from_well_known(WinBuiltinAdministratorsSid, None).unwrap(); // Setting the owner to Administrators requires an elevated token; skip otherwise. // The equivalent owner/DACL combinations are covered by the SDDL-based tests above. - if set_security(temp.path(), Some(&admins), &[]).is_err() { + if set_security(&path, Some(&admins), &[]).is_err() { return; } let system = Sid::from_well_known(WinLocalSystemSid, None).unwrap(); let users = Sid::from_well_known(windows::Win32::Security::WinBuiltinUsersSid, None).unwrap(); set_security( - temp.path(), + &path, None, &[ grant(GENERIC_ALL.0, system), @@ -1230,11 +1388,45 @@ mod tests { ) .unwrap(); - let file = File::open(temp.path()).unwrap(); + let file = OpenOptions::new() + .read(true) + .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT.0) + .open(&path) + .unwrap(); + verify_policy_file_path(&file, &path).expect("ordinary policy path must be accepted"); verify_policy_file_security(&file).expect("SYSTEM/Administrators-only policy file must be accepted"); } + #[test] + fn policy_reparse_ancestor_is_rejected() { + let temp = tempfile::tempdir().unwrap(); + let target = temp.path().join("target"); + std::fs::create_dir(&target).unwrap(); + let link = temp.path().join("link"); + std::os::windows::fs::symlink_dir(&target, &link).unwrap(); + + let error = verify_policy_path_ancestors(&link.join("policy.json")).unwrap_err(); + assert!(error.to_string().contains("reparse point"), "unexpected error: {error}"); + } + + #[test] + fn policy_leaf_reparse_is_rejected() { + let temp = tempfile::tempdir().unwrap(); + let target = temp.path().join("target.json"); + std::fs::write(&target, "{}").unwrap(); + let link = temp.path().join("policy.json"); + std::os::windows::fs::symlink_file(&target, &link).unwrap(); + let file = OpenOptions::new() + .read(true) + .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT.0) + .open(&link) + .unwrap(); + + let error = verify_policy_file_path(&file, &link).unwrap_err(); + assert!(error.to_string().contains("reparse point"), "unexpected error: {error}"); + } + #[test] fn everyone_writable_executable_path_is_rejected() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/crates/now-package-broker/src/policy_store/mod.rs b/crates/now-package-broker/src/policy_store/mod.rs new file mode 100644 index 000000000..abcb682a3 --- /dev/null +++ b/crates/now-package-broker/src/policy_store/mod.rs @@ -0,0 +1,1057 @@ +//! Serialized policy management, validation, persistence, and reload. +//! Store tokens serialize API writers and reloads, not privileged out-of-band writes. +//! Conditional handle-relative publication is deferred. + +use std::fs::{File, OpenOptions}; +use std::io::{Read as _, Write as _}; +use std::mem::size_of; +use std::os::windows::ffi::OsStrExt as _; +use std::os::windows::fs::OpenOptionsExt as _; +use std::os::windows::io::AsRawHandle as _; +use std::path::{Component, Path, PathBuf}; +use std::sync::{Arc, RwLock}; + +use anyhow::Context as _; +use chrono::Utc; +use now_policy::PolicyDocument; +use now_policy_api::{ + API_VERSION_STR, ErrorCode, ErrorResponse, ErrorResponseKind, InvalidPolicyDiagnostics, PolicyConfigurationSource, + PolicyManagementSnapshot, PolicyManagementState, PolicyReadOnlyReason, PolicyReplacementOperation, + PolicyReplacementRequest, PolicyStoreToken, PolicyValidationResult, PolicyWriteCapability, ServerContext, + Transport, +}; +use sha2::{Digest as _, Sha256}; +use windows::Win32::Foundation::HANDLE; +use windows::Win32::Storage::FileSystem::{ + FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_ID_INFO, FILE_READ_ATTRIBUTES, FILE_SHARE_DELETE, + FILE_SHARE_READ, FILE_SHARE_WRITE, FileIdInfo, GetFileInformationByHandleEx, GetVolumeInformationW, + GetVolumePathNameW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW, READ_CONTROL, +}; +use windows::core::PCWSTR; + +use crate::policy_security; +mod receipt; +mod validation; + +#[derive(Clone, Copy, Debug)] +pub enum ReloadCause { + ExternalChange, +} + +#[derive(Clone, PartialEq, Eq)] +struct DiskFingerprint([u8; 32]); + +struct Observation { + state: PolicyManagementState, + policy: Option, + invalid_diagnostics: Option, + write_capability: PolicyWriteCapability, + read_only_reason: Option, + configured_path: PathBuf, + fingerprint: DiskFingerprint, +} + +struct PersistedPolicy { + policy: PolicyDocument, + observation: Observation, +} + +enum WriteFailure { + PrePublication(anyhow::Error), + PostPublication(anyhow::Error), +} + +trait PolicyStorage: Send + Sync { + fn observe(&self, source: PolicyConfigurationSource, path: &Path) -> Observation; + fn create( + &self, + configured_path: &Path, + observation: &Observation, + bytes: &[u8], + ) -> Result; + fn replace( + &self, + configured_path: &Path, + observation: &Observation, + bytes: &[u8], + ) -> Result; +} + +struct FilePolicyStorage; + +impl PolicyStorage for FilePolicyStorage { + fn observe(&self, source: PolicyConfigurationSource, path: &Path) -> Observation { + observe_file(source, path) + } + + fn create( + &self, + configured_path: &Path, + observation: &Observation, + bytes: &[u8], + ) -> Result { + publish_file(configured_path, observation, bytes, false) + } + + fn replace( + &self, + configured_path: &Path, + observation: &Observation, + bytes: &[u8], + ) -> Result { + publish_file(configured_path, observation, bytes, true) + } +} + +struct Snapshot { + state: PolicyManagementState, + policy: Option>, + invalid_diagnostics: Option, + write_capability: PolicyWriteCapability, + read_only_reason: Option, + configured_path: PathBuf, + store_token: PolicyStoreToken, + fingerprint: DiskFingerprint, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Monitoring { + Initializing, + Available, + Unavailable, +} + +#[derive(Debug)] +pub struct ReplaceSuccess { + pub policy: PolicyDocument, + pub validation: PolicyValidationResult, + pub management: PolicyManagementSnapshot, +} + +pub struct PolicyStore { + configured_path: PathBuf, + source: PolicyConfigurationSource, + snapshot: RwLock>, + writer: tokio::sync::Mutex, + storage: Arc, + receipt_key: receipt::ReceiptKey, +} + +impl PolicyStore { + pub fn load(configured_path: Option) -> Arc { + Self::load_with_storage(configured_path, Arc::new(FilePolicyStorage), Monitoring::Initializing) + } + + fn load_with_storage( + configured_path: Option, + storage: Arc, + monitoring: Monitoring, + ) -> Arc { + let (configured_path, source) = match configured_path { + Some(path) => (path, PolicyConfigurationSource::ConfiguredPath), + None => ( + crate::policy_loader::find_default_policy() + .unwrap_or_else(|_| crate::policy_loader::default_policy_candidate()), + PolicyConfigurationSource::DefaultPath, + ), + }; + let observation = storage.observe(source, &configured_path); + let snapshot = Arc::new(snapshot_from_observation(observation, random_store_token())); + Arc::new(Self { + configured_path, + source, + snapshot: RwLock::new(snapshot), + writer: tokio::sync::Mutex::new(monitoring), + storage, + receipt_key: receipt::ReceiptKey::generate(), + }) + } + + fn snapshot(&self) -> Arc { + Arc::clone(&self.snapshot.read().expect("policy store snapshot lock poisoned")) + } + + pub fn active_policy(&self) -> Option> { + self.snapshot().policy.clone() + } + + pub fn management_snapshot(&self) -> PolicyManagementSnapshot { + let snapshot = self.snapshot(); + management_from_snapshot(&snapshot, self.source) + } + + pub(crate) fn configured_path(&self) -> PathBuf { + self.snapshot().configured_path.clone() + } + + pub fn validate_draft(&self, raw: &serde_json::Value) -> PolicyValidationResult { + let mut result = validation::validate_draft(raw); + if let Some(draft) = &result.canonical_draft { + result.validation_receipt = Some(self.receipt_key.issue( + &result.validator_version, + draft, + &result.findings, + )); + } + result + } + + pub async fn reload_from_disk(&self, cause: ReloadCause) -> PolicyManagementSnapshot { + let monitoring = self.writer.lock().await; + if *monitoring != Monitoring::Available { + return self.management_snapshot(); + } + let observation = self.storage.observe(self.source, &self.configured_path); + let management = self.publish_observation(observation); + tracing::info!(?cause, state = ?management.state, "Reloaded package broker policy"); + management + } + + pub(crate) async fn mark_monitoring_ready(&self) -> PolicyManagementSnapshot { + let mut monitoring = self.writer.lock().await; + if *monitoring != Monitoring::Initializing { + return self.management_snapshot(); + } + let management = self.publish_observation(self.storage.observe(self.source, &self.configured_path)); + *monitoring = Monitoring::Available; + management + } + + pub(crate) async fn mark_watcher_unavailable(&self) { + let mut monitoring = self.writer.lock().await; + *monitoring = Monitoring::Unavailable; + let previous = self.snapshot(); + let observation = Observation { + state: PolicyManagementState::Invalid, + policy: None, + invalid_diagnostics: Some(InvalidPolicyDiagnostics { + diagnostics_version: API_VERSION_STR.into(), + findings: vec![validation::disk_failure_finding( + validation::DiskFailureReason::WatcherUnavailable, + )], + }), + write_capability: PolicyWriteCapability::ReadOnly, + read_only_reason: Some(PolicyReadOnlyReason::ManagementDisabled), + configured_path: previous.configured_path.clone(), + fingerprint: DiskFingerprint(Sha256::digest(b"watcher unavailable").into()), + }; + self.publish_observation(observation); + } + + pub async fn replace(&self, request: PolicyReplacementRequest) -> Result { + let monitoring = self.writer.lock().await; + if *monitoring != Monitoring::Available { + return Err(error_with_management( + ErrorCode::BrokerPaused, + "policy change monitoring is unavailable", + self.management_snapshot(), + )); + } + let previous = self.snapshot(); + let observation = self.storage.observe(self.source, &self.configured_path); + let fresh_token = token_for(&previous, &observation.fingerprint); + + // Both conflict modes require this exact token. + // ConfirmOverwrite records retry intent without retaining token history. + if fresh_token != request.expected_store_token { + let management = self.publish_observation(observation); + return Err(error_with_management( + ErrorCode::StalePolicyStoreToken, + "the configured policy changed after the supplied store token was observed", + management, + )); + } + + if observation.write_capability != PolicyWriteCapability::Writable { + let code = match observation.read_only_reason { + Some(PolicyReadOnlyReason::UnsupportedFileSystem) => ErrorCode::UnsupportedPolicyFilesystem, + Some(PolicyReadOnlyReason::UnsupportedFormat) => ErrorCode::UnsupportedPolicyFormat, + _ => ErrorCode::UnsafePolicyPath, + }; + return Err(error_response(code, "the configured policy path is not writable")); + } + + let validation = self.validate_draft(&request.draft); + if !validation.is_valid { + return Err(error_with_validation( + ErrorCode::InvalidPolicy, + "the submitted draft failed authoritative validation", + validation, + )); + } + let draft = validation + .canonical_draft + .clone() + .expect("valid validation carries a canonical draft"); + if !self.receipt_key.verify( + &validation.validator_version, + &draft, + &validation.findings, + &request.validation_receipt, + ) { + return Err(error_with_validation( + ErrorCode::ValidationFailed, + "the validation receipt does not match this draft", + validation, + )); + } + if !validation.findings.is_empty() && !request.warnings_acknowledged { + return Err(error_with_validation( + ErrorCode::WarningConfirmationRequired, + "validation warnings must be explicitly acknowledged", + validation, + )); + } + + let revision = plan_revision( + request.operation, + observation.state, + observation.policy.as_ref(), + &draft.metadata.id.0, + ) + .map_err(|message| error_response(ErrorCode::Conflict, message))?; + let policy = draft.into_policy_document(revision, Utc::now()).map_err(|_| { + error_response( + ErrorCode::ValidationFailed, + "failed to commit the validated policy draft", + ) + })?; + let bytes = serde_json::to_vec_pretty(&policy) + .map_err(|_| error_response(ErrorCode::InternalError, "failed to serialize the committed policy"))?; + + let persisted = if request.operation == PolicyReplacementOperation::Create { + self.storage.create(&self.configured_path, &observation, &bytes) + } else { + self.storage.replace(&self.configured_path, &observation, &bytes) + }; + let persisted = match persisted { + Ok(persisted) => persisted, + Err(WriteFailure::PrePublication(error)) => { + tracing::warn!(error = format!("{error:#}"), "Policy persistence failed"); + let current = self.storage.observe(self.source, &self.configured_path); + if current.fingerprint != observation.fingerprint { + let management = self.publish_observation(current); + return Err(error_with_management( + ErrorCode::StalePolicyStoreToken, + "the policy storage changed before publication; retry with the current store token", + management, + )); + } + return Err(error_response( + ErrorCode::PolicyPersistenceFailed, + "failed to persist the policy", + )); + } + Err(WriteFailure::PostPublication(error)) => { + tracing::warn!( + error = format!("{error:#}"), + "Published policy failed authoritative reload" + ); + let current = self.storage.observe(self.source, &self.configured_path); + let management = self.publish_observation(current); + return Err(error_with_management( + ErrorCode::PolicyActivationFailed, + "the policy was published but failed authoritative reload", + management, + )); + } + }; + + if persisted.observation.state != PolicyManagementState::Active { + let management = self.publish_observation(persisted.observation); + return Err(error_with_management( + ErrorCode::PolicyActivationFailed, + "the policy was published but failed authoritative reload", + management, + )); + } + let token = token_for(&previous, &persisted.observation.fingerprint); + let snapshot = Arc::new(snapshot_from_observation(persisted.observation, token)); + *self.snapshot.write().expect("policy store snapshot lock poisoned") = snapshot; + + Ok(ReplaceSuccess { + policy: persisted.policy, + validation, + management: self.management_snapshot(), + }) + } + + fn publish_observation(&self, observation: Observation) -> PolicyManagementSnapshot { + let previous = self.snapshot(); + if previous.fingerprint == observation.fingerprint + && previous.write_capability == observation.write_capability + && previous.read_only_reason == observation.read_only_reason + { + return management_from_snapshot(&previous, self.source); + } + let token = token_for(&previous, &observation.fingerprint); + let snapshot = Arc::new(snapshot_from_observation(observation, token)); + let management = management_from_snapshot(&snapshot, self.source); + *self.snapshot.write().expect("policy store snapshot lock poisoned") = snapshot; + management + } + + #[cfg(test)] + pub(crate) fn for_tests(policy: Option) -> Arc { + let storage = Arc::new(TestStorage::new(policy)); + Self::load_with_storage(Some(PathBuf::from(r"C:\policy.json")), storage, Monitoring::Available) + } + + #[cfg(test)] + pub(crate) fn test_set_active(&self, policy: Arc) { + let previous = self.snapshot(); + let fingerprint = DiskFingerprint( + Sha256::digest(serde_json::to_vec(policy.as_ref()).expect("test policy serializes")).into(), + ); + let snapshot = Arc::new(Snapshot { + state: PolicyManagementState::Active, + policy: Some(policy), + invalid_diagnostics: None, + write_capability: PolicyWriteCapability::Writable, + read_only_reason: None, + configured_path: previous.configured_path.clone(), + store_token: token_for(&previous, &fingerprint), + fingerprint, + }); + *self.snapshot.write().expect("policy store snapshot lock poisoned") = snapshot; + } +} + +fn plan_revision( + operation: PolicyReplacementOperation, + state: PolicyManagementState, + current_policy: Option<&PolicyDocument>, + new_id: &str, +) -> Result { + match operation { + PolicyReplacementOperation::Update => { + let current = current_policy.ok_or_else(|| "Update requires an Active policy".to_owned())?; + if current.metadata.id.0 != new_id { + return Err("Update must preserve the active policy identity".to_owned()); + } + current + .metadata + .revision + .checked_add(1) + .filter(|revision| i32::try_from(*revision).is_ok()) + .ok_or_else(|| "the policy revision reached its maximum value".to_owned()) + } + PolicyReplacementOperation::ReplaceIdentity => { + let current = current_policy.ok_or_else(|| "ReplaceIdentity requires an Active policy".to_owned())?; + if current.metadata.id.0 == new_id { + return Err("ReplaceIdentity requires a different policy identity".to_owned()); + } + Ok(1) + } + PolicyReplacementOperation::Create if state == PolicyManagementState::Missing => Ok(1), + PolicyReplacementOperation::Repair if state == PolicyManagementState::Invalid => Ok(1), + PolicyReplacementOperation::Create => Err("Create requires a Missing policy".to_owned()), + PolicyReplacementOperation::Repair => Err("Repair requires an Invalid policy".to_owned()), + } +} + +fn snapshot_from_observation(observation: Observation, store_token: PolicyStoreToken) -> Snapshot { + Snapshot { + state: observation.state, + policy: observation.policy.map(Arc::new), + invalid_diagnostics: observation.invalid_diagnostics, + write_capability: observation.write_capability, + read_only_reason: observation.read_only_reason, + configured_path: observation.configured_path, + store_token, + fingerprint: observation.fingerprint, + } +} + +fn management_from_snapshot(snapshot: &Snapshot, source: PolicyConfigurationSource) -> PolicyManagementSnapshot { + PolicyManagementSnapshot { + state: snapshot.state, + configured_path: snapshot.configured_path.display().to_string(), + store_token: snapshot.store_token.clone(), + source, + write_capability: snapshot.write_capability, + read_only_reason: snapshot.read_only_reason, + elevation_required: true, + policy: snapshot.policy.as_deref().cloned(), + invalid_diagnostics: snapshot.invalid_diagnostics.clone(), + } +} + +fn token_for(previous: &Snapshot, fingerprint: &DiskFingerprint) -> PolicyStoreToken { + if previous.fingerprint == *fingerprint { + previous.store_token.clone() + } else { + random_store_token() + } +} + +fn random_store_token() -> PolicyStoreToken { + format!("store:{}", uuid::Uuid::new_v4().simple()).into() +} + +fn error_response(code: ErrorCode, message: impl Into) -> ErrorResponse { + ErrorResponse { + response_kind: ErrorResponseKind, + response_version: API_VERSION_STR.into(), + server: ServerContext { + server_version: env!("CARGO_PKG_VERSION").to_owned(), + transport: Transport::HttpNamedPipe, + }, + code, + message: message.into(), + details: Vec::new(), + validation: None, + management: None, + } +} + +fn error_with_validation( + code: ErrorCode, + message: impl Into, + validation: PolicyValidationResult, +) -> ErrorResponse { + let mut response = error_response(code, message); + response.validation = Some(validation); + response +} + +fn error_with_management( + code: ErrorCode, + message: impl Into, + management: PolicyManagementSnapshot, +) -> ErrorResponse { + let mut response = error_response(code, message); + response.management = Some(management); + response +} + +fn observe_file(_source: PolicyConfigurationSource, configured_path: &Path) -> Observation { + let mut hasher = Sha256::new(); + for unit in configured_path.as_os_str().encode_wide() { + hasher.update(unit.to_le_bytes()); + } + + if !is_safe_path_shape(configured_path) { + return invalid_observation( + configured_path.to_owned(), + PolicyWriteCapability::Unsupported, + Some(PolicyReadOnlyReason::UnsafePath), + validation::DiskFailureReason::InsecureStorage, + hasher, + ); + } + + let extension = configured_path + .extension() + .and_then(|value| value.to_str()) + .unwrap_or_default() + .to_ascii_lowercase(); + if extension != "json" { + return invalid_observation( + configured_path.to_owned(), + PolicyWriteCapability::Unsupported, + Some(PolicyReadOnlyReason::UnsupportedFormat), + validation::DiskFailureReason::UnsupportedFormat, + hasher, + ); + } + + let display_path = match canonical_display_path(configured_path) { + Ok(path) => path, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return missing_observation( + configured_path.to_owned(), + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::InsufficientPermissions), + hasher, + ); + } + Err(error) => { + tracing::warn!(error = %error, "Failed to resolve policy path"); + return invalid_observation( + configured_path.to_owned(), + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::UnsafePath), + validation::DiskFailureReason::InsecureStorage, + hasher, + ); + } + }; + for unit in display_path.as_os_str().encode_wide() { + hasher.update(unit.to_le_bytes()); + } + + let Some(parent) = display_path.parent() else { + return invalid_observation( + display_path, + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::UnsafePath), + validation::DiskFailureReason::Unreadable, + hasher, + ); + }; + let directory = match open_directory(parent) { + Ok(directory) => directory, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return missing_observation( + display_path, + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::InsufficientPermissions), + hasher, + ); + } + Err(error) => { + tracing::warn!(error = %error, "Failed to open policy directory"); + return invalid_observation( + display_path, + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::UnsafePath), + validation::DiskFailureReason::InsecureStorage, + hasher, + ); + } + }; + hash_file_identity(&directory, &mut hasher); + let directory_safe = match policy_security::verify_policy_path_ancestors(configured_path) + .and_then(|()| policy_security::verify_policy_path_ancestors(&display_path)) + .and_then(|()| { + let current_path = canonical_display_path(configured_path) + .context("failed to resolve policy path after security validation")?; + if policy_security::windows_paths_equal(&display_path, ¤t_path) { + Ok(()) + } else { + anyhow::bail!("policy path canonical chain changed during security validation") + } + }) + .and_then(|()| policy_security::verify_policy_directory_security(&directory)) + .and_then(|()| policy_security::security_state_digest(&directory)) + { + Ok(digest) => { + hasher.update(digest); + true + } + Err(error) => { + tracing::warn!( + error = format!("{error:#}"), + "Policy directory security validation failed" + ); + false + } + }; + if !directory_safe { + return invalid_observation( + display_path, + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::UnsafePath), + validation::DiskFailureReason::InsecureStorage, + hasher, + ); + } + let atomic_filesystem = directory_safe && supports_atomic_replace(parent); + let capability = if !atomic_filesystem { + PolicyWriteCapability::Unsupported + } else { + PolicyWriteCapability::Writable + }; + let read_only_reason = match capability { + PolicyWriteCapability::Writable => None, + PolicyWriteCapability::Unsupported => Some(PolicyReadOnlyReason::UnsupportedFileSystem), + PolicyWriteCapability::ReadOnly => unreachable!("unsafe directories returned above"), + }; + + let mut file = match OpenOptions::new() + .read(true) + .custom_flags(FILE_FLAG_OPEN_REPARSE_POINT.0) + .open(&display_path) + { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return missing_observation(display_path, capability, read_only_reason, hasher); + } + Err(error) => { + tracing::warn!(error = %error, "Failed to open configured policy"); + return invalid_observation( + display_path, + capability, + read_only_reason, + validation::DiskFailureReason::Unreadable, + hasher, + ); + } + }; + hash_file_identity(&file, &mut hasher); + if let Err(error) = policy_security::verify_policy_file_path(&file, &display_path) + .and_then(|()| policy_security::verify_policy_file_security(&file)) + { + tracing::warn!( + error = format!("{error:#}"), + "Configured policy security validation failed" + ); + return invalid_observation( + display_path, + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::UnsafePath), + validation::DiskFailureReason::InsecureStorage, + hasher, + ); + } + match policy_security::security_state_digest(&file) { + Ok(digest) => hasher.update(digest), + Err(error) => { + tracing::warn!( + error = format!("{error:#}"), + "Failed to fingerprint configured policy security" + ); + return invalid_observation( + display_path, + PolicyWriteCapability::ReadOnly, + Some(PolicyReadOnlyReason::UnsafePath), + validation::DiskFailureReason::InsecureStorage, + hasher, + ); + } + } + let mut bytes = Vec::new(); + if let Err(error) = file.read_to_end(&mut bytes) { + tracing::warn!(error = %error, "Failed to read configured policy"); + return invalid_observation( + display_path, + capability, + read_only_reason, + validation::DiskFailureReason::Unreadable, + hasher, + ); + } + hasher.update(&bytes); + let policy = serde_json::from_slice::(&bytes); + let policy = match policy { + Ok(policy) => policy, + Err(error) => { + tracing::warn!(error = %error, "Configured policy parsing failed"); + return invalid_observation( + display_path, + capability, + read_only_reason, + validation::DiskFailureReason::MalformedContent, + hasher, + ); + } + }; + let committed_validation = validation::validate_committed_policy(&policy); + if !committed_validation.is_valid { + tracing::warn!( + findings = ?committed_validation.findings, + "Configured policy semantic validation failed" + ); + return invalid_observation( + display_path, + capability, + read_only_reason, + validation::DiskFailureReason::FailedSemanticValidation, + hasher, + ); + } + + Observation { + state: PolicyManagementState::Active, + policy: Some(policy), + invalid_diagnostics: None, + write_capability: capability, + read_only_reason, + configured_path: display_path, + fingerprint: DiskFingerprint(hasher.finalize().into()), + } +} + +fn publish_file( + configured_path: &Path, + observation: &Observation, + bytes: &[u8], + replace: bool, +) -> Result { + let path = &observation.configured_path; + let parent = path + .parent() + .ok_or_else(|| WriteFailure::PrePublication(anyhow::anyhow!("policy path has no parent")))?; + let leaf = path + .file_name() + .ok_or_else(|| WriteFailure::PrePublication(anyhow::anyhow!("policy path has no file name")))?; + let temp_path = parent.join(format!( + ".{}.{}.tmp", + leaf.to_string_lossy(), + uuid::Uuid::new_v4().simple() + )); + let prepared = (|| { + let mut temp = OpenOptions::new().write(true).create_new(true).open(&temp_path)?; + temp.write_all(bytes)?; + temp.sync_all()?; + policy_security::verify_policy_file_security(&temp)?; + drop(temp); + + let from = wide_path(&temp_path); + let to = wide_path(path); + let flags = if replace { + MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH + } else { + MOVEFILE_WRITE_THROUGH + }; + // SAFETY: Both buffers are live, nul-terminated absolute paths. + unsafe { MoveFileExW(PCWSTR(from.as_ptr()), PCWSTR(to.as_ptr()), flags) }?; + Ok::<(), anyhow::Error>(()) + })(); + if let Err(error) = prepared { + let _ = std::fs::remove_file(&temp_path); + return Err(WriteFailure::PrePublication(error)); + } + + let reloaded = (|| { + let reloaded = observe_file(PolicyConfigurationSource::ConfiguredPath, configured_path); + let policy = reloaded + .policy + .clone() + .ok_or_else(|| anyhow::anyhow!("published policy failed authoritative reload"))?; + let expected: serde_json::Value = serde_json::from_slice(bytes)?; + if serde_json::to_value(&policy)? != expected { + anyhow::bail!("published policy does not match the requested committed document"); + } + Ok(PersistedPolicy { + policy, + observation: reloaded, + }) + })(); + reloaded.map_err(WriteFailure::PostPublication) +} + +fn missing_observation( + path: PathBuf, + capability: PolicyWriteCapability, + reason: Option, + mut hasher: Sha256, +) -> Observation { + hasher.update(b"missing"); + Observation { + state: PolicyManagementState::Missing, + policy: None, + invalid_diagnostics: None, + write_capability: capability, + read_only_reason: reason, + configured_path: path, + fingerprint: DiskFingerprint(hasher.finalize().into()), + } +} + +fn invalid_observation( + path: PathBuf, + capability: PolicyWriteCapability, + reason: Option, + failure: validation::DiskFailureReason, + mut hasher: Sha256, +) -> Observation { + hasher.update(format!("{failure:?}")); + Observation { + state: PolicyManagementState::Invalid, + policy: None, + invalid_diagnostics: Some(InvalidPolicyDiagnostics { + diagnostics_version: API_VERSION_STR.into(), + findings: vec![validation::disk_failure_finding(failure)], + }), + write_capability: capability, + read_only_reason: reason, + configured_path: path, + fingerprint: DiskFingerprint(hasher.finalize().into()), + } +} + +fn is_safe_path_shape(path: &Path) -> bool { + let raw = path.as_os_str().to_string_lossy(); + path.is_absolute() + && path.file_name().is_some() + && !raw.split(['\\', '/']).any(|segment| matches!(segment, "." | "..")) + && path + .components() + .all(|component| !matches!(component, Component::CurDir | Component::ParentDir)) +} + +fn canonical_display_path(path: &Path) -> std::io::Result { + let parent = path + .parent() + .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "policy path has no parent"))?; + let leaf = path + .file_name() + .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "policy path has no file name"))?; + Ok(parent.canonicalize()?.join(leaf)) +} + +fn open_directory(path: &Path) -> std::io::Result { + OpenOptions::new() + .access_mode(FILE_READ_ATTRIBUTES.0 | READ_CONTROL.0) + .share_mode((FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE).0) + .custom_flags(FILE_FLAG_BACKUP_SEMANTICS.0) + .open(path) +} + +fn supports_atomic_replace(path: &Path) -> bool { + let path = wide_path(path); + let mut root = vec![0; 512]; + // SAFETY: The path is nul-terminated and the root buffer is writable. + if unsafe { GetVolumePathNameW(PCWSTR(path.as_ptr()), &mut root) }.is_err() { + return false; + } + let mut filesystem = vec![0; 261]; + // SAFETY: GetVolumePathNameW returned a nul-terminated root and the output buffer is writable. + if unsafe { GetVolumeInformationW(PCWSTR(root.as_ptr()), None, None, None, None, Some(&mut filesystem)) }.is_err() { + return false; + } + let length = filesystem + .iter() + .position(|unit| *unit == 0) + .unwrap_or(filesystem.len()); + matches!( + String::from_utf16_lossy(&filesystem[..length]).as_str(), + "NTFS" | "ReFS" + ) +} + +fn hash_file_identity(file: &File, hasher: &mut Sha256) { + let mut info = FILE_ID_INFO::default(); + let size = u32::try_from(size_of::()).expect("FILE_ID_INFO size fits u32"); + // SAFETY: The file handle and correctly sized output buffer are valid for the call. + if unsafe { GetFileInformationByHandleEx(HANDLE(file.as_raw_handle()), FileIdInfo, (&raw mut info).cast(), size) } + .is_ok() + { + hasher.update(info.VolumeSerialNumber.to_le_bytes()); + hasher.update(info.FileId.Identifier); + } +} + +fn wide_path(path: &Path) -> Vec { + path.as_os_str().encode_wide().chain(std::iter::once(0)).collect() +} + +#[cfg(test)] +struct TestStorage { + observation: parking_lot::Mutex, + fail_persist: std::sync::atomic::AtomicBool, +} + +#[cfg(test)] +impl TestStorage { + fn new(policy: Option) -> Self { + let state = if policy.is_some() { + PolicyManagementState::Active + } else { + PolicyManagementState::Missing + }; + Self { + observation: parking_lot::Mutex::new(Observation { + state, + policy, + invalid_diagnostics: None, + write_capability: PolicyWriteCapability::Writable, + read_only_reason: None, + configured_path: PathBuf::from(r"C:\policy.json"), + fingerprint: DiskFingerprint([0; 32]), + }), + fail_persist: std::sync::atomic::AtomicBool::new(false), + } + } + + fn invalid() -> Self { + let mut storage = Self::new(None); + storage.observation = parking_lot::Mutex::new(Observation { + state: PolicyManagementState::Invalid, + policy: None, + invalid_diagnostics: Some(InvalidPolicyDiagnostics { + diagnostics_version: API_VERSION_STR.into(), + findings: vec![validation::disk_failure_finding( + validation::DiskFailureReason::MalformedContent, + )], + }), + write_capability: PolicyWriteCapability::Writable, + read_only_reason: None, + configured_path: PathBuf::from(r"C:\policy.json"), + fingerprint: DiskFingerprint([1; 32]), + }); + storage + } + + fn set_disk_state(&self, policy: Option, invalid: bool, marker: u8) { + let mut observation = self.observation.lock(); + observation.state = if invalid { + PolicyManagementState::Invalid + } else if policy.is_some() { + PolicyManagementState::Active + } else { + PolicyManagementState::Missing + }; + observation.policy = policy; + observation.invalid_diagnostics = invalid.then(|| InvalidPolicyDiagnostics { + diagnostics_version: API_VERSION_STR.into(), + findings: vec![validation::disk_failure_finding( + validation::DiskFailureReason::MalformedContent, + )], + }); + observation.fingerprint = DiskFingerprint([marker; 32]); + } +} + +#[cfg(test)] +impl PolicyStorage for TestStorage { + fn observe(&self, _source: PolicyConfigurationSource, _path: &Path) -> Observation { + clone_observation(&self.observation.lock()) + } + + fn create( + &self, + _configured_path: &Path, + observation: &Observation, + bytes: &[u8], + ) -> Result { + self.persist(observation, bytes) + } + + fn replace( + &self, + _configured_path: &Path, + observation: &Observation, + bytes: &[u8], + ) -> Result { + self.persist(observation, bytes) + } +} + +#[cfg(test)] +impl TestStorage { + fn persist(&self, observation: &Observation, bytes: &[u8]) -> Result { + if self.fail_persist.load(std::sync::atomic::Ordering::SeqCst) { + return Err(WriteFailure::PrePublication(anyhow::anyhow!( + "injected persistence failure" + ))); + } + let policy: PolicyDocument = + serde_json::from_slice(bytes).map_err(|error| WriteFailure::PrePublication(error.into()))?; + let mut next = clone_observation(observation); + next.state = PolicyManagementState::Active; + next.policy = Some(policy.clone()); + next.invalid_diagnostics = None; + next.fingerprint = DiskFingerprint(Sha256::digest(bytes).into()); + *self.observation.lock() = clone_observation(&next); + Ok(PersistedPolicy { + policy, + observation: next, + }) + } +} + +#[cfg(test)] +fn clone_observation(observation: &Observation) -> Observation { + Observation { + state: observation.state, + policy: observation.policy.clone(), + invalid_diagnostics: observation.invalid_diagnostics.clone(), + write_capability: observation.write_capability, + read_only_reason: observation.read_only_reason, + configured_path: observation.configured_path.clone(), + fingerprint: observation.fingerprint.clone(), + } +} diff --git a/crates/now-package-broker/src/policy_store/receipt.rs b/crates/now-package-broker/src/policy_store/receipt.rs new file mode 100644 index 000000000..8bcaf2e55 --- /dev/null +++ b/crates/now-package-broker/src/policy_store/receipt.rs @@ -0,0 +1,540 @@ +use hmac::{Hmac, Mac as _}; +use now_policy::PolicyDraftDocument; +use now_policy_api::{PolicyFinding, PolicyValidationReceipt}; +use sha2::Sha256; + +type HmacSha256 = Hmac; + +const RECEIPT_PREFIX: &str = "hmac-sha256:"; + +pub(super) struct ReceiptKey([u8; 32]); + +impl ReceiptKey { + pub(super) fn generate() -> Self { + let mut key = [0; 32]; + key[..16].copy_from_slice(uuid::Uuid::new_v4().as_bytes()); + key[16..].copy_from_slice(uuid::Uuid::new_v4().as_bytes()); + Self(key) + } + + fn mac( + &self, + validator_version: &str, + canonical_draft: &PolicyDraftDocument, + findings: &[PolicyFinding], + ) -> HmacSha256 { + let canonical_json = serde_json::to_vec(canonical_draft).expect("canonical policy draft serializes"); + let finding_identities: Vec<_> = findings + .iter() + .map(|finding| (&finding.severity, &finding.code, &finding.rule_id, &finding.arguments)) + .collect(); + let findings_json = serde_json::to_vec(&finding_identities).expect("policy finding identities serialize"); + let mut mac = HmacSha256::new_from_slice(&self.0).expect("HMAC accepts any key length"); + mac.update(validator_version.as_bytes()); + mac.update(b"\0"); + mac.update(&canonical_json); + mac.update(b"\0"); + mac.update(&findings_json); + mac + } + + pub(super) fn issue( + &self, + validator_version: &str, + canonical_draft: &PolicyDraftDocument, + findings: &[PolicyFinding], + ) -> PolicyValidationReceipt { + let tag = self + .mac(validator_version, canonical_draft, findings) + .finalize() + .into_bytes(); + format!("{RECEIPT_PREFIX}{}", hex::encode(tag)).into() + } + + pub(super) fn verify( + &self, + validator_version: &str, + canonical_draft: &PolicyDraftDocument, + findings: &[PolicyFinding], + candidate: &PolicyValidationReceipt, + ) -> bool { + let Some(encoded) = candidate.strip_prefix(RECEIPT_PREFIX) else { + return false; + }; + let Ok(tag) = hex::decode(encoded) else { + return false; + }; + self.mac(validator_version, canonical_draft, findings) + .verify_slice(&tag) + .is_ok() + } +} + +#[cfg(test)] +mod tests { + use std::path::PathBuf; + use std::sync::Arc; + + use chrono::Utc; + use now_policy::{PolicyDocument, PolicyDraftDocument}; + use now_policy_api::{ + API_VERSION_STR, ErrorCode, PolicyConfigurationSource, PolicyConflictHandling, PolicyFindingCode, + PolicyFindingSeverity, PolicyManagementState, PolicyReadOnlyReason, PolicyReplacementOperation, + PolicyReplacementRequest, PolicyReplacementRequestKind, PolicyStoreToken, PolicyValidationResult, + PolicyWriteCapability, + }; + + use super::*; + use crate::policy_store::{ + Monitoring, PolicyStorage, PolicyStore, ReloadCause, TestStorage, observe_file, plan_revision, + }; + use crate::policy_watcher::{WatcherFailure, fail_closed}; + fn draft(id: &str) -> PolicyDraftDocument { + serde_json::from_value(serde_json::json!({ + "$schema": now_policy::POLICY_DRAFT_SCHEMA_URI, + "PolicyVersion": "1.0.0", + "PolicyType": "PackageBrokerPolicy", + "Metadata": { "Id": id, "Publisher": "Test" }, + "Enforcement": { "DefaultDecision": "Deny", "RulePrecedence": "PriorityThenDeny" }, + "Rules": [] + })) + .expect("valid draft") + } + fn policy(id: &str, revision: u32) -> PolicyDocument { + draft(id) + .into_policy_document(revision, Utc::now()) + .expect("valid committed policy") + } + fn request( + store: &PolicyStore, + operation: PolicyReplacementOperation, + raw: serde_json::Value, + ) -> PolicyReplacementRequest { + let validation = store.validate_draft(&raw); + PolicyReplacementRequest { + request_kind: PolicyReplacementRequestKind, + request_version: API_VERSION_STR.into(), + expected_store_token: store.management_snapshot().store_token, + operation, + conflict_handling: PolicyConflictHandling::Reject, + warnings_acknowledged: false, + draft: raw, + validation_receipt: validation.validation_receipt.expect("valid receipt"), + } + } + fn warning() -> PolicyFinding { + PolicyFinding { + finding_version: "1.0".into(), + severity: PolicyFindingSeverity::Warning, + code: PolicyFindingCode::DefaultAllow, + path: "/Enforcement/DefaultDecision".to_owned(), + rule_id: None, + arguments: Default::default(), + message: "warning".to_owned(), + } + } + #[test] + fn receipt_is_stable_for_one_key_and_exact_input() { + let key = ReceiptKey::generate(); + let draft = draft("policy-a"); + let first = key.issue("v1", &draft, &[]); + let second = key.issue("v1", &draft, &[]); + assert_eq!(first, second); + assert!(key.verify("v1", &draft, &[], &first)); + } + #[test] + fn receipt_rejects_other_keys_and_tampering() { + let key = ReceiptKey::generate(); + let original = draft("policy-a"); + let receipt = key.issue("v1", &original, &[]); + assert!(!ReceiptKey::generate().verify("v1", &original, &[], &receipt)); + assert!(!key.verify("v1", &draft("policy-b"), &[], &receipt)); + assert!(!key.verify("v2", &original, &[], &receipt)); + assert!(!key.verify("v1", &original, &[warning()], &receipt)); + } + #[test] + fn receipt_ignores_diagnostic_location_but_binds_semantic_warning_identity() { + let key = ReceiptKey::generate(); + let draft = draft("policy-a"); + let original = warning(); + let receipt = key.issue("v1", &draft, std::slice::from_ref(&original)); + let mut relocated = original; + relocated.path = "/Rules/0".to_owned(); + relocated.message = "localized warning".to_owned(); + assert!(key.verify("v1", &draft, std::slice::from_ref(&relocated), &receipt)); + relocated + .arguments + .insert("option".to_owned(), serde_json::json!("SkipHashCheck")); + assert!(!key.verify("v1", &draft, &[relocated], &receipt)); + } + #[test] + fn malformed_receipts_are_rejected() { + let key = ReceiptKey::generate(); + let draft = draft("policy-a"); + for receipt in ["invalid", "hmac-sha256:not-hex", "hmac-sha256:00"] { + assert!(!key.verify("v1", &draft, &[], &receipt.into())); + } + } + #[test] + fn unsafe_path_shape_reports_insecure_storage() { + let observation = observe_file( + PolicyConfigurationSource::ConfiguredPath, + PathBuf::from(r"..\policy.json").as_path(), + ); + assert_eq!(observation.state, PolicyManagementState::Invalid); + assert_eq!(observation.write_capability, PolicyWriteCapability::Unsupported); + assert_eq!(observation.read_only_reason, Some(PolicyReadOnlyReason::UnsafePath)); + assert_eq!( + observation.invalid_diagnostics.expect("invalid diagnostics").findings[0].message, + "the configured policy file failed storage security validation" + ); + } + #[tokio::test] + async fn store_rejects_stale_request_after_retargeting_and_tampered_receipts() { + let storage = Arc::new(TestStorage::new(Some(policy("current", 1)))); + let store = PolicyStore::load_with_storage( + Some(PathBuf::from(r"C:\policy.json")), + Arc::clone(&storage) as Arc, + Monitoring::Available, + ); + let raw = serde_json::to_value(draft("current")).expect("serialize draft"); + let mut stale_request = request(&store, PolicyReplacementOperation::Update, raw.clone()); + storage.set_disk_state(Some(policy("retargeted", 9)), false, 9); + stale_request.conflict_handling = PolicyConflictHandling::ConfirmOverwrite; + let stale_error = store.replace(stale_request).await.expect_err("stale token rejected"); + assert_eq!(stale_error.code, ErrorCode::StalePolicyStoreToken); + assert!(stale_error.management.is_some()); + assert_eq!( + store.active_policy().expect("retargeted policy loaded").metadata.id.0, + "retargeted" + ); + let mut tampered = request(&store, PolicyReplacementOperation::Update, raw); + tampered.draft["Metadata"]["Publisher"] = "Tampered".into(); + let receipt_error = store.replace(tampered).await.expect_err("tampered draft rejected"); + assert_eq!(receipt_error.code, ErrorCode::ValidationFailed); + } + #[tokio::test] + async fn store_requires_warning_acknowledgement() { + let store = PolicyStore::for_tests(None); + let mut risky = serde_json::to_value(draft("risky")).expect("serialize draft"); + risky["Rules"] = serde_json::Value::Array( + (0..64) + .map(|index| { + serde_json::json!({ + "Id": format!("allow-{index}"), + "Priority": index, + "Decision": "Allow", + "Match": { "Managers": ["Winget"] } + }) + }) + .collect(), + ); + let validation = store.validate_draft(&risky); + let repeated = store.validate_draft(&risky); + assert!(validation.is_valid); + assert!(validation.canonical_draft.is_some()); + assert!(validation.validation_receipt.is_some()); + assert_eq!(validation.validation_receipt, repeated.validation_receipt); + assert_eq!( + serde_json::to_value(&validation.findings).expect("serialize findings"), + serde_json::to_value(&repeated.findings).expect("serialize repeated findings") + ); + assert_eq!(validation.findings.len(), 128); + assert!( + validation + .findings + .iter() + .all(|finding| finding.severity == PolicyFindingSeverity::Warning) + ); + let serialized = serde_json::to_value(&validation).expect("serialize complete validation result"); + let round_trip: PolicyValidationResult = + serde_json::from_value(serialized.clone()).expect("deserialize complete validation result"); + assert_eq!( + serde_json::to_value(round_trip).expect("serialize round-tripped validation result"), + serialized + ); + let mut replacement = request(&store, PolicyReplacementOperation::Create, risky); + let error = store + .replace(replacement.clone()) + .await + .expect_err("warning must be acknowledged"); + assert_eq!(error.code, ErrorCode::WarningConfirmationRequired); + replacement.warnings_acknowledged = true; + store.replace(replacement).await.expect("acknowledged warning succeeds"); + } + #[tokio::test] + async fn canonical_sensitive_warnings_accept_the_original_receipt() { + let options = [ + ("SkipHashCheck", "SkipHashCheck", "AllowSkipHashCheck"), + ("PreRelease", "PreRelease", "AllowPreRelease"), + ( + "AllowCustomInstallLocation", + "HasCustomInstallLocation", + "AllowCustomInstallLocation", + ), + ("AllowPrePostCommands", "HasPrePostCommands", "AllowPrePostCommands"), + ( + "AllowKillBeforeOperation", + "HasKillBeforeOperation", + "AllowKillBeforeOperation", + ), + ( + "AllowUninstallPrevious", + "HasUninstallPrevious", + "AllowUninstallPrevious", + ), + ("AllowCustomParameters", "HasCustomParameters", "AllowCustomParameters"), + ]; + for (option, match_field, constraint_field) in options { + for explicit in ["Constraint", "EmptyMatch", "Default"] { + let store = PolicyStore::for_tests(None); + let mut raw = serde_json::to_value(draft(&format!("{option}-{explicit}"))).expect("serialize draft"); + let mut rule = serde_json::json!({ + "Id": "allow-sensitive", + "Priority": 1, + "Decision": "Allow", + "Match": { "Managers": ["Winget"] } + }); + match explicit { + "Constraint" => { + rule["Constraints"] = serde_json::json!({}); + rule["Constraints"][constraint_field] = serde_json::json!(true); + } + "EmptyMatch" => rule["Match"][match_field] = serde_json::json!([]), + "Default" => {} + _ => unreachable!(), + } + raw["Rules"] = serde_json::json!([rule]); + let validation = store.validate_draft(&raw); + assert!(validation.is_valid, "{option} via {explicit}"); + assert!( + validation + .findings + .iter() + .any(|finding| finding.arguments.get("option") == Some(&serde_json::json!(option))), + "missing {option} warning via {explicit}" + ); + let receipt = validation.validation_receipt.expect("valid receipt"); + let canonical = serde_json::to_value(validation.canonical_draft.expect("valid canonical draft")) + .expect("serialize canonical draft"); + let replacement = PolicyReplacementRequest { + request_kind: PolicyReplacementRequestKind, + request_version: API_VERSION_STR.into(), + expected_store_token: store.management_snapshot().store_token, + operation: PolicyReplacementOperation::Create, + conflict_handling: PolicyConflictHandling::Reject, + warnings_acknowledged: true, + draft: canonical.clone(), + validation_receipt: receipt.clone(), + }; + if option == "SkipHashCheck" && explicit == "Constraint" { + let mut changed = replacement.clone(); + changed.draft["Rules"][0]["Constraints"]["AllowSkipHashCheck"] = serde_json::json!(false); + let error = store + .replace(changed) + .await + .expect_err("meaningful option change invalidates receipt"); + assert_eq!(error.code, ErrorCode::ValidationFailed); + } + store + .replace(replacement) + .await + .unwrap_or_else(|error| panic!("{option} via {explicit} failed: {error:?}")); + } + } + } + #[tokio::test] + async fn all_replacement_operations_enforce_state_identity_and_revision() { + let create = PolicyStore::for_tests(None); + let raw = serde_json::to_value(draft("created")).expect("serialize draft"); + let created = create + .replace(request(&create, PolicyReplacementOperation::Create, raw)) + .await + .expect("create succeeds"); + assert_eq!(created.policy.metadata.revision, 1); + let update = PolicyStore::for_tests(Some(policy("current", 7))); + let raw = serde_json::to_value(draft("current")).expect("serialize draft"); + let updated = update + .replace(request(&update, PolicyReplacementOperation::Update, raw)) + .await + .expect("update succeeds"); + assert_eq!(updated.policy.metadata.revision, 8); + let replace = PolicyStore::for_tests(Some(policy("current", 7))); + let raw = serde_json::to_value(draft("replacement")).expect("serialize draft"); + let replaced = replace + .replace(request(&replace, PolicyReplacementOperation::ReplaceIdentity, raw)) + .await + .expect("identity replacement succeeds"); + assert_eq!(replaced.policy.metadata.revision, 1); + let storage = Arc::new(TestStorage::invalid()); + let repair = PolicyStore::load_with_storage( + Some(PathBuf::from(r"C:\policy.json")), + Arc::clone(&storage) as Arc, + Monitoring::Available, + ); + let raw = serde_json::to_value(draft("repaired")).expect("serialize draft"); + let repaired = repair + .replace(request(&repair, PolicyReplacementOperation::Repair, raw)) + .await + .expect("repair succeeds"); + assert_eq!(repaired.policy.metadata.revision, 1); + let wrong_identity = PolicyStore::for_tests(Some(policy("current", 1))); + let raw = serde_json::to_value(draft("different")).expect("serialize draft"); + let error = wrong_identity + .replace(request(&wrong_identity, PolicyReplacementOperation::Update, raw)) + .await + .expect_err("update must preserve identity"); + assert_eq!(error.code, ErrorCode::Conflict); + } + #[tokio::test] + async fn concurrent_writers_cannot_commit_from_one_token() { + let store = PolicyStore::for_tests(Some(policy("current", 1))); + let raw = serde_json::to_value(draft("current")).expect("serialize draft"); + let first = request(&store, PolicyReplacementOperation::Update, raw); + let (first, second) = tokio::join!(store.replace(first.clone()), store.replace(first)); + let outcomes = [first, second]; + assert_eq!(outcomes.iter().filter(|result| result.is_ok()).count(), 1); + assert_eq!( + outcomes + .iter() + .filter(|result| matches!(result, Err(error) if error.code == ErrorCode::StalePolicyStoreToken)) + .count(), + 1 + ); + } + #[tokio::test] + async fn failed_persistence_preserves_snapshot_and_reload_rotates_token() { + let storage = Arc::new(TestStorage::new(Some(policy("current", 3)))); + let store = PolicyStore::load_with_storage( + Some(PathBuf::from(r"C:\policy.json")), + Arc::clone(&storage) as Arc, + Monitoring::Available, + ); + storage.fail_persist.store(true, std::sync::atomic::Ordering::SeqCst); + let raw = serde_json::to_value(draft("current")).expect("serialize draft"); + let error = store + .replace(request(&store, PolicyReplacementOperation::Update, raw)) + .await + .expect_err("persistence failure"); + assert_eq!(error.code, ErrorCode::PolicyPersistenceFailed); + assert_eq!(store.active_policy().expect("old policy remains").metadata.revision, 3); + let old_token = store.management_snapshot().store_token; + storage.set_disk_state(None, true, 2); + let management = store.reload_from_disk(ReloadCause::ExternalChange).await; + assert_eq!(management.state, PolicyManagementState::Invalid); + assert_ne!(management.store_token, old_token); + assert!(store.active_policy().is_none()); + } + #[tokio::test] + async fn readiness_reloads_each_disk_state_after_provisional_load() { + for (disk_policy, invalid, expected) in [ + (Some(policy("changed", 2)), false, PolicyManagementState::Active), + (None, false, PolicyManagementState::Missing), + (None, true, PolicyManagementState::Invalid), + ] { + let storage = Arc::new(TestStorage::new(Some(policy("provisional", 1)))); + let store = PolicyStore::load_with_storage( + Some(PathBuf::from(r"C:\policy.json")), + Arc::clone(&storage) as Arc, + Monitoring::Initializing, + ); + storage.set_disk_state(disk_policy, invalid, 3); + assert_eq!(store.mark_monitoring_ready().await.state, expected); + assert_eq!( + store.active_policy().is_some(), + expected == PolicyManagementState::Active + ); + } + } + #[tokio::test] + async fn monitoring_failure_stays_unavailable_across_reload_and_put() { + for failure in [ + WatcherFailure::Creation, + WatcherFailure::Registration, + WatcherFailure::Notification, + WatcherFailure::ChannelClosed, + WatcherFailure::TaskTerminated, + ] { + let storage = Arc::new(TestStorage::new(Some(policy("current", 3)))); + let store = PolicyStore::load_with_storage( + Some(PathBuf::from(r"C:\policy.json")), + Arc::clone(&storage) as Arc, + Monitoring::Available, + ); + fail_closed(&store, failure).await; + let first_unavailable = store.management_snapshot(); + fail_closed(&store, failure).await; + let unavailable = store.management_snapshot(); + assert_eq!(unavailable.store_token, first_unavailable.store_token); + assert_eq!(unavailable.write_capability, PolicyWriteCapability::ReadOnly); + assert_eq!( + unavailable.read_only_reason, + Some(PolicyReadOnlyReason::ManagementDisabled) + ); + storage.set_disk_state(Some(policy("external", 9)), false, 9); + assert_eq!( + store.reload_from_disk(ReloadCause::ExternalChange).await.store_token, + unavailable.store_token + ); + for token in [PolicyStoreToken::from("store:stale"), unavailable.store_token.clone()] { + let mut replacement = request( + &store, + PolicyReplacementOperation::Update, + serde_json::to_value(draft("current")).expect("serialize draft"), + ); + replacement.expected_store_token = token; + let error = store + .replace(replacement) + .await + .expect_err("monitoring failure blocks PUT"); + assert_eq!(error.code, ErrorCode::BrokerPaused); + assert_eq!( + error.management.expect("management snapshot").store_token, + unavailable.store_token + ); + } + assert!(store.active_policy().is_none()); + assert_eq!( + storage + .observation + .lock() + .policy + .as_ref() + .expect("disk policy") + .metadata + .revision, + 9 + ); + } + } + #[test] + fn revision_planning_rejects_invalid_state_transitions_and_overflow() { + assert!( + plan_revision( + PolicyReplacementOperation::Create, + PolicyManagementState::Active, + None, + "id" + ) + .is_err() + ); + assert!( + plan_revision( + PolicyReplacementOperation::Repair, + PolicyManagementState::Missing, + None, + "id" + ) + .is_err() + ); + assert!( + plan_revision( + PolicyReplacementOperation::Update, + PolicyManagementState::Active, + Some(&policy("id", i32::MAX as u32)), + "id" + ) + .is_err() + ); + } +} diff --git a/crates/now-package-broker/src/policy_store/validation.rs b/crates/now-package-broker/src/policy_store/validation.rs new file mode 100644 index 000000000..92a0758f6 --- /dev/null +++ b/crates/now-package-broker/src/policy_store/validation.rs @@ -0,0 +1,1271 @@ +//! Strict deterministic validation for editable policy documents. + +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; + +use now_policy::{Decision, PolicyConstraints, PolicyDraftDocument, PolicyDraftMetadata, PolicyMatch, PolicyRule}; +use now_policy_api::{ + API_VERSION_STR, PolicyFinding, PolicyFindingCode, PolicyFindingSeverity, PolicyValidationResult, +}; + +pub(super) const VALIDATOR_VERSION: &str = "now-package-broker-policy-validator/8"; +const MAX_RULES: usize = 1024; +const MAX_RULE_PRIORITY: u32 = i32::MAX as u32; +const MAX_FINDING_MESSAGE_CHARS: usize = 2048; +const MAX_FINDINGS: usize = 128; +const MATCH_COLLECTION_MAXIMA: &[(&str, usize)] = &[ + ("Operations", 3), + ("Managers", 16), + ("Sources", 128), + ("PackageIdentifiers", 1024), + ("PackageNames", 1024), + ("Versions", 256), + ("Scopes", 2), + ("Architectures", 5), + ("Elevation", 2), +]; +const BOOLEAN_MATCH_FIELDS: &[&str] = &[ + "Interactive", + "SkipHashCheck", + "PreRelease", + "HasCustomParameters", + "HasCustomInstallLocation", + "HasPrePostCommands", + "HasKillBeforeOperation", + "HasUninstallPrevious", +]; +const CONSTRAINT_COLLECTION_MAXIMA: &[(&str, usize)] = &[ + ("AllowedInstallLocationPatterns", 64), + ("AllowedCustomParameters", 128), + ("AllowedCustomParameterPatterns", 128), + ("DeniedCustomParameters", 128), +]; +struct Findings { + values: Vec, + has_error: bool, +} +impl Findings { + fn new() -> Self { + Self { + values: Vec::with_capacity(MAX_FINDINGS), + has_error: false, + } + } + fn push(&mut self, finding: PolicyFinding) { + let is_error = finding.severity == PolicyFindingSeverity::Error; + self.has_error |= is_error; + if self.values.len() < MAX_FINDINGS { + self.values.push(finding); + } else if is_error + && !self + .values + .iter() + .any(|existing| existing.severity == PolicyFindingSeverity::Error) + { + self.values[MAX_FINDINGS - 1] = finding; + } + } + fn is_saturated(&self) -> bool { + self.values.len() == MAX_FINDINGS + } +} +pub(super) fn validate_draft(raw: &serde_json::Value) -> PolicyValidationResult { + let mut findings = Findings::new(); + if !raw.is_object() { + findings.push(error( + PolicyFindingCode::SchemaViolation, + "", + "the policy draft must be a JSON object", + )); + return invalid_result(findings); + } + check_constant( + raw, + "$schema", + "/$schema", + now_policy::POLICY_DRAFT_SCHEMA_URI, + PolicyFindingCode::UnsupportedSchema, + &mut findings, + ); + check_constant( + raw, + "PolicyType", + "/PolicyType", + "PackageBrokerPolicy", + PolicyFindingCode::UnsupportedPolicyType, + &mut findings, + ); + check_policy_version(raw, &mut findings); + if has_error(&findings) { + return invalid_result(findings); + } + if check_raw_collection_bounds(raw, &mut findings) { + return invalid_result(findings); + } + match serde_json::from_value::(raw.clone()) { + Ok(draft) => { + semantic_checks(raw, &draft, &mut findings); + if has_error(&findings) { + invalid_result(findings) + } else { + valid_result(draft, findings) + } + } + Err(parse_error) => { + findings.push(classify_parse_error(&parse_error)); + invalid_result(findings) + } + } +} +pub(super) fn validate_committed_policy(policy: &now_policy::PolicyDocument) -> PolicyValidationResult { + let raw = serde_json::to_value(policy.to_draft()).expect("committed policy draft serializes"); + validate_draft(&raw) +} +fn has_error(findings: &Findings) -> bool { + findings.has_error +} +fn invalid_result(findings: Findings) -> PolicyValidationResult { + PolicyValidationResult { + result_version: API_VERSION_STR.into(), + validator_version: VALIDATOR_VERSION.to_owned(), + is_valid: false, + canonical_draft: None, + validation_receipt: None, + findings: findings.values, + } +} +fn valid_result(draft: PolicyDraftDocument, findings: Findings) -> PolicyValidationResult { + PolicyValidationResult { + result_version: API_VERSION_STR.into(), + validator_version: VALIDATOR_VERSION.to_owned(), + is_valid: true, + canonical_draft: Some(draft), + validation_receipt: None, + findings: findings.values, + } +} +fn finding( + severity: PolicyFindingSeverity, + code: PolicyFindingCode, + path: impl Into, + message: impl Into, +) -> PolicyFinding { + let message = message.into(); + let message = if message.chars().count() <= MAX_FINDING_MESSAGE_CHARS { + message + } else { + let mut bounded = message.chars().take(MAX_FINDING_MESSAGE_CHARS - 3).collect::(); + bounded.push_str("..."); + bounded + }; + PolicyFinding { + finding_version: API_VERSION_STR.into(), + severity, + code, + path: path.into(), + rule_id: None, + arguments: BTreeMap::new(), + message, + } +} +fn error(code: PolicyFindingCode, path: impl Into, message: impl Into) -> PolicyFinding { + finding(PolicyFindingSeverity::Error, code, path, message) +} +fn warning(code: PolicyFindingCode, path: impl Into, message: impl Into) -> PolicyFinding { + finding(PolicyFindingSeverity::Warning, code, path, message) +} +fn rule_finding( + rule: &PolicyRule, + severity: PolicyFindingSeverity, + code: PolicyFindingCode, + path: impl Into, + message: impl Into, +) -> PolicyFinding { + let mut finding = finding(severity, code, path, message); + finding.rule_id = Some(now_policy_api::ResourceId::from(rule.id.0.as_str())); + finding +} +fn check_constant( + raw: &serde_json::Value, + key: &str, + path: &str, + expected: &str, + mismatch_code: PolicyFindingCode, + findings: &mut Findings, +) { + match raw.get(key) { + None => findings.push(error( + PolicyFindingCode::MissingRequiredField, + path, + format!("missing required field '{key}'"), + )), + Some(serde_json::Value::String(value)) if value == expected => {} + Some(serde_json::Value::String(value)) => findings.push(error( + mismatch_code, + path, + format!("unsupported value '{value}'; expected '{expected}'"), + )), + Some(_) => findings.push(error( + PolicyFindingCode::InvalidFieldType, + path, + format!("'{key}' must be a string"), + )), + } +} +fn check_policy_version(raw: &serde_json::Value, findings: &mut Findings) { + const PATH: &str = "/PolicyVersion"; + match raw.get("PolicyVersion") { + None => findings.push(error( + PolicyFindingCode::MissingRequiredField, + PATH, + "missing required field 'PolicyVersion'", + )), + Some(serde_json::Value::String(value)) if value.len() > 128 => findings.push(error( + PolicyFindingCode::InvalidFieldValue, + PATH, + "PolicyVersion exceeds the maximum length of 128", + )), + Some(serde_json::Value::String(value)) => match semver::Version::parse(value) { + Ok(version) if version.major == 1 => {} + Ok(version) => findings.push(error( + PolicyFindingCode::UnsupportedPolicyVersion, + PATH, + format!("unsupported PolicyVersion major '{}'; expected 1.x", version.major), + )), + Err(parse_error) => findings.push(error( + PolicyFindingCode::InvalidFieldValue, + PATH, + format!("PolicyVersion is not a valid semantic version: {parse_error}"), + )), + }, + Some(_) => findings.push(error( + PolicyFindingCode::InvalidFieldType, + PATH, + "'PolicyVersion' must be a string", + )), + } +} +pub(crate) fn classify_parse_error(parse_error: &serde_json::Error) -> PolicyFinding { + let message = parse_error.to_string(); + let code = if message.contains("boolean match arrays") { + PolicyFindingCode::IneffectiveBooleanMatch + } else if message.contains("missing field") { + PolicyFindingCode::MissingRequiredField + } else if message.contains("unknown field") { + PolicyFindingCode::UnknownField + } else if message.contains("invalid type") { + PolicyFindingCode::InvalidFieldType + } else { + PolicyFindingCode::InvalidFieldValue + }; + error( + code, + "", + format!("policy draft does not match the expected schema: {message}"), + ) +} +fn check_raw_collection_bounds(raw: &serde_json::Value, findings: &mut Findings) -> bool { + let Some(rules) = raw.get("Rules").and_then(serde_json::Value::as_array) else { + return false; + }; + if rules.len() > MAX_RULES { + check_max_len(rules.len(), MAX_RULES, "/Rules", findings); + return true; + } + for (index, rule) in rules.iter().enumerate() { + let Some(rule) = rule.as_object() else { + continue; + }; + let base = format!("/Rules/{index}"); + if let Some(matches) = rule.get("Match").and_then(serde_json::Value::as_object) { + for &(field, max) in MATCH_COLLECTION_MAXIMA { + check_raw_set_array(matches, field, max, &format!("{base}/Match/{field}"), findings); + } + for &field in BOOLEAN_MATCH_FIELDS { + if matches + .get(field) + .and_then(serde_json::Value::as_array) + .is_some_and(|values| values.len() > 1) + { + findings.push(error( + PolicyFindingCode::IneffectiveBooleanMatch, + format!("{base}/Match/{field}"), + "boolean match arrays may contain at most one value", + )); + } + } + } + if let Some(constraints) = rule.get("Constraints").and_then(serde_json::Value::as_object) { + for &(field, max) in CONSTRAINT_COLLECTION_MAXIMA { + check_raw_array_len( + constraints, + field, + max, + &format!("{base}/Constraints/{field}"), + findings, + ); + } + } + if findings.is_saturated() { + break; + } + } + has_error(findings) +} +fn check_raw_array_len( + object: &serde_json::Map, + field: &str, + max: usize, + path: &str, + findings: &mut Findings, +) { + if let Some(values) = object.get(field).and_then(serde_json::Value::as_array) { + check_max_len(values.len(), max, path, findings); + } +} +fn check_raw_set_array( + object: &serde_json::Map, + field: &str, + max: usize, + path: &str, + findings: &mut Findings, +) { + let Some(values) = object.get(field).and_then(serde_json::Value::as_array) else { + return; + }; + check_max_len(values.len(), max, path, findings); + if values.len() > max { + return; + } + let mut seen = HashSet::with_capacity(values.len()); + for value in values { + let Some(value) = value.as_str() else { + return; + }; + if !seen.insert(value) { + findings.push(error( + PolicyFindingCode::SchemaViolation, + path, + format!("{path} contains duplicate value '{value}'"), + )); + return; + } + } +} +#[derive(Debug, Clone, Copy)] +pub(crate) enum DiskFailureReason { + Unreadable, + InsecureStorage, + MalformedContent, + UnsupportedFormat, + FailedSemanticValidation, + WatcherUnavailable, +} +pub(crate) fn disk_failure_finding(reason: DiskFailureReason) -> PolicyFinding { + let message = match reason { + DiskFailureReason::Unreadable => "the configured policy file could not be opened or read", + DiskFailureReason::InsecureStorage => "the configured policy file failed storage security validation", + DiskFailureReason::MalformedContent => { + "the configured policy file does not contain a policy matching the expected schema" + } + DiskFailureReason::UnsupportedFormat => "the configured policy path uses an unsupported format", + DiskFailureReason::FailedSemanticValidation => "the configured policy file failed semantic validation", + DiskFailureReason::WatcherUnavailable => "policy change monitoring is unavailable", + }; + error(PolicyFindingCode::SchemaViolation, "", message) +} +fn semantic_checks(raw: &serde_json::Value, draft: &PolicyDraftDocument, findings: &mut Findings) { + if draft.rules.len() > MAX_RULES { + check_max_len(draft.rules.len(), MAX_RULES, "/Rules", findings); + return; + } + check_metadata(&draft.metadata, findings); + check_duplicate_rule_ids(&draft.rules, findings); + for (index, rule) in draft.rules.iter().enumerate() { + if findings.is_saturated() { + return; + } + check_rule(index, rule, findings); + } + if has_error(findings) { + return; + } + if draft.enforcement.audit_mode == Some(true) { + findings.push(warning( + PolicyFindingCode::AuditModeEnabled, + "/Enforcement/AuditMode", + "audit mode is enabled; decisions are not enforced", + )); + } + if draft.enforcement.default_decision == Decision::Allow { + findings.push(warning( + PolicyFindingCode::DefaultAllow, + "/Enforcement/DefaultDecision", + "the default decision is Allow", + )); + } + for (index, rule) in draft.rules.iter().enumerate() { + if findings.is_saturated() { + return; + } + check_sensitive_options(raw, index, rule, findings); + } +} +fn check_metadata(metadata: &PolicyDraftMetadata, findings: &mut Findings) { + check_string_len(&metadata.publisher, 1, 128, "/Metadata/Publisher", findings); + if let Some(description) = &metadata.description { + check_string_len(description, 0, 512, "/Metadata/Description", findings); + } + if let (Some(valid_from), Some(valid_until)) = (metadata.valid_from, metadata.valid_until) + && valid_from >= valid_until + { + findings.push(error( + PolicyFindingCode::InvalidValidityInterval, + "/Metadata/ValidUntil", + "ValidUntil must be after ValidFrom", + )); + } +} +fn check_duplicate_rule_ids(rules: &[PolicyRule], findings: &mut Findings) { + let mut seen: HashMap<&str, usize> = HashMap::new(); + for (index, rule) in rules.iter().enumerate() { + if findings.is_saturated() { + return; + } + if let Some(first_index) = seen.insert(&rule.id.0, index) { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::DuplicateRuleId, + format!("/Rules/{index}/Id"), + format!("rule id '{}' duplicates rule at index {first_index}", rule.id), + )); + } + } +} +fn check_rule(index: usize, rule: &PolicyRule, findings: &mut Findings) { + let base = format!("/Rules/{index}"); + let matches = &rule.match_criteria; + if rule.priority > MAX_RULE_PRIORITY { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::InvalidFieldValue, + format!("{base}/Priority"), + format!("Priority exceeds {MAX_RULE_PRIORITY}"), + )); + } + if let Some(reason) = &rule.reason { + check_string_len(reason, 0, 512, &format!("{base}/Reason"), findings); + } + check_max_len(matches.managers.len(), 16, &format!("{base}/Match/Managers"), findings); + check_max_len(matches.sources.len(), 128, &format!("{base}/Match/Sources"), findings); + check_max_len( + matches.package_identifiers.len(), + 1024, + &format!("{base}/Match/PackageIdentifiers"), + findings, + ); + check_max_len( + matches.package_names.len(), + 1024, + &format!("{base}/Match/PackageNames"), + findings, + ); + check_max_len(matches.versions.len(), 256, &format!("{base}/Match/Versions"), findings); + if !matches.package_names.is_empty() { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::InvalidFieldValue, + format!("{base}/Match/PackageNames"), + "PackageNames is unsupported because requests do not provide a package display name", + )); + } + check_version_range(index, rule, findings); + check_patterns( + index, + rule, + "Match/Sources", + matches.sources.iter().take(128).map(AsRef::as_ref), + findings, + ); + check_patterns( + index, + rule, + "Match/PackageIdentifiers", + matches.package_identifiers.iter().take(1024).map(AsRef::as_ref), + findings, + ); + if let Some(constraints) = &rule.constraints { + check_constraints(index, rule, constraints, findings); + } +} +fn check_constraints(index: usize, rule: &PolicyRule, constraints: &PolicyConstraints, findings: &mut Findings) { + let base = format!("/Rules/{index}/Constraints"); + check_max_len( + constraints.allowed_install_location_patterns.len(), + 64, + &format!("{base}/AllowedInstallLocationPatterns"), + findings, + ); + check_max_len( + constraints.allowed_custom_parameters.len(), + 128, + &format!("{base}/AllowedCustomParameters"), + findings, + ); + check_max_len( + constraints.allowed_custom_parameter_patterns.len(), + 128, + &format!("{base}/AllowedCustomParameterPatterns"), + findings, + ); + check_max_len( + constraints.denied_custom_parameters.len(), + 128, + &format!("{base}/DeniedCustomParameters"), + findings, + ); + check_patterns( + index, + rule, + "Constraints/AllowedInstallLocationPatterns", + constraints + .allowed_install_location_patterns + .iter() + .take(64) + .map(AsRef::as_ref), + findings, + ); + check_patterns( + index, + rule, + "Constraints/AllowedCustomParameterPatterns", + constraints + .allowed_custom_parameter_patterns + .iter() + .take(128) + .map(AsRef::as_ref), + findings, + ); + let matches = &rule.match_criteria; + for (values, allowed, name) in [ + (&matches.interactive, constraints.allow_interactive, "Interactive"), + ( + &matches.skip_hash_check, + constraints.allow_skip_hash_check, + "SkipHashCheck", + ), + (&matches.pre_release, constraints.allow_pre_release, "PreRelease"), + ( + &matches.has_custom_install_location, + constraints.allow_custom_install_location, + "HasCustomInstallLocation", + ), + ( + &matches.has_custom_parameters, + constraints.allow_custom_parameters, + "HasCustomParameters", + ), + ( + &matches.has_pre_post_commands, + constraints.allow_pre_post_commands, + "HasPrePostCommands", + ), + ( + &matches.has_kill_before_operation, + constraints.allow_kill_before_operation, + "HasKillBeforeOperation", + ), + ( + &matches.has_uninstall_previous, + constraints.allow_uninstall_previous, + "HasUninstallPrevious", + ), + ] { + if findings.is_saturated() { + return; + } + if !allowed && values.len() == 1 && values.contains(&true) { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::ContradictoryConstraints, + &base, + format!("rule requires {name}=true but its constraints deny {name}"), + )); + } + } +} +fn check_version_range(index: usize, rule: &PolicyRule, findings: &mut Findings) { + let Some(range) = &rule.match_criteria.version_range else { + return; + }; + let base = format!("/Rules/{index}/Match/VersionRange"); + let min = parse_version_bound( + range.min_version.as_deref(), + &format!("{base}/MinVersion"), + rule, + findings, + ); + let max = parse_version_bound( + range.max_version.as_deref(), + &format!("{base}/MaxVersion"), + rule, + findings, + ); + if range.min_version.is_none() && range.max_version.is_none() { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::EmptyVersionRange, + &base, + "version range must specify MinVersion or MaxVersion", + )); + } else if let (Some(min), Some(max)) = (min.as_ref(), max.as_ref()) + && min > max + { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::EmptyVersionRange, + &base, + "MinVersion is greater than MaxVersion", + )); + } else if !range.include_prerelease + && let Some(max) = max + && let Some(mut first_stable) = + min.or_else(|| range.min_version.is_none().then(|| semver::Version::new(0, 0, 0))) + { + first_stable.pre = semver::Prerelease::EMPTY; + first_stable.build = semver::BuildMetadata::EMPTY; + if max < first_stable { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::EmptyVersionRange, + &base, + "version range contains no stable version", + )); + } + } +} +fn parse_version_bound( + value: Option<&str>, + path: &str, + rule: &PolicyRule, + findings: &mut Findings, +) -> Option { + let value = value?; + if value.is_empty() || value.len() > 128 { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::InvalidVersionRange, + path, + "version bound must contain 1 to 128 characters", + )); + return None; + } + match semver::Version::parse(value) { + Ok(version) => Some(version), + Err(parse_error) => { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::InvalidVersionRange, + path, + format!("invalid semantic version: {parse_error}"), + )); + None + } + } +} +fn check_patterns>( + index: usize, + rule: &PolicyRule, + field: &str, + patterns: impl Iterator, + findings: &mut Findings, +) { + for pattern in patterns { + if findings.is_saturated() { + return; + } + let pattern = pattern.as_ref(); + let regex = format!("^{}$", regex::escape(pattern).replace(r"\*", ".*")); + if regex::RegexBuilder::new(®ex).case_insensitive(true).build().is_err() { + findings.push(rule_finding( + rule, + PolicyFindingSeverity::Error, + PolicyFindingCode::InvalidWildcardPattern, + format!("/Rules/{index}/{field}"), + "wildcard pattern is too complex to evaluate", + )); + } + } +} +fn check_sensitive_options(raw: &serde_json::Value, index: usize, rule: &PolicyRule, findings: &mut Findings) { + if !rule.enabled || rule.decision != Decision::Allow { + return; + } + let defaults = PolicyConstraints::default(); + let constraints = rule.constraints.as_ref().unwrap_or(&defaults); + let matches: &PolicyMatch = &rule.match_criteria; + let reachable = |values: &BTreeSet| values.is_empty() || values.contains(&true); + let options = [ + ( + constraints.allow_skip_hash_check && reachable(&matches.skip_hash_check), + "SkipHashCheck", + "SkipHashCheck", + "AllowSkipHashCheck", + ), + ( + constraints.allow_pre_release && reachable(&matches.pre_release), + "PreRelease", + "PreRelease", + "AllowPreRelease", + ), + ( + constraints.allow_custom_install_location && reachable(&matches.has_custom_install_location), + "AllowCustomInstallLocation", + "HasCustomInstallLocation", + "AllowCustomInstallLocation", + ), + ( + constraints.allow_pre_post_commands && reachable(&matches.has_pre_post_commands), + "AllowPrePostCommands", + "HasPrePostCommands", + "AllowPrePostCommands", + ), + ( + constraints.allow_kill_before_operation && reachable(&matches.has_kill_before_operation), + "AllowKillBeforeOperation", + "HasKillBeforeOperation", + "AllowKillBeforeOperation", + ), + ( + constraints.allow_uninstall_previous && reachable(&matches.has_uninstall_previous), + "AllowUninstallPrevious", + "HasUninstallPrevious", + "AllowUninstallPrevious", + ), + ( + constraints.allow_custom_parameters + && reachable(&matches.has_custom_parameters) + && !constraints + .denied_custom_parameters + .iter() + .take(128) + .any(|pattern| pattern.as_ref() == "*"), + "AllowCustomParameters", + "HasCustomParameters", + "AllowCustomParameters", + ), + ]; + for (enabled, option, match_field, constraint_field) in options { + if findings.is_saturated() { + return; + } + if enabled { + let rule_path = format!("/Rules/{index}"); + let match_path = format!("{rule_path}/Match/{match_field}"); + let constraint_path = format!("{rule_path}/Constraints/{constraint_field}"); + let path = if raw.pointer(&match_path).is_some() { + match_path + } else if raw.pointer(&constraint_path).is_some() { + constraint_path + } else { + rule_path + }; + let mut finding = rule_finding( + rule, + PolicyFindingSeverity::Warning, + PolicyFindingCode::SensitiveOptionAllowed, + path, + format!("rule '{}' allows {option}", rule.id), + ); + finding + .arguments + .insert("option".to_owned(), serde_json::Value::from(option)); + findings.push(finding); + } + } +} +fn check_string_len(value: &str, min: usize, max: usize, path: &str, findings: &mut Findings) { + let length = value.chars().count(); + if !(min..=max).contains(&length) { + findings.push(error( + PolicyFindingCode::SchemaViolation, + path, + format!("{path} must contain between {min} and {max} characters"), + )); + } +} +fn check_max_len(len: usize, max: usize, path: &str, findings: &mut Findings) { + if len > max { + findings.push(error( + PolicyFindingCode::SchemaViolation, + path, + format!("{path} has {len} entries, exceeding the maximum of {max}"), + )); + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + fn draft() -> serde_json::Value { + json!({ + "$schema": now_policy::POLICY_DRAFT_SCHEMA_URI, + "PolicyVersion": "1.0.0", + "PolicyType": "PackageBrokerPolicy", + "Metadata": { "Id": "policy-a", "Publisher": "Test" }, + "Enforcement": { "DefaultDecision": "Deny", "RulePrecedence": "PriorityThenDeny" }, + "Rules": [] + }) + } + fn rule(id: &str, match_value: serde_json::Value) -> serde_json::Value { + json!({ + "Id": id, + "Priority": 1, + "Decision": "Deny", + "Match": match_value + }) + } + fn has_code(result: &PolicyValidationResult, code: PolicyFindingCode) -> bool { + result.findings.iter().any(|finding| finding.code == code) + } + #[test] + fn strict_valid_draft_is_canonicalized_deterministically() { + let raw = draft(); + let first = validate_draft(&raw); + let second = validate_draft(&raw); + assert!(first.is_valid); + assert_eq!( + serde_json::to_value(first.canonical_draft).expect("serialize canonical draft"), + serde_json::to_value(second.canonical_draft).expect("serialize canonical draft") + ); + } + #[test] + fn constants_and_unknown_fields_are_rejected() { + for (pointer, value, code) in [ + ("/$schema", json!("wrong"), PolicyFindingCode::UnsupportedSchema), + ( + "/PolicyType", + json!("OtherPolicy"), + PolicyFindingCode::UnsupportedPolicyType, + ), + ( + "/PolicyVersion", + json!("2.0.0"), + PolicyFindingCode::UnsupportedPolicyVersion, + ), + ] { + let mut raw = draft(); + *raw.pointer_mut(pointer).expect("pointer exists") = value; + assert!(has_code(&validate_draft(&raw), code)); + } + let mut raw = draft(); + raw["Unexpected"] = json!(true); + assert!(has_code(&validate_draft(&raw), PolicyFindingCode::UnknownField)); + } + #[test] + fn structural_bounds_and_duplicate_ids_are_rejected() { + let mut raw = draft(); + raw["Metadata"]["Publisher"] = json!("x".repeat(129)); + assert!(has_code(&validate_draft(&raw), PolicyFindingCode::SchemaViolation)); + let mut raw = draft(); + raw["Rules"] = json!([ + rule("duplicate", json!({ "Managers": ["Winget"] })), + rule("duplicate", json!({ "Managers": ["Npm"] })) + ]); + assert!(has_code(&validate_draft(&raw), PolicyFindingCode::DuplicateRuleId)); + } + #[test] + fn oversized_rules_stop_after_one_structural_finding() { + let mut raw = draft(); + raw["Rules"] = serde_json::Value::Array( + (0..=MAX_RULES) + .map(|_| rule("duplicate", json!({ "Managers": ["Winget"] }))) + .collect(), + ); + let started = std::time::Instant::now(); + let result = validate_draft(&raw); + assert!(started.elapsed() < std::time::Duration::from_secs(5)); + assert!(!result.is_valid); + assert!(result.canonical_draft.is_none()); + assert!(result.validation_receipt.is_none()); + assert_eq!(result.findings.len(), 1); + assert_eq!(result.findings[0].code, PolicyFindingCode::SchemaViolation); + assert_eq!(result.findings[0].path, "/Rules"); + } + #[test] + fn warning_findings_are_capped_without_invalidating_the_draft() { + let mut raw = draft(); + raw["Rules"] = serde_json::Value::Array( + (0..64) + .map(|index| { + let mut value = rule(&format!("allow-{index}"), json!({ "Managers": ["Winget"] })); + value["Decision"] = json!("Allow"); + value + }) + .collect(), + ); + let started = std::time::Instant::now(); + let first = validate_draft(&raw); + let second = validate_draft(&raw); + assert!(started.elapsed() < std::time::Duration::from_secs(5)); + assert!(first.is_valid); + assert!(first.canonical_draft.is_some()); + assert!(first.validation_receipt.is_none()); + assert_eq!(first.findings.len(), MAX_FINDINGS); + assert!( + first + .findings + .iter() + .all(|finding| finding.severity == PolicyFindingSeverity::Warning) + ); + assert_eq!( + serde_json::to_value(&first.findings).expect("serialize findings"), + serde_json::to_value(&second.findings).expect("serialize findings") + ); + } + + #[test] + fn warning_heavy_draft_with_a_late_error_remains_invalid() { + let mut raw = draft(); + let mut rules: Vec<_> = (0..64) + .map(|index| { + let mut value = rule(&format!("allow-{index}"), json!({ "Managers": ["Winget"] })); + value["Decision"] = json!("Allow"); + value + }) + .collect(); + let mut invalid = rule("invalid-last", json!({ "Managers": ["Winget"] })); + invalid["Priority"] = json!(u64::from(MAX_RULE_PRIORITY) + 1); + rules.push(invalid); + raw["Rules"] = serde_json::Value::Array(rules); + let result = validate_draft(&raw); + assert!(!result.is_valid); + assert!(result.canonical_draft.is_none()); + assert!(result.validation_receipt.is_none()); + assert!(result.findings.iter().any(|finding| { + finding.severity == PolicyFindingSeverity::Error + && finding.code == PolicyFindingCode::InvalidFieldValue + && finding.path == "/Rules/64/Priority" + })); + } + #[test] + fn oversized_pattern_collections_have_bounded_ordered_findings() { + let mut raw = draft(); + let sources: Vec<_> = (0..2048).map(|index| json!(format!("source-{index}"))).collect(); + let packages: Vec<_> = (0..2048).map(|index| json!(format!("package-{index}"))).collect(); + raw["Rules"] = json!([rule( + "oversized", + json!({ "Sources": sources, "PackageIdentifiers": packages }) + )]); + let result = validate_draft(&raw); + assert!(!result.is_valid); + assert_eq!(result.findings.len(), 2); + assert_eq!(result.findings[0].path, "/Rules/0/Match/Sources"); + assert_eq!(result.findings[1].path, "/Rules/0/Match/PackageIdentifiers"); + } + #[test] + fn every_schema_array_bound_is_rejected_before_typed_parsing() { + let collections = MATCH_COLLECTION_MAXIMA + .iter() + .map(|&(field, max)| ("Match", field, max)) + .chain(BOOLEAN_MATCH_FIELDS.iter().map(|&field| ("Match", field, 1))) + .chain( + CONSTRAINT_COLLECTION_MAXIMA + .iter() + .map(|&(field, max)| ("Constraints", field, max)), + ); + for (section, field, max) in collections { + let mut raw = draft(); + let mut value = rule("bounded", json!({ "Managers": ["Winget"] })); + value[section][field] = serde_json::Value::Array(vec![json!(false); max + 1]); + raw["Rules"] = json!([value]); + let result = validate_draft(&raw); + assert!(!result.is_valid, "{section}/{field}"); + assert!(result.canonical_draft.is_none(), "{section}/{field}"); + assert!(result.validation_receipt.is_none(), "{section}/{field}"); + assert_eq!(result.findings.len(), 1, "{section}/{field}"); + assert_eq!(result.findings[0].path, format!("/Rules/0/{section}/{field}")); + } + } + + #[test] + fn set_backed_match_arrays_reject_exact_duplicates() { + for (field, value) in [ + ("Operations", "Install"), + ("Managers", "Winget"), + ("Sources", "source"), + ("PackageIdentifiers", "package"), + ("PackageNames", "name"), + ("Versions", "1.0.0"), + ("Scopes", "User"), + ("Architectures", "X64"), + ("Elevation", "Elevated"), + ] { + let mut raw = draft(); + let mut duplicate = rule("duplicate", json!({ "Managers": ["Winget"] })); + duplicate["Match"][field] = json!([value, value]); + raw["Rules"] = json!([duplicate]); + let result = validate_draft(&raw); + assert!(!result.is_valid, "{field}"); + assert!(result.canonical_draft.is_none(), "{field}"); + assert!(result.validation_receipt.is_none(), "{field}"); + assert!(result.findings.iter().any(|finding| { + finding.code == PolicyFindingCode::SchemaViolation + && finding.path == format!("/Rules/0/Match/{field}") + && finding.message.contains("duplicate value") + })); + } + } + + #[test] + fn raw_uniqueness_is_case_sensitive_and_excludes_constraint_vectors() { + let mut raw = draft(); + let mut distinct = rule( + "distinct", + json!({ + "Operations": ["Install", "Update"], + "Managers": ["Winget", "Npm"], + "Sources": ["source", "Source"], + "PackageIdentifiers": ["package", "Package"], + "Versions": ["1.0.0", "2.0.0"], + "Scopes": ["User", "Machine"], + "Architectures": ["X64", "Arm64"], + "Elevation": ["Standard", "Elevated"] + }), + ); + distinct["Constraints"] = json!({ + "AllowedInstallLocationPatterns": ["C:\\Tools", "C:\\Tools"] + }); + raw["Rules"] = json!([distinct]); + let result = validate_draft(&raw); + assert!(result.is_valid); + assert_eq!( + result.canonical_draft.expect("valid canonical draft").rules[0] + .constraints + .as_ref() + .expect("constraints") + .allowed_install_location_patterns + .len(), + 2 + ); + } + + #[test] + fn large_boolean_arrays_are_rejected_quickly_and_deterministically() { + let oversized = serde_json::Value::Array(vec![json!(true); 125_000]); + let mut match_value = serde_json::Map::new(); + for field in BOOLEAN_MATCH_FIELDS { + match_value.insert((*field).to_owned(), oversized.clone()); + } + let mut raw = draft(); + raw["Rules"] = json!([rule("booleans", match_value.into())]); + let started = std::time::Instant::now(); + let first = validate_draft(&raw); + let second = validate_draft(&raw); + assert!(started.elapsed() < std::time::Duration::from_secs(5)); + assert!(!first.is_valid && first.canonical_draft.is_none() && first.validation_receipt.is_none()); + assert_eq!(first.findings.len(), BOOLEAN_MATCH_FIELDS.len()); + assert_eq!( + serde_json::to_value(first.findings).expect("serialize findings"), + serde_json::to_value(second.findings).expect("serialize findings") + ); + } + + #[test] + fn ineffective_boolean_matches_and_unsupported_criteria_are_rejected() { + let mut raw = draft(); + raw["Rules"] = json!([rule("r1", json!({ "Interactive": [false, true] }))]); + assert!(has_code( + &validate_draft(&raw), + PolicyFindingCode::IneffectiveBooleanMatch + )); + raw["Rules"] = json!([rule("r1", json!({ "PackageNames": ["Display Name"] }))]); + assert!(has_code(&validate_draft(&raw), PolicyFindingCode::InvalidFieldValue)); + } + + #[test] + fn invalid_ranges_validity_and_contradictions_are_rejected() { + let mut raw = draft(); + raw["Rules"] = json!([rule( + "r1", + json!({ "VersionRange": { "MinVersion": "2.0.0", "MaxVersion": "1.0.0" } }) + )]); + assert!(has_code(&validate_draft(&raw), PolicyFindingCode::EmptyVersionRange)); + let mut raw = draft(); + let mut contradictory = rule("r1", json!({ "Interactive": [true] })); + contradictory["Constraints"] = json!({ "AllowInteractive": false }); + raw["Rules"] = json!([contradictory]); + assert!(has_code( + &validate_draft(&raw), + PolicyFindingCode::ContradictoryConstraints + )); + } + + #[test] + fn prerelease_exclusion_rejects_ranges_without_stable_versions() { + for (min, max, include_prerelease, expected_valid) in [ + (Some("1.0.0-alpha"), Some("1.0.0-beta"), false, false), + (Some("1.0.0-alpha"), Some("1.0.0-beta"), true, true), + (Some("1.0.0-alpha"), Some("1.0.0"), false, true), + (None, Some("0.0.0-alpha"), false, false), + (None, Some("0.0.0"), false, true), + (Some("1.0.0"), Some("2.0.0-alpha"), false, true), + ] { + let mut raw = draft(); + let mut range = json!({ "IncludePrerelease": include_prerelease }); + if let Some(min) = min { + range["MinVersion"] = json!(min); + } + if let Some(max) = max { + range["MaxVersion"] = json!(max); + } + raw["Rules"] = json!([rule("range", json!({ "VersionRange": range }))]); + let result = validate_draft(&raw); + assert_eq!( + result.is_valid, expected_valid, + "{min:?}..{max:?}, prerelease={include_prerelease}" + ); + assert_eq!(result.validator_version, VALIDATOR_VERSION); + if expected_valid { + assert!(result.canonical_draft.is_some()); + } else { + assert!(result.canonical_draft.is_none()); + assert!(result.validation_receipt.is_none()); + let finding = result + .findings + .iter() + .find(|finding| finding.code == PolicyFindingCode::EmptyVersionRange) + .expect("empty range finding"); + assert_eq!(finding.path, "/Rules/0/Match/VersionRange"); + } + } + } + + #[test] + fn validity_interval_requires_strictly_increasing_instants() { + for (valid_from, valid_until, expected_valid) in [ + (None, None, true), + (Some("2026-01-01T00:00:00Z"), None, true), + (None, Some("2026-01-01T00:00:00Z"), true), + (Some("2026-01-01T00:00:00Z"), Some("2026-01-01T00:00:01Z"), true), + (Some("2026-01-01T00:00:00Z"), Some("2026-01-01T00:00:00Z"), false), + (Some("2026-01-01T00:00:00Z"), Some("2025-12-31T19:00:00-05:00"), false), + (Some("2026-02-01T00:00:00Z"), Some("2026-01-01T00:00:00Z"), false), + ] { + let mut raw = draft(); + if let Some(valid_from) = valid_from { + raw["Metadata"]["ValidFrom"] = json!(valid_from); + } + if let Some(valid_until) = valid_until { + raw["Metadata"]["ValidUntil"] = json!(valid_until); + } + let result = validate_draft(&raw); + assert_eq!(result.is_valid, expected_valid, "{valid_from:?}..{valid_until:?}"); + assert_eq!(result.validator_version, VALIDATOR_VERSION); + if expected_valid { + assert!(result.canonical_draft.is_some()); + } else { + assert!(result.canonical_draft.is_none()); + assert!(result.validation_receipt.is_none()); + let finding = result + .findings + .iter() + .find(|finding| finding.code == PolicyFindingCode::InvalidValidityInterval) + .expect("invalid interval finding"); + assert_eq!(finding.path, "/Metadata/ValidUntil"); + assert_eq!(finding.message, "ValidUntil must be after ValidFrom"); + } + } + } + + #[test] + fn risky_postures_produce_ordered_warnings() { + let mut raw = draft(); + raw["Enforcement"]["AuditMode"] = json!(true); + raw["Enforcement"]["DefaultDecision"] = json!("Allow"); + let mut allow = rule("allow", json!({ "Managers": ["Winget"] })); + allow["Decision"] = json!("Allow"); + raw["Rules"] = json!([allow]); + let result = validate_draft(&raw); + assert!(result.is_valid); + assert_eq!(result.findings[0].code, PolicyFindingCode::AuditModeEnabled); + assert_eq!(result.findings[1].code, PolicyFindingCode::DefaultAllow); + assert!(has_code(&result, PolicyFindingCode::SensitiveOptionAllowed)); + } + + #[test] + fn sensitive_option_warnings_point_into_the_submitted_draft() { + let options = [ + ("SkipHashCheck", "SkipHashCheck", "AllowSkipHashCheck"), + ("PreRelease", "PreRelease", "AllowPreRelease"), + ( + "AllowCustomInstallLocation", + "HasCustomInstallLocation", + "AllowCustomInstallLocation", + ), + ("AllowPrePostCommands", "HasPrePostCommands", "AllowPrePostCommands"), + ( + "AllowKillBeforeOperation", + "HasKillBeforeOperation", + "AllowKillBeforeOperation", + ), + ( + "AllowUninstallPrevious", + "HasUninstallPrevious", + "AllowUninstallPrevious", + ), + ("AllowCustomParameters", "HasCustomParameters", "AllowCustomParameters"), + ]; + for (option, match_field, constraint_field) in options { + for explicit in ["Match", "Constraints", "Default"] { + let mut raw = draft(); + let mut allow = rule("allow", json!({ "Managers": ["Winget"] })); + allow["Decision"] = json!("Allow"); + match explicit { + "Match" => allow["Match"][match_field] = json!([true]), + "Constraints" => { + allow["Constraints"] = json!({}); + allow["Constraints"][constraint_field] = json!(true); + } + "Default" => {} + _ => unreachable!(), + } + raw["Rules"] = json!([allow]); + let result = validate_draft(&raw); + assert!(result.is_valid, "{option} via {explicit}"); + let finding = result + .findings + .iter() + .find(|finding| finding.arguments.get("option") == Some(&json!(option))) + .unwrap_or_else(|| panic!("missing {option} finding via {explicit}")); + let expected_path = match explicit { + "Match" => format!("/Rules/0/Match/{match_field}"), + "Constraints" => format!("/Rules/0/Constraints/{constraint_field}"), + "Default" => "/Rules/0".to_owned(), + _ => unreachable!(), + }; + assert_eq!(finding.path, expected_path); + assert!(raw.pointer(&finding.path).is_some(), "missing {}", finding.path); + assert!(!finding.arguments.contains_key("Option")); + } + } + } + + #[test] + fn disk_diagnostics_are_sanitized_and_bounded() { + let finding = disk_failure_finding(DiskFailureReason::MalformedContent); + assert_eq!(finding.code, PolicyFindingCode::SchemaViolation); + assert!(!finding.message.contains("secret")); + assert!(finding.message.chars().count() <= MAX_FINDING_MESSAGE_CHARS); + } +} diff --git a/crates/now-package-broker/src/policy_watcher.rs b/crates/now-package-broker/src/policy_watcher.rs index e839698d1..04d83a128 100644 --- a/crates/now-package-broker/src/policy_watcher.rs +++ b/crates/now-package-broker/src/policy_watcher.rs @@ -1,55 +1,60 @@ //! Policy file watcher with live reload. //! //! Watches the policy file for changes and reloads it when modified. -//! If the file becomes unavailable or corrupted, the broker pauses -//! (denies all requests) until a valid policy is available again. -use std::path::{Path, PathBuf}; +use std::path::Path; use std::sync::Arc; use std::time::Duration; use notify::{RecommendedWatcher, RecursiveMode, Watcher}; -use now_policy::PolicyDocument; -use tokio::sync::watch; use tokio_util::sync::CancellationToken; -use tracing::{error, info, warn}; +use tracing::{error, info}; -use crate::policy_loader; +use crate::policy_store::{PolicyStore, ReloadCause}; -/// State of the policy: either loaded and active, or unavailable. -#[derive(Debug, Clone)] -pub enum PolicyState { - /// A valid policy is loaded and active. - Active(Arc), - /// The policy file is missing or corrupted; broker should deny all requests. - Unavailable { reason: String }, +fn affects_policy(event: ¬ify::Event, path: &Path) -> bool { + (event.kind.is_create() || event.kind.is_modify() || event.kind.is_remove()) + && event + .paths + .iter() + .any(|event_path| crate::policy_security::windows_paths_equal(event_path, path)) } -/// Watches a JSON policy file and sends updates via a channel. -/// -/// On startup, attempts to load the policy. If it fails, starts in `Unavailable` state. -/// When the file is modified, reloads it. If reload fails, transitions to `Unavailable`. -/// When a valid file becomes available again, transitions back to `Active`. -pub struct PolicyWatcher { - path: PathBuf, - state_tx: watch::Sender, +async fn debounce_change( + changes: &mut tokio::sync::mpsc::Receiver, + failures: &mut tokio::sync::mpsc::UnboundedReceiver, + shutdown: &CancellationToken, + deadline: tokio::time::Instant, +) -> Result { + loop { + tokio::select! { + biased; + _ = shutdown.cancelled() => return Ok(false), + failure = failures.recv() => return Err(failure.unwrap_or(WatcherFailure::ChannelClosed)), + _ = tokio::time::sleep_until(deadline) => { + while changes.try_recv().is_ok() {} + return Ok(true); + } + Some(_) = changes.recv() => {} + } + } } -impl PolicyWatcher { - /// Create a new watcher for the given policy file path. - /// - /// Returns the watcher and a receiver for policy state changes. - pub fn new(path: PathBuf) -> (Self, watch::Receiver) { - let initial_state = match policy_loader::load_policy(&path) { - Ok(policy) => PolicyState::Active(Arc::new(policy)), - Err(e) => PolicyState::Unavailable { reason: e.to_string() }, - }; - - let (state_tx, state_rx) = watch::channel(initial_state); +#[derive(Clone, Copy, Debug)] +pub(crate) enum WatcherFailure { + Creation, + Registration, + Notification, + ChannelClosed, + TaskTerminated, +} - let watcher = Self { path, state_tx }; +/// Watches a JSON policy file and reloads the shared policy store on change. +pub struct PolicyWatcher(Arc); - (watcher, state_rx) +impl PolicyWatcher { + pub fn new(store: Arc) -> Self { + Self(store) } /// Start watching the policy file for changes. @@ -57,87 +62,138 @@ impl PolicyWatcher { /// This spawns a background task that watches the policy file's parent directory /// and reloads the policy when the file is modified, created, or removed. /// The task runs until the shutdown notify is triggered. - pub async fn watch(self, shutdown: CancellationToken) { - let path = self.path.clone(); - let state_tx = self.state_tx; + pub(crate) async fn watch( + self, + shutdown: CancellationToken, + ready: tokio::sync::oneshot::Sender>, + ) { + let store = self.0; + let path = store.configured_path(); let dir = path.parent().unwrap_or_else(|| Path::new(".")).to_owned(); - let (fs_tx, mut fs_rx) = tokio::sync::mpsc::channel::<()>(16); + let (change_tx, mut changes) = tokio::sync::mpsc::channel(1); + let (failure_tx, mut failures) = tokio::sync::mpsc::unbounded_channel(); let (watcher_stop_tx, watcher_stop_rx) = std::sync::mpsc::channel::<()>(); - // Set up file watcher in a blocking context. - let watch_path = dir.clone(); - let setup_state_tx = state_tx.clone(); let _watcher_handle = tokio::task::spawn_blocking(move || { - let rt_tx = fs_tx; let mut watcher: RecommendedWatcher = - match notify::recommended_watcher(move |res: notify::Result| { - if let Ok(event) = res { - // Only react to modify/create/remove events. - use notify::EventKind; - match event.kind { - EventKind::Modify(_) | EventKind::Create(_) | EventKind::Remove(_) => { - let _ = rt_tx.blocking_send(()); - } - _ => {} - } - } + match notify::recommended_watcher(move |result: notify::Result| match result { + Ok(event) if affects_policy(&event, &path) => _ = change_tx.try_send(tokio::time::Instant::now()), + Ok(_) => {} + Err(_) => _ = failure_tx.send(WatcherFailure::Notification), }) { Ok(watcher) => watcher, Err(error) => { error!(%error, "Failed to create policy file watcher"); - let _ = setup_state_tx.send(PolicyState::Unavailable { - reason: format!("failed to create policy file watcher: {error}"), - }); + let _ = ready.send(Err(WatcherFailure::Creation)); return; } }; - if let Err(error) = watcher.watch(&watch_path, RecursiveMode::NonRecursive) { - error!(%error, path = %watch_path.display(), "Failed to watch policy directory"); - let _ = setup_state_tx.send(PolicyState::Unavailable { - reason: format!("failed to watch policy directory {}: {error}", watch_path.display()), - }); + if let Err(error) = watcher.watch(&dir, RecursiveMode::NonRecursive) { + error!(%error, path = %dir.display(), "Failed to watch policy directory"); + let _ = ready.send(Err(WatcherFailure::Registration)); return; } + let _ = ready.send(Ok(())); let _ = watcher_stop_rx.recv(); }); - // Debounce interval to avoid rapid reloads. let debounce = Duration::from_millis(500); - loop { + let failure = loop { tokio::select! { - _ = shutdown.cancelled() => { - info!("Policy watcher shutting down"); - let _ = watcher_stop_tx.send(()); - break; - } - Some(()) = fs_rx.recv() => { - // Debounce: drain any additional events that arrived. - tokio::time::sleep(debounce).await; - while fs_rx.try_recv().is_ok() {} - - // Attempt reload. - match policy_loader::load_policy(&path) { - Ok(policy) => { - info!( - policy_id = %policy.metadata.id, - revision = policy.metadata.revision, - "Policy reloaded successfully" - ); - let _ = state_tx.send(PolicyState::Active(Arc::new(policy))); - } - Err(e) => { - warn!(error = %e, "Policy reload failed; broker paused"); - let _ = state_tx.send(PolicyState::Unavailable { - reason: e.to_string(), - }); - } + biased; + _ = shutdown.cancelled() => break None, + failure = failures.recv() => break Some(failure.unwrap_or(WatcherFailure::ChannelClosed)), + Some(changed_at) = changes.recv() => { + match debounce_change(&mut changes, &mut failures, &shutdown, changed_at + debounce).await { + Ok(true) => _ = store.reload_from_disk(ReloadCause::ExternalChange).await, + Ok(false) => break None, + Err(failure) => break Some(failure), } } } + }; + match failure { + Some(failure) => fail_closed(&store, failure).await, + None => info!("Policy watcher shutting down"), } + let _ = watcher_stop_tx.send(()); + } +} + +pub(crate) async fn fail_closed(store: &PolicyStore, failure: WatcherFailure) { + error!(?failure, "Policy watcher failed; broker paused"); + store.mark_watcher_unavailable().await; +} + +pub(crate) async fn monitor_watcher_task( + store: Arc, + shutdown: CancellationToken, + handle: tokio::task::JoinHandle<()>, +) { + let _ = handle.await; + if !shutdown.is_cancelled() { + fail_closed(&store, WatcherFailure::TaskTerminated).await; + } +} + +#[cfg(test)] +mod tests { + use notify::EventKind; + use notify::event::{CreateKind, ModifyKind, RemoveKind, RenameMode}; + + use super::*; + + #[tokio::test] + async fn fatal_error_preempts_expired_deadline_which_preempts_queued_change() { + let (tx, mut changes) = tokio::sync::mpsc::channel(2); + let (failure_tx, mut failures) = tokio::sync::mpsc::unbounded_channel(); + tx.send(tokio::time::Instant::now()).await.expect("send change"); + let stop = CancellationToken::new(); + let result = debounce_change(&mut changes, &mut failures, &stop, tokio::time::Instant::now()).await; + assert!(matches!(result, Ok(true))); + failure_tx.send(WatcherFailure::Notification).expect("send failure"); + let result = debounce_change(&mut changes, &mut failures, &stop, tokio::time::Instant::now()).await; + assert!(matches!(result, Err(WatcherFailure::Notification))); + } + + #[test] + fn event_filter_ignores_siblings_and_accepts_relevant_kinds() { + let policy = Path::new(r"C:\POLICY.json"); + let event = |name| notify::Event::new(EventKind::Modify(ModifyKind::Any)).add_path(policy.with_file_name(name)); + assert!(!affects_policy(&event("sibling.json"), policy)); + let relevant = |kind| { + affects_policy( + ¬ify::Event::new(kind).add_path(policy.with_file_name("policy.json")), + policy, + ) + }; + assert!(relevant(EventKind::Create(CreateKind::Any))); + assert!(relevant(EventKind::Modify(ModifyKind::Any))); + assert!(relevant(EventKind::Remove(RemoveKind::Any))); + assert!(relevant(EventKind::Modify(ModifyKind::Name(RenameMode::From)))); + assert!(relevant(EventKind::Modify(ModifyKind::Name(RenameMode::To)))); + assert!(relevant(EventKind::Modify(ModifyKind::Name(RenameMode::Both)))); + assert!(affects_policy( + ¬ify::Event::new(EventKind::Modify(ModifyKind::Any)).add_path(Path::new(r"C:\pölicy.json").to_owned()), + Path::new(r"C:\PÖLICY.json"), + )); + } + + #[tokio::test] + async fn watcher_task_exit_fails_closed_but_shutdown_does_not() { + let store = PolicyStore::for_tests(None); + let initial = store.management_snapshot().state; + monitor_watcher_task(Arc::clone(&store), CancellationToken::new(), tokio::spawn(async {})).await; + assert_ne!(store.management_snapshot().state, initial); + let store = PolicyStore::for_tests(None); + let initial = store.management_snapshot().state; + let shutdown = CancellationToken::new(); + shutdown.cancel(); + monitor_watcher_task(Arc::clone(&store), shutdown, tokio::spawn(async {})).await; + assert_eq!(store.management_snapshot().state, initial); } } diff --git a/crates/now-package-broker/src/server/mod.rs b/crates/now-package-broker/src/server/mod.rs index be684b76b..e7724f06f 100644 --- a/crates/now-package-broker/src/server/mod.rs +++ b/crates/now-package-broker/src/server/mod.rs @@ -1,12 +1,13 @@ //! Runtime implementation of the shared NOW package broker server facade. use std::collections::HashMap; -use std::sync::{Arc, RwLock}; +use std::sync::Arc; use std::time::{Duration, Instant}; use async_trait::async_trait; -use axum::extract::Request; -use axum::http::{Method, StatusCode, header}; +use axum::Json; +use axum::extract::{Extension, Request, State}; +use axum::http::{Method, StatusCode}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; use chrono::{DateTime, Utc}; @@ -28,6 +29,7 @@ use crate::command_builder::build_command; use crate::evaluator; use crate::executor::{CommandExecutor, ExecutionContext}; use crate::operation_tracker::OperationTracker; +use crate::policy_store::PolicyStore; mod connection; mod execution; @@ -39,6 +41,10 @@ use responses::{ policy_info, policy_validity_failure, request_summary, server_context, supported_manager_capabilities, }; +tokio::task_local! { + static POLICY_MANAGEMENT_AUTHENTICATED: (); +} + /// How long a per-user manager availability probe stays fresh before it is re-run. const MANAGER_PROBE_TTL: Duration = Duration::from_secs(60); @@ -81,8 +87,7 @@ impl ManagerProbeCache { /// Shared server state. pub struct BrokerState { - /// Current policy. `None` means the broker is paused (policy file missing or corrupted). - pub policy: RwLock>>, + pub policy_store: Arc, pub executor: Arc, pub pipe_name: String, pub tracker: OperationTracker, @@ -101,19 +106,53 @@ struct EvaluatedRequest { /// Build the axum router for a single authenticated pipe client. pub(crate) fn build_router_for_client(state: Arc, client: PipeClient) -> axum::Router { - let server: SharedPackageBrokerServer = Arc::new(BrokerConnection { state, client }); + let server: SharedPackageBrokerServer = Arc::new(BrokerConnection { + state: Arc::clone(&state), + client: client.clone(), + }); axum::Router::from(now_policy_server_template::api_router_from_shared(server)) - .layer(middleware::from_fn(restrict_phase_one_policy_routes)) + .layer(middleware::from_fn_with_state(state, authenticate_policy_management)) + .layer(Extension(client)) } -async fn restrict_phase_one_policy_routes(request: Request, next: Next) -> Response { - match (request.method(), request.uri().path()) { - (_, "/v1/policy/management" | "/v1/policy/validate") => StatusCode::NOT_FOUND.into_response(), - (method, "/v1/policy") if method != Method::GET && method != Method::HEAD => { - (StatusCode::METHOD_NOT_ALLOWED, [(header::ALLOW, "GET, HEAD")]).into_response() +async fn authenticate_policy_management( + State(state): State>, + Extension(client): Extension, + request: Request, + next: Next, +) -> Response { + let protected = matches!( + (request.method(), request.uri().path()), + (&Method::GET, "/v1/policy/management") + | (&Method::HEAD, "/v1/policy/management") + | (&Method::POST, "/v1/policy/validate") + | (&Method::PUT, "/v1/policy") + ); + if protected { + if let Err(error) = client.validate_connection(state.skip_signature_validation) { + warn!(error = format!("{error:#}"), "Rejected policy management request"); + return ( + StatusCode::UNAUTHORIZED, + Json(error_response( + ErrorCode::Unauthorized, + "pipe client authentication failed", + )), + ) + .into_response(); } - _ => next.run(request).await, + return POLICY_MANAGEMENT_AUTHENTICATED.scope((), next.run(request)).await; } + next.run(request).await +} + +#[expect( + clippy::result_large_err, + reason = "the shared API contract requires ErrorResponse values" +)] +fn require_policy_management_authentication() -> Result<(), ErrorResponse> { + POLICY_MANAGEMENT_AUTHENTICATED + .try_with(|()| ()) + .map_err(|_| error_response(ErrorCode::Unauthorized, "pipe client authentication failed")) } struct BrokerConnection { @@ -143,30 +182,51 @@ impl PackageBrokerServer for BrokerConnection { } async fn policy_management(&self) -> Result { - Err(error_response( - ErrorCode::UnsupportedEndpoint, - "policy management is unavailable", - )) + require_policy_management_authentication()?; + Ok(PolicyManagementResponse { + response_kind: now_policy_api::PolicyManagementResponseKind, + response_version: api_version(), + server: server_context(), + management: self.state.policy_store.management_snapshot(), + }) } async fn validate_policy( &self, - _request: PolicyValidationRequest, + request: PolicyValidationRequest, ) -> Result { - Err(error_response( - ErrorCode::UnsupportedEndpoint, - "policy validation is unavailable", - )) + require_policy_management_authentication()?; + Ok(PolicyValidationResponse { + response_kind: now_policy_api::PolicyValidationResponseKind, + response_version: api_version(), + server: server_context(), + validation: self.state.policy_store.validate_draft(&request.draft), + }) } async fn replace_policy( &self, - _request: PolicyReplacementRequest, + request: PolicyReplacementRequest, ) -> Result { - Err(error_response( - ErrorCode::UnsupportedEndpoint, - "policy replacement is unavailable", - )) + require_policy_management_authentication()?; + if !self.client.is_elevated_administrator() { + return Err(error_response( + ErrorCode::AdministratorRequired, + "policy replacement requires an elevated Administrator", + )); + } + self.state + .policy_store + .replace(request) + .await + .map(|success| PolicyReplacementResponse { + response_kind: now_policy_api::PolicyReplacementResponseKind, + response_version: api_version(), + server: server_context(), + policy: success.policy, + validation: success.validation, + management: success.management, + }) } async fn evaluate(&self, request: PackageRequest) -> Result { @@ -218,8 +278,7 @@ impl PackageBrokerServer for BrokerConnection { impl BrokerState { fn active_policy_snapshot(&self) -> Option> { - let guard = self.policy.read().expect("policy lock poisoned"); - guard.as_ref().map(Arc::clone) + self.policy_store.active_policy() } #[expect( @@ -240,8 +299,8 @@ impl BrokerState { } async fn health(&self) -> HealthResponse { - let policy_guard = self.policy.read().expect("policy lock poisoned"); - let (status, policy_id) = match policy_guard.as_ref() { + let policy = self.active_policy_snapshot(); + let (status, policy_id) = match policy.as_ref() { Some(policy) => (HealthStatus::Ready, policy.metadata.id.to_string()), None => (HealthStatus::Paused, String::new()), }; @@ -597,6 +656,7 @@ mod tests { use axum::body::{Body, to_bytes}; use axum::http::{Method, Request, StatusCode}; + use axum::response::Response; use chrono::Utc; use now_policy::{ PackageBrokerPolicy, PolicyEnforcement, PolicyMetadata, PolicySchemaUri, ResourceId, RulePrecedence, @@ -669,7 +729,7 @@ mod tests { fn state() -> BrokerState { BrokerState { - policy: RwLock::new(Some(Arc::new(permissive_policy()))), + policy_store: PolicyStore::for_tests(Some(permissive_policy())), executor: Arc::new(NoopExecutor), pipe_name: "test-pipe".to_owned(), tracker: OperationTracker::new(), @@ -680,21 +740,30 @@ mod tests { fn shared_state(policy: Option) -> Arc { let mut state = state(); - state.policy = RwLock::new(policy.map(Arc::new)); + state.policy_store = PolicyStore::for_tests(policy); Arc::new(state) } async fn route_request(state: Arc, method: Method, uri: &str) -> Response { let client = PipeClient::from_current_process().expect("capture current test process"); + route_raw(state, client, method, uri, None, Body::empty()).await + } + + async fn route_raw( + state: Arc, + client: PipeClient, + method: Method, + uri: &str, + content_type: Option<&str>, + body: Body, + ) -> Response { + let mut builder = Request::builder().method(method).uri(uri); + if let Some(content_type) = content_type { + builder = builder.header("content-type", content_type); + } let mut router = build_router_for_client(state, client); router - .call( - Request::builder() - .method(method) - .uri(uri) - .body(Body::empty()) - .expect("valid test request"), - ) + .call(builder.body(body).expect("valid test request")) .await .expect("router is infallible") } @@ -706,6 +775,28 @@ mod tests { serde_json::from_slice(&body).expect("response is valid JSON") } + #[cfg(feature = "dev-skip-broker-signature")] + async fn route_json( + state: Arc, + client: PipeClient, + method: Method, + uri: &str, + body: serde_json::Value, + ) -> Response { + let mut router = build_router_for_client(state, client); + router + .call( + Request::builder() + .method(method) + .uri(uri) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(&body).expect("serialize request"))) + .expect("valid test request"), + ) + .await + .expect("router is infallible") + } + #[tokio::test] async fn policy_route_rejects_unsigned_client() { let mut state = state(); @@ -721,6 +812,61 @@ mod tests { assert!(body.get("Policy").is_none()); } + #[tokio::test] + async fn policy_management_authentication_precedes_body_extraction() { + let mut state = state(); + state.skip_signature_validation = false; + let state = Arc::new(state); + let client = PipeClient::from_current_process().expect("capture current test process"); + let response = route_raw( + Arc::clone(&state), + client.clone(), + Method::HEAD, + "/v1/policy/management", + None, + Body::empty(), + ) + .await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + for (method, path, body) in [ + (Method::POST, "/v1/policy/validate", Body::from("{")), + ( + Method::POST, + "/v1/policy/validate", + Body::from(vec![ + b'x'; + now_policy_server_template::MAX_POLICY_MANAGEMENT_BODY_BYTES + 1 + ]), + ), + (Method::PUT, "/v1/policy", Body::from("{")), + ] { + let response = route_raw( + Arc::clone(&state), + client.clone(), + method, + path, + Some("application/json"), + body, + ) + .await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!(response_json(response).await["Code"], "Unauthorized"); + } + } + + #[tokio::test] + async fn policy_management_handlers_require_the_middleware_marker() { + let server = BrokerConnection { + state: shared_state(None), + client: PipeClient::from_current_process().expect("capture current test process"), + }; + let error = server + .policy_management() + .await + .expect_err("direct handler invocation must be rejected"); + assert_eq!(error.code, ErrorCode::Unauthorized); + } + #[test] fn policy_response_returns_not_found_when_unavailable() { let Err(error) = shared_state(None).policy_response() else { @@ -731,21 +877,90 @@ mod tests { } #[tokio::test] - async fn phase_one_router_does_not_expose_policy_management_routes() { + async fn shared_router_exposes_policy_management_routes() { + let (management_status, body_status) = if cfg!(feature = "dev-skip-broker-signature") { + (StatusCode::OK, StatusCode::UNSUPPORTED_MEDIA_TYPE) + } else { + (StatusCode::UNAUTHORIZED, StatusCode::UNAUTHORIZED) + }; for (method, uri, expected_status) in [ - (Method::GET, "/v1/policy/management", StatusCode::NOT_FOUND), - (Method::POST, "/v1/policy/validate", StatusCode::NOT_FOUND), - (Method::PUT, "/v1/policy", StatusCode::METHOD_NOT_ALLOWED), + (Method::GET, "/v1/policy/management", management_status), + (Method::POST, "/v1/policy/validate", body_status), + (Method::PUT, "/v1/policy", body_status), (Method::DELETE, "/v1/policy", StatusCode::METHOD_NOT_ALLOWED), ] { let response = route_request(shared_state(Some(permissive_policy())), method, uri).await; assert_eq!(response.status(), expected_status, "{uri}"); - if expected_status == StatusCode::METHOD_NOT_ALLOWED { - assert_eq!(response.headers().get(header::ALLOW).unwrap(), "GET, HEAD"); - } } } + #[cfg(feature = "dev-skip-broker-signature")] + #[tokio::test] + async fn management_is_authenticated_but_only_elevated_administrators_can_write() { + let unelevated = PipeClient::test_with_authority(false, false).expect("test client"); + let management = route_json( + shared_state(None), + unelevated.clone(), + Method::GET, + "/v1/policy/management", + serde_json::Value::Null, + ) + .await; + assert_eq!(management.status(), StatusCode::OK); + + let draft = serde_json::json!({ + "$schema": now_policy::POLICY_DRAFT_SCHEMA_URI, + "PolicyVersion": "1.0.0", + "PolicyType": "PackageBrokerPolicy", + "Metadata": { "Id": "created", "Publisher": "Test" }, + "Enforcement": { "DefaultDecision": "Deny", "RulePrecedence": "PriorityThenDeny" }, + "Rules": [] + }); + let validated = route_json( + shared_state(None), + unelevated.clone(), + Method::POST, + "/v1/policy/validate", + serde_json::json!({ + "RequestKind": "PolicyValidationRequest", + "RequestVersion": "1.0", + "Draft": draft.clone() + }), + ) + .await; + assert_eq!(validated.status(), StatusCode::OK); + + let state = shared_state(None); + let validation = state.policy_store.validate_draft(&draft); + let replacement = serde_json::json!({ + "RequestKind": "PolicyReplacementRequest", + "RequestVersion": "1.0", + "ExpectedStoreToken": state.policy_store.management_snapshot().store_token, + "Operation": "Create", + "ConflictHandling": "Reject", + "WarningsAcknowledged": false, + "Draft": draft, + "ValidationReceipt": validation.validation_receipt.expect("valid receipt") + }); + let denied = route_json( + Arc::clone(&state), + unelevated, + Method::PUT, + "/v1/policy", + replacement.clone(), + ) + .await; + assert_eq!(denied.status(), StatusCode::FORBIDDEN); + assert_eq!( + response_json(denied).await["Code"], + serde_json::Value::String("AdministratorRequired".to_owned()) + ); + + let elevated = PipeClient::test_with_authority(true, true).expect("test client"); + let accepted = route_json(state, elevated, Method::PUT, "/v1/policy", replacement).await; + assert_eq!(accepted.status(), StatusCode::OK); + } + #[test] fn concurrent_policy_replacement_returns_only_complete_snapshots() { let policy_a = permissive_policy(); @@ -759,8 +974,7 @@ mod tests { 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)); + let state = shared_state(Some((*policy_a).clone())); const READER_COUNT: usize = 4; const ITERATIONS: usize = 1_000; @@ -793,7 +1007,7 @@ mod tests { } else { Arc::clone(&policy_a) }; - *state.policy.write().expect("policy lock") = Some(replacement); + state.policy_store.test_set_active(replacement); std::thread::yield_now(); } }); @@ -924,7 +1138,7 @@ mod tests { probe_count: AtomicUsize::new(0), }); let state = Arc::new(BrokerState { - policy: RwLock::new(None), + policy_store: PolicyStore::for_tests(None), executor: Arc::clone(&executor) as Arc, pipe_name: "test-pipe".to_owned(), tracker: OperationTracker::new(), @@ -1031,7 +1245,7 @@ mod tests { fn state_with_executor(executor: Arc) -> BrokerState { BrokerState { - policy: RwLock::new(Some(Arc::new(permissive_policy()))), + policy_store: PolicyStore::for_tests(Some(permissive_policy())), executor, pipe_name: "test-pipe".to_owned(), tracker: OperationTracker::new(), diff --git a/crates/now-package-broker/src/task.rs b/crates/now-package-broker/src/task.rs index 3e64a4012..fdeb01fc7 100644 --- a/crates/now-package-broker/src/task.rs +++ b/crates/now-package-broker/src/task.rs @@ -1,17 +1,17 @@ //! Package broker entry point -use std::sync::{Arc, RwLock}; +use std::sync::Arc; use anyhow::Context as _; use async_trait::async_trait; use devolutions_gateway_task::{ShutdownSignal, Task}; use tokio_util::sync::CancellationToken; -use tracing::{info, warn}; +use tracing::info; use crate::executor::{self, CommandExecutor}; use crate::pipe::DEFAULT_PIPE_NAME; -use crate::policy_loader; -use crate::policy_watcher::{PolicyState, PolicyWatcher}; +use crate::policy_store::PolicyStore; +use crate::policy_watcher::{PolicyWatcher, WatcherFailure, fail_closed, monitor_watcher_task}; use crate::server::BrokerState; /// Configuration for the broker task. @@ -53,53 +53,13 @@ impl Task for BrokerTask { const NAME: &'static str = "package-broker"; async fn run(self, mut shutdown_signal: ShutdownSignal) -> Self::Output { - // Resolve policy file path. - - let policy_path = match &self.config.policy_path { - Some(path) => std::path::PathBuf::from(path), - None => policy_loader::find_default_policy().unwrap_or_else(|error| { - let candidate = policy_loader::default_policy_candidate(); - warn!( - %error, - path = %candidate.display(), - "Default broker policy is unavailable; broker will pause until this file is provided" - ); - candidate - }), - }; - - // Create policy watcher with initial load attempt. - let (watcher, mut state_rx) = PolicyWatcher::new(policy_path.clone()); - - // Log initial state. - match &*state_rx.borrow() { - PolicyState::Active(policy) => { - info!( - policy_id = %policy.metadata.id, - policy_revision = %policy.metadata.revision, - path = %policy_path.display(), - "Loaded package broker policy" - ); - } - PolicyState::Unavailable { reason } => { - warn!( - %reason, - path = %policy_path.display(), - "Policy unavailable at startup; broker will pause until a valid policy is provided" - ); - } - } + let policy_store = PolicyStore::load(self.config.policy_path.clone().map(std::path::PathBuf::from)); + let watcher = PolicyWatcher::new(Arc::clone(&policy_store)); let executor: Arc = executor::create_platform_executor().into(); - // Initialize BrokerState with current policy (or None if unavailable). - let initial_policy = match &*state_rx.borrow() { - PolicyState::Active(policy) => Some(Arc::clone(policy)), - PolicyState::Unavailable { .. } => None, - }; - let state = Arc::new(BrokerState { - policy: RwLock::new(initial_policy), + policy_store, executor, pipe_name: self.config.pipe_name.clone(), tracker: crate::operation_tracker::OperationTracker::new(), @@ -113,41 +73,22 @@ impl Task for BrokerTask { // Spawn policy watcher task. let watcher_shutdown = shutdown.clone(); - tokio::spawn(async move { - watcher.watch(watcher_shutdown).await; + let (watcher_ready_tx, watcher_ready_rx) = tokio::sync::oneshot::channel(); + let watcher_handle = tokio::spawn(async move { + watcher.watch(watcher_shutdown, watcher_ready_tx).await; }); - - // Spawn policy state relay: updates BrokerState when policy watcher reports changes. - let relay_state = Arc::clone(&state); - let relay_shutdown = shutdown.clone(); - tokio::spawn(async move { - loop { - tokio::select! { - _ = relay_shutdown.cancelled() => break, - result = state_rx.changed() => { - if result.is_err() { - // Sender dropped (watcher exited). - break; - } - let new_policy = match &*state_rx.borrow_and_update() { - PolicyState::Active(policy) => { - info!( - policy_id = %policy.metadata.id, - revision = policy.metadata.revision, - "Policy hot-reloaded; broker resumed" - ); - Some(Arc::clone(policy)) - } - PolicyState::Unavailable { reason } => { - warn!(%reason, "Policy became unavailable; broker paused"); - None - } - }; - *relay_state.policy.write().expect("policy lock poisoned") = new_policy; - } - } + match watcher_ready_rx.await { + Ok(Ok(())) => { + state.policy_store.mark_monitoring_ready().await; } - }); + Ok(Err(failure)) => fail_closed(&state.policy_store, failure).await, + Err(_) => fail_closed(&state.policy_store, WatcherFailure::TaskTerminated).await, + } + tokio::spawn(monitor_watcher_task( + Arc::clone(&state.policy_store), + shutdown.clone(), + watcher_handle, + )); // Spawn pipe server. let server_shutdown = shutdown.clone(); diff --git a/crates/win-api-wrappers/src/token.rs b/crates/win-api-wrappers/src/token.rs index deefeddba..eb12db39a 100644 --- a/crates/win-api-wrappers/src/token.rs +++ b/crates/win-api-wrappers/src/token.rs @@ -397,6 +397,22 @@ impl Token { Ok(is_elevated) } + /// Determines whether `sid` is an enabled group in this token. + pub fn is_member(&self, sid: &Sid) -> anyhow::Result { + use windows::Win32::Security::CheckTokenMembership; + + let token = self + .duplicate(TOKEN_QUERY, None, SecurityIdentification, Security::TokenImpersonation) + .context("duplicate token for membership check")?; + let mut is_member = windows::core::BOOL(0); + + // SAFETY: The duplicated token, SID, and output pointer remain valid for the call. + unsafe { CheckTokenMembership(Some(token.handle.raw()), sid.as_psid_const(), &mut is_member) } + .context("CheckTokenMembership failed")?; + + Ok(is_member.as_bool()) + } + pub fn linked_token(&self) -> anyhow::Result { // SAFETY: The TokenLinkedToken info class is associated to a HANDLE. let handle = unsafe {