Function init_ws

src/environmentd/src/http.rs:1264–1380

(
    WsState {
        frontegg,
        oidc_rx,
        authenticator_kind,
        adapter_client_rx,
        active_connection_counter,
        helm_chart_version,
        allowed_roles,
    }: W

}

async fn init_ws(
    WsState {
        frontegg,
        oidc_rx,
        authenticator_kind,
        adapter_client_rx,
        active_connection_counter,
        helm_chart_version,
        allowed_roles,
    }: WsState,
    existing_user: Option<ExistingUser>,
    peer_addr: IpAddr,
    ws: &mut WebSocket,
) -> Result<AuthedClient, anyhow::Error> {
    // TODO: Add a timeout here to prevent resource leaks by clients that
    // connect then never send a message.
    let ws_auth: WebSocketAuth = loop {
        let init_msg = ws.recv().await.ok_or_else(|| anyhow::anyhow!("closed"))??;
        match init_msg {
            Message::Text(data) => break serde_json::from_str(&data)?,
            Message::Binary(data) => break serde_json::from_slice(&data)?,
            // Handled automatically by the server.
            Message::Ping(_) => {
                continue;
            }
            Message::Pong(_) => {
                continue;
            }
            Message::Close(_) => {
                anyhow::bail!("closed");
            }
        }
    };

    // If credentials are provided, we perform a new authentication,
    // separate from the existing session.
    let (creds, options) = match ws_auth {
        WebSocketAuth::Basic {
            user,
            password,
            options,
        } => {
            let creds = Credentials::Password {
                username: user,
                password,
            };
            (Some(creds), options)
        }
        WebSocketAuth::Bearer { token, options } => {
            let creds = Credentials::Token { token };
            (Some(creds), options)
        }
        WebSocketAuth::OptionsOnly { options } => (None, options),
    };

    let user = match (existing_user, creds) {
        (Some(ExistingUser::XMaterializeUserHeader(_)), Some(_creds)) => {
            warn!("Unexpected bearer or basic auth provided when using user header");