use crate::StateSynched;
use super::{
wallet::Wallet, ActorEvent, Error, Outcome, ReplicaValidator, Result, TernaryResult,
TransferInitiated, TransferRegistrationSent, TransferValidated, TransferValidationReceived,
TransfersSynched,
};
use crdts::Dot;
use itertools::Itertools;
use log::debug;
use sn_data_types::{
ActorHistory, Credit, CreditAgreementProof, CreditId, Debit, DebitId, OwnerType, PublicKey,
SignatureShare, SignedCredit, SignedDebit, Signing, Token, TransferAgreementProof, WalletInfo,
};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt;
use threshold_crypto::PublicKeySet;
#[derive(Clone)]
pub struct Actor<V: ReplicaValidator, S: Signing> {
id: OwnerType,
signing: S,
wallet: Wallet,
next_expected_debit: u64,
accumulating_validations: HashMap<DebitId, HashMap<usize, TransferValidated>>,
replicas: PublicKeySet,
replica_validator: V,
history: ActorHistory,
}
impl<V: ReplicaValidator, S: Signing> Actor<V, S> {
pub fn new(signing: S, replicas: PublicKeySet, replica_validator: V) -> Actor<V, S> {
let id = signing.id();
let wallet = Wallet::new(id.clone());
Actor {
id,
signing,
replicas,
replica_validator,
wallet,
next_expected_debit: 0,
accumulating_validations: Default::default(),
history: ActorHistory::empty(),
}
}
pub fn from_info(signing: S, info: WalletInfo, replica_validator: V) -> Result<Actor<V, S>> {
let mut actor = Self::new(signing, info.replicas, replica_validator);
if let Some(e) = actor.from_history(info.history)? {
actor.apply(ActorEvent::TransfersSynched(e))?;
}
Ok(actor)
}
pub fn from_snapshot(
wallet: Wallet,
signing: S,
replicas: PublicKeySet,
replica_validator: V,
) -> Actor<V, S> {
let id = wallet.id().clone();
Actor {
id,
signing,
replicas,
replica_validator,
wallet,
next_expected_debit: 0,
accumulating_validations: Default::default(),
history: ActorHistory::empty(),
}
}
pub fn id(&self) -> PublicKey {
self.id.public_key()
}
pub fn owner(&self) -> &OwnerType {
&self.id
}
pub fn balance(&self) -> Token {
self.wallet.balance()
}
pub fn replicas_public_key(&self) -> PublicKey {
PublicKey::Bls(self.replicas.public_key())
}
pub fn replicas_key_set(&self) -> PublicKeySet {
self.replicas.clone()
}
pub fn history(&self) -> ActorHistory {
self.history.clone()
}
pub fn transfer(
&self,
amount: Token,
recipient: PublicKey,
msg: String,
) -> Outcome<TransferInitiated> {
if recipient == self.id() {
return Outcome::rejected(Error::SameSenderAndRecipient);
}
let id = Dot::new(self.id(), self.wallet.next_debit());
if self.next_expected_debit != self.wallet.next_debit() {
return Outcome::rejected(Error::DebitPending);
}
if self.next_expected_debit != id.counter {
return Outcome::rejected(Error::DebitProposed);
}
if amount > self.balance() {
return Outcome::rejected(Error::InsufficientBalance);
}
if amount == Token::from_nano(0) {
return Outcome::rejected(Error::ZeroValueTransfer);
}
let debit = Debit { id, amount };
let credit = Credit {
id: debit.credit_id()?,
recipient,
amount,
msg,
};
let actor_signature = self.signing.sign(&debit)?;
let signed_debit = SignedDebit {
debit,
actor_signature,
};
let actor_signature = self.signing.sign(&credit)?;
let signed_credit = SignedCredit {
credit,
actor_signature,
};
Outcome::success(TransferInitiated {
signed_debit,
signed_credit,
})
}
pub fn receive(&self, validation: TransferValidated) -> Outcome<TransferValidationReceived> {
if self.verify(&validation).is_err() {
return Err(Error::InvalidSignature);
}
debug!("Actor: Verified validation.");
let signed_debit = &validation.signed_debit;
let signed_credit = &validation.signed_credit;
if signed_credit.id() != &signed_debit.credit_id()? {
return Err(Error::CreditDebitIdMismatch);
}
if self.id() != signed_debit.sender() {
return Err(Error::WrongValidationActor);
}
if self.next_expected_debit != signed_debit.id().counter + 1 {
return Err(Error::OperationOutOfOrder(
signed_debit.id().counter,
self.next_expected_debit,
));
}
if let Some(map) = self.accumulating_validations.get(&validation.id()) {
if map.contains_key(&validation.replica_debit_sig.index) {
return Err(Error::ValidatedAlready);
}
} else {
return Err(Error::NoSetForDebitId(validation.id()));
}
let map = self
.accumulating_validations
.get(&validation.id())
.ok_or_else(|| Error::NoSetForTransferId(validation.id()))?;
let mut proof = None;
let majority =
map.len() + 1 > self.replicas.threshold() && self.replicas == validation.replicas;
if majority {
let debit_bytes = match bincode::serialize(&signed_debit) {
Err(_) => return Err(Error::Serialisation("Serialization Error".to_string())),
Ok(data) => data,
};
let credit_bytes = match bincode::serialize(&signed_credit) {
Err(_) => return Err(Error::Serialisation("Serialization Error".to_string())),
Ok(data) => data,
};
let debit_sig_shares: BTreeMap<_, _> = map
.values()
.chain(vec![&validation])
.map(|v| v.replica_debit_sig.clone())
.map(|s| (s.index, s.share))
.collect();
let credit_sig_shares: BTreeMap<_, _> = map
.values()
.chain(vec![&validation])
.map(|v| v.replica_credit_sig.clone())
.map(|s| (s.index, s.share))
.collect();
let debit_sig = self
.replicas
.combine_signatures(&debit_sig_shares)
.map_err(|_| Error::CannotAggregate)?;
let credit_sig = self
.replicas
.combine_signatures(&credit_sig_shares)
.map_err(|_| Error::CannotAggregate)?;
let valid_debit = self.replicas.public_key().verify(&debit_sig, debit_bytes);
let valid_credit = self.replicas.public_key().verify(&credit_sig, credit_bytes);
if valid_debit && valid_credit {
proof = Some(TransferAgreementProof {
signed_debit: signed_debit.clone(),
debit_sig: sn_data_types::Signature::Bls(debit_sig),
signed_credit: signed_credit.clone(),
credit_sig: sn_data_types::Signature::Bls(credit_sig),
debiting_replicas_keys: self.replicas.clone(),
});
}
}
Outcome::success(TransferValidationReceived { validation, proof })
}
pub fn register(
&self,
transfer_proof: TransferAgreementProof,
) -> Outcome<TransferRegistrationSent> {
if self.verify_transfer_proof(&transfer_proof).is_err() {
return Err(Error::InvalidSignature);
}
if self.wallet.next_debit() == transfer_proof.id().counter {
Outcome::success(TransferRegistrationSent { transfer_proof })
} else {
Err(Error::OperationOutOfOrder(
transfer_proof.id().counter,
self.wallet.next_debit(),
))
}
}
pub fn synch(
&self,
balance: Token,
debit_version: u64,
credit_ids: HashSet<CreditId>,
) -> Outcome<StateSynched> {
Outcome::success(StateSynched {
id: self.id(),
balance,
debit_version,
credit_ids,
})
}
pub fn from_history(&self, history: ActorHistory) -> Outcome<TransfersSynched> {
if history.is_empty() {
return Outcome::no_change();
}
let credits = self.validate_credits(&history.credits);
let debits = self.validate_debits(&history.debits);
if !credits.is_empty() || !debits.is_empty() {
Outcome::success(TransfersSynched(ActorHistory { credits, debits }))
} else {
Err(Error::NothingToSync)
}
}
fn validate_credits(&self, credits: &[CreditAgreementProof]) -> Vec<CreditAgreementProof> {
let valid_credits: Vec<_> = credits
.iter()
.cloned()
.unique_by(|e| *e.id())
.filter(|_credit_proof| {
#[cfg(feature = "simulated-payouts")]
return true;
#[cfg(not(feature = "simulated-payouts"))]
self.verify_credit_proof(_credit_proof).is_ok()
})
.filter(|credit| self.id() == credit.recipient())
.filter(|credit| !self.wallet.contains(&credit.id()))
.collect();
valid_credits
}
#[allow(clippy::explicit_counter_loop)]
fn validate_debits(&self, debits: &[TransferAgreementProof]) -> Vec<TransferAgreementProof> {
let mut debits: Vec<_> = debits
.iter()
.unique_by(|e| e.id())
.filter(|transfer| self.id() == transfer.sender())
.filter(|transfer| transfer.id().counter >= self.wallet.next_debit())
.filter(|transfer| self.verify_transfer_proof(transfer).is_ok())
.collect();
debits.sort_by_key(|t| t.id().counter);
let mut iter = 0;
let mut valid_debits = vec![];
for out in debits {
let version = out.id().counter;
let expected_version = iter + self.wallet.next_debit();
if version != expected_version {
break;
}
valid_debits.push(out.clone());
iter += 1;
}
valid_debits
}
pub fn apply(&mut self, event: ActorEvent) -> Result<()> {
debug!("Applying event {:?}", event);
match event {
ActorEvent::TransferInitiated(e) => {
self.next_expected_debit = e.id().counter + 1;
let _ = self.accumulating_validations.insert(e.id(), HashMap::new());
Ok(())
}
ActorEvent::TransferValidationReceived(e) => {
if e.proof.is_some() {
self.replicas = e.validation.replicas.clone();
}
match self.accumulating_validations.get_mut(&e.validation.id()) {
Some(map) => {
let _ = map.insert(e.validation.replica_debit_sig.index, e.validation);
}
None => return Err(Error::PendingTransferNotFound),
}
Ok(())
}
ActorEvent::TransferRegistrationSent(e) => {
self.wallet
.apply_debit(e.transfer_proof.signed_debit.debit.clone())?;
self.accumulating_validations.clear();
self.history.debits.push(e.transfer_proof);
Ok(())
}
ActorEvent::TransfersSynched(e) => {
for credit in e.0.credits {
self.wallet
.apply_credit(credit.signed_credit.credit.clone())?;
self.history.credits.push(credit);
}
for debit in e.0.debits {
self.wallet.apply_debit(debit.signed_debit.debit.clone())?;
self.history.debits.push(debit);
}
self.next_expected_debit = self.wallet.next_debit();
Ok(())
}
ActorEvent::StateSynched(e) => {
self.wallet = Wallet::from(
self.owner().clone(),
e.balance,
e.debit_version,
e.credit_ids,
);
self.next_expected_debit = self.wallet.next_debit();
Ok(())
}
}
}
fn verify(&self, event: &TransferValidated) -> Result<()> {
let signed_debit = &event.signed_debit;
let signed_credit = &event.signed_credit;
if let error @ Err(_) = self.verify_is_our_transfer(signed_debit, signed_credit) {
return error;
}
let valid_debit = self
.verify_share(signed_debit, &event.replica_debit_sig, &event.replicas)
.is_ok();
let valid_credit = self
.verify_share(signed_credit, &event.replica_credit_sig, &event.replicas)
.is_ok();
if valid_debit && valid_credit {
Ok(())
} else {
Err(Error::InvalidSignature)
}
}
fn verify_share<T: serde::Serialize>(
&self,
item: T,
replica_signature: &SignatureShare,
replicas: &PublicKeySet,
) -> Result<()> {
let sig_share = &replica_signature.share;
let share_index = replica_signature.index;
match bincode::serialize(&item) {
Err(_) => Err(Error::Serialisation("Could not serialise item".into())),
Ok(data) => {
let verified = replicas
.public_key_share(share_index)
.verify(sig_share, data);
if verified {
Ok(())
} else {
Err(Error::InvalidSignature)
}
}
}
}
fn verify_transfer_proof(&self, proof: &TransferAgreementProof) -> Result<()> {
let signed_debit = &proof.signed_debit;
let signed_credit = &proof.signed_credit;
if let error @ Err(_) = self.verify_is_our_transfer(signed_debit, signed_credit) {
return error;
}
let valid_debit = match bincode::serialize(&proof.signed_debit) {
Err(_) => return Err(Error::Serialisation("Could not serialise debit".into())),
Ok(data) => {
let public_key = sn_data_types::PublicKey::Bls(self.replicas.public_key());
public_key.verify(&proof.debit_sig, &data).is_ok()
}
};
let valid_credit = match bincode::serialize(&proof.signed_credit) {
Err(_) => return Err(Error::Serialisation("Could not serialise credit".into())),
Ok(data) => {
let public_key = sn_data_types::PublicKey::Bls(self.replicas.public_key());
public_key.verify(&proof.credit_sig, &data).is_ok()
}
};
if valid_debit && valid_credit {
Ok(())
} else {
Err(Error::InvalidSignature)
}
}
#[cfg(not(feature = "simulated-payouts"))]
fn verify_credit_proof(&self, proof: &CreditAgreementProof) -> Result<()> {
let debiting_replicas_keys = PublicKey::Bls(proof.debiting_replicas_keys.public_key());
if !self.replica_validator.is_valid(debiting_replicas_keys) {
return Err(Error::Unknown(format!(
"Unknown debiting replica keys: {}",
debiting_replicas_keys
)));
}
debug!("Verfying debiting_replicas_sig..!");
match bincode::serialize(&proof.signed_credit) {
Err(_) => Err(Error::Serialisation("Could not serialise credit".into())),
Ok(data) => debiting_replicas_keys
.verify(&proof.debiting_replicas_sig, &data)
.map_err(Error::NetworkDataError),
}
}
fn verify_is_our_transfer(
&self,
signed_debit: &SignedDebit,
signed_credit: &SignedCredit,
) -> Result<()> {
debug!("Actor: Verifying is our transfer!");
let valid_debit = self
.signing
.verify(&signed_debit.actor_signature, &signed_debit.debit);
let valid_credit = self
.signing
.verify(&signed_credit.actor_signature, &signed_credit.credit);
if !(valid_debit && valid_credit) {
debug!(
"Actor: Valid debit sig? {}, Valid credit sig? {}",
valid_debit, valid_credit
);
Err(Error::InvalidSignature)
} else if signed_credit.id() != &signed_debit.credit_id()? {
Err(Error::CreditDebitIdMismatch)
} else {
Ok(())
}
}
}
impl<V: ReplicaValidator + fmt::Debug, S: Signing + fmt::Debug> fmt::Debug for Actor<V, S> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"Actor {{ id: {:?}, signing: {:?}, wallet: {:?}, next_expected_debit: {:?}, accumulating_validations: {:?}, replicas: PkSet {{ public_key: {:?} }}, replica_validator: {:?} }}",
self.id,
self.signing,
self.wallet,
self.next_expected_debit,
self.accumulating_validations,
self.replicas.public_key(),
self.replica_validator
)
}
}
#[cfg(test)]
mod test {
use super::{
Actor, ActorEvent, Error, OwnerType, ReplicaValidator, Result, TransferInitiated,
TransferRegistrationSent, Wallet,
};
use crdts::Dot;
use serde::Serialize;
use sn_data_types::{
Credit, Debit, Keypair, PublicKey, Signature, SignatureShare, Token,
TransferAgreementProof, TransferValidated,
};
use std::collections::BTreeMap;
use threshold_crypto::{SecretKey, SecretKeySet};
struct Validator {}
impl ReplicaValidator for Validator {
fn is_valid(&self, _replica_group: PublicKey) -> bool {
true
}
}
#[test]
fn creates_actor() -> Result<()> {
let (_actor, _sk_set) = get_actor_and_replicas_sk_set(10)?;
Ok(())
}
#[test]
fn initial_state_is_applied() -> Result<()> {
let initial_amount = 10;
let (actor, _sk_set) = get_actor_and_replicas_sk_set(initial_amount)?;
assert_eq!(actor.balance(), Token::from_nano(initial_amount));
Ok(())
}
#[test]
fn initiates_transfers() -> Result<()> {
let (actor, _sk_set) = get_actor_and_replicas_sk_set(10)?;
let debit = get_debit(&actor)?;
let mut actor = actor;
actor.apply(ActorEvent::TransferInitiated(debit))?;
Ok(())
}
#[test]
fn cannot_initiate_0_value_transfers() -> anyhow::Result<()> {
let (actor, _sk_set) = get_actor_and_replicas_sk_set(10)?;
match actor.transfer(Token::from_nano(0), get_random_pk(), "asfd".to_string()) {
Ok(_) => Err(anyhow::anyhow!(
"Should not be able to send 0 value transfers",
)),
Err(error) => {
assert!(error
.to_string()
.contains("Transfer amount must be greater than zero"));
Ok(())
}
}
}
#[test]
fn can_apply_completed_transfer() -> Result<()> {
let (actor, sk_set) = get_actor_and_replicas_sk_set(15)?;
let debit = get_debit(&actor)?;
let mut actor = actor;
actor.apply(ActorEvent::TransferInitiated(debit.clone()))?;
let transfer_event = get_transfer_registration_sent(debit, &sk_set)?;
actor.apply(ActorEvent::TransferRegistrationSent(transfer_event))?;
assert_eq!(Token::from_nano(5), actor.balance());
Ok(())
}
#[test]
fn can_apply_completed_transfers_in_succession() -> Result<()> {
let (actor, sk_set) = get_actor_and_replicas_sk_set(22)?;
let debit = get_debit(&actor)?;
let mut actor = actor;
actor.apply(ActorEvent::TransferInitiated(debit.clone()))?;
let transfer_event = get_transfer_registration_sent(debit, &sk_set)?;
actor.apply(ActorEvent::TransferRegistrationSent(transfer_event))?;
assert_eq!(Token::from_nano(12), actor.balance());
let debit2 = get_debit(&actor)?;
actor.apply(ActorEvent::TransferInitiated(debit2.clone()))?;
let transfer_event = get_transfer_registration_sent(debit2, &sk_set)?;
actor.apply(ActorEvent::TransferRegistrationSent(transfer_event))?;
assert_eq!(Token::from_nano(2), actor.balance());
Ok(())
}
#[allow(clippy::needless_range_loop)]
#[test]
fn can_return_proof_for_validated_transfers() -> Result<()> {
let (actor, sk_set) = get_actor_and_replicas_sk_set(22)?;
let debit = get_debit(&actor)?;
let mut actor = actor;
actor.apply(ActorEvent::TransferInitiated(debit.clone()))?;
let validations = get_transfer_validation_vec(debit, &sk_set)?;
for i in 0..7 {
let transfer_validation = actor
.receive(validations[i].clone())?
.ok_or(Error::UnexpectedOutcome)?;
if i < 1
{
assert_eq!(transfer_validation.clone().proof, None);
} else {
assert_ne!(transfer_validation.proof, None);
}
actor.apply(ActorEvent::TransferValidationReceived(
transfer_validation.clone(),
))?;
}
Ok(())
}
fn get_debit(actor: &Actor<Validator, Keypair>) -> Result<TransferInitiated> {
let event = actor
.transfer(Token::from_nano(10), get_random_pk(), "asdf".to_string())?
.ok_or(Error::UnexpectedOutcome)?;
Ok(event)
}
fn try_serialize<T: Serialize>(value: T) -> Result<Vec<u8>> {
match bincode::serialize(&value) {
Ok(res) => Ok(res),
_ => Err(Error::Serialisation("Serialisation error".to_string())),
}
}
fn get_transfer_validation_vec(
transfer: TransferInitiated,
sk_set: &SecretKeySet,
) -> Result<Vec<TransferValidated>> {
let signed_debit = transfer.signed_debit;
let signed_credit = transfer.signed_credit;
let serialized_signed_debit = try_serialize(&signed_debit)?;
let serialized_signed_credit = try_serialize(&signed_credit)?;
let sk_shares: Vec<_> = (0..7).map(|i| sk_set.secret_key_share(i)).collect();
let pk_set = sk_set.public_keys();
let debit_sig_shares: BTreeMap<_, _> = (0..7)
.map(|i| (i, sk_shares[i].sign(serialized_signed_debit.clone())))
.collect();
let credit_sig_shares: BTreeMap<_, _> = (0..7)
.map(|i| (i, sk_shares[i].sign(serialized_signed_credit.clone())))
.collect();
let mut validated_transfers = vec![];
for i in 0..7 {
let debit_sig_share = &debit_sig_shares[&i];
let credit_sig_share = &credit_sig_shares[&i];
assert!(pk_set
.public_key_share(i)
.verify(debit_sig_share, serialized_signed_debit.clone()));
assert!(pk_set
.public_key_share(i)
.verify(credit_sig_share, serialized_signed_credit.clone()));
validated_transfers.push(TransferValidated {
signed_debit: signed_debit.clone(),
signed_credit: signed_credit.clone(),
replica_debit_sig: SignatureShare {
index: i,
share: debit_sig_share.clone(),
},
replica_credit_sig: SignatureShare {
index: i,
share: credit_sig_share.clone(),
},
replicas: pk_set.clone(),
})
}
Ok(validated_transfers)
}
fn get_transfer_registration_sent(
transfer: TransferInitiated,
sk_set: &SecretKeySet,
) -> Result<TransferRegistrationSent> {
let signed_debit = transfer.signed_debit;
let signed_credit = transfer.signed_credit;
let serialized_signed_debit = try_serialize(&signed_debit)?;
let serialized_signed_credit = try_serialize(&signed_credit)?;
let sk_shares: Vec<_> = (0..7).map(|i| sk_set.secret_key_share(i)).collect();
let pk_set = sk_set.public_keys();
let debit_sig_shares: BTreeMap<_, _> = (0..7)
.map(|i| (i, sk_shares[i].sign(serialized_signed_debit.clone())))
.collect();
let credit_sig_shares: BTreeMap<_, _> = (0..7)
.map(|i| (i, sk_shares[i].sign(serialized_signed_credit.clone())))
.collect();
let debit_sig = match pk_set.combine_signatures(&debit_sig_shares) {
Ok(s) => s,
_ => return Err(Error::InvalidSignature),
};
let credit_sig = match pk_set.combine_signatures(&credit_sig_shares) {
Ok(s) => s,
_ => return Err(Error::InvalidSignature),
};
assert!(pk_set
.public_key()
.verify(&debit_sig, serialized_signed_debit));
assert!(pk_set
.public_key()
.verify(&credit_sig, serialized_signed_credit));
let debit_sig = Signature::Bls(debit_sig);
let credit_sig = Signature::Bls(credit_sig);
let transfer_agreement_proof = TransferAgreementProof {
signed_debit,
signed_credit,
debit_sig,
credit_sig,
debiting_replicas_keys: pk_set,
};
Ok(TransferRegistrationSent {
transfer_proof: transfer_agreement_proof,
})
}
fn get_actor_and_replicas_sk_set(
amount: u64,
) -> Result<(Actor<Validator, Keypair>, SecretKeySet)> {
let mut rng = rand::thread_rng();
let keypair = Keypair::new_ed25519(&mut rng);
let client_pubkey = keypair.public_key();
let bls_secret_key = SecretKeySet::random(1, &mut rng);
let replicas_id = bls_secret_key.public_keys();
let balance = Token::from_nano(amount);
let sender = Dot::new(get_random_pk(), 0);
let credit = get_credit(sender, client_pubkey, balance)?;
let replica_validator = Validator {};
let mut wallet = Wallet::new(OwnerType::Single(credit.recipient()));
wallet.apply_credit(credit)?;
let actor = Actor::from_snapshot(wallet, keypair, replicas_id, replica_validator);
Ok((actor, bls_secret_key))
}
fn get_credit(from: Dot<PublicKey>, recipient: PublicKey, amount: Token) -> Result<Credit> {
let debit = Debit { id: from, amount };
Ok(Credit {
id: debit.credit_id()?,
recipient,
amount,
msg: "asdf".to_string(),
})
}
#[allow(unused)]
fn get_random_dot() -> Dot<PublicKey> {
Dot::new(get_random_pk(), 0)
}
fn get_random_pk() -> PublicKey {
PublicKey::from(SecretKey::random().public_key())
}
}