//! Payloads exchanged during the post-handshake state.
use std::collections::HashMap;
use educe::Educe;
use libthreema_macros::{DebugVariantNames, Name, VariantNames};
use crate::{
common::D2xDeviceId,
d2m::{
D2mProtocolError, D2mProtocolInternalEncodingErrorCause, D2mProtocolInternalErrorCause,
payload::{PayloadDecoder, PayloadEncoder, ReflectFlags},
},
protobuf,
utils::{
bytes::{ByteReader as _, ByteWriter, OwnedVecByteReader},
debug::{Name as _, debug_slice_length},
time::Duration,
},
};
#[derive(Clone, Copy, strum::FromRepr, VariantNames, DebugVariantNames)]
#[repr(u8)]
enum IncomingPayloadType {
// Chat server proxying
Proxy = 0x00,
// States
ReflectionQueueDry = 0x20,
RolePromotedToLeader = 0x21,
// Device management
DevicesInfo = 0x31,
DropDeviceAck = 0x33,
// Transactions
BeginTransactionAck = 0x41,
CommitTransactionAck = 0x43,
TransactionRejected = 0x44,
TransactionEnded = 0x45,
// Reflection
ReflectAck = 0x81,
Reflected = 0x82,
}
/// Acknowledges that a message to be reflected to all other devices has been stored in their respective
/// reflection queues.
#[derive(Debug, Name)]
pub struct ReflectAck {
/// Refers to the `Reflect ID` as sent in the `Reflect` message.
pub reflect_id: u32,
/// Unix-ish timestamp in milliseconds when the message has been stored in the reflection queue of the
/// mediator server.
pub timestamp: u64,
}
impl ReflectAck {
fn decode(mut reader: OwnedVecByteReader) -> Result<Self, D2mProtocolError> {
reader
.run(|reader| {
// Skip reserved bytes
reader.skip(4)?;
// Decode reflect-id and timestamp
let reflect_id = reader.read_u32_le()?;
let timestamp = reader.read_u64_le()?;
// Note: Intentionally **not** calling `reader.expect_consumed()` since we may extend the
// struct in a backwards-compatible way.
Ok(Self {
reflect_id,
timestamp,
})
})
.map_err(|error| D2mProtocolError::DecodingFailed {
name: Self::NAME,
source: error.into(),
})
}
}
/// A message from the device's reflection queue.
#[derive(Name, Educe)]
#[educe(Debug)]
pub struct Reflected {
/// Flags set for the message. See [`ReflectFlags`] for known values.
pub flags: ReflectFlags,
/// Monotonically increasing unique number (per device slot) used for acknowledgement. May wrap.
pub reflect_id: u32,
/// Unix-ish timestamp in milliseconds when the message has been stored in the reflection queue of the
/// mediator server.
pub timestamp: u64,
/// The protobuf-encoded and encrypted data to be reflected, encrypted by `DGRK.secret` and prefixed with
/// a random nonce.
#[educe(Debug(method(debug_slice_length)))]
pub envelope: Vec<u8>,
}
impl Reflected {
const MIN_HEADER_LENGTH: u8 = 16;
fn decode(mut reader: OwnedVecByteReader) -> Result<Self, D2mProtocolError> {
// Decode header length
//
// Note: Checking the minimum header length is necessary as we might otherwise read garbage header
// data from the envelope that's behind it.
let header_length = reader
.read_u8()
.map_err(|error| D2mProtocolError::DecodingFailed {
name: Self::NAME,
source: error.into(),
})?;
if header_length < Self::MIN_HEADER_LENGTH {
return Err(D2mProtocolError::InvalidMessage {
name: Self::NAME,
cause: format!(
"Header needs to be at least {min_length} bytes but got {header_length}",
min_length = Self::MIN_HEADER_LENGTH
),
});
}
// Decode known header fields
let (flags, reflect_id, timestamp) = reader
.run_at(0, |mut reader| {
// Skip reserved byte
reader.skip(1)?;
// Decode flags, reflected-id and timestamp
let flags = reader.read_u16_le()?;
let reflect_id = reader.read_u32_le()?;
let timestamp = reader.read_u64_le()?;
Ok((flags, reflect_id, timestamp))
})
.map_err(|error| D2mProtocolError::DecodingFailed {
name: Self::NAME,
source: error.into(),
})?;
// Move beyond the header and extract the envelope
let envelope = reader
.run_owned(|mut reader| {
reader.skip(header_length as usize)?;
Ok(reader.read_remaining_owned())
})
.map_err(|error| D2mProtocolError::DecodingFailed {
name: Self::NAME,
source: error.into(),
})?;
Ok(Self {
reflect_id,
timestamp,
flags: ReflectFlags(flags),
envelope,
})
}
}
/// Augmented device information.
#[derive(Educe)]
#[educe(Debug)]
pub struct AugmentedDeviceInfo {
/// Device info (`d2d.DeviceInfo`), encrypted by `DGDIK.secret` and prefixed with a random nonce.
#[educe(Debug(method(debug_slice_length)))]
pub encrypted_device_info: Vec<u8>,
/// Expiration policy of the device.
pub device_slot_expiration_policy: Option<protobuf::d2m::DeviceSlotExpirationPolicy>,
/// Connection state of the device.
pub connection_state: Option<protobuf::d2m::devices_info::augmented_device_info::ConnectionState>,
}
impl From<protobuf::d2m::devices_info::AugmentedDeviceInfo> for AugmentedDeviceInfo {
fn from(
protobuf::d2m::devices_info::AugmentedDeviceInfo {
encrypted_device_info,
device_slot_expiration_policy,
connection_state,
}: protobuf::d2m::devices_info::AugmentedDeviceInfo,
) -> Self {
Self {
encrypted_device_info,
device_slot_expiration_policy: protobuf::d2m::DeviceSlotExpirationPolicy::try_from(
device_slot_expiration_policy,
)
.ok(),
connection_state,
}
}
}
/// Device information of all devices.
#[derive(Debug)]
pub struct DevicesInfo(pub HashMap<D2xDeviceId, AugmentedDeviceInfo>);
impl From<protobuf::d2m::DevicesInfo> for DevicesInfo {
fn from(
protobuf::d2m::DevicesInfo {
augmented_device_info,
}: protobuf::d2m::DevicesInfo,
) -> Self {
Self(
augmented_device_info
.into_iter()
.map(|(id, info)| (D2xDeviceId(id), AugmentedDeviceInfo::from(info)))
.collect(),
)
}
}
/// A begin transaction request has been rejected because another transaction is already in process.
#[derive(Educe)]
#[educe(Debug)]
pub struct TransactionRejected {
/// The device that currently holds the lock.
pub device_id: D2xDeviceId,
/// The encrypted transaction scope (`d2d.TransactionScope`) associated with the currently locked
/// transaction, encrypted by `DGTSK.secret` and prefixed with a random nonce.
#[educe(Debug(method(debug_slice_length)))]
pub encrypted_scope: Vec<u8>,
}
impl From<protobuf::d2m::TransactionRejected> for TransactionRejected {
fn from(
protobuf::d2m::TransactionRejected {
device_id,
encrypted_scope,
}: protobuf::d2m::TransactionRejected,
) -> Self {
Self {
device_id: D2xDeviceId(device_id),
encrypted_scope,
}
}
}
/// When a transaction ends (either because it was committed or because the device disconnected), this message
/// is sent to all connected devices except for the device that committed the transaction.
///
/// This can be used by the other devices as a _retry signal_ if a previous `BeginTransaction` attempt was
/// unsuccessful.
#[derive(Educe)]
#[educe(Debug)]
pub struct TransactionEnded {
/// The device that held the lock up until now.
pub device_id: D2xDeviceId,
/// The encrypted transaction scope (`d2d.TransactionScope`) associated with the transaction that just
/// ended, encrypted by `DGTSK.secret` and prefixed with a random nonce.
#[educe(Debug(method(debug_slice_length)))]
pub encrypted_scope: Vec<u8>,
}
impl From<protobuf::d2m::TransactionEnded> for TransactionEnded {
fn from(
protobuf::d2m::TransactionEnded {
device_id,
encrypted_scope,
}: protobuf::d2m::TransactionEnded,
) -> Self {
Self {
device_id: D2xDeviceId(device_id),
encrypted_scope,
}
}
}
/// An incoming payload received from the mediator server during the post-handshake phase.
#[derive(Debug, Name, VariantNames)]
pub enum IncomingPayload {
/// Proxied message from the chat server.
Proxy(Vec<u8>),
/// Indicates that the device's reflection queue on the server has been fully transmitted to the device.
ReflectionQueueDry,
/// The device's role has been promoted to leader, indicating that the device should now request to
/// receive and reflect messages from the chat server.
RolePromotedToLeader,
/// Device information of all devices.
DevicesInfo(DevicesInfo),
/// Acknowledges that a device has been dropped and the device slot has been free'd.
DropDeviceAck(D2xDeviceId),
/// Acknowledges that the device group lock has been acquired and that the transaction has been started.
BeginTransactionAck,
/// Commits a transaction, releases a device group lock.
CommitTransactionAck,
/// A begin transaction request has been rejected because another transaction is already in process.
TransactionRejected(TransactionRejected),
/// Another device just ended a transaction, releasing the device group lock.
TransactionEnded(TransactionEnded),
/// Acknowledges that a message to be reflected to all other devices has been stored in their respective
/// reflection queues.
ReflectAck(ReflectAck),
/// A message from the device's reflection queue.
Reflected(Reflected),
/// The payload type is not known, either due to server misbehavior or as consequence of running an old
/// version of libthreema. This information might be helpful for debugging.
UnknownPayload {
/// The (unsupported) type of the payload.
payload_type: u8,
/// The length of the payload.
length: usize,
},
}
impl PayloadDecoder for IncomingPayload {
fn decode(r#type: u8, reader: OwnedVecByteReader) -> Result<Self, D2mProtocolError> {
// Parse payload type
let Some(r#type) = IncomingPayloadType::from_repr(r#type) else {
return Ok(Self::UnknownPayload {
payload_type: r#type,
length: reader.remaining(),
});
};
// Create the corresponding variant by decoding the payload accordingly
Ok(match r#type {
IncomingPayloadType::Proxy => Self::Proxy(reader.read_remaining_owned()),
IncomingPayloadType::ReflectionQueueDry => {
let protobuf::d2m::ReflectionQueueDry {} =
Self::decode_protobuf::<protobuf::d2m::ReflectionQueueDry>(reader)?;
Self::ReflectionQueueDry
},
IncomingPayloadType::RolePromotedToLeader => {
let protobuf::d2m::RolePromotedToLeader {} =
Self::decode_protobuf::<protobuf::d2m::RolePromotedToLeader>(reader)?;
Self::RolePromotedToLeader
},
IncomingPayloadType::DevicesInfo => {
Self::DevicesInfo(DevicesInfo::from(Self::decode_protobuf::<
protobuf::d2m::DevicesInfo,
>(reader)?))
},
IncomingPayloadType::DropDeviceAck => {
let protobuf::d2m::DropDeviceAck { device_id } =
Self::decode_protobuf::<protobuf::d2m::DropDeviceAck>(reader)?;
Self::DropDeviceAck(D2xDeviceId(device_id))
},
IncomingPayloadType::BeginTransactionAck => {
let protobuf::d2m::BeginTransactionAck {} =
Self::decode_protobuf::<protobuf::d2m::BeginTransactionAck>(reader)?;
Self::BeginTransactionAck
},
IncomingPayloadType::CommitTransactionAck => {
let protobuf::d2m::CommitTransactionAck {} =
Self::decode_protobuf::<protobuf::d2m::CommitTransactionAck>(reader)?;
Self::CommitTransactionAck
},
IncomingPayloadType::TransactionRejected => {
Self::TransactionRejected(TransactionRejected::from(Self::decode_protobuf::<
protobuf::d2m::TransactionRejected,
>(reader)?))
},
IncomingPayloadType::TransactionEnded => {
Self::TransactionEnded(TransactionEnded::from(Self::decode_protobuf::<
protobuf::d2m::TransactionEnded,
>(reader)?))
},
IncomingPayloadType::ReflectAck => Self::ReflectAck(ReflectAck::decode(reader)?),
IncomingPayloadType::Reflected => Self::Reflected(Reflected::decode(reader)?),
})
}
}
#[derive(Clone, Copy, strum::FromRepr, VariantNames, DebugVariantNames)]
#[repr(u8)]
enum OutgoingPayloadType {
// Chat server proxying
Proxy = 0x00,
// Device management
GetDevicesInfo = 0x30,
DropDevice = 0x32,
SetSharedDeviceData = 0x34,
// Transactions
BeginTransaction = 0x40,
CommitTransaction = 0x42,
// Reflection
Reflect = 0x80,
ReflectedAck = 0x83,
}
/// Acquires a device group lock for an atomic operation shared across the device group.
///
/// See [`protobuf::d2m::BeginTransaction`] for further details.
#[derive(Educe)]
#[educe(Debug)]
pub struct BeginTransaction {
/// The transaction scope (`d2d.TransactionScope`), encrypted by `DGTSK.secret` and prefixed with a
/// random nonce.
#[educe(Debug(method(debug_slice_length)))]
pub encrypted_scope: Vec<u8>,
/// Time-to-live for this transaction. Once the TTL is reached, the mediator server will abort the
/// transaction and disconnect the client. When not provided, the server's maximum transaction TTL will
/// be used.
pub ttl: Option<Duration>,
}
impl TryFrom<BeginTransaction> for protobuf::d2m::BeginTransaction {
type Error = D2mProtocolError;
fn try_from(BeginTransaction { encrypted_scope, ttl }: BeginTransaction) -> Result<Self, Self::Error> {
Ok(Self {
encrypted_scope,
ttl: ttl.unwrap_or_default().as_secs().try_into().map_err(|_| {
D2mProtocolError::ConversionError("Transaction TTL in seconds must fit a u32")
})?,
})
}
}
/// Set the shared device data which is being sent to each device during login.
#[derive(Educe)]
#[educe(Debug)]
pub struct SharedDeviceData {
/// Device data shared among devices (`d2d.SharedDeviceData`), encrypted by `DGSDDK.secret` and prefixed
/// with a random nonce.
#[educe(Debug(method(debug_slice_length)))]
pub encrypted_shared_device_data: Vec<u8>,
}
impl From<SharedDeviceData> for protobuf::d2m::SetSharedDeviceData {
fn from(
SharedDeviceData {
encrypted_shared_device_data,
}: SharedDeviceData,
) -> Self {
Self {
encrypted_shared_device_data,
}
}
}
/// Reflect a message into the reflection queue of all other devices.
#[derive(Name, Educe)]
#[educe(Debug)]
pub struct Reflect {
/// [`ReflectFlags`] set for this reflected message.
pub flags: ReflectFlags,
/// Unique number (per connection), used for acknowledgement.
#[expect(clippy::struct_field_names, reason = "Consistency with protocol")]
pub reflect_id: u32,
/// The protobuf-encoded and encrypted data to be reflected, encrypted by `DGRK.secret` and prefixed with
/// a random nonce.
#[educe(Debug(method(debug_slice_length)))]
pub envelope: Vec<u8>,
}
impl Reflect {
const HEADER_LENGTH: u8 = 8;
fn length(&self) -> usize {
(Self::HEADER_LENGTH as usize).saturating_add(self.envelope.len())
}
fn encode_into(self, writer: &mut impl ByteWriter) -> Result<(), D2mProtocolError> {
writer
.run(|writer| {
// Write header-length, reserved, flags, reflect-id and the envelope
writer.write_u8(Self::HEADER_LENGTH)?;
writer.write_u8(0)?;
writer.write_u16_le(self.flags.0)?;
writer.write_u32_le(self.reflect_id)?;
writer.write(&self.envelope)
})
.map_err(|error| {
D2mProtocolError::InternalError(D2mProtocolInternalErrorCause::EncodingFailed {
name: Self::NAME,
source: D2mProtocolInternalEncodingErrorCause::ByteWriterError(error),
})
})
}
}
/// Acknowledges that a reflected message has been processed by the device.
#[derive(Name, Educe)]
#[educe(Debug)]
pub struct ReflectedAck {
/// Refers to the reflect id as sent in the `reflected` message.
pub reflect_id: u32,
}
impl ReflectedAck {
const LENGTH: usize = Self::RESERVED.len() + 4;
const RESERVED: [u8; 4] = [0_u8; 4];
fn encode_into(self, writer: &mut impl ByteWriter) -> Result<(), D2mProtocolError> {
writer
.run(|writer| {
// Write reserved bytes and reflect-id
writer.write(&Self::RESERVED)?;
writer.write_u32_le(self.reflect_id)
})
.map_err(|error| {
D2mProtocolError::InternalError(D2mProtocolInternalErrorCause::EncodingFailed {
name: Self::NAME,
source: D2mProtocolInternalEncodingErrorCause::ByteWriterError(error),
})
})
}
}
/// An outgoing payload to be sent to the mediator server during the post-handshake phase.
#[derive(Educe, Name, VariantNames)]
#[educe(Debug)]
pub enum OutgoingPayload {
/// Proxied message to the chat server.
Proxy(Vec<u8>),
/// Request device information of all devices.
GetDevicesInfo,
/// Request to drop a device and free its device slot.
DropDevice(D2xDeviceId),
/// Acquires a device group lock for an atomic operation shared across the device group.
BeginTransaction(BeginTransaction),
/// Set the shared device data which is being sent to each device during login.
SetSharedDeviceData(SharedDeviceData),
/// Commits a transaction, releases a device group lock.
CommitTransaction,
/// Reflect a message into the reflection queue of all other devices.
Reflect(Reflect),
/// Acknowledges that a reflected message has been processed by the device.
ReflectedAck(ReflectedAck),
}
impl PayloadEncoder for OutgoingPayload {
fn type_and_length(&self) -> (u8, Option<usize>) {
match self {
Self::Proxy(data) => (OutgoingPayloadType::Proxy as u8, Some(data.len())),
Self::GetDevicesInfo => (OutgoingPayloadType::GetDevicesInfo as u8, None),
Self::DropDevice(_) => (OutgoingPayloadType::DropDevice as u8, None),
Self::BeginTransaction(_) => (OutgoingPayloadType::BeginTransaction as u8, None),
Self::SetSharedDeviceData(_) => (OutgoingPayloadType::SetSharedDeviceData as u8, None),
Self::CommitTransaction => (OutgoingPayloadType::CommitTransaction as u8, None),
Self::Reflect(reflect) => (OutgoingPayloadType::Reflect as u8, Some(reflect.length())),
Self::ReflectedAck(_) => (
OutgoingPayloadType::ReflectedAck as u8,
Some(ReflectedAck::LENGTH),
),
}
}
fn encode_into<TWriter>(self, writer: &mut TWriter) -> Result<(), D2mProtocolError>
where
TWriter: ByteWriter,
{
let name = self.variant_name();
match self {
Self::Proxy(data) => writer.write(&data).map_err(|error| {
D2mProtocolError::InternalError(D2mProtocolInternalErrorCause::EncodingFailed {
name,
source: D2mProtocolInternalEncodingErrorCause::ByteWriterError(error),
})
}),
Self::GetDevicesInfo => Self::encode_protobuf(protobuf::d2m::GetDevicesInfo {}, writer),
Self::DropDevice(device_id) => Self::encode_protobuf(
protobuf::d2m::DropDevice {
device_id: device_id.0,
},
writer,
),
Self::BeginTransaction(begin_transaction) => Self::encode_protobuf(
protobuf::d2m::BeginTransaction::try_from(begin_transaction)?,
writer,
),
Self::SetSharedDeviceData(shared_device_data) => Self::encode_protobuf(
protobuf::d2m::SetSharedDeviceData::from(shared_device_data),
writer,
),
Self::CommitTransaction => Self::encode_protobuf(protobuf::d2m::CommitTransaction {}, writer),
Self::Reflect(reflect) => reflect.encode_into(writer),
Self::ReflectedAck(reflected_ack) => reflected_ack.encode_into(writer),
}
}
}
¤ Dauer der Verarbeitung: 0.7 Sekunden
¤
*© Formatika GbR, Deutschland
|
|