Function handle_upgrade

crates/openshell-supervisor-network/src/l7/relay.rs:591–677

Handle an upgraded connection (101 Switching Protocols). Forwards any overflow bytes from the upgrade response to the client, then either switches to a parsed WebSocket relay for opted-in message policy / credential rewriting or to raw bidirectional TCP copy for other upgrades.

(
    client: &mut C,
    upstream: &mut U,
    overflow: Vec<u8>,
    host: &str,
    port: u16,
    options: UpgradeRelayOptions<'_>,
)

/// either switches to a parsed WebSocket relay for opted-in message policy /
/// credential rewriting or to raw bidirectional TCP copy for other upgrades.
pub(crate) async fn handle_upgrade<C, U>(
    client: &mut C,
    upstream: &mut U,
    overflow: Vec<u8>,
    host: &str,
    port: u16,
    options: UpgradeRelayOptions<'_>,
) -> Result<()>
where
    C: AsyncRead + AsyncWrite + Unpin + Send,
    U: AsyncRead + AsyncWrite + Unpin + Send,
{
    let use_websocket_relay = options.websocket_request
        && (options.websocket.message_policy.inspects_messages()
            || options.websocket.permessage_deflate
            || (options.websocket.credential_rewrite && options.secret_resolver.is_some()));
    let relay_mode = if use_websocket_relay {
        "websocket parsed relay"
    } else {
        "raw bidirectional relay (L7 enforcement no longer active)"
    };
    ocsf_emit!(
        NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx())
            .activity(ActivityId::Other)
            .activity_name("Upgrade")
            .severity(SeverityId::Informational)
            .dst_endpoint(Endpoint::from_domain(host, port))
            .message(format!(
                "101 Switching Protocols — {relay_mode} [host:{host} port:{port} overflow_bytes:{}]",
                overflow.len()
            ))
            .build()
    );
    if use_websocket_relay {
        let resolver = if options.websocket.credential_rewrite {
            options.secret_resolver.as_deref()
        } else {
            None
        };
        let inspector = if options.websocket.message_policy.inspects_messages() {
            match (options.engine, options.ctx) {
                (Some(engine), Some(ctx)) => Some(crate::l7::websocket::InspectionOptions {
                    engine,
                    ctx,
                    enforcement: options.enforcement,
                    target: options.target.clone(),
                    query_params: options.query_params.clone(),
                    graphql_policy: options.websocket.message_policy.is_graphql(),
                }),
                _ => {
                    return Err(miette!(
                        "websocket message inspection missing policy context"
                    ));
                }
            }
        } else {
            None
        };