Compare commits
1 Commits
push-yyxvk
...
b22be1627a
| Author | SHA1 | Date | |
|---|---|---|---|
|
b22be1627a |
@@ -1,178 +0,0 @@
|
||||
{
|
||||
"configVersion": 2,
|
||||
"packages": [
|
||||
{
|
||||
"name": "async",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/async-2.13.0",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.4"
|
||||
},
|
||||
{
|
||||
"name": "boolean_selector",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/boolean_selector-2.1.2",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.1"
|
||||
},
|
||||
{
|
||||
"name": "characters",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/characters-1.4.0",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.4"
|
||||
},
|
||||
{
|
||||
"name": "clock",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/clock-1.1.2",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.4"
|
||||
},
|
||||
{
|
||||
"name": "collection",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/collection-1.19.1",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.4"
|
||||
},
|
||||
{
|
||||
"name": "cupertino_icons",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/cupertino_icons-1.0.8",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.1"
|
||||
},
|
||||
{
|
||||
"name": "fake_async",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/fake_async-1.3.3",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.3"
|
||||
},
|
||||
{
|
||||
"name": "flutter",
|
||||
"rootUri": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable/packages/flutter",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.8"
|
||||
},
|
||||
{
|
||||
"name": "flutter_lints",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/flutter_lints-6.0.0",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.8"
|
||||
},
|
||||
{
|
||||
"name": "flutter_test",
|
||||
"rootUri": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable/packages/flutter_test",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.8"
|
||||
},
|
||||
{
|
||||
"name": "leak_tracker",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/leak_tracker-11.0.2",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.2"
|
||||
},
|
||||
{
|
||||
"name": "leak_tracker_flutter_testing",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/leak_tracker_flutter_testing-3.0.10",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.2"
|
||||
},
|
||||
{
|
||||
"name": "leak_tracker_testing",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/leak_tracker_testing-3.0.2",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.2"
|
||||
},
|
||||
{
|
||||
"name": "lints",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/lints-6.1.0",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.8"
|
||||
},
|
||||
{
|
||||
"name": "matcher",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/matcher-0.12.17",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.4"
|
||||
},
|
||||
{
|
||||
"name": "material_color_utilities",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/material_color_utilities-0.11.1",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "2.17"
|
||||
},
|
||||
{
|
||||
"name": "meta",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/meta-1.17.0",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.5"
|
||||
},
|
||||
{
|
||||
"name": "path",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/path-1.9.1",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.4"
|
||||
},
|
||||
{
|
||||
"name": "sky_engine",
|
||||
"rootUri": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable/bin/cache/pkg/sky_engine",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.8"
|
||||
},
|
||||
{
|
||||
"name": "source_span",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/source_span-1.10.2",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.1"
|
||||
},
|
||||
{
|
||||
"name": "stack_trace",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/stack_trace-1.12.1",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.4"
|
||||
},
|
||||
{
|
||||
"name": "stream_channel",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/stream_channel-2.1.4",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.3"
|
||||
},
|
||||
{
|
||||
"name": "string_scanner",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/string_scanner-1.4.1",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.1"
|
||||
},
|
||||
{
|
||||
"name": "term_glyph",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/term_glyph-1.2.2",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.1"
|
||||
},
|
||||
{
|
||||
"name": "test_api",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/test_api-0.7.7",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.5"
|
||||
},
|
||||
{
|
||||
"name": "vector_math",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/vector_math-2.2.0",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.1"
|
||||
},
|
||||
{
|
||||
"name": "vm_service",
|
||||
"rootUri": "file:///Users/kaska/.pub-cache/hosted/pub.dev/vm_service-15.0.2",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.5"
|
||||
},
|
||||
{
|
||||
"name": "app",
|
||||
"rootUri": "../",
|
||||
"packageUri": "lib/",
|
||||
"languageVersion": "3.10"
|
||||
}
|
||||
],
|
||||
"generator": "pub",
|
||||
"generatorVersion": "3.10.8",
|
||||
"flutterRoot": "file:///Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable",
|
||||
"flutterVersion": "3.38.9",
|
||||
"pubCache": "file:///Users/kaska/.pub-cache"
|
||||
}
|
||||
@@ -1,230 +0,0 @@
|
||||
{
|
||||
"roots": [
|
||||
"app"
|
||||
],
|
||||
"packages": [
|
||||
{
|
||||
"name": "app",
|
||||
"version": "1.0.0+1",
|
||||
"dependencies": [
|
||||
"cupertino_icons",
|
||||
"flutter"
|
||||
],
|
||||
"devDependencies": [
|
||||
"flutter_lints",
|
||||
"flutter_test"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "flutter_lints",
|
||||
"version": "6.0.0",
|
||||
"dependencies": [
|
||||
"lints"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "flutter_test",
|
||||
"version": "0.0.0",
|
||||
"dependencies": [
|
||||
"clock",
|
||||
"collection",
|
||||
"fake_async",
|
||||
"flutter",
|
||||
"leak_tracker_flutter_testing",
|
||||
"matcher",
|
||||
"meta",
|
||||
"path",
|
||||
"stack_trace",
|
||||
"stream_channel",
|
||||
"test_api",
|
||||
"vector_math"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "cupertino_icons",
|
||||
"version": "1.0.8",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "flutter",
|
||||
"version": "0.0.0",
|
||||
"dependencies": [
|
||||
"characters",
|
||||
"collection",
|
||||
"material_color_utilities",
|
||||
"meta",
|
||||
"sky_engine",
|
||||
"vector_math"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "lints",
|
||||
"version": "6.1.0",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "stream_channel",
|
||||
"version": "2.1.4",
|
||||
"dependencies": [
|
||||
"async"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "meta",
|
||||
"version": "1.17.0",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "collection",
|
||||
"version": "1.19.1",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "leak_tracker_flutter_testing",
|
||||
"version": "3.0.10",
|
||||
"dependencies": [
|
||||
"flutter",
|
||||
"leak_tracker",
|
||||
"leak_tracker_testing",
|
||||
"matcher",
|
||||
"meta"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "vector_math",
|
||||
"version": "2.2.0",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "stack_trace",
|
||||
"version": "1.12.1",
|
||||
"dependencies": [
|
||||
"path"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "clock",
|
||||
"version": "1.1.2",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "fake_async",
|
||||
"version": "1.3.3",
|
||||
"dependencies": [
|
||||
"clock",
|
||||
"collection"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "path",
|
||||
"version": "1.9.1",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "matcher",
|
||||
"version": "0.12.17",
|
||||
"dependencies": [
|
||||
"async",
|
||||
"meta",
|
||||
"stack_trace",
|
||||
"term_glyph",
|
||||
"test_api"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "test_api",
|
||||
"version": "0.7.7",
|
||||
"dependencies": [
|
||||
"async",
|
||||
"boolean_selector",
|
||||
"collection",
|
||||
"meta",
|
||||
"source_span",
|
||||
"stack_trace",
|
||||
"stream_channel",
|
||||
"string_scanner",
|
||||
"term_glyph"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "sky_engine",
|
||||
"version": "0.0.0",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "material_color_utilities",
|
||||
"version": "0.11.1",
|
||||
"dependencies": [
|
||||
"collection"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "characters",
|
||||
"version": "1.4.0",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "async",
|
||||
"version": "2.13.0",
|
||||
"dependencies": [
|
||||
"collection",
|
||||
"meta"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "leak_tracker_testing",
|
||||
"version": "3.0.2",
|
||||
"dependencies": [
|
||||
"leak_tracker",
|
||||
"matcher",
|
||||
"meta"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "leak_tracker",
|
||||
"version": "11.0.2",
|
||||
"dependencies": [
|
||||
"clock",
|
||||
"collection",
|
||||
"meta",
|
||||
"path",
|
||||
"vm_service"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "term_glyph",
|
||||
"version": "1.2.2",
|
||||
"dependencies": []
|
||||
},
|
||||
{
|
||||
"name": "string_scanner",
|
||||
"version": "1.4.1",
|
||||
"dependencies": [
|
||||
"source_span"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "source_span",
|
||||
"version": "1.10.2",
|
||||
"dependencies": [
|
||||
"collection",
|
||||
"path",
|
||||
"term_glyph"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "boolean_selector",
|
||||
"version": "2.1.2",
|
||||
"dependencies": [
|
||||
"source_span",
|
||||
"string_scanner"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "vm_service",
|
||||
"version": "15.0.2",
|
||||
"dependencies": []
|
||||
}
|
||||
],
|
||||
"configVersion": 1
|
||||
}
|
||||
@@ -1 +0,0 @@
|
||||
3.38.9
|
||||
@@ -1,11 +0,0 @@
|
||||
// This is a generated file; do not edit or check into version control.
|
||||
FLUTTER_ROOT=/Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable
|
||||
FLUTTER_APPLICATION_PATH=/Users/kaska/Documents/Projects/Major/arbiter/app
|
||||
COCOAPODS_PARALLEL_CODE_SIGN=true
|
||||
FLUTTER_BUILD_DIR=build
|
||||
FLUTTER_BUILD_NAME=1.0.0
|
||||
FLUTTER_BUILD_NUMBER=1
|
||||
DART_OBFUSCATION=false
|
||||
TRACK_WIDGET_CREATION=true
|
||||
TREE_SHAKE_ICONS=false
|
||||
PACKAGE_CONFIG=.dart_tool/package_config.json
|
||||
@@ -1,12 +0,0 @@
|
||||
#!/bin/sh
|
||||
# This is a generated file; do not edit or check into version control.
|
||||
export "FLUTTER_ROOT=/Users/kaska/.local/share/mise/installs/flutter/3.38.9-stable"
|
||||
export "FLUTTER_APPLICATION_PATH=/Users/kaska/Documents/Projects/Major/arbiter/app"
|
||||
export "COCOAPODS_PARALLEL_CODE_SIGN=true"
|
||||
export "FLUTTER_BUILD_DIR=build"
|
||||
export "FLUTTER_BUILD_NAME=1.0.0"
|
||||
export "FLUTTER_BUILD_NUMBER=1"
|
||||
export "DART_OBFUSCATION=false"
|
||||
export "TRACK_WIDGET_CREATION=true"
|
||||
export "TREE_SHAKE_ICONS=false"
|
||||
export "PACKAGE_CONFIG=.dart_tool/package_config.json"
|
||||
@@ -19,7 +19,6 @@ rustls-pki-types.workspace = true
|
||||
base64 = "0.22.1"
|
||||
tracing.workspace = true
|
||||
|
||||
|
||||
[build-dependencies]
|
||||
tonic-prost-build = "0.14.3"
|
||||
|
||||
|
||||
@@ -1,371 +1,125 @@
|
||||
//! Transport abstraction layer for bridging gRPC bidirectional streaming with kameo actors.
|
||||
//!
|
||||
//! This module provides a clean separation between the gRPC transport layer and business logic
|
||||
//! by modeling the connection as two linked kameo actors:
|
||||
//!
|
||||
//! - A **transport actor** ([`GrpcTransportActor`]) that owns the gRPC stream and channel,
|
||||
//! forwarding inbound messages to the business actor and outbound messages to the client.
|
||||
//! - A **business logic actor** that receives inbound messages from the transport actor and
|
||||
//! sends outbound messages back through the transport actor.
|
||||
//!
|
||||
//! The [`wire()`] function sets up bidirectional linking between the two actors, ensuring
|
||||
//! that if either actor dies, the other is notified and can shut down gracefully.
|
||||
//!
|
||||
//! # Terminology
|
||||
//!
|
||||
//! - **InboundMessage**: a message received by the transport actor from the channel/socket
|
||||
//! and forwarded to the business actor.
|
||||
//! - **OutboundMessage**: a message produced by the business actor and sent to the transport
|
||||
//! actor to be forwarded to the channel/socket.
|
||||
//!
|
||||
//! # Architecture
|
||||
//!
|
||||
//! ```text
|
||||
//! gRPC Stream ──InboundMessage──▶ GrpcTransportActor ──tell(InboundMessage)──▶ BusinessActor
|
||||
//! ▲ │
|
||||
//! └─tell(Result<OutboundMessage, _>)────┘
|
||||
//! │
|
||||
//! mpsc::Sender ──▶ Client
|
||||
//! ```
|
||||
//!
|
||||
//! # Example
|
||||
//!
|
||||
//! ```rust,ignore
|
||||
//! let (tx, rx) = mpsc::channel(1000);
|
||||
//! let context = server_context.clone();
|
||||
//!
|
||||
//! wire(
|
||||
//! |transport_ref| MyBusinessActor::new(context, transport_ref),
|
||||
//! |business_recipient, business_id| GrpcTransportActor {
|
||||
//! sender: tx,
|
||||
//! receiver: grpc_stream,
|
||||
//! business_logic_actor: business_recipient,
|
||||
//! business_logic_actor_id: business_id,
|
||||
//! },
|
||||
//! ).await;
|
||||
//!
|
||||
//! Ok(Response::new(ReceiverStream::new(rx)))
|
||||
//! ```
|
||||
use std::marker::PhantomData;
|
||||
|
||||
use futures::{Stream, StreamExt};
|
||||
use kameo::{
|
||||
Actor,
|
||||
actor::{ActorRef, PreparedActor, Recipient, Spawn, WeakActorRef},
|
||||
mailbox::Signal,
|
||||
prelude::Message,
|
||||
};
|
||||
use tokio::{
|
||||
select,
|
||||
sync::mpsc::{self, error::SendError},
|
||||
};
|
||||
use futures::StreamExt;
|
||||
use tokio::sync::mpsc;
|
||||
use tonic::{Status, Streaming};
|
||||
use tracing::{debug, error};
|
||||
|
||||
/// A bidirectional stream abstraction for sans-io testing.
|
||||
/// Errors returned by transport adapters implementing [`Bi`].
|
||||
pub enum Error {
|
||||
/// The outbound side of the transport is no longer accepting messages.
|
||||
ChannelClosed,
|
||||
}
|
||||
|
||||
/// Minimal bidirectional transport abstraction used by protocol code.
|
||||
///
|
||||
/// Combines a [`Stream`] of incoming messages with the ability to [`send`](Bi::send)
|
||||
/// outgoing responses. This trait allows business logic to be tested without a real
|
||||
/// gRPC connection by swapping in an in-memory implementation.
|
||||
/// `Bi<T, U, E>` models a duplex channel with:
|
||||
/// - inbound items of type `T` read via [`Bi::recv`]
|
||||
/// - outbound success items of type `U` or domain errors of type `E` written via [`Bi::send`]
|
||||
///
|
||||
/// # Type Parameters
|
||||
/// - `T`: `InboundMessage` received from the channel/socket (e.g., `UserAgentRequest`)
|
||||
/// - `U`: `OutboundMessage` sent to the channel/socket (e.g., `UserAgentResponse`)
|
||||
pub trait Bi<T, U>: Stream<Item = Result<T, Status>> + Send + Sync + 'static {
|
||||
type Error;
|
||||
/// The trait intentionally exposes only the operations the protocol layer needs,
|
||||
/// allowing it to work with gRPC streams and other transport implementations.
|
||||
///
|
||||
/// # Stream termination and errors
|
||||
///
|
||||
/// [`Bi::recv`] returns:
|
||||
/// - `Some(item)` when a new inbound message is available
|
||||
/// - `None` when the inbound stream ends or the underlying transport reports an error
|
||||
///
|
||||
/// Implementations may collapse transport-specific receive errors into `None`
|
||||
/// when the protocol does not need to distinguish them from normal stream
|
||||
/// termination.
|
||||
pub trait Bi<T, U, E>: Send + Sync + 'static {
|
||||
/// Sends one outbound result to the peer.
|
||||
fn send(
|
||||
&mut self,
|
||||
item: Result<U, Status>,
|
||||
) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
|
||||
item: Result<U, E>,
|
||||
) -> impl std::future::Future<Output = Result<(), Error>> + Send;
|
||||
|
||||
/// Receives the next inbound item.
|
||||
///
|
||||
/// Returns `None` when the inbound stream is finished or can no longer
|
||||
/// produce items.
|
||||
fn recv(&mut self) -> impl std::future::Future<Output = Option<T>> + Send;
|
||||
}
|
||||
|
||||
/// Concrete [`Bi`] implementation backed by a tonic gRPC [`Streaming`] and an [`mpsc::Sender`].
|
||||
/// [`Bi`] adapter backed by a tonic gRPC bidirectional stream.
|
||||
///
|
||||
/// This is the production implementation used in gRPC service handlers. The `request_stream`
|
||||
/// receives messages from the client, and `response_sender` sends responses back.
|
||||
pub struct BiStream<T, U> {
|
||||
pub request_stream: Streaming<T>,
|
||||
pub response_sender: mpsc::Sender<Result<U, Status>>,
|
||||
/// Outbound items are sent through a Tokio MPSC sender, while inbound items are
|
||||
/// read from tonic [`Streaming`].
|
||||
pub struct GrpcAdapter<Inbound, Outbound, E> {
|
||||
sender: mpsc::Sender<Result<Outbound, Status>>,
|
||||
receiver: Streaming<Inbound>,
|
||||
_error: PhantomData<E>,
|
||||
}
|
||||
|
||||
impl<T, U> Stream for BiStream<T, U>
|
||||
where
|
||||
T: Send + 'static,
|
||||
U: Send + 'static,
|
||||
{
|
||||
type Item = Result<T, Status>;
|
||||
impl<Inbound, Outbound, E> GrpcAdapter<Inbound, Outbound, E> {
|
||||
|
||||
fn poll_next(
|
||||
mut self: std::pin::Pin<&mut Self>,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
) -> std::task::Poll<Option<Self::Item>> {
|
||||
self.request_stream.poll_next_unpin(cx)
|
||||
}
|
||||
}
|
||||
|
||||
impl<T, U> Bi<T, U> for BiStream<T, U>
|
||||
where
|
||||
T: Send + 'static,
|
||||
U: Send + 'static,
|
||||
{
|
||||
type Error = SendError<Result<U, Status>>;
|
||||
|
||||
async fn send(&mut self, item: Result<U, Status>) -> Result<(), Self::Error> {
|
||||
self.response_sender.send(item).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Marker trait for transport actors that can receive outbound messages of type `T`.
|
||||
///
|
||||
/// Implement this on your transport actor to indicate it can handle outbound messages
|
||||
/// produced by the business actor. Requires the actor to implement [`Message<Result<T, E>>`]
|
||||
/// so business logic can forward responses via [`tell()`](ActorRef::tell).
|
||||
///
|
||||
/// # Example
|
||||
///
|
||||
/// ```rust,ignore
|
||||
/// #[derive(Actor)]
|
||||
/// struct MyTransportActor { /* ... */ }
|
||||
///
|
||||
/// impl Message<Result<MyResponse, MyError>> for MyTransportActor {
|
||||
/// type Reply = ();
|
||||
/// async fn handle(&mut self, msg: Result<MyResponse, MyError>, _ctx: &mut Context<Self, Self::Reply>) -> Self::Reply {
|
||||
/// // forward outbound message to channel/socket
|
||||
/// }
|
||||
/// }
|
||||
///
|
||||
/// impl TransportActor<MyResponse, MyError> for MyTransportActor {}
|
||||
/// ```
|
||||
pub trait TransportActor<Outbound: Send + 'static, DomainError: Send + 'static>:
|
||||
Actor + Send + Message<Result<Outbound, DomainError>>
|
||||
{
|
||||
}
|
||||
|
||||
/// A kameo actor that bridges a gRPC bidirectional stream with a business logic actor.
|
||||
///
|
||||
/// This actor owns the gRPC [`Streaming`] receiver and an [`mpsc::Sender`] for responses.
|
||||
/// It multiplexes between its own mailbox (for outbound messages from the business actor)
|
||||
/// and the gRPC stream (for inbound client messages) using [`tokio::select!`].
|
||||
///
|
||||
/// # Message Flow
|
||||
///
|
||||
/// - **Inbound**: Messages from the gRPC stream are forwarded to `business_logic_actor`
|
||||
/// via [`tell()`](Recipient::tell).
|
||||
/// - **Outbound**: The business actor sends `Result<Outbound, DomainError>` messages to this
|
||||
/// actor, which forwards them through the `sender` channel to the gRPC response stream.
|
||||
///
|
||||
/// # Lifecycle
|
||||
///
|
||||
/// - If the business logic actor dies (detected via actor linking), this actor stops,
|
||||
/// which closes the gRPC stream.
|
||||
/// - If the gRPC stream closes or errors, this actor stops, which (via linking) notifies
|
||||
/// the business actor.
|
||||
/// - Error responses (`Err(DomainError)`) are forwarded to the client and then the actor stops,
|
||||
/// closing the connection.
|
||||
///
|
||||
/// # Type Parameters
|
||||
/// - `Outbound`: `OutboundMessage` sent to the client (e.g., `UserAgentResponse`)
|
||||
/// - `Inbound`: `InboundMessage` received from the client (e.g., `UserAgentRequest`)
|
||||
/// - `E`: The domain error type, must implement `Into<tonic::Status>` for gRPC conversion
|
||||
pub struct GrpcTransportActor<Outbound, Inbound, DomainError>
|
||||
where
|
||||
Outbound: Send + 'static,
|
||||
Inbound: Send + 'static,
|
||||
DomainError: Into<tonic::Status> + Send + 'static,
|
||||
{
|
||||
sender: mpsc::Sender<Result<Outbound, tonic::Status>>,
|
||||
receiver: tonic::Streaming<Inbound>,
|
||||
business_logic_actor: Recipient<Inbound>,
|
||||
_error: std::marker::PhantomData<DomainError>,
|
||||
}
|
||||
|
||||
impl<Outbound, Inbound, DomainError> GrpcTransportActor<Outbound, Inbound, DomainError>
|
||||
where
|
||||
Outbound: Send + 'static,
|
||||
Inbound: Send + 'static,
|
||||
DomainError: Into<tonic::Status> + Send + 'static,
|
||||
{
|
||||
pub fn new(
|
||||
sender: mpsc::Sender<Result<Outbound, tonic::Status>>,
|
||||
receiver: tonic::Streaming<Inbound>,
|
||||
business_logic_actor: Recipient<Inbound>,
|
||||
) -> Self {
|
||||
/// Creates a new gRPC-backed [`Bi`] adapter.
|
||||
pub fn new(sender: mpsc::Sender<Result<Outbound, Status>>, receiver: Streaming<Inbound>) -> Self {
|
||||
Self {
|
||||
sender,
|
||||
receiver,
|
||||
business_logic_actor,
|
||||
_error: std::marker::PhantomData,
|
||||
_error: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Outbound, Inbound, E> Actor for GrpcTransportActor<Outbound, Inbound, E>
|
||||
impl<Inbound, Outbound, E> Bi<Inbound, Outbound, E> for GrpcAdapter<Inbound, Outbound, E>
|
||||
where
|
||||
Outbound: Send + 'static,
|
||||
Inbound: Send + 'static,
|
||||
E: Into<tonic::Status> + Send + 'static,
|
||||
Outbound: Send + 'static,
|
||||
E: Into<Status> + Send + Sync + 'static,
|
||||
{
|
||||
type Args = Self;
|
||||
|
||||
type Error = ();
|
||||
|
||||
async fn on_start(args: Self::Args, _: ActorRef<Self>) -> Result<Self, Self::Error> {
|
||||
Ok(args)
|
||||
#[tracing::instrument(level = "trace", skip(self, item))]
|
||||
async fn send(&mut self, item: Result<Outbound, E>) -> Result<(), Error> {
|
||||
self.sender
|
||||
.send(item.map_err(Into::into))
|
||||
.await
|
||||
.map_err(|_| Error::ChannelClosed)
|
||||
}
|
||||
|
||||
fn on_link_died(
|
||||
&mut self,
|
||||
_: WeakActorRef<Self>,
|
||||
id: kameo::prelude::ActorId,
|
||||
_: kameo::prelude::ActorStopReason,
|
||||
) -> impl Future<
|
||||
Output = Result<std::ops::ControlFlow<kameo::prelude::ActorStopReason>, Self::Error>,
|
||||
> + Send {
|
||||
async move {
|
||||
if id == self.business_logic_actor.id() {
|
||||
error!("Business logic actor died, stopping GrpcTransportActor");
|
||||
Ok(std::ops::ControlFlow::Break(
|
||||
kameo::prelude::ActorStopReason::Normal,
|
||||
))
|
||||
} else {
|
||||
debug!(
|
||||
"Linked actor {} died, but it's not the business logic actor, ignoring",
|
||||
id
|
||||
);
|
||||
Ok(std::ops::ControlFlow::Continue(()))
|
||||
}
|
||||
}
|
||||
#[tracing::instrument(level = "trace", skip(self))]
|
||||
async fn recv(&mut self) -> Option<Inbound> {
|
||||
self.receiver.next().await.transpose().ok().flatten()
|
||||
}
|
||||
}
|
||||
|
||||
async fn next(
|
||||
&mut self,
|
||||
_: WeakActorRef<Self>,
|
||||
mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
|
||||
) -> Option<kameo::mailbox::Signal<Self>> {
|
||||
select! {
|
||||
msg = mailbox_rx.recv() => {
|
||||
msg
|
||||
}
|
||||
recv_msg = self.receiver.next() => {
|
||||
match recv_msg {
|
||||
Some(Ok(msg)) => {
|
||||
match self.business_logic_actor.tell(msg).await {
|
||||
Ok(_) => None,
|
||||
Err(e) => {
|
||||
// TODO: this would probably require better error handling - or resending if backpressure is the issue
|
||||
error!("Failed to send message to business logic actor: {}", e);
|
||||
Some(Signal::Stop)
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(Err(e)) => {
|
||||
error!("Received error from stream: {}, stopping GrpcTransportActor", e);
|
||||
Some(Signal::Stop)
|
||||
}
|
||||
None => {
|
||||
error!("Receiver channel closed, stopping GrpcTransportActor");
|
||||
Some(Signal::Stop)
|
||||
}
|
||||
}
|
||||
}
|
||||
/// No-op [`Bi`] transport for tests and manual actor usage.
|
||||
///
|
||||
/// `send` drops all items and succeeds. [`Bi::recv`] never resolves and therefore
|
||||
/// does not busy-wait or spuriously close the stream.
|
||||
pub struct DummyTransport<T, U, E> {
|
||||
_marker: PhantomData<(T, U, E)>,
|
||||
}
|
||||
|
||||
impl<T, U, E> DummyTransport<T, U, E> {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
_marker: PhantomData,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<Outbound, Inbound, E> Message<Result<Outbound, E>> for GrpcTransportActor<Outbound, Inbound, E>
|
||||
where
|
||||
Outbound: Send + 'static,
|
||||
Inbound: Send + 'static,
|
||||
E: Into<tonic::Status> + Send + 'static,
|
||||
{
|
||||
type Reply = ();
|
||||
|
||||
async fn handle(
|
||||
&mut self,
|
||||
msg: Result<Outbound, E>,
|
||||
ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
|
||||
) -> Self::Reply {
|
||||
let is_err = msg.is_err();
|
||||
let grpc_msg = msg.map_err(Into::into);
|
||||
match self.sender.send(grpc_msg).await {
|
||||
Ok(_) => {
|
||||
if is_err {
|
||||
ctx.stop();
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to send message: {}", e);
|
||||
ctx.stop();
|
||||
}
|
||||
}
|
||||
impl<T, U, E> Default for DummyTransport<T, U, E> {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl<Outbound, Inbound, E> TransportActor<Outbound, E> for GrpcTransportActor<Outbound, Inbound, E>
|
||||
impl<T, U, E> Bi<T, U, E> for DummyTransport<T, U, E>
|
||||
where
|
||||
Outbound: Send + 'static,
|
||||
Inbound: Send + 'static,
|
||||
E: Into<tonic::Status> + Send + 'static,
|
||||
T: Send + Sync + 'static,
|
||||
U: Send + Sync + 'static,
|
||||
E: Send + Sync + 'static,
|
||||
{
|
||||
}
|
||||
|
||||
/// Wires together a transport actor and a business logic actor with bidirectional linking.
|
||||
///
|
||||
/// This function handles the chicken-and-egg problem of two actors that need references
|
||||
/// to each other at construction time. It uses kameo's [`PreparedActor`] to obtain
|
||||
/// [`ActorRef`]s before spawning, then links both actors so that if either dies,
|
||||
/// the other is notified via [`on_link_died`](Actor::on_link_died).
|
||||
///
|
||||
/// The business actor receives a type-erased [`Recipient<Result<Outbound, DomainError>>`] instead of an
|
||||
/// `ActorRef<Transport>`, keeping it decoupled from the concrete transport implementation.
|
||||
///
|
||||
/// # Type Parameters
|
||||
/// - `Transport`: The transport actor type (e.g., [`GrpcTransportActor`])
|
||||
/// - `Inbound`: `InboundMessage` received by the business actor from the transport
|
||||
/// - `Outbound`: `OutboundMessage` sent by the business actor back to the transport
|
||||
/// - `Business`: The business logic actor
|
||||
/// - `BusinessCtor`: Closure that receives a prepared business actor and transport recipient,
|
||||
/// spawns the business actor, and returns its [`ActorRef`]
|
||||
/// - `TransportCtor`: Closure that receives a prepared transport actor, a recipient for
|
||||
/// inbound messages, and the business actor id, then spawns the transport actor
|
||||
///
|
||||
/// # Returns
|
||||
/// A tuple of `(transport_ref, business_ref)` — actor references for both spawned actors.
|
||||
pub async fn wire<
|
||||
Transport,
|
||||
Inbound,
|
||||
Outbound,
|
||||
DomainError,
|
||||
Business,
|
||||
BusinessCtor,
|
||||
TransportCtor,
|
||||
>(
|
||||
business_ctor: BusinessCtor,
|
||||
transport_ctor: TransportCtor,
|
||||
) -> (ActorRef<Transport>, ActorRef<Business>)
|
||||
where
|
||||
Transport: TransportActor<Outbound, DomainError>,
|
||||
Inbound: Send + 'static,
|
||||
Outbound: Send + 'static,
|
||||
DomainError: Send + 'static,
|
||||
Business: Actor + Message<Inbound> + Send + 'static,
|
||||
BusinessCtor: FnOnce(PreparedActor<Business>, Recipient<Result<Outbound, DomainError>>),
|
||||
TransportCtor:
|
||||
FnOnce(PreparedActor<Transport>, Recipient<Inbound>),
|
||||
{
|
||||
let prepared_business: PreparedActor<Business> = Spawn::prepare();
|
||||
let prepared_transport: PreparedActor<Transport> = Spawn::prepare();
|
||||
|
||||
let business_ref = prepared_business.actor_ref().clone();
|
||||
let transport_ref = prepared_transport.actor_ref().clone();
|
||||
|
||||
transport_ref.link(&business_ref).await;
|
||||
business_ref.link(&transport_ref).await;
|
||||
|
||||
let recipient = transport_ref.clone().recipient();
|
||||
business_ctor(prepared_business, recipient);
|
||||
let business_recipient = business_ref.clone().recipient();
|
||||
transport_ctor(prepared_transport, business_recipient);
|
||||
|
||||
|
||||
(transport_ref, business_ref)
|
||||
async fn send(&mut self, _item: Result<U, E>) -> Result<(), Error> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn recv(&mut self) -> impl std::future::Future<Output = Option<T>> + Send {
|
||||
async {
|
||||
std::future::pending::<()>().await;
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,6 +7,6 @@ use crate::ServerContext;
|
||||
|
||||
pub(crate) async fn handle_client(
|
||||
_context: ServerContext,
|
||||
_bistream: impl Bi<ClientRequest, ClientResponse>,
|
||||
_bistream: impl Bi<ClientRequest, ClientResponse, tonic::Status>,
|
||||
) {
|
||||
}
|
||||
|
||||
@@ -1,57 +0,0 @@
|
||||
use tonic::Status;
|
||||
|
||||
use crate::db;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum UserAgentError {
|
||||
#[error("Missing payload in request")]
|
||||
MissingPayload,
|
||||
|
||||
#[error("Invalid bootstrap token")]
|
||||
InvalidBootstrapToken,
|
||||
|
||||
#[error("Public key not registered")]
|
||||
PubkeyNotRegistered,
|
||||
|
||||
#[error("Invalid public key format")]
|
||||
InvalidPubkey,
|
||||
|
||||
#[error("Invalid signature length")]
|
||||
InvalidSignatureLength,
|
||||
|
||||
#[error("Invalid challenge solution")]
|
||||
InvalidChallengeSolution,
|
||||
|
||||
#[error("Invalid state for operation")]
|
||||
InvalidState,
|
||||
|
||||
#[error("Actor unavailable")]
|
||||
ActorUnavailable,
|
||||
|
||||
#[error("Database error")]
|
||||
Database(#[from] diesel::result::Error),
|
||||
|
||||
#[error("Database pool error")]
|
||||
DatabasePool(#[from] db::PoolError),
|
||||
}
|
||||
|
||||
impl From<UserAgentError> for Status {
|
||||
fn from(err: UserAgentError) -> Self {
|
||||
match err {
|
||||
UserAgentError::MissingPayload
|
||||
| UserAgentError::InvalidBootstrapToken
|
||||
| UserAgentError::InvalidPubkey
|
||||
| UserAgentError::InvalidSignatureLength => Status::invalid_argument(err.to_string()),
|
||||
|
||||
UserAgentError::PubkeyNotRegistered | UserAgentError::InvalidChallengeSolution => {
|
||||
Status::unauthenticated(err.to_string())
|
||||
}
|
||||
|
||||
UserAgentError::InvalidState => Status::failed_precondition(err.to_string()),
|
||||
|
||||
UserAgentError::ActorUnavailable
|
||||
| UserAgentError::Database(_)
|
||||
| UserAgentError::DatabasePool(_) => Status::internal(err.to_string()),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,23 +1,27 @@
|
||||
use std::{ops::DerefMut, sync::Mutex};
|
||||
|
||||
use arbiter_proto::proto::{
|
||||
UnsealEncryptedKey, UnsealResult, UnsealStart, UnsealStartResponse, UserAgentRequest,
|
||||
UserAgentResponse,
|
||||
auth::{
|
||||
self, AuthChallengeRequest, AuthOk, ClientMessage as ClientAuthMessage,
|
||||
ServerMessage as AuthServerMessage,
|
||||
client_message::Payload as ClientAuthPayload,
|
||||
server_message::Payload as ServerAuthPayload,
|
||||
use arbiter_proto::{
|
||||
proto::{
|
||||
UnsealEncryptedKey, UnsealResult, UnsealStart, UnsealStartResponse, UserAgentRequest,
|
||||
UserAgentResponse,
|
||||
auth::{
|
||||
self, AuthChallengeRequest, AuthOk, ClientMessage as ClientAuthMessage,
|
||||
ServerMessage as AuthServerMessage, client_message::Payload as ClientAuthPayload,
|
||||
server_message::Payload as ServerAuthPayload,
|
||||
},
|
||||
user_agent_request::Payload as UserAgentRequestPayload,
|
||||
user_agent_response::Payload as UserAgentResponsePayload,
|
||||
},
|
||||
user_agent_request::Payload as UserAgentRequestPayload,
|
||||
user_agent_response::Payload as UserAgentResponsePayload,
|
||||
transport::{Bi, DummyTransport},
|
||||
};
|
||||
use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
|
||||
use diesel::{ExpressionMethods as _, OptionalExtension as _, QueryDsl, dsl::update};
|
||||
use diesel_async::RunQueryDsl;
|
||||
use ed25519_dalek::VerifyingKey;
|
||||
use kameo::{Actor, actor::Recipient, error::SendError, messages, prelude::Message};
|
||||
use kameo::{Actor, error::SendError};
|
||||
use memsafe::MemSafe;
|
||||
use tokio::select;
|
||||
use tonic::Status;
|
||||
use tracing::{error, info};
|
||||
use x25519_dalek::{EphemeralSecret, PublicKey};
|
||||
|
||||
@@ -35,24 +39,103 @@ use crate::{
|
||||
db::{self, schema},
|
||||
};
|
||||
|
||||
mod error;
|
||||
mod state;
|
||||
|
||||
pub use error::UserAgentError;
|
||||
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
|
||||
pub enum UserAgentError {
|
||||
#[error("Expected message with payload")]
|
||||
MissingRequestPayload,
|
||||
#[error("Expected message with payload")]
|
||||
UnexpectedRequestPayload,
|
||||
#[error("Invalid state for challenge solution")]
|
||||
InvalidStateForChallengeSolution,
|
||||
#[error("Invalid state for unseal encrypted key")]
|
||||
InvalidStateForUnsealEncryptedKey,
|
||||
#[error("client_pubkey must be 32 bytes")]
|
||||
InvalidClientPubkeyLength,
|
||||
#[error("Expected pubkey to have specific length")]
|
||||
InvalidAuthPubkeyLength,
|
||||
#[error("Failed to convert pubkey to VerifyingKey")]
|
||||
InvalidAuthPubkeyEncoding,
|
||||
#[error("Invalid signature length")]
|
||||
InvalidSignatureLength,
|
||||
#[error("Invalid bootstrap token")]
|
||||
InvalidBootstrapToken,
|
||||
#[error("Public key not registered")]
|
||||
PublicKeyNotRegistered,
|
||||
#[error("Invalid challenge solution")]
|
||||
InvalidChallengeSolution,
|
||||
#[error("State machine error")]
|
||||
StateTransitionFailed,
|
||||
#[error("Bootstrap token consumption failed")]
|
||||
BootstrapperActorUnreachable,
|
||||
#[error("Vault is not available")]
|
||||
KeyHolderActorUnreachable,
|
||||
#[error("Database pool error")]
|
||||
DatabasePoolUnavailable,
|
||||
#[error("Database error")]
|
||||
DatabaseOperationFailed,
|
||||
}
|
||||
|
||||
#[derive(Actor)]
|
||||
pub struct UserAgentActor {
|
||||
impl From<UserAgentError> for Status {
|
||||
fn from(value: UserAgentError) -> Self {
|
||||
match value {
|
||||
UserAgentError::MissingRequestPayload | UserAgentError::UnexpectedRequestPayload => {
|
||||
Status::invalid_argument("Expected message with payload")
|
||||
}
|
||||
UserAgentError::InvalidStateForChallengeSolution => {
|
||||
Status::invalid_argument("Invalid state for challenge solution")
|
||||
}
|
||||
UserAgentError::InvalidStateForUnsealEncryptedKey => {
|
||||
Status::failed_precondition("Invalid state for unseal encrypted key")
|
||||
}
|
||||
UserAgentError::InvalidClientPubkeyLength => {
|
||||
Status::invalid_argument("client_pubkey must be 32 bytes")
|
||||
}
|
||||
UserAgentError::InvalidAuthPubkeyLength => {
|
||||
Status::invalid_argument("Expected pubkey to have specific length")
|
||||
}
|
||||
UserAgentError::InvalidAuthPubkeyEncoding => {
|
||||
Status::invalid_argument("Failed to convert pubkey to VerifyingKey")
|
||||
}
|
||||
UserAgentError::InvalidSignatureLength => {
|
||||
Status::invalid_argument("Invalid signature length")
|
||||
}
|
||||
UserAgentError::InvalidBootstrapToken => {
|
||||
Status::invalid_argument("Invalid bootstrap token")
|
||||
}
|
||||
UserAgentError::PublicKeyNotRegistered => {
|
||||
Status::unauthenticated("Public key not registered")
|
||||
}
|
||||
UserAgentError::InvalidChallengeSolution => {
|
||||
Status::unauthenticated("Invalid challenge solution")
|
||||
}
|
||||
UserAgentError::StateTransitionFailed => Status::internal("State machine error"),
|
||||
UserAgentError::BootstrapperActorUnreachable => {
|
||||
Status::internal("Bootstrap token consumption failed")
|
||||
}
|
||||
UserAgentError::KeyHolderActorUnreachable => Status::internal("Vault is not available"),
|
||||
UserAgentError::DatabasePoolUnavailable => Status::internal("Database pool error"),
|
||||
UserAgentError::DatabaseOperationFailed => Status::internal("Database error"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct UserAgentActor<Transport>
|
||||
where
|
||||
Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
|
||||
{
|
||||
db: db::DatabasePool,
|
||||
actors: GlobalActors,
|
||||
state: UserAgentStateMachine<DummyContext>,
|
||||
transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
|
||||
transport: Transport,
|
||||
}
|
||||
|
||||
impl UserAgentActor {
|
||||
pub(crate) fn new(
|
||||
context: ServerContext,
|
||||
transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
|
||||
) -> Self {
|
||||
impl<Transport> UserAgentActor<Transport>
|
||||
where
|
||||
Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
|
||||
{
|
||||
pub(crate) fn new(context: ServerContext, transport: Transport) -> Self {
|
||||
Self {
|
||||
db: context.db.clone(),
|
||||
actors: context.actors.clone(),
|
||||
@@ -61,23 +144,18 @@ impl UserAgentActor {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn new_manual(
|
||||
db: db::DatabasePool,
|
||||
actors: GlobalActors,
|
||||
transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
db,
|
||||
actors,
|
||||
state: UserAgentStateMachine::new(DummyContext),
|
||||
transport,
|
||||
}
|
||||
fn transition(&mut self, event: UserAgentEvents) -> Result<(), UserAgentError> {
|
||||
self.state.process_event(event).map_err(|e| {
|
||||
error!(?e, "State transition failed");
|
||||
UserAgentError::StateTransitionFailed
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn process_request(&mut self, req: UserAgentRequest) -> Output {
|
||||
pub async fn process_transport_inbound(&mut self, req: UserAgentRequest) -> Output {
|
||||
let msg = req.payload.ok_or_else(|| {
|
||||
error!(actor = "useragent", "Received message with no payload");
|
||||
UserAgentError::MissingPayload
|
||||
UserAgentError::MissingRequestPayload
|
||||
})?;
|
||||
|
||||
match msg {
|
||||
@@ -93,23 +171,15 @@ impl UserAgentActor {
|
||||
UserAgentRequestPayload::UnsealEncryptedKey(unseal_encrypted_key) => {
|
||||
self.handle_unseal_encrypted_key(unseal_encrypted_key).await
|
||||
}
|
||||
_ => Err(UserAgentError::MissingPayload),
|
||||
_ => Err(UserAgentError::UnexpectedRequestPayload),
|
||||
}
|
||||
}
|
||||
|
||||
fn transition(&mut self, event: UserAgentEvents) -> Result<(), UserAgentError> {
|
||||
self.state.process_event(event).map_err(|e| {
|
||||
error!(?e, "State transition failed");
|
||||
UserAgentError::InvalidState
|
||||
})?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn auth_with_bootstrap_token(
|
||||
&mut self,
|
||||
pubkey: ed25519_dalek::VerifyingKey,
|
||||
token: String,
|
||||
) -> Output {
|
||||
) -> Result<UserAgentResponse, UserAgentError> {
|
||||
let token_ok: bool = self
|
||||
.actors
|
||||
.bootstrapper
|
||||
@@ -117,7 +187,7 @@ impl UserAgentActor {
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!(?pubkey, "Failed to consume bootstrap token: {e}");
|
||||
UserAgentError::ActorUnavailable
|
||||
UserAgentError::BootstrapperActorUnreachable
|
||||
})?;
|
||||
|
||||
if !token_ok {
|
||||
@@ -126,7 +196,10 @@ impl UserAgentActor {
|
||||
}
|
||||
|
||||
{
|
||||
let mut conn = self.db.get().await?;
|
||||
let mut conn = self.db.get().await.map_err(|e| {
|
||||
error!(error = ?e, "Database pool error");
|
||||
UserAgentError::DatabasePoolUnavailable
|
||||
})?;
|
||||
|
||||
diesel::insert_into(schema::useragent_client::table)
|
||||
.values((
|
||||
@@ -134,7 +207,11 @@ impl UserAgentActor {
|
||||
schema::useragent_client::nonce.eq(1),
|
||||
))
|
||||
.execute(&mut conn)
|
||||
.await?;
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!(error = ?e, "Database error");
|
||||
UserAgentError::DatabaseOperationFailed
|
||||
})?;
|
||||
}
|
||||
|
||||
self.transition(UserAgentEvents::ReceivedBootstrapToken)?;
|
||||
@@ -144,7 +221,10 @@ impl UserAgentActor {
|
||||
|
||||
async fn auth_with_challenge(&mut self, pubkey: VerifyingKey, pubkey_bytes: Vec<u8>) -> Output {
|
||||
let nonce: Option<i32> = {
|
||||
let mut db_conn = self.db.get().await?;
|
||||
let mut db_conn = self.db.get().await.map_err(|e| {
|
||||
error!(error = ?e, "Database pool error");
|
||||
UserAgentError::DatabasePoolUnavailable
|
||||
})?;
|
||||
db_conn
|
||||
.exclusive_transaction(|conn| {
|
||||
Box::pin(async move {
|
||||
@@ -168,12 +248,16 @@ impl UserAgentActor {
|
||||
})
|
||||
})
|
||||
.await
|
||||
.optional()?
|
||||
.optional()
|
||||
.map_err(|e| {
|
||||
error!(error = ?e, "Database error");
|
||||
UserAgentError::DatabaseOperationFailed
|
||||
})?
|
||||
};
|
||||
|
||||
let Some(nonce) = nonce else {
|
||||
error!(?pubkey, "Public key not found in database");
|
||||
return Err(UserAgentError::PubkeyNotRegistered);
|
||||
return Err(UserAgentError::PublicKeyNotRegistered);
|
||||
};
|
||||
|
||||
let challenge = auth::AuthChallenge {
|
||||
@@ -202,7 +286,7 @@ impl UserAgentActor {
|
||||
let UserAgentStates::WaitingForChallengeSolution(challenge_context) = self.state.state()
|
||||
else {
|
||||
error!("Received challenge solution in invalid state");
|
||||
return Err(UserAgentError::InvalidState);
|
||||
return Err(UserAgentError::InvalidStateForChallengeSolution);
|
||||
};
|
||||
let formatted_challenge = arbiter_proto::format_challenge(&challenge_context.challenge);
|
||||
|
||||
@@ -236,17 +320,18 @@ fn unseal_response(payload: UserAgentResponsePayload) -> UserAgentResponse {
|
||||
}
|
||||
}
|
||||
|
||||
#[messages]
|
||||
impl UserAgentActor {
|
||||
#[message]
|
||||
pub async fn handle_unseal_request(&mut self, req: UnsealStart) -> Output {
|
||||
impl<Transport> UserAgentActor<Transport>
|
||||
where
|
||||
Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
|
||||
{
|
||||
async fn handle_unseal_request(&mut self, req: UnsealStart) -> Output {
|
||||
let secret = EphemeralSecret::random();
|
||||
let public_key = PublicKey::from(&secret);
|
||||
|
||||
let client_pubkey_bytes: [u8; 32] = req
|
||||
.client_pubkey
|
||||
.try_into()
|
||||
.map_err(|_| UserAgentError::InvalidPubkey)?;
|
||||
.map_err(|_| UserAgentError::InvalidClientPubkeyLength)?;
|
||||
|
||||
let client_public_key = PublicKey::from(client_pubkey_bytes);
|
||||
|
||||
@@ -262,11 +347,10 @@ impl UserAgentActor {
|
||||
))
|
||||
}
|
||||
|
||||
#[message]
|
||||
pub async fn handle_unseal_encrypted_key(&mut self, req: UnsealEncryptedKey) -> Output {
|
||||
async fn handle_unseal_encrypted_key(&mut self, req: UnsealEncryptedKey) -> Output {
|
||||
let UserAgentStates::WaitingForUnsealKey(unseal_context) = self.state.state() else {
|
||||
error!("Received unseal encrypted key in invalid state");
|
||||
return Err(UserAgentError::InvalidState);
|
||||
return Err(UserAgentError::InvalidStateForUnsealEncryptedKey);
|
||||
};
|
||||
let ephemeral_secret = {
|
||||
let mut secret_lock = unseal_context.secret.lock().unwrap();
|
||||
@@ -330,7 +414,7 @@ impl UserAgentActor {
|
||||
Err(err) => {
|
||||
error!(?err, "Failed to send unseal request to keyholder");
|
||||
self.transition(UserAgentEvents::ReceivedInvalidKey)?;
|
||||
Err(UserAgentError::ActorUnavailable)
|
||||
Err(UserAgentError::KeyHolderActorUnreachable)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -344,15 +428,14 @@ impl UserAgentActor {
|
||||
}
|
||||
}
|
||||
|
||||
#[message]
|
||||
pub async fn handle_auth_challenge_request(&mut self, req: AuthChallengeRequest) -> Output {
|
||||
async fn handle_auth_challenge_request(&mut self, req: AuthChallengeRequest) -> Output {
|
||||
let pubkey = req
|
||||
.pubkey
|
||||
.as_array()
|
||||
.ok_or(UserAgentError::InvalidPubkey)?;
|
||||
.ok_or(UserAgentError::InvalidAuthPubkeyLength)?;
|
||||
let pubkey = VerifyingKey::from_bytes(pubkey).map_err(|_err| {
|
||||
error!(?pubkey, "Failed to convert to VerifyingKey");
|
||||
UserAgentError::InvalidPubkey
|
||||
UserAgentError::InvalidAuthPubkeyEncoding
|
||||
})?;
|
||||
|
||||
self.transition(UserAgentEvents::AuthRequest)?;
|
||||
@@ -363,8 +446,7 @@ impl UserAgentActor {
|
||||
}
|
||||
}
|
||||
|
||||
#[message]
|
||||
pub async fn handle_auth_challenge_solution(
|
||||
async fn handle_auth_challenge_solution(
|
||||
&mut self,
|
||||
solution: auth::AuthChallengeSolution,
|
||||
) -> Output {
|
||||
@@ -385,17 +467,67 @@ impl UserAgentActor {
|
||||
}
|
||||
}
|
||||
|
||||
impl Message<UserAgentRequest> for UserAgentActor {
|
||||
type Reply = ();
|
||||
|
||||
async fn handle(
|
||||
impl<Transport> Actor for UserAgentActor<Transport>
|
||||
where
|
||||
Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
|
||||
{
|
||||
type Args = Self;
|
||||
|
||||
type Error = ();
|
||||
|
||||
async fn on_start(
|
||||
args: Self::Args,
|
||||
_: kameo::prelude::ActorRef<Self>,
|
||||
) -> Result<Self, Self::Error> {
|
||||
Ok(args)
|
||||
}
|
||||
|
||||
async fn next(
|
||||
&mut self,
|
||||
msg: UserAgentRequest,
|
||||
_ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
|
||||
) -> Self::Reply {
|
||||
let result = self.process_request(msg).await;
|
||||
if let Err(e) = self.transport.tell(result).await {
|
||||
error!(actor = "useragent", "Failed to send response to transport: {}", e);
|
||||
_actor_ref: kameo::prelude::WeakActorRef<Self>,
|
||||
mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
|
||||
) -> Option<kameo::mailbox::Signal<Self>> {
|
||||
loop {
|
||||
select! {
|
||||
signal = mailbox_rx.recv() => {
|
||||
return signal;
|
||||
}
|
||||
msg = self.transport.recv() => {
|
||||
match msg {
|
||||
Some(request) => {
|
||||
match self.process_transport_inbound(request).await {
|
||||
Ok(response) => {
|
||||
if self.transport.send(Ok(response)).await.is_err() {
|
||||
error!(actor = "useragent", reason = "channel closed", "send.failed");
|
||||
return Some(kameo::mailbox::Signal::Stop);
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
let _ = self.transport.send(Err(err)).await;
|
||||
return Some(kameo::mailbox::Signal::Stop);
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
info!(actor = "useragent", "transport.closed");
|
||||
return Some(kameo::mailbox::Signal::Stop);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
impl UserAgentActor<DummyTransport<UserAgentRequest, UserAgentResponse, UserAgentError>> {
|
||||
pub fn new_manual(db: db::DatabasePool, actors: GlobalActors) -> Self {
|
||||
Self {
|
||||
db,
|
||||
actors,
|
||||
state: UserAgentStateMachine::new(DummyContext),
|
||||
transport: DummyTransport::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
24
server/crates/arbiter-server/src/errors.rs
Normal file
24
server/crates/arbiter-server/src/errors.rs
Normal file
@@ -0,0 +1,24 @@
|
||||
use tonic::Status;
|
||||
use tracing::error;
|
||||
|
||||
pub trait GrpcStatusExt<T> {
|
||||
fn to_status(self) -> Result<T, Status>;
|
||||
}
|
||||
|
||||
impl<T> GrpcStatusExt<T> for Result<T, diesel::result::Error> {
|
||||
fn to_status(self) -> Result<T, Status> {
|
||||
self.map_err(|e| {
|
||||
error!(error = ?e, "Database error");
|
||||
Status::internal("Database error")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> GrpcStatusExt<T> for Result<T, crate::db::PoolError> {
|
||||
fn to_status(self) -> Result<T, Status> {
|
||||
self.map_err(|e| {
|
||||
error!(error = ?e, "Database pool error");
|
||||
Status::internal("Database pool error")
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,26 +1,25 @@
|
||||
#![forbid(unsafe_code)]
|
||||
use arbiter_proto::{
|
||||
proto::{ClientRequest, ClientResponse, UserAgentRequest, UserAgentResponse},
|
||||
transport::{BiStream, GrpcTransportActor, wire},
|
||||
transport::GrpcAdapter,
|
||||
};
|
||||
use async_trait::async_trait;
|
||||
use kameo::actor::PreparedActor;
|
||||
use kameo::actor::Spawn;
|
||||
use tokio_stream::wrappers::ReceiverStream;
|
||||
|
||||
use tokio::sync::mpsc;
|
||||
use tonic::{Request, Response, Status};
|
||||
use tracing::info;
|
||||
|
||||
use crate::{
|
||||
actors::{
|
||||
client::handle_client,
|
||||
user_agent::UserAgentActor,
|
||||
},
|
||||
actors::user_agent::UserAgentActor,
|
||||
context::ServerContext,
|
||||
};
|
||||
|
||||
pub mod actors;
|
||||
pub mod context;
|
||||
pub mod db;
|
||||
mod errors;
|
||||
|
||||
const DEFAULT_CHANNEL_SIZE: usize = 1000;
|
||||
|
||||
@@ -41,42 +40,24 @@ impl arbiter_proto::proto::arbiter_service_server::ArbiterService for Server {
|
||||
|
||||
async fn client(
|
||||
&self,
|
||||
request: Request<tonic::Streaming<ClientRequest>>,
|
||||
_request: Request<tonic::Streaming<ClientRequest>>,
|
||||
) -> Result<Response<Self::ClientStream>, Status> {
|
||||
let req_stream = request.into_inner();
|
||||
let (tx, rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
|
||||
tokio::spawn(handle_client(
|
||||
self.context.clone(),
|
||||
BiStream {
|
||||
request_stream: req_stream,
|
||||
response_sender: tx,
|
||||
},
|
||||
));
|
||||
|
||||
Ok(Response::new(ReceiverStream::new(rx)))
|
||||
todo!()
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip(self))]
|
||||
async fn user_agent(
|
||||
&self,
|
||||
request: Request<tonic::Streaming<UserAgentRequest>>,
|
||||
) -> Result<Response<Self::UserAgentStream>, Status> {
|
||||
let req_stream = request.into_inner();
|
||||
let (tx, rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
|
||||
let context = self.context.clone();
|
||||
|
||||
wire(
|
||||
|prepared: PreparedActor<UserAgentActor>, recipient| {
|
||||
prepared.spawn(UserAgentActor::new(context, recipient));
|
||||
},
|
||||
|prepared: PreparedActor<GrpcTransportActor<_, _, _>>, business_recipient| {
|
||||
prepared.spawn(GrpcTransportActor::new(
|
||||
tx,
|
||||
req_stream,
|
||||
business_recipient,
|
||||
));
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let adapter = GrpcAdapter::new(tx, req_stream);
|
||||
|
||||
UserAgentActor::spawn(UserAgentActor::new(self.context.clone(), adapter));
|
||||
|
||||
info!(event = "connection established", "grpc.user_agent");
|
||||
|
||||
Ok(Response::new(ReceiverStream::new(rx)))
|
||||
}
|
||||
|
||||
@@ -1,30 +1,5 @@
|
||||
mod common;
|
||||
|
||||
use arbiter_proto::proto::UserAgentResponse;
|
||||
use arbiter_server::actors::user_agent::UserAgentError;
|
||||
use kameo::{Actor, actor::Recipient, actor::Spawn, prelude::Message};
|
||||
|
||||
/// A no-op actor that discards any messages it receives.
|
||||
#[derive(Actor)]
|
||||
struct NullSink;
|
||||
|
||||
impl Message<Result<UserAgentResponse, UserAgentError>> for NullSink {
|
||||
type Reply = ();
|
||||
|
||||
async fn handle(
|
||||
&mut self,
|
||||
_msg: Result<UserAgentResponse, UserAgentError>,
|
||||
_ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
|
||||
) -> Self::Reply {
|
||||
}
|
||||
}
|
||||
|
||||
/// Creates a `Recipient` that silently discards all messages.
|
||||
fn null_recipient() -> Recipient<Result<UserAgentResponse, UserAgentError>> {
|
||||
let actor_ref = NullSink::spawn(NullSink);
|
||||
actor_ref.recipient()
|
||||
}
|
||||
|
||||
#[path = "user_agent/auth.rs"]
|
||||
mod auth;
|
||||
#[path = "user_agent/unseal.rs"]
|
||||
|
||||
@@ -1,20 +1,29 @@
|
||||
use arbiter_proto::proto::{
|
||||
UserAgentResponse,
|
||||
auth::{self, AuthChallengeRequest, AuthOk},
|
||||
UserAgentRequest,
|
||||
auth::{self, AuthChallengeRequest, AuthOk, ClientMessage, client_message::Payload as ClientAuthPayload},
|
||||
user_agent_request::Payload as UserAgentRequestPayload,
|
||||
user_agent_response::Payload as UserAgentResponsePayload,
|
||||
};
|
||||
use arbiter_server::{
|
||||
actors::{
|
||||
GlobalActors,
|
||||
bootstrap::GetToken,
|
||||
user_agent::{HandleAuthChallengeRequest, HandleAuthChallengeSolution, UserAgentActor},
|
||||
user_agent::{UserAgentActor, UserAgentError},
|
||||
},
|
||||
db::{self, schema},
|
||||
};
|
||||
use diesel::{ExpressionMethods as _, QueryDsl, insert_into};
|
||||
use diesel_async::RunQueryDsl;
|
||||
use ed25519_dalek::Signer as _;
|
||||
use kameo::actor::Spawn;
|
||||
|
||||
fn auth_request(payload: ClientAuthPayload) -> UserAgentRequest {
|
||||
UserAgentRequest {
|
||||
payload: Some(UserAgentRequestPayload::AuthMessage(ClientMessage {
|
||||
payload: Some(payload),
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[test_log::test]
|
||||
@@ -23,22 +32,20 @@ pub async fn test_bootstrap_token_auth() {
|
||||
|
||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||
let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
|
||||
let user_agent =
|
||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
||||
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||
|
||||
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
||||
|
||||
let result = user_agent_ref
|
||||
.ask(HandleAuthChallengeRequest {
|
||||
req: AuthChallengeRequest {
|
||||
let result = user_agent
|
||||
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||
AuthChallengeRequest {
|
||||
pubkey: pubkey_bytes,
|
||||
bootstrap_token: Some(token),
|
||||
},
|
||||
})
|
||||
)))
|
||||
.await
|
||||
.expect("Shouldn't fail to send message");
|
||||
.expect("Shouldn't fail to process message");
|
||||
|
||||
assert_eq!(
|
||||
result,
|
||||
@@ -68,31 +75,25 @@ pub async fn test_bootstrap_invalid_token_auth() {
|
||||
let db = db::create_test_pool().await;
|
||||
|
||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||
let user_agent =
|
||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
||||
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||
|
||||
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
||||
|
||||
let result = user_agent_ref
|
||||
.ask(HandleAuthChallengeRequest {
|
||||
req: AuthChallengeRequest {
|
||||
let result = user_agent
|
||||
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||
AuthChallengeRequest {
|
||||
pubkey: pubkey_bytes,
|
||||
bootstrap_token: Some("invalid_token".to_string()),
|
||||
},
|
||||
})
|
||||
)))
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Err(kameo::error::SendError::HandlerError(err)) => {
|
||||
assert!(
|
||||
matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidBootstrapToken),
|
||||
"Expected InvalidBootstrapToken, got {err:?}"
|
||||
);
|
||||
}
|
||||
Err(other) => {
|
||||
panic!("Expected SendError::HandlerError, got {other:?}");
|
||||
Err(err) => {
|
||||
assert_eq!(err, UserAgentError::InvalidBootstrapToken);
|
||||
let status: tonic::Status = err.into();
|
||||
assert_eq!(status.code(), tonic::Code::InvalidArgument);
|
||||
}
|
||||
Ok(_) => {
|
||||
panic!("Expected error due to invalid bootstrap token, but got success");
|
||||
@@ -106,9 +107,7 @@ pub async fn test_challenge_auth() {
|
||||
let db = db::create_test_pool().await;
|
||||
|
||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||
let user_agent =
|
||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
||||
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||
|
||||
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
||||
@@ -122,15 +121,15 @@ pub async fn test_challenge_auth() {
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let result = user_agent_ref
|
||||
.ask(HandleAuthChallengeRequest {
|
||||
req: AuthChallengeRequest {
|
||||
let result = user_agent
|
||||
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||
AuthChallengeRequest {
|
||||
pubkey: pubkey_bytes,
|
||||
bootstrap_token: None,
|
||||
},
|
||||
})
|
||||
)))
|
||||
.await
|
||||
.expect("Shouldn't fail to send message");
|
||||
.expect("Shouldn't fail to process message");
|
||||
|
||||
let UserAgentResponse {
|
||||
payload:
|
||||
@@ -147,14 +146,14 @@ pub async fn test_challenge_auth() {
|
||||
let signature = new_key.sign(&formatted_challenge);
|
||||
let serialized_signature = signature.to_bytes().to_vec();
|
||||
|
||||
let result = user_agent_ref
|
||||
.ask(HandleAuthChallengeSolution {
|
||||
solution: auth::AuthChallengeSolution {
|
||||
let result = user_agent
|
||||
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeSolution(
|
||||
auth::AuthChallengeSolution {
|
||||
signature: serialized_signature,
|
||||
},
|
||||
})
|
||||
)))
|
||||
.await
|
||||
.expect("Shouldn't fail to send message");
|
||||
.expect("Shouldn't fail to process message");
|
||||
|
||||
assert_eq!(
|
||||
result,
|
||||
|
||||
@@ -1,27 +1,51 @@
|
||||
use arbiter_proto::proto::{
|
||||
UnsealEncryptedKey, UnsealResult, UnsealStart, auth::AuthChallengeRequest,
|
||||
UnsealEncryptedKey, UnsealResult, UnsealStart, UserAgentRequest, UserAgentResponse,
|
||||
auth::{AuthChallengeRequest, ClientMessage, client_message::Payload as ClientAuthPayload},
|
||||
user_agent_request::Payload as UserAgentRequestPayload,
|
||||
user_agent_response::Payload as UserAgentResponsePayload,
|
||||
};
|
||||
use arbiter_proto::transport::DummyTransport;
|
||||
use arbiter_server::{
|
||||
actors::{
|
||||
GlobalActors,
|
||||
bootstrap::GetToken,
|
||||
keyholder::{Bootstrap, Seal},
|
||||
user_agent::{
|
||||
HandleAuthChallengeRequest, HandleUnsealEncryptedKey, HandleUnsealRequest,
|
||||
UserAgentActor,
|
||||
},
|
||||
user_agent::{UserAgentActor, UserAgentError},
|
||||
},
|
||||
db,
|
||||
};
|
||||
use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
|
||||
use kameo::actor::{ActorRef, Spawn};
|
||||
use memsafe::MemSafe;
|
||||
use x25519_dalek::{EphemeralSecret, PublicKey};
|
||||
|
||||
type TestUserAgent = UserAgentActor<DummyTransport<UserAgentRequest, UserAgentResponse, UserAgentError>>;
|
||||
|
||||
fn auth_request(payload: ClientAuthPayload) -> UserAgentRequest {
|
||||
UserAgentRequest {
|
||||
payload: Some(UserAgentRequestPayload::AuthMessage(ClientMessage {
|
||||
payload: Some(payload),
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
fn unseal_start_request(req: UnsealStart) -> UserAgentRequest {
|
||||
UserAgentRequest {
|
||||
payload: Some(UserAgentRequestPayload::UnsealStart(req)),
|
||||
}
|
||||
}
|
||||
|
||||
fn unseal_key_request(req: UnsealEncryptedKey) -> UserAgentRequest {
|
||||
UserAgentRequest {
|
||||
payload: Some(UserAgentRequestPayload::UnsealEncryptedKey(req)),
|
||||
}
|
||||
}
|
||||
|
||||
async fn setup_authenticated_user_agent(
|
||||
seal_key: &[u8],
|
||||
) -> (arbiter_server::db::DatabasePool, ActorRef<UserAgentActor>) {
|
||||
) -> (
|
||||
arbiter_server::db::DatabasePool,
|
||||
TestUserAgent,
|
||||
) {
|
||||
let db = db::create_test_pool().await;
|
||||
|
||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||
@@ -34,38 +58,34 @@ async fn setup_authenticated_user_agent(
|
||||
.unwrap();
|
||||
actors.key_holder.ask(Seal).await.unwrap();
|
||||
|
||||
let user_agent =
|
||||
UserAgentActor::new_manual(db.clone(), actors.clone(), super::null_recipient());
|
||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
||||
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors.clone());
|
||||
|
||||
let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
|
||||
let auth_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||
user_agent_ref
|
||||
.ask(HandleAuthChallengeRequest {
|
||||
req: AuthChallengeRequest {
|
||||
user_agent
|
||||
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||
AuthChallengeRequest {
|
||||
pubkey: auth_key.verifying_key().to_bytes().to_vec(),
|
||||
bootstrap_token: Some(token),
|
||||
},
|
||||
})
|
||||
)))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
(db, user_agent_ref)
|
||||
(db, user_agent)
|
||||
}
|
||||
|
||||
async fn client_dh_encrypt(
|
||||
user_agent_ref: &ActorRef<UserAgentActor>,
|
||||
user_agent: &mut TestUserAgent,
|
||||
key_to_send: &[u8],
|
||||
) -> UnsealEncryptedKey {
|
||||
let client_secret = EphemeralSecret::random();
|
||||
let client_public = PublicKey::from(&client_secret);
|
||||
|
||||
let response = user_agent_ref
|
||||
.ask(HandleUnsealRequest {
|
||||
req: UnsealStart {
|
||||
client_pubkey: client_public.as_bytes().to_vec(),
|
||||
},
|
||||
})
|
||||
let response = user_agent
|
||||
.process_transport_inbound(unseal_start_request(UnsealStart {
|
||||
client_pubkey: client_public.as_bytes().to_vec(),
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
@@ -95,12 +115,12 @@ async fn client_dh_encrypt(
|
||||
#[test_log::test]
|
||||
pub async fn test_unseal_success() {
|
||||
let seal_key = b"test-seal-key";
|
||||
let (_db, user_agent_ref) = setup_authenticated_user_agent(seal_key).await;
|
||||
let (_db, mut user_agent) = setup_authenticated_user_agent(seal_key).await;
|
||||
|
||||
let encrypted_key = client_dh_encrypt(&user_agent_ref, seal_key).await;
|
||||
let encrypted_key = client_dh_encrypt(&mut user_agent, seal_key).await;
|
||||
|
||||
let response = user_agent_ref
|
||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
||||
let response = user_agent
|
||||
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
@@ -113,12 +133,12 @@ pub async fn test_unseal_success() {
|
||||
#[tokio::test]
|
||||
#[test_log::test]
|
||||
pub async fn test_unseal_wrong_seal_key() {
|
||||
let (_db, user_agent_ref) = setup_authenticated_user_agent(b"correct-key").await;
|
||||
let (_db, mut user_agent) = setup_authenticated_user_agent(b"correct-key").await;
|
||||
|
||||
let encrypted_key = client_dh_encrypt(&user_agent_ref, b"wrong-key").await;
|
||||
let encrypted_key = client_dh_encrypt(&mut user_agent, b"wrong-key").await;
|
||||
|
||||
let response = user_agent_ref
|
||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
||||
let response = user_agent
|
||||
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
@@ -131,28 +151,24 @@ pub async fn test_unseal_wrong_seal_key() {
|
||||
#[tokio::test]
|
||||
#[test_log::test]
|
||||
pub async fn test_unseal_corrupted_ciphertext() {
|
||||
let (_db, user_agent_ref) = setup_authenticated_user_agent(b"test-key").await;
|
||||
let (_db, mut user_agent) = setup_authenticated_user_agent(b"test-key").await;
|
||||
|
||||
let client_secret = EphemeralSecret::random();
|
||||
let client_public = PublicKey::from(&client_secret);
|
||||
|
||||
user_agent_ref
|
||||
.ask(HandleUnsealRequest {
|
||||
req: UnsealStart {
|
||||
client_pubkey: client_public.as_bytes().to_vec(),
|
||||
},
|
||||
})
|
||||
user_agent
|
||||
.process_transport_inbound(unseal_start_request(UnsealStart {
|
||||
client_pubkey: client_public.as_bytes().to_vec(),
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let response = user_agent_ref
|
||||
.ask(HandleUnsealEncryptedKey {
|
||||
req: UnsealEncryptedKey {
|
||||
nonce: vec![0u8; 24],
|
||||
ciphertext: vec![0u8; 32],
|
||||
associated_data: vec![],
|
||||
},
|
||||
})
|
||||
let response = user_agent
|
||||
.process_transport_inbound(unseal_key_request(UnsealEncryptedKey {
|
||||
nonce: vec![0u8; 24],
|
||||
ciphertext: vec![0u8; 32],
|
||||
associated_data: vec![],
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
@@ -168,27 +184,20 @@ pub async fn test_unseal_start_without_auth_fails() {
|
||||
let db = db::create_test_pool().await;
|
||||
|
||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||
let user_agent =
|
||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
||||
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||
|
||||
let client_secret = EphemeralSecret::random();
|
||||
let client_public = PublicKey::from(&client_secret);
|
||||
|
||||
let result = user_agent_ref
|
||||
.ask(HandleUnsealRequest {
|
||||
req: UnsealStart {
|
||||
client_pubkey: client_public.as_bytes().to_vec(),
|
||||
},
|
||||
})
|
||||
let result = user_agent
|
||||
.process_transport_inbound(unseal_start_request(UnsealStart {
|
||||
client_pubkey: client_public.as_bytes().to_vec(),
|
||||
}))
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Err(kameo::error::SendError::HandlerError(err)) => {
|
||||
assert!(
|
||||
matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidState),
|
||||
"Expected InvalidState, got {err:?}"
|
||||
);
|
||||
Err(err) => {
|
||||
assert_eq!(err, UserAgentError::StateTransitionFailed);
|
||||
}
|
||||
other => panic!("Expected state machine error, got {other:?}"),
|
||||
}
|
||||
@@ -198,13 +207,13 @@ pub async fn test_unseal_start_without_auth_fails() {
|
||||
#[test_log::test]
|
||||
pub async fn test_unseal_retry_after_invalid_key() {
|
||||
let seal_key = b"real-seal-key";
|
||||
let (_db, user_agent_ref) = setup_authenticated_user_agent(seal_key).await;
|
||||
let (_db, mut user_agent) = setup_authenticated_user_agent(seal_key).await;
|
||||
|
||||
{
|
||||
let encrypted_key = client_dh_encrypt(&user_agent_ref, b"wrong-key").await;
|
||||
let encrypted_key = client_dh_encrypt(&mut user_agent, b"wrong-key").await;
|
||||
|
||||
let response = user_agent_ref
|
||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
||||
let response = user_agent
|
||||
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
@@ -215,10 +224,10 @@ pub async fn test_unseal_retry_after_invalid_key() {
|
||||
}
|
||||
|
||||
{
|
||||
let encrypted_key = client_dh_encrypt(&user_agent_ref, seal_key).await;
|
||||
let encrypted_key = client_dh_encrypt(&mut user_agent, seal_key).await;
|
||||
|
||||
let response = user_agent_ref
|
||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
||||
let response = user_agent
|
||||
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
|
||||
@@ -1,66 +1,13 @@
|
||||
use arbiter_proto::{proto::UserAgentRequest, transport::TransportActor};
|
||||
use ed25519_dalek::SigningKey;
|
||||
use kameo::{
|
||||
Actor, Reply,
|
||||
actor::{ActorRef, WeakActorRef},
|
||||
prelude::Message,
|
||||
};
|
||||
use smlang::statemachine;
|
||||
use kameo::Actor;
|
||||
use tonic::transport::CertificateDer;
|
||||
use tracing::{debug, error};
|
||||
|
||||
struct Storage {
|
||||
pub identity: SigningKey,
|
||||
pub server_ca_cert: CertificateDer<'static>,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum InitError {
|
||||
StorageError,
|
||||
Other(String),
|
||||
}
|
||||
|
||||
statemachine! {
|
||||
name: UserAgentStateMachine,
|
||||
custom_error: false,
|
||||
transitions: {
|
||||
*Init + SendAuthChallenge = WaitingForAuthSolution
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
pub struct UserAgentActor<A: TransportActor<UserAgentRequest>> {
|
||||
key: SigningKey,
|
||||
server_ca_cert: CertificateDer<'static>,
|
||||
sender: ActorRef<A>,
|
||||
}
|
||||
impl<A: TransportActor<UserAgentRequest>> Actor for UserAgentActor<A> {
|
||||
type Args = Self;
|
||||
|
||||
type Error = InitError;
|
||||
|
||||
async fn on_start(args: Self::Args, actor_ref: ActorRef<Self>) -> Result<Self, Self::Error> {
|
||||
todo!()
|
||||
}
|
||||
|
||||
async fn on_link_died(
|
||||
&mut self,
|
||||
_: WeakActorRef<Self>,
|
||||
id: kameo::prelude::ActorId,
|
||||
_: kameo::prelude::ActorStopReason,
|
||||
) -> Result<std::ops::ControlFlow<kameo::prelude::ActorStopReason>, Self::Error> {
|
||||
if id == self.sender.id() {
|
||||
error!("Transport actor died, stopping UserAgentActor");
|
||||
Ok(std::ops::ControlFlow::Break(
|
||||
kameo::prelude::ActorStopReason::Normal,
|
||||
))
|
||||
} else {
|
||||
debug!(
|
||||
"Linked actor {} died, but it's not the transport actor, ignoring",
|
||||
id
|
||||
);
|
||||
Ok(std::ops::ControlFlow::Continue(()))
|
||||
}
|
||||
}
|
||||
#[derive(Actor)]
|
||||
pub struct UserAgent {
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user