JWT implementation (#9)

* first steps

* monolith architecture

* prettier in github workflow

* kubernetes setup

* envs

* basic jwt impl

* basic jwt tests

* fix clippy

* cargo fmt

* add KSOX_SERVER_JWT_SECRET env to workflows

.github/workflows/rust.yml

@@ -8,14 +8,21 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
-  KSOX_SERVER_WORKER_ADDRESS: 0.0.0.0:80
+  KSOX_SERVER_JWT_SECRET: 620c6abfe023e6e30645e48be1cc78c52792dfc6a2914aa1eb1f6dc567216d79
+  KSOX_SERVER_API_BIND: 0.0.0.0:80
 
 jobs:
+  prettier:
+    runs-on: self-hosted
+    steps:
+      - uses: actions/setup-node@v3
+      - name: Check formatting with prettier
+        run: npx prettier --check .
   fmt:
     runs-on: self-hosted
     steps:
       - uses: actions/checkout@v3
-      - name: Check formatting
+      - name: Check formatting with cargo fmt
         run: cargo fmt -- --check
   clippy:
     runs-on: self-hosted

.gitignore

@@ -1,4 +1,5 @@
 # Envs
+**/envs/
 **/*.env
 **/*.local
 **/*.cargo

Cargo.toml

@@ -1,6 +1,12 @@
 [workspace]
 resolver = "2"
-members = ["engine", "packages/shutdown", "worker"]
+members = [
+    "api",
+    "crates/balance",
+    "crates/exchange",
+    "crates/pay",
+    "crates/shutdown",
+]
 
 [workspace.package]
 edition = "2021"
@@ -17,11 +23,20 @@ axum = { version = "0.6", features = [
     "tracing",
     "ws",
 ] }
-chrono = { version = "0.4.26", features = ["serde"] }
+chrono = { version = "0.4", features = ["serde"] }
 futures = "0.3"
 hyper = { version = "0.14", features = ["full"] }
+jsonwebtoken = "8"
+once_cell = "1"
+proptest = "1"
+ring = "0.17"
+seq-macro = "0.3"
+serde = { version = "1", features = ["derive"] }
+surrealdb = { version = "1", features = ["kv-mem"] }
 thiserror = "1"
-tokio = { version = "1.32", features = ["full", "tracing"] }
-tokio-stream = "0.1.14"
+tokio = { version = "1", features = ["full", "tracing"] }
+tokio-stream = "0.1"
+tower = "0.4"
 tracing = "0.1"
 tracing-subscriber = "0.3"
+url = "2"

README.md

@@ -1,2 +1,25 @@
 # ksox-template
+
 The KSOX Project
+
+#### `./k8s/patches/dev/envs/config.env`:
+
+```
+SURREAL_BIND=0.0.0.0:80
+```
+
+#### `./k8s/patches/dev/envs/secrets.env`:
+
+```
+SURREAL_USER=surrealuser
+SURREAL_PASS=surrealp4ssword
+```
+
+#### `./.cargo/config.toml`:
+
+```
+[env]
+KSOX_SERVER_SURREALDB_URL = "http://surrealdb.test/"
+KSOX_SERVER_REDIS_URL = "redis://redis.test/"
+KSOX_SERVER_API_BIND = "0.0.0.0:8080"
+```

api/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "api"
+edition.workspace = true
+version.workspace = true
+authors.workspace = true
+description.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+axum.workspace = true
+chrono.workspace = true
+futures.workspace = true
+hyper.workspace = true
+jsonwebtoken.workspace = true
+once_cell.workspace = true
+proptest.workspace = true
+ring.workspace = true
+seq-macro.workspace = true
+serde.workspace = true
+shutdown = { version = "0.1.0", path = "../crates/shutdown" }
+surrealdb.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+tokio-stream.workspace = true
+tower.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+url.workspace = true

worker/src/app.rs → api/src/app.rs

@@ -3,16 +3,23 @@ use axum::{routing, Router};
 pub fn get_app() -> Router {
     Router::new()
         .route("/", routing::get(http::root))
+        .route("/me", routing::get(http::get_subject))
         .route("/sse", routing::get(sse::root))
         .route("/ws", routing::get(ws::root))
 }
 
 mod http {
     use chrono::Utc;
 
+    use crate::jwt::Claims;
+
     pub async fn root() -> String {
         format!("Hello from server! Time: {}\n", Utc::now())
     }
+
+    pub async fn get_subject(claims: Claims) -> String {
+        claims.sub
+    }
 }
 
 mod sse {

api/src/errors.rs

@@ -0,0 +1,40 @@
+use axum::{
+    response::{IntoResponse, Response},
+    Json,
+};
+use hyper::StatusCode;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, thiserror::Error)]
+pub enum ApiError {
+    #[error("parse address error")]
+    AddressParse(#[from] std::net::AddrParseError),
+
+    #[error("axum server error")]
+    Hyper(#[from] hyper::Error),
+
+    #[error("tracing setup error")]
+    Tracing(#[from] tracing::subscriber::SetGlobalDefaultError),
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+struct AuthErrorResponse {
+    error: String,
+}
+#[derive(Debug)]
+pub enum AuthError {
+    WrongCredentials,
+    InvalidToken,
+}
+impl IntoResponse for AuthError {
+    fn into_response(self) -> Response {
+        let (status, error_message) = match self {
+            AuthError::WrongCredentials => (StatusCode::UNAUTHORIZED, "Wrong credentials"),
+            AuthError::InvalidToken => (StatusCode::BAD_REQUEST, "Invalid token"),
+        };
+        let body = Json(AuthErrorResponse {
+            error: error_message.to_string(),
+        });
+        (status, body).into_response()
+    }
+}

api/src/jwt.rs

@@ -0,0 +1,79 @@
+use axum::{
+    async_trait,
+    extract::FromRequestParts,
+    headers::{authorization::Bearer, Authorization},
+    http::request::Parts,
+    RequestPartsExt, TypedHeader,
+};
+use chrono::Utc;
+use jsonwebtoken::{
+    decode, encode, Algorithm, DecodingKey, EncodingKey, Header, TokenData, Validation,
+};
+use once_cell::sync::Lazy;
+use serde::{Deserialize, Serialize};
+
+use crate::errors::AuthError;
+
+pub static KEYS: Lazy<Keys> = Lazy::new(|| {
+    let secret =
+        std::env::var("KSOX_SERVER_JWT_SECRET").expect("KSOX_SERVER_JWT_SECRET must be set");
+    Keys::new(secret.as_bytes())
+});
+pub struct Keys {
+    pub decoding: DecodingKey,
+    pub encoding: EncodingKey,
+}
+impl Keys {
+    fn new(secret: &[u8]) -> Self {
+        Self {
+            decoding: DecodingKey::from_secret(secret),
+            encoding: EncodingKey::from_secret(secret),
+        }
+    }
+}
+
+pub trait JwtEncodeDecode<T> {
+    fn decode(token: &str) -> jsonwebtoken::errors::Result<TokenData<T>>;
+    fn encode(&self) -> jsonwebtoken::errors::Result<String>;
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Claims {
+    pub sub: String,
+    pub exp: usize,
+}
+impl JwtEncodeDecode<Self> for Claims {
+    fn decode(token: &str) -> jsonwebtoken::errors::Result<TokenData<Self>> {
+        decode::<Claims>(token, &KEYS.decoding, &Validation::new(Algorithm::HS256))
+    }
+    fn encode(&self) -> jsonwebtoken::errors::Result<String> {
+        encode(&Header::new(Algorithm::HS256), self, &KEYS.encoding)
+    }
+}
+
+#[async_trait]
+impl<S> FromRequestParts<S> for Claims
+where
+    S: Send + Sync,
+{
+    type Rejection = AuthError;
+
+    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
+        // Extract the token from the authorization header
+        let TypedHeader(Authorization(bearer)) = parts
+            .extract::<TypedHeader<Authorization<Bearer>>>()
+            .await
+            .map_err(|_| AuthError::InvalidToken)?;
+        // Decode the user data
+
+        let token_data = Claims::decode(bearer.token()).map_err(|_| AuthError::InvalidToken)?;
+
+        if token_data.claims.exp
+            <= usize::try_from(Utc::now().timestamp()).map_err(|_| AuthError::WrongCredentials)?
+        {
+            return Err(AuthError::WrongCredentials);
+        }
+
+        Ok(token_data.claims)
+    }
+}

worker/src/main.rs → api/src/main.rs

@@ -1,10 +1,19 @@
+use once_cell::sync::Lazy;
+
 mod app;
 mod errors;
+mod jwt;
+#[cfg(test)]
+mod tests;
+mod user;
+
+static API_BIND: Lazy<String> =
+    Lazy::new(|| std::env::var("KSOX_SERVER_API_BIND").expect("KSOX_SERVER_API_BIND must be set"));
 
 #[tokio::main]
-async fn main() -> Result<(), errors::WorkerError> {
+async fn main() -> Result<(), errors::ApiError> {
     let subscriber = tracing_subscriber::fmt()
-        .with_max_level(tracing::Level::DEBUG)
+        .with_max_level(tracing::Level::INFO)
         .with_timer(tracing_subscriber::fmt::time::uptime())
         .with_level(true)
         .with_thread_ids(true)
@@ -16,8 +25,8 @@ async fn main() -> Result<(), errors::WorkerError> {
 
     let app = app::get_app();
 
-    let addr = std::env::var("KSOX_SERVER_WORKER_ADDRESS")?.parse()?;
-    tracing::debug!("server starting at {}", addr);
+    let addr = API_BIND.parse()?;
+    tracing::info!("🚀 server starting at {}", addr);
     axum::Server::bind(&addr)
         .serve(app.into_make_service())
         .with_graceful_shutdown(async {

api/src/tests.rs

@@ -0,0 +1 @@
+mod api_test;

api/src/tests/api_test.rs

@@ -0,0 +1,39 @@
+use axum::{body::Body, http::Request};
+use chrono::Utc;
+use futures::executor::block_on;
+use hyper::StatusCode;
+use proptest::prelude::*;
+use seq_macro::seq;
+use tower::ServiceExt;
+
+use crate::{
+    app::get_app,
+    jwt::{Claims, JwtEncodeDecode},
+};
+
+seq!(N in 0..15 {
+proptest! {
+    #[test]
+    fn test_me_endpoint~N(s in "[a-zA-Z0-9]{256}") {
+        let app = get_app();
+        let jwt = Claims {
+            sub: s.to_owned(),
+            exp: usize::try_from(Utc::now().timestamp()).unwrap() + 60,
+        };
+        let request = Request::builder()
+            .method("GET")
+            .uri("/me")
+            .header("Authorization", format!("Bearer {}", jwt.encode().unwrap()))
+            .body(Body::empty())
+            .unwrap();
+
+        let resp = block_on(async { app.oneshot(request).await.unwrap() });
+        let (parts, body) = resp.into_parts();
+        let bytes = block_on(hyper::body::to_bytes(body)).unwrap();
+        let body_str = String::from_utf8(bytes.to_vec()).expect("Response body is not a valid UTF-8 string");
+
+        assert_eq!(parts.status, StatusCode::OK);
+        assert_eq!(body_str, s);
+    }
+}
+});