Add argon2, rusty_pool, oneshot dependencies; remove auth_token.rs; r…

…efactor auth module usage; introduce Credential enum; add error logging; misc refactoring and cleanup

binaries/geph5-broker/Cargo.toml

@@ -38,3 +38,6 @@ sillad-sosistab3={path="../../libraries/sillad-sosistab3"}
 smol-timeout = "0.6.0"
 stdcode = "0.1.14"
 bytes = {version="1.5.0", features=["serde"]}
+argon2 = "0.5.3"
+rusty_pool = "0.7.0"
+oneshot = "0.1.6"

binaries/geph5-broker/src/main.rs

@@ -9,10 +9,10 @@ use once_cell::sync::{Lazy, OnceCell};
 use rpc_impl::BrokerImpl;
 use serde::Deserialize;
 use smolscale::immortal::{Immortal, RespawnStrategy};
-use std::{fs, net::SocketAddr, path::PathBuf};
+use std::{fmt::Debug, fs, net::SocketAddr, path::PathBuf};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 
-mod auth_token;
+mod auth;
 mod database;
 mod routes;
 mod rpc_impl;
@@ -117,3 +117,7 @@ async fn main() -> anyhow::Result<()> {
 async fn rpc(Json(payload): Json<JrpcRequest>) -> Json<JrpcResponse> {
     Json(BrokerService(BrokerImpl {}).respond_raw(payload).await)
 }
+
+fn log_error(e: &impl Debug) {
+    tracing::warn!(err = debug(e), "transient error")
+}

binaries/geph5-broker/src/rpc_impl.rs

@@ -2,7 +2,7 @@ use std::{net::SocketAddr, ops::Deref, sync::Arc, time::Duration};
 
 use async_trait::async_trait;
 use bytes::Bytes;
-use ed25519_dalek::{SigningKey, VerifyingKey};
+use ed25519_dalek::{VerifyingKey};
 use geph5_broker_protocol::{
     AccountLevel, AuthError, BridgeDescriptor, BrokerProtocol, Credential, ExitDescriptor,
     ExitList, GenericError, Mac, RouteDescriptor, Signed, DOMAIN_EXIT_DESCRIPTOR,
@@ -11,10 +11,10 @@ use isocountry::CountryCode;
 use mizaru2::{BlindedClientToken, BlindedSignature, ClientToken, UnblindedSignature};
 use moka::future::Cache;
 use once_cell::sync::Lazy;
-use rand::Rng as _;
+
 
 use crate::{
-    auth_token::{self, new_auth_token, valid_auth_token},
+    auth::{new_auth_token, valid_auth_token, validate_username_pwd},
     database::{insert_exit, query_bridges, ExitRow, POSTGRES},
     routes::bridge_to_leaf_route,
     CONFIG_FILE, FREE_MIZARU_SK, MASTER_SECRET, PLUS_MIZARU_SK,
@@ -39,7 +39,10 @@ impl BrokerProtocol for BrokerImpl {
 
     async fn get_auth_token(&self, credential: Credential) -> Result<String, AuthError> {
         let user_id = match credential {
-            Credential::TestDummy => 42, // User ID for TestDummy
+            Credential::TestDummy => 42,
+            Credential::LegacyUsernamePassword { username, password } => {
+                validate_username_pwd(&username, &password).await?
+            }
         };
 
         let token = new_auth_token(user_id)
@@ -124,7 +127,7 @@ impl BrokerProtocol for BrokerImpl {
         exit: SocketAddr,
     ) -> Result<RouteDescriptor, GenericError> {
         // authenticate the token
-        let account_level = if PLUS_MIZARU_SK
+        let _account_level = if PLUS_MIZARU_SK
             .to_public_key()
             .blind_verify(token, &sig)
             .is_ok()

binaries/geph5-client/src/auth.rs

@@ -4,7 +4,7 @@ use anyctx::AnyCtx;
 use anyhow::Context as _;
 use blind_rsa_signatures as brs;
 use geph5_broker_protocol::{AccountLevel, AuthError, Credential};
-use mizaru2::{BlindedSignature, ClientToken, UnblindedSignature};
+use mizaru2::{ClientToken, UnblindedSignature};
 use stdcode::StdcodeSerializeExt;
 
 use crate::{
@@ -28,7 +28,7 @@ pub async fn auth_loop(ctx: &AnyCtx<Config>) -> anyhow::Result<()> {
         String::from_utf8_lossy(&token).to_string()
     } else {
         let auth_token = broker_client(ctx)?
-            .get_auth_token(Credential::TestDummy)
+            .get_auth_token(ctx.init().credentials.clone())
             .await??;
         db_write(ctx, "auth_token", auth_token.as_bytes()).await?;
         auth_token

binaries/geph5-client/src/client.rs

@@ -1,17 +1,15 @@
 use anyctx::AnyCtx;
 use clone_macro::clone;
 use futures_util::TryFutureExt;
+use geph5_broker_protocol::Credential;
 use smol::future::FutureExt as _;
 use std::{net::SocketAddr, path::PathBuf, time::Duration};
 
 use serde::{Deserialize, Serialize};
 use smolscale::immortal::{Immortal, RespawnStrategy};
 
 use crate::{
-    auth::{auth_loop, get_connect_token},
-    broker::BrokerSource,
-    client_inner::client_once,
-    route::ExitConstraint,
+    auth::auth_loop, broker::BrokerSource, client_inner::client_once, route::ExitConstraint,
     socks5::socks5_loop,
 };
 
@@ -21,6 +19,8 @@ pub struct Config {
     pub exit_constraint: ExitConstraint,
     pub cache: Option<PathBuf>,
     pub broker: Option<BrokerSource>,
+    #[serde(default)]
+    pub credentials: Credential,
 }
 
 pub struct Client {

binaries/geph5-client/src/database.rs

@@ -1,7 +1,7 @@
 use anyctx::AnyCtx;
 use event_listener::Event;
 use sqlx::{pool::PoolOptions, Row};
-use sqlx::{sqlite::SqliteConnectOptions, Pool, SqlitePool};
+use sqlx::{sqlite::SqliteConnectOptions, SqlitePool};
 use std::str::FromStr;
 
 use crate::client::{Config, CtxField};

binaries/geph5-client/src/socks5.rs

@@ -38,7 +38,7 @@ pub async fn socks5_loop(ctx: &AnyCtx<Config>) -> anyhow::Result<()> {
                     remote_addr = display(&remote_addr),
                     "socks5 request received"
                 );
-                let stream = open_conn(&ctx, &remote_addr).await?;
+                let stream = open_conn(ctx, &remote_addr).await?;
                 write_request_status(
                     &mut write_client,
                     SocksV5RequestStatus::Success,

binaries/geph5-exit/src/listen.rs

@@ -113,7 +113,7 @@ async fn b2e_loop() -> anyhow::Result<()> {
     loop {
         let b2e_raw = listener.accept().await?;
         let (read, write) = b2e_raw.split();
-        let mut b2e_mux = PicoMux::new(read, write);
+        let b2e_mux = PicoMux::new(read, write);
         let b2e_table = b2e_table.clone();
         smolscale::spawn::<anyhow::Result<()>>(async move {
             loop {
@@ -176,7 +176,7 @@ async fn handle_client(mut client: impl Pipe) -> anyhow::Result<()> {
     };
 
     let (client_read, client_write) = client.split();
-    let mut mux = PicoMux::new(client_read, client_write);
+    let mux = PicoMux::new(client_read, client_write);
     loop {
         let stream = mux.accept().await?;
         smolscale::spawn(proxy_stream(stream).map_err(|e| tracing::debug!("stream died with {e}")))

libraries/geph5-broker-protocol/Cargo.toml

@@ -20,3 +20,4 @@ isocountry = "0.3.2"
 language-tags = {version="0.3.2", features=["serde"]}
 serde_with = {version="3.6.1", features=["hex"]}
 mizaru2={path="../mizaru2"}
+tracing = "0.1.40"

libraries/geph5-broker-protocol/src/lib.rs

@@ -52,6 +52,7 @@ pub enum AccountLevel {
 }
 
 #[derive(Clone, Debug, Error, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
 pub enum AuthError {
     #[error("rate limited")]
     RateLimited,
@@ -62,8 +63,16 @@ pub enum AuthError {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
 pub enum Credential {
     TestDummy,
+    LegacyUsernamePassword { username: String, password: String },
+}
+
+impl Default for Credential {
+    fn default() -> Self {
+        Self::TestDummy
+    }
 }
 
 pub const DOMAIN_EXIT_DESCRIPTOR: &str = "exit-descriptor";

libraries/mizaru2/src/lib.rs

@@ -9,7 +9,7 @@ use std::{
     },
     time::SystemTime,
 };
-use stdcode::StdcodeSerializeExt;
+
 const KEY_COUNT: usize = 65536;
 const KEY_BITS: usize = 2048;
 
@@ -245,7 +245,7 @@ mod tests {
     fn test_blind_sign() {
         let secret_key = SecretKey::generate();
         let token = ClientToken::random();
-        let (blinded_digest, secret) = token.blind(&secret_key.get_subkey(0).public_key().unwrap());
+        let (blinded_digest, _secret) = token.blind(&secret_key.get_subkey(0).public_key().unwrap());
         let blinded_signature = secret_key.blind_sign(0, &blinded_digest);
 
         assert_eq!(blinded_signature.epoch, 0);

libraries/picomux/examples/picomux-socks.rs

@@ -71,7 +71,7 @@ fn main() -> anyhow::Result<()> {
                 loop {
                     let tcp_stream = listener.accept().await?;
                     let (read, write) = tcp_stream.split();
-                    let mut mux = PicoMux::new(read, write);
+                    let mux = PicoMux::new(read, write);
                     smolscale::spawn::<anyhow::Result<()>>(async move {
                         loop {
                             let client = mux.accept().await?;

libraries/picomux/src/lib.rs

@@ -611,7 +611,7 @@ mod tests {
     #[test]
     fn test_picomux_basic() {
         smolscale::block_on(async move {
-            let (picomux_a, mut picomux_b) = setup_picomux_pair().await;
+            let (picomux_a, picomux_b) = setup_picomux_pair().await;
 
             let a_proc = async move {
                 let mut stream_a = picomux_a.open(b"").await.unwrap();