fedimint_server/net/api/
mod.rs

1pub mod announcement;
2pub mod guardian_metadata;
3mod http_auth;
4pub mod pkarr_publish;
5
6use std::fmt::{self, Formatter};
7use std::future::Future;
8use std::net::SocketAddr;
9use std::panic::AssertUnwindSafe;
10use std::str::FromStr;
11use std::time::Duration;
12
13use anyhow::{Context, bail};
14use async_trait::async_trait;
15use fedimint_core::core::ModuleInstanceId;
16use fedimint_core::encoding::{Decodable, Encodable};
17use fedimint_core::module::{ApiEndpoint, ApiEndpointContext, ApiError, ApiRequestErased};
18use fedimint_logging::LOG_NET_API;
19use futures::FutureExt;
20use jsonrpsee::RpcModule;
21use jsonrpsee::server::{PingConfig, RpcServiceBuilder, ServerBuilder, ServerHandle};
22use jsonrpsee::types::ErrorObject;
23use tracing::{error, info};
24
25use crate::metrics;
26use crate::net::api::http_auth::HttpAuthLayer;
27
28#[derive(Clone, Encodable, Decodable, Default)]
29pub struct ApiSecrets(Vec<String>);
30
31impl fmt::Debug for ApiSecrets {
32    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
33        f.debug_struct("ApiSecrets")
34            .field("num_secrets", &self.0.len())
35            .finish()
36    }
37}
38
39impl FromStr for ApiSecrets {
40    type Err = anyhow::Error;
41
42    fn from_str(s: &str) -> anyhow::Result<Self> {
43        if s.is_empty() {
44            return Ok(Self(vec![]));
45        }
46
47        let secrets = s
48            .split(',')
49            .map(str::trim)
50            .map(|s| {
51                if s.is_empty() {
52                    bail!("Empty Api Secret is not allowed")
53                }
54                Ok(s.to_string())
55            })
56            .collect::<anyhow::Result<_>>()?;
57        Ok(ApiSecrets(secrets))
58    }
59}
60
61impl ApiSecrets {
62    pub fn is_empty(&self) -> bool {
63        self.0.is_empty()
64    }
65
66    /// Get "active" secret - one that should be used to call other peers
67    pub fn get_active(&self) -> Option<String> {
68        self.0.first().cloned()
69    }
70
71    /// Get all secrets
72    pub fn get_all(&self) -> &[String] {
73        &self.0
74    }
75
76    /// Get empty value - meaning no secrets to use
77    pub fn none() -> ApiSecrets {
78        Self(vec![])
79    }
80}
81
82/// How long an API request may spend preparing its context and running its
83/// handler.
84pub(crate) const API_ENDPOINT_TIMEOUT: Duration = Duration::from_mins(1);
85
86/// Apply the same endpoint deadline and timeout error across API transports.
87pub(crate) async fn with_api_endpoint_timeout<T>(
88    handler: impl Future<Output = Result<T, ApiError>>,
89) -> Result<T, ApiError> {
90    tokio::time::timeout(API_ENDPOINT_TIMEOUT, handler)
91        .await
92        .unwrap_or_else(|_| Err(ApiError::new(-32000, "Request timeout".to_owned())))
93}
94
95/// Has the context necessary for serving API endpoints
96///
97/// Returns the specific `State` the endpoint requires and the
98/// `ApiEndpointContext` which all endpoints can access.
99#[async_trait]
100pub trait HasApiContext<State> {
101    async fn context(
102        &self,
103        request: &ApiRequestErased,
104        id: Option<ModuleInstanceId>,
105    ) -> (&State, ApiEndpointContext);
106}
107
108pub async fn spawn<T>(
109    name: &'static str,
110    api_bind: SocketAddr,
111    module: RpcModule<T>,
112    max_connections: u32,
113    api_secrets: ApiSecrets,
114) -> ServerHandle {
115    info!(target: LOG_NET_API, "Starting http api on ws://{api_bind}");
116
117    let builder = tower::ServiceBuilder::new().layer(HttpAuthLayer::new(api_secrets.get_all()));
118
119    ServerBuilder::new()
120        .max_connections(max_connections)
121        .enable_ws_ping(PingConfig::new().ping_interval(Duration::from_secs(10)))
122        .set_rpc_middleware(
123            RpcServiceBuilder::new()
124                .layer(metrics::jsonrpsee::MetricsLayer::new(module.method_names())),
125        )
126        .set_http_middleware(builder)
127        .build(&api_bind.to_string())
128        .await
129        .context(format!("Bind address: {api_bind}"))
130        .context(format!("API name: {name}"))
131        .expect("Could not build API server")
132        .start(module)
133}
134
135pub fn attach_endpoints<State, T>(
136    rpc_module: &mut RpcModule<T>,
137    endpoints: Vec<ApiEndpoint<State>>,
138    module_instance_id: Option<ModuleInstanceId>,
139) where
140    T: HasApiContext<State> + Sync + Send + 'static,
141    State: Sync + Send + 'static,
142{
143    for endpoint in endpoints {
144        let path = if let Some(module_instance_id) = module_instance_id {
145            // This memory leak is fine because it only happens on server startup
146            // and path has to live till the end of program anyways.
147            Box::leak(format!("module_{}_{}", module_instance_id, endpoint.path).into_boxed_str())
148        } else {
149            endpoint.path
150        };
151        // Check if paths contain any abnormal characters
152        assert!(
153            !path.contains(|c: char| !matches!(c, '0'..='9' | 'a'..='z' | '_')),
154            "Constructing bad path name {path}"
155        );
156
157        // Another memory leak that is fine because the function is only called once at
158        // startup
159        let handler: &'static _ = Box::leak(endpoint.handler);
160
161        rpc_module
162            .register_async_method(path, move |params, rpc_state, _extensions| async move {
163                let params = params.one::<serde_json::Value>()?;
164
165                // Using AssertUnwindSafe here is far from ideal. In theory this means we could
166                // end up with an inconsistent state in theory. In practice most API functions
167                // are only reading and the few that do write anything are atomic. Lastly, this
168                // is only the last line of defense
169                AssertUnwindSafe(with_api_endpoint_timeout(async {
170                    let request = serde_json::from_value(params)
171                        .map_err(|e| ApiError::bad_request(e.to_string()))?;
172
173                    let (state, context) = rpc_state.context(&request, module_instance_id).await;
174
175                    (handler)(state, context, request).await
176                }))
177                .catch_unwind()
178                .await
179                .map_err(|_| {
180                    error!(
181                        target: LOG_NET_API,
182                        path, "API handler panicked, DO NOT IGNORE, FIX IT!!!"
183                    );
184                    ErrorObject::owned(500, "API handler panicked", None::<()>)
185                })?
186                .map_err(|e| ErrorObject::owned(e.code, e.message, None::<()>))
187            })
188            .expect("Failed to register async method");
189    }
190}
191
192#[cfg(test)]
193mod tests;
fedimint_server/net/api/mod.rs

fedimint_server/net/api/
mod.rs
