moq-e2ee 0.0.2

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

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

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");
}

#[tokio::test]
async fn datagrams_over_lite() {
	tokio::time::timeout(TEST_TIMEOUT, async {
		let mut fixture = connect_protected("moq-lite-05".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");
}