Add get_connect_token function and integrate with existing flows; fix…

… database memory cache setting; re-enable client loops and integrate auth with socks5 loop
geph-official · Mar 26, 2024 · f97f6dd · f97f6dd
1 parent 0d16ee7
commit f97f6dd
Show file tree

Hide file tree

Showing 5 changed files with 31 additions and 20 deletions.
diff --git a/binaries/geph5-client/src/auth.rs b/binaries/geph5-client/src/auth.rs
@@ -4,16 +4,23 @@ use anyctx::AnyCtx;
 use anyhow::Context as _;
 use blind_rsa_signatures as brs;
 use geph5_broker_protocol::{AccountLevel, AuthError, Credential};
-use mizaru2::ClientToken;
+use mizaru2::{BlindedSignature, ClientToken, UnblindedSignature};
 use stdcode::StdcodeSerializeExt;
 
 use crate::{
     broker::broker_client,
     client::Config,
-    database::{db_read, db_write},
+    database::{db_read, db_read_or_wait, db_write},
 };
 
-// Basic workflow, we have a maintenance task that, given an auth token, refreshes the connection token for this and the next epoch, every 24 hours.
+pub async fn get_connect_token(
+    ctx: &AnyCtx<Config>,
+) -> anyhow::Result<(AccountLevel, ClientToken, UnblindedSignature)> {
+    let epoch = mizaru2::current_epoch();
+    Ok(stdcode::deserialize(
+        &db_read_or_wait(ctx, &format!("conn_token_{epoch}")).await?,
+    )?)
+}
 
 pub async fn auth_loop(ctx: &AnyCtx<Config>) -> anyhow::Result<()> {
     // Dummy authentication for now!
@@ -64,7 +71,7 @@ async fn refresh_conn_token(ctx: &AnyCtx<Config>, auth_token: &str) -> anyhow::R
                         db_write(
                             ctx,
                             &format!("conn_token_{epoch}"),
-                            &(token, u_sig).stdcode(),
+                            &(level, token, u_sig).stdcode(),
                         )
                         .await?;
                         break;

diff --git a/binaries/geph5-client/src/client.rs b/binaries/geph5-client/src/client.rs
@@ -1,13 +1,17 @@
 use anyctx::AnyCtx;
 use clone_macro::clone;
 use futures_util::TryFutureExt;
+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, broker::BrokerSource, client_inner::client_once, route::ExitConstraint,
+    auth::{auth_loop, get_connect_token},
+    broker::BrokerSource,
+    client_inner::client_once,
+    route::ExitConstraint,
     socks5::socks5_loop,
 };
 
@@ -40,15 +44,14 @@ impl Client {
 pub type CtxField<T> = fn(&AnyCtx<Config>) -> T;
 
 async fn client_main(ctx: AnyCtx<Config>) -> anyhow::Result<()> {
-    // let _client_loops: Vec<_> = (0..6)
-    //     .map(|_| {
-    //         Immortal::respawn(
-    //             RespawnStrategy::JitterDelay(Duration::from_secs(1), Duration::from_secs(5)),
-    //             clone!([ctx], move || client_once(ctx.clone())
-    //                 .inspect_err(|e| tracing::warn!("client_inner died: {:?}", e))),
-    //         )
-    //     })
-    //     .collect();
-    // socks5_loop(ctx).await
-    auth_loop(&ctx).await
+    let _client_loops: Vec<_> = (0..6)
+        .map(|_| {
+            Immortal::respawn(
+                RespawnStrategy::JitterDelay(Duration::from_secs(1), Duration::from_secs(5)),
+                clone!([ctx], move || client_once(ctx.clone())
+                    .inspect_err(|e| tracing::warn!("client_inner died: {:?}", e))),
+            )
+        })
+        .collect();
+    socks5_loop(&ctx).race(auth_loop(&ctx)).await
 }
diff --git a/binaries/geph5-client/src/database.rs b/binaries/geph5-client/src/database.rs
@@ -12,7 +12,7 @@ static DATABASE: CtxField<SqlitePool> = |ctx| {
         .cache
         .as_ref()
         .map(|s| s.to_string_lossy().to_string())
-        .unwrap_or_else(|| ":memory:".into());
+        .unwrap_or_else(|| ":memory:?cache=shared".into());
     tracing::debug!("INITIALIZING DATABASE");
     let options = SqliteConnectOptions::from_str(&db_path)
         .unwrap()

diff --git a/binaries/geph5-client/src/route.rs b/binaries/geph5-client/src/route.rs
@@ -14,7 +14,7 @@ use sillad::{
 };
 use sillad_sosistab3::{dialer::SosistabDialer, Cookie};
 
-use crate::{broker::broker_client, client::Config};
+use crate::{auth::get_connect_token, broker::broker_client, client::Config};
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 #[serde(rename_all = "snake_case")]
@@ -97,8 +97,9 @@ pub async fn get_dialer(ctx: &AnyCtx<Config>) -> anyhow::Result<(VerifyingKey, D
     };
 
     // Also obtain the bridges
+    let (_, conn_token, sig) = get_connect_token(ctx).await?;
     let bridge_routes = broker
-        .get_routes(todo!(), todo!(), exit.b2e_listen)
+        .get_routes(conn_token, sig, exit.b2e_listen)
         .await?
         .map_err(|e| anyhow::anyhow!("broker refused to serve bridge routes: {e}"))?;
     tracing::debug!(

diff --git a/binaries/geph5-client/src/socks5.rs b/binaries/geph5-client/src/socks5.rs
@@ -13,7 +13,7 @@ use std::net::Ipv4Addr;
 use super::Config;
 
 #[tracing::instrument(skip_all)]
-pub async fn socks5_loop(ctx: AnyCtx<Config>) -> anyhow::Result<()> {
+pub async fn socks5_loop(ctx: &AnyCtx<Config>) -> anyhow::Result<()> {
     let mut listener = sillad::tcp::TcpListener::bind(ctx.init().socks5_listen).await?;
     nursery!({
         loop {