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"
|
base64 = "0.22.1"
|
||||||
tracing.workspace = true
|
tracing.workspace = true
|
||||||
|
|
||||||
|
|
||||||
[build-dependencies]
|
[build-dependencies]
|
||||||
tonic-prost-build = "0.14.3"
|
tonic-prost-build = "0.14.3"
|
||||||
|
|
||||||
|
|||||||
@@ -1,371 +1,125 @@
|
|||||||
//! Transport abstraction layer for bridging gRPC bidirectional streaming with kameo actors.
|
use std::marker::PhantomData;
|
||||||
//!
|
|
||||||
//! 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 futures::{Stream, StreamExt};
|
use futures::StreamExt;
|
||||||
use kameo::{
|
use tokio::sync::mpsc;
|
||||||
Actor,
|
|
||||||
actor::{ActorRef, PreparedActor, Recipient, Spawn, WeakActorRef},
|
|
||||||
mailbox::Signal,
|
|
||||||
prelude::Message,
|
|
||||||
};
|
|
||||||
use tokio::{
|
|
||||||
select,
|
|
||||||
sync::mpsc::{self, error::SendError},
|
|
||||||
};
|
|
||||||
use tonic::{Status, Streaming};
|
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)
|
/// `Bi<T, U, E>` models a duplex channel with:
|
||||||
/// outgoing responses. This trait allows business logic to be tested without a real
|
/// - inbound items of type `T` read via [`Bi::recv`]
|
||||||
/// gRPC connection by swapping in an in-memory implementation.
|
/// - outbound success items of type `U` or domain errors of type `E` written via [`Bi::send`]
|
||||||
///
|
///
|
||||||
/// # Type Parameters
|
/// The trait intentionally exposes only the operations the protocol layer needs,
|
||||||
/// - `T`: `InboundMessage` received from the channel/socket (e.g., `UserAgentRequest`)
|
/// allowing it to work with gRPC streams and other transport implementations.
|
||||||
/// - `U`: `OutboundMessage` sent to the channel/socket (e.g., `UserAgentResponse`)
|
///
|
||||||
pub trait Bi<T, U>: Stream<Item = Result<T, Status>> + Send + Sync + 'static {
|
/// # Stream termination and errors
|
||||||
type Error;
|
///
|
||||||
|
/// [`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(
|
fn send(
|
||||||
&mut self,
|
&mut self,
|
||||||
item: Result<U, Status>,
|
item: Result<U, E>,
|
||||||
) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send;
|
) -> 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`
|
/// Outbound items are sent through a Tokio MPSC sender, while inbound items are
|
||||||
/// receives messages from the client, and `response_sender` sends responses back.
|
/// read from tonic [`Streaming`].
|
||||||
pub struct BiStream<T, U> {
|
pub struct GrpcAdapter<Inbound, Outbound, E> {
|
||||||
pub request_stream: Streaming<T>,
|
sender: mpsc::Sender<Result<Outbound, Status>>,
|
||||||
pub response_sender: mpsc::Sender<Result<U, Status>>,
|
receiver: Streaming<Inbound>,
|
||||||
|
_error: PhantomData<E>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<T, U> Stream for BiStream<T, U>
|
impl<Inbound, Outbound, E> GrpcAdapter<Inbound, Outbound, E> {
|
||||||
where
|
|
||||||
T: Send + 'static,
|
|
||||||
U: Send + 'static,
|
|
||||||
{
|
|
||||||
type Item = Result<T, Status>;
|
|
||||||
|
|
||||||
fn poll_next(
|
/// Creates a new gRPC-backed [`Bi`] adapter.
|
||||||
mut self: std::pin::Pin<&mut Self>,
|
pub fn new(sender: mpsc::Sender<Result<Outbound, Status>>, receiver: Streaming<Inbound>) -> 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 {
|
|
||||||
Self {
|
Self {
|
||||||
sender,
|
sender,
|
||||||
receiver,
|
receiver,
|
||||||
business_logic_actor,
|
_error: PhantomData,
|
||||||
_error: std::marker::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
|
where
|
||||||
Outbound: Send + 'static,
|
|
||||||
Inbound: Send + 'static,
|
|
||||||
E: Into<tonic::Status> + Send + 'static,
|
|
||||||
{
|
|
||||||
type Args = Self;
|
|
||||||
|
|
||||||
type Error = ();
|
|
||||||
|
|
||||||
async fn on_start(args: Self::Args, _: ActorRef<Self>) -> Result<Self, Self::Error> {
|
|
||||||
Ok(args)
|
|
||||||
}
|
|
||||||
|
|
||||||
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(()))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
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<Outbound, Inbound, E> TransportActor<Outbound, E> for GrpcTransportActor<Outbound, Inbound, E>
|
|
||||||
where
|
|
||||||
Outbound: Send + 'static,
|
|
||||||
Inbound: Send + 'static,
|
|
||||||
E: Into<tonic::Status> + Send + '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,
|
Inbound: Send + 'static,
|
||||||
Outbound: Send + 'static,
|
Outbound: Send + 'static,
|
||||||
DomainError: Send + 'static,
|
E: Into<Status> + Send + Sync + '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();
|
#[tracing::instrument(level = "trace", skip(self, item))]
|
||||||
let prepared_transport: PreparedActor<Transport> = Spawn::prepare();
|
async fn send(&mut self, item: Result<Outbound, E>) -> Result<(), Error> {
|
||||||
|
self.sender
|
||||||
|
.send(item.map_err(Into::into))
|
||||||
|
.await
|
||||||
|
.map_err(|_| Error::ChannelClosed)
|
||||||
|
}
|
||||||
|
|
||||||
let business_ref = prepared_business.actor_ref().clone();
|
#[tracing::instrument(level = "trace", skip(self))]
|
||||||
let transport_ref = prepared_transport.actor_ref().clone();
|
async fn recv(&mut self) -> Option<Inbound> {
|
||||||
|
self.receiver.next().await.transpose().ok().flatten()
|
||||||
transport_ref.link(&business_ref).await;
|
}
|
||||||
business_ref.link(&transport_ref).await;
|
}
|
||||||
|
|
||||||
let recipient = transport_ref.clone().recipient();
|
/// No-op [`Bi`] transport for tests and manual actor usage.
|
||||||
business_ctor(prepared_business, recipient);
|
///
|
||||||
let business_recipient = business_ref.clone().recipient();
|
/// `send` drops all items and succeeds. [`Bi::recv`] never resolves and therefore
|
||||||
transport_ctor(prepared_transport, business_recipient);
|
/// does not busy-wait or spuriously close the stream.
|
||||||
|
pub struct DummyTransport<T, U, E> {
|
||||||
|
_marker: PhantomData<(T, U, E)>,
|
||||||
(transport_ref, business_ref)
|
}
|
||||||
|
|
||||||
|
impl<T, U, E> DummyTransport<T, U, E> {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
_marker: PhantomData,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T, U, E> Default for DummyTransport<T, U, E> {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self::new()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<T, U, E> Bi<T, U, E> for DummyTransport<T, U, E>
|
||||||
|
where
|
||||||
|
T: Send + Sync + 'static,
|
||||||
|
U: Send + Sync + 'static,
|
||||||
|
E: Send + Sync + 'static,
|
||||||
|
{
|
||||||
|
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(
|
pub(crate) async fn handle_client(
|
||||||
_context: ServerContext,
|
_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 std::{ops::DerefMut, sync::Mutex};
|
||||||
|
|
||||||
use arbiter_proto::proto::{
|
use arbiter_proto::{
|
||||||
|
proto::{
|
||||||
UnsealEncryptedKey, UnsealResult, UnsealStart, UnsealStartResponse, UserAgentRequest,
|
UnsealEncryptedKey, UnsealResult, UnsealStart, UnsealStartResponse, UserAgentRequest,
|
||||||
UserAgentResponse,
|
UserAgentResponse,
|
||||||
auth::{
|
auth::{
|
||||||
self, AuthChallengeRequest, AuthOk, ClientMessage as ClientAuthMessage,
|
self, AuthChallengeRequest, AuthOk, ClientMessage as ClientAuthMessage,
|
||||||
ServerMessage as AuthServerMessage,
|
ServerMessage as AuthServerMessage, client_message::Payload as ClientAuthPayload,
|
||||||
client_message::Payload as ClientAuthPayload,
|
|
||||||
server_message::Payload as ServerAuthPayload,
|
server_message::Payload as ServerAuthPayload,
|
||||||
},
|
},
|
||||||
user_agent_request::Payload as UserAgentRequestPayload,
|
user_agent_request::Payload as UserAgentRequestPayload,
|
||||||
user_agent_response::Payload as UserAgentResponsePayload,
|
user_agent_response::Payload as UserAgentResponsePayload,
|
||||||
|
},
|
||||||
|
transport::{Bi, DummyTransport},
|
||||||
};
|
};
|
||||||
use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
|
use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
|
||||||
use diesel::{ExpressionMethods as _, OptionalExtension as _, QueryDsl, dsl::update};
|
use diesel::{ExpressionMethods as _, OptionalExtension as _, QueryDsl, dsl::update};
|
||||||
use diesel_async::RunQueryDsl;
|
use diesel_async::RunQueryDsl;
|
||||||
use ed25519_dalek::VerifyingKey;
|
use ed25519_dalek::VerifyingKey;
|
||||||
use kameo::{Actor, actor::Recipient, error::SendError, messages, prelude::Message};
|
use kameo::{Actor, error::SendError};
|
||||||
use memsafe::MemSafe;
|
use memsafe::MemSafe;
|
||||||
|
use tokio::select;
|
||||||
|
use tonic::Status;
|
||||||
use tracing::{error, info};
|
use tracing::{error, info};
|
||||||
use x25519_dalek::{EphemeralSecret, PublicKey};
|
use x25519_dalek::{EphemeralSecret, PublicKey};
|
||||||
|
|
||||||
@@ -35,24 +39,103 @@ use crate::{
|
|||||||
db::{self, schema},
|
db::{self, schema},
|
||||||
};
|
};
|
||||||
|
|
||||||
mod error;
|
|
||||||
mod state;
|
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)]
|
impl From<UserAgentError> for Status {
|
||||||
pub struct UserAgentActor {
|
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,
|
db: db::DatabasePool,
|
||||||
actors: GlobalActors,
|
actors: GlobalActors,
|
||||||
state: UserAgentStateMachine<DummyContext>,
|
state: UserAgentStateMachine<DummyContext>,
|
||||||
transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
|
transport: Transport,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl UserAgentActor {
|
impl<Transport> UserAgentActor<Transport>
|
||||||
pub(crate) fn new(
|
where
|
||||||
context: ServerContext,
|
Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
|
||||||
transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
|
{
|
||||||
) -> Self {
|
pub(crate) fn new(context: ServerContext, transport: Transport) -> Self {
|
||||||
Self {
|
Self {
|
||||||
db: context.db.clone(),
|
db: context.db.clone(),
|
||||||
actors: context.actors.clone(),
|
actors: context.actors.clone(),
|
||||||
@@ -61,23 +144,18 @@ impl UserAgentActor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn new_manual(
|
fn transition(&mut self, event: UserAgentEvents) -> Result<(), UserAgentError> {
|
||||||
db: db::DatabasePool,
|
self.state.process_event(event).map_err(|e| {
|
||||||
actors: GlobalActors,
|
error!(?e, "State transition failed");
|
||||||
transport: Recipient<Result<UserAgentResponse, UserAgentError>>,
|
UserAgentError::StateTransitionFailed
|
||||||
) -> Self {
|
})?;
|
||||||
Self {
|
Ok(())
|
||||||
db,
|
|
||||||
actors,
|
|
||||||
state: UserAgentStateMachine::new(DummyContext),
|
|
||||||
transport,
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
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(|| {
|
let msg = req.payload.ok_or_else(|| {
|
||||||
error!(actor = "useragent", "Received message with no payload");
|
error!(actor = "useragent", "Received message with no payload");
|
||||||
UserAgentError::MissingPayload
|
UserAgentError::MissingRequestPayload
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
match msg {
|
match msg {
|
||||||
@@ -93,23 +171,15 @@ impl UserAgentActor {
|
|||||||
UserAgentRequestPayload::UnsealEncryptedKey(unseal_encrypted_key) => {
|
UserAgentRequestPayload::UnsealEncryptedKey(unseal_encrypted_key) => {
|
||||||
self.handle_unseal_encrypted_key(unseal_encrypted_key).await
|
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(
|
async fn auth_with_bootstrap_token(
|
||||||
&mut self,
|
&mut self,
|
||||||
pubkey: ed25519_dalek::VerifyingKey,
|
pubkey: ed25519_dalek::VerifyingKey,
|
||||||
token: String,
|
token: String,
|
||||||
) -> Output {
|
) -> Result<UserAgentResponse, UserAgentError> {
|
||||||
let token_ok: bool = self
|
let token_ok: bool = self
|
||||||
.actors
|
.actors
|
||||||
.bootstrapper
|
.bootstrapper
|
||||||
@@ -117,7 +187,7 @@ impl UserAgentActor {
|
|||||||
.await
|
.await
|
||||||
.map_err(|e| {
|
.map_err(|e| {
|
||||||
error!(?pubkey, "Failed to consume bootstrap token: {e}");
|
error!(?pubkey, "Failed to consume bootstrap token: {e}");
|
||||||
UserAgentError::ActorUnavailable
|
UserAgentError::BootstrapperActorUnreachable
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
if !token_ok {
|
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)
|
diesel::insert_into(schema::useragent_client::table)
|
||||||
.values((
|
.values((
|
||||||
@@ -134,7 +207,11 @@ impl UserAgentActor {
|
|||||||
schema::useragent_client::nonce.eq(1),
|
schema::useragent_client::nonce.eq(1),
|
||||||
))
|
))
|
||||||
.execute(&mut conn)
|
.execute(&mut conn)
|
||||||
.await?;
|
.await
|
||||||
|
.map_err(|e| {
|
||||||
|
error!(error = ?e, "Database error");
|
||||||
|
UserAgentError::DatabaseOperationFailed
|
||||||
|
})?;
|
||||||
}
|
}
|
||||||
|
|
||||||
self.transition(UserAgentEvents::ReceivedBootstrapToken)?;
|
self.transition(UserAgentEvents::ReceivedBootstrapToken)?;
|
||||||
@@ -144,7 +221,10 @@ impl UserAgentActor {
|
|||||||
|
|
||||||
async fn auth_with_challenge(&mut self, pubkey: VerifyingKey, pubkey_bytes: Vec<u8>) -> Output {
|
async fn auth_with_challenge(&mut self, pubkey: VerifyingKey, pubkey_bytes: Vec<u8>) -> Output {
|
||||||
let nonce: Option<i32> = {
|
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
|
db_conn
|
||||||
.exclusive_transaction(|conn| {
|
.exclusive_transaction(|conn| {
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
@@ -168,12 +248,16 @@ impl UserAgentActor {
|
|||||||
})
|
})
|
||||||
})
|
})
|
||||||
.await
|
.await
|
||||||
.optional()?
|
.optional()
|
||||||
|
.map_err(|e| {
|
||||||
|
error!(error = ?e, "Database error");
|
||||||
|
UserAgentError::DatabaseOperationFailed
|
||||||
|
})?
|
||||||
};
|
};
|
||||||
|
|
||||||
let Some(nonce) = nonce else {
|
let Some(nonce) = nonce else {
|
||||||
error!(?pubkey, "Public key not found in database");
|
error!(?pubkey, "Public key not found in database");
|
||||||
return Err(UserAgentError::PubkeyNotRegistered);
|
return Err(UserAgentError::PublicKeyNotRegistered);
|
||||||
};
|
};
|
||||||
|
|
||||||
let challenge = auth::AuthChallenge {
|
let challenge = auth::AuthChallenge {
|
||||||
@@ -202,7 +286,7 @@ impl UserAgentActor {
|
|||||||
let UserAgentStates::WaitingForChallengeSolution(challenge_context) = self.state.state()
|
let UserAgentStates::WaitingForChallengeSolution(challenge_context) = self.state.state()
|
||||||
else {
|
else {
|
||||||
error!("Received challenge solution in invalid state");
|
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);
|
let formatted_challenge = arbiter_proto::format_challenge(&challenge_context.challenge);
|
||||||
|
|
||||||
@@ -236,17 +320,18 @@ fn unseal_response(payload: UserAgentResponsePayload) -> UserAgentResponse {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[messages]
|
impl<Transport> UserAgentActor<Transport>
|
||||||
impl UserAgentActor {
|
where
|
||||||
#[message]
|
Transport: Bi<UserAgentRequest, UserAgentResponse, UserAgentError>,
|
||||||
pub async fn handle_unseal_request(&mut self, req: UnsealStart) -> Output {
|
{
|
||||||
|
async fn handle_unseal_request(&mut self, req: UnsealStart) -> Output {
|
||||||
let secret = EphemeralSecret::random();
|
let secret = EphemeralSecret::random();
|
||||||
let public_key = PublicKey::from(&secret);
|
let public_key = PublicKey::from(&secret);
|
||||||
|
|
||||||
let client_pubkey_bytes: [u8; 32] = req
|
let client_pubkey_bytes: [u8; 32] = req
|
||||||
.client_pubkey
|
.client_pubkey
|
||||||
.try_into()
|
.try_into()
|
||||||
.map_err(|_| UserAgentError::InvalidPubkey)?;
|
.map_err(|_| UserAgentError::InvalidClientPubkeyLength)?;
|
||||||
|
|
||||||
let client_public_key = PublicKey::from(client_pubkey_bytes);
|
let client_public_key = PublicKey::from(client_pubkey_bytes);
|
||||||
|
|
||||||
@@ -262,11 +347,10 @@ impl UserAgentActor {
|
|||||||
))
|
))
|
||||||
}
|
}
|
||||||
|
|
||||||
#[message]
|
async fn handle_unseal_encrypted_key(&mut self, req: UnsealEncryptedKey) -> Output {
|
||||||
pub async fn handle_unseal_encrypted_key(&mut self, req: UnsealEncryptedKey) -> Output {
|
|
||||||
let UserAgentStates::WaitingForUnsealKey(unseal_context) = self.state.state() else {
|
let UserAgentStates::WaitingForUnsealKey(unseal_context) = self.state.state() else {
|
||||||
error!("Received unseal encrypted key in invalid state");
|
error!("Received unseal encrypted key in invalid state");
|
||||||
return Err(UserAgentError::InvalidState);
|
return Err(UserAgentError::InvalidStateForUnsealEncryptedKey);
|
||||||
};
|
};
|
||||||
let ephemeral_secret = {
|
let ephemeral_secret = {
|
||||||
let mut secret_lock = unseal_context.secret.lock().unwrap();
|
let mut secret_lock = unseal_context.secret.lock().unwrap();
|
||||||
@@ -330,7 +414,7 @@ impl UserAgentActor {
|
|||||||
Err(err) => {
|
Err(err) => {
|
||||||
error!(?err, "Failed to send unseal request to keyholder");
|
error!(?err, "Failed to send unseal request to keyholder");
|
||||||
self.transition(UserAgentEvents::ReceivedInvalidKey)?;
|
self.transition(UserAgentEvents::ReceivedInvalidKey)?;
|
||||||
Err(UserAgentError::ActorUnavailable)
|
Err(UserAgentError::KeyHolderActorUnreachable)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -344,15 +428,14 @@ impl UserAgentActor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[message]
|
async fn handle_auth_challenge_request(&mut self, req: AuthChallengeRequest) -> Output {
|
||||||
pub async fn handle_auth_challenge_request(&mut self, req: AuthChallengeRequest) -> Output {
|
|
||||||
let pubkey = req
|
let pubkey = req
|
||||||
.pubkey
|
.pubkey
|
||||||
.as_array()
|
.as_array()
|
||||||
.ok_or(UserAgentError::InvalidPubkey)?;
|
.ok_or(UserAgentError::InvalidAuthPubkeyLength)?;
|
||||||
let pubkey = VerifyingKey::from_bytes(pubkey).map_err(|_err| {
|
let pubkey = VerifyingKey::from_bytes(pubkey).map_err(|_err| {
|
||||||
error!(?pubkey, "Failed to convert to VerifyingKey");
|
error!(?pubkey, "Failed to convert to VerifyingKey");
|
||||||
UserAgentError::InvalidPubkey
|
UserAgentError::InvalidAuthPubkeyEncoding
|
||||||
})?;
|
})?;
|
||||||
|
|
||||||
self.transition(UserAgentEvents::AuthRequest)?;
|
self.transition(UserAgentEvents::AuthRequest)?;
|
||||||
@@ -363,8 +446,7 @@ impl UserAgentActor {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[message]
|
async fn handle_auth_challenge_solution(
|
||||||
pub async fn handle_auth_challenge_solution(
|
|
||||||
&mut self,
|
&mut self,
|
||||||
solution: auth::AuthChallengeSolution,
|
solution: auth::AuthChallengeSolution,
|
||||||
) -> Output {
|
) -> 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,
|
&mut self,
|
||||||
msg: UserAgentRequest,
|
_actor_ref: kameo::prelude::WeakActorRef<Self>,
|
||||||
_ctx: &mut kameo::prelude::Context<Self, Self::Reply>,
|
mailbox_rx: &mut kameo::prelude::MailboxReceiver<Self>,
|
||||||
) -> Self::Reply {
|
) -> Option<kameo::mailbox::Signal<Self>> {
|
||||||
let result = self.process_request(msg).await;
|
loop {
|
||||||
if let Err(e) = self.transport.tell(result).await {
|
select! {
|
||||||
error!(actor = "useragent", "Failed to send response to transport: {}", e);
|
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)]
|
#![forbid(unsafe_code)]
|
||||||
use arbiter_proto::{
|
use arbiter_proto::{
|
||||||
proto::{ClientRequest, ClientResponse, UserAgentRequest, UserAgentResponse},
|
proto::{ClientRequest, ClientResponse, UserAgentRequest, UserAgentResponse},
|
||||||
transport::{BiStream, GrpcTransportActor, wire},
|
transport::GrpcAdapter,
|
||||||
};
|
};
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use kameo::actor::PreparedActor;
|
use kameo::actor::Spawn;
|
||||||
use tokio_stream::wrappers::ReceiverStream;
|
use tokio_stream::wrappers::ReceiverStream;
|
||||||
|
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
use tonic::{Request, Response, Status};
|
use tonic::{Request, Response, Status};
|
||||||
|
use tracing::info;
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
actors::{
|
actors::user_agent::UserAgentActor,
|
||||||
client::handle_client,
|
|
||||||
user_agent::UserAgentActor,
|
|
||||||
},
|
|
||||||
context::ServerContext,
|
context::ServerContext,
|
||||||
};
|
};
|
||||||
|
|
||||||
pub mod actors;
|
pub mod actors;
|
||||||
pub mod context;
|
pub mod context;
|
||||||
pub mod db;
|
pub mod db;
|
||||||
|
mod errors;
|
||||||
|
|
||||||
const DEFAULT_CHANNEL_SIZE: usize = 1000;
|
const DEFAULT_CHANNEL_SIZE: usize = 1000;
|
||||||
|
|
||||||
@@ -41,42 +40,24 @@ impl arbiter_proto::proto::arbiter_service_server::ArbiterService for Server {
|
|||||||
|
|
||||||
async fn client(
|
async fn client(
|
||||||
&self,
|
&self,
|
||||||
request: Request<tonic::Streaming<ClientRequest>>,
|
_request: Request<tonic::Streaming<ClientRequest>>,
|
||||||
) -> Result<Response<Self::ClientStream>, Status> {
|
) -> Result<Response<Self::ClientStream>, Status> {
|
||||||
let req_stream = request.into_inner();
|
todo!()
|
||||||
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)))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tracing::instrument(level = "debug", skip(self))]
|
||||||
async fn user_agent(
|
async fn user_agent(
|
||||||
&self,
|
&self,
|
||||||
request: Request<tonic::Streaming<UserAgentRequest>>,
|
request: Request<tonic::Streaming<UserAgentRequest>>,
|
||||||
) -> Result<Response<Self::UserAgentStream>, Status> {
|
) -> Result<Response<Self::UserAgentStream>, Status> {
|
||||||
let req_stream = request.into_inner();
|
let req_stream = request.into_inner();
|
||||||
let (tx, rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
|
let (tx, rx) = mpsc::channel(DEFAULT_CHANNEL_SIZE);
|
||||||
let context = self.context.clone();
|
|
||||||
|
|
||||||
wire(
|
let adapter = GrpcAdapter::new(tx, req_stream);
|
||||||
|prepared: PreparedActor<UserAgentActor>, recipient| {
|
|
||||||
prepared.spawn(UserAgentActor::new(context, recipient));
|
UserAgentActor::spawn(UserAgentActor::new(self.context.clone(), adapter));
|
||||||
},
|
|
||||||
|prepared: PreparedActor<GrpcTransportActor<_, _, _>>, business_recipient| {
|
info!(event = "connection established", "grpc.user_agent");
|
||||||
prepared.spawn(GrpcTransportActor::new(
|
|
||||||
tx,
|
|
||||||
req_stream,
|
|
||||||
business_recipient,
|
|
||||||
));
|
|
||||||
},
|
|
||||||
)
|
|
||||||
.await;
|
|
||||||
|
|
||||||
Ok(Response::new(ReceiverStream::new(rx)))
|
Ok(Response::new(ReceiverStream::new(rx)))
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,30 +1,5 @@
|
|||||||
mod common;
|
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"]
|
#[path = "user_agent/auth.rs"]
|
||||||
mod auth;
|
mod auth;
|
||||||
#[path = "user_agent/unseal.rs"]
|
#[path = "user_agent/unseal.rs"]
|
||||||
|
|||||||
@@ -1,20 +1,29 @@
|
|||||||
use arbiter_proto::proto::{
|
use arbiter_proto::proto::{
|
||||||
UserAgentResponse,
|
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,
|
user_agent_response::Payload as UserAgentResponsePayload,
|
||||||
};
|
};
|
||||||
use arbiter_server::{
|
use arbiter_server::{
|
||||||
actors::{
|
actors::{
|
||||||
GlobalActors,
|
GlobalActors,
|
||||||
bootstrap::GetToken,
|
bootstrap::GetToken,
|
||||||
user_agent::{HandleAuthChallengeRequest, HandleAuthChallengeSolution, UserAgentActor},
|
user_agent::{UserAgentActor, UserAgentError},
|
||||||
},
|
},
|
||||||
db::{self, schema},
|
db::{self, schema},
|
||||||
};
|
};
|
||||||
use diesel::{ExpressionMethods as _, QueryDsl, insert_into};
|
use diesel::{ExpressionMethods as _, QueryDsl, insert_into};
|
||||||
use diesel_async::RunQueryDsl;
|
use diesel_async::RunQueryDsl;
|
||||||
use ed25519_dalek::Signer as _;
|
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]
|
#[tokio::test]
|
||||||
#[test_log::test]
|
#[test_log::test]
|
||||||
@@ -23,22 +32,20 @@ pub async fn test_bootstrap_token_auth() {
|
|||||||
|
|
||||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||||
let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
|
let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
|
||||||
let user_agent =
|
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
|
||||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
|
||||||
|
|
||||||
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||||
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
||||||
|
|
||||||
let result = user_agent_ref
|
let result = user_agent
|
||||||
.ask(HandleAuthChallengeRequest {
|
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||||
req: AuthChallengeRequest {
|
AuthChallengeRequest {
|
||||||
pubkey: pubkey_bytes,
|
pubkey: pubkey_bytes,
|
||||||
bootstrap_token: Some(token),
|
bootstrap_token: Some(token),
|
||||||
},
|
},
|
||||||
})
|
)))
|
||||||
.await
|
.await
|
||||||
.expect("Shouldn't fail to send message");
|
.expect("Shouldn't fail to process message");
|
||||||
|
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
result,
|
result,
|
||||||
@@ -68,31 +75,25 @@ pub async fn test_bootstrap_invalid_token_auth() {
|
|||||||
let db = db::create_test_pool().await;
|
let db = db::create_test_pool().await;
|
||||||
|
|
||||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||||
let user_agent =
|
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
|
||||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
|
||||||
|
|
||||||
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||||
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
||||||
|
|
||||||
let result = user_agent_ref
|
let result = user_agent
|
||||||
.ask(HandleAuthChallengeRequest {
|
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||||
req: AuthChallengeRequest {
|
AuthChallengeRequest {
|
||||||
pubkey: pubkey_bytes,
|
pubkey: pubkey_bytes,
|
||||||
bootstrap_token: Some("invalid_token".to_string()),
|
bootstrap_token: Some("invalid_token".to_string()),
|
||||||
},
|
},
|
||||||
})
|
)))
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
match result {
|
match result {
|
||||||
Err(kameo::error::SendError::HandlerError(err)) => {
|
Err(err) => {
|
||||||
assert!(
|
assert_eq!(err, UserAgentError::InvalidBootstrapToken);
|
||||||
matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidBootstrapToken),
|
let status: tonic::Status = err.into();
|
||||||
"Expected InvalidBootstrapToken, got {err:?}"
|
assert_eq!(status.code(), tonic::Code::InvalidArgument);
|
||||||
);
|
|
||||||
}
|
|
||||||
Err(other) => {
|
|
||||||
panic!("Expected SendError::HandlerError, got {other:?}");
|
|
||||||
}
|
}
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
panic!("Expected error due to invalid bootstrap token, but got success");
|
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 db = db::create_test_pool().await;
|
||||||
|
|
||||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||||
let user_agent =
|
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
|
||||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
|
||||||
|
|
||||||
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
let new_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||||
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
let pubkey_bytes = new_key.verifying_key().to_bytes().to_vec();
|
||||||
@@ -122,15 +121,15 @@ pub async fn test_challenge_auth() {
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
let result = user_agent_ref
|
let result = user_agent
|
||||||
.ask(HandleAuthChallengeRequest {
|
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||||
req: AuthChallengeRequest {
|
AuthChallengeRequest {
|
||||||
pubkey: pubkey_bytes,
|
pubkey: pubkey_bytes,
|
||||||
bootstrap_token: None,
|
bootstrap_token: None,
|
||||||
},
|
},
|
||||||
})
|
)))
|
||||||
.await
|
.await
|
||||||
.expect("Shouldn't fail to send message");
|
.expect("Shouldn't fail to process message");
|
||||||
|
|
||||||
let UserAgentResponse {
|
let UserAgentResponse {
|
||||||
payload:
|
payload:
|
||||||
@@ -147,14 +146,14 @@ pub async fn test_challenge_auth() {
|
|||||||
let signature = new_key.sign(&formatted_challenge);
|
let signature = new_key.sign(&formatted_challenge);
|
||||||
let serialized_signature = signature.to_bytes().to_vec();
|
let serialized_signature = signature.to_bytes().to_vec();
|
||||||
|
|
||||||
let result = user_agent_ref
|
let result = user_agent
|
||||||
.ask(HandleAuthChallengeSolution {
|
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeSolution(
|
||||||
solution: auth::AuthChallengeSolution {
|
auth::AuthChallengeSolution {
|
||||||
signature: serialized_signature,
|
signature: serialized_signature,
|
||||||
},
|
},
|
||||||
})
|
)))
|
||||||
.await
|
.await
|
||||||
.expect("Shouldn't fail to send message");
|
.expect("Shouldn't fail to process message");
|
||||||
|
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
result,
|
result,
|
||||||
|
|||||||
@@ -1,27 +1,51 @@
|
|||||||
use arbiter_proto::proto::{
|
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,
|
user_agent_response::Payload as UserAgentResponsePayload,
|
||||||
};
|
};
|
||||||
|
use arbiter_proto::transport::DummyTransport;
|
||||||
use arbiter_server::{
|
use arbiter_server::{
|
||||||
actors::{
|
actors::{
|
||||||
GlobalActors,
|
GlobalActors,
|
||||||
bootstrap::GetToken,
|
bootstrap::GetToken,
|
||||||
keyholder::{Bootstrap, Seal},
|
keyholder::{Bootstrap, Seal},
|
||||||
user_agent::{
|
user_agent::{UserAgentActor, UserAgentError},
|
||||||
HandleAuthChallengeRequest, HandleUnsealEncryptedKey, HandleUnsealRequest,
|
|
||||||
UserAgentActor,
|
|
||||||
},
|
|
||||||
},
|
},
|
||||||
db,
|
db,
|
||||||
};
|
};
|
||||||
use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
|
use chacha20poly1305::{AeadInPlace, XChaCha20Poly1305, XNonce, aead::KeyInit};
|
||||||
use kameo::actor::{ActorRef, Spawn};
|
|
||||||
use memsafe::MemSafe;
|
use memsafe::MemSafe;
|
||||||
use x25519_dalek::{EphemeralSecret, PublicKey};
|
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(
|
async fn setup_authenticated_user_agent(
|
||||||
seal_key: &[u8],
|
seal_key: &[u8],
|
||||||
) -> (arbiter_server::db::DatabasePool, ActorRef<UserAgentActor>) {
|
) -> (
|
||||||
|
arbiter_server::db::DatabasePool,
|
||||||
|
TestUserAgent,
|
||||||
|
) {
|
||||||
let db = db::create_test_pool().await;
|
let db = db::create_test_pool().await;
|
||||||
|
|
||||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||||
@@ -34,38 +58,34 @@ async fn setup_authenticated_user_agent(
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
actors.key_holder.ask(Seal).await.unwrap();
|
actors.key_holder.ask(Seal).await.unwrap();
|
||||||
|
|
||||||
let user_agent =
|
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors.clone());
|
||||||
UserAgentActor::new_manual(db.clone(), actors.clone(), super::null_recipient());
|
|
||||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
|
||||||
|
|
||||||
let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
|
let token = actors.bootstrapper.ask(GetToken).await.unwrap().unwrap();
|
||||||
let auth_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
let auth_key = ed25519_dalek::SigningKey::generate(&mut rand::rng());
|
||||||
user_agent_ref
|
user_agent
|
||||||
.ask(HandleAuthChallengeRequest {
|
.process_transport_inbound(auth_request(ClientAuthPayload::AuthChallengeRequest(
|
||||||
req: AuthChallengeRequest {
|
AuthChallengeRequest {
|
||||||
pubkey: auth_key.verifying_key().to_bytes().to_vec(),
|
pubkey: auth_key.verifying_key().to_bytes().to_vec(),
|
||||||
bootstrap_token: Some(token),
|
bootstrap_token: Some(token),
|
||||||
},
|
},
|
||||||
})
|
)))
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
(db, user_agent_ref)
|
(db, user_agent)
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn client_dh_encrypt(
|
async fn client_dh_encrypt(
|
||||||
user_agent_ref: &ActorRef<UserAgentActor>,
|
user_agent: &mut TestUserAgent,
|
||||||
key_to_send: &[u8],
|
key_to_send: &[u8],
|
||||||
) -> UnsealEncryptedKey {
|
) -> UnsealEncryptedKey {
|
||||||
let client_secret = EphemeralSecret::random();
|
let client_secret = EphemeralSecret::random();
|
||||||
let client_public = PublicKey::from(&client_secret);
|
let client_public = PublicKey::from(&client_secret);
|
||||||
|
|
||||||
let response = user_agent_ref
|
let response = user_agent
|
||||||
.ask(HandleUnsealRequest {
|
.process_transport_inbound(unseal_start_request(UnsealStart {
|
||||||
req: UnsealStart {
|
|
||||||
client_pubkey: client_public.as_bytes().to_vec(),
|
client_pubkey: client_public.as_bytes().to_vec(),
|
||||||
},
|
}))
|
||||||
})
|
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
@@ -95,12 +115,12 @@ async fn client_dh_encrypt(
|
|||||||
#[test_log::test]
|
#[test_log::test]
|
||||||
pub async fn test_unseal_success() {
|
pub async fn test_unseal_success() {
|
||||||
let seal_key = b"test-seal-key";
|
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
|
let response = user_agent
|
||||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
@@ -113,12 +133,12 @@ pub async fn test_unseal_success() {
|
|||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
#[test_log::test]
|
#[test_log::test]
|
||||||
pub async fn test_unseal_wrong_seal_key() {
|
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
|
let response = user_agent
|
||||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
@@ -131,28 +151,24 @@ pub async fn test_unseal_wrong_seal_key() {
|
|||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
#[test_log::test]
|
#[test_log::test]
|
||||||
pub async fn test_unseal_corrupted_ciphertext() {
|
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_secret = EphemeralSecret::random();
|
||||||
let client_public = PublicKey::from(&client_secret);
|
let client_public = PublicKey::from(&client_secret);
|
||||||
|
|
||||||
user_agent_ref
|
user_agent
|
||||||
.ask(HandleUnsealRequest {
|
.process_transport_inbound(unseal_start_request(UnsealStart {
|
||||||
req: UnsealStart {
|
|
||||||
client_pubkey: client_public.as_bytes().to_vec(),
|
client_pubkey: client_public.as_bytes().to_vec(),
|
||||||
},
|
}))
|
||||||
})
|
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
let response = user_agent_ref
|
let response = user_agent
|
||||||
.ask(HandleUnsealEncryptedKey {
|
.process_transport_inbound(unseal_key_request(UnsealEncryptedKey {
|
||||||
req: UnsealEncryptedKey {
|
|
||||||
nonce: vec![0u8; 24],
|
nonce: vec![0u8; 24],
|
||||||
ciphertext: vec![0u8; 32],
|
ciphertext: vec![0u8; 32],
|
||||||
associated_data: vec![],
|
associated_data: vec![],
|
||||||
},
|
}))
|
||||||
})
|
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
@@ -168,27 +184,20 @@ pub async fn test_unseal_start_without_auth_fails() {
|
|||||||
let db = db::create_test_pool().await;
|
let db = db::create_test_pool().await;
|
||||||
|
|
||||||
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
let actors = GlobalActors::spawn(db.clone()).await.unwrap();
|
||||||
let user_agent =
|
let mut user_agent = UserAgentActor::new_manual(db.clone(), actors);
|
||||||
UserAgentActor::new_manual(db.clone(), actors, super::null_recipient());
|
|
||||||
let user_agent_ref = UserAgentActor::spawn(user_agent);
|
|
||||||
|
|
||||||
let client_secret = EphemeralSecret::random();
|
let client_secret = EphemeralSecret::random();
|
||||||
let client_public = PublicKey::from(&client_secret);
|
let client_public = PublicKey::from(&client_secret);
|
||||||
|
|
||||||
let result = user_agent_ref
|
let result = user_agent
|
||||||
.ask(HandleUnsealRequest {
|
.process_transport_inbound(unseal_start_request(UnsealStart {
|
||||||
req: UnsealStart {
|
|
||||||
client_pubkey: client_public.as_bytes().to_vec(),
|
client_pubkey: client_public.as_bytes().to_vec(),
|
||||||
},
|
}))
|
||||||
})
|
|
||||||
.await;
|
.await;
|
||||||
|
|
||||||
match result {
|
match result {
|
||||||
Err(kameo::error::SendError::HandlerError(err)) => {
|
Err(err) => {
|
||||||
assert!(
|
assert_eq!(err, UserAgentError::StateTransitionFailed);
|
||||||
matches!(err, arbiter_server::actors::user_agent::UserAgentError::InvalidState),
|
|
||||||
"Expected InvalidState, got {err:?}"
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
other => panic!("Expected state machine error, got {other:?}"),
|
other => panic!("Expected state machine error, got {other:?}"),
|
||||||
}
|
}
|
||||||
@@ -198,13 +207,13 @@ pub async fn test_unseal_start_without_auth_fails() {
|
|||||||
#[test_log::test]
|
#[test_log::test]
|
||||||
pub async fn test_unseal_retry_after_invalid_key() {
|
pub async fn test_unseal_retry_after_invalid_key() {
|
||||||
let seal_key = b"real-seal-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
|
let response = user_agent
|
||||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.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
|
let response = user_agent
|
||||||
.ask(HandleUnsealEncryptedKey { req: encrypted_key })
|
.process_transport_inbound(unseal_key_request(encrypted_key))
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
|
|
||||||
|
|||||||
@@ -1,66 +1,13 @@
|
|||||||
use arbiter_proto::{proto::UserAgentRequest, transport::TransportActor};
|
|
||||||
use ed25519_dalek::SigningKey;
|
use ed25519_dalek::SigningKey;
|
||||||
use kameo::{
|
use kameo::Actor;
|
||||||
Actor, Reply,
|
|
||||||
actor::{ActorRef, WeakActorRef},
|
|
||||||
prelude::Message,
|
|
||||||
};
|
|
||||||
use smlang::statemachine;
|
|
||||||
use tonic::transport::CertificateDer;
|
use tonic::transport::CertificateDer;
|
||||||
use tracing::{debug, error};
|
|
||||||
|
|
||||||
struct Storage {
|
struct Storage {
|
||||||
pub identity: SigningKey,
|
pub identity: SigningKey,
|
||||||
pub server_ca_cert: CertificateDer<'static>,
|
pub server_ca_cert: CertificateDer<'static>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Actor)]
|
||||||
pub enum InitError {
|
pub struct UserAgent {
|
||||||
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(()))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
Reference in New Issue
Block a user