futu_opend/startup/phase4/
preauth.rs

1fn telnet_bind_addr(bind_ip: &str, port: Option<u16>) -> Option<String> {
2    port.map(|port| format!("{bind_ip}:{port}"))
3}
4
5#[derive(Debug, Clone, Copy, Default)]
6struct VerificationOwnerFacts {
7    ftapi_loopback_legacy: bool,
8    rest_loopback_legacy: bool,
9    ws_loopback_legacy: bool,
10    grpc_loopback_legacy: bool,
11    rest_auth_setup: bool,
12    ws_auth_setup: bool,
13    grpc_auth_setup: bool,
14}
15
16fn external_verification_owner_available(facts: VerificationOwnerFacts) -> bool {
17    facts.ftapi_loopback_legacy
18        || facts.rest_loopback_legacy
19        || facts.ws_loopback_legacy
20        || facts.grpc_loopback_legacy
21        || facts.rest_auth_setup
22        || facts.ws_auth_setup
23        || facts.grpc_auth_setup
24}
25
26fn bind_includes_loopback(bind_ip: &str) -> bool {
27    bind_ip.eq_ignore_ascii_case("localhost")
28        || bind_ip
29            .parse::<std::net::IpAddr>()
30            .is_ok_and(|ip| ip.is_loopback() || ip.is_unspecified())
31}
32
33async fn activate_restricted_listeners_before_auth(
34    readiness: ListenerReadiness,
35    enabled: EnabledListenerSurfaces,
36    shutdown_rx: &mut tokio::sync::watch::Receiver<bool>,
37) -> Result<ActivatedListenerReadiness> {
38    let activated = readiness.activate_pre_auth(shutdown_rx).await?;
39    tracing::info!(
40        stage = "listeners_opened",
41        access_mode = "pre_login_restricted",
42        ftapi_enabled = enabled.ftapi,
43        websocket_port_present = enabled.websocket,
44        rest_port_present = enabled.rest,
45        grpc_port_present = enabled.grpc,
46        telnet_deferred_until_ready = enabled.telnet,
47        marker = %activated.marker(),
48        "readiness-gated listener sockets opened before authentication"
49    );
50    Ok(activated)
51}
52
53async fn wait_for_sigterm() {
54    #[cfg(unix)]
55    {
56        use tokio::signal::unix::{SignalKind, signal};
57
58        let mut sigterm = match signal(SignalKind::terminate()) {
59            Ok(signal) => signal,
60            Err(error) => {
61                tracing::error!(
62                    error = %error,
63                    "failed to install SIGTERM handler; graceful SIGTERM shutdown unavailable"
64                );
65                std::future::pending::<()>().await;
66                return;
67            }
68        };
69        let _ = sigterm.recv().await;
70    }
71
72    #[cfg(not(unix))]
73    {
74        std::future::pending::<()>().await;
75    }
76}
77
78fn transition_startup_to_shutdown(readiness: &futu_server::identity::StartupReadiness) {
79    for _ in 0..4 {
80        let current = readiness.snapshot();
81        if current.state == futu_server::identity::StartupState::ShuttingDown {
82            return;
83        }
84        match readiness.transition(
85            current.generation,
86            futu_server::identity::StartupEvent::Shutdown,
87        ) {
88            Ok(_) => return,
89            Err(futu_server::identity::StartupTransitionError::StaleGeneration) => continue,
90            Err(error) => {
91                tracing::warn!(?error, "startup identity rejected shutdown transition");
92                return;
93            }
94        }
95    }
96    tracing::warn!("startup identity remained unstable during shutdown transition");
97}
98
99async fn execute_auth_plan_until_shutdown(
100    bridge: Arc<GatewayBridge>,
101    auth_plan: AuthPlan,
102    shutdown_rx: &mut tokio::sync::watch::Receiver<bool>,
103    surface_tasks: &mut SurfaceTasks,
104) -> futu_core::error::Result<tokio::sync::mpsc::Receiver<futu_gateway_core::bridge::PushEvent>> {
105    await_auth_or_surface_exit(
106        execute_auth_plan(bridge, auth_plan),
107        shutdown_rx,
108        surface_tasks,
109    )
110    .await
111}
112
113async fn await_auth_or_surface_exit<F, T>(
114    auth: F,
115    shutdown_rx: &mut tokio::sync::watch::Receiver<bool>,
116    surface_tasks: &mut SurfaceTasks,
117) -> futu_core::error::Result<T>
118where
119    F: std::future::Future<Output = futu_core::error::Result<T>>,
120{
121    fn surface_error(surface: &str, result: anyhow::Result<()>) -> futu_core::error::FutuError {
122        let detail = match result {
123            Ok(()) => "listener exited without an error".to_string(),
124            Err(error) => error.to_string(),
125        };
126        futu_core::error::FutuError::Codec(format!(
127            "{surface} listener exited during startup authentication: {detail}"
128        ))
129    }
130
131    tokio::select! {
132        result = auth => result,
133        result = surface_task_result_or_pending(&mut surface_tasks.ftapi) => {
134            surface_tasks.ftapi = None;
135            Err(surface_error("FTAPI", result))
136        },
137        result = surface_task_result_or_pending(&mut surface_tasks.websocket) => {
138            surface_tasks.websocket = None;
139            Err(surface_error("WebSocket", result))
140        },
141        result = surface_task_result_or_pending(&mut surface_tasks.rest) => {
142            surface_tasks.rest = None;
143            Err(surface_error("REST", result))
144        },
145        result = surface_task_result_or_pending(&mut surface_tasks.grpc) => {
146            surface_tasks.grpc = None;
147            Err(surface_error("gRPC", result))
148        },
149        result = surface_task_result_or_pending(&mut surface_tasks.telnet) => {
150            surface_tasks.telnet = None;
151            Err(surface_error("Telnet", result))
152        },
153        signal = tokio::signal::ctrl_c() => Err(futu_core::error::FutuError::Codec(
154            match signal {
155                Ok(()) => "startup authentication interrupted by Ctrl+C".to_string(),
156                Err(error) => format!("Ctrl+C handler failed during startup authentication: {error}"),
157            }
158        )),
159        _ = wait_for_sigterm() => Err(futu_core::error::FutuError::Codec(
160            "startup authentication interrupted by SIGTERM".to_string(),
161        )),
162        changed = shutdown_rx.changed() => Err(futu_core::error::FutuError::Codec(
163            match changed {
164                Ok(()) => "startup authentication interrupted by shutdown request".to_string(),
165                Err(error) => format!("startup shutdown channel closed during authentication: {error}"),
166            }
167        )),
168    }
169}
170
171fn startup_auth_failure_after_restricted_bind(error: futu_core::error::FutuError) -> anyhow::Error {
172    anyhow::Error::new(error)
173        .context("startup authentication failed after restricted listeners opened")
174}
futu_opend/startup/phase4/preauth.rs

futu_opend/startup/phase4/
preauth.rs
