diff --git a/CHANGELOG.md b/CHANGELOG.md index 0e5c1a77..8a51bd4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- Support for configuring which reverse proxies the webserver trusts `X-Forwarded-*` headers from, via `spec.webservers.roleConfig.trustedProxies` ([#835]). + ### Changed - Internal operator refactoring: introduce a build() step in the reconciler that @@ -16,6 +20,7 @@ [#821]: https://github.com/stackabletech/airflow-operator/pull/821 [#827]: https://github.com/stackabletech/airflow-operator/pull/827 [#828]: https://github.com/stackabletech/airflow-operator/pull/828 +[#835]: https://github.com/stackabletech/airflow-operator/pull/835 ## [26.7.0] - 2026-07-21 diff --git a/docs/modules/airflow/pages/usage-guide/reverse-proxy.adoc b/docs/modules/airflow/pages/usage-guide/reverse-proxy.adoc new file mode 100644 index 00000000..d2bfbffc --- /dev/null +++ b/docs/modules/airflow/pages/usage-guide/reverse-proxy.adoc @@ -0,0 +1,78 @@ += Running behind a reverse proxy +:description: Configure which reverse proxies the Airflow webserver trusts X-Forwarded-* headers from. + +When the Airflow webserver is reached through an ingress controller or another reverse proxy, every +request arrives from the proxy rather than from the client. Unless the webserver is told to trust +the proxy, it records the proxy's address as the client address, and it treats a TLS-terminated +request as plain HTTP -- which, among other things, means session cookies are not marked as +`Secure`. + +Set `trustedProxies` on the webserver role to the addresses your proxy sends requests from: + +[source,yaml] +---- +spec: + webservers: + roleConfig: + listenerClass: external-stable + trustedProxies: + - 10.244.0.0/16 # <1> + roleGroups: + default: + replicas: 1 +---- + +<1> IP addresses (`10.0.0.1`), CIDR networks (`10.244.0.0/16`), and `*` are accepted. `*` must be +the only entry in the list: combining it with other entries is rejected, since it would silently +degrade to trusting only those other entries. + +On Airflow 3.x the operator starts the api-server with `--proxy-headers` and restricts the headers +to the listed peers. Airflow 2.x has no equivalent restriction: see +<>. + +== Choosing the value + +Use the narrowest range that covers your proxy, and give it as a network address (host bits zero, +e.g. `10.244.0.0/16`) or as a bare host IP (no `/prefix` at all, e.g. `10.244.0.5`). + +For an ingress controller running in the cluster, list the Pod addresses as bare host IPs rather +than guessing a network: + +[source,bash] +---- +kubectl get pods -n ingress-nginx -o jsonpath='{.items[*].status.podIP}' +---- + +If your ingress controller's Pods share a known, stable CIDR (for example a dedicated node pool or +a documented Pod CIDR range for that namespace), you can use that network instead -- but confirm it +against your cluster's actual CNI configuration rather than inferring it from a single Pod IP, +since a Pod IP alone does not tell you where the network boundary is. + +`*` trusts forwarded headers from any peer that can reach the webserver. Only use it when access +to the webserver is restricted by other means -- with a `cluster-internal` or +`external-unstable` ListenerClass, clients reach the Pod directly and can set the headers +themselves. + +WARNING: A peer that is trusted can set the client address that ends up in the webserver's access +log. Do not list ranges wider than the proxies you operate. + +[#airflow-2-x-has-no-peer-restriction] +== Airflow 2.x has no peer restriction + +Airflow 2.x's webserver has no analogue support for trusted network addresses. Once enabled, it +unconditionally trusts `X-Forwarded-*` from *any* peer, the same as `trustedProxies: ["*"]` on 3.x. + +This means that for Airflow 2.x, `trustedProxies: ["*"]` is the only valid configuration. Using +any other value than `*` will lead the operator to reject reconciliation. + +== Interaction with overrides + +On Airflow 3.x, the operator sets `FORWARDED_ALLOW_IPS` from this field, before `envOverrides` are +applied -- so an `envOverrides` entry for that variable wins over the value derived from +`trustedProxies`. This override only takes effect while `trustedProxies` is non-empty: an empty +list means the api-server is not started with `--proxy-headers` at all, so uvicorn never installs +the middleware that reads `FORWARDED_ALLOW_IPS`, and the override has no effect regardless of its +value. + +On Airflow 2.x, the same applies to `AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX` and +`AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR`. diff --git a/docs/modules/airflow/partials/nav.adoc b/docs/modules/airflow/partials/nav.adoc index e84b005f..cf425e30 100644 --- a/docs/modules/airflow/partials/nav.adoc +++ b/docs/modules/airflow/partials/nav.adoc @@ -8,6 +8,7 @@ ** xref:airflow:usage-guide/mounting-dags.adoc[] ** xref:airflow:usage-guide/applying-custom-resources.adoc[] ** xref:airflow:usage-guide/listenerclass.adoc[] +** xref:airflow:usage-guide/reverse-proxy.adoc[] ** xref:airflow:usage-guide/storage-resources.adoc[] ** xref:airflow:usage-guide/security.adoc[] ** xref:airflow:usage-guide/logging.adoc[] diff --git a/extra/crds.yaml b/extra/crds.yaml index d5e18603..e5768124 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -5648,6 +5648,7 @@ spec: podDisruptionBudget: enabled: true maxUnavailable: null + trustedProxies: [] description: This is a product-agnostic RoleConfig, which is sufficient for most of the products. properties: listenerClass: @@ -5688,6 +5689,24 @@ spec: nullable: true type: integer type: object + trustedProxies: + default: [] + description: |- + Enable trusted proxies when Airflow is deployed behind a reverse proxy like Istio or nginx. + + The reverse proxies whose `X-Forwarded-*` headers the webserver trusts, as IP addresses + (`10.0.0.1`), CIDR networks (`10.244.0.0/16`), or `*` for every peer. `*` must be the + only entry in the list if used: combining it with other entries is rejected, since it + would silently degrade to trusting only those other entries. + + On Airflow 3.x this restricts trust to the listed peers. On 2.x it only switches + forwarded-header handling on or off: any non-empty list trusts every peer, the same as + `*`, since Airflow 2.x has no way to restrict it further. + + Leave this empty (the default) and forwarded headers are ignored entirely. + items: + type: string + type: array type: object roleGroups: additionalProperties: @@ -11854,6 +11873,7 @@ spec: podDisruptionBudget: enabled: true maxUnavailable: null + trustedProxies: [] description: This is a product-agnostic RoleConfig, which is sufficient for most of the products. properties: listenerClass: @@ -11894,6 +11914,24 @@ spec: nullable: true type: integer type: object + trustedProxies: + default: [] + description: |- + Enable trusted proxies when Airflow is deployed behind a reverse proxy like Istio or nginx. + + The reverse proxies whose `X-Forwarded-*` headers the webserver trusts, as IP addresses + (`10.0.0.1`), CIDR networks (`10.244.0.0/16`), or `*` for every peer. `*` must be the + only entry in the list if used: combining it with other entries is rejected, since it + would silently degrade to trusting only those other entries. + + On Airflow 3.x this restricts trust to the listed peers. On 2.x it only switches + forwarded-header handling on or off: any non-empty list trusts every peer, the same as + `*`, since Airflow 2.x has no way to restrict it further. + + Leave this empty (the default) and forwarded headers are ignored entirely. + items: + type: string + type: array type: object roleGroups: additionalProperties: diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index b41f4dc4..5d36cbb4 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -209,6 +209,16 @@ pub(crate) mod test_support { /// carries several resolved types (git-sync resources, validated logging, …) that are /// impractical to construct by hand. pub fn validated_cluster(executor_key: &str, executor_config: &str) -> ValidatedCluster { + validated_cluster_with(executor_key, executor_config, |_| {}) + } + + /// As [`validated_cluster`], but `patch` may modify the parsed CR before it is validated — + /// used to add role-level config that the base fixture does not carry. + pub fn validated_cluster_with( + executor_key: &str, + executor_config: &str, + patch: impl FnOnce(&mut serde_yaml::Value), + ) -> ValidatedCluster { let cluster_yaml = r#" apiVersion: airflow.stackable.tech/v1alpha2 kind: AirflowCluster @@ -250,6 +260,9 @@ pub(crate) mod test_support { executor_key.into(), serde_yaml::from_str(executor_config).expect("the executor config is valid YAML"), ); + + patch(&mut cluster_value); + let cluster: v1alpha2::AirflowCluster = serde_yaml::with::singleton_map_recursive::deserialize(cluster_value) .expect("the test CR deserialises"); @@ -268,6 +281,31 @@ pub(crate) mod test_support { .expect("test cluster validates") } + /// A Celery-executor cluster whose webserver trusts the given reverse proxies. + pub fn cluster_with_trusted_proxies(trusted_proxies: &[&str]) -> ValidatedCluster { + let trusted_proxies: Vec = trusted_proxies + .iter() + .map(|proxy| serde_yaml::Value::String((*proxy).to_owned())) + .collect(); + + validated_cluster_with( + "celeryExecutors", + "{config: {}, roleGroups: {}}", + |cluster| { + cluster["spec"]["webservers"] + .as_mapping_mut() + .expect("the webservers role is a mapping") + .insert( + "roleConfig".into(), + serde_yaml::Value::Mapping(serde_yaml::Mapping::from_iter([( + serde_yaml::Value::String("trustedProxies".to_owned()), + serde_yaml::Value::Sequence(trusted_proxies), + )])), + ); + }, + ) + } + /// Validated cluster with a Celery executor (its workers are provisioned via the queue, so no /// executor pod template is built). pub fn celery_executor_cluster() -> ValidatedCluster { @@ -289,8 +327,12 @@ mod tests { use super::{ build, - test_support::{app_version_label, celery_executor_cluster, kubernetes_executor_cluster}, + test_support::{ + app_version_label, celery_executor_cluster, cluster_with_trusted_proxies, + kubernetes_executor_cluster, validated_cluster_with, + }, }; + use crate::controller::ValidatedCluster; fn sorted_names(resources: &[impl Resource]) -> Vec<&str> { let mut names: Vec<&str> = resources @@ -301,6 +343,85 @@ mod tests { names } + /// The bash arguments of the `airflow` container of the given role group's StatefulSet. + fn airflow_container_args(cluster: &ValidatedCluster, stateful_set_name: &str) -> String { + let resources = build(cluster).expect("build succeeds"); + let stateful_set = resources + .stateful_sets + .iter() + .find(|sts| sts.meta().name.as_deref() == Some(stateful_set_name)) + .expect("the StatefulSet exists"); + + stateful_set + .spec + .as_ref() + .expect("the StatefulSet has a spec") + .template + .spec + .as_ref() + .expect("the Pod template has a spec") + .containers + .iter() + .find(|container| container.name == "airflow") + .expect("the airflow container exists") + .args + .as_ref() + .expect("the airflow container has args") + .join("\n") + } + + /// The environment of the `airflow` container of the given StatefulSet, as name/value pairs. + fn airflow_container_env( + cluster: &ValidatedCluster, + stateful_set_name: &str, + ) -> BTreeMap> { + let resources = build(cluster).expect("build succeeds"); + let stateful_set = resources + .stateful_sets + .iter() + .find(|sts| sts.meta().name.as_deref() == Some(stateful_set_name)) + .expect("the StatefulSet exists"); + + stateful_set + .spec + .as_ref() + .expect("the StatefulSet has a spec") + .template + .spec + .as_ref() + .expect("the Pod template has a spec") + .containers + .iter() + .find(|container| container.name == "airflow") + .expect("the airflow container exists") + .env + .as_ref() + .expect("the airflow container has env vars") + .iter() + .map(|env_var| (env_var.name.clone(), env_var.value.clone())) + .collect() + } + + #[test] + fn webserver_start_command_passes_proxy_headers_when_proxies_are_trusted() { + let cluster = cluster_with_trusted_proxies(&["10.244.0.0/16"]); + let args = airflow_container_args(&cluster, "my-airflow-webserver-default"); + + assert!( + args.contains("airflow api-server --proxy-headers &"), + "args were:\n{args}" + ); + } + + #[test] + fn webserver_start_command_omits_proxy_headers_by_default() { + let cluster = celery_executor_cluster(); + let args = airflow_container_args(&cluster, "my-airflow-webserver-default"); + + assert!(args.contains("airflow api-server &"), "args were:\n{args}"); + assert!(!args.contains("--proxy-headers"), "args were:\n{args}"); + } + #[test] fn build_produces_expected_resource_names() { let cluster = celery_executor_cluster(); @@ -402,4 +523,100 @@ mod tests { ] ); } + + #[test] + fn trusted_proxies_are_rendered_as_a_comma_separated_list() { + let cluster = cluster_with_trusted_proxies(&["10.244.0.0/16", "192.168.1.1"]); + let env = airflow_container_env(&cluster, "my-airflow-webserver-default"); + + assert_eq!( + env.get("FORWARDED_ALLOW_IPS"), + Some(&Some("10.244.0.0/16,192.168.1.1".to_string())) + ); + } + + #[test] + fn no_proxy_env_var_without_trusted_proxies() { + let cluster = celery_executor_cluster(); + let env = airflow_container_env(&cluster, "my-airflow-webserver-default"); + + assert_eq!(env.get("FORWARDED_ALLOW_IPS"), None); + } + + /// The scheduler runs no HTTP server, so it must not carry the proxy configuration even when + /// the webserver does. + #[test] + fn trusted_proxies_are_not_set_on_other_roles() { + let cluster = cluster_with_trusted_proxies(&["10.244.0.0/16"]); + let env = airflow_container_env(&cluster, "my-airflow-scheduler-default"); + + assert_eq!(env.get("FORWARDED_ALLOW_IPS"), None); + } + + /// Builds a 2.x Celery-executor cluster whose webserver trusts the given reverse proxies. + fn cluster_with_trusted_proxies_on_airflow_2x(trusted_proxies: &[&str]) -> ValidatedCluster { + let trusted_proxies: Vec = trusted_proxies + .iter() + .map(|proxy| serde_yaml::Value::String((*proxy).to_owned())) + .collect(); + + validated_cluster_with( + "celeryExecutors", + "{config: {}, roleGroups: {}}", + move |cluster| { + cluster["spec"]["image"]["productVersion"] = serde_yaml::Value::from("2.9.3"); + cluster["spec"]["webservers"] + .as_mapping_mut() + .expect("the webservers role is a mapping") + .insert( + "roleConfig".into(), + serde_yaml::Value::Mapping(serde_yaml::Mapping::from_iter([( + serde_yaml::Value::String("trustedProxies".to_owned()), + serde_yaml::Value::Sequence(trusted_proxies.clone()), + )])), + ); + }, + ) + } + + #[test] + fn trusted_proxies_enables_proxy_fix_on_airflow_2x() { + let cluster = cluster_with_trusted_proxies_on_airflow_2x(&["10.244.0.0/16"]); + + let args = airflow_container_args(&cluster, "my-airflow-webserver-default"); + assert!(args.contains("airflow webserver &"), "args were:\n{args}"); + assert!(!args.contains("--proxy-headers"), "args were:\n{args}"); + + let env = airflow_container_env(&cluster, "my-airflow-webserver-default"); + assert_eq!(env.get("FORWARDED_ALLOW_IPS"), None); + assert_eq!( + env.get("AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX"), + Some(&Some("True".to_string())) + ); + assert_eq!( + env.get("AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR"), + Some(&Some("1".to_string())) + ); + } + + #[test] + fn wildcard_trusted_proxies_enables_proxy_fix_on_airflow_2x() { + let cluster = cluster_with_trusted_proxies_on_airflow_2x(&["*"]); + + let env = airflow_container_env(&cluster, "my-airflow-webserver-default"); + assert_eq!( + env.get("AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX"), + Some(&Some("True".to_string())) + ); + } + + /// No `trustedProxies` means no `ProxyFix` on Airflow 2.x either. + #[test] + fn no_proxy_fix_without_trusted_proxies_on_airflow_2x() { + let cluster = cluster_with_trusted_proxies_on_airflow_2x(&[]); + + let env = airflow_container_env(&cluster, "my-airflow-webserver-default"); + assert_eq!(env.get("AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX"), None); + assert_eq!(env.get("AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR"), None); + } } diff --git a/rust/operator-binary/src/controller/build/properties/env_vars.rs b/rust/operator-binary/src/controller/build/properties/env_vars.rs index 1d4ec9b7..fbb0e89f 100644 --- a/rust/operator-binary/src/controller/build/properties/env_vars.rs +++ b/rust/operator-binary/src/controller/build/properties/env_vars.rs @@ -23,6 +23,7 @@ use crate::{ internal_secret::{ FERNET_KEY_SECRET_KEY, INTERNAL_SECRET_SECRET_KEY, JWT_SECRET_SECRET_KEY, }, + trusted_proxies::TrustedProxy, }, util::{env_var_from_secret, role_service_name}, }; @@ -513,6 +514,38 @@ fn add_version_specific_env_vars( ..Default::default() }, ); + + // `--proxy-headers` (added to the start command by + // `AirflowRole::proxy_headers_argument`) only makes the api-server *look* at + // `X-Forwarded-*`; this is what restricts the peers whose headers it trusts. + // Without it uvicorn trusts only 127.0.0.1, which no Kubernetes ingress ever is, + // and the flag has no effect at all. + // + // This covers the uvicorn backend, the only one the SDP image can run. With + // `[api] server_type = gunicorn` the variable is ignored, because Airflow passes + // `forwarded_allow_ips="*"` explicitly - see the plan's "Known gaps". + // + // Inserted before `envOverrides` are applied, so a user can still override it. + let trusted_proxies = cluster + .role_configs + .get(airflow_role) + .map(|role_config| role_config.trusted_proxies.as_slice()) + .unwrap_or_default() + .iter() + .map(TrustedProxy::to_string) + .collect::>() + .join(","); + + if !trusted_proxies.is_empty() { + env.insert( + "FORWARDED_ALLOW_IPS".into(), + EnvVar { + name: "FORWARDED_ALLOW_IPS".into(), + value: Some(trusted_proxies), + ..Default::default() + }, + ); + } } } else { env.insert( @@ -525,6 +558,46 @@ fn add_version_specific_env_vars( ..Default::default() }, ); + + // The 2.x uses Werkzeug's `ProxyFix` to allow forwarded-headers and it does so regardless + // of the peer source. The only valid value for `spec.webservers.roleConfig.trustedProxies` is `["*"]`. + if airflow_role == &AirflowRole::Webserver { + let trusted_proxies = cluster + .role_configs + .get(airflow_role) + .map(|role_config| role_config.trusted_proxies.as_slice()) + .unwrap_or_default(); + + if !trusted_proxies.is_empty() { + env.insert( + "AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX".into(), + EnvVar { + name: "AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX".into(), + value: Some("True".into()), + ..Default::default() + }, + ); + env.insert( + "AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR".into(), + EnvVar { + name: "AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR".into(), + value: Some("1".into()), + ..Default::default() + }, + ); + + if !trusted_proxies.iter().any(TrustedProxy::is_wildcard) { + let product_version = &cluster.image.product_version; + tracing::warn!( + "spec.webservers.roleConfig.trustedProxies lists specific addresses, but \ + Airflow {product_version}'s webserver has no way to restrict forwarded \ + headers to specific peers -- once enabled it trusts X-Forwarded-* from \ + any peer, the same as \"*\"", + ); + } + } + } + if cluster.has_role(&AirflowRole::DagProcessor) { // In airflow 2.x the dag-processor can optionally be started as a // standalone process (rather then as a scheduler subprocess), diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index f4ebef5a..93058b42 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -56,6 +56,7 @@ use crate::{ databases::{ CeleryBrokerConnection, CeleryResultBackendConnection, MetadataDatabaseConnection, }, + trusted_proxies::TrustedProxy, v1alpha2, }, }; @@ -96,6 +97,7 @@ pub struct ValidatedRoleConfig { pub pdb: Option, pub listener_class: Option, pub group_listener_name: Option, + pub trusted_proxies: Vec, } /// Per-rolegroup configuration: the merged CRD config plus overrides. diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 971ed425..ca5a0a4f 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -92,6 +92,11 @@ pub enum Error { #[snafu(display("invalid git-sync specification"))] InvalidGitSyncSpec { source: git_sync::v1alpha2::Error }, + + #[snafu(display("failed to parse the trusted proxies configured for the webserver role"))] + ParseTrustedProxies { + source: crate::crd::trusted_proxies::Error, + }, } pub fn validate_cluster( @@ -142,6 +147,9 @@ pub fn validate_cluster( .map(|rc| rc.pod_disruption_budget), listener_class: role.listener_class_name(airflow), group_listener_name: airflow.group_listener_name(&role), + trusted_proxies: role + .trusted_proxies(airflow) + .context(ParseTrustedProxiesSnafu)?, }, ); diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 961c073d..3147abe1 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -64,6 +64,7 @@ use crate::{ databases::{ CeleryBrokerConnection, CeleryResultBackendConnection, MetadataDatabaseConnection, }, + trusted_proxies::TrustedProxy, }, util::role_service_name, }; @@ -73,6 +74,7 @@ pub mod authentication; pub mod authorization; pub mod databases; pub mod internal_secret; +pub mod trusted_proxies; pub const APP_NAME: &str = "airflow"; pub const FIELD_MANAGER: &str = "airflow-operator"; @@ -331,6 +333,21 @@ pub mod versioned { /// This field controls which [ListenerClass](https://docs.stackable.tech/home/nightly/listener-operator/listenerclass.html) is used to expose the webserver. #[serde(default = "webserver_default_listener_class")] pub listener_class: ListenerClassName, + + /// Enable trusted proxies when Airflow is deployed behind a reverse proxy like Istio or nginx. + /// + /// The reverse proxies whose `X-Forwarded-*` headers the webserver trusts, as IP addresses + /// (`10.0.0.1`), CIDR networks (`10.244.0.0/16`), or `*` for every peer. `*` must be the + /// only entry in the list if used: combining it with other entries is rejected, since it + /// would silently degrade to trusting only those other entries. + /// + /// On Airflow 3.x this restricts trust to the listed peers. On 2.x it only switches + /// forwarded-header handling on or off: any non-empty list trusts every peer, the same as + /// `*`, since Airflow 2.x has no way to restrict it further. + /// + /// Leave this empty (the default) and forwarded headers are ignored entirely. + #[serde(default)] + pub trusted_proxies: Vec, } } @@ -360,6 +377,7 @@ impl Default for v1alpha2::WebserverRoleConfig { fn default() -> Self { v1alpha2::WebserverRoleConfig { listener_class: webserver_default_listener_class(), + trusted_proxies: Vec::new(), common: Default::default(), } } @@ -574,7 +592,10 @@ impl AirflowRole { command.extend(vec![ "prepare_signal_handlers".to_string(), container_debug_command(), - "airflow api-server &".to_string(), + format!( + "airflow api-server{} &", + self.proxy_headers_argument(cluster) + ), ]); } AirflowRole::Scheduler => { @@ -687,6 +708,31 @@ impl AirflowRole { command } + /// The api-server only looks at `X-Forwarded-*` headers when started with `--proxy-headers`. + /// Which peers those headers are trusted from is configured separately, through environment + /// variables — see `env_vars::add_version_specific_env_vars`. Both halves are driven by the + /// same `trustedProxies` field, because either one alone has no effect. + /// + /// Only the webserver runs the api-server; every other role returns the empty string, + /// regardless of what is configured, so that a future call from another role's match arm + /// cannot silently emit the flag for a role with no api-server. + fn proxy_headers_argument(&self, cluster: &ValidatedCluster) -> &'static str { + if !matches!(self, AirflowRole::Webserver) { + return ""; + } + + let has_trusted_proxies = cluster + .role_configs + .get(&AirflowRole::Webserver) + .is_some_and(|role_config| !role_config.trusted_proxies.is_empty()); + + if has_trusted_proxies { + " --proxy-headers" + } else { + "" + } + } + fn authentication_start_commands( auth_config: &AirflowClientAuthenticationDetailsResolved, ) -> Vec { @@ -744,6 +790,30 @@ impl AirflowRole { Self::Worker | Self::Scheduler | Self::DagProcessor | Self::Triggerer => None, } } + + /// The reverse proxies this role trusts `X-Forwarded-*` headers from. + /// + /// Only the webserver serves HTTP, so every other role returns an empty list regardless of + /// what the webserver role configured. + pub fn trusted_proxies( + &self, + airflow: &v1alpha2::AirflowCluster, + ) -> Result, trusted_proxies::Error> { + match self { + Self::Webserver => { + let entries: Vec = airflow + .spec + .webservers + .iter() + .flat_map(|webserver| &webserver.role_config.trusted_proxies) + .map(|trusted_proxy| TrustedProxy::from_str(trusted_proxy)) + .collect::>()?; + trusted_proxies::ensure_wildcard_is_sole_entry(&entries)?; + Ok(entries) + } + Self::Worker | Self::Scheduler | Self::DagProcessor | Self::Triggerer => Ok(Vec::new()), + } + } } fn container_debug_command() -> String { @@ -970,7 +1040,10 @@ mod tests { versioned::test_utils::RoundtripTestData, }; - use crate::{v1alpha1, v1alpha2}; + use crate::{ + crd::{AirflowRole, trusted_proxies::TrustedProxy}, + v1alpha1, v1alpha2, + }; #[test] fn test_cluster_config() { @@ -1025,6 +1098,146 @@ mod tests { assert!(cluster.spec.cluster_config.database_initialization.enabled); } + #[test] + fn webserver_role_config_defaults_to_no_trusted_proxies() { + let role_config: v1alpha2::WebserverRoleConfig = + serde_yaml::from_str("listenerClass: external-stable") + .expect("the role config deserialises"); + + assert!(role_config.trusted_proxies.is_empty()); + } + + #[test] + fn webserver_role_config_accepts_trusted_proxies() { + let role_config: v1alpha2::WebserverRoleConfig = serde_yaml::from_str( + " + trustedProxies: + - 10.244.0.0/16 + - 192.168.1.1 + ", + ) + .expect("the role config deserialises"); + + assert_eq!( + role_config.trusted_proxies, + ["10.244.0.0/16", "192.168.1.1"] + ); + } + + /// A cluster CR with the given `webservers.roleConfig` block spliced in. + fn test_cluster_with_webserver_role_config(role_config: &str) -> v1alpha2::AirflowCluster { + let cluster = formatdoc! {" + apiVersion: airflow.stackable.tech/v1alpha2 + kind: AirflowCluster + metadata: + name: airflow + spec: + image: + productVersion: 3.2.2 + clusterConfig: + credentialsSecretName: airflow-admin-credentials + metadataDatabase: + postgresql: + host: airflow-postgresql + database: airflow + credentialsSecretName: airflow-postgresql-credentials + webservers: + roleConfig: + {role_config} + roleGroups: + default: + config: {{}} + kubernetesExecutors: + config: {{}} + "}; + + let deserializer = serde_yaml::Deserializer::from_str(&cluster); + serde_yaml::with::singleton_map_recursive::deserialize(deserializer) + .expect("the test CR deserialises") + } + + #[test] + fn webserver_trusted_proxies_are_parsed() { + let cluster = test_cluster_with_webserver_role_config( + " trustedProxies:\n - 10.244.0.0/16\n - 192.168.1.1", + ); + + let trusted_proxies = AirflowRole::Webserver + .trusted_proxies(&cluster) + .expect("the trusted proxies are valid"); + + let rendered: Vec = trusted_proxies + .iter() + .map(TrustedProxy::to_string) + .collect(); + assert_eq!(rendered, ["10.244.0.0/16", "192.168.1.1"]); + } + + #[test] + fn wildcard_combined_with_another_entry_is_rejected() { + let cluster = test_cluster_with_webserver_role_config( + " trustedProxies:\n - \"*\"\n - 10.0.0.0/8", + ); + + let error = AirflowRole::Webserver + .trusted_proxies(&cluster) + .expect_err("* combined with another entry must be rejected"); + assert!( + matches!( + error, + crate::crd::trusted_proxies::Error::WildcardMustBeSoleEntry + ), + "error was: {error:?}" + ); + } + + #[test] + fn an_invalid_trusted_proxy_is_rejected() { + let cluster = test_cluster_with_webserver_role_config( + " trustedProxies:\n - airflow.example.com", + ); + + AirflowRole::Webserver + .trusted_proxies(&cluster) + .expect_err("a hostname is not a valid trusted proxy"); + } + + /// Only the webserver serves HTTP, so no other role may pick the setting up even if a + /// webserver configured it. + #[test] + fn non_webserver_roles_have_no_trusted_proxies() { + let cluster = test_cluster_with_webserver_role_config( + " trustedProxies:\n - 10.244.0.0/16", + ); + + for role in [ + AirflowRole::Scheduler, + AirflowRole::Worker, + AirflowRole::DagProcessor, + AirflowRole::Triggerer, + ] { + assert!( + role.trusted_proxies(&cluster) + .expect("no proxies to parse") + .is_empty(), + "role {role} must not have trusted proxies" + ); + } + } + + #[test] + fn a_webserver_without_trusted_proxies_yields_an_empty_list() { + let cluster = + test_cluster_with_webserver_role_config(" listenerClass: external-stable"); + + assert!( + AirflowRole::Webserver + .trusted_proxies(&cluster) + .expect("nothing to parse") + .is_empty() + ); + } + impl RoundtripTestData for v1alpha1::AirflowClusterSpec { fn roundtrip_test_data() -> Vec { let git_sync_section = r#" diff --git a/rust/operator-binary/src/crd/trusted_proxies.rs b/rust/operator-binary/src/crd/trusted_proxies.rs new file mode 100644 index 00000000..abb3bc14 --- /dev/null +++ b/rust/operator-binary/src/crd/trusted_proxies.rs @@ -0,0 +1,362 @@ +use std::{ + fmt::Display, + net::{IpAddr, Ipv4Addr, Ipv6Addr}, + num::ParseIntError, + str::FromStr, +}; + +use snafu::{ResultExt, Snafu, ensure}; + +/// Trusts every peer, regardless of its address. +const WILDCARD: &str = "*"; + +#[derive(Debug, Snafu)] +pub enum Error { + #[snafu(display( + "the trusted proxy {value:?} is neither an IP address, a CIDR network, nor {WILDCARD:?}" + ))] + InvalidIpAddress { + value: String, + source: std::net::AddrParseError, + }, + + #[snafu(display( + "the trusted proxy {value:?} has a prefix length that is not a whole number between 0 \ + and the address family's maximum" + ))] + InvalidPrefixLength { + value: String, + source: ParseIntError, + }, + + #[snafu(display( + "the trusted proxy {value:?} has a prefix length of {prefix_length}, which exceeds the \ + maximum of {maximum} for its address family" + ))] + PrefixLengthOutOfRange { + value: String, + prefix_length: u8, + maximum: u8, + }, + + #[snafu(display("the trusted proxy {value:?} has host bits set; did you mean {masked:?}?"))] + HostBitsSet { value: String, masked: String }, + + #[snafu(display( + "{WILDCARD:?} trusts every peer, so it must be the only entry in the trusted proxy list; \ + remove the other entries or replace them all with {WILDCARD:?}" + ))] + WildcardMustBeSoleEntry, +} + +/// A single entry of the trusted-proxy list: an IP address (`10.0.0.1`), a CIDR network +/// (`10.244.0.0/16`), or `*` for every peer. +/// +/// The value is handed to Airflow verbatim, which is why it is kept as a string rather than a +/// parsed network: uvicorn accepts all three notations, and round-tripping through an +/// `IpAddr`/network type would normalise the user's input for no benefit. +/// +/// Parsing still happens up front: an entry uvicorn cannot parse is silently treated as +/// an opaque literal that matches no peer, which disables proxy trust without any error. Failing +/// reconciliation instead makes the misconfiguration visible. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TrustedProxy(String); + +impl TrustedProxy { + pub fn is_wildcard(&self) -> bool { + self.0 == WILDCARD + } +} + +impl FromStr for TrustedProxy { + type Err = Error; + + fn from_str(value: &str) -> Result { + if value == WILDCARD { + return Ok(Self(value.to_owned())); + } + + // Split at the *last* `/` so that a doubled prefix like `10.0.0.0/16/24` leaves + // `10.0.0.0/16` as the address part and is rejected as an address rather than as a + // prefix length, which is the clearer of the two messages. + let (address, prefix_length) = match value.rsplit_once('/') { + Some((address, prefix_length)) => (address, Some(prefix_length)), + None => (value, None), + }; + + let address = address + .parse::() + .with_context(|_| InvalidIpAddressSnafu { + value: value.to_owned(), + })?; + + if let Some(prefix_length) = prefix_length { + // `u8::from_str` accepts a leading `+` (e.g. `+16`), but Python's `ipaddress` prefix + // parser accepts ASCII digits only. An entry we accept but uvicorn rejects is filed by + // uvicorn as an opaque string literal that matches no peer -- silently disabling proxy + // trust for exactly this entry. Require a non-empty run of ASCII digits up front so we + // reject exactly what Python's parser would reject. `chars().all(...)` is vacuously + // true for the empty string, so `10.0.0.0/` still needs the explicit `is_empty` check. + let is_invalid_prefix_length = + prefix_length.is_empty() || !prefix_length.chars().all(|c| c.is_ascii_digit()); + let prefix_length = if is_invalid_prefix_length { + // Force a `ParseIntError` of the right shape to carry as `source`, rather than + // hand-rolling one. + "".parse::() + } else { + prefix_length.parse::() + } + .with_context(|_| InvalidPrefixLengthSnafu { + value: value.to_owned(), + })?; + + let maximum = match address { + IpAddr::V4(_) => 32, + IpAddr::V6(_) => 128, + }; + + ensure!( + prefix_length <= maximum, + PrefixLengthOutOfRangeSnafu { + value, + prefix_length, + maximum, + } + ); + + // uvicorn parses this address with `ipaddress.ip_network(host, strict=True)`, which + // rejects a network address with host bits set. When that happens uvicorn falls back + // to treating the whole entry as an opaque string literal that matches no peer -- + // silently disabling proxy trust for exactly this entry. Reject it here instead, and + // suggest the masked network the user probably meant. + let masked = masked_network(address, prefix_length); + ensure!( + masked == address, + HostBitsSetSnafu { + value, + masked: format!("{masked}/{prefix_length}"), + } + ); + } + + Ok(Self(value.to_owned())) + } +} + +/// The network address for `address/prefix_length`: `address` with every bit past +/// `prefix_length` cleared. +fn masked_network(address: IpAddr, prefix_length: u8) -> IpAddr { + match address { + IpAddr::V4(addr) => { + let bits = u32::from(addr); + // A shift equal to the full width panics (debug) or masks the shift amount (release) + // rather than shifting by the full width, so the all-bits-masked-out case (prefix + // length 0) is handled separately. + let mask = if prefix_length == 0 { + 0 + } else { + u32::MAX << (32 - prefix_length) + }; + IpAddr::V4(Ipv4Addr::from(bits & mask)) + } + IpAddr::V6(addr) => { + let bits = u128::from(addr); + let mask = if prefix_length == 0 { + 0 + } else { + u128::MAX << (128 - prefix_length) + }; + IpAddr::V6(Ipv6Addr::from(bits & mask)) + } + } +} + +impl Display for TrustedProxy { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +/// Checks the list-level rule that a single `FromStr` call cannot see: uvicorn only ever trusts +/// every peer when the whole trusted-hosts value is exactly `*` (`trusted_hosts in ("*", ["*"])` +/// in uvicorn's own check), not when `*` merely appears alongside other entries. Combining `*` +/// with anything else is accepted by `FromStr` but silently degrades to trusting only the other +/// entries -- exactly the kind of silent surprise this feature exists to prevent. +pub fn ensure_wildcard_is_sole_entry(entries: &[TrustedProxy]) -> Result<(), Error> { + let has_wildcard = entries.iter().any(|entry| entry.0 == WILDCARD); + ensure!( + !has_wildcard || entries.len() == 1, + WildcardMustBeSoleEntrySnafu + ); + Ok(()) +} + +#[cfg(test)] +mod tests { + use rstest::rstest; + + use super::*; + + #[rstest] + #[case("10.244.0.0/16")] + #[case("192.168.1.1")] + #[case("::1")] + #[case("fd00::/8")] + #[case("0.0.0.0/0")] + #[case("*")] + // Leading zeros are plain ASCII digits as far as Python's `str.isdigit()` (and thus uvicorn) + // is concerned -- `'016'.isdigit()` is true and `int('016') == 16` -- so this must stay + // accepted rather than newly diverging from uvicorn in the opposite direction. + #[case("10.0.0.0/016")] + fn accepts_addresses_networks_and_wildcard(#[case] value: &str) { + let proxy = TrustedProxy::from_str(value).expect("must be accepted"); + // The string form is what is handed to Airflow, so it must survive verbatim. + assert_eq!(proxy.to_string(), value); + } + + #[rstest] + #[case("")] + #[case("not-an-ip")] + #[case("10.244.0.0/16/24")] + #[case("airflow.example.com")] + fn rejects_values_that_are_not_addresses(#[case] value: &str) { + assert!(matches!( + TrustedProxy::from_str(value), + Err(Error::InvalidIpAddress { .. }) + )); + } + + #[test] + fn rejects_a_non_numeric_prefix_length() { + assert!(matches!( + TrustedProxy::from_str("10.244.0.0/sixteen"), + Err(Error::InvalidPrefixLength { .. }) + )); + } + + /// `u8::from_str` accepts a leading `+` (`"+16".parse::()` succeeds), but Python's + /// `ipaddress` prefix parser only accepts ASCII digits, so uvicorn would reject these and + /// file them as dead literals that match no peer. `10.0.0.0/+0` in particular must be + /// reported as an invalid prefix length rather than the misleading "did you mean 0.0.0.0/0" + /// host-bits suggestion that a naive `parse::() == 0` would produce. + #[rstest] + #[case("10.0.0.0/+16")] + #[case("10.0.0.0/+32")] + #[case("fd00::/+8")] + #[case("10.0.0.0/+0")] + fn rejects_a_prefix_length_with_a_leading_sign(#[case] value: &str) { + assert!(matches!( + TrustedProxy::from_str(value), + Err(Error::InvalidPrefixLength { .. }) + )); + } + + #[rstest] + #[case("10.244.0.0/33", 33, 32)] + #[case("fd00::/129", 129, 128)] + fn rejects_a_prefix_length_beyond_the_address_family_maximum( + #[case] value: &str, + #[case] expected_prefix_length: u8, + #[case] expected_maximum: u8, + ) { + let error = TrustedProxy::from_str(value).expect_err("must be rejected"); + assert!(matches!( + error, + Error::PrefixLengthOutOfRange { + prefix_length, + maximum, + .. + } if prefix_length == expected_prefix_length && maximum == expected_maximum + )); + } + + /// A too-large numeric prefix length must not be reported as "not a number" -- it is a + /// number, just out of `u8` range, and the two failure modes need different messages. + #[test] + fn a_too_large_prefix_length_is_not_reported_as_non_numeric() { + let error = TrustedProxy::from_str("10.0.0.0/300").expect_err("must be rejected"); + assert!( + matches!(error, Error::InvalidPrefixLength { .. }), + "error was: {error:?}" + ); + assert!( + !error.to_string().contains("is not a number"), + "message was: {error}" + ); + } + + #[rstest] + #[case("10.244.0.1/16", "10.244.0.0/16")] + #[case("fd00::1/8", "fd00::/8")] + fn rejects_a_cidr_with_host_bits_set(#[case] value: &str, #[case] expected_masked: &str) { + let error = TrustedProxy::from_str(value).expect_err("must be rejected"); + assert!( + matches!(error, Error::HostBitsSet { .. }), + "error was: {error:?}" + ); + assert!( + error.to_string().contains(expected_masked), + "message was: {error}, expected it to suggest {expected_masked}" + ); + } + + /// The rendered message must name the offending value, because it is the only thing that tells + /// a user which of their list entries is wrong. + #[test] + fn error_message_names_the_offending_value() { + let error = TrustedProxy::from_str("not-an-ip").expect_err("must be rejected"); + assert!( + error.to_string().contains("not-an-ip"), + "message was: {error}" + ); + } + + #[test] + fn wildcard_alone_is_accepted() { + let entries = [TrustedProxy::from_str("*").expect("must be accepted")]; + ensure_wildcard_is_sole_entry(&entries).expect("a sole wildcard is fine"); + } + + #[test] + fn wildcard_combined_with_another_entry_is_rejected() { + let entries = [ + TrustedProxy::from_str("*").expect("must be accepted"), + TrustedProxy::from_str("10.0.0.0/8").expect("must be accepted"), + ]; + assert!(matches!( + ensure_wildcard_is_sole_entry(&entries), + Err(Error::WildcardMustBeSoleEntry) + )); + } + + #[test] + fn a_list_without_a_wildcard_is_accepted() { + let entries = [TrustedProxy::from_str("10.0.0.0/8").expect("must be accepted")]; + ensure_wildcard_is_sole_entry(&entries).expect("no wildcard involved"); + } + + /// The underlying parse error must survive as `source`, so it reaches logs and Kubernetes + /// events rather than being discarded. + #[test] + fn invalid_ip_address_keeps_the_parse_error_as_source() { + let error = TrustedProxy::from_str("not-an-ip").expect_err("must be rejected"); + match error { + Error::InvalidIpAddress { source, .. } => { + // Constructing this proves `source` is a real `AddrParseError`. + let _: std::net::AddrParseError = source; + } + other => panic!("expected InvalidIpAddress, got {other:?}"), + } + } + + #[test] + fn invalid_prefix_length_keeps_the_parse_error_as_source() { + let error = TrustedProxy::from_str("10.244.0.0/sixteen").expect_err("must be rejected"); + match error { + Error::InvalidPrefixLength { source, .. } => { + let _: ParseIntError = source; + } + other => panic!("expected InvalidPrefixLength, got {other:?}"), + } + } +} diff --git a/tests/templates/kuttl/external-access/45-assert.yaml.j2 b/tests/templates/kuttl/external-access/45-assert.yaml.j2 new file mode 100644 index 00000000..f4790ce1 --- /dev/null +++ b/tests/templates/kuttl/external-access/45-assert.yaml.j2 @@ -0,0 +1,28 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +{% if test_scenario['values']['airflow'].startswith("2") %} +commands: + # ProxyFix must be enabled on Airflow 2 + - script: | + kubectl -n $NAMESPACE get statefulset airflow-webserver-default \ + -o jsonpath='{.spec.template.spec.containers[?(@.name=="airflow")].env[?(@.name=="AIRFLOW__WEBSERVER__ENABLE_PROXY_FIX")].value}' \ + | grep -qx 'True' + - script: | + kubectl -n $NAMESPACE get statefulset airflow-webserver-default \ + -o jsonpath='{.spec.template.spec.containers[?(@.name=="airflow")].env[?(@.name=="AIRFLOW__WEBSERVER__PROXY_FIX_X_FOR")].value}' \ + | grep -qx '1' +{% else %} +commands: + # The api-server must be started with --proxy-headers ... + - script: | + kubectl -n $NAMESPACE get statefulset airflow-webserver-default \ + -o jsonpath='{.spec.template.spec.containers[?(@.name=="airflow")].args}' \ + | grep -q -- '--proxy-headers' + # ... and told which peers to trust it from. + - script: | + kubectl -n $NAMESPACE get statefulset airflow-webserver-default \ + -o jsonpath='{.spec.template.spec.containers[?(@.name=="airflow")].env[?(@.name=="FORWARDED_ALLOW_IPS")].value}' \ + | grep -qx '10.244.0.0/16' +{% endif %} diff --git a/tests/templates/kuttl/external-access/install-airflow-cluster.yaml.j2 b/tests/templates/kuttl/external-access/install-airflow-cluster.yaml.j2 index e689d94b..e4f8639b 100644 --- a/tests/templates/kuttl/external-access/install-airflow-cluster.yaml.j2 +++ b/tests/templates/kuttl/external-access/install-airflow-cluster.yaml.j2 @@ -62,6 +62,8 @@ spec: webservers: roleConfig: listenerClass: test-external-stable-$NAMESPACE + trustedProxies: + - 10.244.0.0/16 config: resources: cpu: diff --git a/tests/templates/kuttl/oidc/45-assert.yaml b/tests/templates/kuttl/oidc/45-assert.yaml new file mode 100644 index 00000000..53c338b3 --- /dev/null +++ b/tests/templates/kuttl/oidc/45-assert.yaml @@ -0,0 +1,14 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +metadata: + name: install-reverse-proxy +timeout: 300 +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: airflow-reverse-proxy +status: + readyReplicas: 1 + replicas: 1 diff --git a/tests/templates/kuttl/oidc/45-install-reverse-proxy.yaml b/tests/templates/kuttl/oidc/45-install-reverse-proxy.yaml new file mode 100644 index 00000000..cfcf9744 --- /dev/null +++ b/tests/templates/kuttl/oidc/45-install-reverse-proxy.yaml @@ -0,0 +1,96 @@ +--- +# A TLS-terminating reverse proxy in front of the webserver, used to exercise +# `trustedProxies`: the webserver is told to trust this proxy's peer address +# (see install-airflow.yaml.j2), and 60-login.yaml checks that a login done +# through this proxy actually benefits from that trust (see login.py). +apiVersion: v1 +kind: ConfigMap +metadata: + name: airflow-reverse-proxy-config +data: + nginx.conf: | + events {} + http { + server { + listen 8443 ssl; + server_name _; + + ssl_certificate /stackable/tls/tls.crt; + ssl_certificate_key /stackable/tls/tls.key; + + location / { + proxy_pass http://airflow-webserver:8080; + proxy_set_header Host $host:8443; + proxy_set_header X-Forwarded-Proto https; + proxy_set_header X-Forwarded-For $remote_addr; + } + } + } +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: airflow-reverse-proxy +spec: + replicas: 1 + selector: + matchLabels: + app: airflow-reverse-proxy + template: + metadata: + labels: + app: airflow-reverse-proxy + spec: + # The cert volume defaults to being readable only by its owning group; nginx-unprivileged + # runs as a non-root UID that is only granted access to it via this group membership. + securityContext: + fsGroup: 1000 + containers: + - name: nginx + image: nginxinc/nginx-unprivileged:alpine + ports: + - containerPort: 8443 + volumeMounts: + - name: config + mountPath: /etc/nginx/nginx.conf + subPath: nginx.conf + - name: tls + mountPath: /stackable/tls + readOnly: true + - name: cache + mountPath: /var/cache/nginx + - name: run + mountPath: /var/run + volumes: + - name: config + configMap: + name: airflow-reverse-proxy-config + - name: tls + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: tls + secrets.stackable.tech/scope: service=airflow-reverse-proxy + spec: + storageClassName: secrets.stackable.tech + accessModes: + - ReadWriteOnce + resources: + requests: + storage: "1" + - name: cache + emptyDir: {} + - name: run + emptyDir: {} +--- +apiVersion: v1 +kind: Service +metadata: + name: airflow-reverse-proxy +spec: + selector: + app: airflow-reverse-proxy + ports: + - port: 8443 + targetPort: 8443 diff --git a/tests/templates/kuttl/oidc/install-airflow.yaml.j2 b/tests/templates/kuttl/oidc/install-airflow.yaml.j2 index ea73899e..93a13ca9 100644 --- a/tests/templates/kuttl/oidc/install-airflow.yaml.j2 +++ b/tests/templates/kuttl/oidc/install-airflow.yaml.j2 @@ -71,6 +71,10 @@ spec: webservers: roleConfig: listenerClass: external-unstable + # Trusts the reverse proxy installed in 45-install-reverse-proxy.yaml (its Pod address + # falls within the cluster's Pod CIDR); see login.py for what this is expected to enable. + trustedProxies: + - 10.244.0.0/16 config: logging: enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} diff --git a/tests/templates/kuttl/oidc/login.py b/tests/templates/kuttl/oidc/login.py index 0b9da8f3..dc5a0440 100644 --- a/tests/templates/kuttl/oidc/login.py +++ b/tests/templates/kuttl/oidc/login.py @@ -37,59 +37,110 @@ def userinfo_page(base_url: str, airflow_version: str) -> str: return f"{base_url}/users/userinfo/" -session = requests.Session() -url = "http://airflow-webserver:8080" - -# Click on "Sign In with keycloak" in Airflow -login_page = session.get(login_page(url, os.environ["AIRFLOW_VERSION"])) - -assert login_page.ok, "Redirection from Airflow to Keycloak failed" - -assert_startwith( - login_page.url, - f"https://keycloak1.{os.environ['NAMESPACE']}.svc.cluster.local:8443/realms/test1/protocol/openid-connect/auth?response_type=code&client_id=airflow1", - "Redirection to the Keycloak login page expected", -) - -# Enter username and password into the Keycloak login page and click on "Sign In" -login_page_html = BeautifulSoup(login_page.text, "html.parser") -authenticate_url = login_page_html.form["action"] -welcome_page = session.post( - authenticate_url, data={"username": "jane.doe", "password": "T8mn72D9"} -) - -assert welcome_page.ok, "Login failed" -assert_equal( - welcome_page.url, f"{url}/", "Redirection to the Airflow home page expected" -) - -# Open the user information page in Airflow -userinfo_url = userinfo_page(url, os.environ["AIRFLOW_VERSION"]) -userinfo_page = session.get(userinfo_url) - -assert userinfo_page.ok, "Retrieving user information failed" -assert_equal( - userinfo_page.url, - userinfo_url, - "Redirection to the Airflow user info page expected", -) - -# Expect the user data provided by Keycloak in Airflow -userinfo_page_html = BeautifulSoup(userinfo_page.text, "html.parser") -table_rows = userinfo_page_html.find_all("tr") -user_data = {tr.find("th").text: tr.find("td").text for tr in table_rows} - -log.debug(f"{user_data=}") - -assert user_data["First Name"] == "Jane", ( - "The first name of the user in Airflow should match the one provided by Keycloak" -) -assert user_data["Last Name"] == "Doe", ( - "The last name of the user in Airflow should match the one provided by Keycloak" -) -assert user_data["Email"] == "jane.doe@stackable.tech", ( - "The email of the user in Airflow should match the one provided by Keycloak" +def auth_cookie(session: requests.Session, airflow_version: str): + # Airflow 3's api-server authenticates via this JWT cookie; there is no equivalent on + # Airflow 2, which uses Flask's own session cookie instead. + if not airflow_version.startswith("3"): + return None + return next((c for c in session.cookies if c.name == "_token"), None) + + +def login(url: str, airflow_version: str) -> requests.Session: + """Log in to Airflow via Keycloak at the given base URL and check that the OIDC data + Keycloak provides ends up correctly reflected in Airflow. Returns the session so callers + can inspect e.g. cookies afterwards.""" + + session = requests.Session() + + # Click on "Sign In with keycloak" in Airflow + login_response = session.get(login_page(url, airflow_version)) + + assert login_response.ok, "Redirection from Airflow to Keycloak failed" + + assert_startwith( + login_response.url, + f"https://keycloak1.{os.environ['NAMESPACE']}.svc.cluster.local:8443/realms/test1/protocol/openid-connect/auth?response_type=code&client_id=airflow1", + "Redirection to the Keycloak login page expected", + ) + + # Enter username and password into the Keycloak login page and click on "Sign In" + login_response_html = BeautifulSoup(login_response.text, "html.parser") + authenticate_url = login_response_html.form["action"] + welcome_response = session.post( + authenticate_url, data={"username": "jane.doe", "password": "T8mn72D9"} + ) + + assert welcome_response.ok, "Login failed" + assert_equal( + welcome_response.url, f"{url}/", "Redirection to the Airflow home page expected" + ) + + # Open the user information page in Airflow + userinfo_url = userinfo_page(url, airflow_version) + userinfo_response = session.get(userinfo_url) + + assert userinfo_response.ok, "Retrieving user information failed" + assert_equal( + userinfo_response.url, + userinfo_url, + "Redirection to the Airflow user info page expected", + ) + + # Expect the user data provided by Keycloak in Airflow + userinfo_response_html = BeautifulSoup(userinfo_response.text, "html.parser") + table_rows = userinfo_response_html.find_all("tr") + user_data = {tr.find("th").text: tr.find("td").text for tr in table_rows} + + log.debug(f"{user_data=}") + + assert user_data["First Name"] == "Jane", ( + "The first name of the user in Airflow should match the one provided by Keycloak" + ) + assert user_data["Last Name"] == "Doe", ( + "The last name of the user in Airflow should match the one provided by Keycloak" + ) + assert user_data["Email"] == "jane.doe@stackable.tech", ( + "The email of the user in Airflow should match the one provided by Keycloak" + ) + + return session + + +airflow_version = os.environ["AIRFLOW_VERSION"] + +# Log in directly against the webserver, bypassing the reverse proxy. +direct_session = login("http://airflow-webserver:8080", airflow_version) +log.info("Direct OIDC login test passed") + +# The webserver is plain HTTP here, so its auth cookie must not be marked Secure. +direct_cookie = auth_cookie(direct_session, airflow_version) +if airflow_version.startswith("3"): + assert direct_cookie is not None, "Expected an auth cookie after a successful login" + assert not direct_cookie.secure, ( + "The auth cookie must not be marked Secure when accessed directly over HTTP" + ) + +# Log in again through the TLS-terminating reverse proxy installed in +# 45-install-reverse-proxy.yaml, which is covered by trustedProxies in install-airflow.yaml.j2. +# If the webserver did not trust the proxy's forwarded headers, this either fails outright +# (the OIDC redirect_uri would be built with the wrong scheme) or silently loses the point of +# running behind a proxy (the auth cookie would not be marked Secure despite being sent over +# TLS). See docs/modules/airflow/pages/usage-guide/reverse-proxy.adoc. +proxy_url = ( + f"https://airflow-reverse-proxy.{os.environ['NAMESPACE']}.svc.cluster.local:8443" ) +proxied_session = login(proxy_url, airflow_version) +log.info("Reverse-proxied OIDC login test passed") + +proxied_cookie = auth_cookie(proxied_session, airflow_version) +if airflow_version.startswith("3"): + assert proxied_cookie is not None, ( + "Expected an auth cookie after a successful login" + ) + assert proxied_cookie.secure, ( + "The auth cookie must be marked Secure when trustedProxies allows the webserver to " + "trust the reverse proxy's X-Forwarded-Proto: https" + ) log.info("OIDC login test passed")