Merge pull request #6 from cspr-rad/server-1

A simple mockup of L2 logic without proofs or signatures.

Cargo.toml

@@ -3,6 +3,7 @@ resolver = "2"
 
 members = [
   "kairos-cli",
+  "kairos-server",
 ]
 
 [workspace.package]

kairos-server/.env

@@ -0,0 +1 @@
+KAIROS_SERVER_PORT="8000"

kairos-server/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "kairos-server"
+version.workspace = true
+edition.workspace = true
+
+[lib]
+
+[[bin]]
+name = "kairos-server"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+dotenvy = "0.15"
+axum = { version = "0.7", features = ["tracing"] }
+axum-extra = { version = "0.9", features = [
+    "typed-routing",
+    "typed-header",
+    "json-deserializer",
+] }
+thiserror = "1.0"
+anyhow = "1.0"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+tokio = { version = "1.35", features = ["full", "tracing", "macros"] }
+tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["std", "env-filter"] }
+
+[dev-dependencies]
+proptest = "1.4"
+axum-test = "14"

kairos-server/src/config.rs

@@ -0,0 +1,26 @@
+use std::{fmt, str::FromStr};
+
+#[derive(Debug)]
+pub struct ServerConfig {
+    pub port: u16,
+}
+
+impl ServerConfig {
+    pub fn from_env() -> Result<Self, String> {
+        let port = parse_env_as::<u16>("KAIROS_SERVER_PORT")?;
+        Ok(Self { port })
+    }
+}
+
+fn parse_env_as<T>(env: &str) -> Result<T, String>
+where
+    T: FromStr,
+    <T as FromStr>::Err: fmt::Display,
+{
+    std::env::var(env)
+        .map_err(|e| format!("Failed to fetch environment variable {}: {}", env, e))
+        .and_then(|val| {
+            val.parse::<T>()
+                .map_err(|e| format!("Failed to parse {}: {}", env, e))
+        })
+}

kairos-server/src/errors.rs

@@ -0,0 +1,68 @@
+use std::{
+    fmt,
+    ops::{Deref, DerefMut},
+};
+
+use axum::{
+    http::StatusCode,
+    response::{IntoResponse, Response},
+};
+
+#[derive(Debug)]
+pub struct AppErr {
+    error: anyhow::Error,
+    status: Option<StatusCode>,
+}
+
+impl AppErr {
+    pub fn set_status(err: impl Into<Self>, status: StatusCode) -> Self {
+        let mut err = err.into();
+        err.status = Some(status);
+        err
+    }
+}
+
+impl IntoResponse for AppErr {
+    fn into_response(self) -> Response {
+        (
+            self.status.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
+            format!("{}", self.error),
+        )
+            .into_response()
+    }
+}
+
+impl Deref for AppErr {
+    type Target = anyhow::Error;
+    fn deref(&self) -> &Self::Target {
+        &self.error
+    }
+}
+
+impl DerefMut for AppErr {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.error
+    }
+}
+
+impl fmt::Display for AppErr {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "{}: {}",
+            self.status.unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
+            self.error
+        )
+    }
+}
+
+impl std::error::Error for AppErr {}
+
+impl From<anyhow::Error> for AppErr {
+    fn from(error: anyhow::Error) -> Self {
+        Self {
+            error,
+            status: None,
+        }
+    }
+}

kairos-server/src/lib.rs

@@ -0,0 +1,21 @@
+pub mod config;
+pub mod errors;
+pub mod routes;
+pub mod state;
+
+use axum::Router;
+use axum_extra::routing::RouterExt;
+use state::LockedBatchState;
+
+pub use errors::AppErr;
+
+type PublicKey = String;
+type Signature = String;
+
+pub fn app_router(state: LockedBatchState) -> Router {
+    Router::new()
+        .typed_post(routes::deposit_handler)
+        .typed_post(routes::withdraw_handler)
+        .typed_post(routes::transfer_handler)
+        .with_state(state)
+}

kairos-server/src/main.rs

@@ -0,0 +1,21 @@
+use std::net::SocketAddr;
+
+use dotenvy::dotenv;
+use kairos_server::{config::ServerConfig, state::BatchState};
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt::init();
+    dotenv().ok();
+
+    let config = ServerConfig::from_env()
+        .unwrap_or_else(|e| panic!("Failed to parse server config from environment: {}", e));
+
+    let app = kairos_server::app_router(BatchState::new());
+
+    let axum_addr = SocketAddr::from(([127, 0, 0, 1], config.port));
+
+    tracing::info!("starting http server on `{}`", axum_addr);
+    let listener = tokio::net::TcpListener::bind(axum_addr).await.unwrap();
+    axum::serve(listener, app).await.unwrap();
+}

kairos-server/src/routes/deposit.rs

@@ -0,0 +1,51 @@
+use std::ops::Deref;
+
+use anyhow::anyhow;
+use axum::{extract::State, http::StatusCode, Json};
+use axum_extra::routing::TypedPath;
+use serde::{Deserialize, Serialize};
+use tracing::*;
+
+use crate::{state::LockedBatchState, AppErr, PublicKey};
+
+#[derive(TypedPath, Debug, Clone, Copy)]
+#[typed_path("/api/v1/deposit")]
+pub struct DepositPath;
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+pub struct Deposit {
+    pub public_key: PublicKey,
+    pub amount: u64,
+}
+
+#[instrument(level = "trace", skip(state), ret)]
+pub async fn deposit_handler(
+    _: DepositPath,
+    state: State<LockedBatchState>,
+    Json(Deposit { public_key, amount }): Json<Deposit>,
+) -> Result<(), AppErr> {
+    tracing::info!("TODO: verifying deposit");
+
+    tracing::info!("TODO: adding deposit to batch");
+
+    let mut state = state.deref().write().await;
+    let account = state.balances.entry(public_key.clone());
+
+    let balance = account.or_insert(0);
+    let updated_balance = balance.checked_add(amount).ok_or_else(|| {
+        AppErr::set_status(
+            anyhow!("deposit would overflow account"),
+            StatusCode::CONFLICT,
+        )
+    })?;
+
+    *balance = updated_balance;
+
+    tracing::info!(
+        "Updated account public_key={} balance={}",
+        public_key,
+        updated_balance
+    );
+
+    Ok(())
+}

kairos-server/src/routes/mod.rs

@@ -0,0 +1,7 @@
+pub mod deposit;
+pub mod transfer;
+pub mod withdraw;
+
+pub use deposit::deposit_handler;
+pub use transfer::transfer_handler;
+pub use withdraw::withdraw_handler;

kairos-server/src/routes/transfer.rs

@@ -0,0 +1,108 @@
+use anyhow::anyhow;
+use axum::{extract::State, http::StatusCode, Json};
+use axum_extra::routing::TypedPath;
+use serde::{Deserialize, Serialize};
+use tracing::instrument;
+
+use crate::{state::LockedBatchState, AppErr, PublicKey, Signature};
+
+#[derive(TypedPath)]
+#[typed_path("/api/v1/transfer")]
+pub struct TransferPath;
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+pub struct Transfer {
+    pub from: PublicKey,
+    pub signature: Signature,
+    pub to: PublicKey,
+    pub amount: u64,
+}
+
+#[instrument(level = "trace", skip(state), ret)]
+pub async fn transfer_handler(
+    _: TransferPath,
+    State(state): State<LockedBatchState>,
+    Json(transfer): Json<Transfer>,
+) -> Result<(), AppErr> {
+    if transfer.amount == 0 {
+        return Err(AppErr::set_status(
+            anyhow!("transfer amount must be greater than 0"),
+            StatusCode::BAD_REQUEST,
+        ));
+    }
+
+    tracing::info!("TODO: verifying transfer signature");
+
+    // We pre-check this read-only to error early without acquiring the write lock.
+    // This prevents a DoS attack exploiting the write lock.
+    tracing::info!("verifying transfer sender has sufficient funds");
+    check_sender_funds(&state, &transfer).await?;
+
+    let mut state = state.write().await;
+    let from_balance = state.balances.get_mut(&transfer.from).ok_or_else(|| {
+        AppErr::set_status(
+            anyhow!(
+                "Sender no longer has an account.
+                The sender just removed all their funds."
+            ),
+            StatusCode::CONFLICT,
+        )
+    })?;
+
+    *from_balance = from_balance.checked_sub(transfer.amount).ok_or_else(|| {
+        AppErr::set_status(
+            anyhow!(
+                "Sender no longer has sufficient funds, balance={}, transfer_amount={}.
+                The sender just moved their funds in a concurrent request",
+                from_balance,
+                transfer.amount
+            ),
+            StatusCode::CONFLICT,
+        )
+    })?;
+
+    let to_balance = state
+        .balances
+        .entry(transfer.to.clone())
+        .or_insert_with(|| {
+            tracing::info!("creating new account for receiver");
+            0
+        });
+
+    *to_balance = to_balance.checked_add(transfer.amount).ok_or_else(|| {
+        AppErr::set_status(anyhow!("Receiver balance overflow"), StatusCode::CONFLICT)
+    })?;
+
+    Ok(())
+}
+
+async fn check_sender_funds(state: &LockedBatchState, transfer: &Transfer) -> Result<(), AppErr> {
+    let state = state.read().await;
+    let from_balance = state.balances.get(&transfer.from).ok_or_else(|| {
+        AppErr::set_status(
+            anyhow!("Sender does not have an account"),
+            StatusCode::BAD_REQUEST,
+        )
+    })?;
+
+    from_balance.checked_sub(transfer.amount).ok_or_else(|| {
+        AppErr::set_status(
+            anyhow!(
+                "Sender does not have sufficient funds, balance={}, transfer_amount={}",
+                from_balance,
+                transfer.amount
+            ),
+            StatusCode::FORBIDDEN,
+        )
+    })?;
+
+    let to_balance = state.balances.get(&transfer.to).unwrap_or(&0);
+    if to_balance.checked_add(transfer.amount).is_none() {
+        return Err(AppErr::set_status(
+            anyhow!("Receiver balance overflow"),
+            StatusCode::CONFLICT,
+        ));
+    }
+
+    Ok(())
+}