servius/songbird
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: v8 encryption modes (#264)

Browse Source 
This PR adds support for the new AEAD cryptosystems advertised by Discord, AES256-GCM and XChaCha20Poly1305. These schemes will shortly become mandatory, and provider stronger integrity/authentication guarantees over the cleartext portions of any voice packet by correctly specifying additional authenticated data.

To provide smooth switchover, we've added basic negotiation over the `CryptoMode`. This ensures that any clients who are manually specifying one of the legacy modes will automatically migrate to `Aes256Gcm` when Discord cease to advertise their original preference.

Closes #246.

---------

Co-authored-by: Kyle Simpson <kyleandrew.simpson@gmail.com>
This commit is contained in:
tig
2024-11-11 21:30:15 +09:00
committed by GitHub
parent fe9b156906
commit 10ce458456
17 changed files with 705 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/config.rs
View File

@@ -34,13 +34,13 @@ pub struct Config {
#[cfg(feature = "driver")]
/// Selected tagging mode for voice packet encryption.
///
/// Defaults to [`CryptoMode::Normal`].
/// Defaults to [`CryptoMode::Aes256Gcm`].
///
/// Changes to this field will not immediately apply if the
/// driver is actively connected, but will apply to subsequent
/// sessions.
///
/// [`CryptoMode::Normal`]: CryptoMode::Normal
/// [`CryptoMode::Aes256Gcm`]: CryptoMode::Aes256Gcm
pub crypto_mode: CryptoMode,
#[cfg(all(feature = "driver", feature = "receive"))]
@@ -211,7 +211,7 @@ impl Default for Config {
fn default() -> Self {
Self {
#[cfg(feature = "driver")]
crypto_mode: CryptoMode::Normal,
crypto_mode: CryptoMode::Aes256Gcm,
#[cfg(all(feature = "driver", feature = "receive"))]
decode_mode: DecodeMode::Decrypt,
#[cfg(all(feature = "driver", feature = "receive"))]

2
src/constants.rs
View File

@@ -81,7 +81,7 @@ pub const RTP_VERSION: u8 = 2;
pub const RTP_PROFILE_TYPE: RtpType = RtpType::Dynamic(120);
#[cfg(test)]
#[allow(clippy::doc_markdown)]
#[allow(clippy::doc_markdown, missing_docs)]
pub mod test_data {
/// URL for a source which YTDL must extract.
///

2
src/driver/bench_internals.rs
View File

@@ -7,7 +7,7 @@
pub use super::tasks::{message as task_message, mixer};
pub use super::crypto::CryptoState;
pub use super::crypto::{Cipher, CryptoState};
use crate::{
driver::tasks::message::TrackContext,

39
src/driver/connection/mod.rs
View File

@@ -3,6 +3,7 @@ pub mod error;
#[cfg(feature = "receive")]
use super::tasks::udp_rx;
use super::{
crypto::Cipher,
tasks::{
message::*,
ws::{self as ws_task, AuxNetwork},
@@ -20,7 +21,6 @@ use crate::{
ws::WsStream,
ConnectionInfo,
};
use crypto_secretbox::{KeyInit, XSalsa20Poly1305 as Cipher};
use discortp::discord::{IpDiscoveryPacket, IpDiscoveryType, MutableIpDiscoveryPacket};
use error::{Error, Result};
use flume::Sender;
@@ -103,9 +103,12 @@ impl Connection {
let ready =
ready.expect("Ready packet expected in connection initialisation, but not found.");
if !has_valid_mode(&ready.modes, config.crypto_mode) {
return Err(Error::CryptoModeUnavailable);
}
let chosen_crypto = CryptoMode::negotiate(&ready.modes, Some(config.crypto_mode))?;
info!(
"Crypto scheme negotiation -- wanted {:?}. Chose {:?} from modes {:?}.",
config.crypto_mode, chosen_crypto, ready.modes
);
let udp = UdpSocket::bind("0.0.0.0:0").await?;
@@ -115,7 +118,7 @@ impl Connection {
} else {
let socket = Socket::from(udp.into_std()?);
// Some operating systems does not allow setting the recv buffer to 0.
// Some operating systems do not allow setting the recv buffer to 0.
#[cfg(any(target_os = "linux", target_os = "windows"))]
socket.set_recv_buffer_size(0)?;
@@ -159,28 +162,25 @@ impl Connection {
let address_str = std::str::from_utf8(&view.get_address_raw()[..nul_byte_index])
.map_err(|_| Error::IllegalIp)?;
let address = IpAddr::from_str(address_str).map_err(|e| {
println!("{e:?}");
Error::IllegalIp
})?;
let address = IpAddr::from_str(address_str).map_err(|_| Error::IllegalIp)?;
client
.send_json(&GatewayEvent::from(SelectProtocol {
protocol: "udp".into(),
data: ProtocolData {
address,
mode: config.crypto_mode.to_request_str().into(),
mode: chosen_crypto.to_request_str().into(),
port: view.get_port(),
},
}))
.await?;
}
let cipher = init_cipher(&mut client, config.crypto_mode).await?;
let cipher = init_cipher(&mut client, chosen_crypto).await?;
info!("Connected to: {}", info.endpoint);
info!("WS heartbeat duration {}ms.", hello.heartbeat_interval,);
info!("WS heartbeat duration {}ms.", hello.heartbeat_interval);
let (ws_msg_tx, ws_msg_rx) = flume::unbounded();
#[cfg(feature = "receive")]
@@ -209,7 +209,7 @@ impl Connection {
cipher: cipher.clone(),
#[cfg(not(feature = "receive"))]
cipher,
crypto_state: config.crypto_mode.into(),
crypto_state: chosen_crypto.into(),
#[cfg(feature = "receive")]
udp_rx: udp_receiver_msg_tx,
udp_tx,
@@ -244,6 +244,7 @@ impl Connection {
interconnect.clone(),
udp_receiver_msg_rx,
cipher,
chosen_crypto,
config.clone(),
udp_rx,
ssrc_tracker,
@@ -349,7 +350,8 @@ async fn init_cipher(client: &mut WsStream, mode: CryptoMode) -> Result<Cipher>
return Err(Error::CryptoModeInvalid);
}
return Cipher::new_from_slice(&desc.secret_key)
return mode
.cipher_from_key(&desc.secret_key)
.map_err(|_| Error::CryptoInvalidLength);
},
other => {
@@ -362,12 +364,3 @@ async fn init_cipher(client: &mut WsStream, mode: CryptoMode) -> Result<Cipher>
}
}
}
#[inline]
fn has_valid_mode<T, It>(modes: It, mode: CryptoMode) -> bool
where
T: for<'a> PartialEq<&'a str>,
It: IntoIterator<Item = T>,
{
modes.into_iter().any(|s| s == mode.to_request_str())
}

704
src/driver/crypto.rs
View File

@@ -1,36 +1,76 @@
//! Encryption schemes supported by Discord's secure RTP negotiation.
use super::tasks::error::Error as InternalError;
use aead::AeadCore;
use aes_gcm::{AeadInPlace, Aes256Gcm, KeyInit};
use byteorder::{NetworkEndian, WriteBytesExt};
#[cfg(any(feature = "receive", test))]
use crypto_secretbox::Tag;
use crypto_secretbox::{
aead::{AeadInPlace, Error as CryptoError},
Nonce,
SecretBox,
XSalsa20Poly1305 as Cipher,
};
use chacha20poly1305::XChaCha20Poly1305;
use crypto_secretbox::{cipher::InvalidLength, Error as CryptoError, XSalsa20Poly1305};
#[cfg(feature = "receive")]
use discortp::rtcp::MutableRtcpPacket;
use discortp::{rtp::RtpPacket, MutablePacket};
#[cfg(any(feature = "receive", test))]
use discortp::{
rtp::{MutableRtpPacket, RtpExtensionPacket},
Packet,
};
use rand::Rng;
use std::num::Wrapping;
use std::{num::Wrapping, str::FromStr};
use typenum::Unsigned;
#[cfg(test)]
pub const KEY_SIZE: usize = SecretBox::<()>::KEY_SIZE;
pub const NONCE_SIZE: usize = SecretBox::<()>::NONCE_SIZE;
pub const TAG_SIZE: usize = SecretBox::<()>::TAG_SIZE;
use crate::error::ConnectionError;
/// Variants of the `XSalsa20Poly1305` encryption scheme.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[derive(Clone, Copy, Debug, Eq, PartialEq, Default, Hash)]
#[non_exhaustive]
pub enum CryptoMode {
#[default]
/// Discord's currently preferred non-E2EE encryption scheme.
///
/// Packets are encrypted and decrypted using the `AES256GCM` encryption scheme.
/// An additional random 4B suffix is used as the source of nonce bytes for the packet.
/// This nonce value increments by `1` with each packet.
///
/// Encrypted content begins *after* the RTP header and extensions, following the SRTP
/// specification.
///
/// Nonce width of 4B (32b), at an extra 4B per packet (~0.2 kB/s).
Aes256Gcm,
/// A fallback non-E2EE encryption scheme.
///
/// Packets are encrypted and decrypted using the `XChaCha20Poly1305` encryption scheme.
/// An additional random 4B suffix is used as the source of nonce bytes for the packet.
/// This nonce value increments by `1` with each packet.
///
/// Encrypted content begins *after* the RTP header and extensions, following the SRTP
/// specification.
///
/// Nonce width of 4B (32b), at an extra 4B per packet (~0.2 kB/s).
XChaCha20Poly1305,
#[deprecated(
since = "0.4.4",
note = "This voice encryption mode will no longer be accepted by Discord\
as of 2024-11-18. This variant will be removed in `v0.5`."
)]
/// The RTP header is used as the source of nonce bytes for the packet.
///
/// Equivalent to a nonce of at most 48b (6B) at no extra packet overhead:
/// the RTP sequence number and timestamp are the varying quantities.
Normal,
#[deprecated(
since = "0.4.4",
note = "This voice encryption mode will no longer be accepted by Discord\
as of 2024-11-18. This variant will be removed in `v0.5`."
)]
/// An additional random 24B suffix is used as the source of nonce bytes for the packet.
/// This is regenerated randomly for each packet.
///
/// Full nonce width of 24B (192b), at an extra 24B per packet (~1.2 kB/s).
Suffix,
#[deprecated(
since = "0.4.4",
note = "This voice encryption mode will no longer be accepted by Discord\
as of 2024-11-18. This variant will be removed in `v0.5`."
)]
/// An additional random 4B suffix is used as the source of nonce bytes for the packet.
/// This nonce value increments by `1` with each packet.
///
@@ -38,60 +78,229 @@ pub enum CryptoMode {
Lite,
}
#[allow(deprecated)]
impl From<CryptoState> for CryptoMode {
fn from(val: CryptoState) -> Self {
match val {
CryptoState::Normal => Self::Normal,
CryptoState::Suffix => Self::Suffix,
CryptoState::Lite(_) => Self::Lite,
CryptoState::Aes256Gcm(_) => Self::Aes256Gcm,
CryptoState::XChaCha20Poly1305(_) => Self::XChaCha20Poly1305,
}
}
}
/// The input string could not be parsed as an encryption scheme supported by songbird.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
pub struct UnrecognisedCryptoMode;
impl FromStr for CryptoMode {
type Err = UnrecognisedCryptoMode;
#[allow(deprecated)]
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"aead_aes256_gcm_rtpsize" => Ok(Self::Aes256Gcm),
"aead_xchacha20_poly1305_rtpsize" => Ok(Self::XChaCha20Poly1305),
"xsalsa20_poly1305" => Ok(Self::Normal),
"xsalsa20_poly1305_suffix" => Ok(Self::Suffix),
"xsalsa20_poly1305_lite" => Ok(Self::Lite),
_ => Err(UnrecognisedCryptoMode),
}
}
}
#[allow(deprecated)]
impl CryptoMode {
/// Returns the underlying crypto algorithm used by a given [`CryptoMode`].
#[must_use]
pub(crate) const fn algorithm(self) -> EncryptionAlgorithm {
match self {
CryptoMode::Aes256Gcm => EncryptionAlgorithm::Aes256Gcm,
CryptoMode::XChaCha20Poly1305 => EncryptionAlgorithm::XChaCha20Poly1305,
CryptoMode::Normal | CryptoMode::Suffix | CryptoMode::Lite =>
EncryptionAlgorithm::XSalsa20Poly1305,
}
}
/// Returns whether this [`CryptoMode`] dynamically sizes the ciphertext region
/// to begin in the middle of RTP extensions.
///
/// Compliant SRTP would leave all extensions in cleartext, hence 'more' SRTP
/// compliant.
#[must_use]
#[cfg(any(feature = "receive", test))]
pub(crate) const fn is_more_srtp_compliant(self) -> bool {
match self {
CryptoMode::Aes256Gcm | CryptoMode::XChaCha20Poly1305 => true,
CryptoMode::Normal | CryptoMode::Suffix | CryptoMode::Lite => false,
}
}
/// Returns an encryption cipher based on the supplied key.
///
/// Creation fails if the key is the incorrect length for the target cipher.
pub(crate) fn cipher_from_key(self, key: &[u8]) -> Result<Cipher, InvalidLength> {
match self.algorithm() {
EncryptionAlgorithm::Aes256Gcm => Aes256Gcm::new_from_slice(key)
.map(Box::new)
.map(Cipher::Aes256Gcm),
EncryptionAlgorithm::XChaCha20Poly1305 =>
XChaCha20Poly1305::new_from_slice(key).map(Cipher::XChaCha20Poly1305),
EncryptionAlgorithm::XSalsa20Poly1305 =>
XSalsa20Poly1305::new_from_slice(key).map(|v| Cipher::XSalsa20Poly1305(v, self)),
}
}
/// Returns a local priority score for a given [`CryptoMode`].
///
/// Higher values are preferred.
#[must_use]
pub(crate) fn priority(self) -> u64 {
match self {
CryptoMode::Aes256Gcm => 4,
CryptoMode::XChaCha20Poly1305 => 3,
CryptoMode::Normal => 2,
CryptoMode::Suffix => 1,
CryptoMode::Lite => 0,
}
}
/// Returns the best available crypto mode, given the `modes` offered by the Discord voice server.
///
/// If `preferred` is set and the mode exists in the server's supported algorithms, then that
/// mode will be chosen. Otherwise we select the highest-scoring option which is mutually understood.
pub(crate) fn negotiate<It, T>(
modes: It,
preferred: Option<Self>,
) -> Result<Self, ConnectionError>
where
T: AsRef<str>,
It: IntoIterator<Item = T>,
{
let mut best = None;
for el in modes {
let Ok(el) = CryptoMode::from_str(el.as_ref()) else {
// Unsupported mode. Ignore.
continue;
};
let mut el_priority = el.priority();
if let Some(preferred) = preferred {
if el == preferred {
el_priority = u64::MAX;
}
}
let accept = match best {
None => true,
Some((_, score)) if el_priority > score => true,
_ => false,
};
if accept {
best = Some((el, el_priority));
}
}
best.map(|(v, _)| v)
.ok_or(ConnectionError::CryptoModeUnavailable)
}
/// Returns the name of a mode as it will appear during negotiation.
#[must_use]
pub fn to_request_str(self) -> &'static str {
pub const fn to_request_str(self) -> &'static str {
match self {
Self::Normal => "xsalsa20_poly1305",
Self::Suffix => "xsalsa20_poly1305_suffix",
Self::Lite => "xsalsa20_poly1305_lite",
Self::Aes256Gcm => "aead_aes256_gcm_rtpsize",
Self::XChaCha20Poly1305 => "aead_xchacha20_poly1305_rtpsize",
}
}
/// Returns the nonce length in bytes required by algorithm.
#[must_use]
pub const fn algorithm_nonce_size(self) -> usize {
use typenum::Unsigned as _;
match self {
Self::Lite | Self::Normal | Self::Suffix => XSalsa20Poly1305::NONCE_SIZE,
Self::XChaCha20Poly1305 => <XChaCha20Poly1305 as AeadCore>::NonceSize::USIZE, // => 24
Self::Aes256Gcm => <Aes256Gcm as AeadCore>::NonceSize::USIZE, // => 12
}
}
/// Returns the number of bytes each nonce is stored as within
/// a packet.
#[must_use]
pub fn nonce_size(self) -> usize {
pub const fn nonce_size(self) -> usize {
match self {
Self::Aes256Gcm | Self::XChaCha20Poly1305 | Self::Lite => 4,
Self::Normal => RtpPacket::minimum_packet_size(),
Self::Suffix => NONCE_SIZE,
Self::Lite => 4,
Self::Suffix => XSalsa20Poly1305::NONCE_SIZE,
}
}
/// Returns the number of bytes occupied by the XSalsa20Poly1305
/// encryption schemes which fall before the payload.
#[must_use]
#[deprecated(
since = "0.4.4",
note = "This method returns the fixed payload prefix for older encryption modes,\
which will no longer be accepted by Discord as of 2024-11-18. It is an\
implementation detail and will be removed in `v0.5`."
)]
pub fn payload_prefix_len() -> usize {
XSalsa20Poly1305::TAG_SIZE
}
/// Returns the number of bytes occupied by the encryption scheme
/// which fall before the payload.
///
/// Method name duplicated until v0.5, to prevent breaking change.
#[must_use]
pub fn payload_prefix_len() -> usize {
TAG_SIZE
pub(crate) const fn payload_prefix_len2(self) -> usize {
match self {
CryptoMode::Aes256Gcm | CryptoMode::XChaCha20Poly1305 => 0,
CryptoMode::Normal | CryptoMode::Suffix | CryptoMode::Lite =>
XSalsa20Poly1305::TAG_SIZE,
}
}
/// Returns the tag length in bytes.
#[must_use]
pub(crate) const fn encryption_tag_len(self) -> usize {
self.algorithm().encryption_tag_len()
}
/// Returns the number of bytes occupied by the encryption scheme
/// which fall after the payload.
#[must_use]
pub fn payload_suffix_len(self) -> usize {
pub const fn payload_suffix_len(self) -> usize {
match self {
Self::Normal => 0,
Self::Suffix | Self::Lite => self.nonce_size(),
Self::Aes256Gcm | Self::XChaCha20Poly1305 =>
self.nonce_size() + self.encryption_tag_len(),
}
}
/// Returns the number of bytes occupied by an encryption scheme's tag which
/// fall *after* the payload.
#[must_use]
pub const fn tag_suffix_len(self) -> usize {
match self {
Self::Normal | Self::Suffix | Self::Lite => 0,
Self::Aes256Gcm | Self::XChaCha20Poly1305 => self.encryption_tag_len(),
}
}
/// Calculates the number of additional bytes required compared
/// to an unencrypted payload.
#[must_use]
pub fn payload_overhead(self) -> usize {
Self::payload_prefix_len() + self.payload_suffix_len()
pub const fn payload_overhead(self) -> usize {
self.payload_prefix_len2() + self.payload_suffix_len()
}
/// Extracts the byte slice in a packet used as the nonce, and the remaining mutable
@@ -103,97 +312,74 @@ impl CryptoMode {
) -> Result<(&'a [u8], &'a mut [u8]), CryptoError> {
match self {
Self::Normal => Ok((header, body)),
Self::Suffix | Self::Lite => {
Self::Suffix | Self::Lite | Self::Aes256Gcm | Self::XChaCha20Poly1305 => {
let len = body.len();
if len < self.payload_suffix_len() {
Err(CryptoError)
} else {
let (body_left, nonce_loc) = body.split_at_mut(len - self.payload_suffix_len());
Ok((&nonce_loc[..self.nonce_size()], body_left))
let (body_left, nonce_loc) = body.split_at_mut(len - self.nonce_size());
Ok((nonce_loc, body_left))
}
},
}
}
#[cfg(any(feature = "receive", test))]
/// Decrypts a Discord RT(C)P packet using the given key.
///
/// If successful, this returns the number of bytes to be ignored from the
/// start and end of the packet payload.
#[inline]
pub(crate) fn decrypt_in_place(
self,
packet: &mut impl MutablePacket,
cipher: &Cipher,
) -> Result<(usize, usize), CryptoError> {
// FIXME on next: packet encrypt/decrypt should use an internal error
// to denote "too small" vs. "opaque".
let header_len = packet.packet().len() - packet.payload().len();
let (header, body) = packet.packet_mut().split_at_mut(header_len);
let (slice_to_use, body_remaining) = self.nonce_slice(header, body)?;
let mut nonce = Nonce::default();
let nonce_slice = if slice_to_use.len() == NONCE_SIZE {
Nonce::from_slice(&slice_to_use[..NONCE_SIZE])
} else {
let max_bytes_avail = slice_to_use.len();
nonce[..self.nonce_size().min(max_bytes_avail)].copy_from_slice(slice_to_use);
&nonce
};
let body_start = Self::payload_prefix_len();
let body_tail = self.payload_suffix_len();
if body_start > body_remaining.len() {
return Err(CryptoError);
}
let (tag_bytes, data_bytes) = body_remaining.split_at_mut(body_start);
let tag = Tag::from_slice(tag_bytes);
cipher
.decrypt_in_place_detached(nonce_slice, b"", data_bytes, tag)
.map(|()| (body_start, body_tail))
}
/// Encrypts a Discord RT(C)P packet using the given key.
/// Encrypts a Discord RT(C)P packet using the given XSalsa20Poly1305 cipher.
///
/// Use of this requires that the input packet has had a nonce generated in the correct location,
/// and `payload_len` specifies the number of bytes after the header including this nonce.
#[deprecated(
since = "0.4.4",
note = "This method performs encryption for older encryption modes,\
which will no longer be accepted by Discord as of 2024-11-18. It is an\
implementation detail and will be removed in `v0.5`."
)]
#[inline]
pub fn encrypt_in_place(
self,
packet: &mut impl MutablePacket,
cipher: &Cipher,
cipher: &XSalsa20Poly1305,
payload_len: usize,
) -> Result<(), CryptoError> {
let header_len = packet.packet().len() - packet.payload().len();
let (header, body) = packet.packet_mut().split_at_mut(header_len);
let (slice_to_use, body_remaining) = self.nonce_slice(header, &mut body[..payload_len])?;
let mut nonce = Nonce::default();
let nonce_slice = if slice_to_use.len() == NONCE_SIZE {
Nonce::from_slice(&slice_to_use[..NONCE_SIZE])
let nonce_size = self.nonce_size();
let tag_size = self.encryption_tag_len();
let mut nonce = crypto_secretbox::Nonce::default();
let nonce_slice = if slice_to_use.len() == nonce_size {
crypto_secretbox::Nonce::from_slice(&slice_to_use[..nonce_size])
} else {
nonce[..self.nonce_size()].copy_from_slice(slice_to_use);
nonce[..slice_to_use.len()].copy_from_slice(slice_to_use);
&nonce
};
// body_remaining is now correctly truncated by this point.
// the true_payload to encrypt follows after the first TAG_LEN bytes.
let tag =
cipher.encrypt_in_place_detached(nonce_slice, b"", &mut body_remaining[TAG_SIZE..])?;
body_remaining[..TAG_SIZE].copy_from_slice(&tag[..]);
cipher.encrypt_in_place_detached(nonce_slice, b"", &mut body_remaining[tag_size..])?;
body_remaining[..tag_size].copy_from_slice(&tag[..]);
Ok(())
}
}
/// State used in nonce generation for the `XSalsa20Poly1305` encryption variants
/// in [`CryptoMode`].
/// State used in nonce generation for the encryption variants in [`CryptoMode`].
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
#[non_exhaustive]
pub enum CryptoState {
/// An additional random 4B suffix is used as the source of nonce bytes for the packet.
/// This nonce value increments by `1` with each packet.
///
/// The last used nonce is stored.
Aes256Gcm(Wrapping<u32>),
/// An additional random 4B suffix is used as the source of nonce bytes for the packet.
/// This nonce value increments by `1` with each packet.
///
/// The last used nonce is stored.
XChaCha20Poly1305(Wrapping<u32>),
/// The RTP header is used as the source of nonce bytes for the packet.
///
/// No state is required.
@@ -210,12 +396,16 @@ pub enum CryptoState {
Lite(Wrapping<u32>),
}
#[allow(deprecated)]
impl From<CryptoMode> for CryptoState {
fn from(val: CryptoMode) -> Self {
match val {
CryptoMode::Normal => CryptoState::Normal,
CryptoMode::Suffix => CryptoState::Suffix,
CryptoMode::Lite => CryptoState::Lite(Wrapping(rand::random::<u32>())),
CryptoMode::Aes256Gcm => CryptoState::Aes256Gcm(Wrapping(rand::random::<u32>())),
CryptoMode::XChaCha20Poly1305 =>
CryptoState::XChaCha20Poly1305(Wrapping(rand::random::<u32>())),
}
}
}
@@ -229,20 +419,23 @@ impl CryptoState {
) -> usize {
let mode = self.kind();
let endpoint = payload_end + mode.payload_suffix_len();
let startpoint = endpoint - mode.nonce_size();
match self {
Self::Suffix => {
rand::thread_rng().fill(&mut packet.payload_mut()[payload_end..endpoint]);
rand::thread_rng().fill(&mut packet.payload_mut()[startpoint..endpoint]);
},
Self::Lite(mut i) => {
(&mut packet.payload_mut()[payload_end..endpoint])
Self::Lite(ref mut i)
| Self::Aes256Gcm(ref mut i)
| Self::XChaCha20Poly1305(ref mut i) => {
(&mut packet.payload_mut()[startpoint..endpoint])
.write_u32::<NetworkEndian>(i.0)
.expect(
"Nonce size is guaranteed to be sufficient to write u32 for lite tagging.",
);
i += Wrapping(1);
*i += Wrapping(1);
},
_ => {},
Self::Normal => {},
}
endpoint
@@ -255,55 +448,372 @@ impl CryptoState {
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)]
pub(crate) enum EncryptionAlgorithm {
Aes256Gcm,
XChaCha20Poly1305,
XSalsa20Poly1305,
}
impl EncryptionAlgorithm {
#[must_use]
pub(crate) const fn encryption_tag_len(self) -> usize {
match self {
Self::Aes256Gcm => <Aes256Gcm as AeadCore>::TagSize::USIZE, // 16
Self::XChaCha20Poly1305 => <XChaCha20Poly1305 as AeadCore>::TagSize::USIZE, // 16
Self::XSalsa20Poly1305 => XSalsa20Poly1305::TAG_SIZE, // 16
}
}
}
impl From<&Cipher> for EncryptionAlgorithm {
fn from(value: &Cipher) -> Self {
match value {
Cipher::XSalsa20Poly1305(..) => EncryptionAlgorithm::XSalsa20Poly1305,
Cipher::XChaCha20Poly1305(_) => EncryptionAlgorithm::XChaCha20Poly1305,
Cipher::Aes256Gcm(_) => EncryptionAlgorithm::Aes256Gcm,
}
}
}
#[derive(Clone)]
pub enum Cipher {
XSalsa20Poly1305(XSalsa20Poly1305, CryptoMode),
XChaCha20Poly1305(XChaCha20Poly1305),
Aes256Gcm(Box<Aes256Gcm>),
}
impl Cipher {
#[must_use]
pub(crate) fn mode(&self) -> CryptoMode {
match self {
Cipher::XSalsa20Poly1305(_, mode) => *mode,
Cipher::XChaCha20Poly1305(_) => CryptoMode::XChaCha20Poly1305,
Cipher::Aes256Gcm(_) => CryptoMode::Aes256Gcm,
}
}
#[must_use]
pub(crate) fn encryption_tag_len(&self) -> usize {
EncryptionAlgorithm::from(self).encryption_tag_len()
}
/// Encrypts a Discord RT(C)P packet using the given key.
///
/// Use of this requires that the input packet has had a nonce generated in the correct location,
/// and `payload_len` specifies the number of bytes after the header including this nonce.
#[inline]
pub fn encrypt_pkt_in_place(
&self,
packet: &mut impl MutablePacket,
payload_len: usize,
) -> Result<(), CryptoError> {
let mode = self.mode();
let header_len = packet.packet().len() - packet.payload().len();
let (header, body) = packet.packet_mut().split_at_mut(header_len);
let (slice_to_use, body_remaining) = mode.nonce_slice(header, &mut body[..payload_len])?;
let tag_size = self.encryption_tag_len();
// body_remaining is now correctly truncated to exclude the nonce by this point.
// the true_payload to encrypt is within the buf[prefix:-suffix].
let (pre_payload, body_remaining) = body_remaining.split_at_mut(mode.payload_prefix_len2());
let (body, post_payload) =
body_remaining.split_at_mut(body_remaining.len() - mode.tag_suffix_len());
// All these Nonce types are distinct at the type level
// (96b for AES, 192b for XSalsa/XChaCha).
match self {
// Older modes place the tag before the payload and do not authenticate
// cleartext.
Self::XSalsa20Poly1305(secret_box, _) => {
let mut nonce = crypto_secretbox::Nonce::default();
nonce[..mode.nonce_size()].copy_from_slice(slice_to_use);
let tag = secret_box.encrypt_in_place_detached(&nonce, b"", body)?;
pre_payload[..tag_size].copy_from_slice(&tag[..]);
},
// The below variants follow part of the SRTP spec (RFC3711, sec 3.1)
// by requiring that we include the cleartext header portion as
// authenticated data.
Self::Aes256Gcm(aes_gcm) => {
let mut nonce = aes_gcm::Nonce::default();
nonce[..mode.nonce_size()].copy_from_slice(slice_to_use);
let tag = aes_gcm.encrypt_in_place_detached(&nonce, header, body)?;
post_payload[..tag_size].copy_from_slice(&tag[..]);
},
Self::XChaCha20Poly1305(cha_cha_poly1305) => {
let mut nonce = chacha20poly1305::XNonce::default();
nonce[..mode.nonce_size()].copy_from_slice(slice_to_use);
let tag = cha_cha_poly1305.encrypt_in_place_detached(&nonce, header, body)?;
post_payload[..tag_size].copy_from_slice(&tag[..]);
},
}
Ok(())
}
#[cfg(any(feature = "receive", test))]
pub(crate) fn decrypt_rtp_in_place(
&self,
packet: &mut MutableRtpPacket<'_>,
) -> Result<(usize, usize), InternalError> {
let mode = self.mode();
// An exciting difference from the SRTP spec: Discord begins encryption
// after the RTP extension *header*, encrypting the extensions themselves,
// whereas the spec leaves all extensions in the clear.
// This header is described as the 'extension preamble'.
let has_extension = packet.get_extension() != 0;
let plain_bytes = if mode.is_more_srtp_compliant() && has_extension {
// CSRCs and extension bytes will be in the plaintext segment.
// We will need these demarcated to select the right bytes to
// decrypt, and to use as auth data.
RtpExtensionPacket::minimum_packet_size()
} else {
0
};
let (mut start_estimate, end) = self.decrypt_pkt_in_place(packet, plain_bytes)?;
// Update the start estimate to account for bytes occupied by extension headers.
if has_extension {
let packet = packet.packet();
if let Some((_, exts_and_opus)) = split_at_checked(packet, start_estimate) {
let extension = RtpExtensionPacket::new(exts_and_opus)
.ok_or(InternalError::IllegalVoicePacket)?;
start_estimate += extension.packet().len() - extension.payload().len();
}
}
Ok((start_estimate, end))
}
#[cfg(feature = "receive")]
pub(crate) fn decrypt_rtcp_in_place(
&self,
packet: &mut MutableRtcpPacket<'_>,
) -> Result<(usize, usize), InternalError> {
// RTCP/SRTCP have identical handling -- no var-length elements
// are included as part of the plaintext.
self.decrypt_pkt_in_place(packet, 0)
}
/// Decrypts an arbitrary packet using the given key.
///
/// If successful, this returns the number of bytes to be ignored from the
/// start and end of the packet payload.
#[inline]
#[cfg(any(feature = "receive", test))]
pub(crate) fn decrypt_pkt_in_place(
&self,
packet: &mut impl MutablePacket,
n_plaintext_body_bytes: usize,
) -> Result<(usize, usize), InternalError> {
let mode = self.mode();
let header_len = packet.packet().len() - packet.payload().len();
let plaintext_end = header_len + n_plaintext_body_bytes;
let (plaintext, ciphertext) =
split_at_mut_checked(packet.packet_mut(), plaintext_end).ok_or(CryptoError)?;
let (slice_to_use, body_remaining) = mode.nonce_slice(plaintext, ciphertext)?;
let (pre_payload, body_remaining) =
split_at_mut_checked(body_remaining, mode.payload_prefix_len2()).ok_or(CryptoError)?;
let suffix_split_point = body_remaining
.len()
.checked_sub(mode.tag_suffix_len())
.ok_or(CryptoError)?;
let (body, post_payload) =
split_at_mut_checked(body_remaining, suffix_split_point).ok_or(CryptoError)?;
let tag_size = self.encryption_tag_len();
match self {
// Older modes place the tag before the payload and do not authenticate
// cleartext.
Self::XSalsa20Poly1305(secret_box, _) => {
let mut nonce = crypto_secretbox::Nonce::default();
nonce[..mode.nonce_size().min(slice_to_use.len())].copy_from_slice(slice_to_use);
let tag = crypto_secretbox::Tag::from_slice(&pre_payload[..tag_size]);
secret_box.decrypt_in_place_detached(&nonce, b"", body, tag)?;
},
// The below variants follow part of the SRTP spec (RFC3711, sec 3.1)
// by requiring that we include the cleartext header portion as
// authenticated data.
Self::Aes256Gcm(aes_gcm) => {
let mut nonce = aes_gcm::Nonce::default();
nonce[..mode.nonce_size()].copy_from_slice(slice_to_use);
let tag = aes_gcm::Tag::from_slice(&post_payload[..tag_size]);
aes_gcm.decrypt_in_place_detached(&nonce, plaintext, body, tag)?;
},
Self::XChaCha20Poly1305(cha_cha_poly1305) => {
let mut nonce = chacha20poly1305::XNonce::default();
nonce[..mode.nonce_size()].copy_from_slice(slice_to_use);
let tag = chacha20poly1305::Tag::from_slice(&post_payload[..tag_size]);
cha_cha_poly1305.decrypt_in_place_detached(&nonce, plaintext, body, tag)?;
},
}
Ok((plaintext_end + pre_payload.len(), post_payload.len()))
}
}
// Temporary functions -- MSRV is ostensibly 1.74, slice::split_at(_mut)_checked is 1.80+.
// TODO: Remove in v0.5+ with MSRV bump to 1.81+.
#[cfg(any(feature = "receive", test))]
#[inline]
#[must_use]
const fn split_at_checked(els: &[u8], mid: usize) -> Option<(&[u8], &[u8])> {
if mid <= els.len() {
Some(els.split_at(mid))
} else {
None
}
}
#[cfg(any(feature = "receive", test))]
#[inline]
#[must_use]
fn split_at_mut_checked(els: &mut [u8], mid: usize) -> Option<(&mut [u8], &mut [u8])> {
if mid <= els.len() {
Some(els.split_at_mut(mid))
} else {
None
}
}
#[cfg(test)]
mod test {
use super::*;
use crypto_secretbox::KeyInit;
use discortp::rtp::MutableRtpPacket;
#[test]
#[allow(deprecated)]
fn small_packet_decrypts_error() {
let mut buf = [0u8; MutableRtpPacket::minimum_packet_size()];
let modes = [CryptoMode::Normal, CryptoMode::Suffix, CryptoMode::Lite];
let modes = [
CryptoMode::Normal,
CryptoMode::Suffix,
CryptoMode::Lite,
CryptoMode::Aes256Gcm,
CryptoMode::XChaCha20Poly1305,
];
let mut pkt = MutableRtpPacket::new(&mut buf[..]).unwrap();
let cipher = Cipher::new_from_slice(&[1u8; KEY_SIZE]).unwrap();
for mode in modes {
// Coincidentally, these are all 32B for now.
let cipher = mode.cipher_from_key(&[1u8; 32]).unwrap();
// AIM: should error, and not panic.
assert!(mode.decrypt_in_place(&mut pkt, &cipher).is_err());
assert!(cipher.decrypt_rtp_in_place(&mut pkt).is_err());
}
}
#[test]
fn symmetric_encrypt_decrypt() {
#[allow(deprecated)]
fn symmetric_encrypt_decrypt_xsalsa20() {
const TRUE_PAYLOAD: [u8; 8] = [1, 2, 3, 4, 5, 6, 7, 8];
let mut buf = [0u8; MutableRtpPacket::minimum_packet_size()
+ TRUE_PAYLOAD.len()
+ TAG_SIZE
+ NONCE_SIZE];
+ XSalsa20Poly1305::TAG_SIZE
+ XSalsa20Poly1305::NONCE_SIZE];
let modes = [CryptoMode::Normal, CryptoMode::Lite, CryptoMode::Suffix];
let cipher = Cipher::new_from_slice(&[7u8; KEY_SIZE]).unwrap();
for mode in modes {
buf.fill(0);
let cipher = mode
.cipher_from_key(&[7u8; XSalsa20Poly1305::KEY_SIZE])
.unwrap();
let mut pkt = MutableRtpPacket::new(&mut buf[..]).unwrap();
let mut crypto_state = CryptoState::from(mode);
let payload = pkt.payload_mut();
payload[TAG_SIZE..TAG_SIZE + TRUE_PAYLOAD.len()].copy_from_slice(&TRUE_PAYLOAD[..]);
payload[XSalsa20Poly1305::TAG_SIZE..XSalsa20Poly1305::TAG_SIZE + TRUE_PAYLOAD.len()]
.copy_from_slice(&TRUE_PAYLOAD[..]);
let final_payload_size =
crypto_state.write_packet_nonce(&mut pkt, TAG_SIZE + TRUE_PAYLOAD.len());
let final_payload_size = crypto_state
.write_packet_nonce(&mut pkt, XSalsa20Poly1305::TAG_SIZE + TRUE_PAYLOAD.len());
let enc_succ = mode.encrypt_in_place(&mut pkt, &cipher, final_payload_size);
let enc_succ = cipher.encrypt_pkt_in_place(&mut pkt, final_payload_size);
assert!(enc_succ.is_ok());
let final_pkt_len = MutableRtpPacket::minimum_packet_size() + final_payload_size;
let mut pkt = MutableRtpPacket::new(&mut buf[..final_pkt_len]).unwrap();
assert!(mode.decrypt_in_place(&mut pkt, &cipher).is_ok());
assert!(cipher.decrypt_rtp_in_place(&mut pkt).is_ok());
}
}
#[test]
fn symmetric_encrypt_decrypt_tag_after_data() {
const TRUE_PAYLOAD: [u8; 8] = [1, 2, 3, 4, 5, 6, 7, 8];
for mode in [CryptoMode::Aes256Gcm, CryptoMode::XChaCha20Poly1305] {
let mut buf = vec![
0u8;
MutableRtpPacket::minimum_packet_size()
+ TRUE_PAYLOAD.len()
+ mode.nonce_size()
+ mode.encryption_tag_len()
];
buf.fill(0);
let cipher = mode.cipher_from_key(&[7u8; 32]).unwrap();
let mut pkt = MutableRtpPacket::new(&mut buf[..]).unwrap();
let mut crypto_state = CryptoState::from(mode);
let payload = pkt.payload_mut();
payload[mode.payload_prefix_len2()..TRUE_PAYLOAD.len()].copy_from_slice(&TRUE_PAYLOAD);
let final_payload_size = crypto_state.write_packet_nonce(&mut pkt, TRUE_PAYLOAD.len());
let enc_succ = cipher.encrypt_pkt_in_place(&mut pkt, final_payload_size);
assert!(enc_succ.is_ok());
let final_pkt_len = MutableRtpPacket::minimum_packet_size() + final_payload_size;
let mut pkt = MutableRtpPacket::new(&mut buf[..final_pkt_len]).unwrap();
assert!(cipher.decrypt_rtp_in_place(&mut pkt).is_ok());
}
}
#[test]
#[allow(deprecated)]
fn negotiate_cryptomode() {
// If we have no preference (or our preference is missing), choose the highest available in the set.
let test_set = [
CryptoMode::Suffix,
CryptoMode::XChaCha20Poly1305,
CryptoMode::Lite,
]
.map(CryptoMode::to_request_str);
assert_eq!(
CryptoMode::negotiate(test_set, None).unwrap(),
CryptoMode::XChaCha20Poly1305
);
assert_eq!(
CryptoMode::negotiate(test_set, Some(CryptoMode::Aes256Gcm)).unwrap(),
CryptoMode::XChaCha20Poly1305
);
// Preference wins in spite of the defined `priority` value.
assert_eq!(
CryptoMode::negotiate(test_set, Some(CryptoMode::Suffix)).unwrap(),
CryptoMode::Suffix
);
// If there is no mutual intelligibility, return an error.
let bad_modes = ["not_real", "des", "rc5"];
assert!(CryptoMode::negotiate(&bad_modes, None).is_err());
assert!(CryptoMode::negotiate(&bad_modes, Some(CryptoMode::Aes256Gcm)).is_err());
}
}

2
src/driver/tasks/error.rs
View File

@@ -20,7 +20,7 @@ pub type Result<T> = std::result::Result<T, Error>;
#[non_exhaustive]
pub enum Error {
Crypto(CryptoError),
#[cfg(feature = "receive")]
#[cfg(any(feature = "receive", test))]
/// Received an illegal voice packet on the voice UDP socket.
IllegalVoicePacket,
InterconnectFailure(Recipient),

3
src/driver/tasks/message/mixer.rs
View File

@@ -5,10 +5,9 @@ use super::UdpRxMessage;
use super::{Interconnect, TrackContext, WsMessage};
use crate::{
driver::{Bitrate, Config, CryptoState},
driver::{crypto::Cipher, Bitrate, Config, CryptoState},
input::{AudioStreamError, Compose, Parsed},
};
use crypto_secretbox::XSalsa20Poly1305 as Cipher;
use flume::Sender;
use std::{net::UdpSocket, sync::Arc};
use symphonia_core::{errors::Error as SymphoniaError, formats::SeekedTo};

32
src/driver/tasks/mixer/mod.rs
View File

@@ -15,10 +15,9 @@ use super::{
error::{Error, Result},
message::*,
};
use crate::driver::crypto::TAG_SIZE;
use crate::{
constants::*,
driver::MixMode,
driver::{CryptoMode, MixMode},
events::EventStore,
input::{Input, Parsed},
tracks::{Action, LoopState, PlayError, PlayMode, TrackCommand, TrackHandle, TrackState, View},
@@ -511,6 +510,15 @@ impl Mixer {
}
}
pub fn crypto_mode(&self) -> CryptoMode {
let mode = self.conn_active.as_ref().map(|v| v.crypto_state.kind());
if cfg!(not(test)) {
mode.expect("Shouldn't be mixing packets without access to a cipher + UDP dest.")
} else {
mode.unwrap_or_else(|| self.config.crypto_mode)
}
}
#[inline]
pub fn mix_and_build_packet(&mut self, packet: &mut [u8]) -> Result<usize> {
// symph_mix is an `AudioBuffer` (planar format), we need to convert this
@@ -545,8 +553,9 @@ impl Mixer {
);
let payload = rtp.payload_mut();
let pre_len = self.crypto_mode().payload_prefix_len2();
payload[TAG_SIZE..TAG_SIZE + SILENT_FRAME.len()].copy_from_slice(&SILENT_FRAME[..]);
payload[pre_len..pre_len + SILENT_FRAME.len()].copy_from_slice(&SILENT_FRAME[..]);
mix_len = MixType::Passthrough(SILENT_FRAME.len());
} else {
@@ -580,7 +589,8 @@ impl Mixer {
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload();
let opus_frame = (payload[TAG_SIZE..][..len]).to_vec();
let opus_frame =
(payload[self.crypto_mode().payload_prefix_len2()..][..len]).to_vec();
OutputMessage::Passthrough(opus_frame)
},
@@ -626,6 +636,7 @@ impl Mixer {
let payload = rtp.payload_mut();
let crypto_mode = conn.crypto_state.kind();
let first_payload_byte = crypto_mode.payload_prefix_len2();
// If passthrough, Opus payload in place already.
// Else encode into buffer with space for AEAD encryption headers.
@@ -635,14 +646,14 @@ impl Mixer {
let total_payload_space = payload.len() - crypto_mode.payload_suffix_len();
self.encoder.encode_float(
&send_buffer[..self.config.mix_mode.sample_count_in_frame()],
&mut payload[TAG_SIZE..total_payload_space],
&mut payload[first_payload_byte..total_payload_space],
)?
},
};
let final_payload_size = conn
.crypto_state
.write_packet_nonce(&mut rtp, TAG_SIZE + payload_len);
.write_packet_nonce(&mut rtp, first_payload_byte + payload_len);
// Packet encryption ignored in test modes.
#[cfg(not(test))]
@@ -651,11 +662,8 @@ impl Mixer {
let encrypt = self.config.override_connection.is_none();
if encrypt {
conn.crypto_state.kind().encrypt_in_place(
&mut rtp,
&conn.cipher,
final_payload_size,
)?;
conn.cipher
.encrypt_pkt_in_place(&mut rtp, final_payload_size)?;
}
Ok(RtpPacket::minimum_packet_size() + final_payload_size)
@@ -746,7 +754,7 @@ impl Mixer {
(Blame: VOICE_PACKET_MAX?)",
);
let payload = rtp.payload_mut();
let opus_frame = &mut payload[TAG_SIZE..];
let opus_frame = &mut payload[self.crypto_mode().payload_prefix_len2()..];
// Opus frame passthrough.
// This requires that we have only one PLAYING track, who has volume 1.0, and an

26
src/driver/tasks/udp_rx/mod.rs
View File

@@ -5,14 +5,14 @@ mod ssrc_state;
use self::{decode_sizes::*, playout_buffer::*, ssrc_state::*};
use super::message::*;
use crate::driver::CryptoMode;
use crate::{
constants::*,
driver::CryptoMode,
driver::crypto::Cipher,
events::{context_data::VoiceTick, internal_data::*, CoreContext},
Config,
};
use bytes::BytesMut;
use crypto_secretbox::XSalsa20Poly1305 as Cipher;
use discortp::{
demux::{self, DemuxedMut},
rtp::RtpPacket,
@@ -33,6 +33,7 @@ type RtpSsrc = u32;
struct UdpRx {
cipher: Cipher,
crypto_mode: CryptoMode,
decoder_map: HashMap<RtpSsrc, SsrcState>,
config: Config,
rx: Receiver<UdpRxMessage>,
@@ -143,7 +144,7 @@ impl UdpRx {
// For simplicity, if the event task fails then we nominate the mixing thread
// to rebuild their context etc. (hence, the `let _ =` statements.), as it will
// try to make contact every 20ms.
let crypto_mode = self.config.crypto_mode;
let crypto_mode = self.crypto_mode;
match demux::demux_mut(packet.as_mut()) {
DemuxedMut::Rtp(mut rtp) => {
@@ -153,11 +154,12 @@ impl UdpRx {
}
let packet_data = if self.config.decode_mode.should_decrypt() {
let out = crypto_mode
.decrypt_in_place(&mut rtp, &self.cipher)
let out = self
.cipher
.decrypt_rtp_in_place(&mut rtp)
.map(|(s, t)| (s, t, true));
if let Err(e) = out {
if let Err(ref e) = out {
warn!("RTP decryption failed: {:?}", e);
}
@@ -169,7 +171,7 @@ impl UdpRx {
let rtp = rtp.to_immutable();
let (rtp_body_start, rtp_body_tail, decrypted) = packet_data.unwrap_or_else(|| {
(
CryptoMode::payload_prefix_len(),
crypto_mode.payload_prefix_len2(),
crypto_mode.payload_suffix_len(),
false,
)
@@ -178,7 +180,7 @@ impl UdpRx {
let entry = self
.decoder_map
.entry(rtp.get_ssrc())
.or_insert_with(|| SsrcState::new(&rtp, &self.config));
.or_insert_with(|| SsrcState::new(&rtp, crypto_mode, &self.config));
// Only do this on RTP, rather than RTCP -- this pins decoder state liveness
// to *speech* rather than just presence.
@@ -201,9 +203,9 @@ impl UdpRx {
},
DemuxedMut::Rtcp(mut rtcp) => {
let packet_data = if self.config.decode_mode.should_decrypt() {
let out = crypto_mode.decrypt_in_place(&mut rtcp, &self.cipher);
let out = self.cipher.decrypt_rtcp_in_place(&mut rtcp);
if let Err(e) = out {
if let Err(ref e) = out {
warn!("RTCP decryption failed: {:?}", e);
}
@@ -214,7 +216,7 @@ impl UdpRx {
let (start, tail) = packet_data.unwrap_or_else(|| {
(
CryptoMode::payload_prefix_len(),
crypto_mode.payload_prefix_len2(),
crypto_mode.payload_suffix_len(),
)
});
@@ -242,6 +244,7 @@ pub(crate) async fn runner(
mut interconnect: Interconnect,
rx: Receiver<UdpRxMessage>,
cipher: Cipher,
crypto_mode: CryptoMode,
config: Config,
udp_socket: UdpSocket,
ssrc_signalling: Arc<SsrcTracker>,
@@ -250,6 +253,7 @@ pub(crate) async fn runner(
let mut state = UdpRx {
cipher,
crypto_mode,
decoder_map: HashMap::new(),
config,
rx,

8
src/driver/tasks/udp_rx/ssrc_state.rs
View File

@@ -19,6 +19,7 @@ use tracing::{error, warn};
#[derive(Debug)]
pub struct SsrcState {
playout_buffer: PlayoutBuffer,
crypto_mode: CryptoMode,
decoder: OpusDecoder,
decode_size: PacketDecodeSize,
pub(crate) prune_time: Instant,
@@ -26,11 +27,12 @@ pub struct SsrcState {
}
impl SsrcState {
pub fn new(pkt: &RtpPacket<'_>, config: &Config) -> Self {
pub fn new(pkt: &RtpPacket<'_>, crypto_mode: CryptoMode, config: &Config) -> Self {
let playout_capacity = config.playout_buffer_length.get() + config.playout_spike_length;
Self {
playout_buffer: PlayoutBuffer::new(playout_capacity, pkt.get_sequence().0),
crypto_mode,
decoder: OpusDecoder::new(SAMPLE_RATE, Channels::Stereo)
.expect("Failed to create new Opus decoder for source."),
decode_size: PacketDecodeSize::TwentyMillis,
@@ -72,8 +74,8 @@ impl SsrcState {
let extensions = rtp.get_extension() != 0;
let payload = rtp.payload();
let payload_offset = CryptoMode::payload_prefix_len();
let payload_end_pad = payload.len() - config.crypto_mode.payload_suffix_len();
let payload_offset = self.crypto_mode.payload_prefix_len2();
let payload_end_pad = payload.len() - self.crypto_mode.payload_suffix_len();
// We still need to compute missed packets here in case of long loss chains or similar.
// This occurs due to the fallback in 'store_packet' (i.e., empty buffer and massive seq difference).

29
src/driver/test_impls.rs
View File

@@ -1,8 +1,12 @@
#![allow(missing_docs)]
use super::{
scheduler::*,
tasks::{message::*, mixer::Mixer},
*,
};
use crate::{
constants::*,
driver::crypto::KEY_SIZE,
input::{
cached::Compressed,
codecs::{CODEC_REGISTRY, PROBE},
@@ -11,17 +15,11 @@ use crate::{
test_utils,
tracks::LoopState,
};
use crypto_secretbox::{KeyInit, XSalsa20Poly1305 as Cipher};
use crypto_secretbox::XSalsa20Poly1305;
use flume::Receiver;
use std::{io::Cursor, net::UdpSocket, sync::Arc};
use tokio::runtime::Handle;
use super::{
scheduler::*,
tasks::{message::*, mixer::Mixer},
*,
};
// create a dummied task + interconnect.
// measure perf at varying numbers of sources (binary 1--64) without passthrough support.
@@ -65,18 +63,25 @@ impl Mixer {
.connect("127.0.0.1:5316")
.expect("Failed to connect to local dest port.");
#[allow(deprecated)]
let mode = CryptoMode::Normal;
let cipher = mode
.cipher_from_key(&[0u8; XSalsa20Poly1305::KEY_SIZE])
.unwrap();
let crypto_state = mode.into();
#[cfg(feature = "receive")]
let fake_conn = MixerConnection {
cipher: Cipher::new_from_slice(&[0u8; KEY_SIZE]).unwrap(),
crypto_state: CryptoState::Normal,
cipher,
crypto_state,
udp_rx: udp_receiver_tx,
udp_tx,
};
#[cfg(not(feature = "receive"))]
let fake_conn = MixerConnection {
cipher: Cipher::new_from_slice(&[0u8; KEY_SIZE]).unwrap(),
crypto_state: CryptoState::Normal,
cipher,
crypto_state,
udp_tx,
};

2
src/input/input_tests.rs
View File

@@ -1,3 +1,5 @@
#![allow(missing_docs)]
use crate::{
driver::Driver,
tracks::{PlayMode, ReadyState, Track},

5
src/input/sources/mod.rs
View File

@@ -38,13 +38,16 @@ impl AsyncReadOnlySource {
}
/// Gets a reference to the underlying reader.
#[allow(clippy::borrowed_box)] // FIXME: Changing this may break compatibility. remedy in v0.5
#[must_use]
pub fn get_ref(&self) -> &Box<dyn AsyncRead + Send + Sync + Unpin> {
&self.stream
}
/// Unwraps this `AsyncReadOnlySource`, returning the underlying reader.
#[must_use]
pub fn into_inner<R>(self) -> Box<dyn AsyncRead + Send + Sync + Unpin> {
self.stream.into()
self.stream
}
}

4
src/tracks/looping.rs
View File

@@ -53,7 +53,7 @@ mod tests {
}
#[tokio::test]
#[ntest::timeout(10_000)]
#[ntest::timeout(15_000)]
async fn finite_track_loops_work() {
let (t_handle, config) = Config::test_cfg(true);
let mut driver = Driver::new(config.clone());
@@ -89,7 +89,7 @@ mod tests {
}
#[tokio::test]
#[ntest::timeout(10_000)]
#[ntest::timeout(15_000)]
async fn infinite_track_loops_work() {
let (t_handle, config) = Config::test_cfg(true);
let mut driver = Driver::new(config.clone());

4
src/tracks/queue.rs
View File

@@ -416,7 +416,7 @@ mod tests {
}
#[tokio::test]
#[ntest::timeout(10_000)]
#[ntest::timeout(15_000)]
async fn next_track_plays_on_skip() {
let (t_handle, config) = Config::test_cfg(true);
let mut driver = Driver::new(config.clone());
@@ -456,7 +456,7 @@ mod tests {
}
#[tokio::test]
#[ntest::timeout(10_000)]
#[ntest::timeout(15_000)]
async fn next_track_plays_on_err() {
let (t_handle, config) = Config::test_cfg(true);
let mut driver = Driver::new(config.clone());