Page MenuHomePhabricator
Differential D6382

[backup-service] Rename 'GetClient' to 'BlobDownloader'
ClosedPublic
Actions

Authored by bartek on Jan 25 2023, 2:11 AM.
Tags
None
Referenced Files
F6173968: D6382.diff
Wed, Apr 23, 3:14 AM
Unknown Object (File)
Mar 16 2025, 2:04 PM
Unknown Object (File)
Mar 5 2025, 10:37 PM
Unknown Object (File)
Mar 5 2025, 10:37 PM
Unknown Object (File)
Mar 5 2025, 10:36 PM
Unknown Object (File)
Mar 5 2025, 10:30 PM
Unknown Object (File)
Feb 24 2025, 8:03 AM
Unknown Object (File)
Feb 18 2025, 9:12 AM
View All Files
Subscribers
ashoat
atul

Details

Reviewers
varun
• jon
michal
tomek
Commits
rCOMM0e95d2c872ce: [backup-service] Rename 'GetClient' to 'BlobDownloader'
Summary

Part of https://linear.app/comm/issue/ENG-2688/backup-service-reuse-single-blob-client-instance

Renamed GetClient to BlobDownloader, because this name is more accurate - these instances are not clients, but rather single RPC call instances. Also, old name could easily be confused with actual Blob Client instance, introduced in further diffs.

Test Plan

Backup service compiles and starts.

Diff Detail

Repository
rCOMM Comm
Lint
No Lint Coverage
Unit
No Test Coverage

Revision Contents
Changeset List

PathSize
services/
backup/
src/
blob/
downloader.rs
get_client.rs
12 lines
4 lines
service/
handlers/
14 lines

Diff 21296

View Options

services/backup/src/blob/downloader.rs

#![deny(clippy::all)] use argon2::Argon2;
use digest::{generic_array::GenericArray, Digest};
use napi::bindgen_prelude::*;
use opaque_ke::{
ciphersuite::CipherSuite, errors::InternalPakeError, hash::Hash,
slow_hash::SlowHash, ClientLogin, ClientLoginFinishParameters,
ClientLoginStartParameters, ClientLoginStartResult, ClientRegistration,
ClientRegistrationFinishParameters, CredentialFinalization,
CredentialResponse, RegistrationResponse, RegistrationUpload,
};
use rand::{rngs::OsRng, CryptoRng, Rng};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::Request;
use tracing::{error, instrument};
mod identity {
tonic::include_proto!("identity");
}
use identity::identity_service_client::IdentityServiceClient;
use identity::{
pake_login_response::Data::AccessToken,
pake_login_response::Data::PakeCredentialResponse,
registration_request::Data::PakeCredentialFinalization as RegistrationPakeCredentialFinalization,
registration_request::Data::PakeRegistrationRequestAndUserId,
registration_request::Data::PakeRegistrationUploadAndCredentialRequest,
registration_response::Data::PakeLoginResponse as RegistrationPakeLoginResponse,
registration_response::Data::PakeRegistrationResponse,
PakeLoginResponse as PakeLoginResponseStruct,
PakeRegistrationRequestAndUserId as PakeRegistrationRequestAndUserIdStruct,
PakeRegistrationUploadAndCredentialRequest as PakeRegistrationUploadAndCredentialRequestStruct,
RegistrationRequest, RegistrationResponse as RegistrationResponseMessage,
};
#[macro_use] #[macro_use]
extern crate napi_derive; extern crate napi_derive;
const IDENTITY_SERVICE_SOCKET_ADDR: &str = "https://[::1]:50051";
pub struct Cipher;
impl CipherSuite for Cipher {
type Group = curve25519_dalek::ristretto::RistrettoPoint;
type KeyExchange = opaque_ke::key_exchange::tripledh::TripleDH;
type Hash = sha2::Sha512;
type SlowHash = ArgonWrapper;
}
pub struct ArgonWrapper(Argon2<'static>);
impl<D: Hash> SlowHash<D> for ArgonWrapper {
fn hash(
input: GenericArray<u8, <D as Digest>::OutputSize>,
) -> std::result::Result<Vec<u8>, InternalPakeError> {
let params = Argon2::default();
let mut output = vec![0u8; <D as Digest>::output_size()];
params
.hash_password_into(&input, &[0; argon2::MIN_SALT_LEN], &mut output)
.map_err(|_| InternalPakeError::SlowHashError)?;
Ok(output)
}
}
#[napi]
varunAuthorUnsubmitted
Done
Inline Actions

this macro exports the function to js

varun: this macro exports the function to js
#[instrument]
pub async fn register_user(
user_id: String,
device_id: String,
username: String,
password: String,
user_public_key: String,
) -> Result<String> {
// Create a gRPC client
let mut identity_client =
varunAuthorUnsubmitted
Done
Inline Actions

we should probably just instantiate a single client wrapped in an Arc. see linear task

varun: we should probably just instantiate a single client wrapped in an Arc. see linear task
IdentityServiceClient::connect(IDENTITY_SERVICE_SOCKET_ADDR)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Create a RegistrationRequest channel and use ReceiverStream to turn the
// MPSC receiver into a Stream for outbound messages
let (tx, rx) = mpsc::channel(1);
let stream = ReceiverStream::new(rx);
let request = Request::new(stream);
// `response` is the Stream for inbound messages
let mut response = identity_client
.register_user(request)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?
.into_inner();
// Start PAKE registration on client and send initial registration request
// to Identity service
let mut client_rng = OsRng;
let (registration_request, client_registration) = pake_registration_start(
&mut client_rng,
user_id,
&password,
device_id,
username,
user_public_key,
)?;
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
varunAuthorUnsubmitted
Done
Inline Actions

linear task covers improving these errors

varun: linear task covers improving these errors
}
// Handle responses from Identity service sequentially, making sure we get
// messages in the correct order
// Finish PAKE registration and begin PAKE login; send the final
// registration request and initial login request together to reduce the
// number of trips
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
let client_login = handle_registration_response(
message,
&mut client_rng,
client_registration,
&password,
tx.clone(),
)
.await?;
// Finish PAKE login; send final login request to Identity service
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_credential_response(message, client_login, tx)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Return access token
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_token_response(message)
}
fn pake_registration_start(
rng: &mut (impl Rng + CryptoRng),
user_id: String,
password: &str,
device_id: String,
username: String,
user_public_key: String,
) -> Result<(RegistrationRequest, ClientRegistration<Cipher>)> {
let client_registration_start_result =
ClientRegistration::<Cipher>::start(rng, password.as_bytes()).map_err(
|e| {
error!("Failed to start PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
},
)?;
let pake_registration_request =
client_registration_start_result.message.serialize();
Ok((
RegistrationRequest {
data: Some(PakeRegistrationRequestAndUserId(
PakeRegistrationRequestAndUserIdStruct {
user_id,
device_id,
pake_registration_request,
username,
user_public_key,
},
)),
},
client_registration_start_result.state,
))
}
async fn handle_registration_response(
message: Option<RegistrationResponseMessage>,
client_rng: &mut (impl Rng + CryptoRng),
client_registration: ClientRegistration<Cipher>,
password: &str,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<ClientLogin<Cipher>> {
if let Some(RegistrationResponseMessage {
data: Some(PakeRegistrationResponse(registration_response_bytes)),
..
}) = message
{
let pake_registration_upload = pake_registration_finish(
client_rng,
&registration_response_bytes,
client_registration,
)?
.serialize();
let client_login_start_result = pake_login_start(client_rng, password)?;
// `registration_request` is a gRPC message containing serialized bytes to
// complete PAKE registration and begin PAKE login
let registration_request = RegistrationRequest {
data: Some(PakeRegistrationUploadAndCredentialRequest(
PakeRegistrationUploadAndCredentialRequestStruct {
pake_registration_upload,
pake_credential_request: client_login_start_result
.message
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
},
)),
};
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(client_login_start_result.state)
} else {
Err(handle_unexpected_response(message))
}
}
async fn handle_registration_credential_response(
message: Option<RegistrationResponseMessage>,
client_login: ClientLogin<Cipher>,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<()> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(PakeCredentialResponse(credential_response_bytes)),
})),
}) = message
{
let registration_request = RegistrationRequest {
data: Some(RegistrationPakeCredentialFinalization(
pake_login_finish(&credential_response_bytes, client_login)?
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
)),
};
send_to_mpsc(tx, registration_request).await
} else {
Err(handle_unexpected_response(message))
}
}
fn handle_registration_token_response(
message: Option<RegistrationResponseMessage>,
) -> Result<String> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(AccessToken(access_token)),
})),
}) = message
{
Ok(access_token)
} else {
Err(handle_unexpected_response(message))
}
}
async fn send_to_mpsc<T>(tx: mpsc::Sender<T>, request: T) -> Result<()> {
if let Err(e) = tx.send(request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(())
}
fn handle_unexpected_response<T: std::fmt::Debug>(message: Option<T>) -> Error {
error!("Received an unexpected message: {:?}", message);
Error::from_status(Status::GenericFailure)
}
fn pake_registration_finish(
rng: &mut (impl Rng + CryptoRng),
registration_response_bytes: &[u8],
client_registration: ClientRegistration<Cipher>,
) -> Result<RegistrationUpload<Cipher>> {
client_registration
.finish(
rng,
RegistrationResponse::deserialize(registration_response_bytes).map_err(
|e| {
error!("Could not deserialize registration response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientRegistrationFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
fn pake_login_start(
rng: &mut (impl Rng + CryptoRng),
password: &str,
) -> Result<ClientLoginStartResult<Cipher>> {
ClientLogin::<Cipher>::start(
rng,
password.as_bytes(),
ClientLoginStartParameters::default(),
)
.map_err(|e| {
error!("Failed to start PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
}
fn pake_login_finish(
credential_response_bytes: &[u8],
client_login: ClientLogin<Cipher>,
) -> Result<CredentialFinalization<Cipher>> {
client_login
.finish(
CredentialResponse::deserialize(credential_response_bytes).map_err(
|e| {
error!("Could not deserialize credential response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientLoginFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
#[napi] #[napi]
pub fn sum(a: i32, b: i32) -> i32 { pub fn sum(a: i32, b: i32) -> i32 {
a + b a + b
} }
View Options

services/backup/src/blob/get_client.rs

#![deny(clippy::all)] use argon2::Argon2;
use digest::{generic_array::GenericArray, Digest};
use napi::bindgen_prelude::*;
use opaque_ke::{
ciphersuite::CipherSuite, errors::InternalPakeError, hash::Hash,
slow_hash::SlowHash, ClientLogin, ClientLoginFinishParameters,
ClientLoginStartParameters, ClientLoginStartResult, ClientRegistration,
ClientRegistrationFinishParameters, CredentialFinalization,
CredentialResponse, RegistrationResponse, RegistrationUpload,
};
use rand::{rngs::OsRng, CryptoRng, Rng};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::Request;
use tracing::{error, instrument};
mod identity {
tonic::include_proto!("identity");
}
use identity::identity_service_client::IdentityServiceClient;
use identity::{
pake_login_response::Data::AccessToken,
pake_login_response::Data::PakeCredentialResponse,
registration_request::Data::PakeCredentialFinalization as RegistrationPakeCredentialFinalization,
registration_request::Data::PakeRegistrationRequestAndUserId,
registration_request::Data::PakeRegistrationUploadAndCredentialRequest,
registration_response::Data::PakeLoginResponse as RegistrationPakeLoginResponse,
registration_response::Data::PakeRegistrationResponse,
PakeLoginResponse as PakeLoginResponseStruct,
PakeRegistrationRequestAndUserId as PakeRegistrationRequestAndUserIdStruct,
PakeRegistrationUploadAndCredentialRequest as PakeRegistrationUploadAndCredentialRequestStruct,
RegistrationRequest, RegistrationResponse as RegistrationResponseMessage,
};
#[macro_use] #[macro_use]
extern crate napi_derive; extern crate napi_derive;
const IDENTITY_SERVICE_SOCKET_ADDR: &str = "https://[::1]:50051";
pub struct Cipher;
impl CipherSuite for Cipher {
type Group = curve25519_dalek::ristretto::RistrettoPoint;
type KeyExchange = opaque_ke::key_exchange::tripledh::TripleDH;
type Hash = sha2::Sha512;
type SlowHash = ArgonWrapper;
}
pub struct ArgonWrapper(Argon2<'static>);
impl<D: Hash> SlowHash<D> for ArgonWrapper {
fn hash(
input: GenericArray<u8, <D as Digest>::OutputSize>,
) -> std::result::Result<Vec<u8>, InternalPakeError> {
let params = Argon2::default();
let mut output = vec![0u8; <D as Digest>::output_size()];
params
.hash_password_into(&input, &[0; argon2::MIN_SALT_LEN], &mut output)
.map_err(|_| InternalPakeError::SlowHashError)?;
Ok(output)
}
}
#[napi]
varunAuthorUnsubmitted
Done
Inline Actions

this macro exports the function to js

varun: this macro exports the function to js
#[instrument]
pub async fn register_user(
user_id: String,
device_id: String,
username: String,
password: String,
user_public_key: String,
) -> Result<String> {
// Create a gRPC client
let mut identity_client =
varunAuthorUnsubmitted
Done
Inline Actions

we should probably just instantiate a single client wrapped in an Arc. see linear task

varun: we should probably just instantiate a single client wrapped in an Arc. see linear task
IdentityServiceClient::connect(IDENTITY_SERVICE_SOCKET_ADDR)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Create a RegistrationRequest channel and use ReceiverStream to turn the
// MPSC receiver into a Stream for outbound messages
let (tx, rx) = mpsc::channel(1);
let stream = ReceiverStream::new(rx);
let request = Request::new(stream);
// `response` is the Stream for inbound messages
let mut response = identity_client
.register_user(request)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?
.into_inner();
// Start PAKE registration on client and send initial registration request
// to Identity service
let mut client_rng = OsRng;
let (registration_request, client_registration) = pake_registration_start(
&mut client_rng,
user_id,
&password,
device_id,
username,
user_public_key,
)?;
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
varunAuthorUnsubmitted
Done
Inline Actions

linear task covers improving these errors

varun: linear task covers improving these errors
}
// Handle responses from Identity service sequentially, making sure we get
// messages in the correct order
// Finish PAKE registration and begin PAKE login; send the final
// registration request and initial login request together to reduce the
// number of trips
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
let client_login = handle_registration_response(
message,
&mut client_rng,
client_registration,
&password,
tx.clone(),
)
.await?;
// Finish PAKE login; send final login request to Identity service
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_credential_response(message, client_login, tx)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Return access token
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_token_response(message)
}
fn pake_registration_start(
rng: &mut (impl Rng + CryptoRng),
user_id: String,
password: &str,
device_id: String,
username: String,
user_public_key: String,
) -> Result<(RegistrationRequest, ClientRegistration<Cipher>)> {
let client_registration_start_result =
ClientRegistration::<Cipher>::start(rng, password.as_bytes()).map_err(
|e| {
error!("Failed to start PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
},
)?;
let pake_registration_request =
client_registration_start_result.message.serialize();
Ok((
RegistrationRequest {
data: Some(PakeRegistrationRequestAndUserId(
PakeRegistrationRequestAndUserIdStruct {
user_id,
device_id,
pake_registration_request,
username,
user_public_key,
},
)),
},
client_registration_start_result.state,
))
}
async fn handle_registration_response(
message: Option<RegistrationResponseMessage>,
client_rng: &mut (impl Rng + CryptoRng),
client_registration: ClientRegistration<Cipher>,
password: &str,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<ClientLogin<Cipher>> {
if let Some(RegistrationResponseMessage {
data: Some(PakeRegistrationResponse(registration_response_bytes)),
..
}) = message
{
let pake_registration_upload = pake_registration_finish(
client_rng,
&registration_response_bytes,
client_registration,
)?
.serialize();
let client_login_start_result = pake_login_start(client_rng, password)?;
// `registration_request` is a gRPC message containing serialized bytes to
// complete PAKE registration and begin PAKE login
let registration_request = RegistrationRequest {
data: Some(PakeRegistrationUploadAndCredentialRequest(
PakeRegistrationUploadAndCredentialRequestStruct {
pake_registration_upload,
pake_credential_request: client_login_start_result
.message
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
},
)),
};
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(client_login_start_result.state)
} else {
Err(handle_unexpected_response(message))
}
}
async fn handle_registration_credential_response(
message: Option<RegistrationResponseMessage>,
client_login: ClientLogin<Cipher>,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<()> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(PakeCredentialResponse(credential_response_bytes)),
})),
}) = message
{
let registration_request = RegistrationRequest {
data: Some(RegistrationPakeCredentialFinalization(
pake_login_finish(&credential_response_bytes, client_login)?
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
)),
};
send_to_mpsc(tx, registration_request).await
} else {
Err(handle_unexpected_response(message))
}
}
fn handle_registration_token_response(
message: Option<RegistrationResponseMessage>,
) -> Result<String> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(AccessToken(access_token)),
})),
}) = message
{
Ok(access_token)
} else {
Err(handle_unexpected_response(message))
}
}
async fn send_to_mpsc<T>(tx: mpsc::Sender<T>, request: T) -> Result<()> {
if let Err(e) = tx.send(request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(())
}
fn handle_unexpected_response<T: std::fmt::Debug>(message: Option<T>) -> Error {
error!("Received an unexpected message: {:?}", message);
Error::from_status(Status::GenericFailure)
}
fn pake_registration_finish(
rng: &mut (impl Rng + CryptoRng),
registration_response_bytes: &[u8],
client_registration: ClientRegistration<Cipher>,
) -> Result<RegistrationUpload<Cipher>> {
client_registration
.finish(
rng,
RegistrationResponse::deserialize(registration_response_bytes).map_err(
|e| {
error!("Could not deserialize registration response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientRegistrationFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
fn pake_login_start(
rng: &mut (impl Rng + CryptoRng),
password: &str,
) -> Result<ClientLoginStartResult<Cipher>> {
ClientLogin::<Cipher>::start(
rng,
password.as_bytes(),
ClientLoginStartParameters::default(),
)
.map_err(|e| {
error!("Failed to start PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
}
fn pake_login_finish(
credential_response_bytes: &[u8],
client_login: ClientLogin<Cipher>,
) -> Result<CredentialFinalization<Cipher>> {
client_login
.finish(
CredentialResponse::deserialize(credential_response_bytes).map_err(
|e| {
error!("Could not deserialize credential response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientLoginFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
#[napi] #[napi]
pub fn sum(a: i32, b: i32) -> i32 { pub fn sum(a: i32, b: i32) -> i32 {
a + b a + b
} }
View Options

services/backup/src/blob/mod.rs

#![deny(clippy::all)] use argon2::Argon2;
use digest::{generic_array::GenericArray, Digest};
use napi::bindgen_prelude::*;
use opaque_ke::{
ciphersuite::CipherSuite, errors::InternalPakeError, hash::Hash,
slow_hash::SlowHash, ClientLogin, ClientLoginFinishParameters,
ClientLoginStartParameters, ClientLoginStartResult, ClientRegistration,
ClientRegistrationFinishParameters, CredentialFinalization,
CredentialResponse, RegistrationResponse, RegistrationUpload,
};
use rand::{rngs::OsRng, CryptoRng, Rng};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::Request;
use tracing::{error, instrument};
mod identity {
tonic::include_proto!("identity");
}
use identity::identity_service_client::IdentityServiceClient;
use identity::{
pake_login_response::Data::AccessToken,
pake_login_response::Data::PakeCredentialResponse,
registration_request::Data::PakeCredentialFinalization as RegistrationPakeCredentialFinalization,
registration_request::Data::PakeRegistrationRequestAndUserId,
registration_request::Data::PakeRegistrationUploadAndCredentialRequest,
registration_response::Data::PakeLoginResponse as RegistrationPakeLoginResponse,
registration_response::Data::PakeRegistrationResponse,
PakeLoginResponse as PakeLoginResponseStruct,
PakeRegistrationRequestAndUserId as PakeRegistrationRequestAndUserIdStruct,
PakeRegistrationUploadAndCredentialRequest as PakeRegistrationUploadAndCredentialRequestStruct,
RegistrationRequest, RegistrationResponse as RegistrationResponseMessage,
};
#[macro_use] #[macro_use]
extern crate napi_derive; extern crate napi_derive;
const IDENTITY_SERVICE_SOCKET_ADDR: &str = "https://[::1]:50051";
pub struct Cipher;
impl CipherSuite for Cipher {
type Group = curve25519_dalek::ristretto::RistrettoPoint;
type KeyExchange = opaque_ke::key_exchange::tripledh::TripleDH;
type Hash = sha2::Sha512;
type SlowHash = ArgonWrapper;
}
pub struct ArgonWrapper(Argon2<'static>);
impl<D: Hash> SlowHash<D> for ArgonWrapper {
fn hash(
input: GenericArray<u8, <D as Digest>::OutputSize>,
) -> std::result::Result<Vec<u8>, InternalPakeError> {
let params = Argon2::default();
let mut output = vec![0u8; <D as Digest>::output_size()];
params
.hash_password_into(&input, &[0; argon2::MIN_SALT_LEN], &mut output)
.map_err(|_| InternalPakeError::SlowHashError)?;
Ok(output)
}
}
#[napi]
varunAuthorUnsubmitted
Done
Inline Actions

this macro exports the function to js

varun: this macro exports the function to js
#[instrument]
pub async fn register_user(
user_id: String,
device_id: String,
username: String,
password: String,
user_public_key: String,
) -> Result<String> {
// Create a gRPC client
let mut identity_client =
varunAuthorUnsubmitted
Done
Inline Actions

we should probably just instantiate a single client wrapped in an Arc. see linear task

varun: we should probably just instantiate a single client wrapped in an Arc. see linear task
IdentityServiceClient::connect(IDENTITY_SERVICE_SOCKET_ADDR)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Create a RegistrationRequest channel and use ReceiverStream to turn the
// MPSC receiver into a Stream for outbound messages
let (tx, rx) = mpsc::channel(1);
let stream = ReceiverStream::new(rx);
let request = Request::new(stream);
// `response` is the Stream for inbound messages
let mut response = identity_client
.register_user(request)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?
.into_inner();
// Start PAKE registration on client and send initial registration request
// to Identity service
let mut client_rng = OsRng;
let (registration_request, client_registration) = pake_registration_start(
&mut client_rng,
user_id,
&password,
device_id,
username,
user_public_key,
)?;
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
varunAuthorUnsubmitted
Done
Inline Actions

linear task covers improving these errors

varun: linear task covers improving these errors
}
// Handle responses from Identity service sequentially, making sure we get
// messages in the correct order
// Finish PAKE registration and begin PAKE login; send the final
// registration request and initial login request together to reduce the
// number of trips
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
let client_login = handle_registration_response(
message,
&mut client_rng,
client_registration,
&password,
tx.clone(),
)
.await?;
// Finish PAKE login; send final login request to Identity service
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_credential_response(message, client_login, tx)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Return access token
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_token_response(message)
}
fn pake_registration_start(
rng: &mut (impl Rng + CryptoRng),
user_id: String,
password: &str,
device_id: String,
username: String,
user_public_key: String,
) -> Result<(RegistrationRequest, ClientRegistration<Cipher>)> {
let client_registration_start_result =
ClientRegistration::<Cipher>::start(rng, password.as_bytes()).map_err(
|e| {
error!("Failed to start PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
},
)?;
let pake_registration_request =
client_registration_start_result.message.serialize();
Ok((
RegistrationRequest {
data: Some(PakeRegistrationRequestAndUserId(
PakeRegistrationRequestAndUserIdStruct {
user_id,
device_id,
pake_registration_request,
username,
user_public_key,
},
)),
},
client_registration_start_result.state,
))
}
async fn handle_registration_response(
message: Option<RegistrationResponseMessage>,
client_rng: &mut (impl Rng + CryptoRng),
client_registration: ClientRegistration<Cipher>,
password: &str,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<ClientLogin<Cipher>> {
if let Some(RegistrationResponseMessage {
data: Some(PakeRegistrationResponse(registration_response_bytes)),
..
}) = message
{
let pake_registration_upload = pake_registration_finish(
client_rng,
&registration_response_bytes,
client_registration,
)?
.serialize();
let client_login_start_result = pake_login_start(client_rng, password)?;
// `registration_request` is a gRPC message containing serialized bytes to
// complete PAKE registration and begin PAKE login
let registration_request = RegistrationRequest {
data: Some(PakeRegistrationUploadAndCredentialRequest(
PakeRegistrationUploadAndCredentialRequestStruct {
pake_registration_upload,
pake_credential_request: client_login_start_result
.message
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
},
)),
};
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(client_login_start_result.state)
} else {
Err(handle_unexpected_response(message))
}
}
async fn handle_registration_credential_response(
message: Option<RegistrationResponseMessage>,
client_login: ClientLogin<Cipher>,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<()> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(PakeCredentialResponse(credential_response_bytes)),
})),
}) = message
{
let registration_request = RegistrationRequest {
data: Some(RegistrationPakeCredentialFinalization(
pake_login_finish(&credential_response_bytes, client_login)?
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
)),
};
send_to_mpsc(tx, registration_request).await
} else {
Err(handle_unexpected_response(message))
}
}
fn handle_registration_token_response(
message: Option<RegistrationResponseMessage>,
) -> Result<String> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(AccessToken(access_token)),
})),
}) = message
{
Ok(access_token)
} else {
Err(handle_unexpected_response(message))
}
}
async fn send_to_mpsc<T>(tx: mpsc::Sender<T>, request: T) -> Result<()> {
if let Err(e) = tx.send(request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(())
}
fn handle_unexpected_response<T: std::fmt::Debug>(message: Option<T>) -> Error {
error!("Received an unexpected message: {:?}", message);
Error::from_status(Status::GenericFailure)
}
fn pake_registration_finish(
rng: &mut (impl Rng + CryptoRng),
registration_response_bytes: &[u8],
client_registration: ClientRegistration<Cipher>,
) -> Result<RegistrationUpload<Cipher>> {
client_registration
.finish(
rng,
RegistrationResponse::deserialize(registration_response_bytes).map_err(
|e| {
error!("Could not deserialize registration response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientRegistrationFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
fn pake_login_start(
rng: &mut (impl Rng + CryptoRng),
password: &str,
) -> Result<ClientLoginStartResult<Cipher>> {
ClientLogin::<Cipher>::start(
rng,
password.as_bytes(),
ClientLoginStartParameters::default(),
)
.map_err(|e| {
error!("Failed to start PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
}
fn pake_login_finish(
credential_response_bytes: &[u8],
client_login: ClientLogin<Cipher>,
) -> Result<CredentialFinalization<Cipher>> {
client_login
.finish(
CredentialResponse::deserialize(credential_response_bytes).map_err(
|e| {
error!("Could not deserialize credential response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientLoginFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
#[napi] #[napi]
pub fn sum(a: i32, b: i32) -> i32 { pub fn sum(a: i32, b: i32) -> i32 {
a + b a + b
} }
View Options

services/backup/src/service/handlers/pull_backup.rs

#![deny(clippy::all)] use argon2::Argon2;
use digest::{generic_array::GenericArray, Digest};
use napi::bindgen_prelude::*;
use opaque_ke::{
ciphersuite::CipherSuite, errors::InternalPakeError, hash::Hash,
slow_hash::SlowHash, ClientLogin, ClientLoginFinishParameters,
ClientLoginStartParameters, ClientLoginStartResult, ClientRegistration,
ClientRegistrationFinishParameters, CredentialFinalization,
CredentialResponse, RegistrationResponse, RegistrationUpload,
};
use rand::{rngs::OsRng, CryptoRng, Rng};
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::Request;
use tracing::{error, instrument};
mod identity {
tonic::include_proto!("identity");
}
use identity::identity_service_client::IdentityServiceClient;
use identity::{
pake_login_response::Data::AccessToken,
pake_login_response::Data::PakeCredentialResponse,
registration_request::Data::PakeCredentialFinalization as RegistrationPakeCredentialFinalization,
registration_request::Data::PakeRegistrationRequestAndUserId,
registration_request::Data::PakeRegistrationUploadAndCredentialRequest,
registration_response::Data::PakeLoginResponse as RegistrationPakeLoginResponse,
registration_response::Data::PakeRegistrationResponse,
PakeLoginResponse as PakeLoginResponseStruct,
PakeRegistrationRequestAndUserId as PakeRegistrationRequestAndUserIdStruct,
PakeRegistrationUploadAndCredentialRequest as PakeRegistrationUploadAndCredentialRequestStruct,
RegistrationRequest, RegistrationResponse as RegistrationResponseMessage,
};
#[macro_use] #[macro_use]
extern crate napi_derive; extern crate napi_derive;
const IDENTITY_SERVICE_SOCKET_ADDR: &str = "https://[::1]:50051";
pub struct Cipher;
impl CipherSuite for Cipher {
type Group = curve25519_dalek::ristretto::RistrettoPoint;
type KeyExchange = opaque_ke::key_exchange::tripledh::TripleDH;
type Hash = sha2::Sha512;
type SlowHash = ArgonWrapper;
}
pub struct ArgonWrapper(Argon2<'static>);
impl<D: Hash> SlowHash<D> for ArgonWrapper {
fn hash(
input: GenericArray<u8, <D as Digest>::OutputSize>,
) -> std::result::Result<Vec<u8>, InternalPakeError> {
let params = Argon2::default();
let mut output = vec![0u8; <D as Digest>::output_size()];
params
.hash_password_into(&input, &[0; argon2::MIN_SALT_LEN], &mut output)
.map_err(|_| InternalPakeError::SlowHashError)?;
Ok(output)
}
}
#[napi]
varunAuthorUnsubmitted
Done
Inline Actions

this macro exports the function to js

varun: this macro exports the function to js
#[instrument]
pub async fn register_user(
user_id: String,
device_id: String,
username: String,
password: String,
user_public_key: String,
) -> Result<String> {
// Create a gRPC client
let mut identity_client =
varunAuthorUnsubmitted
Done
Inline Actions

we should probably just instantiate a single client wrapped in an Arc. see linear task

varun: we should probably just instantiate a single client wrapped in an Arc. see linear task
IdentityServiceClient::connect(IDENTITY_SERVICE_SOCKET_ADDR)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Create a RegistrationRequest channel and use ReceiverStream to turn the
// MPSC receiver into a Stream for outbound messages
let (tx, rx) = mpsc::channel(1);
let stream = ReceiverStream::new(rx);
let request = Request::new(stream);
// `response` is the Stream for inbound messages
let mut response = identity_client
.register_user(request)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?
.into_inner();
// Start PAKE registration on client and send initial registration request
// to Identity service
let mut client_rng = OsRng;
let (registration_request, client_registration) = pake_registration_start(
&mut client_rng,
user_id,
&password,
device_id,
username,
user_public_key,
)?;
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
varunAuthorUnsubmitted
Done
Inline Actions

linear task covers improving these errors

varun: linear task covers improving these errors
}
// Handle responses from Identity service sequentially, making sure we get
// messages in the correct order
// Finish PAKE registration and begin PAKE login; send the final
// registration request and initial login request together to reduce the
// number of trips
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
let client_login = handle_registration_response(
message,
&mut client_rng,
client_registration,
&password,
tx.clone(),
)
.await?;
// Finish PAKE login; send final login request to Identity service
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_credential_response(message, client_login, tx)
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
// Return access token
let message = response
.message()
.await
.map_err(|_| Error::from_status(Status::GenericFailure))?;
handle_registration_token_response(message)
}
fn pake_registration_start(
rng: &mut (impl Rng + CryptoRng),
user_id: String,
password: &str,
device_id: String,
username: String,
user_public_key: String,
) -> Result<(RegistrationRequest, ClientRegistration<Cipher>)> {
let client_registration_start_result =
ClientRegistration::<Cipher>::start(rng, password.as_bytes()).map_err(
|e| {
error!("Failed to start PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
},
)?;
let pake_registration_request =
client_registration_start_result.message.serialize();
Ok((
RegistrationRequest {
data: Some(PakeRegistrationRequestAndUserId(
PakeRegistrationRequestAndUserIdStruct {
user_id,
device_id,
pake_registration_request,
username,
user_public_key,
},
)),
},
client_registration_start_result.state,
))
}
async fn handle_registration_response(
message: Option<RegistrationResponseMessage>,
client_rng: &mut (impl Rng + CryptoRng),
client_registration: ClientRegistration<Cipher>,
password: &str,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<ClientLogin<Cipher>> {
if let Some(RegistrationResponseMessage {
data: Some(PakeRegistrationResponse(registration_response_bytes)),
..
}) = message
{
let pake_registration_upload = pake_registration_finish(
client_rng,
&registration_response_bytes,
client_registration,
)?
.serialize();
let client_login_start_result = pake_login_start(client_rng, password)?;
// `registration_request` is a gRPC message containing serialized bytes to
// complete PAKE registration and begin PAKE login
let registration_request = RegistrationRequest {
data: Some(PakeRegistrationUploadAndCredentialRequest(
PakeRegistrationUploadAndCredentialRequestStruct {
pake_registration_upload,
pake_credential_request: client_login_start_result
.message
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
},
)),
};
if let Err(e) = tx.send(registration_request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(client_login_start_result.state)
} else {
Err(handle_unexpected_response(message))
}
}
async fn handle_registration_credential_response(
message: Option<RegistrationResponseMessage>,
client_login: ClientLogin<Cipher>,
tx: mpsc::Sender<RegistrationRequest>,
) -> Result<()> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(PakeCredentialResponse(credential_response_bytes)),
})),
}) = message
{
let registration_request = RegistrationRequest {
data: Some(RegistrationPakeCredentialFinalization(
pake_login_finish(&credential_response_bytes, client_login)?
.serialize()
.map_err(|e| {
error!("Could not serialize credential request: {}", e);
Error::from_status(Status::GenericFailure)
})?,
)),
};
send_to_mpsc(tx, registration_request).await
} else {
Err(handle_unexpected_response(message))
}
}
fn handle_registration_token_response(
message: Option<RegistrationResponseMessage>,
) -> Result<String> {
if let Some(RegistrationResponseMessage {
data:
Some(RegistrationPakeLoginResponse(PakeLoginResponseStruct {
data: Some(AccessToken(access_token)),
})),
}) = message
{
Ok(access_token)
} else {
Err(handle_unexpected_response(message))
}
}
async fn send_to_mpsc<T>(tx: mpsc::Sender<T>, request: T) -> Result<()> {
if let Err(e) = tx.send(request).await {
error!("Response was dropped: {}", e);
return Err(Error::from_status(Status::GenericFailure));
}
Ok(())
}
fn handle_unexpected_response<T: std::fmt::Debug>(message: Option<T>) -> Error {
error!("Received an unexpected message: {:?}", message);
Error::from_status(Status::GenericFailure)
}
fn pake_registration_finish(
rng: &mut (impl Rng + CryptoRng),
registration_response_bytes: &[u8],
client_registration: ClientRegistration<Cipher>,
) -> Result<RegistrationUpload<Cipher>> {
client_registration
.finish(
rng,
RegistrationResponse::deserialize(registration_response_bytes).map_err(
|e| {
error!("Could not deserialize registration response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientRegistrationFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE registration: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
fn pake_login_start(
rng: &mut (impl Rng + CryptoRng),
password: &str,
) -> Result<ClientLoginStartResult<Cipher>> {
ClientLogin::<Cipher>::start(
rng,
password.as_bytes(),
ClientLoginStartParameters::default(),
)
.map_err(|e| {
error!("Failed to start PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
}
fn pake_login_finish(
credential_response_bytes: &[u8],
client_login: ClientLogin<Cipher>,
) -> Result<CredentialFinalization<Cipher>> {
client_login
.finish(
CredentialResponse::deserialize(credential_response_bytes).map_err(
|e| {
error!("Could not deserialize credential response bytes: {}", e);
Error::from_status(Status::GenericFailure)
},
)?,
ClientLoginFinishParameters::default(),
)
.map_err(|e| {
error!("Failed to finish PAKE login: {}", e);
Error::from_status(Status::GenericFailure)
})
.map(|res| res.message)
}
#[napi] #[napi]
pub fn sum(a: i32, b: i32) -> i32 { pub fn sum(a: i32, b: i32) -> i32 {
a + b a + b
} }