|
| 1 | +// Copyright (c), Mysten Labs, Inc. |
| 2 | +// SPDX-License-Identifier: Apache-2.0 |
| 3 | + |
| 4 | +//! Aggregator server for Seal committee mode. It fetches encrypted partial keys from committee |
| 5 | +//! servers, verifies and aggregates them into a single response. |
| 6 | +
|
| 7 | +#![allow(dead_code)] |
| 8 | +#![allow(unused_variables)] |
| 9 | + |
| 10 | +use anyhow::{Context, Result}; |
| 11 | +use axum::{ |
| 12 | + extract::State, |
| 13 | + http::{HeaderMap, StatusCode}, |
| 14 | + response::{IntoResponse, Response}, |
| 15 | + routing::post, |
| 16 | + Json, Router, |
| 17 | +}; |
| 18 | +use mysten_service::{get_mysten_service, package_name, package_version}; |
| 19 | +use seal_committee::{grpc_helper::create_grpc_client, move_types::PartialKeyServer, Network}; |
| 20 | +use seal_sdk::{FetchKeyRequest, FetchKeyResponse}; |
| 21 | +use serde::Deserialize; |
| 22 | +use std::env; |
| 23 | +use std::sync::Arc; |
| 24 | +use sui_rpc::client::Client as SuiGrpcClient; |
| 25 | +use sui_sdk_types::Address; |
| 26 | +use tower_http::cors::{Any, CorsLayer}; |
| 27 | +use tracing::info; |
| 28 | + |
| 29 | +/// Minimum required version for committee members' responses (matches typescript). |
| 30 | +const MIN_SERVER_VERSION: &str = ">=0.4.1"; |
| 31 | + |
| 32 | +/// Default port for aggregator server. |
| 33 | +const DEFAULT_PORT: u16 = 2024; |
| 34 | + |
| 35 | +/// Configuration for aggregator server. |
| 36 | +#[derive(Deserialize)] |
| 37 | +struct Config { |
| 38 | + network: Network, |
| 39 | + key_server_object_id: Address, |
| 40 | +} |
| 41 | + |
| 42 | +/// Application state. |
| 43 | +#[derive(Clone)] |
| 44 | +struct AppState { |
| 45 | + key_server_object_id: Address, |
| 46 | + network: Network, |
| 47 | + grpc_client: SuiGrpcClient, |
| 48 | + threshold: u16, |
| 49 | + committee_members: Arc<Vec<PartialKeyServer>>, |
| 50 | + // TODO: API storage and rotation. |
| 51 | +} |
| 52 | + |
| 53 | +/// Custom error type for aggregator responses. |
| 54 | +struct AggregatorError { |
| 55 | + status: StatusCode, |
| 56 | + message: String, |
| 57 | + headers: HeaderMap, |
| 58 | +} |
| 59 | + |
| 60 | +impl IntoResponse for AggregatorError { |
| 61 | + fn into_response(self) -> Response { |
| 62 | + let mut response = (self.status, self.message).into_response(); |
| 63 | + *response.headers_mut() = self.headers; |
| 64 | + response |
| 65 | + } |
| 66 | +} |
| 67 | + |
| 68 | +#[tokio::main] |
| 69 | +async fn main() -> Result<()> { |
| 70 | + tracing_subscriber::fmt::init(); |
| 71 | + |
| 72 | + // Load configuration from file. |
| 73 | + let config_path = |
| 74 | + env::var("CONFIG_PATH").context("CONFIG_PATH environment variable not set")?; |
| 75 | + info!("Loading config file: {}", config_path); |
| 76 | + |
| 77 | + let config: Config = serde_yaml::from_reader( |
| 78 | + std::fs::File::open(&config_path) |
| 79 | + .context(format!("Cannot open configuration file {config_path}"))?, |
| 80 | + ) |
| 81 | + .context("Failed to parse configuration file")?; |
| 82 | + let grpc_client = create_grpc_client(&config.network)?; |
| 83 | + |
| 84 | + info!( |
| 85 | + "Starting aggregator for KeyServer {} on {:?}", |
| 86 | + config.key_server_object_id, config.network |
| 87 | + ); |
| 88 | + |
| 89 | + let state = AppState { |
| 90 | + key_server_object_id: config.key_server_object_id, |
| 91 | + network: config.network, |
| 92 | + grpc_client, |
| 93 | + threshold: 0, // TODO: Load from onchain |
| 94 | + committee_members: Arc::new(vec![]), // TODO: Load from onchain |
| 95 | + }; |
| 96 | + |
| 97 | + // TODO: Spawn background task to watch onchain for committee version updates: |
| 98 | + // 1. Every 30s, fetch KeyServerV2.version from onchain |
| 99 | + // 2. If version changes, refresh committee_members in AppState |
| 100 | + info!( |
| 101 | + "Loaded committee with {} members, threshold {}", |
| 102 | + state.committee_members.len(), |
| 103 | + state.threshold |
| 104 | + ); |
| 105 | + |
| 106 | + let port: u16 = env::var("PORT") |
| 107 | + .unwrap_or_else(|_| DEFAULT_PORT.to_string()) |
| 108 | + .parse() |
| 109 | + .context("Invalid PORT")?; |
| 110 | + |
| 111 | + let app = get_mysten_service::<AppState>(package_name!(), package_version!()) |
| 112 | + .merge(Router::new().route("/v1/fetch_key", post(handle_fetch_key))) |
| 113 | + .with_state(state) |
| 114 | + .layer(CorsLayer::new().allow_origin(Any)); |
| 115 | + |
| 116 | + let addr = std::net::SocketAddr::from(([0, 0, 0, 0], port)); |
| 117 | + let listener = tokio::net::TcpListener::bind(addr).await?; |
| 118 | + info!("Aggregator server listening on http://localhost:{}", port); |
| 119 | + |
| 120 | + axum::serve(listener, app).await?; |
| 121 | + Ok(()) |
| 122 | +} |
| 123 | + |
| 124 | +/// Handle fetch_key request by fanning out to committee members and aggregating responses. |
| 125 | +async fn handle_fetch_key( |
| 126 | + State(state): State<AppState>, |
| 127 | + Json(request): Json<FetchKeyRequest>, |
| 128 | +) -> Result<(HeaderMap, Json<FetchKeyResponse>), AggregatorError> { |
| 129 | + // TODO: |
| 130 | + // 1. Call fetch_from_member for all committee members in parallel. |
| 131 | + // 2. Collect responses until we have t successful responses and abort others. |
| 132 | + // 3. Track versions from each response header (X-KeyServer-Version). Use the oldest version as |
| 133 | + // the aggregator's response version (?) |
| 134 | + // 4. Upon sufficient responses, Aaggregate encrypted responses using crypto::elgamal::aggregate_encrypted |
| 135 | + // 5. Return with appropriate headers |
| 136 | + |
| 137 | + unimplemented!("depends on crypto code") |
| 138 | +} |
| 139 | + |
| 140 | +/// Fetch encrypted partial key from a single committee member. |
| 141 | +async fn fetch_from_member( |
| 142 | + member: &PartialKeyServer, |
| 143 | + request: &FetchKeyRequest, |
| 144 | +) -> Result<(FetchKeyResponse, String), String> { |
| 145 | + // TODO: |
| 146 | + // 1. Implement HTTP client call to each member URL. |
| 147 | + // 2. Extract each and validate X-KeyServer-Version header with MIN_SERVER_VERSION |
| 148 | + // 3. Parse response body as FetchKeyResponse |
| 149 | + // 4. Verify encrypted signatures using crypto::ibe::verify_encrypted_signature. |
| 150 | + // 5. Return (response, version_string) |
| 151 | + |
| 152 | + unimplemented!("not implemented yet") |
| 153 | +} |
| 154 | + |
| 155 | +/// Load committee state from onchain KeyServerV2 object. |
| 156 | +async fn load_committee_state(key_server_obj_id: &Address, network: Network) -> Result<AppState> { |
| 157 | + // TODO: |
| 158 | + // 1. Fetch KeyServerV2 object from chain. |
| 159 | + // 2. Parse committee members and threshold. |
| 160 | + // 3. Return AppState with loaded data. |
| 161 | + |
| 162 | + unimplemented!("not implemented yet") |
| 163 | +} |
0 commit comments