From 51fe982ccf3edf80e306b044d804ed7b5aac4f33 Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Mon, 3 Aug 2026 23:35:37 -0400 Subject: [PATCH 1/4] feat(platform): inspect PF redirects without privilege --- crates/platform/src/lib.rs | 5 +- crates/platform/src/pf.rs | 119 ++++++++++++++++++++++++++++++++++++- 2 files changed, 119 insertions(+), 5 deletions(-) diff --git a/crates/platform/src/lib.rs b/crates/platform/src/lib.rs index c4ae3cfa..0f810381 100644 --- a/crates/platform/src/lib.rs +++ b/crates/platform/src/lib.rs @@ -26,8 +26,9 @@ pub use launch_agent::{ }; pub use listener::{loopback_tcp_listener_ports, loopback_tcp_port_has_listener}; pub use pf::{ - PfConfReference, PfFileState, PfRedirectConfig, SYSTEM_PF_ANCHOR_PATH, SYSTEM_PF_CONF_PATH, - active_pf_redirect_config, active_pf_redirect_config_with_privilege_mode, + ActivePfRedirectInspection, PfConfReference, PfFileState, PfRedirectConfig, + SYSTEM_PF_ANCHOR_PATH, SYSTEM_PF_CONF_PATH, active_pf_redirect_config, + active_pf_redirect_config_with_privilege_mode, inspect_active_pf_redirects_unprivileged, inspect_pf_anchor_file, inspect_pf_conf_reference, install_pf_redirects, remove_pf_redirects, }; pub use process::{ diff --git a/crates/platform/src/pf.rs b/crates/platform/src/pf.rs index 7b5008c8..ae4c9680 100644 --- a/crates/platform/src/pf.rs +++ b/crates/platform/src/pf.rs @@ -1,3 +1,4 @@ +use std::collections::BTreeSet; use std::io; use camino::{Utf8Path, Utf8PathBuf}; @@ -24,6 +25,12 @@ pub struct PfRedirectConfig { pub https_port: u16, } +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ActivePfRedirectInspection { + pub pv_config: Option, + pub loopback_target_ports: BTreeSet, +} + #[derive(Copy, Clone, Debug, Default, Eq, PartialEq)] pub struct PfConfReference; @@ -226,6 +233,11 @@ pub fn active_pf_redirect_config() -> Result, PlatformE active_pf_redirect_config_with_privilege_mode(PrivilegeMode::NonInteractive) } +pub fn inspect_active_pf_redirects_unprivileged() +-> Result { + inspect_active_pf_redirects_unprivileged_with_runner(&mut run_system_command_output) +} + pub fn active_pf_redirect_config_with_privilege_mode( privilege_mode: PrivilegeMode, ) -> Result, PlatformError> { @@ -252,6 +264,41 @@ fn active_pf_redirect_config_with_runner( Ok(PfRedirectConfig::parse_active_rules(&anchor_nat_rules)) } +fn inspect_active_pf_redirects_unprivileged_with_runner( + run_system_output: &mut impl FnMut(&str, &[&str]) -> Result, +) -> Result { + let main_nat_rules = run_system_output("/sbin/pfctl", &["-s", "nat"])?; + let pv_config = if main_nat_rules_load_pv_rdr_anchor(&main_nat_rules) { + let anchor_nat_rules = + run_system_output("/sbin/pfctl", &["-a", "com.prvious.pv", "-s", "nat"])?; + + PfRedirectConfig::parse_active_rules(&anchor_nat_rules) + } else { + None + }; + let recursive_nat_rules = run_system_output("/sbin/pfctl", &["-a", "*", "-s", "nat"])?; + + Ok(ActivePfRedirectInspection { + pv_config, + loopback_target_ports: loopback_redirect_target_ports(&recursive_nat_rules), + }) +} + +fn loopback_redirect_target_ports(rules: &str) -> BTreeSet { + rules + .lines() + .filter_map(active_pf_line) + .filter_map(|line| line.rsplit_once("-> 127.0.0.1 port ")) + .filter_map(|(_prefix, port)| { + port.strip_prefix("= ") + .unwrap_or(port) + .split_whitespace() + .next() + }) + .filter_map(|port| port.parse::().ok()) + .collect() +} + fn active_pf_rules_with_runner( pfctl_args: &[&'static str], privilege_mode: PrivilegeMode, @@ -810,11 +857,77 @@ mod tests { use camino_tempfile::tempdir; use super::{ - PfConfReference, PfRedirectConfig, active_pf_redirect_config_with_runner, - install_pf_redirects_with_runner, read_platform_file, remove_pf_redirects_with_runner, - temporary_pf_conf_candidate_path, + ActivePfRedirectInspection, PfConfReference, PfRedirectConfig, + active_pf_redirect_config_with_runner, + inspect_active_pf_redirects_unprivileged_with_runner, install_pf_redirects_with_runner, + read_platform_file, remove_pf_redirects_with_runner, temporary_pf_conf_candidate_path, }; + #[test] + fn active_pf_redirect_inspection_never_invokes_sudo() { + let mut commands = Vec::new(); + + let result = inspect_active_pf_redirects_unprivileged_with_runner(&mut |program, args| { + let command = format!("{program} {}", args.join(" ")); + commands.push(command.clone()); + + Err(crate::PlatformError::SystemIntegrationCommandStatus { + command, + status: "exit status: 1".to_string(), + }) + }); + + assert!(matches!( + result, + Err(crate::PlatformError::SystemIntegrationCommandStatus { .. }) + )); + assert_eq!(commands, ["/sbin/pfctl -s nat"]); + } + + #[test] + fn active_pf_redirect_inspection_reports_recursive_loopback_targets() -> anyhow::Result<()> { + let mut commands = Vec::new(); + + let inspection = inspect_active_pf_redirects_unprivileged_with_runner( + &mut |program, args| { + let command = format!("{program} {}", args.join(" ")); + commands.push(command.clone()); + + match command.as_str() { + "/sbin/pfctl -s nat" => Ok( + "rdr-anchor \"com.prvious.pv\" all\nrdr-anchor \"other\" all\n".to_string(), + ), + "/sbin/pfctl -a com.prvious.pv -s nat" => Ok( + "rdr pass on lo0 inet proto tcp from any to 127.0.0.1 port 80 -> 127.0.0.1 port 48080\nrdr pass on lo0 inet proto tcp from any to 127.0.0.1 port 443 -> 127.0.0.1 port 48443\n" + .to_string(), + ), + _ => Ok( + "rdr pass on lo0 inet proto tcp from any to 127.0.0.1 port 80 -> 127.0.0.1 port 48080\nrdr pass on lo0 inet proto tcp from any to 127.0.0.1 port 8080 -> 127.0.0.1 port = 45080 round-robin\n" + .to_string(), + ), + } + }, + )?; + + assert_eq!( + inspection, + ActivePfRedirectInspection { + pv_config: Some(PfRedirectConfig::new(48080, 48443)), + loopback_target_ports: [45080, 48080].into_iter().collect(), + } + ); + assert_eq!( + commands, + [ + "/sbin/pfctl -s nat", + "/sbin/pfctl -a com.prvious.pv -s nat", + "/sbin/pfctl -a * -s nat", + ] + ); + + Ok(()) + } + #[test] fn active_pf_redirect_config_reads_loaded_rdr_anchor_reference() -> anyhow::Result<()> { let mut commands = Vec::new(); From bb7029c7528b0ab5e3746e516a61f761f435796f Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Mon, 3 Aug 2026 23:53:19 -0400 Subject: [PATCH 2/4] fix(daemon): preserve Gateway during PF drift --- crates/daemon/src/gateway.rs | 442 ++++++++++++++---- crates/daemon/src/gateway_config.rs | 9 + crates/daemon/src/supervisor.rs | 82 +++- .../gateway/fake-frankenphp-server.py | 11 + crates/daemon/tests/gateway_reconciliation.rs | 28 +- ...nderers_quote_path_tokens_with_spaces.snap | 15 + ...mports_project_configs_when_requested.snap | 15 + ...nderer_outputs_empty_gateway_listener.snap | 16 +- ...ig_renderer_outputs_gateway_caddyfile.snap | 15 + ..._gateway_and_one_worker_per_php_track.snap | 4 +- 10 files changed, 542 insertions(+), 95 deletions(-) diff --git a/crates/daemon/src/gateway.rs b/crates/daemon/src/gateway.rs index 1bc258f7..5408f0be 100644 --- a/crates/daemon/src/gateway.rs +++ b/crates/daemon/src/gateway.rs @@ -20,8 +20,9 @@ use tokio::io::{AsyncRead, AsyncReadExt}; use tokio::time::timeout; use crate::gateway_config::{ - GatewayConfigInput, GatewayProjectRoute, PhpWorkerConfigInput, PhpWorkerProject, - PromotedConfigDir, PromotedConfigTree, promote_config_dir, promote_validated_config_tree_async, + GATEWAY_HEALTH_HOSTNAME, GATEWAY_HEALTH_PATH, GATEWAY_HEALTH_RESPONSE, GatewayConfigInput, + GatewayProjectRoute, PhpWorkerConfigInput, PhpWorkerProject, PromotedConfigDir, + PromotedConfigTree, promote_config_dir, promote_validated_config_tree_async, render_gateway_config, render_gateway_project_config, render_php_worker_config, render_php_worker_project_config, }; @@ -41,6 +42,7 @@ type FrankenphpProcessCommand = tokio::process::Command; const PHP_INI_ENVIRONMENT_KEYS: [&str; 2] = ["PHPRC", "PHP_INI_SCAN_DIR"]; const CONFIG_VALIDATION_TIMEOUT: Duration = Duration::from_secs(10); const RUNTIME_READINESS_TIMEOUT: Duration = Duration::from_secs(60); +const PF_PUBLIC_READINESS_TIMEOUT: Duration = Duration::from_secs(2); const FOREIGN_LISTENER_PROBE_TIMEOUT: Duration = Duration::from_millis(100); const PUBLIC_HTTP_PORT: u16 = 80; const PUBLIC_HTTPS_PORT: u16 = 443; @@ -89,6 +91,15 @@ pub struct GatewayRuntimePlan { pub storage_path: Utf8PathBuf, } +#[doc(hidden)] +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +pub enum GatewayPfRoutingState { + Active, + Inactive, + Drifted, + Unknown, +} + #[derive(Clone, Debug, Eq, PartialEq)] pub struct PhpWorkerRuntimePlan { pub php_track: String, @@ -130,6 +141,23 @@ pub async fn reconcile_gateway_runtimes(paths: &PvPaths) -> Result Result { + reconcile_gateway_runtimes_with_pf_state(paths, readiness_timeout, None).await +} + +#[doc(hidden)] +pub async fn reconcile_gateway_runtimes_with_pf_state_for_test( + paths: &PvPaths, + readiness_timeout: Duration, + pf_routing_state: GatewayPfRoutingState, +) -> Result { + reconcile_gateway_runtimes_with_pf_state(paths, readiness_timeout, Some(pf_routing_state)).await +} + +async fn reconcile_gateway_runtimes_with_pf_state( + paths: &PvPaths, + readiness_timeout: Duration, + pf_routing_state: Option, ) -> Result { let Some(gateway_command) = first_installed_frankenphp_command(paths)? else { record_runtime_observed( @@ -209,37 +237,84 @@ pub async fn reconcile_gateway_runtimes_with_readiness_timeout( host: "127.0.0.1".to_owned(), port: worker.port, }, - subject, + subject.clone(), readiness_timeout, + ReadinessFailurePolicy::FailRuntime, ) .await?; + record_runtime_observed( + paths, + subject, + RuntimeObservedStatus::Running, + Some(GATEWAY_RUNTIME_RECONCILED), + )?; } let gateway_config = reconcile_gateway_config(paths, &gateway_command, &plan).await?; - let gateway_readiness = - gateway_readiness_check(paths, &plan, gateway_config.readiness_hostname.clone()); - start_or_adopt_promoted_runtime( + let pf_routing_state = + pf_routing_state.unwrap_or_else(|| gateway_pf_routing_state(paths, &plan)); + let gateway_readiness = gateway_readiness_plan( + &plan, + gateway_config.readiness_hostname.clone(), + pf_routing_state, + readiness_timeout, + ); + let readiness_outcome = start_or_adopt_promoted_runtime( paths, &supervisor, gateway_config.promoted_config, gateway_process_spec(paths, &gateway_command), - gateway_readiness, + gateway_readiness.check, RuntimeSubject::Gateway, - readiness_timeout, + gateway_readiness.timeout, + gateway_readiness.failure_policy, ) .await?; + record_gateway_runtime_observed(paths, pf_routing_state, readiness_outcome)?; stop_stale_worker_runtimes(paths, &supervisor, &plan).await?; Ok(GATEWAY_RUNTIME_RECONCILED.to_owned()) } -fn gateway_readiness_check( - paths: &PvPaths, +fn gateway_readiness_plan( plan: &RuntimePlan, readiness_hostname: Option, -) -> ReadinessCheck { - let ports = gateway_readiness_ports(paths, plan); + pf_routing_state: GatewayPfRoutingState, + readiness_timeout: Duration, +) -> GatewayReadinessPlan { + let ports = gateway_readiness_ports(plan, pf_routing_state); + let failure_policy = match pf_routing_state { + GatewayPfRoutingState::Active | GatewayPfRoutingState::Inactive => { + ReadinessFailurePolicy::FailRuntime + } + GatewayPfRoutingState::Drifted | GatewayPfRoutingState::Unknown => { + ReadinessFailurePolicy::PreserveRuntime + } + }; + let timeout = match failure_policy { + ReadinessFailurePolicy::FailRuntime => readiness_timeout, + ReadinessFailurePolicy::PreserveRuntime => { + readiness_timeout.min(PF_PUBLIC_READINESS_TIMEOUT) + } + }; - gateway_readiness_check_for_ports(plan, readiness_hostname, ports) + let check = if pf_routing_state == GatewayPfRoutingState::Inactive { + gateway_readiness_check_for_ports(plan, readiness_hostname, ports) + } else { + gateway_public_readiness_check(plan, ports) + }; + + GatewayReadinessPlan { + check, + failure_policy, + timeout, + } +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct GatewayReadinessPlan { + check: ReadinessCheck, + failure_policy: ReadinessFailurePolicy, + timeout: Duration, } #[derive(Copy, Clone, Debug, Eq, PartialEq)] @@ -248,11 +323,12 @@ struct GatewayReadinessPorts { https: u16, } -fn gateway_readiness_ports(paths: &PvPaths, plan: &RuntimePlan) -> GatewayReadinessPorts { +fn gateway_readiness_ports( + plan: &RuntimePlan, + pf_routing_state: GatewayPfRoutingState, +) -> GatewayReadinessPorts { // macOS pf can make direct connections to an active rdr target port hang. - if active_gateway_redirects_match_plan(plan) - || prepared_gateway_redirects_match_plan(paths, plan) - { + if pf_routing_state != GatewayPfRoutingState::Inactive { return GatewayReadinessPorts { http: PUBLIC_HTTP_PORT, https: PUBLIC_HTTPS_PORT, @@ -265,22 +341,76 @@ fn gateway_readiness_ports(paths: &PvPaths, plan: &RuntimePlan) -> GatewayReadin } } -fn active_gateway_redirects_match_plan(plan: &RuntimePlan) -> bool { - matches!( - platform::active_pf_redirect_config(), - Ok(Some(config)) - if config - == platform::PfRedirectConfig::new(plan.gateway.http_port, plan.gateway.https_port) - ) +fn gateway_pf_routing_state(paths: &PvPaths, plan: &RuntimePlan) -> GatewayPfRoutingState { + let expected = platform::PfRedirectConfig::new(plan.gateway.http_port, plan.gateway.https_port); + let files_current = pf_files_current(paths, &expected); + + match platform::inspect_active_pf_redirects_unprivileged() { + Ok(inspection) => classify_gateway_pf_routing_state( + &expected, + inspection.pv_config.as_ref(), + &inspection.loopback_target_ports, + true, + files_current, + ), + Err(_error) => classify_gateway_pf_routing_state( + &expected, + None, + &BTreeSet::new(), + false, + files_current, + ), + } } -fn prepared_gateway_redirects_match_plan(paths: &PvPaths, plan: &RuntimePlan) -> bool { - let expected = platform::PfRedirectConfig::new(plan.gateway.http_port, plan.gateway.https_port); +fn classify_gateway_pf_routing_state( + expected: &platform::PfRedirectConfig, + active: Option<&platform::PfRedirectConfig>, + loopback_target_ports: &BTreeSet, + inspection_available: bool, + files_current: bool, +) -> GatewayPfRoutingState { + if !inspection_available { + return if files_current { + GatewayPfRoutingState::Unknown + } else { + GatewayPfRoutingState::Drifted + }; + } - matches!( - platform::inspect_pf_anchor_file(&paths.pf_anchor_config(), Some(&expected)), - platform::PfFileState::Current { .. } - ) + match active { + Some(active) if active == expected && files_current => GatewayPfRoutingState::Active, + Some(active) if active == expected => GatewayPfRoutingState::Drifted, + Some(_active) => GatewayPfRoutingState::Drifted, + None if loopback_target_ports.contains(&expected.http_port) + || loopback_target_ports.contains(&expected.https_port) => + { + GatewayPfRoutingState::Drifted + } + None => GatewayPfRoutingState::Inactive, + } +} + +fn pf_files_current(paths: &PvPaths, expected: &platform::PfRedirectConfig) -> bool { + let prepared_anchor = + platform::inspect_pf_anchor_file(&paths.pf_anchor_config(), Some(expected)); + let prepared_reference = platform::inspect_pf_conf_reference( + &paths.pf_conf_reference_config(), + Some(&platform::PfConfReference), + ); + let system_anchor = platform::inspect_pf_anchor_file( + Utf8Path::new(platform::SYSTEM_PF_ANCHOR_PATH), + Some(expected), + ); + let system_reference = platform::inspect_pf_conf_reference( + Utf8Path::new(platform::SYSTEM_PF_CONF_PATH), + Some(&platform::PfConfReference), + ); + + matches!(prepared_anchor, platform::PfFileState::Current { .. }) + && matches!(prepared_reference, platform::PfFileState::Current { .. }) + && matches!(system_anchor, platform::PfFileState::Current { .. }) + && matches!(system_reference, platform::PfFileState::Current { .. }) } fn gateway_readiness_check_for_ports( @@ -304,6 +434,22 @@ fn gateway_readiness_check_for_ports( } } +fn gateway_public_readiness_check( + plan: &RuntimePlan, + ports: GatewayReadinessPorts, +) -> ReadinessCheck { + ReadinessCheck::GatewayIdentity { + http_host: "127.0.0.1".to_owned(), + http_port: ports.http, + https_host: "127.0.0.1".to_owned(), + https_port: ports.https, + server_name: GATEWAY_HEALTH_HOSTNAME.to_owned(), + path: GATEWAY_HEALTH_PATH.to_owned(), + expected_body: GATEWAY_HEALTH_RESPONSE.to_owned(), + ca_certificate_path: plan.gateway.ca_certificate_path.clone(), + } +} + fn gateway_readiness_hostname(fragments: &[ProjectConfigFragment]) -> Option { fragments .first() @@ -968,7 +1114,8 @@ async fn start_or_adopt_promoted_runtime( readiness: ReadinessCheck, subject: RuntimeSubject, readiness_timeout: Duration, -) -> Result<(), DaemonError> { + failure_policy: ReadinessFailurePolicy, +) -> Result { let result = start_or_adopt_runtime( paths, supervisor, @@ -976,18 +1123,19 @@ async fn start_or_adopt_promoted_runtime( readiness, subject.clone(), readiness_timeout, + failure_policy, ) .await; match result { - Ok(()) => { + Ok(outcome) => { if let Err(error) = promoted_config.commit() { record_runtime_error(paths, subject, &error)?; return Err(error); } - Ok(()) + Ok(outcome) } Err(error) => { if let Err(rollback_error) = promoted_config.rollback() { @@ -1009,13 +1157,22 @@ async fn start_or_adopt_runtime( readiness: ReadinessCheck, subject: RuntimeSubject, readiness_timeout: Duration, -) -> Result<(), DaemonError> { + failure_policy: ReadinessFailurePolicy, +) -> Result { let result = async { if supervisor.adopt(&spec)?.is_some() { if supervisor.reload(&spec)? { - wait_for_readiness(readiness, readiness_timeout).await?; + if let Err(error) = wait_for_readiness(readiness, readiness_timeout).await { + if failure_policy == ReadinessFailurePolicy::PreserveRuntime + && supervisor.verify_ownership(&spec)?.is_some() + { + return Ok(RuntimeReadinessOutcome::Unverified); + } + + return Err(error); + } - return Ok(()); + return Ok(RuntimeReadinessOutcome::Verified); } return Err(DaemonError::UnexpectedProtocolResponse { @@ -1040,6 +1197,10 @@ async fn start_or_adopt_runtime( let mut process = supervisor.start(spec.clone()).await?; if let Err(error) = wait_for_readiness(readiness, readiness_timeout).await { record_runtime_readiness_diagnostics(paths, &spec, &mut process, &error); + if failure_policy == ReadinessFailurePolicy::PreserveRuntime && !process.has_exited()? { + return Ok(RuntimeReadinessOutcome::Unverified); + } + process.stop(Duration::from_secs(1)).await?; return Err(error); @@ -1054,17 +1215,12 @@ async fn start_or_adopt_runtime( }); } - Ok(()) + Ok(RuntimeReadinessOutcome::Verified) } .await; match result { - Ok(()) => record_runtime_observed( - paths, - subject, - RuntimeObservedStatus::Running, - Some(GATEWAY_RUNTIME_RECONCILED), - ), + Ok(outcome) => Ok(outcome), Err(error) => { record_runtime_error(paths, subject, &error)?; @@ -1073,6 +1229,18 @@ async fn start_or_adopt_runtime( } } +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +enum ReadinessFailurePolicy { + FailRuntime, + PreserveRuntime, +} + +#[derive(Copy, Clone, Debug, Eq, PartialEq)] +enum RuntimeReadinessOutcome { + Verified, + Unverified, +} + async fn foreign_listener_is_ready(readiness: &ReadinessCheck) -> bool { matches!( timeout( @@ -1482,6 +1650,33 @@ fn record_runtime_error( ) } +fn record_gateway_runtime_observed( + paths: &PvPaths, + pf_routing_state: GatewayPfRoutingState, + readiness_outcome: RuntimeReadinessOutcome, +) -> Result<(), DaemonError> { + let (status, message) = match (pf_routing_state, readiness_outcome) { + (GatewayPfRoutingState::Active, _) + | (GatewayPfRoutingState::Unknown, RuntimeReadinessOutcome::Verified) => { + (RuntimeObservedStatus::Running, GATEWAY_RUNTIME_RECONCILED) + } + (GatewayPfRoutingState::Inactive, _) => ( + RuntimeObservedStatus::Degraded, + "Low-port routing is inactive; run `pv ports:install` to restore ports 80 and 443", + ), + (GatewayPfRoutingState::Drifted, _) => ( + RuntimeObservedStatus::Degraded, + "Low-port routing is drifted; run `pv ports:install` to restore ports 80 and 443", + ), + (GatewayPfRoutingState::Unknown, RuntimeReadinessOutcome::Unverified) => ( + RuntimeObservedStatus::Degraded, + "Low-port routing is unknown; run `pv ports:install` to verify ports 80 and 443", + ), + }; + + record_runtime_observed(paths, RuntimeSubject::Gateway, status, Some(message)) +} + fn record_runtime_observed( paths: &PvPaths, subject: RuntimeSubject, @@ -1566,6 +1761,9 @@ fn worker_config_private_environment( #[cfg(test)] mod tests { + use std::collections::BTreeSet; + use std::time::Duration; + use anyhow::Result; use camino::Utf8PathBuf; use camino_tempfile::tempdir; @@ -1576,8 +1774,10 @@ mod tests { use crate::gateway_config::GatewayProjectRoute; use super::{ - GatewayReadinessPorts, GatewayRuntimePlan, RuntimePlan, gateway_project_config_fragments, - gateway_readiness_check_for_ports, gateway_readiness_hostname, gateway_readiness_ports, + GatewayPfRoutingState, GatewayReadinessPorts, GatewayRuntimePlan, ReadinessFailurePolicy, + RuntimePlan, classify_gateway_pf_routing_state, gateway_project_config_fragments, + gateway_public_readiness_check, gateway_readiness_check_for_ports, + gateway_readiness_hostname, gateway_readiness_plan, gateway_readiness_ports, project_config_file_name, }; @@ -1610,12 +1810,11 @@ mod tests { } #[test] - fn gateway_readiness_uses_public_ports_when_redirects_are_active() -> Result<()> { + fn gateway_readiness_uses_identity_probes_on_public_ports() -> Result<()> { let plan = runtime_plan(); - let readiness = gateway_readiness_check_for_ports( + let readiness = gateway_public_readiness_check( &plan, - Some("project.test".to_string()), GatewayReadinessPorts { http: 80, https: 443, @@ -1624,12 +1823,14 @@ mod tests { assert_eq!( readiness, - ReadinessCheck::GatewayHttps { + ReadinessCheck::GatewayIdentity { http_host: "127.0.0.1".to_string(), http_port: 80, https_host: "127.0.0.1".to_string(), https_port: 443, - server_name: "project.test".to_string(), + server_name: "pv-gateway.localhost".to_string(), + path: "/__pv/health".to_string(), + expected_body: "pv-gateway-health-v1".to_string(), ca_certificate_path: Utf8PathBuf::from("/tmp/pv-missing-ca.pem"), } ); @@ -1638,67 +1839,132 @@ mod tests { } #[test] - fn gateway_readiness_uses_public_http_for_empty_gateway() -> Result<()> { + fn gateway_readiness_uses_public_ports_for_active_drifted_and_unknown_pf() { let plan = runtime_plan(); - let readiness = gateway_readiness_check_for_ports( - &plan, - None, - GatewayReadinessPorts { - http: 80, - https: 443, - }, - ); + for state in [ + GatewayPfRoutingState::Active, + GatewayPfRoutingState::Drifted, + GatewayPfRoutingState::Unknown, + ] { + assert_eq!( + gateway_readiness_ports(&plan, state), + GatewayReadinessPorts { + http: 80, + https: 443, + } + ); + } + } + + #[test] + fn gateway_readiness_uses_backend_ports_only_when_pf_is_inactive() { + let plan = runtime_plan(); assert_eq!( - readiness, - ReadinessCheck::Tcp { - host: "127.0.0.1".to_string(), - port: 80, + gateway_readiness_ports(&plan, GatewayPfRoutingState::Inactive), + GatewayReadinessPorts { + http: plan.gateway.http_port, + https: plan.gateway.https_port, } ); + } - Ok(()) + #[test] + fn current_prepared_files_do_not_override_confirmed_inactive_rules() { + let expected = PfRedirectConfig::new(45080, 45443); + + assert_eq!( + classify_gateway_pf_routing_state(&expected, None, &BTreeSet::new(), true, false), + GatewayPfRoutingState::Inactive + ); } #[test] - fn gateway_readiness_uses_public_ports_when_prepared_redirect_matches_plan() -> Result<()> { - let tempdir = tempdir()?; - let paths = PvPaths::for_home(tempdir.path().join("home")); - let plan = runtime_plan(); - let prepared_redirect = - PfRedirectConfig::new(plan.gateway.http_port, plan.gateway.https_port); - state::fs::write_sensitive_file( - &paths.pf_anchor_config(), - &prepared_redirect.render_anchor(), - )?; + fn matching_loaded_rules_are_active_only_with_current_files() { + let expected = PfRedirectConfig::new(45080, 45443); assert_eq!( - gateway_readiness_ports(&paths, &plan), - GatewayReadinessPorts { - http: 80, - https: 443, - } + classify_gateway_pf_routing_state( + &expected, + Some(&expected), + &BTreeSet::new(), + true, + true, + ), + GatewayPfRoutingState::Active + ); + assert_eq!( + classify_gateway_pf_routing_state( + &expected, + Some(&expected), + &BTreeSet::new(), + true, + false, + ), + GatewayPfRoutingState::Drifted ); + } - Ok(()) + #[test] + fn redirects_targeting_backend_ports_are_drifted_not_inactive() { + let expected = PfRedirectConfig::new(45080, 45443); + + assert_eq!( + classify_gateway_pf_routing_state( + &expected, + None, + &[45080].into_iter().collect(), + true, + true, + ), + GatewayPfRoutingState::Drifted + ); } #[test] - fn gateway_readiness_uses_backend_ports_without_prepared_redirect() -> Result<()> { - let tempdir = tempdir()?; - let paths = PvPaths::for_home(tempdir.path().join("home")); + fn unavailable_rule_inspection_uses_bounded_advisory_public_readiness() { let plan = runtime_plan(); + let readiness = gateway_readiness_plan( + &plan, + Some("project.test".to_string()), + GatewayPfRoutingState::Unknown, + Duration::from_secs(60), + ); + assert_eq!( - gateway_readiness_ports(&paths, &plan), - GatewayReadinessPorts { - http: plan.gateway.http_port, - https: plan.gateway.https_port, + readiness.failure_policy, + ReadinessFailurePolicy::PreserveRuntime + ); + assert_eq!(readiness.timeout, Duration::from_secs(2)); + assert_eq!( + readiness.check, + ReadinessCheck::GatewayIdentity { + http_host: "127.0.0.1".to_string(), + http_port: 80, + https_host: "127.0.0.1".to_string(), + https_port: 443, + server_name: "pv-gateway.localhost".to_string(), + path: "/__pv/health".to_string(), + expected_body: "pv-gateway-health-v1".to_string(), + ca_certificate_path: Utf8PathBuf::from("/tmp/pv-missing-ca.pem"), } ); + } - Ok(()) + #[test] + fn unavailable_rule_inspection_is_unknown_only_when_files_are_current() { + let expected = PfRedirectConfig::new(45080, 45443); + + assert_eq!( + classify_gateway_pf_routing_state(&expected, None, &BTreeSet::new(), false, true,), + GatewayPfRoutingState::Unknown + ); + assert_eq!( + classify_gateway_pf_routing_state(&expected, None, &BTreeSet::new(), false, false,), + GatewayPfRoutingState::Drifted + ); } #[test] diff --git a/crates/daemon/src/gateway_config.rs b/crates/daemon/src/gateway_config.rs index 7ff52421..e3b2e37c 100644 --- a/crates/daemon/src/gateway_config.rs +++ b/crates/daemon/src/gateway_config.rs @@ -7,6 +7,9 @@ use state::fs; use crate::DaemonError; static CANDIDATE_CONFIG_COUNTER: AtomicU64 = AtomicU64::new(0); +pub(crate) const GATEWAY_HEALTH_HOSTNAME: &str = "pv-gateway.localhost"; +pub(crate) const GATEWAY_HEALTH_PATH: &str = "/__pv/health"; +pub(crate) const GATEWAY_HEALTH_RESPONSE: &str = "pv-gateway-health-v1"; #[derive(Clone, Debug, Eq, PartialEq)] pub struct GatewayConfigInput { @@ -73,6 +76,12 @@ pub fn render_gateway_config(input: &GatewayConfigInput) -> Result { + format!( + "gateway-identity:https:{server_name}:{https_host}:{https_port}{path};http:{http_host}:{http_port}{path}" + ) + } Self::RedisPing { host, port } => format!("redis-ping:{host}:{port}"), Self::Http { host, port, path } => format!("http:{host}:{port}{path}"), } @@ -524,6 +547,32 @@ async fn check_once(check: &ReadinessCheck) -> Result<(), DaemonError> { check_tcp_once(http_host, *http_port).await?; check_https_once(https_host, *https_port, server_name, ca_certificate_path).await } + ReadinessCheck::GatewayIdentity { + http_host, + http_port, + https_host, + https_port, + server_name, + path, + expected_body, + ca_certificate_path, + } => { + let http_stream = TcpStream::connect((http_host.as_str(), *http_port)).await?; + check_gateway_identity_response(http_stream, server_name, path, expected_body).await?; + let tcp_stream = TcpStream::connect((https_host.as_str(), *https_port)).await?; + let connector = TlsConnector::from(tls_client_config(ca_certificate_path)?); + let server_name_text = server_name.to_owned(); + let tls_server_name = + ServerName::try_from(server_name_text.clone()).map_err(|error| { + io::Error::new( + io::ErrorKind::InvalidInput, + format!("invalid TLS server name `{server_name_text}`: {error}"), + ) + })?; + let tls_stream = connector.connect(tls_server_name, tcp_stream).await?; + + check_gateway_identity_response(tls_stream, server_name, path, expected_body).await + } ReadinessCheck::RedisPing { host, port } => { let url = format!("redis://{host}:{port}/"); let client = redis::Client::open(url)?; @@ -554,6 +603,37 @@ async fn check_once(check: &ReadinessCheck) -> Result<(), DaemonError> { } } +async fn check_gateway_identity_response( + mut stream: Stream, + server_name: &str, + path: &str, + expected_body: &str, +) -> Result<(), DaemonError> +where + Stream: AsyncRead + AsyncWrite + Unpin, +{ + let request = + format!("GET {path} HTTP/1.1\r\nHost: {server_name}\r\nConnection: close\r\n\r\n"); + stream.write_all(request.as_bytes()).await?; + + let mut response = Vec::new(); + stream.take(4096).read_to_end(&mut response).await?; + if http_response_has_success_body(&response, expected_body.as_bytes()) { + return Ok(()); + } + + Err(io::Error::other("Gateway identity readiness returned an unexpected response").into()) +} + +fn http_response_has_success_body(response: &[u8], expected_body: &[u8]) -> bool { + let Some(headers_end) = response.windows(4).position(|window| window == b"\r\n\r\n") else { + return false; + }; + let body = &response[headers_end + 4..]; + + http_status_is_success(response, response.len()) && body == expected_body +} + async fn check_tcp_once(host: &str, port: u16) -> Result<(), DaemonError> { let _stream = TcpStream::connect((host, port)).await?; diff --git a/crates/daemon/test-fixtures/gateway/fake-frankenphp-server.py b/crates/daemon/test-fixtures/gateway/fake-frankenphp-server.py index c1456279..030fa918 100644 --- a/crates/daemon/test-fixtures/gateway/fake-frankenphp-server.py +++ b/crates/daemon/test-fixtures/gateway/fake-frankenphp-server.py @@ -32,6 +32,17 @@ class Handler(http.server.SimpleHTTPRequestHandler): def log_message(self, format, *args): pass + def do_GET(self): + if self.path == "/__pv/health": + body = b"pv-gateway-health-v1" + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + + super().do_GET() + class Server(http.server.ThreadingHTTPServer): def server_bind(self): diff --git a/crates/daemon/tests/gateway_reconciliation.rs b/crates/daemon/tests/gateway_reconciliation.rs index 86d4d68f..57dedc7a 100644 --- a/crates/daemon/tests/gateway_reconciliation.rs +++ b/crates/daemon/tests/gateway_reconciliation.rs @@ -3,9 +3,9 @@ use camino::{Utf8Path, Utf8PathBuf}; use camino_tempfile::tempdir; use daemon::DaemonError; use daemon::gateway::{ - FrankenphpCommand, build_runtime_plan, gateway_process_spec, promote_validated_config_for_test, - reconcile_gateway_runtimes, reconcile_gateway_runtimes_with_readiness_timeout, validate_config, - worker_process_spec, + FrankenphpCommand, GatewayPfRoutingState, build_runtime_plan, gateway_process_spec, + promote_validated_config_for_test, reconcile_gateway_runtimes_with_pf_state_for_test, + validate_config, worker_process_spec, }; use insta::{Settings, assert_debug_snapshot}; use rcgen::generate_simple_self_signed; @@ -41,6 +41,27 @@ const FAKE_FRANKENPHP_HANGS_ON_PORT_SERVER_SCRIPT: &str = include_str!(concat!( )); const FAKE_FRANKENPHP_BLOCKED_PORT_SENTINEL: &str = "__PV_BLOCKED_PORT__"; +async fn reconcile_gateway_runtimes(paths: &PvPaths) -> Result { + reconcile_gateway_runtimes_with_pf_state_for_test( + paths, + Duration::from_secs(60), + GatewayPfRoutingState::Inactive, + ) + .await +} + +async fn reconcile_gateway_runtimes_with_readiness_timeout( + paths: &PvPaths, + readiness_timeout: Duration, +) -> Result { + reconcile_gateway_runtimes_with_pf_state_for_test( + paths, + readiness_timeout, + GatewayPfRoutingState::Inactive, + ) + .await +} + #[expect( clippy::disallowed_types, reason = "regression tests spawn a nested test process to control inherited env without unsafe mutation" @@ -2090,6 +2111,7 @@ fn seed_gateway_test_tls(paths: &PvPaths) -> Result<()> { "broken.test".to_owned(), "changed.acme.test".to_owned(), "other.test".to_owned(), + "pv-gateway.localhost".to_owned(), ])?; fs::write_sensitive_file(&paths.ca_certificate(), &certified_key.cert.pem())?; fs::write_sensitive_file( diff --git a/crates/daemon/tests/snapshots/gateway_config__config_renderers_quote_path_tokens_with_spaces.snap b/crates/daemon/tests/snapshots/gateway_config__config_renderers_quote_path_tokens_with_spaces.snap index 901d990e..9de85269 100644 --- a/crates/daemon/tests/snapshots/gateway_config__config_renderers_quote_path_tokens_with_spaces.snap +++ b/crates/daemon/tests/snapshots/gateway_config__config_renderers_quote_path_tokens_with_spaces.snap @@ -22,6 +22,21 @@ Gateway: } } +http://pv-gateway.localhost { + bind 127.0.0.1 ::1 + respond /__pv/health "pv-gateway-health-v1" 200 +} + +https://pv-gateway.localhost { + bind 127.0.0.1 ::1 + tls { + issuer internal { + ca local + } + } + respond /__pv/health "pv-gateway-health-v1" 200 +} + import "/Users/Alice Smith/.pv/config/gateway/projects/*.Caddyfile" Worker: diff --git a/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_imports_project_configs_when_requested.snap b/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_imports_project_configs_when_requested.snap index 2bc6cdd8..dcf6c590 100644 --- a/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_imports_project_configs_when_requested.snap +++ b/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_imports_project_configs_when_requested.snap @@ -21,4 +21,19 @@ expression: render_gateway_config(&input)? } } +http://pv-gateway.localhost { + bind 127.0.0.1 ::1 + respond /__pv/health "pv-gateway-health-v1" 200 +} + +https://pv-gateway.localhost { + bind 127.0.0.1 ::1 + tls { + issuer internal { + ca local + } + } + respond /__pv/health "pv-gateway-health-v1" 200 +} + import "/Users/alice/.pv/config/gateway/projects/*.Caddyfile" diff --git a/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_empty_gateway_listener.snap b/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_empty_gateway_listener.snap index 72ab520d..c89d86c0 100644 --- a/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_empty_gateway_listener.snap +++ b/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_empty_gateway_listener.snap @@ -1,6 +1,5 @@ --- source: crates/daemon/tests/gateway_config.rs -assertion_line: 132 expression: rendered --- # PV_FAKE_PORT 48080 @@ -22,6 +21,21 @@ expression: rendered } } +http://pv-gateway.localhost { + bind 127.0.0.1 ::1 + respond /__pv/health "pv-gateway-health-v1" 200 +} + +https://pv-gateway.localhost { + bind 127.0.0.1 ::1 + tls { + issuer internal { + ca local + } + } + respond /__pv/health "pv-gateway-health-v1" 200 +} + http://127.0.0.1:48080 { bind 127.0.0.1 ::1 respond "PV Gateway is running" 404 diff --git a/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_gateway_caddyfile.snap b/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_gateway_caddyfile.snap index 2bc6cdd8..dcf6c590 100644 --- a/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_gateway_caddyfile.snap +++ b/crates/daemon/tests/snapshots/gateway_config__gateway_config_renderer_outputs_gateway_caddyfile.snap @@ -21,4 +21,19 @@ expression: render_gateway_config(&input)? } } +http://pv-gateway.localhost { + bind 127.0.0.1 ::1 + respond /__pv/health "pv-gateway-health-v1" 200 +} + +https://pv-gateway.localhost { + bind 127.0.0.1 ::1 + tls { + issuer internal { + ca local + } + } + respond /__pv/health "pv-gateway-health-v1" 200 +} + import "/Users/alice/.pv/config/gateway/projects/*.Caddyfile" diff --git a/crates/daemon/tests/snapshots/gateway_reconciliation__gateway_reconciliation_starts_gateway_and_one_worker_per_php_track.snap b/crates/daemon/tests/snapshots/gateway_reconciliation__gateway_reconciliation_starts_gateway_and_one_worker_per_php_track.snap index 5dfcfc81..946661b7 100644 --- a/crates/daemon/tests/snapshots/gateway_reconciliation__gateway_reconciliation_starts_gateway_and_one_worker_per_php_track.snap +++ b/crates/daemon/tests/snapshots/gateway_reconciliation__gateway_reconciliation_starts_gateway_and_one_worker_per_php_track.snap @@ -5,9 +5,9 @@ expression: snapshot [ RuntimeObservedStateRecord { subject: Gateway, - status: Running, + status: Degraded, message: Some( - "Gateway runtime reconciled", + "Low-port routing is inactive; run `pv ports:install` to restore ports 80 and 443", ), observed_at: "", }, From 2f4e1af94918736433b09b5aa28d0cd2cb70112b Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Mon, 3 Aug 2026 23:53:25 -0400 Subject: [PATCH 3/4] test(daemon): cover PF readiness recovery states --- crates/daemon/tests/gateway_reconciliation.rs | 110 +++++++++++++++++- ...ay_running_after_bounded_public_probe.snap | 14 +++ ...lic_readiness_when_rules_are_inactive.snap | 14 +++ ...ay_running_after_bounded_public_probe.snap | 14 +++ crates/daemon/tests/supervisor_foundation.rs | 100 +++++++++++++++- 5 files changed, 250 insertions(+), 2 deletions(-) create mode 100644 crates/daemon/tests/snapshots/gateway_reconciliation__drifted_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap create mode 100644 crates/daemon/tests/snapshots/gateway_reconciliation__prepared_pf_files_do_not_force_public_readiness_when_rules_are_inactive.snap create mode 100644 crates/daemon/tests/snapshots/gateway_reconciliation__unknown_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap diff --git a/crates/daemon/tests/gateway_reconciliation.rs b/crates/daemon/tests/gateway_reconciliation.rs index 57dedc7a..0f62d53e 100644 --- a/crates/daemon/tests/gateway_reconciliation.rs +++ b/crates/daemon/tests/gateway_reconciliation.rs @@ -19,7 +19,7 @@ use std::collections::BTreeMap; use std::ffi::OsString; use std::net::TcpListener; use std::process::Output; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::time::{sleep, timeout}; const GATEWAY_RECONCILIATION_SUMMARY: &str = "Gateway runtime reconciled"; @@ -168,6 +168,114 @@ async fn gateway_reconciliation_starts_gateway_without_linked_projects() -> Resu Ok(()) } +#[tokio::test] +async fn prepared_pf_files_do_not_force_public_readiness_when_rules_are_inactive() -> Result<()> { + let tempdir = tempdir()?; + let paths = PvPaths::for_home(tempdir.path().join("home")); + let release_path = tempdir.path().join("fake-frankenphp-release"); + let fake_frankenphp = release_path.join("bin/frankenphp"); + + write_fake_frankenphp(&fake_frankenphp)?; + + let mut database = Database::open(&paths)?; + database.record_managed_resource_track_installed( + "frankenphp", + "8.4", + "fake-frankenphp-pv1", + &release_path, + )?; + let ports = available_loopback_ports(2)?; + seed_runtime_ports(&paths, &mut database, ports[0], ports[1], &[])?; + drop(database); + + let redirects = platform::PfRedirectConfig::new(ports[0], ports[1]); + fs::write_sensitive_file(&paths.pf_anchor_config(), &redirects.render_anchor())?; + fs::write_sensitive_file( + &paths.pf_conf_reference_config(), + &platform::PfConfReference.render(), + )?; + + let started_at = Instant::now(); + let summary = reconcile_gateway_runtimes_with_pf_state_for_test( + &paths, + Duration::from_millis(100), + GatewayPfRoutingState::Inactive, + ) + .await?; + + assert_eq!(summary, GATEWAY_RECONCILIATION_SUMMARY); + assert!(started_at.elapsed() < Duration::from_secs(2)); + assert_runtime_states_snapshot( + "prepared_pf_files_do_not_force_public_readiness_when_rules_are_inactive", + Database::open(&paths)?.runtime_observed_states()?, + )?; + + stop_runtime_from_pid_file(&paths.gateway_pid()).await?; + + Ok(()) +} + +#[tokio::test] +async fn unknown_pf_state_keeps_owned_gateway_running_after_bounded_public_probe() -> Result<()> { + assert_uncertain_pf_state_preserves_gateway( + GatewayPfRoutingState::Unknown, + "unknown_pf_state_keeps_owned_gateway_running_after_bounded_public_probe", + ) + .await +} + +#[tokio::test] +async fn drifted_pf_state_keeps_owned_gateway_running_after_bounded_public_probe() -> Result<()> { + assert_uncertain_pf_state_preserves_gateway( + GatewayPfRoutingState::Drifted, + "drifted_pf_state_keeps_owned_gateway_running_after_bounded_public_probe", + ) + .await +} + +async fn assert_uncertain_pf_state_preserves_gateway( + pf_routing_state: GatewayPfRoutingState, + snapshot_name: &str, +) -> Result<()> { + let tempdir = tempdir()?; + let paths = PvPaths::for_home(tempdir.path().join("home")); + let release_path = tempdir.path().join("fake-frankenphp-release"); + let fake_frankenphp = release_path.join("bin/frankenphp"); + + write_fake_frankenphp(&fake_frankenphp)?; + + let mut database = Database::open(&paths)?; + database.record_managed_resource_track_installed( + "frankenphp", + "8.4", + "fake-frankenphp-pv1", + &release_path, + )?; + let ports = available_loopback_ports(2)?; + seed_runtime_ports(&paths, &mut database, ports[0], ports[1], &[])?; + drop(database); + + let started_at = Instant::now(); + let summary = reconcile_gateway_runtimes_with_pf_state_for_test( + &paths, + Duration::from_millis(100), + pf_routing_state, + ) + .await?; + + assert_eq!(summary, GATEWAY_RECONCILIATION_SUMMARY); + assert!(started_at.elapsed() < Duration::from_secs(2)); + assert!(paths.gateway_pid().exists()); + assert_runtime_states_snapshot( + snapshot_name, + Database::open(&paths)?.runtime_observed_states()?, + )?; + + stop_runtime_from_pid_file(&paths.gateway_pid()).await?; + + Ok(()) +} + #[tokio::test] async fn gateway_reconciliation_preserves_running_runtimes_on_second_reconcile() -> Result<()> { let tempdir = tempdir()?; diff --git a/crates/daemon/tests/snapshots/gateway_reconciliation__drifted_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap b/crates/daemon/tests/snapshots/gateway_reconciliation__drifted_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap new file mode 100644 index 00000000..d2a6f4e7 --- /dev/null +++ b/crates/daemon/tests/snapshots/gateway_reconciliation__drifted_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap @@ -0,0 +1,14 @@ +--- +source: crates/daemon/tests/gateway_reconciliation.rs +expression: snapshot +--- +[ + RuntimeObservedStateRecord { + subject: Gateway, + status: Degraded, + message: Some( + "Low-port routing is drifted; run `pv ports:install` to restore ports 80 and 443", + ), + observed_at: "", + }, +] diff --git a/crates/daemon/tests/snapshots/gateway_reconciliation__prepared_pf_files_do_not_force_public_readiness_when_rules_are_inactive.snap b/crates/daemon/tests/snapshots/gateway_reconciliation__prepared_pf_files_do_not_force_public_readiness_when_rules_are_inactive.snap new file mode 100644 index 00000000..0990c421 --- /dev/null +++ b/crates/daemon/tests/snapshots/gateway_reconciliation__prepared_pf_files_do_not_force_public_readiness_when_rules_are_inactive.snap @@ -0,0 +1,14 @@ +--- +source: crates/daemon/tests/gateway_reconciliation.rs +expression: snapshot +--- +[ + RuntimeObservedStateRecord { + subject: Gateway, + status: Degraded, + message: Some( + "Low-port routing is inactive; run `pv ports:install` to restore ports 80 and 443", + ), + observed_at: "", + }, +] diff --git a/crates/daemon/tests/snapshots/gateway_reconciliation__unknown_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap b/crates/daemon/tests/snapshots/gateway_reconciliation__unknown_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap new file mode 100644 index 00000000..75b70b82 --- /dev/null +++ b/crates/daemon/tests/snapshots/gateway_reconciliation__unknown_pf_state_keeps_owned_gateway_running_after_bounded_public_probe.snap @@ -0,0 +1,14 @@ +--- +source: crates/daemon/tests/gateway_reconciliation.rs +expression: snapshot +--- +[ + RuntimeObservedStateRecord { + subject: Gateway, + status: Degraded, + message: Some( + "Low-port routing is unknown; run `pv ports:install` to verify ports 80 and 443", + ), + observed_at: "", + }, +] diff --git a/crates/daemon/tests/supervisor_foundation.rs b/crates/daemon/tests/supervisor_foundation.rs index 1f5ab641..4481782f 100644 --- a/crates/daemon/tests/supervisor_foundation.rs +++ b/crates/daemon/tests/supervisor_foundation.rs @@ -15,7 +15,7 @@ use rustix::process::{Pid, test_kill_process}; use rustls::pki_types::PrivateKeyDer; use serde_json::json; use state::PvPaths; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::time::{sleep, timeout}; use tokio_rustls::TlsAcceptor; @@ -246,6 +246,104 @@ async fn gateway_https_readiness_accepts_tls_handshake_without_app_response() -> Ok(()) } +#[tokio::test] +async fn gateway_identity_readiness_verifies_http_and_https_response_bodies() -> Result<()> { + let tempdir = tempdir()?; + let ca_certificate_path = tempdir.path().join("ca.pem"); + let certified_key = + rcgen::generate_simple_self_signed(vec!["pv-gateway.localhost".to_owned()])?; + state::fs::write_sensitive_file(&ca_certificate_path, &certified_key.cert.pem())?; + let server_config = rustls::ServerConfig::builder_with_provider(Arc::new( + rustls::crypto::ring::default_provider(), + )) + .with_safe_default_protocol_versions() + .map_err(|error| anyhow!("TLS protocol configuration failed: {error}"))? + .with_no_client_auth() + .with_single_cert( + vec![certified_key.cert.der().clone()], + PrivateKeyDer::Pkcs8(certified_key.signing_key.serialize_der().into()), + )?; + let acceptor = TlsAcceptor::from(Arc::new(server_config)); + let http_listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let http_port = http_listener.local_addr()?.port(); + let https_listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let https_port = https_listener.local_addr()?.port(); + let http_server = tokio::spawn(async move { + let (mut stream, _address) = http_listener.accept().await?; + write_gateway_identity_response(&mut stream).await + }); + let https_server = tokio::spawn(async move { + let (stream, _address) = https_listener.accept().await?; + let mut stream = acceptor.accept(stream).await?; + + write_gateway_identity_response(&mut stream).await + }); + + wait_for_readiness( + ReadinessCheck::GatewayIdentity { + http_host: "127.0.0.1".to_owned(), + http_port, + https_host: "127.0.0.1".to_owned(), + https_port, + server_name: "pv-gateway.localhost".to_owned(), + path: "/__pv/health".to_owned(), + expected_body: "pv-gateway-health-v1".to_owned(), + ca_certificate_path, + }, + Duration::from_secs(1), + ) + .await?; + http_server.await??; + https_server.await??; + + Ok(()) +} + +#[tokio::test] +async fn gateway_identity_readiness_rejects_generic_tcp_listeners() -> Result<()> { + let tempdir = tempdir()?; + let ca_certificate_path = tempdir.path().join("ca.pem"); + let http_listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let http_port = http_listener.local_addr()?.port(); + let https_listener = TcpListener::bind(("127.0.0.1", 0)).await?; + let https_port = https_listener.local_addr()?.port(); + + let result = wait_for_readiness( + ReadinessCheck::GatewayIdentity { + http_host: "127.0.0.1".to_owned(), + http_port, + https_host: "127.0.0.1".to_owned(), + https_port, + server_name: "pv-gateway.localhost".to_owned(), + path: "/__pv/health".to_owned(), + expected_body: "pv-gateway-health-v1".to_owned(), + ca_certificate_path, + }, + Duration::from_millis(50), + ) + .await; + + assert!(result.is_err()); + drop(http_listener); + drop(https_listener); + + Ok(()) +} + +async fn write_gateway_identity_response(stream: &mut Stream) -> Result<(), std::io::Error> +where + Stream: AsyncRead + AsyncWrite + Unpin, +{ + let mut request = [0_u8; 1024]; + let _bytes = stream.read(&mut request).await?; + stream + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 20\r\nConnection: close\r\n\r\npv-gateway-health-v1", + ) + .await?; + stream.shutdown().await +} + #[tokio::test] async fn http_readiness_times_out_even_when_the_server_keeps_the_socket_open() -> Result<()> { let listener = TcpListener::bind(("127.0.0.1", 0)).await?; From 7ece4cadcc38608aa5c99eabd3a0bfeeb458925d Mon Sep 17 00:00:00 2001 From: Clovis Muneza Date: Mon, 3 Aug 2026 23:55:28 -0400 Subject: [PATCH 4/4] refactor(daemon): group runtime readiness policy --- crates/daemon/src/gateway.rs | 52 +++++++++++++++--------------------- 1 file changed, 22 insertions(+), 30 deletions(-) diff --git a/crates/daemon/src/gateway.rs b/crates/daemon/src/gateway.rs index 5408f0be..c06c1711 100644 --- a/crates/daemon/src/gateway.rs +++ b/crates/daemon/src/gateway.rs @@ -233,13 +233,15 @@ async fn reconcile_gateway_runtimes_with_pf_state( &supervisor, promoted_config, process_spec, - ReadinessCheck::Tcp { - host: "127.0.0.1".to_owned(), - port: worker.port, + RuntimeReadinessPlan { + check: ReadinessCheck::Tcp { + host: "127.0.0.1".to_owned(), + port: worker.port, + }, + failure_policy: ReadinessFailurePolicy::FailRuntime, + timeout: readiness_timeout, }, subject.clone(), - readiness_timeout, - ReadinessFailurePolicy::FailRuntime, ) .await?; record_runtime_observed( @@ -263,10 +265,8 @@ async fn reconcile_gateway_runtimes_with_pf_state( &supervisor, gateway_config.promoted_config, gateway_process_spec(paths, &gateway_command), - gateway_readiness.check, + gateway_readiness, RuntimeSubject::Gateway, - gateway_readiness.timeout, - gateway_readiness.failure_policy, ) .await?; record_gateway_runtime_observed(paths, pf_routing_state, readiness_outcome)?; @@ -280,7 +280,7 @@ fn gateway_readiness_plan( readiness_hostname: Option, pf_routing_state: GatewayPfRoutingState, readiness_timeout: Duration, -) -> GatewayReadinessPlan { +) -> RuntimeReadinessPlan { let ports = gateway_readiness_ports(plan, pf_routing_state); let failure_policy = match pf_routing_state { GatewayPfRoutingState::Active | GatewayPfRoutingState::Inactive => { @@ -303,7 +303,7 @@ fn gateway_readiness_plan( gateway_public_readiness_check(plan, ports) }; - GatewayReadinessPlan { + RuntimeReadinessPlan { check, failure_policy, timeout, @@ -311,7 +311,7 @@ fn gateway_readiness_plan( } #[derive(Clone, Debug, Eq, PartialEq)] -struct GatewayReadinessPlan { +struct RuntimeReadinessPlan { check: ReadinessCheck, failure_policy: ReadinessFailurePolicy, timeout: Duration, @@ -1111,21 +1111,10 @@ async fn start_or_adopt_promoted_runtime( supervisor: &ProcessSupervisor, promoted_config: PromotedConfigTree, spec: ProcessSpec, - readiness: ReadinessCheck, + readiness: RuntimeReadinessPlan, subject: RuntimeSubject, - readiness_timeout: Duration, - failure_policy: ReadinessFailurePolicy, ) -> Result { - let result = start_or_adopt_runtime( - paths, - supervisor, - spec, - readiness, - subject.clone(), - readiness_timeout, - failure_policy, - ) - .await; + let result = start_or_adopt_runtime(paths, supervisor, spec, readiness, subject.clone()).await; match result { Ok(outcome) => { @@ -1154,15 +1143,18 @@ async fn start_or_adopt_runtime( paths: &PvPaths, supervisor: &ProcessSupervisor, spec: ProcessSpec, - readiness: ReadinessCheck, + readiness: RuntimeReadinessPlan, subject: RuntimeSubject, - readiness_timeout: Duration, - failure_policy: ReadinessFailurePolicy, ) -> Result { + let RuntimeReadinessPlan { + check, + failure_policy, + timeout: readiness_timeout, + } = readiness; let result = async { if supervisor.adopt(&spec)?.is_some() { if supervisor.reload(&spec)? { - if let Err(error) = wait_for_readiness(readiness, readiness_timeout).await { + if let Err(error) = wait_for_readiness(check, readiness_timeout).await { if failure_policy == ReadinessFailurePolicy::PreserveRuntime && supervisor.verify_ownership(&spec)?.is_some() { @@ -1185,7 +1177,7 @@ async fn start_or_adopt_runtime( supervisor.adopt_recorded(&spec.pid_path, &spec.metadata_path)? { adopted.stop(Duration::from_secs(1)).await?; - } else if foreign_listener_is_ready(&readiness).await { + } else if foreign_listener_is_ready(&check).await { return Err(DaemonError::UnexpectedProtocolResponse { reason: format!( "runtime `{}` is listening but no PV-owned process could be verified", @@ -1195,7 +1187,7 @@ async fn start_or_adopt_runtime( } let mut process = supervisor.start(spec.clone()).await?; - if let Err(error) = wait_for_readiness(readiness, readiness_timeout).await { + if let Err(error) = wait_for_readiness(check, readiness_timeout).await { record_runtime_readiness_diagnostics(paths, &spec, &mut process, &error); if failure_policy == ReadinessFailurePolicy::PreserveRuntime && !process.has_exited()? { return Ok(RuntimeReadinessOutcome::Unverified);