sui_proxy/
admin.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3use crate::config::{DynamicPeerValidationConfig, RemoteWriteConfig, StaticPeerValidationConfig};
4use crate::handlers::publish_metrics;
5use crate::histogram_relay::HistogramRelay;
6use crate::middleware::{
7    expect_content_length, expect_mysten_proxy_header, expect_valid_public_key,
8};
9use crate::peers::{AllowedPeer, SuiNodeProvider};
10use crate::var;
11use anyhow::Error;
12use anyhow::Result;
13use axum::{
14    Extension, Router, extract::DefaultBodyLimit, http::StatusCode, middleware, routing::post,
15};
16use fastcrypto::ed25519::{Ed25519KeyPair, Ed25519PublicKey};
17use fastcrypto::traits::{KeyPair, ToFromBytes};
18use rustls::pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject};
19use std::net::SocketAddr;
20use std::sync::Arc;
21use std::time::Duration;
22use sui_tls::SUI_VALIDATOR_SERVER_NAME;
23use sui_tls::{
24    AllowAll, ClientCertVerifier, SelfSignedCertificate, TlsAcceptor, rustls::ServerConfig,
25};
26use tokio::signal;
27use tower::ServiceBuilder;
28use tower_http::{
29    LatencyUnit,
30    timeout::TimeoutLayer,
31    trace::{DefaultOnFailure, DefaultOnResponse, TraceLayer},
32};
33use tracing::{Level, info};
34
35/// Configure our graceful shutdown scenarios
36pub async fn shutdown_signal(h: axum_server::Handle<SocketAddr>) {
37    let ctrl_c = async {
38        signal::ctrl_c()
39            .await
40            .expect("failed to install Ctrl+C handler");
41    };
42
43    #[cfg(unix)]
44    let terminate = async {
45        signal::unix::signal(signal::unix::SignalKind::terminate())
46            .expect("failed to install signal handler")
47            .recv()
48            .await;
49    };
50
51    #[cfg(not(unix))]
52    let terminate = std::future::pending::<()>();
53
54    tokio::select! {
55        _ = ctrl_c => {},
56        _ = terminate => {},
57    }
58
59    let grace = 30;
60    info!(
61        "signal received, starting graceful shutdown, grace period {} seconds, if needed",
62        &grace
63    );
64    h.graceful_shutdown(Some(Duration::from_secs(grace)))
65}
66
67/// Reqwest client holds the global client for remote_push api calls
68/// it also holds the username and password.  The client has an underlying
69/// connection pool.  See reqwest documentation for details
70#[derive(Clone)]
71pub struct ReqwestClient {
72    pub client: reqwest::Client,
73    pub settings: RemoteWriteConfig,
74}
75
76pub fn make_reqwest_client(settings: RemoteWriteConfig, user_agent: &str) -> ReqwestClient {
77    ReqwestClient {
78        client: reqwest::Client::builder()
79            .user_agent(user_agent)
80            .pool_max_idle_per_host(settings.pool_max_idle_per_host)
81            .timeout(Duration::from_secs(var!("MIMIR_CLIENT_TIMEOUT", 30)))
82            .build()
83            .expect("cannot create reqwest client"),
84        settings,
85    }
86}
87
88// Labels are adhoc labels we will inject per our config
89#[derive(Clone)]
90pub struct Labels {
91    pub network: String,
92    pub inventory_hostname: String,
93}
94
95/// App will configure our routes. This fn is also used to instrument our tests
96pub fn app(
97    labels: Labels,
98    client: ReqwestClient,
99    relay: HistogramRelay,
100    allower: Option<SuiNodeProvider>,
101    timeout_secs: Option<u64>,
102) -> Router {
103    // build our application with a route and our sender mpsc
104    let mut router = Router::new()
105        .route("/publish/metrics", post(publish_metrics))
106        .route_layer(DefaultBodyLimit::max(var!(
107            "MAX_BODY_SIZE",
108            1024 * 1024 * 5
109        )))
110        .route_layer(middleware::from_fn(expect_mysten_proxy_header))
111        .route_layer(middleware::from_fn(expect_content_length));
112    if let Some(allower) = allower {
113        router = router
114            .route_layer(middleware::from_fn(expect_valid_public_key))
115            .layer(Extension(Arc::new(allower)));
116    }
117    router
118        // Enforce on all routes.
119        // If the request does not complete within the specified timeout it will be aborted
120        // and a 408 Request Timeout response will be sent.
121        .layer(TimeoutLayer::with_status_code(
122            StatusCode::REQUEST_TIMEOUT,
123            Duration::from_secs(timeout_secs.unwrap_or(20)),
124        ))
125        .layer(Extension(relay))
126        .layer(Extension(labels))
127        .layer(Extension(client))
128        .layer(
129            ServiceBuilder::new().layer(
130                TraceLayer::new_for_http()
131                    .on_response(
132                        DefaultOnResponse::new()
133                            .level(Level::INFO)
134                            .latency_unit(LatencyUnit::Seconds),
135                    )
136                    .on_failure(
137                        DefaultOnFailure::new()
138                            .level(Level::ERROR)
139                            .latency_unit(LatencyUnit::Seconds),
140                    ),
141            ),
142        )
143}
144
145/// Server creates our http/https server
146pub async fn server(
147    listener: std::net::TcpListener,
148    app: Router,
149    acceptor: Option<TlsAcceptor>,
150) -> std::io::Result<()> {
151    listener.set_nonblocking(true)?;
152    let listener = tokio::net::TcpListener::from_std(listener)?;
153
154    // setup our graceful shutdown
155    let handle = axum_server::Handle::new();
156    // Spawn a task to gracefully shutdown server.
157    tokio::spawn(shutdown_signal(handle.clone()));
158
159    if let Some(verify_peers) = acceptor {
160        axum_server::Server::from_listener(listener)
161            .acceptor(verify_peers)
162            .handle(handle)
163            .serve(app.into_make_service_with_connect_info::<SocketAddr>())
164            .await
165    } else {
166        axum_server::Server::from_listener(listener)
167            .handle(handle)
168            .serve(app.into_make_service_with_connect_info::<SocketAddr>())
169            .await
170    }
171}
172
173/// CertKeyPair wraps a self signed certificate and the corresponding public key
174pub struct CertKeyPair(pub SelfSignedCertificate, pub Ed25519PublicKey);
175
176/// Generate server certs for use with peer verification
177pub fn generate_self_cert(hostname: String) -> CertKeyPair {
178    let mut rng = rand::thread_rng();
179    let keypair = Ed25519KeyPair::generate(&mut rng);
180    CertKeyPair(
181        SelfSignedCertificate::new(keypair.copy().private(), &hostname),
182        keypair.public().to_owned(),
183    )
184}
185
186/// Load a certificate for use by the listening service
187fn load_certs(filename: &str) -> Vec<CertificateDer<'static>> {
188    CertificateDer::pem_file_iter(filename)
189        .unwrap_or_else(|e| panic!("cannot open certificate file: {}; {}", filename, e))
190        .collect::<Result<Vec<_>, _>>()
191        .unwrap_or_else(|e| panic!("cannot parse certificate file: {}; {}", filename, e))
192}
193
194/// Load a private key
195fn load_private_key(filename: &str) -> PrivateKeyDer<'static> {
196    PrivateKeyDer::from_pem_file(filename).unwrap_or_else(|e| {
197        panic!(
198            "cannot load private key from {} (encrypted keys not supported): {}",
199            filename, e
200        )
201    })
202}
203
204/// load the static keys we'll use to allow external non-validator nodes to push metrics
205fn load_static_peers(
206    static_peers: Option<StaticPeerValidationConfig>,
207) -> Result<Vec<AllowedPeer>, Error> {
208    let Some(static_peers) = static_peers else {
209        return Ok(vec![]);
210    };
211    let static_keys = static_peers
212        .pub_keys
213        .into_iter()
214        .map(|spk| {
215            let peer_id = hex::decode(spk.peer_id).unwrap();
216            let public_key = Ed25519PublicKey::from_bytes(peer_id.as_ref()).unwrap();
217            let s = AllowedPeer {
218                name: spk.name.clone(),
219                public_key,
220            };
221            info!(
222                "loaded static peer: {} public key: {}",
223                &s.name, &s.public_key,
224            );
225            s
226        })
227        .collect();
228    Ok(static_keys)
229}
230
231/// Default allow mode for server, we don't verify clients, everything is accepted
232pub fn create_server_cert_default_allow(
233    hostname: String,
234) -> Result<ServerConfig, sui_tls::rustls::Error> {
235    let CertKeyPair(server_certificate, _) = generate_self_cert(hostname);
236
237    ClientCertVerifier::new(AllowAll, SUI_VALIDATOR_SERVER_NAME.to_string()).rustls_server_config(
238        vec![server_certificate.rustls_certificate()],
239        server_certificate.rustls_private_key(),
240    )
241}
242
243/// Verify clients against sui blockchain, clients that are not found in sui_getValidators
244/// will be rejected
245pub fn create_server_cert_enforce_peer(
246    dynamic_peers: DynamicPeerValidationConfig,
247    static_peers: Option<StaticPeerValidationConfig>,
248) -> Result<(ServerConfig, Option<SuiNodeProvider>), sui_tls::rustls::Error> {
249    // Capture before the cert/key destructure moves out the rest of the struct.
250    let hashi_object_id = dynamic_peers.hashi_object_id.clone();
251    let rpc_url = dynamic_peers.url.clone();
252    let poll_interval = dynamic_peers.interval;
253
254    let (Some(certificate_path), Some(private_key_path)) =
255        (dynamic_peers.certificate_file, dynamic_peers.private_key)
256    else {
257        return Err(sui_tls::rustls::Error::General(
258            "missing certs to initialize server".into(),
259        ));
260    };
261    let static_peers = load_static_peers(static_peers).map_err(|e| {
262        sui_tls::rustls::Error::General(format!("unable to load static pub keys: {}", e))
263    })?;
264    let allower = SuiNodeProvider::new(rpc_url, poll_interval, static_peers, hashi_object_id);
265    allower.poll_peer_list();
266    let c = ClientCertVerifier::new(allower.clone(), SUI_VALIDATOR_SERVER_NAME.to_string())
267        .rustls_server_config(
268            load_certs(&certificate_path),
269            load_private_key(&private_key_path),
270        )?;
271    Ok((c, Some(allower)))
272}
sui_proxy/admin.rs

sui_proxy/
admin.rs
