MarketTakers/arbiter
5
0
Fork 0
Code Issues 27 Pull Requests 5 Packages Projects Releases Wiki Activity

merge: new flow into main

Browse Source
This commit is contained in:
hdbg
2026-03-22 12:23:07 +01:00
parent 8043cdf8d8 2148faa376
commit 4ebe7b6fc4
24 changed files with 995 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
server/crates/arbiter-server/src/actors/client/auth.rs
View File

@@ -1,6 +1,5 @@
use arbiter_proto::{
format_challenge,
transport::{Bi, expect_message},
ClientMetadata, format_challenge, transport::{Bi, expect_message}
};
use chrono::Utc;
use diesel::{
@@ -24,13 +23,6 @@ use crate::{
},
};
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientMetadata {
pub name: String,
pub description: Option<String>,
pub version: Option<String>,
}
#[derive(thiserror::Error, Debug, Clone, PartialEq, Eq)]
pub enum Error {
#[error("Database pool unavailable")]
@@ -72,9 +64,17 @@ pub enum Outbound {
AuthSuccess,
}
pub struct ClientInfo {
pub id: i32,
pub current_nonce: i32,
}
/// Atomically reads and increments the nonce for a known client.
/// Returns `None` if the pubkey is not registered.
async fn get_nonce(db: &db::DatabasePool, pubkey: &VerifyingKey) -> Result<Option<i32>, Error> {
async fn get_client_and_nonce(
db: &db::DatabasePool,
pubkey: &VerifyingKey,
) -> Result<Option<ClientInfo>, Error> {
let pubkey_bytes = pubkey.as_bytes().to_vec();
let mut conn = db.get().await.map_err(|e| {
@@ -85,10 +85,10 @@ async fn get_nonce(db: &db::DatabasePool, pubkey: &VerifyingKey) -> Result<Optio
conn.exclusive_transaction(|conn| {
let pubkey_bytes = pubkey_bytes.clone();
Box::pin(async move {
let Some(current_nonce) = program_client::table
let Some((client_id, current_nonce)) = program_client::table
.filter(program_client::public_key.eq(&pubkey_bytes))
.select(program_client::nonce)
.first::<i32>(conn)
.select((program_client::id, program_client::nonce))
.first::<(i32, i32)>(conn)
.await
.optional()?
else {
@@ -101,7 +101,10 @@ async fn get_nonce(db: &db::DatabasePool, pubkey: &VerifyingKey) -> Result<Optio
.execute(conn)
.await?;
Ok(Some(current_nonce))
Ok(Some(ClientInfo {
id: client_id,
current_nonce,
}))
})
})
.await
@@ -138,9 +141,8 @@ async fn insert_client(
db: &db::DatabasePool,
pubkey: &VerifyingKey,
metadata: &ClientMetadata,
) -> Result<(), Error> {
use crate::db::schema::client_metadata;
) -> Result<i32, Error> {
use crate::db::schema::{client_metadata, program_client};
let mut conn = db.get().await.map_err(|e| {
error!(error = ?e, "Database pool error");
Error::DatabasePoolUnavailable
@@ -160,38 +162,22 @@ async fn insert_client(
Error::DatabaseOperationFailed
})?;
insert_into(program_client::table)
let client_id = insert_into(program_client::table)
.values((
program_client::public_key.eq(pubkey.as_bytes().to_vec()),
program_client::metadata_id.eq(metadata_id),
program_client::nonce.eq(1), // pre-incremented; challenge uses 0
))
.execute(&mut conn)
.on_conflict_do_nothing()
.returning(program_client::id)
.get_result::<i32>(&mut conn)
.await
.map_err(|e| {
error!(error = ?e, "Failed to insert new client");
error!(error = ?e, "Failed to insert client metadata");
Error::DatabaseOperationFailed
})?;
Ok(())
}
async fn get_client_id(db: &db::DatabasePool, pubkey: &VerifyingKey) -> Result<Option<i32>, Error> {
let mut conn = db.get().await.map_err(|e| {
error!(error = ?e, "Database pool error");
Error::DatabasePoolUnavailable
})?;
program_client::table
.filter(program_client::public_key.eq(pubkey.as_bytes().to_vec()))
.select(program_client::id)
.first::<i32>(&mut conn)
.await
.optional()
.map_err(|e| {
error!(error = ?e, "Database error");
Error::DatabaseOperationFailed
})
Ok(client_id)
}
async fn sync_client_metadata(
@@ -312,7 +298,7 @@ where
return Err(Error::Transport);
};
let nonce = match get_nonce(&props.db, &pubkey).await? {
let info = match get_client_and_nonce(&props.db, &pubkey).await? {
Some(nonce) => nonce,
None => {
approve_new_client(
@@ -323,17 +309,18 @@ where
},
)
.await?;
insert_client(&props.db, &pubkey, &metadata).await?;
0
let client_id = insert_client(&props.db, &pubkey, &metadata).await?;
ClientInfo {
id: client_id,
current_nonce: 0,
}
}
};
let client_id = get_client_id(&props.db, &pubkey)
.await?
.ok_or(Error::DatabaseOperationFailed)?;
sync_client_metadata(&props.db, client_id, &metadata).await?;
sync_client_metadata(&props.db, info.id, &metadata).await?;
challenge_client(transport, pubkey, nonce).await?;
challenge_client(transport, pubkey, info.current_nonce).await?;
transport
.send(Ok(Outbound::AuthSuccess))
.await

4
server/crates/arbiter-server/src/actors/client/mod.rs
View File

@@ -1,9 +1,9 @@
use arbiter_proto::transport::Bi;
use arbiter_proto::{ClientMetadata, transport::Bi};
use kameo::actor::Spawn;
use tracing::{error, info};
use crate::{
actors::{GlobalActors, client::{auth::ClientMetadata, session::ClientSession}},
actors::{GlobalActors, client::{ session::ClientSession}},
db,
};

2
server/crates/arbiter-server/src/actors/user_agent/session.rs
View File

@@ -3,7 +3,7 @@ use std::{borrow::Cow, collections::HashMap};
use arbiter_proto::transport::Sender;
use async_trait::async_trait;
use ed25519_dalek::VerifyingKey;
use kameo::{Actor, actor::ActorRef, messages, prelude::Context};
use kameo::{Actor, actor::ActorRef, messages};
use thiserror::Error;
use tracing::error;

9
server/crates/arbiter-server/src/grpc/client/auth.rs
View File

@@ -1,12 +1,11 @@
use arbiter_proto::{
proto::client::{
ClientMetadata, proto::client::{
AuthChallenge as ProtoAuthChallenge, AuthChallengeRequest as ProtoAuthChallengeRequest,
AuthChallengeSolution as ProtoAuthChallengeSolution, AuthResult as ProtoAuthResult,
ClientInfo as ProtoClientInfo, ClientRequest, ClientResponse,
client_request::Payload as ClientRequestPayload,
client_response::Payload as ClientResponsePayload,
},
transport::{Bi, Error as TransportError, Receiver, Sender, grpc::GrpcBi},
}, transport::{Bi, Error as TransportError, Receiver, Sender, grpc::GrpcBi}
};
use async_trait::async_trait;
use tonic::Status;
@@ -170,8 +169,8 @@ impl Receiver<auth::Inbound> for AuthTransportAdapter<'_> {
impl Bi<auth::Inbound, Result<auth::Outbound, auth::Error>> for AuthTransportAdapter<'_> {}
fn client_metadata_from_proto(metadata: ProtoClientInfo) -> auth::ClientMetadata {
auth::ClientMetadata {
fn client_metadata_from_proto(metadata: ProtoClientInfo) -> ClientMetadata {
ClientMetadata {
name: metadata.name,
description: metadata.description,
version: metadata.version,

40
server/crates/arbiter-server/src/grpc/user_agent.rs
View File

@@ -20,15 +20,18 @@ use arbiter_proto::{
},
user_agent::{
BootstrapEncryptedKey as ProtoBootstrapEncryptedKey,
BootstrapResult as ProtoBootstrapResult, ClientConnectionCancel,
ClientConnectionRequest, UnsealEncryptedKey as ProtoUnsealEncryptedKey,
UnsealResult as ProtoUnsealResult, UnsealStart, UserAgentRequest, UserAgentResponse,
VaultState as ProtoVaultState, user_agent_request::Payload as UserAgentRequestPayload,
BootstrapResult as ProtoBootstrapResult,
SdkClientConnectionCancel as ProtoSdkClientConnectionCancel,
SdkClientConnectionRequest as ProtoSdkClientConnectionRequest,
UnsealEncryptedKey as ProtoUnsealEncryptedKey, UnsealResult as ProtoUnsealResult,
UnsealStart, UserAgentRequest, UserAgentResponse, VaultState as ProtoVaultState,
user_agent_request::Payload as UserAgentRequestPayload,
user_agent_response::Payload as UserAgentResponsePayload,
},
},
transport::{Error as TransportError, Receiver, Sender, grpc::GrpcBi},
};
use prost_types::{Timestamp as ProtoTimestamp, };
use async_trait::async_trait;
use chrono::{TimeZone, Utc};
use kameo::{
@@ -261,12 +264,14 @@ async fn dispatch_conn_message(
actor.ask(HandleGrantDelete { grant_id }).await,
))
}
UserAgentRequestPayload::ClientConnectionResponse(resp) => {
UserAgentRequestPayload::SdkClientConnectionResponse(resp) => {
let pubkey_bytes: [u8; 32] = match resp.pubkey.try_into() {
Ok(bytes) => bytes,
Err(_) => {
let _ = bi
.send(Err(Status::invalid_argument("Invalid Ed25519 public key length")))
.send(Err(Status::invalid_argument(
"Invalid Ed25519 public key length",
)))
.await;
return Err(());
}
@@ -289,13 +294,18 @@ async fn dispatch_conn_message(
.await
{
warn!(?err, "Failed to process client connection response");
let _ = bi.send(Err(Status::internal("Failed to process response"))).await;
let _ = bi
.send(Err(Status::internal("Failed to process response")))
.await;
return Err(());
}
return Ok(());
}
UserAgentRequestPayload::AuthChallengeRequest(..) | UserAgentRequestPayload::AuthChallengeSolution(..) => {
UserAgentRequestPayload::SdkClientRevoke(_sdk_client_revoke_request) => todo!(),
UserAgentRequestPayload::SdkClientList(_) => todo!(),
UserAgentRequestPayload::AuthChallengeRequest(..)
| UserAgentRequestPayload::AuthChallengeSolution(..) => {
warn!(?payload, "Unsupported post-auth user agent request");
let _ = bi
.send(Err(Status::invalid_argument(
@@ -304,7 +314,7 @@ async fn dispatch_conn_message(
.await;
return Err(());
}
};
bi.send(Ok(UserAgentResponse {
@@ -321,7 +331,7 @@ async fn send_out_of_band(
) -> Result<(), ()> {
let payload = match oob {
OutOfBand::ClientConnectionRequest { profile } => {
UserAgentResponsePayload::ClientConnectionRequest(ClientConnectionRequest {
UserAgentResponsePayload::SdkClientConnectionRequest(ProtoSdkClientConnectionRequest {
pubkey: profile.pubkey.to_bytes().to_vec(),
info: Some(ProtoClientMetadata {
name: profile.metadata.name,
@@ -331,7 +341,7 @@ async fn send_out_of_band(
})
}
OutOfBand::ClientConnectionCancel { pubkey } => {
UserAgentResponsePayload::ClientConnectionCancel(ClientConnectionCancel {
UserAgentResponsePayload::SdkClientConnectionCancel(ProtoSdkClientConnectionCancel {
pubkey: pubkey.to_bytes().to_vec(),
})
}
@@ -435,9 +445,7 @@ fn u256_from_proto_bytes(bytes: &[u8]) -> Result<U256, Status> {
Ok(U256::from_be_slice(bytes))
}
fn proto_timestamp_to_utc(
timestamp: prost_types::Timestamp,
) -> Result<chrono::DateTime<Utc>, Status> {
fn proto_timestamp_to_utc(timestamp: ProtoTimestamp) -> Result<chrono::DateTime<Utc>, Status> {
Utc.timestamp_opt(timestamp.seconds, timestamp.nanos as u32)
.single()
.ok_or_else(|| Status::invalid_argument("Invalid timestamp"))
@@ -447,11 +455,11 @@ fn shared_settings_to_proto(shared: SharedGrantSettings) -> ProtoSharedSettings
ProtoSharedSettings {
wallet_access_id: shared.wallet_access_id,
chain_id: shared.chain,
valid_from: shared.valid_from.map(|time| prost_types::Timestamp {
valid_from: shared.valid_from.map(|time| ProtoTimestamp {
seconds: time.timestamp(),
nanos: time.timestamp_subsec_nanos() as i32,
}),
valid_until: shared.valid_until.map(|time| prost_types::Timestamp {
valid_until: shared.valid_until.map(|time| ProtoTimestamp {
seconds: time.timestamp(),
nanos: time.timestamp_subsec_nanos() as i32,
}),

2
server/crates/arbiter-server/src/lib.rs
View File

@@ -1,6 +1,4 @@
#![forbid(unsafe_code)]
#![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
use crate::context::ServerContext;
pub mod actors;