futu_backend/auth/

webtcp.rs

//! WebTCP-short request path for broker auth.
//!
//! C++ FTLogin sends broker-auth HTTP payloads through a short-lived TLS
//! channel before falling back to ordinary HTTP. The wire command is 65507 and
//! the protobuf body is an internal `TcpHttpRequest`.

use crate::conn::{BackendConn, PushCallback};
use futu_core::error::{FutuError, Result};
use rustls_pki_types::ServerName;
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::TcpSocket;
use tokio_rustls::TlsConnector;
use tokio_rustls::rustls::{ClientConfig, RootCertStore};

const WEB_REQUEST_CMD: u16 = 65507;
/// Ref: `FTLogin/Src/ftlogin/objc/F3CDefine.h:921-922`
/// (`F3CLIHttpConfig` connect/total timeout defaults are 10s).
const WEBTCP_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
/// Ref: `FTLogin/Src/ftlogin/objc/F3CDefine.h:921-922`.
/// Rust splits TCP connect and TLS handshake but keeps C++'s 10s WebTCP
/// connection-phase bound for both steps.
const WEBTCP_TLS_TIMEOUT: Duration = Duration::from_secs(10);
/// Ref: `FTBasis/Src/ftbasis/protocol/tcp/impl/tcp_client_basic.cpp:330-342`.
const WEBTCP_SEND_BUFFER_SIZE: u32 = 64 * 1024;
/// Ref: `FTBasis/Src/ftbasis/protocol/tcp/impl/tcp_client_basic.cpp:330-342`.
const WEBTCP_RECV_BUFFER_SIZE: u32 = 512 * 1024;
static RUSTLS_PROVIDER_INIT: std::sync::Once = std::sync::Once::new();

/// rustls 0.23 requires an explicit process-level crypto provider when more
/// than one provider feature is present. WebTCP uses rustls directly, so do
/// this before the first `ClientConfig::builder()` call.
pub fn install_default_rustls_crypto_provider() {
    RUSTLS_PROVIDER_INIT.call_once(|| {
        if tokio_rustls::rustls::crypto::ring::default_provider()
            .install_default()
            .is_err()
        {
            tracing::debug!(
                "rustls crypto provider already installed; keeping existing process-level provider"
            );
        }
    });
}

#[derive(Debug)]
pub(super) struct ProtoError(String);

impl std::fmt::Display for ProtoError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(f)
    }
}

impl std::error::Error for ProtoError {}

pub(super) type ProtoResult<T> = std::result::Result<T, ProtoError>;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum WireType {
    Varint = 0,
    Fixed64 = 1,
    LengthDelimited = 2,
    Fixed32 = 5,
}

impl WireType {
    fn from_u64(value: u64) -> ProtoResult<Self> {
        match value {
            0 => Ok(Self::Varint),
            1 => Ok(Self::Fixed64),
            2 => Ok(Self::LengthDelimited),
            5 => Ok(Self::Fixed32),
            other => Err(ProtoError(format!("unsupported wire type {other}"))),
        }
    }
}

pub(super) struct ProtoWriter {
    buf: Vec<u8>,
}

impl ProtoWriter {
    pub(super) fn new() -> Self {
        Self { buf: Vec::new() }
    }

    pub(super) fn finish(self) -> Vec<u8> {
        self.buf
    }

    pub(super) fn write_key(&mut self, field: u32, wire_type: WireType) {
        self.write_varint(((field as u64) << 3) | wire_type as u64);
    }

    pub(super) fn write_varint(&mut self, mut value: u64) {
        while value >= 0x80 {
            self.buf.push((value as u8) | 0x80);
            value >>= 7;
        }
        self.buf.push(value as u8);
    }

    pub(super) fn write_bytes(&mut self, field: u32, value: &[u8]) {
        self.write_key(field, WireType::LengthDelimited);
        self.write_varint(value.len() as u64);
        self.buf.extend_from_slice(value);
    }

    pub(super) fn write_string(&mut self, field: u32, value: &str) {
        self.write_bytes(field, value.as_bytes());
    }

    pub(super) fn write_message(&mut self, field: u32, value: Vec<u8>) {
        self.write_bytes(field, &value);
    }
}

pub(super) struct ProtoReader<'a> {
    data: &'a [u8],
    pos: usize,
}

impl<'a> ProtoReader<'a> {
    pub(super) fn new(data: &'a [u8]) -> Self {
        Self { data, pos: 0 }
    }

    fn eof(&self) -> bool {
        self.pos >= self.data.len()
    }

    fn read_byte(&mut self) -> ProtoResult<u8> {
        let byte = *self
            .data
            .get(self.pos)
            .ok_or_else(|| ProtoError("unexpected eof".into()))?;
        self.pos += 1;
        Ok(byte)
    }

    pub(super) fn read_varint(&mut self) -> ProtoResult<u64> {
        let mut value = 0u64;
        for shift in (0..64).step_by(7) {
            let byte = self.read_byte()? as u64;
            value |= (byte & 0x7f) << shift;
            if byte & 0x80 == 0 {
                return Ok(value);
            }
        }
        Err(ProtoError("varint too long".into()))
    }

    pub(super) fn next_key(&mut self) -> ProtoResult<Option<(u32, WireType)>> {
        if self.eof() {
            return Ok(None);
        }
        let key = self.read_varint()?;
        Ok(Some(((key >> 3) as u32, WireType::from_u64(key & 0x07)?)))
    }

    pub(super) fn read_bytes(&mut self) -> ProtoResult<&'a [u8]> {
        let len = self.read_varint()? as usize;
        let end = self
            .pos
            .checked_add(len)
            .ok_or_else(|| ProtoError("length overflow".into()))?;
        if end > self.data.len() {
            return Err(ProtoError(format!(
                "length-delimited out of bounds len={len} remaining={}",
                self.data.len().saturating_sub(self.pos)
            )));
        }
        let out = &self.data[self.pos..end];
        self.pos = end;
        Ok(out)
    }

    pub(super) fn read_string(&mut self) -> ProtoResult<String> {
        let bytes = self.read_bytes()?;
        std::str::from_utf8(bytes)
            .map(|s| s.to_string())
            .map_err(|e| ProtoError(format!("utf8: {e}")))
    }

    pub(super) fn read_i32(&mut self) -> ProtoResult<i32> {
        Ok(self.read_varint()? as i32)
    }

    fn read_bool(&mut self) -> ProtoResult<bool> {
        Ok(self.read_varint()? != 0)
    }

    pub(super) fn skip_field(&mut self, wire_type: WireType) -> ProtoResult<()> {
        match wire_type {
            WireType::Varint => {
                self.read_varint()?;
            }
            WireType::Fixed64 => {
                self.skip_bytes(8)?;
            }
            WireType::LengthDelimited => {
                let len = self.read_varint()? as usize;
                self.skip_bytes(len)?;
            }
            WireType::Fixed32 => {
                self.skip_bytes(4)?;
            }
        }
        Ok(())
    }

    fn skip_bytes(&mut self, len: usize) -> ProtoResult<()> {
        let end = self
            .pos
            .checked_add(len)
            .ok_or_else(|| ProtoError("skip overflow".into()))?;
        if end > self.data.len() {
            return Err(ProtoError("skip out of bounds".into()));
        }
        self.pos = end;
        Ok(())
    }
}

#[derive(Debug, Clone)]
struct TcpHttpHeader {
    key: String,
    value: String,
}

impl TcpHttpHeader {
    fn encode(&self) -> Vec<u8> {
        let mut writer = ProtoWriter::new();
        writer.write_string(1, &self.key);
        writer.write_string(2, &self.value);
        writer.finish()
    }
}

#[derive(Debug, Clone)]
struct TcpHttpRequest {
    method: String,
    url: String,
    headers: Vec<TcpHttpHeader>,
    body: Vec<u8>,
}

impl TcpHttpRequest {
    fn encode(&self) -> Vec<u8> {
        let mut writer = ProtoWriter::new();
        writer.write_string(1, &self.method);
        writer.write_string(2, &self.url);
        for header in &self.headers {
            writer.write_message(3, header.encode());
        }
        writer.write_bytes(4, &self.body);
        writer.finish()
    }
}

#[derive(Debug, Clone)]
struct TcpHttpResponseBody {
    status_code: Option<i32>,
    message: Option<String>,
    data: Option<Vec<u8>>,
}

impl TcpHttpResponseBody {
    fn decode(bytes: &[u8]) -> ProtoResult<Self> {
        let mut reader = ProtoReader::new(bytes);
        let mut status_code = None;
        let mut message = None;
        let mut data = None;
        while let Some((field, wire_type)) = reader.next_key()? {
            match (field, wire_type) {
                (1, WireType::Varint) => status_code = Some(reader.read_i32()?),
                (2, WireType::LengthDelimited) => message = Some(reader.read_string()?),
                (5, WireType::LengthDelimited) => data = Some(reader.read_bytes()?.to_vec()),
                _ => reader.skip_field(wire_type)?,
            }
        }
        Ok(Self {
            status_code,
            message,
            data,
        })
    }
}

#[derive(Debug, Clone)]
struct TcpHttpResponse {
    code: Option<i32>,
    message: Option<String>,
    response_body: Option<TcpHttpResponseBody>,
    current_need_fallback: Option<bool>,
}

impl TcpHttpResponse {
    fn decode(bytes: &[u8]) -> ProtoResult<Self> {
        let mut reader = ProtoReader::new(bytes);
        let mut code = None;
        let mut message = None;
        let mut response_body = None;
        let mut current_need_fallback = None;
        while let Some((field, wire_type)) = reader.next_key()? {
            match (field, wire_type) {
                (1, WireType::Varint) => code = Some(reader.read_i32()?),
                (2, WireType::LengthDelimited) => message = Some(reader.read_string()?),
                (3, WireType::LengthDelimited) => {
                    response_body = Some(TcpHttpResponseBody::decode(reader.read_bytes()?)?)
                }
                (4, WireType::Varint) => current_need_fallback = Some(reader.read_bool()?),
                _ => reader.skip_field(wire_type)?,
            }
        }
        Ok(Self {
            code,
            message,
            response_body,
            current_need_fallback,
        })
    }
}

/// C++ uses wildcard certificate domains; rustls needs a concrete certificate
/// verification name, but WebTCP's C++ TLS path does not set SNI.
pub(super) fn tls_server_name_for_web_identity(identity: u32) -> &'static str {
    match identity {
        crate::auth::commconfig::CONN_WEB_CN | crate::auth::commconfig::CONN_WEB_HK => {
            "www.futunn.com"
        }
        _ => "www.moomoo.com",
    }
}

fn tls_connector() -> TlsConnector {
    install_default_rustls_crypto_provider();

    let roots = RootCertStore {
        roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(),
    };
    let mut config = ClientConfig::builder()
        .with_root_certificates(roots)
        .with_no_client_auth();
    // Ref: `FTBasis/Src/ftbasis/protocol/tcp/impl/tcp_client_basic.cpp:58-64`.
    // C++ sets RFC2818 certificate verification on `ssl_certificate_domain`
    // but has no `SSL_set_tlsext_host_name` call on the WebTCP TLS path.
    config.enable_sni = false;
    TlsConnector::from(Arc::new(config))
}

fn proto_err(context: &str, err: ProtoError) -> FutuError {
    FutuError::Codec(format!("webtcp {context}: {err}"))
}

#[derive(Debug)]
pub(crate) struct WebTcpRequestError {
    error: FutuError,
    http_fallback_allowed: bool,
}

impl WebTcpRequestError {
    fn fallback_allowed(error: FutuError) -> Self {
        Self {
            error,
            http_fallback_allowed: true,
        }
    }

    fn fallback_forbidden(error: FutuError) -> Self {
        Self {
            error,
            http_fallback_allowed: false,
        }
    }

    pub(crate) fn allows_http_fallback(&self) -> bool {
        self.http_fallback_allowed
    }

    pub(crate) fn into_futu_error(self) -> FutuError {
        self.error
    }
}

impl std::fmt::Display for WebTcpRequestError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.error.fmt(f)
    }
}

impl std::error::Error for WebTcpRequestError {}

fn decode_response_json(
    frame_body: &[u8],
) -> std::result::Result<serde_json::Value, WebTcpRequestError> {
    let response = TcpHttpResponse::decode(frame_body)
        .map_err(|e| WebTcpRequestError::fallback_forbidden(proto_err("decode", e)))?;
    if response.code.unwrap_or(-1) != 0 {
        let error = FutuError::Codec(format!(
            "webtcp response code={} message={:?} fallback={:?}",
            response.code.unwrap_or(-1),
            response.message,
            response.current_need_fallback
        ));
        return Err(if response.current_need_fallback.unwrap_or(false) {
            WebTcpRequestError::fallback_allowed(error)
        } else {
            WebTcpRequestError::fallback_forbidden(error)
        });
    }
    if response.current_need_fallback.unwrap_or(false) {
        return Err(WebTcpRequestError::fallback_allowed(FutuError::Codec(
            format!(
                "webtcp response requested fallback message={:?}",
                response.message
            ),
        )));
    }
    let response_body = response.response_body.ok_or_else(|| {
        WebTcpRequestError::fallback_forbidden(FutuError::Codec(
            "webtcp missing response_body".into(),
        ))
    })?;
    let status_code = response_body.status_code.unwrap_or(0);
    if !(200..300).contains(&status_code) {
        return Err(WebTcpRequestError::fallback_forbidden(FutuError::Codec(
            format!(
                "webtcp http status={status_code} message={:?}",
                response_body.message
            ),
        )));
    }
    let data = response_body.data.ok_or_else(|| {
        WebTcpRequestError::fallback_forbidden(FutuError::Codec(
            "webtcp missing http response data".into(),
        ))
    })?;
    serde_json::from_slice(&data).map_err(|e| {
        WebTcpRequestError::fallback_forbidden(FutuError::Codec(format!("webtcp json decode: {e}")))
    })
}

/// C++ `FTAuthImpl::InitRequest` builds one request object for WebTCP and HTTP,
/// then `SetHttpHeaders` adds only `Content-Type`. `SendRequestImpl` copies
/// OpenD's User-Agent from common params, inserts B3 trace headers, and
/// overwrites `Cookie: device_id=...` for every auth request just before
/// dispatch.
/// Ref:
/// - `FTLogin/Src/ftlogin/auth/impl/auth_impl.cpp:3416-3423`
/// - `FTLogin/Src/ftlogin/auth/impl/auth_impl.cpp:3596-3612`
/// - `FTLogin/Src/ftlogin/auth/impl/auth_impl.cpp:3772-3775`
fn cpp_auth_headers(client_type: u8, host: String, device_id: &str) -> Vec<TcpHttpHeader> {
    let trace_headers = super::http_client::AuthTraceHeaders::new();
    let mut headers = vec![
        TcpHttpHeader {
            key: "Host".to_string(),
            value: host,
        },
        TcpHttpHeader {
            key: "Content-Type".to_string(),
            value: "application/json".to_string(),
        },
        TcpHttpHeader {
            key: "User-Agent".to_string(),
            value: super::http_client::opend_auth_user_agent(client_type),
        },
        TcpHttpHeader {
            key: "Cookie".to_string(),
            value: format!("device_id={device_id}"),
        },
    ];
    headers.extend(
        trace_headers
            .entries()
            .into_iter()
            .map(|(key, value)| TcpHttpHeader {
                key: key.to_string(),
                value: value.to_string(),
            }),
    );
    headers
}

fn webtcp_url_fingerprint(url: &str) -> (String, String, Vec<String>) {
    match reqwest::Url::parse(url) {
        Ok(parsed) => (
            parsed.host_str().unwrap_or("").to_string(),
            parsed.path().to_string(),
            parsed
                .query_pairs()
                .map(|(key, _)| key.into_owned())
                .collect(),
        ),
        Err(_) => (
            "<parse-error>".to_string(),
            "<parse-error>".to_string(),
            Vec::new(),
        ),
    }
}

fn webtcp_json_value_shape(value: &serde_json::Value) -> &'static str {
    match value {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(_) => "bool",
        serde_json::Value::Number(_) => "number",
        serde_json::Value::String(_) => "string",
        serde_json::Value::Array(_) => "array",
        serde_json::Value::Object(_) => "object",
    }
}

fn webtcp_body_field_shapes(body: Option<&[u8]>) -> Vec<String> {
    body.and_then(|body| serde_json::from_slice::<serde_json::Value>(body).ok())
        .and_then(|value| {
            value.as_object().map(|object| {
                object
                    .iter()
                    .map(|(key, value)| {
                        let suffix = value
                            .as_str()
                            .map(|value| format!(":string_len={}", value.len()))
                            .unwrap_or_else(|| format!(":{}", webtcp_json_value_shape(value)));
                        format!("{key}{suffix}")
                    })
                    .collect()
            })
        })
        .unwrap_or_default()
}

fn log_webtcp_request_fingerprint(
    method: &str,
    url: &str,
    headers: &[TcpHttpHeader],
    body: Option<&[u8]>,
) {
    let (url_host, url_path, query_keys) = webtcp_url_fingerprint(url);
    let header_keys: Vec<&str> = headers.iter().map(|header| header.key.as_str()).collect();
    let x_futu_header_keys: Vec<&str> = header_keys
        .iter()
        .copied()
        .filter(|name| name.to_ascii_lowercase().starts_with("x-futu-"))
        .collect();
    let user_agent = headers
        .iter()
        .find(|header| header.key.eq_ignore_ascii_case("User-Agent"))
        .map(|header| header.value.as_str())
        .unwrap_or("");
    let cookie_len = headers
        .iter()
        .find(|header| header.key.eq_ignore_ascii_case("Cookie"))
        .map(|header| header.value.len())
        .unwrap_or(0);
    let body_fields = webtcp_body_field_shapes(body);

    tracing::info!(
        transport = "webtcp-short",
        method,
        url_host,
        url_path,
        query_keys = ?query_keys,
        header_keys = ?header_keys,
        x_futu_header_keys = ?x_futu_header_keys,
        has_explicit_host_header = headers.iter().any(|header| header.key.eq_ignore_ascii_case("Host")),
        user_agent,
        cookie_len,
        body_fields = ?body_fields,
        "primary auth request fingerprint"
    );
}

fn encode_tcp_http_request(
    client_type: u8,
    method: &str,
    url: &str,
    body: Option<&[u8]>,
    device_id: &str,
) -> Result<Vec<u8>> {
    let parsed = reqwest::Url::parse(url)
        .map_err(|e| FutuError::Codec(format!("webtcp invalid url {url}: {e}")))?;
    let host = parsed
        .host_str()
        .ok_or_else(|| FutuError::Codec(format!("webtcp url missing host: {url}")))?
        .to_string();
    let headers = cpp_auth_headers(client_type, host, device_id);
    log_webtcp_request_fingerprint(method, url, &headers, body);
    Ok(TcpHttpRequest {
        method: method.to_string(),
        url: url.to_string(),
        headers,
        body: body.map_or_else(Vec::new, ToOwned::to_owned),
    }
    .encode())
}

pub(super) async fn connect_webtcp(
    ip: &str,
    port: u16,
    tls_server_name: &'static str,
) -> Result<BackendConn> {
    let ip_addr = ip
        .parse::<IpAddr>()
        .map_err(|e| FutuError::Codec(format!("webtcp invalid ip {ip}: {e}")))?;
    let addr = SocketAddr::new(ip_addr, port);
    let socket = match addr {
        SocketAddr::V4(_) => TcpSocket::new_v4(),
        SocketAddr::V6(_) => TcpSocket::new_v6(),
    }
    .map_err(FutuError::Network)?;
    // Ref: `FTBasis/Src/ftbasis/protocol/tcp/impl/tcp_client_basic.cpp:330-342`.
    // C++ enables TCP_NODELAY and fixed socket buffers before connecting
    // WebTCP sockets. Tokio exposes these portable socket options; TCP_MAXSEG,
    // proxy, and NAT64 require platform/config plumbing and are intentionally
    // not guessed here.
    socket.set_nodelay(true)?;
    socket.set_send_buffer_size(WEBTCP_SEND_BUFFER_SIZE)?;
    socket.set_recv_buffer_size(WEBTCP_RECV_BUFFER_SIZE)?;

    let stream = tokio::time::timeout(WEBTCP_CONNECT_TIMEOUT, socket.connect(addr))
        .await
        .map_err(|elapsed| {
            FutuError::Network(std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                format!("webtcp connect {addr} timed out: {elapsed}"),
            ))
        })?
        .map_err(FutuError::Network)?;

    let server_name = ServerName::try_from(tls_server_name.to_string()).map_err(|e| {
        FutuError::Codec(format!(
            "webtcp invalid tls server name {tls_server_name}: {e}"
        ))
    })?;
    let tls = tokio::time::timeout(
        WEBTCP_TLS_TIMEOUT,
        tls_connector().connect(server_name, stream),
    )
    .await
    .map_err(|elapsed| {
        FutuError::Network(std::io::Error::new(
            std::io::ErrorKind::TimedOut,
            format!("webtcp tls connect {addr} timed out: {elapsed}"),
        ))
    })?
    .map_err(|e| FutuError::Codec(format!("webtcp tls connect {addr}: {e}")))?;

    let noop: PushCallback = Arc::new(|_, _, _| {});
    Ok(BackendConn::from_stream_inner(tls, noop))
}

/// Send an auth HTTP-shaped request through FTLogin WebTCP-short.
pub(crate) async fn request_json_via_webtcp(
    client_type: u8,
    web_identity: u32,
    addrs: &[(String, u16)],
    method: &str,
    url: &str,
    body: Option<&[u8]>,
    device_id: &str,
) -> std::result::Result<serde_json::Value, WebTcpRequestError> {
    if addrs.is_empty() {
        return Err(WebTcpRequestError::fallback_allowed(FutuError::Codec(
            format!("webtcp identity {web_identity} has empty addr pool"),
        )));
    }
    let request_body = encode_tcp_http_request(client_type, method, url, body, device_id)
        .map_err(WebTcpRequestError::fallback_forbidden)?;

    let tls_server_name = tls_server_name_for_web_identity(web_identity);
    let mut last_error: Option<WebTcpRequestError> = None;
    for (ip, port) in addrs {
        tracing::debug!(
            web_identity,
            ip,
            port,
            tls_server_name,
            sni_enabled = false,
            "auth webtcp-short connecting"
        );
        let conn = match connect_webtcp(ip, *port, tls_server_name).await {
            Ok(conn) => conn,
            Err(e) => {
                tracing::warn!(
                    web_identity,
                    ip,
                    port,
                    tls_server_name,
                    sni_enabled = false,
                    error = %e,
                    "auth webtcp-short connect failed; trying next IP"
                );
                last_error = Some(WebTcpRequestError::fallback_allowed(e));
                continue;
            }
        };
        let frame = match conn.request(WEB_REQUEST_CMD, request_body.clone()).await {
            Ok(frame) => frame,
            Err(e) => {
                tracing::warn!(
                    web_identity,
                    ip,
                    port,
                    tls_server_name,
                    sni_enabled = false,
                    error = %e,
                    "auth webtcp-short request failed; trying next IP"
                );
                last_error = Some(WebTcpRequestError::fallback_allowed(e));
                continue;
            }
        };
        return decode_response_json(&frame.body);
    }
    Err(last_error.unwrap_or_else(|| {
        WebTcpRequestError::fallback_allowed(FutuError::Network(std::io::Error::other(format!(
            "webtcp identity {web_identity}: all IPs failed"
        ))))
    }))
}

/// POST JSON through FTLogin WebTCP-short.
pub(crate) async fn post_json_via_webtcp(
    client_type: u8,
    web_identity: u32,
    addrs: &[(String, u16)],
    url: &str,
    body: &serde_json::Value,
    device_id: &str,
) -> std::result::Result<serde_json::Value, WebTcpRequestError> {
    let payload = serde_json::to_vec(body).map_err(|e| {
        WebTcpRequestError::fallback_forbidden(FutuError::Codec(format!("webtcp json encode: {e}")))
    })?;
    request_json_via_webtcp(
        client_type,
        web_identity,
        addrs,
        "POST",
        url,
        Some(&payload),
        device_id,
    )
    .await
}

#[cfg(test)]
mod tests;