futu_rest/server/
startup.rs

1use std::sync::Arc;
2
3use futu_auth::{KeyStore, RuntimeCounters};
4use futu_server::listener_status::{
5    ListenerBindEventSender, ListenerSurface, notify_listener_failed, notify_listener_opened,
6};
7use futu_server::router::RequestRouter;
8use tokio::sync::watch;
9
10use crate::ws::WsBroadcaster;
11
12use super::{LEGACY_WS_WARN_MESSAGE, RestAdminHooks, build_router_with_auth_full_admin};
13use axum::Router;
14
15#[derive(Clone)]
16pub struct RestTlsConfig {
17    inner: axum_server::tls_rustls::RustlsConfig,
18}
19
20impl RestTlsConfig {
21    pub async fn from_pem_files(
22        cert_path: impl AsRef<std::path::Path>,
23        key_path: impl AsRef<std::path::Path>,
24    ) -> std::io::Result<Self> {
25        futu_backend::auth::install_default_rustls_crypto_provider();
26        let cert = read_rest_tls_pem("certificate", cert_path).await?;
27        let key = read_rest_tls_pem("private-key", key_path).await?;
28        let inner = axum_server::tls_rustls::RustlsConfig::from_pem(cert, key)
29            .await
30            .map_err(|error| {
31                std::io::Error::new(
32                    error.kind(),
33                    format!("REST TLS PEM validation failed: {error}"),
34                )
35            })?;
36        Ok(Self { inner })
37    }
38}
39
40async fn read_rest_tls_pem(
41    role: &'static str,
42    path: impl AsRef<std::path::Path>,
43) -> std::io::Result<Vec<u8>> {
44    tokio::fs::read(path).await.map_err(|error| {
45        std::io::Error::new(
46            error.kind(),
47            format!("REST TLS {role} file read failed ({:?})", error.kind()),
48        )
49    })
50}
51
52#[derive(Clone)]
53pub enum RestTransport {
54    Plaintext,
55    Tls(RestTlsConfig),
56}
57
58impl RestTransport {
59    #[must_use]
60    pub const fn is_tls(&self) -> bool {
61        matches!(self, Self::Tls(_))
62    }
63}
64
65/// 启动 REST API 服务，挂载 KeyStore 做 Bearer Token 鉴权 +
66/// RuntimeCounters 做限额。
67pub async fn start_with_auth(
68    listen_addr: &str,
69    router: Arc<RequestRouter>,
70    ws_broadcaster: Arc<WsBroadcaster>,
71    key_store: Arc<KeyStore>,
72    counters: Arc<RuntimeCounters>,
73) -> std::io::Result<()> {
74    start_with_auth_and_admin(
75        listen_addr,
76        router,
77        ws_broadcaster,
78        key_store,
79        counters,
80        None,
81    )
82    .await
83}
84
85/// v1.4.32+ 同 `start_with_auth`，但额外接 admin_status_provider。
86/// 让 `/api/admin/status` 能返回实时健康快照。
87/// `push_health_snapshot_provider` 仍需走 `start_with_auth_full_admin` 注入；
88/// 未注入时 `/api/push-subscriber-info` loud-fail 503。
89pub async fn start_with_auth_and_admin(
90    listen_addr: &str,
91    router: Arc<RequestRouter>,
92    ws_broadcaster: Arc<WsBroadcaster>,
93    key_store: Arc<KeyStore>,
94    counters: Arc<RuntimeCounters>,
95    admin_status_provider: Option<crate::adapter::AdminStatusProvider>,
96) -> std::io::Result<()> {
97    start_with_auth_full_admin(
98        listen_addr,
99        router,
100        ws_broadcaster,
101        key_store,
102        counters,
103        RestAdminHooks {
104            admin_status_provider,
105            ..RestAdminHooks::default()
106        },
107    )
108    .await
109}
110
111/// v1.4.32+ 完整 admin 入口：同时接 status provider + reload handler。
112///
113/// v1.4.83 §9 Phase 2 F5: 加 `push_health_snapshot_provider` 参数支持
114/// `/api/push-subscriber-info` 返真实 push 通道健康 state。
115pub async fn start_with_auth_full_admin(
116    listen_addr: &str,
117    router: Arc<RequestRouter>,
118    ws_broadcaster: Arc<WsBroadcaster>,
119    key_store: Arc<KeyStore>,
120    counters: Arc<RuntimeCounters>,
121    hooks: RestAdminHooks,
122) -> std::io::Result<()> {
123    let (_shutdown_tx, shutdown_rx) = watch::channel(false);
124    start_with_auth_full_admin_until_shutdown_with_transport_and_listener_events(
125        listen_addr,
126        router,
127        ws_broadcaster,
128        key_store,
129        counters,
130        hooks,
131        RestTransport::Plaintext,
132        shutdown_rx,
133        None,
134    )
135    .await
136}
137
138/// 同 [`start_with_auth_full_admin`]，但支持 daemon 统一 shutdown 信号。
139pub async fn start_with_auth_full_admin_until_shutdown(
140    listen_addr: &str,
141    router: Arc<RequestRouter>,
142    ws_broadcaster: Arc<WsBroadcaster>,
143    key_store: Arc<KeyStore>,
144    counters: Arc<RuntimeCounters>,
145    hooks: RestAdminHooks,
146    shutdown_rx: watch::Receiver<bool>,
147) -> std::io::Result<()> {
148    start_with_auth_full_admin_until_shutdown_with_transport_and_listener_events(
149        listen_addr,
150        router,
151        ws_broadcaster,
152        key_store,
153        counters,
154        hooks,
155        RestTransport::Plaintext,
156        shutdown_rx,
157        None,
158    )
159    .await
160}
161
162/// Shutdown-aware REST server that reports its exact socket bind result.
163#[allow(clippy::too_many_arguments)]
164pub async fn start_with_auth_full_admin_until_shutdown_with_listener_events(
165    listen_addr: &str,
166    router: Arc<RequestRouter>,
167    ws_broadcaster: Arc<WsBroadcaster>,
168    key_store: Arc<KeyStore>,
169    counters: Arc<RuntimeCounters>,
170    hooks: RestAdminHooks,
171    shutdown_rx: watch::Receiver<bool>,
172    listener_events: Option<ListenerBindEventSender>,
173) -> std::io::Result<()> {
174    start_with_auth_full_admin_until_shutdown_with_transport_and_listener_events(
175        listen_addr,
176        router,
177        ws_broadcaster,
178        key_store,
179        counters,
180        hooks,
181        RestTransport::Plaintext,
182        shutdown_rx,
183        listener_events,
184    )
185    .await
186}
187
188/// Transport-aware REST server with daemon shutdown and exact bind events.
189#[allow(clippy::too_many_arguments)]
190pub async fn start_with_auth_full_admin_until_shutdown_with_transport_and_listener_events(
191    listen_addr: &str,
192    router: Arc<RequestRouter>,
193    ws_broadcaster: Arc<WsBroadcaster>,
194    key_store: Arc<KeyStore>,
195    counters: Arc<RuntimeCounters>,
196    hooks: RestAdminHooks,
197    transport: RestTransport,
198    shutdown_rx: watch::Receiver<bool>,
199    listener_events: Option<ListenerBindEventSender>,
200) -> std::io::Result<()> {
201    let scope_mode = key_store.is_configured();
202    if scope_mode && !transport.is_tls() {
203        notify_listener_failed(&listener_events, ListenerSurface::Rest);
204        return Err(std::io::Error::new(
205            std::io::ErrorKind::PermissionDenied,
206            "configured REST authentication requires native TLS",
207        ));
208    }
209    let app = build_router_with_auth_full_admin(router, ws_broadcaster, key_store, counters, hooks);
210    let listener = tokio::net::TcpListener::bind(listen_addr)
211        .await
212        .map_err(|error| {
213            notify_listener_failed(&listener_events, ListenerSurface::Rest);
214            futu_server::bind_hint::io_bind_error("REST", "--rest-port", listen_addr, error)
215        })?;
216    serve_bound_listener(
217        listener,
218        app,
219        scope_mode,
220        transport,
221        shutdown_rx,
222        listener_events,
223        listen_addr,
224    )
225    .await
226}
227
228#[allow(clippy::too_many_arguments)]
229async fn serve_bound_listener(
230    listener: tokio::net::TcpListener,
231    app: Router,
232    scope_mode: bool,
233    transport: RestTransport,
234    shutdown_rx: watch::Receiver<bool>,
235    listener_events: Option<ListenerBindEventSender>,
236    listen_addr: &str,
237) -> std::io::Result<()> {
238    match transport {
239        RestTransport::Plaintext => {
240            let _serving =
241                notify_listener_opened(&listener_events, ListenerSurface::Rest, &shutdown_rx)
242                    .await?;
243            drop(listener_events);
244            tracing::info!(
245                addr = %listen_addr,
246                scope_mode,
247                transport = "http",
248                "REST API 服务已启动 (WebSocket: /ws)"
249            );
250            if !scope_mode {
251                warn_legacy_mode(listen_addr);
252            }
253            axum::serve(
254                listener,
255                app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
256            )
257            .with_graceful_shutdown(rest_shutdown_requested(shutdown_rx))
258            .await
259        }
260        RestTransport::Tls(tls_config) => {
261            let std_listener = listener.into_std().inspect_err(|_| {
262                notify_listener_failed(&listener_events, ListenerSurface::Rest);
263            })?;
264            let handle = axum_server::Handle::new();
265            let server =
266                axum_server::from_tcp_rustls(std_listener, tls_config.inner).inspect_err(|_| {
267                    notify_listener_failed(&listener_events, ListenerSurface::Rest);
268                })?;
269            let _serving =
270                notify_listener_opened(&listener_events, ListenerSurface::Rest, &shutdown_rx)
271                    .await?;
272            drop(listener_events);
273            tracing::info!(
274                addr = %listen_addr,
275                scope_mode,
276                transport = "https",
277                "REST API 服务已启动 (WebSocket: /ws)"
278            );
279            if !scope_mode {
280                warn_legacy_mode(listen_addr);
281            }
282
283            let shutdown_handle = handle.clone();
284            let shutdown_task = tokio::spawn(async move {
285                rest_shutdown_requested(shutdown_rx).await;
286                shutdown_handle.graceful_shutdown(None);
287            });
288            let result = server
289                .handle(handle)
290                .serve(app.into_make_service_with_connect_info::<std::net::SocketAddr>())
291                .await;
292            shutdown_task.abort();
293            let _ = shutdown_task.await;
294            result
295        }
296    }
297}
298
299fn warn_legacy_mode(listen_addr: &str) {
300    tracing::warn!("REST API in legacy mode (no keys.json); mutating endpoints are blocked");
301    // v1.4.93 P0-5 (NEW-C-02): WS 也对齐 mutating-blocked policy 的"loud
302    // unauth"信号 — REST `/ws` route 在 legacy 模式接受 unauthenticated
303    // handshake (no-token / wrong-bearer / bogus-query 都 HTTP 101)，未授权
304    // 客户端可接收 live push。本版不 reject（保持向后兼容，未来 major 版
305    // 默认 reject），但补 startup loud WARN + CHANGELOG 公告。
306    tracing::warn!("{}", LEGACY_WS_WARN_MESSAGE);
307    // v1.4.86 SEC-003 Q4 真 fix: legacy mode 下 mutating endpoint 硬门禁
308    // (/api/order / modify-order / cancel-order / cancel-all-order / unlock-trade /
309    // reconfirm-order / admin/*). 只读 endpoint 继续 legacy 允许.
310    tracing::warn!(
311        listen_addr = %listen_addr,
312        readonly_endpoints = "qot/account/order-read",
313        blocked_mutating_endpoints = "/api/order,/api/modify-order,/api/cancel-order,/api/cancel-all-order,/api/unlock-trade,/api/reconfirm-order,/api/admin/*",
314        ws_legacy_unauthenticated = true,
315        migration = "futucli gen-key --id my-key --scopes qot:read,acc:read,trade:real; restart with --rest-keys-file /path/to/keys.json",
316        "REST API legacy mode: no keys.json configured; read endpoints remain unauthenticated, mutating/admin endpoints return 401, /ws still accepts unauthenticated connections for compatibility and v2 will default-reject"
317    );
318}
319
320async fn rest_shutdown_requested(mut shutdown_rx: watch::Receiver<bool>) {
321    loop {
322        if *shutdown_rx.borrow() {
323            tracing::info!("REST API server stopped by shutdown signal");
324            return;
325        }
326        if shutdown_rx.changed().await.is_err() {
327            tracing::info!("REST API server stopped after shutdown sender dropped");
328            return;
329        }
330    }
331}
332
333#[cfg(test)]
334mod tests;
335#[cfg(test)]
336mod tls_tests;
futu_rest/server/startup.rs

futu_rest/server/
startup.rs
