feat(realtime): /v1/realtime WebSocket relay (openai-family + azure, browser subprotocol auth, session usage)

Cargo.toml

@@ -33,6 +33,7 @@ authors = ["moonming"]
 # Runtime
 tokio = { version = "1.41", features = ["full"] }
 futures = "0.3"
+tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] }
 async-trait = "0.1"
 
 # HTTP server

crates/aisix-proxy/Cargo.toml

@@ -42,6 +42,7 @@ bytes.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 futures.workspace = true
+tokio-tungstenite.workspace = true
 async-trait.workspace = true
 async-stream = "0.3"
 dashmap.workspace = true

crates/aisix-proxy/src/auth.rs

@@ -38,30 +38,41 @@ where
     async fn from_request_parts(parts: &mut Parts, state: &S) -> Result<Self, Self::Rejection> {
         let token = extract_bearer(parts)?;
         let proxy_state = ProxyState::from_ref(state);
-        let snapshot = proxy_state.snapshot.load();
-        // Self-hosted CP (prd-09a §9A.7B.4): the snapshot stores
-        // SHA-256 hashes of the plaintext bearer, never the plaintext
-        // itself. Hash the incoming token via the canonical helper
-        // and look up by the hex digest. cp-api hashes with the same
-        // function before persistence, so the two sides agree byte
-        // for byte.
-        let entry = snapshot
-            .apikeys
-            .get_by_name(&ApiKey::hash_bearer(&token))
-            .ok_or(ProxyError::InvalidApiKey)?;
-        // Lifecycle enforcement (#933): a known key that is disabled or
-        // past its expiry deadline must be rejected here, at the single
-        // auth choke point, so every proxy surface (chat, messages,
-        // responses, embeddings, audio, passthrough, MCP, …) inherits
-        // the same 401 without per-handler checks.
-        if entry.value.disabled {
-            return Err(ProxyError::ApiKeyDisabled);
-        }
-        if entry.value.is_expired_at(chrono::Utc::now()) {
-            return Err(ProxyError::ApiKeyExpired);
-        }
-        Ok(AuthenticatedKey { entry })
+        authenticate_token(&proxy_state, &token)
+    }
+}
+
+/// Look a plaintext bearer up in the snapshot and enforce key lifecycle.
+/// The single auth choke point behind the [`AuthenticatedKey`] extractor;
+/// also called directly by surfaces whose credentials arrive outside the
+/// standard headers (WebSocket subprotocol auth on `/v1/realtime`).
+pub(crate) fn authenticate_token(
+    state: &ProxyState,
+    token: &str,
+) -> Result<AuthenticatedKey, ProxyError> {
+    let snapshot = state.snapshot.load();
+    // Self-hosted CP (prd-09a §9A.7B.4): the snapshot stores
+    // SHA-256 hashes of the plaintext bearer, never the plaintext
+    // itself. Hash the incoming token via the canonical helper
+    // and look up by the hex digest. cp-api hashes with the same
+    // function before persistence, so the two sides agree byte
+    // for byte.
+    let entry = snapshot
+        .apikeys
+        .get_by_name(&ApiKey::hash_bearer(token))
+        .ok_or(ProxyError::InvalidApiKey)?;
+    // Lifecycle enforcement (#933): a known key that is disabled or
+    // past its expiry deadline must be rejected here, at the single
+    // auth choke point, so every proxy surface (chat, messages,
+    // responses, embeddings, audio, passthrough, MCP, …) inherits
+    // the same 401 without per-handler checks.
+    if entry.value.disabled {
+        return Err(ProxyError::ApiKeyDisabled);
+    }
+    if entry.value.is_expired_at(chrono::Utc::now()) {
+        return Err(ProxyError::ApiKeyExpired);
     }
+    Ok(AuthenticatedKey { entry })
 }
 
 fn extract_bearer(parts: &Parts) -> Result<String, ProxyError> {

crates/aisix-proxy/src/lib.rs

@@ -50,6 +50,7 @@ mod model_resolve;
 mod models;
 mod passthrough;
 mod quota;
+mod realtime;
 mod redact;
 mod render;
 mod request_id;
@@ -106,6 +107,9 @@ pub fn build_router(state: ProxyState) -> Router {
         .route("/v1/audio/transcriptions", post(audio::transcriptions))
         .route("/v1/audio/translations", post(audio::translations))
         .route("/v1/audio/speech", post(audio::speech))
+        // OpenAI Realtime WebSocket relay (#721). Auth/ACL/quota are
+        // enforced pre-upgrade inside the handler.
+        .route("/v1/realtime", get(realtime::realtime))
         // Files / Batches / Fine-tuning jobs surface (#720). Provider
         // routing rides the gateway-encoded resource ids; see jobs.rs.
         .route(
@@ -223,6 +227,7 @@ fn normalize_endpoint_label(path: &str) -> &'static str {
         "/v1/audio/translations" => "/v1/audio/translations",
         "/v1/audio/speech" => "/v1/audio/speech",
         "/mcp" | "/mcp/" => "/mcp",
+        "/v1/realtime" => "/v1/realtime",
         "/v1/files" => "/v1/files",
         "/v1/batches" => "/v1/batches",
         "/v1/fine_tuning/jobs" => "/v1/fine_tuning/jobs",
@@ -242,6 +247,8 @@ fn inbound_protocol_for_endpoint(endpoint: &str) -> &'static str {
         "mcp"
     } else if endpoint == "/a2a" {
         "a2a"
+    } else if endpoint == "/v1/realtime" {
+        "realtime"
     } else {
         "openai"
     }