Compare commits

...

33 Commits

Author SHA1 Message Date
Istio Automation 85a94b6cc4
Automator: update common-files@master in istio/ztunnel@master (#1595) 2025-07-10 07:15:33 -04:00
Istio Automation 3fa6335035
Automator: update common-files@master in istio/ztunnel@master (#1589) 2025-07-08 05:44:30 -04:00
Steven Jin dfa3b58bbc
Buffer inner h2 streams (#1580)
* Buffer h2 streams

* Tests

* naming

* Review

simplify code
2025-07-07 17:52:29 -04:00
Gustavo Meira c2d2534edb
increasing limit for open files (#1586)
* increasing limit for open files

* suggestion from PR

* adding comment

* Update src/main.rs

Co-authored-by: Daniel Hawton <daniel@hawton.org>

---------

Co-authored-by: Daniel Hawton <daniel@hawton.org>
2025-07-07 15:37:29 -04:00
Krinkin, Mike 84f0e52e64
Multinetwork/Support remote networks for services with waypoints (#1565)
* Multinetwork/Support remote networks for services with waypoints

Currently `build_request` when it sees a service with a waypoint
resolves the waypoint backend and routes request there using regular
HBONE.

In multi network scenario though the waypoint may have workload on a
remote network and to reach it we have to go through E/W gateway and use
double HBONE.

This change enables handling of services with waypoint on a remote
network.

Some of the assumptions that were used when I prepared this change:

1. We assume uniformity of configuration (e.g., if service X in local
   cluster has a waypoint, then service X in remote network also has a
   waypoint, if waypoint is service addressable, then it's using service
   to address waypoint both locally and on remote network)
2  Split-horizon representation of waypoint workloads, just like with
   any regular workloads and services (e.g., in the local cluster
   instead of an actual waypoint workload pointing to a pod on another
   network we will have a "proxy" representation that just has network
   gateway).

Both of those can be in hanled by the controlplane (e.g., controlplane
can generate split-horizon workloads and when configuration is
non-uniform, just filter out remote configs for remote networks), though
we don't yet have a complete implementation.

Signed-off-by: Mikhail Krinkin <mkrinkin@microsoft.com>

* Return an error instead of panicking

Signed-off-by: Mikhail Krinkin <mkrinkin@microsoft.com>

* Update comments in src/proxy/outbound.rs

Co-authored-by: Ian Rudie <ilrudie@gmail.com>

* Update comments in src/proxy/outbound.rs

Signed-off-by: Mikhail Krinkin <mkrinkin@microsoft.com>

* Add a debug assert to provide a bit more context to the error in tests

Signed-off-by: Mikhail Krinkin <mkrinkin@microsoft.com>

* Fix formatting

Signed-off-by: Mikhail Krinkin <mkrinkin@microsoft.com>

* Added a few debug logs to be able to trace when a workload on a remote network is picked

Signed-off-by: Mikhail Krinkin <mkrinkin@microsoft.com>

---------

Signed-off-by: Mikhail Krinkin <mkrinkin@microsoft.com>
Co-authored-by: Ian Rudie <ilrudie@gmail.com>
2025-07-07 12:29:29 -04:00
Istio Automation f030073f2f
Automator: update common-files@master in istio/ztunnel@master (#1583) 2025-06-30 16:52:23 -04:00
Ian Rudie 3233bb1017
Improved Service Resolution (#1562)
* initial idea for improved resolution

Signed-off-by: ilrudie <ian.rudie@solo.io>

* handle preferred service namespace; unit testing

Signed-off-by: Ian Rudie <ian.rudie@solo.io>

---------

Signed-off-by: ilrudie <ian.rudie@solo.io>
Signed-off-by: Ian Rudie <ian.rudie@solo.io>
2025-06-30 11:23:22 -04:00
Istio Automation 7df8cf5d08
Automator: update common-files@master in istio/ztunnel@master (#1582) 2025-06-26 08:18:20 -04:00
Jacek Ewertowski 7cddb868e9
tls: add PQC compliance policy (#1561)
* tls: add PQC compliance policy

Signed-off-by: Jacek Ewertowski <jacek.ewertowski1@gmail.com>

* Add global lazy variable PQC_ENABLED

Signed-off-by: Jacek Ewertowski <jacek.ewertowski1@gmail.com>

* Add unused_imports and dead_code to PQC_ENABLED declaration

Signed-off-by: Jacek Ewertowski <jacek.ewertowski1@gmail.com>

---------

Signed-off-by: Jacek Ewertowski <jacek.ewertowski1@gmail.com>
2025-06-23 10:45:24 -04:00
Istio Automation ac477c15a8
Automator: update common-files@master in istio/ztunnel@master (#1578) 2025-06-19 09:56:10 -04:00
Istio Automation 5d0352588c
Automator: update common-files@master in istio/ztunnel@master (#1577) 2025-06-18 14:30:10 -04:00
Ian Rudie b86fd9989b
remove invalid test cases from parsing of ZTUNNEL_WORKER_THREADS (#1576)
Signed-off-by: Ian Rudie <ian.rudie@solo.io>
2025-06-18 13:15:00 -04:00
Istio Automation facd9a28a0
Automator: update common-files@master in istio/ztunnel@master (#1572) 2025-06-08 22:00:13 -04:00
Istio Automation 224b2c34ac
Automator: update common-files@master in istio/ztunnel@master (#1571) 2025-06-06 10:26:53 -04:00
Steven Landow c52e0bbdbf
don't send to empty address (#1570)
* don't send to empty address

* add test
2025-06-05 16:38:51 -04:00
John Howard 442923910b
Allow dynamic configuration of thread count (#1566)
* Allow dynamic configuration of thread count

* fix flakes
2025-06-04 14:31:52 -04:00
Istio Automation c616a29092
Automator: update common-files@master in istio/ztunnel@master (#1568) 2025-06-04 12:44:51 -04:00
Istio Automation d6d3b606ed
Automator: update common-files@master in istio/ztunnel@master (#1563) 2025-05-27 12:51:34 -04:00
Ian Rudie 8d9a56a416
update how io errors are being generated to fix clippy issues (#1564)
Signed-off-by: ilrudie <ian.rudie@solo.io>
2025-05-27 12:43:33 -04:00
zirain 615277a05a
remove git tag (#1559) 2025-05-21 23:47:28 -04:00
Harsh Pratap Singh 3d1223af09
Support TLS for metrics endpoint (#1507)
* add support for secure metrics endpoint

* nit

* no need to rewrite to localhost

* Implement tests for ztunnel identity and metrics routing

* nits

* Refactor to use environment variables for trust domain and service account
environments.

* refactoring

* nit

* Refactor Config and Identity Handling

* made identity and workload info optional and added unit test

* dont need to use env variables in test now

* Added logic to direct traffic to the ztunnel metrics endpoint on port 15020 via HBONE, ensuring proper request forwarding.

* Enhance Proxy to Support ztunnel Inbound Traffic

* Implement ztunnel inbound listener creation in shared proxy mode

* Refactor ztunnel inbound listener creation in shared proxy mode by removing unnecessary checks

* Refactor ztunnel inbound listener creation by encapsulating logic in ProxyFactory. =

* Enhance ProxyInputs to include disable_inbound_freebind flag for controlling source IP preservation.

* Add tests for ztunnel self-scrape and metrics scrape via HBONE

* refactored e2e test for ztunnel identity and inbound

* fmt nit

* clippy nits

* nits

* nit

* fmt nits

* Refactor e2e tests

* Refactor client in e2e test connection logic to perform a standard HTTP GET request to the metrics endpoint, ensuring proper verification of the response.

* fmt nit

* match nit

* remove unnecessary test and error

* Refactor ztunnel listener creation and configuration for shared proxy mode

* Refactor trust domain handling and improve string formatting in configuration
nagement.

* nits

* Revert timeout duration for TCP connection and

* fmt nit

* nits
2025-05-19 16:05:24 -04:00
Istio Automation 79dfd10249
Automator: update common-files@master in istio/ztunnel@master (#1558) 2025-05-15 14:26:14 -04:00
Istio Automation 9f6ae51005
Automator: update common-files@master in istio/ztunnel@master (#1557) 2025-05-13 18:03:14 -04:00
Istio Automation 46acf76463
Automator: update common-files@master in istio/ztunnel@master (#1554) 2025-05-09 10:01:46 -04:00
Istio Automation 58cf2a0f94
Automator: update common-files@master in istio/ztunnel@master (#1553) 2025-05-09 06:16:46 -04:00
Steven Jin 9c01d1276d
show crypto provider in ztunnel version (#1545)
* Version shows crypto provider

* All tls providers

* Move crypto consts
2025-05-08 16:42:45 -04:00
Yuval Kohavi d9ea32ce21
Make csr test stricter and more correct (previously #1432) (#1550)
* Make csr test stricter and more correct

Part of https://github.com/istio/ztunnel/issues/1431

Fails without https://github.com/rustls/rcgen/pull/311

* update rcgen

Signed-off-by: Yuval Kohavi <yuval.kohavi@gmail.com>

* fix merge issue

* format fix

Signed-off-by: Yuval Kohavi <yuval.kohavi@gmail.com>

---------

Signed-off-by: Yuval Kohavi <yuval.kohavi@gmail.com>
Co-authored-by: John Howard <john.howard@solo.io>
2025-05-07 16:01:44 -04:00
Istio Automation c96dd032da
Automator: update common-files@master in istio/ztunnel@master (#1549) 2025-05-07 14:25:48 -04:00
Istio Automation 903cf079de
Automator: update common-files@master in istio/ztunnel@master (#1544) 2025-04-25 04:03:03 -04:00
Istio Automation ad8bea43ef
Automator: update common-files@master in istio/ztunnel@master (#1543) 2025-04-24 16:28:02 -04:00
John Howard 6eaa32e8ac
http2: tune connection window size and add config (#1539)
Fixes https://github.com/istio/ztunnel/issues/1538

See comment for motivation as to why this change is needed.
2025-04-24 09:54:55 -04:00
zirain 3470f4bba2
fix istio_build metric (#1532)
* fix istio_build metric

* fmt
2025-04-18 09:54:24 -04:00
John Howard 93a0973175
Include error info in DNS access logs (#1529) 2025-04-17 15:32:23 -04:00
29 changed files with 1147 additions and 294 deletions

View File

@ -1,6 +1,6 @@
{
"name": "istio build-tools",
"image": "gcr.io/istio-testing/build-tools:master-fcd42145fc132acd1e8f607e9e7aca15058e9fb9",
"image": "gcr.io/istio-testing/build-tools:master-8e6480403f5cf4c9a4cd9d65174d01850e632e1a",
"privileged": true,
"remoteEnv": {
"USE_GKE_GCLOUD_AUTH_PLUGIN": "True",

94
Cargo.lock generated
View File

@ -112,29 +112,13 @@ version = "0.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
[[package]]
name = "asn1-rs"
version = "0.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5493c3bedbacf7fd7382c6346bbd66687d12bbaad3a89a2d2c303ee6cf20b048"
dependencies = [
"asn1-rs-derive 0.5.1",
"asn1-rs-impl",
"displaydoc",
"nom",
"num-traits",
"rusticata-macros",
"thiserror 1.0.69",
"time",
]
[[package]]
name = "asn1-rs"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "56624a96882bb8c26d61312ae18cb45868e5a9992ea73c58e45c3101e56a1e60"
dependencies = [
"asn1-rs-derive 0.6.0",
"asn1-rs-derive",
"asn1-rs-impl",
"displaydoc",
"nom",
@ -144,18 +128,6 @@ dependencies = [
"time",
]
[[package]]
name = "asn1-rs-derive"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "965c2d33e53cb6b267e148a4cb0760bc01f4904c1cd4bb4002a085bb016d1490"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.100",
"synstructure",
]
[[package]]
name = "asn1-rs-derive"
version = "0.6.0"
@ -761,27 +733,13 @@ dependencies = [
"const-oid",
]
[[package]]
name = "der-parser"
version = "9.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5cd0a5c643689626bec213c4d8bd4d96acc8ffdb4ad4bb6bc16abf27d5f4b553"
dependencies = [
"asn1-rs 0.6.2",
"displaydoc",
"nom",
"num-bigint",
"num-traits",
"rusticata-macros",
]
[[package]]
name = "der-parser"
version = "10.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6"
dependencies = [
"asn1-rs 0.7.1",
"asn1-rs",
"displaydoc",
"nom",
"num-bigint",
@ -1844,7 +1802,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34"
dependencies = [
"cfg-if",
"windows-targets 0.52.6",
"windows-targets 0.48.5",
]
[[package]]
@ -2226,22 +2184,13 @@ dependencies = [
"memchr",
]
[[package]]
name = "oid-registry"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a8d8034d9489cdaf79228eb9f6a3b8d7bb32ba00d6645ebd48eef4077ceb5bd9"
dependencies = [
"asn1-rs 0.6.2",
]
[[package]]
name = "oid-registry"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7"
dependencies = [
"asn1-rs 0.7.1",
"asn1-rs",
]
[[package]]
@ -2815,16 +2764,16 @@ dependencies = [
[[package]]
name = "rcgen"
version = "0.13.2"
version = "0.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2"
checksum = "887a643fa081058097896d87764863994f6c32a1716e76adc479bd283974a825"
dependencies = [
"aws-lc-rs",
"pem",
"ring",
"rustls-pki-types",
"time",
"x509-parser 0.16.0",
"x509-parser",
"yasna",
]
@ -4305,36 +4254,18 @@ version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e9df38ee2d2c3c5948ea468a8406ff0db0b29ae1ffde1bcf20ef305bcc95c51"
[[package]]
name = "x509-parser"
version = "0.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fcbc162f30700d6f3f82a24bf7cc62ffe7caea42c0b2cba8bf7f3ae50cf51f69"
dependencies = [
"asn1-rs 0.6.2",
"data-encoding",
"der-parser 9.0.0",
"lazy_static",
"nom",
"oid-registry 0.7.1",
"ring",
"rusticata-macros",
"thiserror 1.0.69",
"time",
]
[[package]]
name = "x509-parser"
version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4569f339c0c402346d4a75a9e39cf8dad310e287eef1ff56d4c68e5067f53460"
dependencies = [
"asn1-rs 0.7.1",
"asn1-rs",
"data-encoding",
"der-parser 10.0.0",
"der-parser",
"lazy_static",
"nom",
"oid-registry 0.8.1",
"oid-registry",
"ring",
"rusticata-macros",
"thiserror 2.0.12",
@ -4510,7 +4441,8 @@ dependencies = [
"matches",
"netns-rs",
"nix 0.29.0",
"oid-registry 0.8.1",
"num_cpus",
"oid-registry",
"once_cell",
"openssl",
"pin-project-lite",
@ -4553,6 +4485,6 @@ dependencies = [
"tracing-log",
"tracing-subscriber",
"url",
"x509-parser 0.17.0",
"x509-parser",
"ztunnel",
]

View File

@ -71,8 +71,9 @@ itertools = "0.14"
keyed_priority_queue = "0.4"
libc = "0.2"
log = "0.4"
nix = { version = "0.29", features = ["socket", "sched", "uio", "fs", "ioctl", "user", "net", "mount"] }
nix = { version = "0.29", features = ["socket", "sched", "uio", "fs", "ioctl", "user", "net", "mount", "resource" ] }
once_cell = "1.21"
num_cpus = "1.16"
ppp = "2.3"
prometheus-client = { version = "0.23" }
prometheus-parse = "0.2"

View File

@ -1 +1 @@
a1d5c4198ab79a14c09c034f2d95245efa3e2bcb
d46067e1a8ba3db4abe2635af5807f00ba1981e6

View File

@ -32,7 +32,7 @@ set -x
####################################################################
# DEFAULT_KIND_IMAGE is used to set the Kubernetes version for KinD unless overridden in params to setup_kind_cluster(s)
DEFAULT_KIND_IMAGE="gcr.io/istio-testing/kind-node:v1.32.0"
DEFAULT_KIND_IMAGE="gcr.io/istio-testing/kind-node:v1.33.1"
# the default kind cluster should be ipv4 if not otherwise specified
KIND_IP_FAMILY="${KIND_IP_FAMILY:-ipv4}"

View File

@ -75,7 +75,7 @@ fi
TOOLS_REGISTRY_PROVIDER=${TOOLS_REGISTRY_PROVIDER:-gcr.io}
PROJECT_ID=${PROJECT_ID:-istio-testing}
if [[ "${IMAGE_VERSION:-}" == "" ]]; then
IMAGE_VERSION=master-fcd42145fc132acd1e8f607e9e7aca15058e9fb9
IMAGE_VERSION=master-8e6480403f5cf4c9a4cd9d65174d01850e632e1a
fi
if [[ "${IMAGE_NAME:-}" == "" ]]; then
IMAGE_NAME=build-tools

1
fuzz/Cargo.lock generated
View File

@ -3843,6 +3843,7 @@ dependencies = [
"log",
"netns-rs",
"nix 0.29.0",
"num_cpus",
"once_cell",
"pin-project-lite",
"pingora-pool",

View File

@ -136,6 +136,25 @@ pub async fn build_with_cert(
if config.proxy_mode == config::ProxyMode::Shared {
tracing::info!("shared proxy mode - in-pod mode enabled");
// Create ztunnel inbound listener only if its specific identity and workload info are configured.
if let Some(inbound) = proxy_gen.create_ztunnel_self_proxy_listener().await? {
// Run the inbound listener in the data plane worker pool
let mut xds_rx_for_inbound = xds_rx.clone();
data_plane_pool.send(DataPlaneTask {
block_shutdown: true,
fut: Box::pin(async move {
tracing::info!("Starting ztunnel inbound listener task");
let _ = xds_rx_for_inbound.changed().await;
tokio::task::spawn(async move {
inbound.run().in_current_span().await;
})
.await?;
Ok(())
}),
})?;
}
let run_future = init_inpod_proxy_mgr(
&mut registry,
&mut admin_server,
@ -247,7 +266,8 @@ fn new_data_plane_pool(num_worker_threads: usize) -> mpsc::Sender<DataPlaneTask>
.thread_name_fn(|| {
static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
format!("ztunnel-proxy-{id}")
// Thread name can only be 16 chars so keep it short
format!("ztunnel-{id}")
})
.enable_all()
.build()

View File

@ -54,10 +54,12 @@ const LOCAL_XDS_PATH: &str = "LOCAL_XDS_PATH";
const LOCAL_XDS: &str = "LOCAL_XDS";
const XDS_ON_DEMAND: &str = "XDS_ON_DEMAND";
const XDS_ADDRESS: &str = "XDS_ADDRESS";
const PREFERED_SERVICE_NAMESPACE: &str = "PREFERED_SERVICE_NAMESPACE";
const CA_ADDRESS: &str = "CA_ADDRESS";
const SECRET_TTL: &str = "SECRET_TTL";
const FAKE_CA: &str = "FAKE_CA";
const ZTUNNEL_WORKER_THREADS: &str = "ZTUNNEL_WORKER_THREADS";
const ZTUNNEL_CPU_LIMIT: &str = "ZTUNNEL_CPU_LIMIT";
const POOL_MAX_STREAMS_PER_CONNECTION: &str = "POOL_MAX_STREAMS_PER_CONNECTION";
const POOL_UNUSED_RELEASE_TIMEOUT: &str = "POOL_UNUSED_RELEASE_TIMEOUT";
// CONNECTION_TERMINATION_DEADLINE configures an explicit deadline
@ -70,6 +72,10 @@ const ENABLE_ORIG_SRC: &str = "ENABLE_ORIG_SRC";
const PROXY_CONFIG: &str = "PROXY_CONFIG";
const IPV6_ENABLED: &str = "IPV6_ENABLED";
const HTTP2_STREAM_WINDOW_SIZE: &str = "HTTP2_STREAM_WINDOW_SIZE";
const HTTP2_CONNECTION_WINDOW_SIZE: &str = "HTTP2_CONNECTION_WINDOW_SIZE";
const HTTP2_FRAME_SIZE: &str = "HTTP2_FRAME_SIZE";
const UNSTABLE_ENABLE_SOCKS5: &str = "UNSTABLE_ENABLE_SOCKS5";
const DEFAULT_WORKER_THREADS: u16 = 2;
@ -237,6 +243,12 @@ pub struct Config {
// Allow custom alternative XDS hostname verification
pub alt_xds_hostname: Option<String>,
/// Prefered service namespace to use for service resolution.
/// If unset, local namespaces is preferred and other namespaces have equal priority.
/// If set, the local namespace is preferred, then the defined prefered_service_namespace
/// and finally other namespaces at an equal priority.
pub prefered_service_namespace: Option<String>,
/// TTL for CSR requests
pub secret_ttl: Duration,
/// YAML config for local XDS workloads
@ -293,6 +305,10 @@ pub struct Config {
// If true, when AppTunnel is set for
pub localhost_app_tunnel: bool,
pub ztunnel_identity: Option<identity::Identity>,
pub ztunnel_workload: Option<state::WorkloadInfo>,
}
#[derive(serde::Serialize, Clone, Copy, Debug)]
@ -399,6 +415,60 @@ fn parse_headers(prefix: &str) -> Result<MetadataVector, Error> {
Ok(metadata)
}
fn get_cpu_count() -> Result<usize, Error> {
// Allow overriding the count with an env var. This can be used to pass the CPU limit on Kubernetes
// from the downward API.
// Note the downward API will return the total thread count ("logical cores") if no limit is set,
// so it is really the same as num_cpus.
// We allow num_cpus for cases its not set (not on Kubernetes, etc).
match parse::<usize>(ZTUNNEL_CPU_LIMIT)? {
Some(limit) => Ok(limit),
// This is *logical cores*
None => Ok(num_cpus::get()),
}
}
/// Parse worker threads configuration, supporting both fixed numbers and percentages
fn parse_worker_threads(default: usize) -> Result<usize, Error> {
match parse::<String>(ZTUNNEL_WORKER_THREADS)? {
Some(value) => {
if let Some(percent_str) = value.strip_suffix('%') {
// Parse as percentage
let percent: f64 = percent_str.parse().map_err(|e| {
Error::EnvVar(
ZTUNNEL_WORKER_THREADS.to_string(),
value.clone(),
format!("invalid percentage: {}", e),
)
})?;
if percent <= 0.0 || percent > 100.0 {
return Err(Error::EnvVar(
ZTUNNEL_WORKER_THREADS.to_string(),
value,
"percentage must be between 0 and 100".to_string(),
));
}
let cpu_count = get_cpu_count()?;
// Round up, minimum of 1
let threads = ((cpu_count as f64 * percent / 100.0).ceil() as usize).max(1);
Ok(threads)
} else {
// Parse as fixed number
value.parse::<usize>().map_err(|e| {
Error::EnvVar(
ZTUNNEL_WORKER_THREADS.to_string(),
value,
format!("invalid number: {}", e),
)
})
}
}
None => Ok(default),
}
}
pub fn parse_config() -> Result<Config, Error> {
let pc = parse_proxy_config()?;
construct_config(pc)
@ -438,6 +508,14 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
.or_else(|| Some(default_istiod_address.clone())),
))?;
let prefered_service_namespace = match parse::<String>(PREFERED_SERVICE_NAMESPACE) {
Ok(ns) => ns,
Err(e) => {
warn!(err=?e, "failed to parse {PREFERED_SERVICE_NAMESPACE}, continuing with default behavior");
None
}
};
let istio_meta_cluster_id = ISTIO_META_PREFIX.to_owned() + CLUSTER_ID;
let cluster_id: String = match parse::<String>(&istio_meta_cluster_id)? {
Some(id) => id,
@ -600,6 +678,29 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
let socket_config_defaults = SocketConfig::default();
// Read ztunnel identity and workload info from Downward API if available
let (ztunnel_identity, ztunnel_workload) = match (
parse::<String>("POD_NAMESPACE")?,
parse::<String>("SERVICE_ACCOUNT")?,
parse::<String>("POD_NAME")?,
) {
(Some(namespace), Some(service_account), Some(pod_name)) => {
let trust_domain = std::env::var("TRUST_DOMAIN")
.unwrap_or_else(|_| crate::identity::manager::DEFAULT_TRUST_DOMAIN.to_string());
let identity = identity::Identity::from_parts(
trust_domain.into(),
namespace.clone().into(),
service_account.clone().into(),
);
let workload = state::WorkloadInfo::new(pod_name, namespace, service_account);
(Some(identity), Some(workload))
}
_ => (None, None),
};
validate_config(Config {
proxy: parse_default(ENABLE_PROXY, true)?,
// Enable by default; running the server is not an issue, clients still need to opt-in to sending their
@ -619,9 +720,15 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
DEFAULT_POOL_UNUSED_RELEASE_TIMEOUT,
)?,
window_size: 4 * 1024 * 1024,
connection_window_size: 4 * 1024 * 1024,
frame_size: 1024 * 1024,
// window size: per-stream limit
window_size: parse_default(HTTP2_STREAM_WINDOW_SIZE, 4 * 1024 * 1024)?,
// connection window size: per connection.
// Setting this to the same value as window_size can introduce deadlocks in some applications
// where clients do not read data on streamA until they receive data on streamB.
// If streamA consumes the entire connection window, we enter a deadlock.
// A 4x limit should be appropriate without introducing too much potential buffering.
connection_window_size: parse_default(HTTP2_CONNECTION_WINDOW_SIZE, 16 * 1024 * 1024)?,
frame_size: parse_default(HTTP2_FRAME_SIZE, 1024 * 1024)?,
self_termination_deadline: match parse_duration(CONNECTION_TERMINATION_DEADLINE)? {
Some(period) => period,
@ -675,6 +782,7 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
xds_address,
xds_root_cert,
prefered_service_namespace,
ca_address,
ca_root_cert,
alt_xds_hostname: parse(ALT_XDS_HOSTNAME)?,
@ -688,8 +796,7 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
fake_ca,
auth,
num_worker_threads: parse_default(
ZTUNNEL_WORKER_THREADS,
num_worker_threads: parse_worker_threads(
pc.concurrency.unwrap_or(DEFAULT_WORKER_THREADS).into(),
)?,
@ -753,6 +860,8 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
ca_headers: parse_headers(ISTIO_CA_HEADER_PREFIX)?,
localhost_app_tunnel: parse_default(LOCALHOST_APP_TUNNEL, true)?,
ztunnel_identity,
ztunnel_workload,
})
}
@ -1055,4 +1164,45 @@ pub mod tests {
assert!(metadata.vec.contains(&(key, value)));
}
}
#[test]
fn test_parse_worker_threads() {
unsafe {
// Test fixed number
env::set_var(ZTUNNEL_WORKER_THREADS, "4");
assert_eq!(parse_worker_threads(2).unwrap(), 4);
// Test percentage with CPU limit
env::set_var(ZTUNNEL_CPU_LIMIT, "8");
env::set_var(ZTUNNEL_WORKER_THREADS, "50%");
assert_eq!(parse_worker_threads(2).unwrap(), 4); // 50% of 8 CPUs = 4 threads
// Test percentage with CPU limit
env::set_var(ZTUNNEL_CPU_LIMIT, "16");
env::set_var(ZTUNNEL_WORKER_THREADS, "30%");
assert_eq!(parse_worker_threads(2).unwrap(), 5); // Round up to 5
// Test low percentage that rounds up to 1
env::set_var(ZTUNNEL_CPU_LIMIT, "4");
env::set_var(ZTUNNEL_WORKER_THREADS, "10%");
assert_eq!(parse_worker_threads(2).unwrap(), 1); // 10% of 4 CPUs = 0.4, rounds up to 1
// Test default when no env var is set
env::remove_var(ZTUNNEL_WORKER_THREADS);
assert_eq!(parse_worker_threads(2).unwrap(), 2);
// Test without CPU limit (should use system CPU count)
env::remove_var(ZTUNNEL_CPU_LIMIT);
let system_cpus = num_cpus::get();
assert_eq!(get_cpu_count().unwrap(), system_cpus);
// Test with CPU limit
env::set_var(ZTUNNEL_CPU_LIMIT, "12");
assert_eq!(get_cpu_count().unwrap(), 12);
// Clean up
env::remove_var(ZTUNNEL_WORKER_THREADS);
env::remove_var(ZTUNNEL_CPU_LIMIT);
}
}
}

View File

@ -47,7 +47,7 @@ use crate::drain::{DrainMode, DrainWatcher};
use crate::metrics::{DeferRecorder, IncrementRecorder, Recorder};
use crate::proxy::Error;
use crate::state::DemandProxyState;
use crate::state::service::IpFamily;
use crate::state::service::{IpFamily, Service};
use crate::state::workload::Workload;
use crate::state::workload::address::Address;
use crate::{config, dns};
@ -85,6 +85,7 @@ impl Server {
drain: DrainWatcher,
socket_factory: &(dyn SocketFactory + Send + Sync),
local_workload_information: Arc<LocalWorkloadFetcher>,
prefered_service_namespace: Option<String>,
) -> Result<Self, Error> {
// if the address we got from config is supposed to be v6-enabled,
// actually check if the local pod context our socketfactory operates in supports V6.
@ -102,6 +103,7 @@ impl Server {
forwarder,
metrics,
local_workload_information,
prefered_service_namespace,
);
let store = Arc::new(store);
let handler = dns::handler::Handler::new(store.clone());
@ -191,6 +193,7 @@ struct Store {
svc_domain: Name,
metrics: Arc<Metrics>,
local_workload: Arc<LocalWorkloadFetcher>,
prefered_service_namespace: Option<String>,
}
impl Store {
@ -200,6 +203,7 @@ impl Store {
forwarder: Arc<dyn Forwarder>,
metrics: Arc<Metrics>,
local_workload_information: Arc<LocalWorkloadFetcher>,
prefered_service_namespace: Option<String>,
) -> Self {
let domain = as_name(domain);
let svc_domain = append_name(as_name("svc"), &domain);
@ -211,6 +215,7 @@ impl Store {
svc_domain,
metrics,
local_workload: local_workload_information,
prefered_service_namespace,
}
}
@ -359,7 +364,7 @@ impl Store {
let search_name_str = search_name.to_string().into();
search_name.set_fqdn(true);
let service = state
let services: Vec<Arc<Service>> = state
.services
.get_by_host(&search_name_str)
.iter()
@ -382,13 +387,30 @@ impl Store {
})
// Get the service matching the client namespace. If no match exists, just
// return the first service.
.find_or_first(|service| service.namespace == client.namespace)
.cloned();
// .find_or_first(|service| service.namespace == client.namespace)
.cloned()
.collect();
// TODO: ideally we'd sort these by creation time so that the oldest would be used if there are no namespace matches
// presently service doesn't have creation time in WDS, but we could add it
// TODO: if the local namespace doesn't define a service, kube service should be prioritized over se
let service = match services
.iter()
.find(|service| service.namespace == client.namespace)
{
Some(service) => Some(service),
None => match self.prefered_service_namespace.as_ref() {
Some(prefered_namespace) => services.iter().find_or_first(|service| {
service.namespace == prefered_namespace.as_str()
}),
None => services.first(),
},
};
// First, lookup the host as a service.
if let Some(service) = service {
return Some(ServerMatch {
server: Address::Service(service),
server: Address::Service(service.clone()),
name: search_name,
alias,
});
@ -564,7 +586,12 @@ impl Resolver for Store {
}
Err(e) => {
// Forwarding failed. Just return the error.
access_log(request, Some(&client), "forwarding failed", 0);
access_log(
request,
Some(&client),
&format!("forwarding failed ({e})"),
0,
);
return Err(e);
}
}
@ -589,7 +616,12 @@ impl Resolver for Store {
}
Err(e) => {
// Forwarding failed. Just return the error.
access_log(request, Some(&client), "forwarding failed", 0);
access_log(
request,
Some(&client),
&format!("forwarding failed ({e})"),
0,
);
return Err(e);
}
}
@ -946,6 +978,7 @@ mod tests {
const NS1: &str = "ns1";
const NS2: &str = "ns2";
const PREFERRED: &str = "preferred-ns";
const NW1: Strng = strng::literal!("nw1");
const NW2: Strng = strng::literal!("nw2");
@ -1053,6 +1086,7 @@ mod tests {
forwarder,
metrics: test_metrics(),
local_workload,
prefered_service_namespace: None,
};
let namespaced_domain = n(format!("{}.svc.cluster.local", c.client_namespace));
@ -1368,6 +1402,18 @@ mod tests {
expect_code: ResponseCode::NXDomain,
..Default::default()
},
Case {
name: "success: preferred namespace is chosen if local namespace is not defined",
host: "preferred.io.",
expect_records: vec![a(n("preferred.io."), ipv4("10.10.10.211"))],
..Default::default()
},
Case {
name: "success: external service resolves to local namespace's address",
host: "everywhere.io.",
expect_records: vec![a(n("everywhere.io."), ipv4("10.10.10.112"))],
..Default::default()
},
];
// Create and start the proxy.
@ -1385,6 +1431,7 @@ mod tests {
drain,
&factory,
local_workload,
Some(PREFERRED.to_string()),
)
.await
.unwrap();
@ -1471,6 +1518,7 @@ mod tests {
drain,
&factory,
local_workload,
None,
)
.await
.unwrap();
@ -1520,6 +1568,7 @@ mod tests {
}),
state.clone(),
),
prefered_service_namespace: None,
};
let ip4n6_client_ip = ip("::ffff:202:202");
@ -1553,6 +1602,7 @@ mod tests {
drain,
&factory,
local_workload,
None,
)
.await
.unwrap();
@ -1669,6 +1719,16 @@ mod tests {
xds_external_service("www.google.com", &[na(NW1, "1.1.1.1")]),
xds_service("productpage", NS1, &[na(NW1, "9.9.9.9")]),
xds_service("example", NS2, &[na(NW1, "10.10.10.10")]),
// Service with the same name in another namespace
// This should not be used if the preferred service namespace is set
xds_namespaced_external_service("everywhere.io", NS2, &[na(NW1, "10.10.10.110")]),
xds_namespaced_external_service("preferred.io", NS2, &[na(NW1, "10.10.10.210")]),
// Preferred service namespace
xds_namespaced_external_service("everywhere.io", PREFERRED, &[na(NW1, "10.10.10.111")]),
xds_namespaced_external_service("preferred.io", PREFERRED, &[na(NW1, "10.10.10.211")]),
// Service with the same name in the same namespace
// Client in NS1 should use this service
xds_namespaced_external_service("everywhere.io", NS1, &[na(NW1, "10.10.10.112")]),
with_fqdn(
"details.ns2.svc.cluster.remote",
xds_service(
@ -1819,9 +1879,17 @@ mod tests {
}
fn xds_external_service<S: AsRef<str>>(hostname: S, addrs: &[NetworkAddress]) -> XdsService {
xds_namespaced_external_service(hostname, NS1, addrs)
}
fn xds_namespaced_external_service<S1: AsRef<str>, S2: AsRef<str>>(
hostname: S1,
ns: S2,
vips: &[NetworkAddress],
) -> XdsService {
with_fqdn(
hostname.as_ref(),
xds_service(hostname.as_ref(), NS1, addrs),
xds_service(hostname.as_ref(), ns.as_ref(), vips),
)
}

View File

@ -58,10 +58,7 @@ async fn load_token(path: &PathBuf) -> io::Result<Vec<u8>> {
let t = tokio::fs::read(path).await?;
if t.is_empty() {
return Err(io::Error::new(
io::ErrorKind::Other,
"token file exists, but was empty",
));
return Err(io::Error::other("token file exists, but was empty"));
}
Ok(t)
}

View File

@ -38,6 +38,9 @@ use keyed_priority_queue::KeyedPriorityQueue;
const CERT_REFRESH_FAILURE_RETRY_DELAY_MAX_INTERVAL: Duration = Duration::from_secs(150);
/// Default trust domain to use if not otherwise specified.
pub const DEFAULT_TRUST_DOMAIN: &str = "cluster.local";
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Hash)]
pub enum Identity {
Spiffe {
@ -130,11 +133,10 @@ impl Identity {
#[cfg(any(test, feature = "testing"))]
impl Default for Identity {
fn default() -> Self {
const TRUST_DOMAIN: &str = "cluster.local";
const SERVICE_ACCOUNT: &str = "ztunnel";
const NAMESPACE: &str = "istio-system";
Identity::Spiffe {
trust_domain: TRUST_DOMAIN.into(),
trust_domain: DEFAULT_TRUST_DOMAIN.into(),
namespace: NAMESPACE.into(),
service_account: SERVICE_ACCOUNT.into(),
}

View File

@ -12,6 +12,9 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use once_cell::sync::Lazy;
use std::env;
pub mod admin;
pub mod app;
pub mod assertions;
@ -42,3 +45,7 @@ pub mod xds;
#[cfg(any(test, feature = "testing"))]
pub mod test_helpers;
#[allow(dead_code)]
static PQC_ENABLED: Lazy<bool> =
Lazy::new(|| env::var("COMPLIANCE_POLICY").unwrap_or_default() == "pqc");

View File

@ -14,8 +14,9 @@
extern crate core;
use nix::sys::resource::{Resource, getrlimit, setrlimit};
use std::sync::Arc;
use tracing::info;
use tracing::{info, warn};
use ztunnel::*;
#[cfg(feature = "jemalloc")]
@ -28,6 +29,26 @@ static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
#[unsafe(export_name = "malloc_conf")]
pub static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0";
// We use this on Unix systems to increase the number of open file descriptors
// if possible. This is useful for high-load scenarios where the default limit
// is too low, which can lead to droopped connections and other issues:
// see: https://github.com/istio/ztunnel/issues/1585
fn increase_open_files_limit() {
#[cfg(unix)]
if let Ok((soft_limit, hard_limit)) = getrlimit(Resource::RLIMIT_NOFILE) {
if let Err(e) = setrlimit(Resource::RLIMIT_NOFILE, hard_limit, hard_limit) {
warn!("failed to set file descriptor limits: {e}");
} else {
info!(
"set file descriptor limits from {} to {}",
soft_limit, hard_limit
);
}
} else {
warn!("failed to get file descriptor limits");
}
}
fn main() -> anyhow::Result<()> {
let _log_flush = telemetry::setup_logging();
@ -74,6 +95,7 @@ fn version() -> anyhow::Result<()> {
async fn proxy(cfg: Arc<config::Config>) -> anyhow::Result<()> {
info!("version: {}", version::BuildInfo::new());
increase_open_files_limit();
info!("running with config: {}", serde_yaml::to_string(&cfg)?);
app::build(cfg).await?.wait_termination().await
}

View File

@ -48,8 +48,9 @@ use crate::state::{DemandProxyState, WorkloadInfo};
use crate::{config, identity, socket, tls};
pub mod connection_manager;
pub mod inbound;
mod h2;
mod inbound;
mod inbound_passthrough;
#[allow(non_camel_case_types)]
pub mod metrics;
@ -259,6 +260,8 @@ pub(super) struct ProxyInputs {
socket_factory: Arc<dyn SocketFactory + Send + Sync>,
local_workload_information: Arc<LocalWorkloadInformation>,
resolver: Option<Arc<dyn Resolver + Send + Sync>>,
// If true, inbound connections created with these inputs will not attempt to preserve the original source IP.
pub disable_inbound_freebind: bool,
}
#[allow(clippy::too_many_arguments)]
@ -271,6 +274,7 @@ impl ProxyInputs {
socket_factory: Arc<dyn SocketFactory + Send + Sync>,
resolver: Option<Arc<dyn Resolver + Send + Sync>>,
local_workload_information: Arc<LocalWorkloadInformation>,
disable_inbound_freebind: bool,
) -> Arc<Self> {
Arc::new(Self {
cfg,
@ -280,6 +284,7 @@ impl ProxyInputs {
socket_factory,
local_workload_information,
resolver,
disable_inbound_freebind,
})
}
}
@ -479,6 +484,9 @@ pub enum Error {
#[error("requested service {0} found, but has no IP addresses")]
NoIPForService(String),
#[error("no service for target address: {0}")]
NoService(SocketAddr),
#[error(
"ip addresses were resolved for workload {0}, but valid dns response had no A/AAAA records"
)]

View File

@ -13,7 +13,7 @@
// limitations under the License.
use crate::copy;
use bytes::{BufMut, Bytes};
use bytes::Bytes;
use futures_core::ready;
use h2::Reason;
use std::io::Error;
@ -85,7 +85,10 @@ pub struct H2StreamWriteHalf {
_dropped: Option<DropCounter>,
}
pub struct TokioH2Stream(H2Stream);
pub struct TokioH2Stream {
stream: H2Stream,
buf: Bytes,
}
struct DropCounter {
// Whether the other end of this shared counter has already dropped.
@ -144,7 +147,10 @@ impl Drop for DropCounter {
// then the specific implementation will conflict with the generic one.
impl TokioH2Stream {
pub fn new(stream: H2Stream) -> Self {
Self(stream)
Self {
stream,
buf: Bytes::new(),
}
}
}
@ -154,24 +160,21 @@ impl tokio::io::AsyncRead for TokioH2Stream {
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
let pinned = std::pin::Pin::new(&mut self.0.read);
copy::ResizeBufRead::poll_bytes(pinned, cx).map(|r| match r {
Ok(bytes) => {
if buf.remaining() < bytes.len() {
Err(Error::new(
std::io::ErrorKind::Other,
format!(
"kould overflow buffer of with {} remaining",
buf.remaining()
),
))
} else {
buf.put(bytes);
Ok(())
}
}
Err(e) => Err(e),
})
// Just return the bytes we have left over and don't poll the stream because
// its unclear what to do if there are bytes left over from the previous read, and when we
// poll, we get an error.
if self.buf.is_empty() {
// If we have no unread bytes, we can poll the stream
// and fill self.buf with the bytes we read.
let pinned = std::pin::Pin::new(&mut self.stream.read);
let res = ready!(copy::ResizeBufRead::poll_bytes(pinned, cx))?;
self.buf = res;
}
// Copy as many bytes as we can from self.buf.
let cnt = Ord::min(buf.remaining(), self.buf.len());
buf.put_slice(&self.buf[..cnt]);
self.buf = self.buf.split_off(cnt);
Poll::Ready(Ok(()))
}
}
@ -181,7 +184,7 @@ impl tokio::io::AsyncWrite for TokioH2Stream {
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, tokio::io::Error>> {
let pinned = std::pin::Pin::new(&mut self.0.write);
let pinned = std::pin::Pin::new(&mut self.stream.write);
let buf = Bytes::copy_from_slice(buf);
copy::AsyncWriteBuf::poll_write_buf(pinned, cx, buf)
}
@ -190,7 +193,7 @@ impl tokio::io::AsyncWrite for TokioH2Stream {
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
let pinned = std::pin::Pin::new(&mut self.0.write);
let pinned = std::pin::Pin::new(&mut self.stream.write);
copy::AsyncWriteBuf::poll_flush(pinned, cx)
}
@ -198,7 +201,7 @@ impl tokio::io::AsyncWrite for TokioH2Stream {
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(), std::io::Error>> {
let pinned = std::pin::Pin::new(&mut self.0.write);
let pinned = std::pin::Pin::new(&mut self.stream.write);
copy::AsyncWriteBuf::poll_shutdown(pinned, cx)
}
}
@ -302,6 +305,6 @@ fn h2_to_io_error(e: h2::Error) -> std::io::Error {
if e.is_io() {
e.into_io().unwrap()
} else {
std::io::Error::new(std::io::ErrorKind::Other, e)
std::io::Error::other(e)
}
}

View File

@ -45,7 +45,7 @@ use crate::state::{DemandProxyState, ProxyRbacContext};
use crate::strng::Strng;
use crate::tls::TlsError;
pub(super) struct Inbound {
pub struct Inbound {
listener: socket::Listener,
drain: DrainWatcher,
pi: Arc<ProxyInputs>,
@ -53,7 +53,7 @@ pub(super) struct Inbound {
}
impl Inbound {
pub(super) async fn new(pi: Arc<ProxyInputs>, drain: DrainWatcher) -> Result<Inbound, Error> {
pub(crate) async fn new(pi: Arc<ProxyInputs>, drain: DrainWatcher) -> Result<Inbound, Error> {
let listener = pi
.socket_factory
.tcp_bind(pi.cfg.inbound_addr)
@ -74,11 +74,12 @@ impl Inbound {
})
}
pub(super) fn address(&self) -> SocketAddr {
/// Returns the socket address this proxy is listening on.
pub fn address(&self) -> SocketAddr {
self.listener.local_addr()
}
pub(super) async fn run(self) {
pub async fn run(self) {
let pi = self.pi.clone();
let acceptor = InboundCertProvider {
local_workload: self.pi.local_workload_information.clone(),
@ -122,7 +123,7 @@ impl Inbound {
let conn = Connection {
src_identity,
src,
dst_network: strng::new(&network), // inbound request must be on our network
dst_network: network.clone(), // inbound request must be on our network
dst,
};
debug!(%conn, "accepted connection");
@ -244,10 +245,18 @@ impl Inbound {
SocketAddr::new(loopback, ri.upstream_addr.port()),
)
} else {
(
enable_original_source.then_some(ri.rbac_ctx.conn.src.ip()),
ri.upstream_addr,
)
// When ztunnel is proxying to its own internal endpoints (metrics server after HBONE termination),
// we must not attempt to use the original external client's IP as the source for this internal connection.
// Setting `disable_inbound_freebind` to true for such self-proxy scenarios ensures `upstream_src_ip` is `None`,
// causing `freebind_connect` to use a local IP for the connection to ztunnel's own service.
// For regular inbound traffic to other workloads, `disable_inbound_freebind` is false, and original source
// preservation depends on `enable_original_source`.
let upstream_src_ip = if pi.disable_inbound_freebind {
None
} else {
enable_original_source.then_some(ri.rbac_ctx.conn.src.ip())
};
(upstream_src_ip, ri.upstream_addr)
};
// Establish upstream connection between original source and destination
@ -536,7 +545,7 @@ impl Inbound {
/// find_inbound_upstream determines the next hop for an inbound request.
#[expect(clippy::type_complexity)]
fn find_inbound_upstream(
pub(super) fn find_inbound_upstream(
cfg: &Config,
state: &DemandProxyState,
conn: &Connection,
@ -545,6 +554,7 @@ impl Inbound {
) -> Result<(SocketAddr, Option<TunnelRequest>, Vec<Arc<Service>>), Error> {
// We always target the local workload IP as the destination. But we need to determine the port to send to.
let target_ip = conn.dst.ip();
// First, fetch the actual target SocketAddr as well as all possible services this could be for.
// Given they may request the pod directly, there may be multiple possible services; we will
// select a final one (if any) later.
@ -640,7 +650,7 @@ impl Inbound {
}
#[derive(Debug)]
struct TunnelRequest {
pub(super) struct TunnelRequest {
tunnel_target: SocketAddr,
protocol: Protocol,
}
@ -702,37 +712,36 @@ fn build_response(status: StatusCode) -> Response<()> {
}
#[cfg(test)]
#[allow(clippy::too_many_arguments)]
mod tests {
use super::{Inbound, ProxyInputs};
use crate::{config, proxy::ConnectionManager, proxy::inbound::HboneAddress, strng};
use crate::{
config,
identity::manager::mock::new_secret_manager,
proxy::{
ConnectionManager, DefaultSocketFactory, LocalWorkloadInformation,
h2::server::RequestParts, inbound::HboneAddress,
},
rbac::Connection,
state::{
self, DemandProxyState,
self, DemandProxyState, WorkloadInfo,
service::{Endpoint, EndpointSet, Service},
workload::{
ApplicationTunnel, GatewayAddress, InboundProtocol, NetworkAddress, Workload,
application_tunnel::Protocol as AppProtocol, gatewayaddress::Destination,
ApplicationTunnel, GatewayAddress, HealthStatus, InboundProtocol, NetworkAddress,
NetworkMode, Workload, application_tunnel::Protocol as AppProtocol,
gatewayaddress::Destination,
},
},
test_helpers,
strng, test_helpers,
};
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
use http::{Method, Uri};
use prometheus_client::registry::Registry;
use std::{
net::SocketAddr,
sync::{Arc, RwLock},
time::Duration,
};
use crate::identity::manager::mock::new_secret_manager;
use crate::proxy::DefaultSocketFactory;
use crate::proxy::LocalWorkloadInformation;
use crate::proxy::h2::server::RequestParts;
use crate::state::WorkloadInfo;
use crate::state::workload::HealthStatus;
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
use http::{Method, Uri};
use prometheus_client::registry::Registry;
use test_case::test_case;
const CLIENT_POD_IP: &str = "10.0.0.1";
@ -904,6 +913,7 @@ mod tests {
sf,
None,
local_workload,
false,
));
let inbound_request = Inbound::build_inbound_request(&pi, conn, &request_parts).await;
match want {
@ -959,7 +969,6 @@ mod tests {
"waypoint",
WAYPOINT_POD_IP,
Waypoint::None,
// the waypoint's _workload_ gets the app tunnel field
server_waypoint.app_tunnel(),
),
("client", CLIENT_POD_IP, Waypoint::None, None),
@ -975,6 +984,7 @@ mod tests {
namespace: "default".into(),
service_account: strng::format!("service-account-{name}"),
application_tunnel: app_tunnel,
network_mode: NetworkMode::Standard,
..test_helpers::test_default_workload()
});

View File

@ -35,10 +35,10 @@ use crate::proxy::{ConnectionOpen, ConnectionResult, DerivedWorkload, metrics};
use crate::drain::DrainWatcher;
use crate::drain::run_with_drain;
use crate::proxy::h2::{H2Stream, client::WorkloadKey};
use crate::state::ServiceResolutionMode;
use crate::state::service::ServiceDescription;
use crate::state::service::{Service, ServiceDescription};
use crate::state::workload::OutboundProtocol;
use crate::state::workload::{InboundProtocol, NetworkAddress, Workload, address::Address};
use crate::state::{ServiceResolutionMode, Upstream};
use crate::{assertions, copy, proxy, socket};
use super::h2::TokioH2Stream;
@ -369,6 +369,83 @@ impl OutboundConnection {
}
}
// This function is called when the select next hop is on a different network,
// so we expect the upstream workload to have a network gatewy configured.
//
// When we use a gateway to reach to a workload on a remote network we have to
// use double HBONE (HBONE incapsulated inside HBONE). The gateway will
// terminate the outer HBONE tunnel and forward the inner HBONE to the actual
// destination as a opaque stream of bytes and the actual destination will
// interpret it as an HBONE connection.
//
// If the upstream workload does not have an E/W gateway this function returns
// an error indicating that it could not find a valid destination.
//
// A note about double HBONE, in double HBONE both inner and outer HBONE use
// destination service name as HBONE target URI.
//
// Having target URI in the outer HBONE tunnel allows E/W gateway to figure out
// where to route the data next witout the need to terminate inner HBONE tunnel.
// In other words, it could forward inner HBONE as if it's an opaque stream of
// bytes without trying to interpret it.
//
// NOTE: when connecting through an E/W gateway, regardless of whether there is
// a waypoint or not, we always use service hostname and the service port. It's
// somewhat different from how regular HBONE works, so I'm calling it out here.
async fn build_request_through_gateway(
&self,
source: Arc<Workload>,
// next hop on the remote network that we picked as our destination.
// It may be a local view of a Waypoint workload on remote network or
// a local view of the service workload (when waypoint is not
// configured).
upstream: Upstream,
// This is a target service we wanted to reach in the first place.
//
// NOTE: Crossing network boundaries is only supported for services
// at the moment, so we should always have a service we could use.
service: &Service,
target: SocketAddr,
) -> Result<Request, Error> {
if let Some(gateway) = &upstream.workload.network_gateway {
let gateway_upstream = self
.pi
.state
.fetch_network_gateway(gateway, &source, target)
.await?;
let hbone_target_destination = Some(HboneAddress::SvcHostname(
service.hostname.clone(),
target.port(),
));
debug!("built request to a destination on another network through an E/W gateway");
Ok(Request {
protocol: OutboundProtocol::DOUBLEHBONE,
source,
hbone_target_destination,
actual_destination_workload: Some(gateway_upstream.workload.clone()),
intended_destination_service: Some(ServiceDescription::from(service)),
actual_destination: gateway_upstream.workload_socket_addr().ok_or(
Error::NoValidDestination(Box::new((*gateway_upstream.workload).clone())),
)?,
// The outer tunnel of double HBONE is terminated by the E/W
// gateway and so for the credentials of the next hop
// (upstream_sans) we use gateway credentials.
upstream_sans: gateway_upstream.workload_and_services_san(),
// The inner HBONE tunnel is terminated by either the server
// we want to reach or a Waypoint in front of it, depending on
// the configuration. So for the final destination credentials
// (final_sans) we use the upstream workload credentials.
final_sans: upstream.service_sans(),
})
} else {
// Do not try to send cross-network traffic without network gateway.
Err(Error::NoValidDestination(Box::new(
(*upstream.workload).clone(),
)))
}
}
// build_request computes all information about the request we should send
// TODO: Do we want a single lock for source and upstream...?
async fn build_request(
@ -381,7 +458,7 @@ impl OutboundConnection {
// If this is to-service traffic check for a service waypoint
// Capture result of whether this is svc addressed
let svc_addressed = if let Some(Address::Service(target_service)) = state
let service = if let Some(Address::Service(target_service)) = state
.fetch_address(&NetworkAddress {
network: self.pi.cfg.network.clone(),
address: target.ip(),
@ -393,6 +470,18 @@ impl OutboundConnection {
.fetch_service_waypoint(&target_service, &source_workload, target)
.await?
{
if waypoint.workload.network != source_workload.network {
debug!("picked a waypoint on remote network");
return self
.build_request_through_gateway(
source_workload.clone(),
waypoint,
&target_service,
target,
)
.await;
}
let upstream_sans = waypoint.workload_and_services_san();
let actual_destination =
waypoint
@ -413,10 +502,10 @@ impl OutboundConnection {
});
}
// this was service addressed but we did not find a waypoint
true
Some(target_service)
} else {
// this wasn't service addressed
false
None
};
let Some(us) = state
@ -428,7 +517,7 @@ impl OutboundConnection {
)
.await?
else {
if svc_addressed {
if service.is_some() {
return Err(Error::NoHealthyUpstream(target));
}
debug!("built request as passthrough; no upstream found");
@ -446,37 +535,26 @@ impl OutboundConnection {
// Check whether we are using an E/W gateway and sending cross network traffic
if us.workload.network != source_workload.network {
if let Some(ew_gtw) = &us.workload.network_gateway {
let gtw_us = {
self.pi
.state
.fetch_network_gateway(ew_gtw, &source_workload, target)
.await?
};
let svc = us
.destination_service
.as_ref()
.expect("Workloads with network gateways must be service addressed.");
let hbone_target_destination =
Some(HboneAddress::SvcHostname(svc.hostname.clone(), us.port));
return Ok(Request {
protocol: OutboundProtocol::DOUBLEHBONE,
source: source_workload,
hbone_target_destination,
actual_destination_workload: Some(gtw_us.workload.clone()),
intended_destination_service: us.destination_service.clone(),
actual_destination: gtw_us.workload_socket_addr().ok_or(
Error::NoValidDestination(Box::new((*gtw_us.workload).clone())),
)?,
upstream_sans: gtw_us.workload_and_services_san(),
final_sans: us.service_sans(),
});
} else {
// Do not try to send cross-network traffic without network gateway.
return Err(Error::NoValidDestination(Box::new((*us.workload).clone())));
}
// Workloads on remote network must be service addressed, so if we got here
// and we don't have a service for the original target address then it's a
// bug either in ztunnel itself or in istiod.
//
// For a double HBONE protocol implementation we have to know the
// destination service and if there is no service for the target it's a bug.
//
// This situation "should never happen" because for workloads fetch_upstream
// above only checks the workloads on the same network as this ztunnel
// instance and therefore it should not be able to find a workload on a
// different network.
debug_assert!(
service.is_some(),
"workload on remote network is not service addressed"
);
debug!("picked a workload on remote network");
let service = service.as_ref().ok_or(Error::NoService(target))?;
return self
.build_request_through_gateway(source_workload.clone(), us, service, target)
.await;
}
// We are not using a network gateway and there is no workload address.
@ -491,7 +569,7 @@ impl OutboundConnection {
// Check if we need to go through a workload addressed waypoint.
// Don't traverse waypoint twice if the source is sandwich-outbound.
// Don't traverse waypoint if traffic was addressed to a service (handled before)
if !from_waypoint && !svc_addressed {
if !from_waypoint && service.is_none() {
// For case upstream server has enabled waypoint
let waypoint = state
.fetch_workload_waypoint(&us.workload, &source_workload, target)
@ -716,6 +794,7 @@ mod tests {
local_workload_information: local_workload_information.clone(),
connection_manager: ConnectionManager::default(),
resolver: None,
disable_inbound_freebind: false,
}),
id: TraceParent::new(),
pool: WorkloadHBONEPool::new(
@ -815,6 +894,8 @@ mod tests {
#[tokio::test]
async fn build_request_double_hbone() {
// example.com service has a workload on remote network.
// E/W gateway is addressed by an IP.
run_build_request_multi(
"127.0.0.1",
"127.0.0.3:80",
@ -866,11 +947,13 @@ mod tests {
],
Some(ExpectedRequest {
protocol: OutboundProtocol::DOUBLEHBONE,
hbone_destination: "example.com:8080",
hbone_destination: "example.com:80",
destination: "10.22.1.1:15009",
}),
)
.await;
// example.com service has a workload on remote network.
// E/W gateway is addressed by a hostname.
run_build_request_multi(
"127.0.0.1",
"127.0.0.3:80",
@ -943,11 +1026,218 @@ mod tests {
],
Some(ExpectedRequest {
protocol: OutboundProtocol::DOUBLEHBONE,
hbone_destination: "example.com:8080",
hbone_destination: "example.com:80",
destination: "127.0.0.5:15008",
}),
)
.await;
// example.com service has a waypoint and waypoint workload is on remote network.
// E/W gateway is addressed by an IP.
run_build_request_multi(
"127.0.0.1",
"127.0.0.3:80",
vec![
XdsAddressType::Service(XdsService {
hostname: "example.com".to_string(),
addresses: vec![XdsNetworkAddress {
network: "".to_string(),
address: vec![127, 0, 0, 3],
}],
ports: vec![Port {
service_port: 80,
target_port: 8080,
}],
waypoint: Some(xds::istio::workload::GatewayAddress {
destination: Some(
xds::istio::workload::gateway_address::Destination::Hostname(
XdsNamespacedHostname {
namespace: Default::default(),
hostname: "waypoint.com".into(),
},
),
),
hbone_mtls_port: 15008,
}),
..Default::default()
}),
XdsAddressType::Service(XdsService {
hostname: "waypoint.com".to_string(),
addresses: vec![XdsNetworkAddress {
network: "".to_string(),
address: vec![127, 0, 0, 4],
}],
ports: vec![Port {
service_port: 15008,
target_port: 15008,
}],
..Default::default()
}),
XdsAddressType::Workload(XdsWorkload {
uid: "Kubernetes//Pod/default/remote-waypoint-pod".to_string(),
addresses: vec![],
network: "remote".to_string(),
network_gateway: Some(xds::istio::workload::GatewayAddress {
destination: Some(
xds::istio::workload::gateway_address::Destination::Address(
XdsNetworkAddress {
network: "remote".to_string(),
address: vec![10, 22, 1, 1],
},
),
),
hbone_mtls_port: 15009,
}),
services: std::collections::HashMap::from([(
"/waypoint.com".to_string(),
PortList {
ports: vec![Port {
service_port: 15008,
target_port: 15008,
}],
},
)]),
..Default::default()
}),
XdsAddressType::Workload(XdsWorkload {
uid: "Kubernetes//Pod/default/remote-ew-gtw".to_string(),
addresses: vec![Bytes::copy_from_slice(&[10, 22, 1, 1])],
network: "remote".to_string(),
..Default::default()
}),
],
Some(ExpectedRequest {
protocol: OutboundProtocol::DOUBLEHBONE,
hbone_destination: "example.com:80",
destination: "10.22.1.1:15009",
}),
)
.await;
}
#[tokio::test]
async fn build_request_failover_to_remote() {
// Similar to the double HBONE test that we already have, but it sets up a scenario when
// load balancing logic will pick a workload on a remote cluster when local workloads are
// unhealthy, thus showing the expected failover behavior.
let service = XdsAddressType::Service(XdsService {
hostname: "example.com".to_string(),
addresses: vec![XdsNetworkAddress {
network: "".to_string(),
address: vec![127, 0, 0, 3],
}],
ports: vec![Port {
service_port: 80,
target_port: 8080,
}],
// Prefer routing to workloads on the same network, but when nothing is healthy locally
// allow failing over to remote networks.
load_balancing: Some(xds::istio::workload::LoadBalancing {
routing_preference: vec![
xds::istio::workload::load_balancing::Scope::Network.into(),
],
mode: xds::istio::workload::load_balancing::Mode::Failover.into(),
..Default::default()
}),
..Default::default()
});
let ew_gateway = XdsAddressType::Workload(XdsWorkload {
uid: "Kubernetes//Pod/default/remote-ew-gtw".to_string(),
addresses: vec![Bytes::copy_from_slice(&[10, 22, 1, 1])],
network: "remote".to_string(),
..Default::default()
});
let remote_workload = XdsAddressType::Workload(XdsWorkload {
uid: "Kubernetes//Pod/default/remote-example.com-pod".to_string(),
addresses: vec![],
network: "remote".to_string(),
network_gateway: Some(xds::istio::workload::GatewayAddress {
destination: Some(xds::istio::workload::gateway_address::Destination::Address(
XdsNetworkAddress {
network: "remote".to_string(),
address: vec![10, 22, 1, 1],
},
)),
hbone_mtls_port: 15009,
}),
services: std::collections::HashMap::from([(
"/example.com".to_string(),
PortList {
ports: vec![Port {
service_port: 80,
target_port: 8080,
}],
},
)]),
..Default::default()
});
let healthy_local_workload = XdsAddressType::Workload(XdsWorkload {
uid: "Kubernetes//Pod/default/local-example.com-pod".to_string(),
addresses: vec![Bytes::copy_from_slice(&[127, 0, 0, 2])],
network: "".to_string(),
tunnel_protocol: xds::istio::workload::TunnelProtocol::Hbone.into(),
services: std::collections::HashMap::from([(
"/example.com".to_string(),
PortList {
ports: vec![Port {
service_port: 80,
target_port: 8080,
}],
},
)]),
status: xds::istio::workload::WorkloadStatus::Healthy.into(),
..Default::default()
});
let unhealthy_local_workload = XdsAddressType::Workload(XdsWorkload {
uid: "Kubernetes//Pod/default/local-example.com-pod".to_string(),
addresses: vec![Bytes::copy_from_slice(&[127, 0, 0, 2])],
network: "".to_string(),
tunnel_protocol: xds::istio::workload::TunnelProtocol::Hbone.into(),
services: std::collections::HashMap::from([(
"/example.com".to_string(),
PortList {
ports: vec![Port {
service_port: 80,
target_port: 8080,
}],
},
)]),
status: xds::istio::workload::WorkloadStatus::Unhealthy.into(),
..Default::default()
});
run_build_request_multi(
"127.0.0.1",
"127.0.0.3:80",
vec![
service.clone(),
ew_gateway.clone(),
remote_workload.clone(),
healthy_local_workload.clone(),
],
Some(ExpectedRequest {
protocol: OutboundProtocol::HBONE,
hbone_destination: "127.0.0.2:8080",
destination: "127.0.0.2:15008",
}),
)
.await;
run_build_request_multi(
"127.0.0.1",
"127.0.0.3:80",
vec![
service.clone(),
ew_gateway.clone(),
remote_workload.clone(),
unhealthy_local_workload.clone(),
],
Some(ExpectedRequest {
protocol: OutboundProtocol::DOUBLEHBONE,
hbone_destination: "example.com:80",
destination: "10.22.1.1:15009",
}),
)
.await;
}
#[tokio::test]

View File

@ -594,9 +594,10 @@ mod test {
}
/// This is really a test for TokioH2Stream, but its nicer here because we have access to
/// streams
/// streams.
/// Most important, we make sure there are no panics.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn small_reads() {
async fn read_buffering() {
let (mut pool, srv) = setup_test(3).await;
let key = key(&srv, 2);
@ -612,13 +613,28 @@ mod test {
let c = pool.send_request_pooled(&key.clone(), req()).await.unwrap();
let mut c = TokioH2Stream::new(c);
c.write_all(b"abcde").await.unwrap();
let mut b = [0u8; 0];
// Crucially, this should error rather than panic.
if let Err(e) = c.read(&mut b).await {
assert_eq!(e.kind(), io::ErrorKind::Other);
} else {
panic!("Should have errored");
}
let mut b = [0u8; 100];
// Properly buffer reads and don't error
assert_eq!(c.read(&mut b).await.unwrap(), 8);
assert_eq!(&b[..8], b"poolsrv\n"); // this is added by itself
assert_eq!(c.read(&mut b[..1]).await.unwrap(), 1);
assert_eq!(&b[..1], b"a");
assert_eq!(c.read(&mut b[..1]).await.unwrap(), 1);
assert_eq!(&b[..1], b"b");
assert_eq!(c.read(&mut b[..1]).await.unwrap(), 1);
assert_eq!(&b[..1], b"c");
assert_eq!(c.read(&mut b).await.unwrap(), 2); // there are only two bytes left
assert_eq!(&b[..2], b"de");
// Once we drop the pool, we should still retained the buffered data,
// but then we should error.
c.write_all(b"abcde").await.unwrap();
assert_eq!(c.read(&mut b[..3]).await.unwrap(), 3);
assert_eq!(&b[..3], b"abc");
drop(pool);
assert_eq!(c.read(&mut b[..2]).await.unwrap(), 2);
assert_eq!(&b[..2], b"de");
assert!(c.read(&mut b).await.is_err());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]

View File

@ -22,10 +22,9 @@ use crate::dns;
use crate::drain::DrainWatcher;
use crate::proxy::connection_manager::ConnectionManager;
use crate::proxy::{DefaultSocketFactory, Proxy, inbound::Inbound};
use crate::proxy::{Error, LocalWorkloadInformation, Metrics};
use crate::proxy::Proxy;
// Proxy factory creates ztunnel proxies using a socket factory.
// this allows us to create our proxies the same way in regular mode and in inpod mode.
pub struct ProxyFactory {
@ -113,6 +112,7 @@ impl ProxyFactory {
drain.clone(),
socket_factory.as_ref(),
local_workload_information.as_fetcher(),
self.config.prefered_service_namespace.clone(),
)
.await?;
resolver = Some(server.resolver());
@ -130,6 +130,7 @@ impl ProxyFactory {
socket_factory.clone(),
resolver,
local_workload_information,
false,
);
result.connection_manager = Some(cm);
result.proxy = Some(Proxy::from_inputs(pi, drain).await?);
@ -137,6 +138,52 @@ impl ProxyFactory {
Ok(result)
}
/// Creates an inbound listener specifically for ztunnel's own internal endpoints (metrics).
/// This allows ztunnel to act as its own workload, enforcing policies on traffic directed to itself.
/// This is distinct from the main inbound listener which handles traffic for other workloads proxied by ztunnel.
pub async fn create_ztunnel_self_proxy_listener(
&self,
) -> Result<Option<crate::proxy::inbound::Inbound>, Error> {
if self.config.proxy_mode != config::ProxyMode::Shared {
return Ok(None);
}
if let (Some(ztunnel_identity), Some(ztunnel_workload)) =
(&self.config.ztunnel_identity, &self.config.ztunnel_workload)
{
tracing::info!(
"creating ztunnel self-proxy listener with identity: {:?}",
ztunnel_identity
);
let local_workload_information = Arc::new(LocalWorkloadInformation::new(
Arc::new(ztunnel_workload.clone()),
self.state.clone(),
self.cert_manager.clone(),
));
let socket_factory = Arc::new(DefaultSocketFactory(self.config.socket_config));
let cm = ConnectionManager::default();
let pi = crate::proxy::ProxyInputs::new(
self.config.clone(),
cm.clone(),
self.state.clone(),
self.proxy_metrics.clone(),
socket_factory,
None,
local_workload_information,
true,
);
let inbound = Inbound::new(pi, self.drain.clone()).await?;
Ok(Some(inbound))
} else {
Ok(None)
}
}
}
#[derive(Default)]

View File

@ -366,6 +366,19 @@ impl ProxyState {
debug!("failed to fetch workload for {}", ep.workload_uid);
return None;
};
let in_network = wl.network == src.network;
let has_network_gateway = wl.network_gateway.is_some();
let has_address = !wl.workload_ips.is_empty() || !wl.hostname.is_empty();
if !has_address {
// Workload has no IP. We can only reach it via a network gateway
// WDS is client-agnostic, so we will get a network gateway for a workload
// even if it's in the same network; we should never use it.
if in_network || !has_network_gateway {
return None;
}
}
match resolution_mode {
ServiceResolutionMode::Standard => {
if target_port.unwrap_or_default() == 0 && !ep.port.contains_key(&svc_port) {
@ -1571,6 +1584,22 @@ mod tests {
},
..test_helpers::test_default_workload()
};
let wl_empty_ip = Workload {
uid: "cluster1//v1/Pod/default/wl_empty_ip".into(),
name: "wl_empty_ip".into(),
namespace: "default".into(),
trust_domain: "cluster.local".into(),
service_account: "default".into(),
workload_ips: vec![], // none!
network: "network".into(),
locality: Locality {
region: "reg".into(),
zone: "zone".into(),
subzone: "".into(),
},
..test_helpers::test_default_workload()
};
let _ep_almost = Workload {
uid: "cluster1//v1/Pod/default/ep_almost".into(),
name: "wl_almost".into(),
@ -1617,6 +1646,11 @@ mod tests {
port: HashMap::from([(80u16, 80u16)]),
status: HealthStatus::Healthy,
},
Endpoint {
workload_uid: "cluster1//v1/Pod/default/wl_empty_ip".into(),
port: HashMap::from([(80u16, 80u16)]),
status: HealthStatus::Healthy,
},
]);
let strict_svc = Service {
endpoints: endpoints.clone(),
@ -1649,6 +1683,7 @@ mod tests {
state.workloads.insert(Arc::new(wl_no_locality.clone()));
state.workloads.insert(Arc::new(wl_match.clone()));
state.workloads.insert(Arc::new(wl_almost.clone()));
state.workloads.insert(Arc::new(wl_empty_ip.clone()));
state.services.insert(strict_svc.clone());
state.services.insert(failover_svc.clone());
@ -1663,6 +1698,15 @@ mod tests {
assert!(want.contains(&got.unwrap()), "{}", desc);
}
};
let assert_not_endpoint =
|src: &Workload, svc: &Service, uid: &str, tries: usize, desc: &str| {
for _ in 0..tries {
let got = state
.load_balance(src, svc, 80, ServiceResolutionMode::Standard)
.map(|(ep, _)| ep.workload_uid.as_str());
assert!(got != Some(uid), "{}", desc);
}
};
assert_endpoint(
&wl_no_locality,
@ -1708,5 +1752,12 @@ mod tests {
vec!["cluster1//v1/Pod/default/wl_match"],
"failover full match selects closest match",
);
assert_not_endpoint(
&wl_no_locality,
&failover_svc,
"cluster1//v1/Pod/default/wl_empty_ip",
10,
"failover no match can select any endpoint",
);
}
}

View File

@ -326,9 +326,7 @@ impl io::Write for WriteAdaptor<'_> {
let s =
std::str::from_utf8(buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
self.fmt_write
.write_str(s)
.map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
self.fmt_write.write_str(s).map_err(io::Error::other)?;
Ok(s.len())
}

View File

@ -52,6 +52,7 @@ pub struct TestApp {
pub namespace: Option<super::netns::Namespace>,
pub shutdown: ShutdownTrigger,
pub ztunnel_identity: Option<identity::Identity>,
}
impl From<(&Bound, Arc<SecretManager>)> for TestApp {
@ -66,6 +67,7 @@ impl From<(&Bound, Arc<SecretManager>)> for TestApp {
cert_manager,
namespace: None,
shutdown: app.shutdown.trigger(),
ztunnel_identity: None,
}
}
}

View File

@ -298,6 +298,7 @@ pub async fn run_dns(responses: HashMap<Name, Vec<IpAddr>>) -> anyhow::Result<Te
}),
state.clone(),
),
Some("prefered-namespace".to_string()),
)
.await?;

View File

@ -124,38 +124,104 @@ impl WorkloadManager {
wli: Option<state::WorkloadInfo>,
) -> anyhow::Result<TestApp> {
let mut inpod_uds: PathBuf = "/dev/null".into();
let ztunnel_server = if self.mode == Shared {
inpod_uds = self.tmp_dir.join(node);
Some(start_ztunnel_server(inpod_uds.clone()).await)
let current_mode = self.mode;
let proxy_mode = match current_mode {
Shared => ProxyMode::Shared,
Dedicated => ProxyMode::Dedicated,
};
let ztunnel_name = format!("ztunnel-{node}");
// Define ztunnel's own identity and workload info if it's a Shared proxy.
// These are used for registering ztunnel as a workload and for cfg.ztunnel_identity/workload.
let ztunnel_shared_identity: Option<identity::Identity> = if proxy_mode == ProxyMode::Shared
{
Some(identity::Identity::Spiffe {
trust_domain: "cluster.local".into(),
namespace: "default".into(),
service_account: ztunnel_name.clone().into(),
})
} else {
None
};
let ns = TestWorkloadBuilder::new(&format!("ztunnel-{node}"), self)
.on_node(node)
.uncaptured()
.register()
.await?;
let ztunnel_shared_workload_info: Option<state::WorkloadInfo> =
if proxy_mode == ProxyMode::Shared {
Some(state::WorkloadInfo::new(
ztunnel_name.clone(),
"default".to_string(),
ztunnel_name.clone(),
))
} else {
None
};
let ztunnel_server = match current_mode {
Shared => {
inpod_uds = self.tmp_dir.join(node);
Some(start_ztunnel_server(inpod_uds.clone()).await)
}
Dedicated => None,
};
let ns = match current_mode {
Shared => {
// Shared mode: Ztunnel has its own identity, registered as HBONE
TestWorkloadBuilder::new(&ztunnel_name, self)
.on_node(node)
.identity(
ztunnel_shared_identity
.clone()
.expect("Shared mode must have an identity for ztunnel registration"),
)
.hbone() // Shared ztunnel uses HBONE protocol
.register()
.await?
}
Dedicated => {
TestWorkloadBuilder::new(&ztunnel_name, self)
.on_node(node)
.uncaptured() // Dedicated ztunnel is treated as uncaptured TCP
.register()
.await?
}
};
let _ztunnel_local_workload = self
.workloads
.last()
.cloned()
.expect("ztunnel workload should be registered");
let ip = ns.ip();
let initial_config = LocalConfig {
workloads: self.workloads.clone(),
policies: self.policies.clone(),
services: self.services.values().cloned().collect_vec(),
};
let proxy_mode = if ztunnel_server.is_some() {
ProxyMode::Shared
} else {
ProxyMode::Dedicated
};
let (mut tx_cfg, rx_cfg) = mpsc_ack(1);
tx_cfg.send(initial_config).await?;
let local_xds_config = Some(ConfigSource::Dynamic(Arc::new(Mutex::new(rx_cfg))));
// Config for ztunnel's own identity and workload, primarily for when it acts as a server (metrics endpoint).
let cfg_ztunnel_identity = ztunnel_shared_identity.clone();
let cfg_ztunnel_workload_info = ztunnel_shared_workload_info.clone();
// Config for the workload this ztunnel instance is proxying for :
// If Shared, ztunnel is effectively proxying for itself
// If Dedicated, it's for the application workload `wli`
let cfg_proxy_workload_information = match proxy_mode {
// Ztunnel's own info for shared mode proxy
ProxyMode::Shared => ztunnel_shared_workload_info.clone(),
// Application's workload info for dedicated mode
ProxyMode::Dedicated => wli,
};
let cfg = config::Config {
xds_address: None,
dns_proxy: true,
fake_ca: true,
local_xds_config,
local_node: Some(node.to_string()),
proxy_workload_information: wli,
proxy_workload_information: cfg_proxy_workload_information,
inpod_uds,
proxy_mode,
// We use packet mark even in dedicated to distinguish proxy from application
@ -166,12 +232,16 @@ impl WorkloadManager {
Some(true)
},
localhost_app_tunnel: true,
ztunnel_identity: cfg_ztunnel_identity,
ztunnel_workload: cfg_ztunnel_workload_info,
..config::parse_config().unwrap()
};
let (tx, rx) = std::sync::mpsc::sync_channel(0);
// Setup the ztunnel...
let cloned_ns = ns.clone();
let cloned_ns2 = ns.clone();
let ztunnel_identity = ztunnel_shared_identity.clone();
// run_ready will spawn a thread and block on it. Run with spawn_blocking so it doesn't block the runtime.
tokio::task::spawn_blocking(move || {
ns.run_ready(move |ready| async move {
@ -210,9 +280,9 @@ impl WorkloadManager {
ip,
)),
cert_manager,
namespace: Some(cloned_ns),
shutdown,
ztunnel_identity: ztunnel_identity.clone(),
};
ta.ready().await;
info!("ready");
@ -512,16 +582,18 @@ impl<'a> TestWorkloadBuilder<'a> {
pub async fn register(mut self) -> anyhow::Result<Namespace> {
let zt = self.manager.ztunnels.get(self.w.workload.node.as_str());
let node = self.w.workload.node.clone();
let network_namespace = if self.manager.mode == Dedicated && zt.is_some() {
// This is a bit of hack. For dedicated mode, we run the app and ztunnel in the same namespace
// We probably should express this more natively in the framework, but for now we just detect it
// and re-use the namespace.
tracing::info!("node already has ztunnel and dedicate mode, sharing");
zt.as_ref().unwrap().namespace.clone()
} else {
self.manager
let network_namespace = match (self.manager.mode, zt.is_some()) {
(Dedicated, true) => {
// This is a bit of hack. For dedicated mode, we run the app and ztunnel in the same namespace
// We probably should express this more natively in the framework, but for now we just detect it
// and re-use the namespace.
tracing::info!("node already has ztunnel and dedicate mode, sharing");
zt.as_ref().unwrap().namespace.clone()
}
_ => self
.manager
.namespaces
.child(&self.w.workload.node, &self.w.workload.name)?
.child(&self.w.workload.node, &self.w.workload.name)?,
};
if self.w.workload.network_gateway.is_some() {
// This is a little inefficient, because we create the
@ -569,7 +641,7 @@ impl<'a> TestWorkloadBuilder<'a> {
let fd = network_namespace.netns().file().as_raw_fd();
let msg = inpod::Message::Start(inpod::StartZtunnelMessage {
uid: uid.to_string(),
workload_info: Some(wli),
workload_info: Some(wli.clone()),
fd,
});
zt_info

View File

@ -124,28 +124,44 @@ impl CsrOptions {
#[cfg(test)]
mod tests {
use crate::tls;
use itertools::Itertools;
#[test]
fn test_csr() {
use x509_parser::prelude::FromDer;
use x509_parser::prelude::*;
let csr = tls::csr::CsrOptions {
san: "spiffe://td/ns/ns1/sa/sa1".to_string(),
}
.generate()
.unwrap();
let (_, der) = x509_parser::pem::parse_x509_pem(csr.csr.as_bytes()).unwrap();
let (_, cert) =
x509_parser::certification_request::X509CertificationRequest::from_der(&der.contents)
.unwrap();
cert.verify_signature().unwrap();
let subject = cert.certification_request_info.subject.iter().collect_vec();
assert_eq!(subject.len(), 0);
let attr = cert
.certification_request_info
.iter_attributes()
.next()
.unwrap();
// SAN is encoded in some format I don't understand how to parse; this could be improved.
// but make sure it's there in a hacky manner
assert!(attr.value.ends_with(b"spiffe://td/ns/ns1/sa/sa1"));
let ParsedCriAttribute::ExtensionRequest(parsed) = attr.parsed_attribute() else {
panic!("not a ExtensionRequest")
};
let ext = parsed.clone().extensions;
assert_eq!(ext.len(), 1);
let ext = ext.into_iter().next().unwrap();
assert!(ext.critical);
let ParsedExtension::SubjectAlternativeName(san) = ext.parsed_extension() else {
panic!("not a SubjectAlternativeName")
};
assert_eq!(
&format!("{san:?}"),
"SubjectAlternativeName { general_names: [URI(\"spiffe://td/ns/ns1/sa/sa1\")] }"
)
}
}

View File

@ -14,6 +14,8 @@
use super::Error;
#[allow(unused_imports)]
use crate::PQC_ENABLED;
use crate::identity::{self, Identity};
use std::fmt::Debug;
@ -40,6 +42,15 @@ pub trait ServerCertProvider: Send + Sync + Clone {
pub(super) static TLS_VERSIONS: &[&rustls::SupportedProtocolVersion] = &[&rustls::version::TLS13];
#[cfg(feature = "tls-aws-lc")]
pub static CRYPTO_PROVIDER: &str = "tls-aws-lc";
#[cfg(feature = "tls-ring")]
pub static CRYPTO_PROVIDER: &str = "tls-ring";
#[cfg(feature = "tls-boring")]
pub static CRYPTO_PROVIDER: &str = "tls-boring";
#[cfg(feature = "tls-openssl")]
pub static CRYPTO_PROVIDER: &str = "tls-openssl";
// Ztunnel use `rustls` with pluggable crypto modules.
// All crypto MUST be done via the below providers.
//
@ -68,14 +79,20 @@ pub(super) fn provider() -> Arc<CryptoProvider> {
#[cfg(feature = "tls-aws-lc")]
pub(super) fn provider() -> Arc<CryptoProvider> {
Arc::new(CryptoProvider {
let mut provider = CryptoProvider {
// Limit to only the subset of ciphers that are FIPS compatible
cipher_suites: vec![
rustls::crypto::aws_lc_rs::cipher_suite::TLS13_AES_256_GCM_SHA384,
rustls::crypto::aws_lc_rs::cipher_suite::TLS13_AES_128_GCM_SHA256,
],
..rustls::crypto::aws_lc_rs::default_provider()
})
};
if *PQC_ENABLED {
provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519MLKEM768]
}
Arc::new(provider)
}
#[cfg(feature = "tls-openssl")]

View File

@ -17,10 +17,11 @@ use std::fmt;
use std::fmt::{Display, Formatter};
use std::string::String;
use crate::tls::CRYPTO_PROVIDER;
const BUILD_VERSION: &str = env!("ZTUNNEL_BUILD_buildVersion");
const BUILD_GIT_REVISION: &str = env!("ZTUNNEL_BUILD_buildGitRevision");
const BUILD_STATUS: &str = env!("ZTUNNEL_BUILD_buildStatus");
const BUILD_TAG: &str = env!("ZTUNNEL_BUILD_buildTag");
const BUILD_RUST_VERSION: &str = env!("ZTUNNEL_BUILD_RUSTC_VERSION");
const BUILD_RUST_PROFILE: &str = env!("ZTUNNEL_BUILD_PROFILE_NAME");
@ -32,8 +33,8 @@ pub struct BuildInfo {
rust_version: String,
build_profile: String,
build_status: String,
git_tag: String,
pub istio_version: String,
crypto_provider: String,
}
impl BuildInfo {
@ -44,8 +45,9 @@ impl BuildInfo {
rust_version: BUILD_RUST_VERSION.to_string(),
build_profile: BUILD_RUST_PROFILE.to_string(),
build_status: BUILD_STATUS.to_string(),
git_tag: BUILD_TAG.to_string(),
istio_version: env::var("ISTIO_VERSION").unwrap_or_else(|_| "unknown".to_string()),
istio_version: env::var("ISTIO_META_ISTIO_VERSION")
.unwrap_or_else(|_| "unknown".to_string()),
crypto_provider: CRYPTO_PROVIDER.to_string(),
}
}
}
@ -54,14 +56,14 @@ impl Display for BuildInfo {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
write!(
f,
"version.BuildInfo{{Version:\"{}\", GitRevision:\"{}\", RustVersion:\"{}\", BuildProfile:\"{}\", BuildStatus:\"{}\", GitTag:\"{}\", IstioVersion:\"{}\"}}",
"version.BuildInfo{{Version:\"{}\", GitRevision:\"{}\", RustVersion:\"{}\", BuildProfile:\"{}\", BuildStatus:\"{}\", IstioVersion:\"{}\", CryptoProvider:\"{}\"}}",
self.version,
self.git_revision,
self.rust_version,
self.build_profile,
self.build_status,
self.git_tag,
self.istio_version
self.istio_version,
self.crypto_provider,
)
}
}

View File

@ -18,14 +18,16 @@ mod namespaced {
use futures::future::poll_fn;
use http_body_util::Empty;
use std::collections::HashMap;
use ztunnel::state::workload::ApplicationTunnel;
use ztunnel::state::workload::application_tunnel::Protocol;
use ztunnel::state::workload::gatewayaddress::Destination;
use ztunnel::state::workload::{GatewayAddress, NamespacedHostname};
use ztunnel::test_helpers::linux::TestMode;
use std::net::{IpAddr, SocketAddr};
use anyhow::Context;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::Duration;
use ztunnel::rbac::{Authorization, RbacMatch, StringMatch};
@ -39,18 +41,17 @@ mod namespaced {
use tokio::time::timeout;
use tracing::{error, info};
use ztunnel::state::workload::{ApplicationTunnel, NetworkAddress};
use ztunnel::test_helpers::app::ParsedMetrics;
use ztunnel::test_helpers::app::TestApp;
use ztunnel::state::workload::NetworkAddress;
use ztunnel::test_helpers::app::{ParsedMetrics, TestApp};
use ztunnel::test_helpers::linux::TestMode::{Dedicated, Shared};
use ztunnel::test_helpers::linux::WorkloadManager;
use ztunnel::test_helpers::netns::{Namespace, Resolver};
use ztunnel::test_helpers::*;
use ztunnel::{identity, strng, telemetry};
use crate::namespaced::WorkloadMode::Captured;
use ztunnel::setup_netns_test;
use ztunnel::test_helpers::linux::TestMode::{Dedicated, Shared};
use ztunnel::test_helpers::linux::WorkloadManager;
use ztunnel::test_helpers::netns::{Namespace, Resolver};
use ztunnel::test_helpers::*;
const WAYPOINT_MESSAGE: &[u8] = b"waypoint\n";
@ -926,27 +927,12 @@ mod namespaced {
// Now shutdown the server. In real world, the server app would shutdown, then ztunnel would remove itself.
// In this test, we will leave the server app running, but shutdown ztunnel.
manager.delete_workload("server").await.unwrap();
// Request should fail now
let tx = Arc::new(Mutex::new(tx));
#[allow(clippy::await_holding_lock)]
assert_eventually(
Duration::from_secs(2),
|| async { tx.lock().unwrap().send_and_wait(()).await.is_err() },
true,
)
.await;
// Close the connection
drop(tx);
// Should fail as the last request fails
assert!(cjh.join().unwrap().is_err());
// Now try to connect and make sure it fails
// In shared mode, verify that new connections succeed but data transfer fails
client
.run_and_wait(move || async move {
let mut stream = TcpStream::connect(srv).await.unwrap();
// We should be able to connect (since client is running), but not send a request
const BODY: &[u8] = b"hello world";
stream.write_all(BODY).await.unwrap();
let mut buf = [0; BODY.len() * 2];
@ -955,6 +941,16 @@ mod namespaced {
Ok(())
})
.unwrap();
// The long running connection should also fail on next attempt
let tx_send_result = tx.send_and_wait(()).await;
assert!(
tx_send_result.is_err(),
"long running connection should fail after workload deletion"
);
drop(tx);
assert!(cjh.join().unwrap().is_err());
Ok(())
}
@ -1220,7 +1216,7 @@ mod namespaced {
vec![
(zt, 15001, Request), // Outbound: should be blocked due to recursive call
(zt, 15006, Request), // Inbound: should be blocked due to recursive call
(zt, 15008, Request), // HBONE: expected TLS, reject
(zt, 15008, Request), // HBONE: Connection succeeds (ztunnel listens) but request fails due to TLS
// Localhost still get connection established, as ztunnel accepts anything. But they are dropped immediately.
(zt, 15080, Request), // socks5: localhost
(zt, 15000, Request), // admin: localhost
@ -1252,7 +1248,7 @@ mod namespaced {
// Ztunnel doesn't listen on these ports...
(zt, 15001, Connection), // Outbound: should be blocked due to recursive call
(zt, 15006, Connection), // Inbound: should be blocked due to recursive call
(zt, 15008, Connection), // HBONE: expected TLS, reject
(zt, 15008, Request), // HBONE: Connection succeeds (ztunnel listens) but request fails due to TLS
// Localhost is not accessible
(zt, 15080, Connection), // socks5: localhost
(zt, 15000, Connection), // admin: localhost
@ -1329,66 +1325,191 @@ mod namespaced {
let id1s = id1.to_string();
let ta = manager.deploy_ztunnel(DEFAULT_NODE).await?;
let ztunnel_identity_obj = ta.ztunnel_identity.as_ref().unwrap().clone();
ta.cert_manager
.fetch_certificate(&ztunnel_identity_obj)
.await?;
let ztunnel_identity_str = ztunnel_identity_obj.to_string();
let check = |want: Vec<String>, help: &str| {
let cm = ta.cert_manager.clone();
let help = help.to_string();
let mut sorted_want = want.clone();
sorted_want.sort();
async move {
// Cert manager is async, so we need to wait
let res = check_eventually(
Duration::from_secs(2),
|| cm.collect_certs(|a, _b| a.to_string()),
want,
|| async {
let mut certs = cm.collect_certs(|a, _b| a.to_string()).await;
certs.sort();
certs
},
sorted_want,
)
.await;
assert!(res.is_ok(), "{}: got {:?}", help, res.err().unwrap());
}
};
check(vec![], "initially empty").await;
check(
vec![ztunnel_identity_str.clone()],
"initially only ztunnel cert",
)
.await;
manager
.workload_builder("id1-a-remote-node", REMOTE_NODE)
.identity(id1.clone())
.register()
.await?;
check(vec![], "we should not prefetch remote nodes").await;
check(
vec![ztunnel_identity_str.clone()],
"we should not prefetch remote nodes",
)
.await;
manager
.workload_builder("id1-a-same-node", DEFAULT_NODE)
.identity(id1.clone())
.register()
.await?;
check(vec![id1s.clone()], "we should prefetch our nodes").await;
check(
vec![ztunnel_identity_str.clone(), id1s.clone()],
"we should prefetch our nodes",
)
.await;
manager
.workload_builder("id1-b-same-node", DEFAULT_NODE)
.identity(id1.clone())
.register()
.await?;
check(
vec![id1s.clone()],
vec![ztunnel_identity_str.clone(), id1s.clone()],
"multiple of same identity shouldn't do anything",
)
.await;
manager.delete_workload("id1-a-remote-node").await?;
// Deleting remote node should not affect local certs if local workloads still exist
check(
vec![id1s.clone()],
vec![ztunnel_identity_str.clone(), id1s.clone()],
"removing remote node shouldn't impact anything",
)
.await;
manager.delete_workload("id1-b-same-node").await?;
// Deleting one local node shouldn't impact certs if another local workload still exists
check(
vec![id1s.clone()],
vec![ztunnel_identity_str.clone(), id1s.clone()],
"removing local node shouldn't impact anything if I still have some running",
)
.await;
manager.delete_workload("id1-a-same-node").await?;
// TODO: this should be vec![], but our testing setup doesn't exercise the real codepath
// After deleting all workloads using sa1, give cert manager time to clean up
tokio::time::sleep(Duration::from_millis(100)).await;
// In shared mode, certificates may be kept alive by the inbound listener
// for handling inbound connections, even after workload deletion
let expected_certs = match manager.mode() {
TestMode::Shared => vec![ztunnel_identity_str.clone(), id1s.clone()],
TestMode::Dedicated => vec![ztunnel_identity_str.clone()],
};
check(
vec![id1s.clone()],
"removing final workload should clear things out",
expected_certs,
"removing final workload should clear certs except those needed by inbound listener",
)
.await;
Ok(())
}
#[tokio::test]
async fn test_hbone_metrics_access() -> Result<(), anyhow::Error> {
let mut manager = setup_netns_test!(Shared);
// Deploy ztunnel for the node
let zt = manager.deploy_ztunnel(DEFAULT_NODE).await?;
let ztunnel_node_ip = manager.resolve("ztunnel-node")?;
// Use the actual metrics address ztunnel is listening on (e.g., [::]:15020)
// but combine it with the node IP for the client to target.
let target_metrics_addr = SocketAddr::new(ztunnel_node_ip, zt.metrics_address.port());
let target_metrics_url = format!("http://{}/metrics", target_metrics_addr);
// Deploy a client workload (simulating Prometheus)
let client = manager
.workload_builder("client", DEFAULT_NODE)
.register()
.await?;
let zt_identity_str = zt.ztunnel_identity.as_ref().unwrap().to_string();
// Client makes a standard HTTP GET request to ztunnel's metrics endpoint
// Ztunnel's outbound capture should intercept this, initiate HBONE to its own inbound,
// which then proxies to the internal metrics server.
client
.run(move || async move {
info!(target=%target_metrics_url, "Client attempting standard HTTP GET to metrics endpoint");
let client = hyper_util::client::legacy::Client::builder(
ztunnel::hyper_util::TokioExecutor,
)
.build_http();
let req = hyper::Request::builder()
.method(Method::GET)
.uri(&target_metrics_url)
.body(Empty::<Bytes>::new())?;
let response = client.request(req).await?;
info!("Received response status: {:?}", response.status());
assert_eq!(response.status(), StatusCode::OK, "GET request failed");
let body_bytes = http_body_util::BodyExt::collect(response.into_body())
.await?
.to_bytes();
let response_str = String::from_utf8_lossy(&body_bytes);
assert!(
response_str.contains("# TYPE"),
"Expected Prometheus metrics (# TYPE) in response, got:\n{}",
response_str
);
info!("Successfully verified metrics response body");
Ok(())
})?
.join()
.unwrap()?;
// Verify metrics from the DESTINATION perspective (ztunnel handling its own inbound)
let metrics = [
(CONNECTIONS_OPENED, 1), // One connection opened (client -> zt inbound via HBONE)
(CONNECTIONS_CLOSED, 1), // One connection closed
];
verify_metrics(&zt, &metrics, &destination_labels()).await;
// Verify INBOUND telemetry log for the metrics connection
let dst_addr_log = format!("{}:15008", ztunnel_node_ip);
let dst_hbone_addr_log = format!("{}", target_metrics_addr);
// We don't know exact byte counts, so omit them from the check for now
let want = HashMap::from([
("scope", "access"),
("src.workload", "client"),
("dst.workload", "ztunnel-node"), // ztunnel's workload name
("dst.addr", dst_addr_log.as_str()), // Connected to HBONE port
("dst.hbone_addr", dst_hbone_addr_log.as_str()), // Original target
("direction", "inbound"),
("message", "connection complete"), // Assuming success
(
"src.identity",
"spiffe://cluster.local/ns/default/sa/client",
), // Client identity
("dst.identity", zt_identity_str.as_str()), // Ztunnel identity
]);
telemetry::testing::assert_contains(want);
Ok(())
}
const TEST_VIP: &str = "10.10.0.1";
const TEST_VIP2: &str = "10.10.0.2";
const TEST_VIP3: &str = "10.10.0.3";
@ -1759,7 +1880,6 @@ mod namespaced {
}
use Failure::*;
use ztunnel::state::WorkloadInfo;
use ztunnel::state::workload::application_tunnel::Protocol;
async fn malicious_calls_test(
client: Namespace,
@ -1786,14 +1906,14 @@ mod namespaced {
let stream = timeout(Duration::from_secs(1), TcpStream::connect(tgt)).await?;
error!("stream {stream:?}");
if failure == Connection {
assert!(stream.is_err());
assert!(stream.is_err(), "expected connection to fail for {tgt}");
continue;
}
let mut stream = stream.unwrap();
let res = timeout(Duration::from_secs(1), send_traffic(&mut stream)).await?;
if failure == Request {
assert!(res.is_err());
assert!(res.is_err(), "expected request to fail for {tgt}");
continue;
}
res.unwrap();