feat(eventhubs): support AMQP-over-WebSockets transport

Cargo.toml

@@ -104,11 +104,12 @@ criterion = { version = "0.8", features = ["async_tokio"] }
 crossbeam = { version = "0.8", default-features = false }
 crossbeam-epoch = { version = "0.9", default-features = false }
 dyn-clone = "1.0"
-fe2o3-amqp = { version = "0.14", features = ["uuid"] }
-fe2o3-amqp-ext = { version = "0.14" }
-fe2o3-amqp-management = { version = "0.14" }
-fe2o3-amqp-cbs = { version = "0.14" }
-fe2o3-amqp-types = { version = "0.14" }
+fe2o3-amqp = { version = "0.16", features = ["uuid"] }
+fe2o3-amqp-ws = { version = "0.16" }
+fe2o3-amqp-ext = { version = "0.16" }
+fe2o3-amqp-management = { version = "0.16" }
+fe2o3-amqp-cbs = { version = "0.16" }
+fe2o3-amqp-types = { version = "0.16" }
 flate2 = "1.1.9"
 futures = "0.3"
 getrandom = { version = "0.4" }
@@ -136,8 +137,10 @@ reqwest = { version = "0.13.2", features = [
 ], default-features = false }
 rust_decimal = "1.40.0"
 rustc_version = "0.4"
+rustls = "0.23"
+rustls-platform-verifier = "0.7"
 serde = { version = "1.0", features = ["derive"] }
-serde_amqp = { version = "0.14", features = ["uuid"] }
+serde_amqp = { version = "0.16", features = ["uuid"] }
 serde_bytes = { version = "0.11" }
 serde_json = "1.0.149"
 serde_test = "1"
@@ -156,6 +159,7 @@ tokio = { version = "1.49", default-features = false, features = [
   "time",
 ] }
 tokio-metrics = "0.4"
+tokio-rustls = { version = "0.26", default-features = false }
 tracing = "0.1.44"
 tracing-subscriber = "0.3"
 url = "2.5"

eng/dict/crates.txt

@@ -54,11 +54,13 @@ fe2o3_amqp_cbs
 fe2o3_amqp_ext
 fe2o3_amqp_management
 fe2o3_amqp_types
+fe2o3_amqp_ws
 fe2o3-amqp
 fe2o3-amqp-cbs
 fe2o3-amqp-ext
 fe2o3-amqp-management
 fe2o3-amqp-types
+fe2o3-amqp-ws
 flate2
 futures
 getrandom
@@ -113,6 +115,8 @@ time
 tokio
 tokio_metrics
 tokio-metrics
+tokio_tungstenite
+tokio-tungstenite
 tracing
 tracing_subscriber
 tracing-subscriber

sdk/core/azure_core_amqp/CHANGELOG.md

@@ -4,15 +4,24 @@
 
 ### Features Added
 
+- Added `AmqpTransport` and an `AmqpConnectionOptions::transport` field to select the connection transport. `AmqpTransport::WebSocket` tunnels AMQP over secure WebSockets (`wss://`, port 443) for networks that block the native AMQP ports.
+- Added the `fe2o3_amqp_ws` and `fe2o3_amqp_ws_rustls` features. `fe2o3_amqp_ws` is the base feature and turns on the WebSocket transport code, and `fe2o3_amqp_ws_rustls` adds the TLS stack that the rest of `sdk/core` uses, rustls with the aws-lc-rs provider. This is the shape that `azure_core` uses for HTTP, where `reqwest` is the base and `reqwest_rustls` adds the stack. The `default` feature selects both. To build the transport on another TLS stack, turn off the default features, name `fe2o3_amqp_ws`, and take a direct dependency on `fe2o3-amqp-ws` with the stack you want; Cargo unifies the features. One stack must be selected somewhere in the graph, and a build that selects none still compiles and reports the missing stack when the connection opens. A build without `fe2o3_amqp_ws` still accepts `AmqpTransport::WebSocket`, and the connection then returns an error when it opens.
+- Added the `fe2o3_amqp_rustls` feature, which adds the TLS stack for AMQP framed directly on TCP (`amqps://`, port 5671) on top of the `fe2o3_amqp` base feature. It is rustls with the aws-lc-rs provider, the stack that the rest of `sdk/core` uses, and `default` selects it in place of the `fe2o3-amqp/native-tls` entry that `default` named before. The connection supplies a TLS connector built on `rustls-platform-verifier`, so the handshake reads the trust store of the operating system and trusts the same roots as `reqwest/rustls` on the HTTP side. The default connector of `fe2o3-amqp` instead fills its root store from `webpki-roots`, a compiled-in copy of the Mozilla root set, which would drop the roots that an operator installs in the operating system. There is no native-tls feature on this crate, because `fe2o3-amqp` accepts one TLS stack only, and two features that cannot both be on would break `--all-features`. ([#4189](https://github.com/Azure/azure-sdk-for-rust/issues/4189))
+
 ### Breaking Changes
 
+- Added the `transport` field to `AmqpConnectionOptions`. The struct is not `#[non_exhaustive]`, so an existing struct literal that names every field no longer compiles. Add `..Default::default()` to the initializer, and put `#[allow(clippy::needless_update)]` on it. The workspace allows the `constructible_struct_adds_field` semver lint for this reason, so `cargo semver-checks` does not report the addition.
+- The `default` feature now selects `fe2o3_amqp_rustls`, so AMQP framed directly on TCP (`amqps://`, port 5671) runs on rustls with the aws-lc-rs provider where it ran on native-tls. Both stacks read the trust store of the operating system, so a broker behind a private or an enterprise certificate authority keeps working. The stacks read that store through different platform APIs, and a deployment that tunes native-tls directly, such as one that sets OpenSSL environment variables, can still see a difference. To keep native-tls, turn off the default features, name `fe2o3_amqp`, and take a direct dependency on `fe2o3-amqp` with its `native-tls` feature. ([#4189](https://github.com/Azure/azure-sdk-for-rust/issues/4189))
+
 ### Bugs Fixed
 
 - Link properties set through `AmqpReceiverOptions::properties` and `AmqpSenderOptions::properties` now reach the Attach frame. They were discarded before the link attached.
 - A failed receive now reports the link-state kind the sender path already reports: `AmqpDescribedError` when the remote closed or detached with an AMQP error, and `LinkClosedByRemote` or `LinkDetachedByRemote` otherwise. All of these previously reported `LinkStateError`.
 
 ### Other Changes
 
+- Updated the `fe2o3-amqp` family of dependencies from 0.14 to 0.16. The rustls backend of 0.16 is built on aws-lc-rs, where 0.14 was built on `ring`, which `deny.toml` bans. This is what makes `fe2o3_amqp_rustls` possible. The update needed no source change.
+
 ## 1.1.0 (2026-07-09)
 
 ### Features Added

sdk/core/azure_core_amqp/Cargo.toml

@@ -21,14 +21,18 @@ edition.workspace = true
 async-trait.workspace = true
 azure_core = { path = "../azure_core", version = "1.2.0-beta.1", default-features = false }
 fe2o3-amqp = { workspace = true, optional = true }
+fe2o3-amqp-ws = { workspace = true, optional = true }
 fe2o3-amqp-cbs = { workspace = true, optional = true }
 fe2o3-amqp-ext = { workspace = true, optional = true }
 fe2o3-amqp-management = { workspace = true, optional = true }
 fe2o3-amqp-types = { workspace = true, optional = true }
+rustls = { workspace = true, optional = true }
+rustls-platform-verifier = { workspace = true, optional = true }
 serde.workspace = true
 serde_amqp = { workspace = true, optional = true }
 serde_bytes = { workspace = true, optional = true }
 tokio.workspace = true
+tokio-rustls = { workspace = true, optional = true }
 tracing.workspace = true
 typespec = { path = "../typespec", version = "1.2.0-beta.1" }
 typespec_macros = { path = "../typespec_macros", version = "1.1.0-beta.1" }
@@ -38,7 +42,12 @@ serde_json.workspace = true
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 
 [features]
-default = ["fe2o3_amqp", "fe2o3-amqp/native-tls"]
+default = [
+  "fe2o3_amqp",
+  "fe2o3_amqp_rustls",
+  "fe2o3_amqp_ws",
+  "fe2o3_amqp_ws_rustls",
+]
 ffi = []
 test = []
 fe2o3_amqp = [
@@ -51,9 +60,100 @@ fe2o3_amqp = [
   "serde_bytes",
   "azure_core/tokio",
 ]
+# The TLS stack for AMQP framed directly on TCP (`amqps://`, port 5671).
+# `fe2o3_amqp` is the base, and this feature adds the stack, in the same shape
+# that `azure_core` uses for HTTP, where `reqwest` is the base and
+# `reqwest_rustls` adds the stack. It is rustls with the aws-lc-rs provider, and
+# `default` selects it.
+#
+# The connection builds its own TLS connector on `rustls-platform-verifier`.
+# The default connector of `fe2o3-amqp` fills its root store from
+# `webpki-roots`, which holds a compiled-in copy of the Mozilla root set and
+# ignores the trust store of the operating system. The platform verifier reads
+# the trust store of the operating system, so this feature trusts the same roots
+# as `reqwest/rustls` on the HTTP side, and a broker behind a private or an
+# enterprise certificate authority keeps working. `webpki-roots` still arrives
+# as a dependency, because the `rustls` feature of `fe2o3-amqp` always names it,
+# but nothing reads it.
+#
+# The direct `rustls` dependency selects the crypto provider.
+# `ClientConfig::builder()` panics when no process-level default is installed
+# and the crate features name no single provider. The default features of
+# `rustls` give aws-lc-rs, std, and tls12, and `deny.toml` bans `ring`, so this
+# repository cannot reach that panic. An application that unifies a second
+# provider into the graph makes the choice ambiguous again, and it must call
+# `CryptoProvider::install_default()` before it opens a connection. The
+# WebSocket feature below carries the same condition.
+#
+# This feature needs `fe2o3-amqp` 0.16 or later. The rustls backend of 0.14 was
+# built on `ring`, which `deny.toml` bans (#4189).
+#
+# There is no matching native-tls feature, because `fe2o3-amqp` accepts one TLS
+# stack only and reports `TlsConnectorNotFound` at run time when both are on.
+# Two features that cannot both be on would break `--all-features`. To use
+# another stack, turn off the default features, name `fe2o3_amqp`, and take a
+# direct dependency on `fe2o3-amqp` with the stack you want. Cargo unifies the
+# features, and nothing pulls rustls in:
+#
+#   azure_core_amqp = { version = "...", default-features = false, features = [
+#     "fe2o3_amqp",
+#   ] }
+#   fe2o3-amqp = { version = "0.16", features = ["native-tls"] }
+#
+# `samples/list_blobs_native_tls` shows the same pattern for `reqwest`.
+fe2o3_amqp_rustls = [
+  "fe2o3_amqp",
+  "fe2o3-amqp/rustls",
+  "dep:rustls",
+  "dep:rustls-platform-verifier",
+  "dep:tokio-rustls",
+]
+# AMQP over WebSockets. `fe2o3_amqp_ws` is the base feature: it turns on the
+# transport code and the `fe2o3-amqp-ws` dependency, and it names no TLS stack.
+# `fe2o3_amqp_ws_rustls` adds the stack that the rest of `sdk/core` uses, rustls
+# with the aws-lc-rs provider, and `default` selects it.
+#
+# To build the transport on another TLS stack, turn off the default features,
+# name `fe2o3_amqp_ws`, and take a direct dependency on the backend crate with
+# the stack you want. Cargo unifies the features, so the transport then uses
+# your stack and nothing pulls rustls in:
+#
+#   azure_core_amqp = { version = "...", default-features = false, features = [
+#     "fe2o3_amqp", "fe2o3_amqp_ws",
+#   ] }
+#   fe2o3-amqp = { version = "0.16", features = ["native-tls"] }
+#   fe2o3-amqp-ws = { version = "0.16", features = ["native-tls"] }
+#
+# `samples/list_blobs_native_tls` shows the same pattern for `reqwest`.
+#
+# One stack must be selected somewhere in the graph. `fe2o3_amqp_ws` on its own
+# builds, and the connection reports `TlsFeatureNotEnabled` when it opens.
+# `reqwest` behaves the same way.
+fe2o3_amqp_ws = ["fe2o3_amqp", "dep:fe2o3-amqp-ws"]
+# The direct `rustls` dependency is only here to select the crypto provider.
+# This crate names no `rustls` type. `reqwest/rustls` selects aws-lc-rs itself,
+# so `typespec_client_core` needs no such dependency. The `tokio-tungstenite`
+# chain under `fe2o3-amqp-ws` is different. It takes rustls with
+# `default-features = false` and names no provider, so `ClientConfig::builder()`
+# panics when no process-level default is installed. The default features of
+# `rustls` give aws-lc-rs, std, and tls12.
+#
+# `fe2o3_amqp_rustls` selects the same provider, so the dependency is redundant
+# when `default` is on. This feature has to stand on its own, because a consumer
+# can turn the default features off and name only the WebSocket transport.
+fe2o3_amqp_ws_rustls = [
+  "fe2o3_amqp_ws",
+  "fe2o3-amqp-ws/rustls-tls-native-roots",
+  "dep:rustls",
+]
 
 [lints]
 workspace = true
 
 [package.metadata.docs.rs]
-features = ["fe2o3_amqp"]
+features = [
+  "fe2o3_amqp",
+  "fe2o3_amqp_rustls",
+  "fe2o3_amqp_ws",
+  "fe2o3_amqp_ws_rustls",
+]

sdk/core/azure_core_amqp/src/connection.rs

@@ -14,7 +14,46 @@ type ConnectionImplementation = super::fe2o3::connection::Fe2o3AmqpConnection;
 #[cfg(not(feature = "fe2o3_amqp"))]
 type ConnectionImplementation = super::noop::NoopAmqpConnection;
 
+/// The transport used to carry the AMQP protocol.
+///
+/// AMQP is normally framed directly over a TCP/TLS socket, but some network
+/// environments (for example corporate firewalls) only permit outbound
+/// connections on port 443. In those cases the AMQP frames can be tunneled
+/// over a WebSocket connection instead.
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum AmqpTransport {
+    /// AMQP framing over a TCP/TLS socket (port 5671). This is the default.
+    #[default]
+    Tcp,
+    /// AMQP framing tunneled over secure WebSockets (`wss://`, port 443).
+    ///
+    /// This variant needs the `fe2o3_amqp_ws` feature, which the `default`
+    /// feature selects. A build without it can still select this variant, but
+    /// the connection then returns an error when it opens.
+    ///
+    /// `fe2o3_amqp_ws` names no TLS stack. `default` adds
+    /// `fe2o3_amqp_ws_rustls`, which selects rustls with the aws-lc-rs
+    /// provider. To use another stack, turn off the default features, name
+    /// `fe2o3_amqp_ws`, and take a direct dependency on `fe2o3-amqp-ws` with
+    /// the stack you want. A build that selects no stack also compiles, and
+    /// the connection reports the missing stack when it opens.
+    WebSocket,
+}
+
 /// Options for configuring an AMQP connection.
+///
+/// Build it from [`Default`] and set only the fields you need, so a later field
+/// addition does not break the call site:
+///
+/// ```
+/// use azure_core_amqp::{AmqpConnectionOptions, AmqpTransport};
+///
+/// #[allow(clippy::needless_update)]
+/// let options = AmqpConnectionOptions {
+///     transport: Some(AmqpTransport::WebSocket),
+///     ..Default::default()
+/// };
+/// ```
 #[derive(Debug, Default, Clone)]
 pub struct AmqpConnectionOptions {
     /// Maximum frame size for the connection in bytes.
@@ -36,11 +75,17 @@ pub struct AmqpConnectionOptions {
     /// Buffer size for the connection.
     pub buffer_size: Option<usize>,
     /// Custom endpoint for the connection. Used to connect to a local AMQP proxy server.
+    ///
+    /// The host and an explicit port both carry into the address that the
+    /// connection dials. Under [`AmqpTransport::WebSocket`] the port carries into
+    /// the `wss://` address, so name the port that the proxy accepts WebSockets
+    /// on, and leave the port out to dial the default port 443. The .NET Azure SDK
+    /// treats `CustomEndpointAddress` the same way.
     pub custom_endpoint: Option<Url>,
+    /// The transport used to carry the AMQP protocol. Defaults to [`AmqpTransport::Tcp`].
+    pub transport: Option<AmqpTransport>,
 }
 
-impl AmqpConnectionOptions {}
-
 /// Trait defining the asynchronous APIs for AMQP connection operations.
 #[async_trait::async_trait]
 pub trait AmqpConnectionApis {
@@ -250,6 +295,7 @@ mod tests {
                     .collect(),
             ),
             buffer_size: Some(1024),
+            transport: Some(AmqpTransport::WebSocket),
         };
 
         assert_eq!(connection_options.max_frame_size, Some(1024));
@@ -281,6 +327,7 @@ mod tests {
             connection_options.custom_endpoint,
             Some(Url::parse("http://localhost:8080").unwrap())
         );
+        assert_eq!(connection_options.transport, Some(AmqpTransport::WebSocket));
     }
 
     // On macOS, there is a periodic issue where loopback TCP connections fail.