feat(server): v2 RPC handler dispatch for all 33 methods
Add v2_handlers module with ServerState, build_registry(), require_auth() helper, and 33 protobuf handlers across 10 files: - auth: 4 OPAQUE handlers (register start/finish, login start/finish) - delivery: 6 handlers (enqueue, fetch, fetch_wait, peek, ack, batch) - keys: 5 handlers (upload/fetch key package, upload/fetch hybrid key/keys) - channel: create_channel - user: resolve_user, resolve_identity - blob: upload_blob, download_blob - device: register, list, revoke - p2p: publish_endpoint, resolve_endpoint, health - federation: 6 stubs (Unimplemented) - account: delete_account All handlers decode protobuf, call domain services, encode response. Auth handlers use full OPAQUE flow with session creation. Delivery handlers include rate limiting and long-poll (fetch_wait).
This commit is contained in:
287
crates/quicproquo-server/src/v2_handlers/mod.rs
Normal file
287
crates/quicproquo-server/src/v2_handlers/mod.rs
Normal file
@@ -0,0 +1,287 @@
|
||||
//! v2 RPC handler dispatch — protobuf in, domain logic, protobuf out.
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
|
||||
use dashmap::DashMap;
|
||||
use opaque_ke::ServerSetup;
|
||||
use quicproquo_core::opaque_auth::OpaqueSuite;
|
||||
use quicproquo_proto::method_ids;
|
||||
use quicproquo_rpc::error::RpcStatus;
|
||||
use quicproquo_rpc::method::{HandlerResult, MethodRegistry, RequestContext};
|
||||
use tokio::sync::Notify;
|
||||
|
||||
use crate::auth::{AuthConfig, PendingLogin, RateEntry, SessionInfo};
|
||||
use crate::hooks::ServerHooks;
|
||||
use crate::storage::Store;
|
||||
|
||||
pub mod account;
|
||||
pub mod auth;
|
||||
pub mod blob;
|
||||
pub mod channel;
|
||||
pub mod delivery;
|
||||
pub mod device;
|
||||
pub mod federation;
|
||||
pub mod keys;
|
||||
pub mod p2p;
|
||||
pub mod user;
|
||||
|
||||
/// Shared server state accessible by all v2 RPC handlers.
|
||||
pub struct ServerState {
|
||||
pub store: Arc<dyn Store>,
|
||||
pub waiters: Arc<DashMap<Vec<u8>, Arc<Notify>>>,
|
||||
pub auth_cfg: Arc<AuthConfig>,
|
||||
pub opaque_setup: Arc<ServerSetup<OpaqueSuite>>,
|
||||
pub pending_logins: Arc<DashMap<String, PendingLogin>>,
|
||||
pub sessions: Arc<DashMap<Vec<u8>, SessionInfo>>,
|
||||
pub rate_limits: Arc<DashMap<Vec<u8>, RateEntry>>,
|
||||
pub sealed_sender: bool,
|
||||
pub hooks: Arc<dyn ServerHooks>,
|
||||
pub signing_key: Arc<quicproquo_core::IdentityKeypair>,
|
||||
pub kt_log: Arc<std::sync::Mutex<quicproquo_kt::MerkleLog>>,
|
||||
pub data_dir: PathBuf,
|
||||
pub redact_logs: bool,
|
||||
}
|
||||
|
||||
/// Validate the session token from the request context and return the
|
||||
/// authenticated caller's identity key. Returns an Unauthorized HandlerResult
|
||||
/// on failure.
|
||||
pub fn require_auth(state: &ServerState, ctx: &RequestContext) -> Result<Vec<u8>, HandlerResult> {
|
||||
let token = ctx
|
||||
.session_token
|
||||
.as_deref()
|
||||
.or(ctx.identity_key.as_deref())
|
||||
.unwrap_or(&[]);
|
||||
|
||||
if token.is_empty() {
|
||||
return Err(HandlerResult::err(
|
||||
RpcStatus::Unauthorized,
|
||||
"missing session token",
|
||||
));
|
||||
}
|
||||
|
||||
// Check session store.
|
||||
if let Some(session) = state.sessions.get(token) {
|
||||
let now = crate::auth::current_timestamp();
|
||||
if session.expires_at > now && !session.identity_key.is_empty() {
|
||||
return Ok(session.identity_key.clone());
|
||||
}
|
||||
}
|
||||
|
||||
// Fall back to static bearer token (dev mode).
|
||||
if state.auth_cfg.allow_insecure_identity_from_request {
|
||||
if let Some(ik) = ctx.identity_key.as_deref() {
|
||||
if !ik.is_empty() {
|
||||
return Ok(ik.to_vec());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Err(HandlerResult::err(
|
||||
RpcStatus::Unauthorized,
|
||||
"invalid or expired session token",
|
||||
))
|
||||
}
|
||||
|
||||
/// Map a domain error to an RPC HandlerResult error.
|
||||
pub fn domain_err(e: crate::domain::types::DomainError) -> HandlerResult {
|
||||
use crate::domain::types::DomainError;
|
||||
match &e {
|
||||
DomainError::InvalidIdentityKey(_)
|
||||
| DomainError::EmptyPackage
|
||||
| DomainError::EmptyHybridKey
|
||||
| DomainError::EmptyUsername
|
||||
| DomainError::BlobHashLength(_)
|
||||
| DomainError::BadParams(_) => HandlerResult::err(RpcStatus::BadRequest, &e.to_string()),
|
||||
|
||||
DomainError::BlobNotFound | DomainError::DeviceNotFound => {
|
||||
HandlerResult::err(RpcStatus::NotFound, &e.to_string())
|
||||
}
|
||||
|
||||
DomainError::PackageTooLarge(_) | DomainError::BlobTooLarge(_) => {
|
||||
HandlerResult::err(RpcStatus::BadRequest, &e.to_string())
|
||||
}
|
||||
|
||||
DomainError::BlobHashMismatch => {
|
||||
HandlerResult::err(RpcStatus::BadRequest, &e.to_string())
|
||||
}
|
||||
|
||||
DomainError::DeviceLimit(_) => HandlerResult::err(RpcStatus::Forbidden, &e.to_string()),
|
||||
|
||||
DomainError::Io(_) | DomainError::Storage(_) => {
|
||||
HandlerResult::err(RpcStatus::Internal, &e.to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the v2 method registry with all 33 handlers registered.
|
||||
pub fn build_registry() -> MethodRegistry<ServerState> {
|
||||
let mut reg = MethodRegistry::new();
|
||||
|
||||
// Auth (100-103)
|
||||
reg.register(
|
||||
method_ids::OPAQUE_REGISTER_START,
|
||||
"OpaqueRegisterStart",
|
||||
auth::handle_opaque_register_start,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::OPAQUE_REGISTER_FINISH,
|
||||
"OpaqueRegisterFinish",
|
||||
auth::handle_opaque_register_finish,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::OPAQUE_LOGIN_START,
|
||||
"OpaqueLoginStart",
|
||||
auth::handle_opaque_login_start,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::OPAQUE_LOGIN_FINISH,
|
||||
"OpaqueLoginFinish",
|
||||
auth::handle_opaque_login_finish,
|
||||
);
|
||||
|
||||
// Delivery (200-205)
|
||||
reg.register(method_ids::ENQUEUE, "Enqueue", delivery::handle_enqueue);
|
||||
reg.register(method_ids::FETCH, "Fetch", delivery::handle_fetch);
|
||||
reg.register(
|
||||
method_ids::FETCH_WAIT,
|
||||
"FetchWait",
|
||||
delivery::handle_fetch_wait,
|
||||
);
|
||||
reg.register(method_ids::PEEK, "Peek", delivery::handle_peek);
|
||||
reg.register(method_ids::ACK, "Ack", delivery::handle_ack);
|
||||
reg.register(
|
||||
method_ids::BATCH_ENQUEUE,
|
||||
"BatchEnqueue",
|
||||
delivery::handle_batch_enqueue,
|
||||
);
|
||||
|
||||
// Keys (300-304)
|
||||
reg.register(
|
||||
method_ids::UPLOAD_KEY_PACKAGE,
|
||||
"UploadKeyPackage",
|
||||
keys::handle_upload_key_package,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::FETCH_KEY_PACKAGE,
|
||||
"FetchKeyPackage",
|
||||
keys::handle_fetch_key_package,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::UPLOAD_HYBRID_KEY,
|
||||
"UploadHybridKey",
|
||||
keys::handle_upload_hybrid_key,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::FETCH_HYBRID_KEY,
|
||||
"FetchHybridKey",
|
||||
keys::handle_fetch_hybrid_key,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::FETCH_HYBRID_KEYS,
|
||||
"FetchHybridKeys",
|
||||
keys::handle_fetch_hybrid_keys,
|
||||
);
|
||||
|
||||
// Channel (400)
|
||||
reg.register(
|
||||
method_ids::CREATE_CHANNEL,
|
||||
"CreateChannel",
|
||||
channel::handle_create_channel,
|
||||
);
|
||||
|
||||
// User (500-501)
|
||||
reg.register(
|
||||
method_ids::RESOLVE_USER,
|
||||
"ResolveUser",
|
||||
user::handle_resolve_user,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::RESOLVE_IDENTITY,
|
||||
"ResolveIdentity",
|
||||
user::handle_resolve_identity,
|
||||
);
|
||||
|
||||
// Blob (600-601)
|
||||
reg.register(
|
||||
method_ids::UPLOAD_BLOB,
|
||||
"UploadBlob",
|
||||
blob::handle_upload_blob,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::DOWNLOAD_BLOB,
|
||||
"DownloadBlob",
|
||||
blob::handle_download_blob,
|
||||
);
|
||||
|
||||
// Device (700-702)
|
||||
reg.register(
|
||||
method_ids::REGISTER_DEVICE,
|
||||
"RegisterDevice",
|
||||
device::handle_register_device,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::LIST_DEVICES,
|
||||
"ListDevices",
|
||||
device::handle_list_devices,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::REVOKE_DEVICE,
|
||||
"RevokeDevice",
|
||||
device::handle_revoke_device,
|
||||
);
|
||||
|
||||
// P2P (800-802)
|
||||
reg.register(
|
||||
method_ids::PUBLISH_ENDPOINT,
|
||||
"PublishEndpoint",
|
||||
p2p::handle_publish_endpoint,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::RESOLVE_ENDPOINT,
|
||||
"ResolveEndpoint",
|
||||
p2p::handle_resolve_endpoint,
|
||||
);
|
||||
reg.register(method_ids::HEALTH, "Health", p2p::handle_health);
|
||||
|
||||
// Federation (900-905)
|
||||
reg.register(
|
||||
method_ids::RELAY_ENQUEUE,
|
||||
"RelayEnqueue",
|
||||
federation::handle_relay_enqueue,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::RELAY_BATCH_ENQUEUE,
|
||||
"RelayBatchEnqueue",
|
||||
federation::handle_relay_batch_enqueue,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::PROXY_FETCH_KEY_PACKAGE,
|
||||
"ProxyFetchKeyPackage",
|
||||
federation::handle_proxy_fetch_key_package,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::PROXY_FETCH_HYBRID_KEY,
|
||||
"ProxyFetchHybridKey",
|
||||
federation::handle_proxy_fetch_hybrid_key,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::PROXY_RESOLVE_USER,
|
||||
"ProxyResolveUser",
|
||||
federation::handle_proxy_resolve_user,
|
||||
);
|
||||
reg.register(
|
||||
method_ids::FEDERATION_HEALTH,
|
||||
"FederationHealth",
|
||||
federation::handle_federation_health,
|
||||
);
|
||||
|
||||
// Account (950)
|
||||
reg.register(
|
||||
method_ids::DELETE_ACCOUNT,
|
||||
"DeleteAccount",
|
||||
account::handle_delete_account,
|
||||
);
|
||||
|
||||
reg
|
||||
}
|
||||
Reference in New Issue
Block a user