moq-e2ee 0.0.11

End-to-end encryption for MoQ groups, datagrams, and track names.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

//! Grouped tracks and datagrams on moq-lite and MoQ Transport.

mod support;

use std::time::Duration;

use moq_e2ee::credential::Config;
use moq_e2ee::{Credential, Epoch};
use moq_net::{Hop, Timestamp, Version};
use support::harness::{MockConnectOptions, MockPair, connect_mock};

const TEST_TIMEOUT: Duration = Duration::from_secs(10);

fn produce_origin(hop: u64) -> moq_net::origin::Producer {
	let (producer, driver) = moq_net::origin::Producer::new(moq_net::origin::Config::new(Hop::new(hop).unwrap()));
	tokio::spawn(support::harness::run(driver));
	producer
}

struct Fixture {
	producer: moq_e2ee::track::Producer,
	consumer: moq_e2ee::track::Consumer,
	_broadcast: moq_net::broadcast::Producer,
	_pair: MockPair,
}

async fn connect_protected(version: Version, track: &str) -> Fixture {
	let cred = Credential::new(Config {
		context: format!("transport-{version}-{track}").into(),
		kid: 7,
		secret: *b"moq-e2ee-00 test secret!!!!!!!!!",
	})
	.unwrap();
	let generation = cred.generation(Epoch::mint());
	let path = cred.path("meeting.hang").unwrap().join(generation.epoch().as_str());
	let name = generation.name(track).unwrap();

	let publisher = produce_origin(1);
	let consumer_origin = produce_origin(2);

	let broadcast = publisher.create_broadcast(&path).unwrap();
	let net = broadcast.create_track(name.as_str(), None).unwrap();
	broadcast.announce(Default::default()).unwrap();

	let mut options = MockConnectOptions::new(version);
	options.server_publish = Some(publisher);
	options.client_subscribe = Some(consumer_origin.clone());
	let pair = connect_mock(options).await;

	// The subscriber knows the opaque prefix and takes the epoch from the discovered path.
	let consumer = consumer_origin.consume();
	consumer.routed(&path).await.unwrap();
	let epoch: Epoch = path.parts().last().unwrap().parse().unwrap();
	let generation = cred.generation(epoch);
	let remote = consumer.request_broadcast(&path).await.unwrap();
	let subscriber = remote
		.track(name.as_str())
		.unwrap()
		.subscribe(Some(
			moq_net::track::Subscription::default().with_max_age(Duration::from_secs(60)),
		))
		.await
		.unwrap();

	Fixture {
		producer: generation.produce(net).unwrap(),
		consumer: generation.consume(subscriber).unwrap(),
		_broadcast: broadcast,
		_pair: pair,
	}
}

#[tokio::test]
async fn grouped_over_lite() {
	tokio::time::timeout(TEST_TIMEOUT, async {
		let mut fixture = connect_protected("moq-lite-05".parse().unwrap(), "video").await;
		let mut group = fixture.producer.append_group().unwrap();
		group
			.write_frame(Timestamp::from_millis(1).unwrap(), b"lite-frame")
			.unwrap();
		group.finish().unwrap();
		let mut group = fixture.consumer.recv_group().await.unwrap().unwrap();
		let frame = group.read_frame().await.unwrap().unwrap();
		assert_eq!(&frame.plaintext[..], b"lite-frame");
	})
	.await
	.expect("timed out");
}

#[tokio::test]
async fn grouped_over_ietf() {
	tokio::time::timeout(TEST_TIMEOUT, async {
		let mut fixture = connect_protected("moq-transport-21".parse().unwrap(), "video").await;
		let mut group = fixture.producer.append_group().unwrap();
		group
			.write_frame(Timestamp::from_millis(1).unwrap(), b"ietf-frame")
			.unwrap();
		group.finish().unwrap();
		let mut group = fixture.consumer.recv_group().await.unwrap().unwrap();
		let frame = group.read_frame().await.unwrap().unwrap();
		assert_eq!(&frame.plaintext[..], b"ietf-frame");
	})
	.await
	.expect("timed out");
}

async fn datagram_roundtrip(version: &str) {
	tokio::time::timeout(TEST_TIMEOUT, async {
		let mut fixture = connect_protected(version.parse().unwrap(), "audio").await;
		fixture
			.producer
			.append_datagram(Timestamp::from_millis(9).unwrap(), b"opus")
			.unwrap();
		match fixture.consumer.recv_datagram().await.unwrap() {
			Some(moq_e2ee::datagram::Event::Datagram(d)) => {
				assert_eq!(&d.plaintext[..], b"opus");
				assert_eq!(d.sequence, 0);
			}
			other => panic!("{other:?}"),
		}
	})
	.await
	.expect("timed out");
}

#[tokio::test]
async fn datagrams_over_lite() {
	datagram_roundtrip("moq-lite-05").await;
}

#[tokio::test]
async fn datagrams_over_ietf() {
	datagram_roundtrip("moq-transport-21").await;
}