- commit
- e267bba082fcdfce41d08ca9554573e4b1a4b918
- parent
- 5a2bab9ce4f6e7da37a7e6f39e94753c55e22dfc
- author
- Dami <dami.etoile@gmail.com>
@@ -5663,6 +5663,7 @@ dependencies = [
56635663 "hang",56645664 "serde",56655665 "serde_json",5666+ "sha2 0.10.9",56665667 "thiserror 2.0.19",56675668 "tokio",56685669 ]@@ -2,10 +2,14 @@ use std::time::Duration;
22 33 use anyhow::{Context, Result};44 use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};5-use sigil_protocol::{GoqCatalogDocument, MAX_MOQ_CATALOG_BYTES};5+use sigil_protocol::{6+ GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES,7+ SignedMediaGenerationCertificate,8+};69 710 811 pub(crate) enum MoqCatalogMode {12+ GoqV2Authenticated,913 GoqV1,1014 AbsentStaticTrackCompat,1115 }@@ -13,6 +17,7 @@ pub(crate) enum MoqCatalogMode {
1317 impl MoqCatalogMode {1418 pub(crate) const fn label(self) -> &'static str {1519 match self {20+ Self::GoqV2Authenticated => "goq-v2-authenticated",1621 Self::GoqV1 => "goq-v1",1722 Self::AbsentStaticTrackCompat => "absent-static-track-compat",1823 }@@ -22,6 +27,7 @@ impl MoqCatalogMode {
2227 pub(crate) struct MoqCatalogSelection {2328 pub(crate) track: TrackConsumer,2429 pub(crate) mode: MoqCatalogMode,30+ pub(crate) certificate: Option<SignedMediaGenerationCertificate>,2531 }2632 2733 fn is_track_not_found(error: &MoqError) -> bool {@@ -36,6 +42,7 @@ fn subscribe_static_video_track(broadcast: &BroadcastConsumer) -> Result<MoqCata
3642 Ok(MoqCatalogSelection {3743 track,3844 mode: MoqCatalogMode::AbsentStaticTrackCompat,45+ certificate: None,3946 })4047 }4148 @@ -98,6 +105,66 @@ pub(crate) async fn subscribe_goq_video_track(
98105 Ok(MoqCatalogSelection {99106 track,100107 mode: MoqCatalogMode::GoqV1,108+ certificate: None,109+ })110+}111+112+pub(crate) async fn subscribe_authenticated_goq_video_track(113+ broadcast: &BroadcastConsumer,114+ timeout: Duration,115+ expected_host: [u8; 32],116+ expected_generation: u64,117+ now_unix: u64,118+) -> Result<MoqCatalogSelection> {119+ let mut catalog_track = broadcast120+ .subscribe_track(&hang::Catalog::default_track())121+ .context("authenticated v2 media requires catalog.json")?;122+ let deadline = tokio::time::Instant::now() + timeout;123+ let mut group = tokio::time::timeout_at(deadline, catalog_track.next_group())124+ .await125+ .context("timed out waiting for authenticated Goq catalog")?126+ .context("reading authenticated Goq catalog")?127+ .context("authenticated Goq catalog ended before its snapshot")?;128+ anyhow::ensure!(129+ group.sequence == 0,130+ "authenticated Goq catalog must be immutable group 0"131+ );132+ let frame_count = tokio::time::timeout_at(deadline, group.finished())133+ .await134+ .context("timed out finishing authenticated Goq catalog")?135+ .context("finishing authenticated Goq catalog")?;136+ anyhow::ensure!(137+ frame_count == 1,138+ "authenticated Goq catalog must contain one frame"139+ );140+ let mut frame = tokio::time::timeout_at(deadline, group.get_frame(0))141+ .await142+ .context("timed out opening authenticated Goq catalog")?143+ .context("opening authenticated Goq catalog")?144+ .context("authenticated Goq catalog has no frame 0")?;145+ anyhow::ensure!(146+ frame.size <= MAX_MOQ_CATALOG_BYTES as u64,147+ "authenticated Goq catalog exceeds {MAX_MOQ_CATALOG_BYTES} bytes"148+ );149+ let snapshot = tokio::time::timeout_at(deadline, frame.read_all())150+ .await151+ .context("timed out reading authenticated Goq catalog")?152+ .context("reading authenticated Goq catalog")?;153+ let document: GoqCatalogDocumentV2 =154+ serde_json::from_slice(&snapshot).context("decoding authenticated Goq catalog")?;155+ let certificate = document156+ .validate()157+ .context("validating authenticated Goq catalog")?;158+ certificate159+ .verify_binding(expected_host, expected_generation, now_unix)160+ .context("verifying media generation certificate binding")?;161+ let track = broadcast162+ .subscribe_track(&Track::new(document.goq.video.track.name))163+ .context("subscribing to authenticated Goq video track")?;164+ Ok(MoqCatalogSelection {165+ track,166+ mode: MoqCatalogMode::GoqV2Authenticated,167+ certificate: Some(certificate),101168 })102169 }103170 @@ -14,19 +14,21 @@ use iroh_moq::{Moq, MoqSession};
1414 use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};1515 use sigil_protocol::{1616 AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1,17- CONTROL_ALPN_V1, CONTROL_ALPN_V2, Capability, ClientControlEnvelopeV2, ClientHello,18- ClientHelloV2, ControllerSlot, FocusCommandActionV2, FocusCommandV2, FocusStateV2, FrameFlags,19- GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN, GAMEPAD_TRIGGER_MAX, GamepadState, INPUT_ALPN_V1,20- INPUT_ALPN_V2, INVITATION_CLOCK_SKEW_SECS, InputAck, InputClientHelloV2, InputEvent,21- InputEventV2, InvitationGrants, KeyframeRequestReasonV3, MAX_INVITATION_TOKEN_LEN,22- MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_ID_V3, MEDIA_ALPN_V3, MEDIA_FEEDBACK_ALPN_V1,23- MediaCodec, MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame,24- MediaObjectV3, PointerPosition, PointerSurfaceDimensions, ProtocolError,25- ServerControlEnvelopeV2, SignedInvitation, decode_media_frame_object, media_moq_broadcast_name,26- read_adaptive_bitrate_decision_v1, read_host_hello, read_host_hello_v2, read_input_ack,27- read_input_ack_v2, read_input_host_hello_v2, read_media_object_v3, read_server_control_v2,28- write_client_control_v2, write_client_hello, write_client_hello_v2,29- write_input_client_hello_v2, write_input_event, write_input_event_v2,17+ AuthenticatedMediaObject, CONTROL_ALPN_V1, CONTROL_ALPN_V2, Capability,18+ ClientControlEnvelopeV2, ClientHello, ClientHelloV2, ControllerSlot, FocusCommandActionV2,19+ FocusCommandV2, FocusStateV2, FrameFlags, GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN,20+ GAMEPAD_TRIGGER_MAX, GamepadState, INPUT_ALPN_V1, INPUT_ALPN_V2, INVITATION_CLOCK_SKEW_SECS,21+ InputAck, InputClientHelloV2, InputEvent, InputEventV2, InvitationGrants,22+ KeyframeRequestReasonV3, MAX_INVITATION_TOKEN_LEN, MAX_MEDIA_GROUP_BYTES_V3,23+ MAX_MEDIA_OBJECT_ID_V3, MEDIA_ALPN_V3, MEDIA_FEEDBACK_ALPN_V1, MediaCodec,24+ MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame,25+ MediaObjectCoordinates, MediaObjectV3, MediaTrack, PointerPosition, PointerSurfaceDimensions,26+ ProtocolError, ServerControlEnvelopeV2, SignedInvitation, SignedMediaGenerationCertificate,27+ SignedSubscriptionCapability, SubscriptionTracks, decode_media_frame_object,28+ media_moq_broadcast_name, read_adaptive_bitrate_decision_v1, read_host_hello,29+ read_host_hello_v2, read_input_ack, read_input_ack_v2, read_input_host_hello_v2,30+ read_media_object_v3, read_server_control_v2, write_client_control_v2, write_client_hello,31+ write_client_hello_v2, write_input_client_hello_v2, write_input_event, write_input_event_v2,3032 write_media_control_request_v3, write_media_feedback_report_v1,3133 };3234 @@ -39,7 +41,9 @@ mod identity;
3941 4042 mod moq_catalog;4143 42-use moq_catalog::{MoqCatalogMode, subscribe_goq_video_track};44+use moq_catalog::{45+ MoqCatalogMode, subscribe_authenticated_goq_video_track, subscribe_goq_video_track,46+};4347 4448 const MEDIA_OBJECT_CAPACITY: usize = 4;4549 @@ -490,6 +494,13 @@ struct MoqProbeReceiver {
490494 maximum_group_objects: usize,491495 maximum_group_bytes: usize,492496 historical_suffix_frames: u64,497+ certificate: Option<SignedMediaGenerationCertificate>,498+}499+500+struct MoqProbeAuthentication {501+ expected_host: [u8; 32],502+ subscription_capability: String,503+ now_unix: u64,493504 }494505 495506 impl MoqProbeReceiver {@@ -498,6 +509,7 @@ impl MoqProbeReceiver {
498509 address: EndpointAddr,499510 session_id: u64,500511 timeout: Duration,512+ authentication: Option<MoqProbeAuthentication>,501513 ) -> Result<Self> {502514 let broadcast_name = media_moq_broadcast_name(session_id)503515 .context("deriving session-scoped MoQ broadcast name")?;@@ -518,9 +530,30 @@ impl MoqProbeReceiver {
518530 format!("timed out subscribing to upstream MoQ broadcast {broadcast_name}")519531 })?520532 .with_context(|| format!("subscribing to upstream MoQ broadcast {broadcast_name}"))?;521- let catalog = subscribe_goq_video_track(&broadcast, timeout)533+ let catalog = if let Some(authentication) = authentication {534+ SignedSubscriptionCapability::decode(&authentication.subscription_capability)?535+ .verify_binding(536+ authentication.expected_host,537+ session_id,538+ *endpoint.id().as_bytes(),539+ SubscriptionTracks::VIDEO_H264,540+ 1,541+ authentication.now_unix,542+ )?;543+ subscribe_authenticated_goq_video_track(544+ &broadcast,545+ timeout,546+ authentication.expected_host,547+ session_id,548+ authentication.now_unix,549+ )522550 .await523- .context("resolving Goq MoQ catalog")?;551+ .context("resolving authenticated Goq MoQ catalog")?552+ } else {553+ subscribe_goq_video_track(&broadcast, timeout)554+ .await555+ .context("resolving Goq MoQ catalog")?556+ };524557 Ok(Self {525558 lifetime: MoqProbeLifetime {526559 _moq: moq,@@ -539,6 +572,7 @@ impl MoqProbeReceiver {
539572 maximum_group_objects: 0,540573 maximum_group_bytes: 0,541574 historical_suffix_frames: 0,575+ certificate: catalog.certificate,542576 })543577 }544578 @@ -625,12 +659,53 @@ impl MoqProbeReceiver {
625659 cursor.object_bytes,626660 object.len(),627661 )?;628- let frame = decode_media_frame_object(&object).with_context(|| {662+ let authenticated_coordinates = self663+ .certificate664+ .as_ref()665+ .map(|_| AuthenticatedMediaObject::coordinates(&object))666+ .transpose()667+ .with_context(|| {668+ format!(669+ "decoding authenticated upstream MoQ group {} object {}",670+ cursor.sequence, cursor.object_count671+ )672+ })?;673+ let payload = if let (Some(certificate), Some(coordinates)) =674+ (&self.certificate, authenticated_coordinates)675+ {676+ AuthenticatedMediaObject::verify(677+ &object,678+ certificate,679+ MediaObjectCoordinates {680+ generation_id: certificate.claims.generation_id,681+ track: MediaTrack::VideoH264,682+ group_id: cursor.sequence,683+ object_id: u32::try_from(cursor.object_count)684+ .context("authenticated MoQ object index overflow")?,685+ flags: coordinates.flags,686+ },687+ )688+ .with_context(|| {689+ format!(690+ "verifying authenticated upstream MoQ group {} object {}",691+ cursor.sequence, cursor.object_count692+ )693+ })?694+ } else {695+ object.as_ref()696+ };697+ let frame = decode_media_frame_object(payload).with_context(|| {629698 format!(630699 "decoding upstream MoQ group {} object {}",631700 cursor.sequence, cursor.object_count632701 )633702 })?;703+ ensure!(704+ authenticated_coordinates.is_none_or(|coordinates| {705+ coordinates.flags == u16::from(frame.header.flags.bits())706+ }),707+ "authenticated media flags do not match the decoded media envelope"708+ );634709 let first_object = cursor.object_count == 0;635710 let contiguous = validate_moq_probe_frame(636711 cursor.sequence,@@ -703,6 +778,9 @@ async fn run_control_v2_smoke(
703778 let session_id = host704779 .session_id705780 .context("host omitted control v2 session id")?;781+ let subscription_capability = host782+ .media_subscription_capability783+ .context("host omitted control v2 media subscription capability")?;706784 let initial = host707785 .snapshot708786 .context("host omitted initial control v2 snapshot")?;@@ -716,6 +794,14 @@ async fn run_control_v2_smoke(
716794 address.clone(),717795 session_id,718796 Duration::from_secs(timeout_seconds),797+ Some(MoqProbeAuthentication {798+ expected_host: *address.id.as_bytes(),799+ subscription_capability,800+ now_unix: SystemTime::now()801+ .duration_since(UNIX_EPOCH)802+ .context("system clock is before the Unix epoch")?803+ .as_secs(),804+ }),719805 )720806 .await?;721807 let focus_request = FocusCommandV2 {@@ -857,6 +943,9 @@ async fn run_control_v2_smoke(
857943 println!("transport=iroh-moq");858944 println!("control_alpn=sigil/control/2");859945 println!("input_alpn=sigil/input/2");946+ println!("media_authentication=ed25519-v1");947+ println!("media_generation_certified=ok");948+ println!("subscription_endpoint_binding=ok");860949 println!("frames={accepted}");861950 println!("initial_snapshot_revision={}", initial.revision);862951 println!("focus_grant_revision={}", focused.revision);@@ -1541,6 +1630,7 @@ async fn main() -> Result<()> {
15411630 address.clone(),15421631 session_id,15431632 Duration::from_secs(args.timeout_seconds),1633+ None,15441634 )15451635 .await?,15461636 )@@ -0,0 +1,291 @@
1+use std::collections::VecDeque;2+use std::sync::Arc;3+use std::time::Instant;4+5+use anyhow::{Context, Result, ensure};6+use clap::Parser;7+use iroh::SecretKey;8+use sigil_protocol::{9+ AuthenticatedMediaObject, FrameFlags, MediaFrame, MediaFrameHeader, MediaGenerationSigningKey,10+ MediaObjectCoordinates, MediaTrack, SignedSubscriptionCapability, SubscriptionClaims,11+ SubscriptionTracks, encode_media_frame_object,12+};13+14+const VIDEO_FPS: u64 = 60;15+const AUDIO_OBJECTS_PER_SECOND: u64 = 50;16+const VIDEO_PAYLOAD_BYTES: usize = 64 * 1024;17+const AUDIO_PAYLOAD_BYTES: usize = 200;18+const RELAY_QUEUE_CAPACITY: usize = 4;19+const AUTHORIZATION_REVISION: u64 = 1;20+21+22+23+24+25+26+struct Args {27+ 28+ video: String,29+ 30+ audio: String,31+ 32+ duration_seconds: u64,33+}34+35+36+struct BenchmarkEvidence {37+ host_bytes: u64,38+ downstream_bytes: u64,39+ sign_nanos: u128,40+ verification_micros: Vec<u64>,41+ verification_nanos: u128,42+ maximum_relay_queue: usize,43+ first_object: Option<(Vec<u8>, MediaObjectCoordinates)>,44+}45+46+fn main() -> Result<()> {47+ let args = Args::parse();48+ ensure!(49+ args.video == "1280x800@60",50+ "only the fixed 1280x800@60 target is supported"51+ );52+ ensure!(53+ args.audio == "opus-48k-stereo",54+ "only Opus 48 kHz stereo is supported"55+ );56+57+ let host = SecretKey::from_bytes(&[0x31; 32]);58+ let portal_a = SecretKey::from_bytes(&[0x41; 32]);59+ let portal_b = SecretKey::from_bytes(&[0x42; 32]);60+ let generation_id = 7_u64;61+ let issued_at = 1_700_000_000_u64;62+ let generation = MediaGenerationSigningKey::from_bytes(&[0x51; 32]);63+ let certificate = generation.certify(64+ *host.public().as_bytes(),65+ &host.to_bytes(),66+ generation_id,67+ issued_at,68+ issued_at + 600,69+ )?;70+ certificate.verify_binding(*host.public().as_bytes(), generation_id, issued_at + 1)?;71+ let subscription = SignedSubscriptionCapability::issue(72+ SubscriptionClaims::new(73+ *host.public().as_bytes(),74+ generation_id,75+ *portal_b.public().as_bytes(),76+ SubscriptionTracks::VIDEO_H264.union(SubscriptionTracks::AUDIO_OPUS),77+ AUTHORIZATION_REVISION,78+ issued_at,79+ issued_at + 300,80+ [0x61; 32],81+ 1,82+ )?,83+ &host.to_bytes(),84+ )?;85+ subscription.verify_binding(86+ *host.public().as_bytes(),87+ generation_id,88+ *portal_b.public().as_bytes(),89+ SubscriptionTracks::VIDEO_H264.union(SubscriptionTracks::AUDIO_OPUS),90+ AUTHORIZATION_REVISION,91+ issued_at + 1,92+ )?;93+94+ let mut evidence = BenchmarkEvidence {95+ verification_micros: Vec::with_capacity(96+ usize::try_from(args.duration_seconds * (VIDEO_FPS + AUDIO_OBJECTS_PER_SECOND))97+ .context("benchmark sample capacity overflow")?,98+ ),99+ ..BenchmarkEvidence::default()100+ };101+ let video_payload: Arc<[u8]> = Arc::from(vec![0x65; VIDEO_PAYLOAD_BYTES]);102+ let audio_payload = vec![0x7a; AUDIO_PAYLOAD_BYTES];103+ let mut relay_queue = VecDeque::with_capacity(RELAY_QUEUE_CAPACITY);104+105+ for sequence in 0..args.duration_seconds * VIDEO_FPS {106+ let group_id = sequence / VIDEO_FPS;107+ let object_id = u32::try_from(sequence % VIDEO_FPS).expect("video object id is bounded");108+ let flags = if object_id == 0 {109+ FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG)110+ } else {111+ FrameFlags::NONE112+ };113+ let frame = MediaFrame::new(114+ MediaFrameHeader::h264(115+ 1280,116+ 800,117+ video_payload.len(),118+ sequence,119+ sequence * 1_000_000 / VIDEO_FPS,120+ i64::try_from(sequence * 1_000_000 / VIDEO_FPS).expect("bounded timestamp"),121+ flags,122+ )?,123+ Arc::clone(&video_payload),124+ )?;125+ let payload = encode_media_frame_object(&frame)?;126+ authenticate_relay_verify(127+ &generation,128+ &certificate,129+ MediaObjectCoordinates {130+ generation_id,131+ track: MediaTrack::VideoH264,132+ group_id,133+ object_id,134+ flags: u16::from(flags.bits()),135+ },136+ &payload,137+ &mut relay_queue,138+ &mut evidence,139+ )?;140+ }141+142+ for sequence in 0..args.duration_seconds * AUDIO_OBJECTS_PER_SECOND {143+ authenticate_relay_verify(144+ &generation,145+ &certificate,146+ MediaObjectCoordinates {147+ generation_id,148+ track: MediaTrack::AudioOpus,149+ group_id: sequence / AUDIO_OBJECTS_PER_SECOND,150+ object_id: u32::try_from(sequence % AUDIO_OBJECTS_PER_SECOND)151+ .expect("audio object id is bounded"),152+ flags: 0,153+ },154+ &audio_payload,155+ &mut relay_queue,156+ &mut evidence,157+ )?;158+ }159+160+ ensure!(relay_queue.is_empty(), "relay queue did not drain");161+ let (first_object, first_coordinates) = evidence162+ .first_object163+ .as_ref()164+ .context("benchmark did not produce an authenticated object")?;165+ let mut tampered = first_object.clone();166+ *tampered167+ .last_mut()168+ .context("authenticated object was empty")? ^= 1;169+ ensure!(170+ AuthenticatedMediaObject::verify(&tampered, &certificate, *first_coordinates).is_err(),171+ "tampered relayed media was trusted"172+ );173+ ensure!(174+ subscription175+ .verify_binding(176+ *host.public().as_bytes(),177+ generation_id,178+ *portal_a.public().as_bytes(),179+ SubscriptionTracks::VIDEO_H264,180+ AUTHORIZATION_REVISION,181+ issued_at + 1,182+ )183+ .is_err(),184+ "wrong subscriber was authorized"185+ );186+ ensure!(187+ subscription188+ .verify_binding(189+ *host.public().as_bytes(),190+ generation_id,191+ *portal_b.public().as_bytes(),192+ SubscriptionTracks::VIDEO_H264,193+ AUTHORIZATION_REVISION,194+ issued_at + 301,195+ )196+ .is_err(),197+ "expired subscription was authorized"198+ );199+ AuthenticatedMediaObject::verify(first_object, &certificate, *first_coordinates)200+ .context("direct-host fallback could not verify the relayed object")?;201+202+ evidence.verification_micros.sort_unstable();203+ let p95_index = evidence204+ .verification_micros205+ .len()206+ .saturating_mul(95)207+ .div_ceil(100)208+ - 1;209+ let verify_p95_us = evidence.verification_micros[p95_index];210+ let host_direct_bytes = evidence.host_bytes.saturating_mul(2);211+ let savings_percent = if host_direct_bytes == 0 {212+ 0213+ } else {214+ 100_u64.saturating_sub(evidence.host_bytes.saturating_mul(100) / host_direct_bytes)215+ };216+ let sign_cpu_percent =217+ evidence.sign_nanos as f64 / (args.duration_seconds as f64 * 1_000_000_000_f64) * 100.0;218+ let verify_cpu_percent = evidence.verification_nanos as f64219+ / (args.duration_seconds as f64 * 1_000_000_000_f64)220+ * 100.0;221+222+ println!("relay_spike=ok");223+ println!("authentication_mode=ed25519-v1");224+ println!("decision=defer-production-mesh");225+ println!("virtual_duration_seconds={}", args.duration_seconds);226+ println!("video_objects={}", args.duration_seconds * VIDEO_FPS);227+ println!(228+ "audio_objects={}",229+ args.duration_seconds * AUDIO_OBJECTS_PER_SECOND230+ );231+ println!("host_upload_bytes={}", evidence.host_bytes);232+ println!("direct_two_viewer_host_upload_bytes={host_direct_bytes}");233+ println!("host_upload_savings_percent={savings_percent}");234+ println!("relay_downstream_bytes={}", evidence.downstream_bytes);235+ println!("sign_cpu_percent_of_virtual_realtime={sign_cpu_percent:.4}");236+ println!("verify_cpu_percent_of_virtual_realtime={verify_cpu_percent:.4}");237+ println!("verify_p95_us={verify_p95_us}");238+ println!("added_local_latency_p95_us={verify_p95_us}");239+ println!("maximum_relay_queue={}", evidence.maximum_relay_queue);240+ println!("tamper_rejected=ok");241+ println!("wrong_subscriber_rejected=ok");242+ println!("expired_subscription_rejected=ok");243+ println!("relay_loss_direct_fallback=ok");244+ println!("withheld_media_trusted=0");245+ Ok(())246+}247+248+fn authenticate_relay_verify(249+ generation: &MediaGenerationSigningKey,250+ certificate: &sigil_protocol::SignedMediaGenerationCertificate,251+ coordinates: MediaObjectCoordinates,252+ payload: &[u8],253+ relay_queue: &mut VecDeque<Vec<u8>>,254+ evidence: &mut BenchmarkEvidence,255+) -> Result<()> {256+ ensure!(257+ relay_queue.len() < RELAY_QUEUE_CAPACITY,258+ "relay queue exceeded its bound"259+ );260+ let sign_started = Instant::now();261+ let object = generation.authenticate(coordinates, payload)?.into_bytes();262+ evidence.sign_nanos = evidence263+ .sign_nanos264+ .saturating_add(sign_started.elapsed().as_nanos());265+ evidence.host_bytes = evidence266+ .host_bytes267+ .saturating_add(u64::try_from(object.len()).unwrap_or(u64::MAX));268+ if evidence.first_object.is_none() {269+ evidence.first_object = Some((object.clone(), coordinates));270+ }271+ relay_queue.push_back(object);272+ evidence.maximum_relay_queue = evidence.maximum_relay_queue.max(relay_queue.len());273+274+ let relayed = relay_queue275+ .pop_front()276+ .context("relay queue lost an object")?;277+ evidence.downstream_bytes = evidence278+ .downstream_bytes279+ .saturating_add(u64::try_from(relayed.len()).unwrap_or(u64::MAX));280+ let verify_started = Instant::now();281+ let verified = AuthenticatedMediaObject::verify(&relayed, certificate, coordinates)?;282+ let verify_elapsed = verify_started.elapsed();283+ evidence.verification_nanos = evidence284+ .verification_nanos285+ .saturating_add(verify_elapsed.as_nanos());286+ evidence287+ .verification_micros288+ .push(u64::try_from(verify_elapsed.as_micros()).unwrap_or(u64::MAX));289+ ensure!(verified == payload, "verified relay payload changed bytes");290+ Ok(())291+}@@ -870,6 +870,7 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
870870 (loaded.config, loaded.revision, lifecycle)871871 };872872 let secret = identity::load(&config.identity_path)?;873+ let host_secret = secret.to_bytes();873874 let sessions = Arc::new(SessionRegistry::default());874875 let publisher = appliance::RuntimePublisher::start(875876 secret.public(),@@ -1011,6 +1012,7 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
10111012 sessions: Arc::clone(&sessions),10121013 authorization: authorization.clone(),10131014 input_operations: Arc::clone(&input_operations),1015+ host_secret,10141016 },10151017 )10161018 .accept(@@ -1,16 +1,22 @@
11 use anyhow::{Context, Result, ensure};22 use moq_net::BroadcastProducer;3-use sigil_protocol::{GoqCatalogDocument, MAX_MOQ_CATALOG_BYTES};3+use sigil_protocol::{4+ GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES,5+ SignedMediaGenerationCertificate,6+};47 5-pub(crate) struct GoqCatalogProducer {6- producer: moq_json::Producer<GoqCatalogDocument>,8+pub(crate) enum GoqCatalogProducer {9+ V1(moq_json::Producer<GoqCatalogDocument>),10+ V2(moq_json::Producer<GoqCatalogDocumentV2>),711 }812 913 impl GoqCatalogProducer {10- pub(crate) fn finish(mut self) -> Result<()> {11- self.producer12- .finish()13- .context("finishing Goq catalog track")14+ pub(crate) fn finish(self) -> Result<()> {15+ match self {16+ Self::V1(mut producer) => producer.finish(),17+ Self::V2(mut producer) => producer.finish(),18+ }19+ .context("finishing Goq catalog track")1420 }1521 }1622 @@ -31,7 +37,33 @@ pub(crate) fn publish_goq_catalog(broadcast: &mut BroadcastProducer) -> Result<G
3137 producer3238 .update(&document)3339 .context("publishing immutable Goq catalog snapshot")?;34- Ok(GoqCatalogProducer { producer })40+ Ok(GoqCatalogProducer::V1(producer))41+}42+43+pub(crate) fn publish_goq_catalog_v2(44+ broadcast: &mut BroadcastProducer,45+ generation_id: u64,46+ certificate: &SignedMediaGenerationCertificate,47+) -> Result<GoqCatalogProducer> {48+ let track = broadcast49+ .create_track(hang::Catalog::default_track())50+ .context("creating authenticated catalog.json track")?;51+ let mut producer = moq_json::Producer::new(track, moq_json::Config::default());52+ let document = GoqCatalogDocumentV2::video_h264(generation_id, certificate)53+ .context("constructing authenticated Goq catalog")?;54+ document55+ .validate()56+ .context("validating authenticated Goq catalog document")?;57+ let snapshot =58+ serde_json::to_vec(&document).context("serializing authenticated Goq catalog snapshot")?;59+ ensure!(60+ snapshot.len() <= MAX_MOQ_CATALOG_BYTES,61+ "authenticated Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"62+ );63+ producer64+ .update(&document)65+ .context("publishing immutable authenticated Goq catalog snapshot")?;66+ Ok(GoqCatalogProducer::V2(producer))3567 }3668 3769 @@ -40,6 +72,20 @@ mod tests {
4072 use moq_net::{Broadcast, Track};4173 use std::time::Duration;4274 75+ fn certificate() -> SignedMediaGenerationCertificate {76+ let host = iroh::SecretKey::from_bytes(&[7; 32]);77+ let generation = sigil_protocol::MediaGenerationSigningKey::from_bytes(&[9; 32]);78+ generation79+ .certify(80+ *host.public().as_bytes(),81+ &host.to_bytes(),82+ 42,83+ 1_700_000_000,84+ 1_700_000_600,85+ )86+ .unwrap()87+ }88+4389 4490 async fn catalog_snapshot_is_late_subscribable_and_hang_compatible() {4591 let mut broadcast = Broadcast::new().produce();@@ -77,4 +123,29 @@ mod tests {
77123 assert_eq!(document, GoqCatalogDocument::video_h264());78124 catalog.finish().unwrap();79125 }126+127+ 128+ async fn authenticated_catalog_snapshot_carries_the_certified_generation() {129+ let mut broadcast = Broadcast::new().produce();130+ let _video = broadcast131+ .create_track(Track {132+ name: sigil_protocol::MOQ_VIDEO_H264_TRACK.to_owned(),133+ priority: sigil_protocol::MOQ_VIDEO_TRACK_PRIORITY,134+ })135+ .unwrap();136+ let certificate = certificate();137+ let catalog = publish_goq_catalog_v2(&mut broadcast, 42, &certificate).unwrap();138+ let document_track = broadcast139+ .consume()140+ .subscribe_track(&hang::Catalog::default_track())141+ .unwrap();142+ let mut consumer = moq_json::Consumer::<GoqCatalogDocumentV2>::new(document_track);143+ let document = tokio::time::timeout(Duration::from_millis(100), consumer.next())144+ .await145+ .unwrap()146+ .unwrap()147+ .unwrap();148+ assert_eq!(document.validate().unwrap(), certificate);149+ catalog.finish().unwrap();150+ }80151 }@@ -71,6 +71,7 @@ pub struct ControlV2Handler {
7171 pub sessions: Arc<SessionRegistry>,7272 pub authorization: AuthorizationPolicy,7373 pub input_operations: Arc<InputOperations>,74+ pub host_secret: [u8; 32],7475 }7576 7677 @@ -270,6 +271,7 @@ impl ProtocolHandler for ControlV2Handler {
270271 &self.sessions,271272 &self.authorization,272273 &self.input_operations,274+ self.host_secret,273275 )274276 .await275277 {@@ -19,12 +19,14 @@ use sigil_protocol::{
1919 MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3,2020 MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MOQ_VIDEO_H264_TRACK, MOQ_VIDEO_TRACK_PRIORITY,2121 MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame, MediaFrameHeader,22- MediaObjectHeaderV3, MediaObjectV3, ServerControlEnvelopeV2, encode_media_frame_object,23- media_moq_broadcast_name, read_client_control_v2, read_client_hello, read_client_hello_v2,24- read_input_client_hello_v2, read_input_event, read_input_event_v2,25- read_media_control_request_v3, read_media_feedback_report_v1,26- write_adaptive_bitrate_decision_v1, write_host_hello, write_host_hello_v2, write_input_ack,27- write_input_ack_v2, write_input_host_hello_v2, write_media_object_v3, write_server_control_v2,22+ MediaGenerationSigningKey, MediaObjectCoordinates, MediaObjectHeaderV3, MediaObjectV3,23+ MediaTrack, ServerControlEnvelopeV2, SignedSubscriptionCapability, SubscriptionClaims,24+ SubscriptionTracks, encode_media_frame_object, media_moq_broadcast_name,25+ read_client_control_v2, read_client_hello, read_client_hello_v2, read_input_client_hello_v2,26+ read_input_event, read_input_event_v2, read_media_control_request_v3,27+ read_media_feedback_report_v1, write_adaptive_bitrate_decision_v1, write_host_hello,28+ write_host_hello_v2, write_input_ack, write_input_ack_v2, write_input_host_hello_v2,29+ write_media_object_v3, write_server_control_v2,2830 };2931 use tracing::{debug, error, info, warn};3032 @@ -34,7 +36,7 @@ use crate::clock::SessionClock;
3436 use crate::config::{GamescopeEncoderBackend, HostConfig, VaapiRateControl, VideoSource};3537 use crate::cursor::{PointerPositionTracker, PointerState};3638 use crate::input::{InputBackend, InputDisposition};37-use crate::moq_catalog::publish_goq_catalog;39+use crate::moq_catalog::{publish_goq_catalog, publish_goq_catalog_v2};3840 use crate::source::{3941 EncodedFrame, EncodedGop, EncodedSource, EncoderControl,4042 spawn_gamescope_pipewire_after_static_preflight, spawn_test_pattern,@@ -22,6 +22,12 @@ struct MoqGroupPublisher {
2222 current_group: Option<GroupProducer>,2323 cursor: MediaV3GroupCursor,2424 object_bytes: usize,25+ authentication: Option<MoqObjectAuthentication>,26+}27+28+struct MoqObjectAuthentication {29+ generation_id: u64,30+ signing_key: MediaGenerationSigningKey,2531 }2632 2733 impl MoqGroupPublisher {@@ -31,6 +37,24 @@ impl MoqGroupPublisher {
3137 current_group: None,3238 cursor: MediaV3GroupCursor::default(),3339 object_bytes: 0,40+ authentication: None,41+ }42+ }43+44+ fn authenticated(45+ track: TrackProducer,46+ generation_id: u64,47+ signing_key: MediaGenerationSigningKey,48+ ) -> Self {49+ Self {50+ track,51+ current_group: None,52+ cursor: MediaV3GroupCursor::default(),53+ object_bytes: 0,54+ authentication: Some(MoqObjectAuthentication {55+ generation_id,56+ signing_key,57+ }),3458 }3559 }3660 @@ -50,23 +74,12 @@ impl MoqGroupPublisher {
5074 return Ok(MoqGroupDecision::EnterResync);5175 }5276 };53- let object = encode_media_frame_object(&media_frame_for_encoded(77+ let media_frame = media_frame_for_encoded(5478 config,5579 frame,5680 replay_discontinuity || position.discontinuity,57- )?)?;58- let next_object_bytes = if position.object_id == 0 {59- Some(object.len())60- } else {61- self.object_bytes.checked_add(object.len())62- };63- let Some(next_object_bytes) =64- next_object_bytes.filter(|bytes| *bytes <= MAX_MEDIA_GROUP_BYTES_V3)65- else {66- self.cursor.request_keyframe();67- self.abort_current();68- return Ok(MoqGroupDecision::EnterResync);69- };81+ )?;82+ let payload = encode_media_frame_object(&media_frame)?;7083 7184 if position.object_id == 0 {7285 // A new independently-decodable GOP supersedes the previous one.@@ -78,10 +91,22 @@ impl MoqGroupPublisher {
7891 .append_group()7992 .context("creating sequential MoQ video group")?;8093 let group_id = group.sequence;94+ let object = self.authenticate_object(95+ group_id,96+ position.object_id,97+ u16::from(media_frame.header.flags.bits()),98+ payload,99+ )?;100+ let object_len = object.len();101+ if object.len() > MAX_MEDIA_GROUP_BYTES_V3 {102+ let _ = group.abort(MoqError::Cancel);103+ self.cursor.request_keyframe();104+ return Ok(MoqGroupDecision::EnterResync);105+ }81106 group82107 .write_frame(object)83108 .context("writing configured keyframe to MoQ group")?;84- self.object_bytes = next_object_bytes;109+ self.object_bytes = object_len;85110 self.current_group = Some(group);86111 return Ok(MoqGroupDecision::Published {87112 group_id,@@ -90,11 +115,30 @@ impl MoqGroupPublisher {
90115 });91116 }92117 118+ let group_id = self119+ .current_group120+ .as_ref()121+ .map(|group| group.sequence)122+ .context("MoQ delta frame has no active configured-keyframe group")?;123+ let object = self.authenticate_object(124+ group_id,125+ position.object_id,126+ u16::from(media_frame.header.flags.bits()),127+ payload,128+ )?;129+ let Some(next_object_bytes) = self130+ .object_bytes131+ .checked_add(object.len())132+ .filter(|bytes| *bytes <= MAX_MEDIA_GROUP_BYTES_V3)133+ else {134+ self.cursor.request_keyframe();135+ self.abort_current();136+ return Ok(MoqGroupDecision::EnterResync);137+ };93138 let group = self94139 .current_group95140 .as_mut()96- .context("MoQ delta frame has no active configured-keyframe group")?;97- let group_id = group.sequence;141+ .context("MoQ delta frame lost its configured-keyframe group")?;98142 group99143 .write_frame(object)100144 .context("writing delta access unit to MoQ group")?;@@ -106,6 +150,32 @@ impl MoqGroupPublisher {
106150 })107151 }108152 153+ fn authenticate_object(154+ &self,155+ group_id: u64,156+ object_id: u32,157+ flags: u16,158+ payload: Vec<u8>,159+ ) -> Result<Vec<u8>> {160+ let Some(authentication) = &self.authentication else {161+ return Ok(payload);162+ };163+ authentication164+ .signing_key165+ .authenticate(166+ MediaObjectCoordinates {167+ generation_id: authentication.generation_id,168+ track: MediaTrack::VideoH264,169+ group_id,170+ object_id,171+ flags,172+ },173+ &payload,174+ )175+ .map(|object| object.into_bytes())176+ .map_err(Into::into)177+ }178+109179 fn request_keyframe(&mut self) -> Option<u64> {110180 self.cursor.request_keyframe();111181 self.abort_current()@@ -331,6 +401,7 @@ pub(super) async fn serve_control_moq(
331401 &mut broadcast,332402 encoder_control,333403 Arc::clone(&lease.media_v3_telemetry),404+ None,334405 )335406 .await;336407 let catalog_result = catalog.finish();@@ -352,6 +423,7 @@ pub(super) async fn serve_control_moq_v2(
352423 sessions: &Arc<SessionRegistry>,353424 authorization: &AuthorizationPolicy,354425 input_operations: &Arc<InputOperations>,426+ host_secret: [u8; 32],355427 ) -> Result<()> {356428 let remote = connection.remote_id();357429 let handshake_permit = sessions@@ -423,6 +495,39 @@ pub(super) async fn serve_control_moq_v2(
423495 let source_task = SourceTaskGuard::new(source_task);424496 sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;425497 498+ let mut generation_secret = [0_u8; 32];499+ getrandom::fill(&mut generation_secret)500+ .context("generating ephemeral media authentication key")?;501+ let generation_signing_key = MediaGenerationSigningKey::from_bytes(&generation_secret);502+ generation_secret.fill(0);503+ let issued_at_unix = unix_timestamp_now()?;504+ let certificate = generation_signing_key.certify(505+ *iroh::SecretKey::from_bytes(&host_secret)506+ .public()507+ .as_bytes(),508+ &host_secret,509+ lease.session_id,510+ issued_at_unix,511+ issued_at_unix.saturating_add(60 * 60),512+ )?;513+ let mut subscription_nonce = [0_u8; 32];514+ getrandom::fill(&mut subscription_nonce).context("generating subscription capability nonce")?;515+ let subscription_capability = SignedSubscriptionCapability::issue(516+ SubscriptionClaims::new(517+ certificate.claims.host_node_id,518+ lease.session_id,519+ *remote.as_bytes(),520+ SubscriptionTracks::VIDEO_H264,521+ 1,522+ issued_at_unix,523+ issued_at_unix.saturating_add(15 * 60),524+ subscription_nonce,525+ 1,526+ )?,527+ &host_secret,528+ )?;529+ let subscription_token = subscription_capability.encode();530+426531 let mut broadcast = Broadcast::new().produce();427532 let track = broadcast428533 .create_track(Track {@@ -430,13 +535,14 @@ pub(super) async fn serve_control_moq_v2(
430535 priority: MOQ_VIDEO_TRACK_PRIORITY,431536 })432537 .context("creating static MoQ H.264 track for control v2")?;433- let catalog = publish_goq_catalog(&mut broadcast)?;538+ let catalog = publish_goq_catalog_v2(&mut broadcast, lease.session_id, &certificate)?;434539 let broadcast_name = media_moq_broadcast_name(lease.session_id)?;435- let attachment = sessions.expect_moq(540+ let attachment = sessions.expect_moq_v2(436541 remote,437542 lease.session_id,438543 broadcast_name.clone(),439544 broadcast.consume(),545+ subscription_capability,440546 )?;441547 442548 let negotiated = MEDIA_CAPABILITIES@@ -445,7 +551,8 @@ pub(super) async fn serve_control_moq_v2(
445551 .filter(|capability| hello.capabilities.contains(capability))446552 .collect();447553 let mut control_hello =448- HostHelloV2::accepted(lease.session_id, negotiated, lease.initial_snapshot.clone());554+ HostHelloV2::accepted(lease.session_id, negotiated, lease.initial_snapshot.clone())555+ .with_media_subscription_capability(subscription_token);449556 if let Some(dimensions) = pointer_surface_dimensions {450557 control_hello = control_hello.with_pointer_surface_dimensions(dimensions);451558 }@@ -487,6 +594,7 @@ pub(super) async fn serve_control_moq_v2(
487594 &mut broadcast,488595 encoder_control,489596 Arc::clone(&lease.media_v3_telemetry),597+ Some((lease.session_id, generation_signing_key)),490598 )491599 .await;492600 let catalog_result = catalog.finish();@@ -604,10 +712,16 @@ async fn run_control_moq_session(
604712 broadcast: &mut BroadcastProducer,605713 encoder_control: Option<EncoderControl>,606714 telemetry: Arc<MediaV3Telemetry>,715+ authentication: Option<(u64, MediaGenerationSigningKey)>,607716 ) -> Result<()> {608717 let maximum_replay_age = maximum_media_replay_age(config.framerate);609718 let mut replay_cursor = MediaReplayCursor::default();610- let mut publisher = MoqGroupPublisher::new(track);719+ let mut publisher = match authentication {720+ Some((generation_id, signing_key)) => {721+ MoqGroupPublisher::authenticated(track, generation_id, signing_key)722+ }723+ None => MoqGroupPublisher::new(track),724+ };611725 let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);612726 let mut control_task = match control_reader {613727 MoqControlReader::V1(control_recv) => tokio::spawn(forward_media_v3_control_requests(@@ -841,6 +955,29 @@ mod tests {
841955 use super::super::{media_v3_encoded_frame, moq_test_config};842956 use super::*;843957 958+ fn authenticated_publisher(959+ track: TrackProducer,960+ ) -> (961+ MoqGroupPublisher,962+ sigil_protocol::SignedMediaGenerationCertificate,963+ ) {964+ let host = iroh::SecretKey::from_bytes(&[7; 32]);965+ let signing_key = MediaGenerationSigningKey::from_bytes(&[9; 32]);966+ let certificate = signing_key967+ .certify(968+ *host.public().as_bytes(),969+ &host.to_bytes(),970+ 42,971+ 1_700_000_000,972+ 1_700_000_600,973+ )974+ .unwrap();975+ (976+ MoqGroupPublisher::authenticated(track, 42, signing_key),977+ certificate,978+ )979+ }980+844981 845982 async fn upstream_moq_groups_are_sequential_and_cancel_the_superseded_gop() {846983 let track_info = Track {@@ -902,6 +1039,42 @@ mod tests {
9021039 assert!(frame.header.flags.contains(FrameFlags::DISCONTINUITY));9031040 }9041041 1042+ 1043+ async fn control_v2_objects_are_signed_after_final_moq_coordinates_are_known() {1044+ let track_info = Track {1045+ name: MOQ_VIDEO_H264_TRACK.to_owned(),1046+ priority: MOQ_VIDEO_TRACK_PRIORITY,1047+ };1048+ let mut broadcast = Broadcast::new().produce();1049+ let track = broadcast.create_track(track_info.clone()).unwrap();1050+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1051+ let (mut publisher, certificate) = authenticated_publisher(track);1052+ let config = moq_test_config();1053+ let encoded = media_v3_encoded_frame(100, true, true, 4);1054+ publisher.publish(&config, &encoded, false).unwrap();1055+ let mut group = consumer.recv_group().await.unwrap().unwrap();1056+ let object = group.read_frame().await.unwrap().unwrap();1057+ let payload = sigil_protocol::AuthenticatedMediaObject::verify(1058+ &object,1059+ &certificate,1060+ MediaObjectCoordinates {1061+ generation_id: 42,1062+ track: MediaTrack::VideoH264,1063+ group_id: group.sequence,1064+ object_id: 0,1065+ flags: u16::from(FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG).bits()),1066+ },1067+ )1068+ .unwrap();1069+ assert_eq!(1070+ sigil_protocol::decode_media_frame_object(payload)1071+ .unwrap()1072+ .header1073+ .sequence,1074+ 1001075+ );1076+ }1077+9051078 9061079 async fn upstream_moq_late_join_preserves_active_static_group() {9071080 let track_info = Track {@@ -1,6 +1,7 @@
11 use std::sync::atomic::{AtomicU64, Ordering};22 use std::sync::{Arc, Mutex};33 use std::time::{Duration, Instant};4+use std::time::{SystemTime, UNIX_EPOCH};45 56 use anyhow::{Context, Result, bail, ensure};67 use iroh::EndpointId;@@ -9,7 +10,8 @@ use moq_net::BroadcastConsumer;
910 use sigil_protocol::{1011 ControllerSlot, FocusCommandActionV2, FocusCommandV2, FocusStateV2, FocusTransitionReasonV2,1112 InvitationGrants, KeyframeRequestReasonV3, MediaGenerationDescriptorV2, SessionSnapshotV2,12- ViewerPresenceId, ViewerPresenceV2, media_moq_broadcast_name,13+ SignedSubscriptionCapability, SubscriptionTracks, ViewerPresenceId, ViewerPresenceV2,14+ media_moq_broadcast_name,1315 };1416 use tracing::{debug, warn};1517 @@ -39,6 +41,7 @@ struct PendingMoqAttachment {
3941 attached: tokio::sync::oneshot::Sender<()>,4042 closed: tokio::sync::oneshot::Sender<()>,4143 telemetry: Arc<MediaV3Telemetry>,44+ subscription_capability: Option<SignedSubscriptionCapability>,4245 }4346 4447 impl std::fmt::Debug for PendingMoqAttachment {@@ -888,6 +891,59 @@ impl SessionRegistry {
888891 attached,889892 closed,890893 telemetry,894+ subscription_capability: None,895+ });896+ Ok(MoqAttachmentWait {897+ attached: attached_rx,898+ closed: closed_rx,899+ })900+ }901+902+ pub(super) fn expect_moq_v2(903+ &self,904+ remote: EndpointId,905+ session_id: u64,906+ broadcast_name: String,907+ broadcast: BroadcastConsumer,908+ subscription_capability: SignedSubscriptionCapability,909+ ) -> Result<MoqAttachmentWait> {910+ ensure!(911+ subscription_capability.claims.media_generation_id == session_id912+ && subscription_capability.claims.subscriber_endpoint_id == *remote.as_bytes()913+ && subscription_capability914+ .claims915+ .tracks916+ .contains(SubscriptionTracks::VIDEO_H264)917+ && subscription_capability.claims.authorization_revision == 1,918+ "subscription capability does not match the v2 viewer attachment"919+ );920+ let active = self.active.lock().expect("session registry poisoned");921+ let telemetry = active922+ .as_ref()923+ .filter(|session| {924+ session.mode == SessionMode::V2Single925+ && session.media_active926+ && session.remote == remote927+ && session.session_id == session_id928+ })929+ .map(|session| Arc::clone(&session.media_v3_telemetry))930+ .context("authenticated MoQ expectation does not match the active v2 session")?;931+ let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");932+ ensure!(933+ pending.is_none(),934+ "active control session already expects MoQ"935+ );936+ let (attached, attached_rx) = tokio::sync::oneshot::channel();937+ let (closed, closed_rx) = tokio::sync::oneshot::channel();938+ *pending = Some(PendingMoqAttachment {939+ remote,940+ session_id,941+ broadcast_name,942+ broadcast,943+ attached,944+ closed,945+ telemetry,946+ subscription_capability: Some(subscription_capability),891947 });892948 Ok(MoqAttachmentWait {893949 attached: attached_rx,@@ -896,6 +952,14 @@ impl SessionRegistry {
896952 }897953 898954 pub(super) fn claim_moq(&self, remote: EndpointId) -> Result<ClaimedMoqAttachment> {955+ let now_unix = SystemTime::now()956+ .duration_since(UNIX_EPOCH)957+ .context("system clock is before the Unix epoch")?958+ .as_secs();959+ self.claim_moq_at(remote, now_unix)960+ }961+962+ fn claim_moq_at(&self, remote: EndpointId, now_unix: u64) -> Result<ClaimedMoqAttachment> {899963 let active = self.active.lock().expect("session registry poisoned");900964 let session = active901965 .as_ref()@@ -908,6 +972,27 @@ impl SessionRegistry {
908972 attachment.remote == remote && attachment.session_id == session.session_id909973 })910974 .context("active control session is not expecting a MoQ connection")?;975+ if session.mode == SessionMode::V2Single {976+ let capability = attachment977+ .subscription_capability978+ .as_ref()979+ .context("v2 MoQ attachment has no subscription capability")?;980+ capability981+ .verify_binding(982+ capability.claims.host_node_id,983+ session.session_id,984+ *remote.as_bytes(),985+ SubscriptionTracks::VIDEO_H264,986+ 1,987+ now_unix,988+ )989+ .context("verifying endpoint-bound MoQ subscription capability")?;990+ } else {991+ ensure!(992+ attachment.subscription_capability.is_none(),993+ "legacy MoQ attachment unexpectedly carries a subscription capability"994+ );995+ }911996 debug_assert_eq!(attachment.remote, session.remote);912997 let attachment = pending913998 .take()@@ -1377,6 +1462,31 @@ mod tests {
13771462 (producer, consumer)13781463 }13791464 1465+ fn test_subscription(1466+ host: &iroh::SecretKey,1467+ remote: EndpointId,1468+ generation_id: u64,1469+ issued_at_unix: u64,1470+ expires_at_unix: u64,1471+ ) -> SignedSubscriptionCapability {1472+ SignedSubscriptionCapability::issue(1473+ sigil_protocol::SubscriptionClaims::new(1474+ *host.public().as_bytes(),1475+ generation_id,1476+ *remote.as_bytes(),1477+ SubscriptionTracks::VIDEO_H264,1478+ 1,1479+ issued_at_unix,1480+ expires_at_unix,1481+ [7; 32],1482+ 1,1483+ )1484+ .unwrap(),1485+ &host.to_bytes(),1486+ )1487+ .unwrap()1488+ }1489+13801490 13811491 fn moq_attachment_requires_exact_active_control_remote_and_is_single_use() {13821492 let sessions = Arc::new(SessionRegistry::default());@@ -1406,6 +1516,47 @@ mod tests {
14061516 assert!(sessions.claim_moq(endpoint(1)).is_err());14071517 }14081518 1519+ 1520+ fn v2_moq_attachment_requires_a_fresh_endpoint_bound_subscription() {1521+ let host = iroh::SecretKey::from_bytes(&[9; 32]);1522+ let remote = endpoint(1);1523+ let sessions = Arc::new(SessionRegistry::default());1524+ let lease = sessions1525+ .claim_v2(remote, [1; 16], InvitationGrants::VIEW)1526+ .unwrap();1527+ let capability = test_subscription(&host, remote, lease.session_id, 100, 200);1528+ let (_producer, consumer) = test_moq_broadcast();1529+ let _wait = sessions1530+ .expect_moq_v2(1531+ remote,1532+ lease.session_id,1533+ media_moq_broadcast_name(lease.session_id).unwrap(),1534+ consumer,1535+ capability,1536+ )1537+ .unwrap();1538+ assert!(sessions.claim_moq_at(endpoint(2), 150).is_err());1539+ assert!(sessions.claim_moq_at(remote, 150).is_ok());1540+ assert!(sessions.claim_moq_at(remote, 150).is_err());1541+1542+ let expired_sessions = Arc::new(SessionRegistry::default());1543+ let expired_lease = expired_sessions1544+ .claim_v2(remote, [2; 16], InvitationGrants::VIEW)1545+ .unwrap();1546+ let expired = test_subscription(&host, remote, expired_lease.session_id, 100, 200);1547+ let (_producer, consumer) = test_moq_broadcast();1548+ let _wait = expired_sessions1549+ .expect_moq_v2(1550+ remote,1551+ expired_lease.session_id,1552+ media_moq_broadcast_name(expired_lease.session_id).unwrap(),1553+ consumer,1554+ expired,1555+ )1556+ .unwrap();1557+ assert!(expired_sessions.claim_moq_at(remote, 201).is_err());1558+ }1559+14091560 14101561 fn competing_moq_connections_cannot_both_claim_one_control_attachment() {14111562 let sessions = Arc::new(SessionRegistry::default());@@ -7,6 +7,7 @@ use crate::framing::{read_json, write_json};
77 use crate::{88 Capability, ControllerSlot, KeyframeRequestReasonV3, MAX_INVITATION_TOKEN_LEN,99 PROTOCOL_VERSION_V2, PointerSurfaceDimensions, ProtocolError, Result, SignedInvitation,10+ SignedSubscriptionCapability,1011 };1112 1213 const MAX_AGENT_LEN: usize = 128;@@ -473,6 +474,8 @@ pub struct HostHelloV2 {
473474 pub pointer_surface_dimensions: Option<PointerSurfaceDimensions>,474475 475476 pub snapshot: Option<SessionSnapshotV2>,477+ 478+ pub media_subscription_capability: Option<String>,476479 }477480 478481 impl HostHelloV2 {@@ -489,6 +492,7 @@ impl HostHelloV2 {
489492 message: None,490493 pointer_surface_dimensions: None,491494 snapshot: Some(snapshot),495+ media_subscription_capability: None,492496 }493497 }494498 @@ -501,6 +505,7 @@ impl HostHelloV2 {
501505 message: Some(message.into()),502506 pointer_surface_dimensions: None,503507 snapshot: None,508+ media_subscription_capability: None,504509 }505510 }506511 @@ -509,6 +514,11 @@ impl HostHelloV2 {
509514 self510515 }511516 517+ pub fn with_media_subscription_capability(mut self, capability: impl Into<String>) -> Self {518+ self.media_subscription_capability = Some(capability.into());519+ self520+ }521+512522 pub fn validate(&self) -> Result<()> {513523 validate_version(self.version)?;514524 validate_capabilities(&self.capabilities)?;@@ -536,8 +546,18 @@ impl HostHelloV2 {
536546 "snapshot session does not match host hello",537547 );538548 }549+ if let Some(capability) = &self.media_subscription_capability {550+ let capability = SignedSubscriptionCapability::decode(capability)?;551+ if capability.claims.media_generation_id != snapshot.media.generation_id {552+ return invalid(553+ "host hello v2",554+ "subscription capability generation does not match the snapshot",555+ );556+ }557+ }539558 } else if self.session_id.is_some()540559 || self.snapshot.is_some()560+ || self.media_subscription_capability.is_some()541561 || self.message.as_ref().is_none_or(|message| {542562 message.is_empty() || message.len() > MAX_REJECTION_MESSAGE_LEN543563 })@@ -15,8 +15,10 @@ mod input;
1515 mod input_v2;1616 mod invitation;1717 mod media;18+mod media_auth;1819 mod media_v3;1920 mod moq_catalog;21+mod subscription;2022 2123 pub use audio::{AUDIO_HEADER_LEN, AudioCodec, AudioFlags, AudioPacket, AudioPacketHeader};2224 pub use control_v2::{@@ -59,6 +61,12 @@ pub use media::{
5961 FrameFlags, MEDIA_HEADER_LEN, MediaCodec, MediaFrame, MediaFrameHeader,6062 decode_media_frame_object, encode_media_frame_object,6163 };64+pub use media_auth::{65+ AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN, AuthenticatedMediaObject,66+ MAX_MEDIA_GENERATION_CERTIFICATE_TOKEN_LEN, MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX,67+ MediaAuthMode, MediaGenerationCertificateClaims, MediaGenerationSigningKey,68+ MediaObjectCoordinates, MediaTrack, SignedMediaGenerationCertificate,69+};6270 pub use media_v3::{6371 KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,6472 MAX_MEDIA_OBJECT_ID_V3, MEDIA_CONTROL_REQUEST_V3_LEN, MEDIA_OBJECT_V3_HEADER_LEN,@@ -67,9 +75,16 @@ pub use media_v3::{
6775 write_media_control_request_v3, write_media_object_v3,6876 };6977 pub use moq_catalog::{70- GoqCatalogDocument, MAX_MOQ_CATALOG_BYTES, MOQ_CATALOG_EXTENSION_VERSION_V1,71- MOQ_GOP_GROUP_FORMAT_V1, MOQ_MEDIA_OBJECT_FORMAT_V1, MOQ_VIDEO_TRACK_PRIORITY,72- MoqCatalogExtensionV1, MoqTrackDescriptorV1, MoqVideoCatalogV1,78+ GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES,79+ MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1, MOQ_CATALOG_EXTENSION_VERSION_V1,80+ MOQ_CATALOG_EXTENSION_VERSION_V2, MOQ_GOP_GROUP_FORMAT_V1, MOQ_MEDIA_OBJECT_FORMAT_V1,81+ MOQ_VIDEO_TRACK_PRIORITY, MoqCatalogExtensionV1, MoqCatalogExtensionV2,82+ MoqTrackAuthenticationV2, MoqTrackDescriptorV1, MoqVideoCatalogV1, MoqVideoCatalogV2,83+};84+pub use subscription::{85+ MAX_SUBSCRIPTION_CAPABILITY_TOKEN_LEN, MAX_SUBSCRIPTION_TTL_SECS,86+ SUBSCRIPTION_CAPABILITY_TOKEN_PREFIX, SignedSubscriptionCapability, SubscriptionClaims,87+ SubscriptionTracks,7388 };7489 7590 /// Protocol version encoded in v1 messages.@@ -0,0 +1,689 @@
1+use base64::Engine as _;2+use base64::engine::general_purpose::URL_SAFE_NO_PAD;3+use ed25519_dalek::{Signature, Signer as _, SigningKey, VerifyingKey};4+use sha2::{Digest as _, Sha256};5+6+use crate::{MAX_MEDIA_PAYLOAD_LEN, MEDIA_HEADER_LEN, ProtocolError, Result};7+8+const CERTIFICATE_MAGIC: [u8; 4] = *b"SGMC";9+const CERTIFICATE_VERSION: u16 = 1;10+const CERTIFICATE_CLAIMS_LEN: usize = 96;11+const SIGNATURE_LEN: usize = 64;12+const CERTIFICATE_LEN: usize = CERTIFICATE_CLAIMS_LEN + SIGNATURE_LEN;13+const CERTIFICATE_DOMAIN: &[u8] = b"goq.sh/sigil-media-certificate/v1\0";14+const OBJECT_MAGIC: [u8; 4] = *b"SGMA";15+const OBJECT_VERSION: u16 = 1;16+const OBJECT_UNSIGNED_HEADER_LEN: usize = 70;17+const OBJECT_DOMAIN: &[u8] = b"goq.sh/sigil-media-object/v1\0";18+const MAX_GENERATION_CERTIFICATE_TTL_SECS: u64 = 24 * 60 * 60;19+20+pub const MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX: &str = "goq-media-cert-v1.";21+pub const MAX_MEDIA_GENERATION_CERTIFICATE_TOKEN_LEN: usize = 256;22+pub const AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN: usize = OBJECT_UNSIGNED_HEADER_LEN + SIGNATURE_LEN;23+const MAX_AUTHENTICATED_PAYLOAD_LEN: usize = MEDIA_HEADER_LEN + MAX_MEDIA_PAYLOAD_LEN;24+25+26+27+pub enum MediaAuthMode {28+ Ed25519 = 1,29+}30+31+impl MediaAuthMode {32+ pub const fn label(self) -> &'static str {33+ match self {34+ Self::Ed25519 => "ed25519-v1",35+ }36+ }37+}38+39+impl TryFrom<u8> for MediaAuthMode {40+ type Error = ProtocolError;41+42+ fn try_from(value: u8) -> Result<Self> {43+ match value {44+ 1 => Ok(Self::Ed25519),45+ _ => invalid(46+ "media authentication mode",47+ "unsupported authentication mode",48+ ),49+ }50+ }51+}52+53+54+55+pub enum MediaTrack {56+ VideoH264 = 1,57+ AudioOpus = 2,58+}59+60+impl MediaTrack {61+ pub const fn name(self) -> &'static str {62+ match self {63+ Self::VideoH264 => crate::MOQ_VIDEO_H264_TRACK,64+ Self::AudioOpus => "audio/opus",65+ }66+ }67+}68+69+impl TryFrom<u8> for MediaTrack {70+ type Error = ProtocolError;71+72+ fn try_from(value: u8) -> Result<Self> {73+ match value {74+ 1 => Ok(Self::VideoH264),75+ 2 => Ok(Self::AudioOpus),76+ _ => invalid("authenticated media object", "unsupported media track"),77+ }78+ }79+}80+81+82+pub struct MediaGenerationCertificateClaims {83+ pub host_node_id: [u8; 32],84+ pub generation_id: u64,85+ pub generation_signing_key: [u8; 32],86+ pub issued_at_unix: u64,87+ pub expires_at_unix: u64,88+ pub authentication_mode: MediaAuthMode,89+}90+91+impl MediaGenerationCertificateClaims {92+ pub fn new(93+ host_node_id: [u8; 32],94+ generation_id: u64,95+ generation_signing_key: [u8; 32],96+ issued_at_unix: u64,97+ expires_at_unix: u64,98+ ) -> Result<Self> {99+ let claims = Self {100+ host_node_id,101+ generation_id,102+ generation_signing_key,103+ issued_at_unix,104+ expires_at_unix,105+ authentication_mode: MediaAuthMode::Ed25519,106+ };107+ claims.validate()?;108+ Ok(claims)109+ }110+111+ pub fn validate(&self) -> Result<()> {112+ if self.authentication_mode != MediaAuthMode::Ed25519 {113+ return invalid(114+ "media generation certificate",115+ "unsupported authentication mode",116+ );117+ }118+ if self.generation_id == 0 {119+ return invalid(120+ "media generation certificate",121+ "generation id must be non-zero",122+ );123+ }124+ if self.issued_at_unix >= self.expires_at_unix125+ || self.expires_at_unix - self.issued_at_unix > MAX_GENERATION_CERTIFICATE_TTL_SECS126+ {127+ return invalid(128+ "media generation certificate",129+ "certificate lifetime must be within 1..=86400 seconds",130+ );131+ }132+ VerifyingKey::from_bytes(&self.host_node_id).map_err(|_| {133+ ProtocolError::InvalidMessage {134+ message_type: "media generation certificate",135+ reason: "host node id is not an Ed25519 public key",136+ }137+ })?;138+ VerifyingKey::from_bytes(&self.generation_signing_key).map_err(|_| {139+ ProtocolError::InvalidMessage {140+ message_type: "media generation certificate",141+ reason: "generation key is not an Ed25519 public key",142+ }143+ })?;144+ Ok(())145+ }146+147+ fn encode(&self) -> [u8; CERTIFICATE_CLAIMS_LEN] {148+ let mut bytes = [0_u8; CERTIFICATE_CLAIMS_LEN];149+ bytes[0..4].copy_from_slice(&CERTIFICATE_MAGIC);150+ bytes[4..6].copy_from_slice(&CERTIFICATE_VERSION.to_be_bytes());151+ bytes[6] = self.authentication_mode as u8;152+ bytes[8..40].copy_from_slice(&self.host_node_id);153+ bytes[40..48].copy_from_slice(&self.generation_id.to_be_bytes());154+ bytes[48..80].copy_from_slice(&self.generation_signing_key);155+ bytes[80..88].copy_from_slice(&self.issued_at_unix.to_be_bytes());156+ bytes[88..96].copy_from_slice(&self.expires_at_unix.to_be_bytes());157+ bytes158+ }159+160+ fn signing_message(&self) -> Vec<u8> {161+ let mut message = Vec::with_capacity(CERTIFICATE_DOMAIN.len() + CERTIFICATE_CLAIMS_LEN);162+ message.extend_from_slice(CERTIFICATE_DOMAIN);163+ message.extend_from_slice(&self.encode());164+ message165+ }166+}167+168+169+pub struct SignedMediaGenerationCertificate {170+ pub claims: MediaGenerationCertificateClaims,171+ signature: [u8; SIGNATURE_LEN],172+}173+174+impl SignedMediaGenerationCertificate {175+ pub fn issue(claims: MediaGenerationCertificateClaims, host_secret: &[u8; 32]) -> Result<Self> {176+ claims.validate()?;177+ let signing_key = SigningKey::from_bytes(host_secret);178+ if signing_key.verifying_key().to_bytes() != claims.host_node_id {179+ return invalid(180+ "media generation certificate",181+ "host signing key does not match the certificate host",182+ );183+ }184+ let signature = signing_key.sign(&claims.signing_message()).to_bytes();185+ Ok(Self { claims, signature })186+ }187+188+ pub fn verify(&self) -> Result<()> {189+ self.claims.validate()?;190+ let key = VerifyingKey::from_bytes(&self.claims.host_node_id).map_err(|_| {191+ ProtocolError::InvalidMessage {192+ message_type: "media generation certificate",193+ reason: "invalid host verification key",194+ }195+ })?;196+ key.verify_strict(197+ &self.claims.signing_message(),198+ &Signature::from_bytes(&self.signature),199+ )200+ .map_err(|_| ProtocolError::InvalidMessage {201+ message_type: "media generation certificate",202+ reason: "host signature verification failed",203+ })204+ }205+206+ pub fn verify_binding(207+ &self,208+ expected_host: [u8; 32],209+ expected_generation: u64,210+ now_unix: u64,211+ ) -> Result<()> {212+ self.verify()?;213+ if self.claims.host_node_id != expected_host {214+ return invalid(215+ "media generation certificate",216+ "certificate is bound to another host",217+ );218+ }219+ if self.claims.generation_id != expected_generation {220+ return invalid(221+ "media generation certificate",222+ "certificate is bound to another generation",223+ );224+ }225+ if now_unix < self.claims.issued_at_unix || now_unix > self.claims.expires_at_unix {226+ return invalid(227+ "media generation certificate",228+ "certificate is not currently valid",229+ );230+ }231+ Ok(())232+ }233+234+ pub fn encode(&self) -> String {235+ let mut bytes = Vec::with_capacity(CERTIFICATE_LEN);236+ bytes.extend_from_slice(&self.claims.encode());237+ bytes.extend_from_slice(&self.signature);238+ format!(239+ "{MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX}{}",240+ URL_SAFE_NO_PAD.encode(bytes)241+ )242+ }243+244+ pub fn decode(token: &str) -> Result<Self> {245+ if token.len() > MAX_MEDIA_GENERATION_CERTIFICATE_TOKEN_LEN246+ || token.contains('=')247+ || token.bytes().any(|byte| byte.is_ascii_whitespace())248+ {249+ return invalid(250+ "media generation certificate",251+ "certificate token is oversized or non-canonical",252+ );253+ }254+ let encoded = token255+ .strip_prefix(MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX)256+ .ok_or(ProtocolError::InvalidMessage {257+ message_type: "media generation certificate",258+ reason: "certificate token prefix is invalid",259+ })?;260+ let bytes = URL_SAFE_NO_PAD261+ .decode(encoded)262+ .map_err(|_| ProtocolError::InvalidMessage {263+ message_type: "media generation certificate",264+ reason: "certificate token is not canonical base64url",265+ })?;266+ if bytes.len() != CERTIFICATE_LEN {267+ return Err(ProtocolError::InvalidMessageLength {268+ actual: bytes.len(),269+ maximum: CERTIFICATE_LEN,270+ });271+ }272+ if bytes[0..4] != CERTIFICATE_MAGIC {273+ return invalid(274+ "media generation certificate",275+ "certificate magic is invalid",276+ );277+ }278+ let version = u16::from_be_bytes(bytes[4..6].try_into().expect("fixed certificate"));279+ if version != CERTIFICATE_VERSION {280+ return Err(ProtocolError::UnsupportedVersion {281+ expected: CERTIFICATE_VERSION,282+ actual: version,283+ });284+ }285+ if bytes[7] != 0 {286+ return invalid(287+ "media generation certificate",288+ "reserved fields must be zero",289+ );290+ }291+ let claims = MediaGenerationCertificateClaims {292+ authentication_mode: MediaAuthMode::try_from(bytes[6])?,293+ host_node_id: bytes[8..40].try_into().expect("fixed certificate"),294+ generation_id: u64::from_be_bytes(bytes[40..48].try_into().expect("fixed certificate")),295+ generation_signing_key: bytes[48..80].try_into().expect("fixed certificate"),296+ issued_at_unix: u64::from_be_bytes(297+ bytes[80..88].try_into().expect("fixed certificate"),298+ ),299+ expires_at_unix: u64::from_be_bytes(300+ bytes[88..96].try_into().expect("fixed certificate"),301+ ),302+ };303+ let certificate = Self {304+ claims,305+ signature: bytes[CERTIFICATE_CLAIMS_LEN..]306+ .try_into()307+ .expect("fixed certificate signature"),308+ };309+ certificate.verify()?;310+ if certificate.encode() != token {311+ return invalid(312+ "media generation certificate",313+ "certificate encoding is not canonical",314+ );315+ }316+ Ok(certificate)317+ }318+}319+320+pub struct MediaGenerationSigningKey(SigningKey);321+322+impl std::fmt::Debug for MediaGenerationSigningKey {323+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {324+ formatter.write_str("MediaGenerationSigningKey([REDACTED])")325+ }326+}327+328+impl MediaGenerationSigningKey {329+ pub fn from_bytes(secret: &[u8; 32]) -> Self {330+ Self(SigningKey::from_bytes(secret))331+ }332+333+ pub fn public_key(&self) -> [u8; 32] {334+ self.0.verifying_key().to_bytes()335+ }336+337+ pub fn certify(338+ &self,339+ host_node_id: [u8; 32],340+ host_secret: &[u8; 32],341+ generation_id: u64,342+ issued_at_unix: u64,343+ expires_at_unix: u64,344+ ) -> Result<SignedMediaGenerationCertificate> {345+ SignedMediaGenerationCertificate::issue(346+ MediaGenerationCertificateClaims::new(347+ host_node_id,348+ generation_id,349+ self.public_key(),350+ issued_at_unix,351+ expires_at_unix,352+ )?,353+ host_secret,354+ )355+ }356+357+ pub fn authenticate(358+ &self,359+ coordinates: MediaObjectCoordinates,360+ payload: &[u8],361+ ) -> Result<AuthenticatedMediaObject> {362+ coordinates.validate(payload.len())?;363+ let digest: [u8; 32] = Sha256::digest(payload).into();364+ let unsigned = encode_unsigned_object_header(coordinates, payload.len(), digest)?;365+ let mut message = Vec::with_capacity(OBJECT_DOMAIN.len() + unsigned.len());366+ message.extend_from_slice(OBJECT_DOMAIN);367+ message.extend_from_slice(&unsigned);368+ let signature = self.0.sign(&message).to_bytes();369+ let mut bytes = Vec::with_capacity(AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN + payload.len());370+ bytes.extend_from_slice(&unsigned);371+ bytes.extend_from_slice(&signature);372+ bytes.extend_from_slice(payload);373+ Ok(AuthenticatedMediaObject { bytes })374+ }375+}376+377+378+pub struct MediaObjectCoordinates {379+ pub generation_id: u64,380+ pub track: MediaTrack,381+ pub group_id: u64,382+ pub object_id: u32,383+ pub flags: u16,384+}385+386+impl MediaObjectCoordinates {387+ fn validate(&self, payload_len: usize) -> Result<()> {388+ if self.generation_id == 0 {389+ return invalid(390+ "authenticated media object",391+ "generation id must be non-zero",392+ );393+ }394+ if self.flags & !0x0007 != 0 {395+ return invalid("authenticated media object", "unknown media flags are set");396+ }397+ if payload_len == 0 || payload_len > MAX_AUTHENTICATED_PAYLOAD_LEN {398+ return Err(ProtocolError::InvalidMediaPayloadLength {399+ actual: payload_len,400+ maximum: MAX_AUTHENTICATED_PAYLOAD_LEN,401+ });402+ }403+ u32::try_from(payload_len).map_err(|_| ProtocolError::InvalidMediaPayloadLength {404+ actual: payload_len,405+ maximum: MAX_AUTHENTICATED_PAYLOAD_LEN,406+ })?;407+ Ok(())408+ }409+}410+411+412+pub struct AuthenticatedMediaObject {413+ bytes: Vec<u8>,414+}415+416+impl AuthenticatedMediaObject {417+ pub fn as_bytes(&self) -> &[u8] {418+ &self.bytes419+ }420+421+ pub fn into_bytes(self) -> Vec<u8> {422+ self.bytes423+ }424+425+ pub fn coordinates(object: &[u8]) -> Result<MediaObjectCoordinates> {426+ if object.len() < AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN {427+ return invalid(428+ "authenticated media object",429+ "object ended before its authentication header",430+ );431+ }432+ let unsigned = &object[..OBJECT_UNSIGNED_HEADER_LEN];433+ if unsigned[0..4] != OBJECT_MAGIC {434+ return invalid(435+ "authenticated media object",436+ "authentication magic is invalid",437+ );438+ }439+ let version = u16::from_be_bytes(unsigned[4..6].try_into().expect("fixed auth header"));440+ if version != OBJECT_VERSION {441+ return Err(ProtocolError::UnsupportedVersion {442+ expected: OBJECT_VERSION,443+ actual: version,444+ });445+ }446+ let header_len = u16::from_be_bytes(unsigned[6..8].try_into().expect("fixed auth header"));447+ if usize::from(header_len) != AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN {448+ return invalid(449+ "authenticated media object",450+ "authentication header length is invalid",451+ );452+ }453+ if MediaAuthMode::try_from(unsigned[8])? != MediaAuthMode::Ed25519454+ || unsigned[12..14].iter().any(|byte| *byte != 0)455+ {456+ return invalid(457+ "authenticated media object",458+ "authentication mode or reserved fields are invalid",459+ );460+ }461+ let coordinates = MediaObjectCoordinates {462+ generation_id: u64::from_be_bytes(463+ unsigned[14..22].try_into().expect("fixed auth header"),464+ ),465+ track: MediaTrack::try_from(unsigned[9])?,466+ group_id: u64::from_be_bytes(unsigned[22..30].try_into().expect("fixed auth header")),467+ object_id: u32::from_be_bytes(unsigned[30..34].try_into().expect("fixed auth header")),468+ flags: u16::from_be_bytes(unsigned[10..12].try_into().expect("fixed auth header")),469+ };470+ let payload_len =471+ u32::from_be_bytes(unsigned[34..38].try_into().expect("fixed auth header")) as usize;472+ coordinates.validate(payload_len)?;473+ Ok(coordinates)474+ }475+476+ pub fn verify<'a>(477+ object: &'a [u8],478+ certificate: &SignedMediaGenerationCertificate,479+ expected: MediaObjectCoordinates,480+ ) -> Result<&'a [u8]> {481+ certificate.verify()?;482+ let actual = Self::coordinates(object)?;483+ let unsigned = &object[..OBJECT_UNSIGNED_HEADER_LEN];484+ let payload_len =485+ u32::from_be_bytes(unsigned[34..38].try_into().expect("fixed auth header")) as usize;486+ if actual != expected || actual.generation_id != certificate.claims.generation_id {487+ return invalid(488+ "authenticated media object",489+ "object coordinates do not match delivery coordinates",490+ );491+ }492+ let payload = &object[AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN..];493+ if payload.len() != payload_len {494+ return invalid(495+ "authenticated media object",496+ "payload length does not match authentication header",497+ );498+ }499+ let expected_digest: [u8; 32] = unsigned[38..70].try_into().expect("fixed auth digest");500+ let actual_digest: [u8; 32] = Sha256::digest(payload).into();501+ if actual_digest != expected_digest {502+ return invalid(503+ "authenticated media object",504+ "payload digest verification failed",505+ );506+ }507+ let key =508+ VerifyingKey::from_bytes(&certificate.claims.generation_signing_key).map_err(|_| {509+ ProtocolError::InvalidMessage {510+ message_type: "authenticated media object",511+ reason: "certificate generation key is invalid",512+ }513+ })?;514+ let mut message = Vec::with_capacity(OBJECT_DOMAIN.len() + unsigned.len());515+ message.extend_from_slice(OBJECT_DOMAIN);516+ message.extend_from_slice(unsigned);517+ let signature = Signature::from_bytes(518+ object[OBJECT_UNSIGNED_HEADER_LEN..AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN]519+ .try_into()520+ .expect("fixed auth signature"),521+ );522+ key.verify_strict(&message, &signature)523+ .map_err(|_| ProtocolError::InvalidMessage {524+ message_type: "authenticated media object",525+ reason: "object signature verification failed",526+ })?;527+ Ok(payload)528+ }529+}530+531+fn encode_unsigned_object_header(532+ coordinates: MediaObjectCoordinates,533+ payload_len: usize,534+ digest: [u8; 32],535+) -> Result<[u8; OBJECT_UNSIGNED_HEADER_LEN]> {536+ coordinates.validate(payload_len)?;537+ let mut bytes = [0_u8; OBJECT_UNSIGNED_HEADER_LEN];538+ bytes[0..4].copy_from_slice(&OBJECT_MAGIC);539+ bytes[4..6].copy_from_slice(&OBJECT_VERSION.to_be_bytes());540+ bytes[6..8].copy_from_slice(&(AUTHENTICATED_MEDIA_OBJECT_HEADER_LEN as u16).to_be_bytes());541+ bytes[8] = MediaAuthMode::Ed25519 as u8;542+ bytes[9] = coordinates.track as u8;543+ bytes[10..12].copy_from_slice(&coordinates.flags.to_be_bytes());544+ bytes[14..22].copy_from_slice(&coordinates.generation_id.to_be_bytes());545+ bytes[22..30].copy_from_slice(&coordinates.group_id.to_be_bytes());546+ bytes[30..34].copy_from_slice(&coordinates.object_id.to_be_bytes());547+ bytes[34..38].copy_from_slice(&(payload_len as u32).to_be_bytes());548+ bytes[38..70].copy_from_slice(&digest);549+ Ok(bytes)550+}551+552+fn invalid<T>(message_type: &'static str, reason: &'static str) -> Result<T> {553+ Err(ProtocolError::InvalidMessage {554+ message_type,555+ reason,556+ })557+}558+559+560+mod tests {561+ use super::*;562+563+ fn fixture() -> (MediaGenerationSigningKey, SignedMediaGenerationCertificate) {564+ let host = SigningKey::from_bytes(&[7; 32]);565+ let generation = MediaGenerationSigningKey::from_bytes(&[9; 32]);566+ let certificate = generation567+ .certify(568+ host.verifying_key().to_bytes(),569+ &[7; 32],570+ 42,571+ 1_700_000_000,572+ 1_700_000_600,573+ )574+ .unwrap();575+ (generation, certificate)576+ }577+578+ 579+ fn generation_certificate_is_canonical_and_host_bound() {580+ let (_, certificate) = fixture();581+ let token = certificate.encode();582+ assert!(token.starts_with(MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX));583+ assert_eq!(584+ SignedMediaGenerationCertificate::decode(&token).unwrap(),585+ certificate586+ );587+ certificate588+ .verify_binding(certificate.claims.host_node_id, 42, 1_700_000_300)589+ .unwrap();590+ assert!(591+ certificate592+ .verify_binding([1; 32], 42, 1_700_000_300)593+ .is_err()594+ );595+ assert!(596+ certificate597+ .verify_binding(certificate.claims.host_node_id, 43, 1_700_000_300)598+ .is_err()599+ );600+ assert!(601+ certificate602+ .verify_binding(certificate.claims.host_node_id, 42, 1_700_001_000)603+ .is_err()604+ );605+ }606+607+ 608+ fn authenticated_object_header_has_stable_golden_prefix_and_round_trips() {609+ let (generation, certificate) = fixture();610+ let coordinates = MediaObjectCoordinates {611+ generation_id: 42,612+ track: MediaTrack::VideoH264,613+ group_id: 3,614+ object_id: 4,615+ flags: 5,616+ };617+ let object = generation.authenticate(coordinates, b"frame").unwrap();618+ assert_eq!(619+ &object.as_bytes()[..38],620+ &[621+ 0x53, 0x47, 0x4d, 0x41, 0, 1, 0, 134, 1, 1, 0, 5, 0, 0, 0, 0, 0, 0, 0, 0, 0, 42, 0,622+ 0, 0, 0, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0, 5,623+ ]624+ );625+ assert_eq!(626+ AuthenticatedMediaObject::verify(object.as_bytes(), &certificate, coordinates).unwrap(),627+ b"frame"628+ );629+ }630+631+ 632+ fn tampering_and_wrong_coordinates_fail_closed() {633+ let (generation, certificate) = fixture();634+ let coordinates = MediaObjectCoordinates {635+ generation_id: 42,636+ track: MediaTrack::VideoH264,637+ group_id: 3,638+ object_id: 4,639+ flags: 0,640+ };641+ let object = generation642+ .authenticate(coordinates, b"frame")643+ .unwrap()644+ .into_bytes();645+ let mut payload_tampered = object.clone();646+ *payload_tampered.last_mut().unwrap() ^= 1;647+ assert!(648+ AuthenticatedMediaObject::verify(&payload_tampered, &certificate, coordinates).is_err()649+ );650+ let mut signature_tampered = object.clone();651+ signature_tampered[OBJECT_UNSIGNED_HEADER_LEN] ^= 1;652+ assert!(653+ AuthenticatedMediaObject::verify(&signature_tampered, &certificate, coordinates)654+ .is_err()655+ );656+ let wrong = MediaObjectCoordinates {657+ object_id: 5,658+ ..coordinates659+ };660+ assert!(AuthenticatedMediaObject::verify(&object, &certificate, wrong).is_err());661+ }662+663+ 664+ fn malformed_lengths_versions_modes_and_reserved_fields_fail_closed() {665+ let (generation, certificate) = fixture();666+ let coordinates = MediaObjectCoordinates {667+ generation_id: 42,668+ track: MediaTrack::VideoH264,669+ group_id: 0,670+ object_id: 0,671+ flags: 0,672+ };673+ let valid = generation674+ .authenticate(coordinates, b"frame")675+ .unwrap()676+ .into_bytes();677+ for offset in [0, 5, 7, 8, 9, 12] {678+ let mut malformed = valid.clone();679+ malformed[offset] ^= 0xff;680+ assert!(681+ AuthenticatedMediaObject::verify(&malformed, &certificate, coordinates).is_err(),682+ "accepted malformed offset {offset}"683+ );684+ }685+ assert!(686+ AuthenticatedMediaObject::verify(&valid[..100], &certificate, coordinates).is_err()687+ );688+ }689+}@@ -1,11 +1,16 @@
11 use serde::{Deserialize, Serialize};22 3-use crate::{MOQ_VIDEO_H264_TRACK, ProtocolError, Result};3+use crate::{4+ MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX, MOQ_VIDEO_H264_TRACK, MediaAuthMode, ProtocolError,5+ Result, SignedMediaGenerationCertificate,6+};47 58 pub const MOQ_CATALOG_EXTENSION_VERSION_V1: u16 = 1;9+pub const MOQ_CATALOG_EXTENSION_VERSION_V2: u16 = 2;610 /// Maximum immutable catalog snapshot accepted from a peer before JSON decoding.711 pub const MAX_MOQ_CATALOG_BYTES: usize = 4 * 1024;812 pub const MOQ_MEDIA_OBJECT_FORMAT_V1: &str = "sigil/media-frame/1";13+pub const MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1: &str = "sigil/authenticated-media-object/1";914 pub const MOQ_GOP_GROUP_FORMAT_V1: &str = "sigil/moq-gop/1";1015 pub const MOQ_VIDEO_TRACK_PRIORITY: u8 = u8::MAX;1116 @@ -89,6 +94,100 @@ impl MoqCatalogExtensionV1 {
8994 }9095 }9196 97+98+99+pub struct MoqTrackAuthenticationV2 {100+ pub mode: String,101+ pub object_format: String,102+ pub generation_certificate: String,103+}104+105+106+107+pub struct MoqVideoCatalogV2 {108+ pub track: MoqTrackDescriptorV1,109+ pub codec: String,110+ pub payload_format: String,111+ pub group_format: String,112+ pub authentication: MoqTrackAuthenticationV2,113+}114+115+116+117+pub struct MoqCatalogExtensionV2 {118+ pub version: u16,119+ pub generation_id: u64,120+ pub video: MoqVideoCatalogV2,121+}122+123+impl MoqCatalogExtensionV2 {124+ pub fn video_h264(125+ generation_id: u64,126+ certificate: &SignedMediaGenerationCertificate,127+ ) -> Result<Self> {128+ let extension = Self {129+ version: MOQ_CATALOG_EXTENSION_VERSION_V2,130+ generation_id,131+ video: MoqVideoCatalogV2 {132+ track: MoqTrackDescriptorV1 {133+ name: MOQ_VIDEO_H264_TRACK.to_owned(),134+ priority: MOQ_VIDEO_TRACK_PRIORITY,135+ },136+ codec: "h264".to_owned(),137+ payload_format: MOQ_MEDIA_OBJECT_FORMAT_V1.to_owned(),138+ group_format: MOQ_GOP_GROUP_FORMAT_V1.to_owned(),139+ authentication: MoqTrackAuthenticationV2 {140+ mode: MediaAuthMode::Ed25519.label().to_owned(),141+ object_format: MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1.to_owned(),142+ generation_certificate: certificate.encode(),143+ },144+ },145+ };146+ extension.validate()?;147+ Ok(extension)148+ }149+150+ pub fn validate(&self) -> Result<SignedMediaGenerationCertificate> {151+ if self.version != MOQ_CATALOG_EXTENSION_VERSION_V2 || self.generation_id == 0 {152+ return invalid_v2("unsupported extension version or zero generation");153+ }154+ if self.video.track.name != MOQ_VIDEO_H264_TRACK155+ || self.video.track.priority != MOQ_VIDEO_TRACK_PRIORITY156+ || self.video.codec != "h264"157+ || self.video.payload_format != MOQ_MEDIA_OBJECT_FORMAT_V1158+ || self.video.group_format != MOQ_GOP_GROUP_FORMAT_V1159+ {160+ return invalid_v2("video track contract does not match authenticated H.264");161+ }162+ if self.video.authentication.mode != MediaAuthMode::Ed25519.label()163+ || self.video.authentication.object_format != MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1164+ || !self165+ .video166+ .authentication167+ .generation_certificate168+ .starts_with(MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX)169+ {170+ return invalid_v2("unknown or malformed media authentication contract");171+ }172+ let certificate = SignedMediaGenerationCertificate::decode(173+ &self.video.authentication.generation_certificate,174+ )?;175+ if certificate.claims.generation_id != self.generation_id176+ || certificate.claims.authentication_mode != MediaAuthMode::Ed25519177+ {178+ return invalid_v2("generation certificate does not match the catalog generation");179+ }180+ Ok(certificate)181+ }182+}183+184+fn invalid_v2<T>(reason: &'static str) -> Result<T> {185+ Err(ProtocolError::InvalidMessage {186+ message_type: "Goq MoQ catalog extension v2",187+ reason,188+ })189+}190+92191 /// The immutable catalog.json document both peers must agree on byte-for-byte:93192 /// a default (empty) Hang catalog envelope carrying the Goq extension. This is94193 /// the single definition — the host producer, Portal subscriber, and probe all@@ -120,9 +219,56 @@ impl GoqCatalogDocument {
120219 }121220 }122221 222+223+224+pub struct GoqCatalogDocumentV2 {225+ 226+ pub media: hang::Catalog,227+ pub goq: MoqCatalogExtensionV2,228+}229+230+impl GoqCatalogDocumentV2 {231+ pub fn video_h264(232+ generation_id: u64,233+ certificate: &SignedMediaGenerationCertificate,234+ ) -> Result<Self> {235+ Ok(Self {236+ media: hang::Catalog::default(),237+ goq: MoqCatalogExtensionV2::video_h264(generation_id, certificate)?,238+ })239+ }240+241+ pub fn validate(&self) -> Result<SignedMediaGenerationCertificate> {242+ if self.media != hang::Catalog::default() {243+ return Err(ProtocolError::InvalidMessage {244+ message_type: "Goq MoQ catalog document v2",245+ reason: "catalog must not advertise a standard Hang rendition for enveloped media",246+ });247+ }248+ self.goq.validate()249+ }250+}251+123252 124253 mod tests {125254 use super::*;255+ use crate::MediaGenerationSigningKey;256+ use ed25519_dalek::SigningKey;257+258+ fn v2_document() -> GoqCatalogDocumentV2 {259+ let host = SigningKey::from_bytes(&[7; 32]);260+ let generation = MediaGenerationSigningKey::from_bytes(&[9; 32]);261+ let certificate = generation262+ .certify(263+ host.verifying_key().to_bytes(),264+ &[7; 32],265+ 42,266+ 1_700_000_000,267+ 1_700_000_600,268+ )269+ .unwrap();270+ GoqCatalogDocumentV2::video_h264(42, &certificate).unwrap()271+ }126272 127273 128274 fn goq_catalog_extension_has_a_stable_golden_document() {@@ -180,4 +326,41 @@ mod tests {
180326 let json = r#"{"version":1,"video":{"track":{"name":"video/h264","priority":255,"extra":true},"codec":"h264","objectFormat":"sigil/media-frame/1","groupFormat":"sigil/moq-gop/1"}}"#;181327 assert!(serde_json::from_str::<MoqCatalogExtensionV1>(json).is_err());182328 }329+330+ 331+ fn authenticated_v2_catalog_is_strict_and_preserves_v1_decoding() {332+ let document = v2_document();333+ let certificate = document.validate().unwrap();334+ assert_eq!(certificate.claims.generation_id, 42);335+ let json = serde_json::to_string(&document).unwrap();336+ assert!(json.contains(r#""version":2"#));337+ assert!(json.contains(r#""mode":"ed25519-v1""#));338+ assert!(json.contains(MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1));339+ assert_eq!(340+ serde_json::from_str::<GoqCatalogDocumentV2>(&json).unwrap(),341+ document342+ );343+344+ let legacy = serde_json::to_string(&GoqCatalogDocument::video_h264()).unwrap();345+ serde_json::from_str::<GoqCatalogDocument>(&legacy)346+ .unwrap()347+ .validate()348+ .unwrap();349+ assert!(serde_json::from_str::<GoqCatalogDocumentV2>(&legacy).is_err());350+ }351+352+ 353+ fn authenticated_v2_catalog_rejects_unknown_mode_generation_and_fields() {354+ let mut wrong_mode = v2_document();355+ wrong_mode.goq.video.authentication.mode = "shared-mac-v1".into();356+ assert!(wrong_mode.validate().is_err());357+358+ let mut wrong_generation = v2_document();359+ wrong_generation.goq.generation_id = 43;360+ assert!(wrong_generation.validate().is_err());361+362+ let mut value = serde_json::to_value(v2_document()).unwrap();363+ value["goq"]["video"]["authentication"]["key"] = serde_json::json!("secret");364+ assert!(serde_json::from_value::<GoqCatalogDocumentV2>(value).is_err());365+ }183366 }@@ -0,0 +1,440 @@
1+use base64::Engine as _;2+use base64::engine::general_purpose::URL_SAFE_NO_PAD;3+use ed25519_dalek::{Signature, Signer as _, SigningKey, VerifyingKey};4+5+use crate::{ProtocolError, Result};6+7+const SUBSCRIPTION_MAGIC: [u8; 4] = *b"SGSC";8+const SUBSCRIPTION_VERSION: u16 = 1;9+const CLAIMS_LEN: usize = 136;10+const SIGNATURE_LEN: usize = 64;11+const CAPABILITY_LEN: usize = CLAIMS_LEN + SIGNATURE_LEN;12+const SIGNATURE_DOMAIN: &[u8] = b"goq.sh/sigil-subscription/v1\0";13+14+pub const SUBSCRIPTION_CAPABILITY_TOKEN_PREFIX: &str = "goq-subscription-v1.";15+pub const MAX_SUBSCRIPTION_CAPABILITY_TOKEN_LEN: usize = 320;16+pub const MAX_SUBSCRIPTION_TTL_SECS: u64 = 15 * 60;17+18+19+pub struct SubscriptionTracks(u8);20+21+impl SubscriptionTracks {22+ pub const VIDEO_H264: Self = Self(1 << 0);23+ pub const AUDIO_OPUS: Self = Self(1 << 1);24+ pub const ALL: Self = Self(Self::VIDEO_H264.0 | Self::AUDIO_OPUS.0);25+ const KNOWN: u8 = Self::ALL.0;26+27+ pub fn new(bits: u8) -> Result<Self> {28+ if bits == 0 || bits & !Self::KNOWN != 0 {29+ return invalid(30+ "subscription tracks",31+ "tracks must contain only known non-zero bits",32+ );33+ }34+ Ok(Self(bits))35+ }36+37+ pub const fn bits(self) -> u8 {38+ self.039+ }40+41+ pub const fn contains(self, required: Self) -> bool {42+ self.0 & required.0 == required.043+ }44+45+ pub const fn union(self, other: Self) -> Self {46+ Self(self.0 | other.0)47+ }48+}49+50+51+pub struct SubscriptionClaims {52+ pub host_node_id: [u8; 32],53+ pub media_generation_id: u64,54+ pub subscriber_endpoint_id: [u8; 32],55+ pub tracks: SubscriptionTracks,56+ pub authorization_revision: u64,57+ pub issued_at_unix: u64,58+ pub expires_at_unix: u64,59+ pub nonce: [u8; 32],60+ pub relay_hops: u8,61+}62+63+impl SubscriptionClaims {64+ 65+ pub fn new(66+ host_node_id: [u8; 32],67+ media_generation_id: u64,68+ subscriber_endpoint_id: [u8; 32],69+ tracks: SubscriptionTracks,70+ authorization_revision: u64,71+ issued_at_unix: u64,72+ expires_at_unix: u64,73+ nonce: [u8; 32],74+ relay_hops: u8,75+ ) -> Result<Self> {76+ let claims = Self {77+ host_node_id,78+ media_generation_id,79+ subscriber_endpoint_id,80+ tracks,81+ authorization_revision,82+ issued_at_unix,83+ expires_at_unix,84+ nonce,85+ relay_hops,86+ };87+ claims.validate()?;88+ Ok(claims)89+ }90+91+ pub fn validate(&self) -> Result<()> {92+ SubscriptionTracks::new(self.tracks.bits())?;93+ if self.media_generation_id == 0 || self.authorization_revision == 0 {94+ return invalid(95+ "subscription capability",96+ "media generation and authorization revision must be non-zero",97+ );98+ }99+ if self.relay_hops != 1 {100+ return invalid(101+ "subscription capability",102+ "exactly one authorized relay hop is required",103+ );104+ }105+ if self.issued_at_unix >= self.expires_at_unix106+ || self.expires_at_unix - self.issued_at_unix > MAX_SUBSCRIPTION_TTL_SECS107+ {108+ return invalid(109+ "subscription capability",110+ "capability lifetime must be within 1..=900 seconds",111+ );112+ }113+ if self.nonce == [0; 32] {114+ return invalid("subscription capability", "nonce must be non-zero");115+ }116+ VerifyingKey::from_bytes(&self.host_node_id).map_err(|_| {117+ ProtocolError::InvalidMessage {118+ message_type: "subscription capability",119+ reason: "host node id is not an Ed25519 public key",120+ }121+ })?;122+ VerifyingKey::from_bytes(&self.subscriber_endpoint_id).map_err(|_| {123+ ProtocolError::InvalidMessage {124+ message_type: "subscription capability",125+ reason: "subscriber endpoint id is not an Ed25519 public key",126+ }127+ })?;128+ Ok(())129+ }130+131+ fn encode(&self) -> [u8; CLAIMS_LEN] {132+ let mut bytes = [0_u8; CLAIMS_LEN];133+ bytes[0..4].copy_from_slice(&SUBSCRIPTION_MAGIC);134+ bytes[4..6].copy_from_slice(&SUBSCRIPTION_VERSION.to_be_bytes());135+ bytes[6] = self.relay_hops;136+ bytes[7] = self.tracks.bits();137+ bytes[8..40].copy_from_slice(&self.host_node_id);138+ bytes[40..48].copy_from_slice(&self.media_generation_id.to_be_bytes());139+ bytes[48..80].copy_from_slice(&self.subscriber_endpoint_id);140+ bytes[80..88].copy_from_slice(&self.authorization_revision.to_be_bytes());141+ bytes[88..96].copy_from_slice(&self.issued_at_unix.to_be_bytes());142+ bytes[96..104].copy_from_slice(&self.expires_at_unix.to_be_bytes());143+ bytes[104..136].copy_from_slice(&self.nonce);144+ bytes145+ }146+147+ fn signing_message(&self) -> Vec<u8> {148+ let mut message = Vec::with_capacity(SIGNATURE_DOMAIN.len() + CLAIMS_LEN);149+ message.extend_from_slice(SIGNATURE_DOMAIN);150+ message.extend_from_slice(&self.encode());151+ message152+ }153+}154+155+156+pub struct SignedSubscriptionCapability {157+ pub claims: SubscriptionClaims,158+ signature: [u8; SIGNATURE_LEN],159+}160+161+impl SignedSubscriptionCapability {162+ pub fn issue(claims: SubscriptionClaims, host_secret: &[u8; 32]) -> Result<Self> {163+ claims.validate()?;164+ let signing_key = SigningKey::from_bytes(host_secret);165+ if signing_key.verifying_key().to_bytes() != claims.host_node_id {166+ return invalid(167+ "subscription capability",168+ "host signing key does not match the capability host",169+ );170+ }171+ let signature = signing_key.sign(&claims.signing_message()).to_bytes();172+ Ok(Self { claims, signature })173+ }174+175+ pub fn verify(&self) -> Result<()> {176+ self.claims.validate()?;177+ let key = VerifyingKey::from_bytes(&self.claims.host_node_id).map_err(|_| {178+ ProtocolError::InvalidMessage {179+ message_type: "subscription capability",180+ reason: "invalid host verification key",181+ }182+ })?;183+ key.verify_strict(184+ &self.claims.signing_message(),185+ &Signature::from_bytes(&self.signature),186+ )187+ .map_err(|_| ProtocolError::InvalidMessage {188+ message_type: "subscription capability",189+ reason: "host signature verification failed",190+ })191+ }192+193+ 194+ pub fn verify_binding(195+ &self,196+ expected_host: [u8; 32],197+ expected_generation: u64,198+ expected_subscriber: [u8; 32],199+ required_tracks: SubscriptionTracks,200+ expected_authorization_revision: u64,201+ now_unix: u64,202+ ) -> Result<()> {203+ self.verify()?;204+ if self.claims.host_node_id != expected_host {205+ return invalid(206+ "subscription capability",207+ "capability is bound to another host",208+ );209+ }210+ if self.claims.media_generation_id != expected_generation {211+ return invalid(212+ "subscription capability",213+ "capability is bound to another media generation",214+ );215+ }216+ if self.claims.subscriber_endpoint_id != expected_subscriber {217+ return invalid(218+ "subscription capability",219+ "capability is bound to another subscriber",220+ );221+ }222+ if !self.claims.tracks.contains(required_tracks) {223+ return invalid(224+ "subscription capability",225+ "capability does not authorize the required tracks",226+ );227+ }228+ if self.claims.authorization_revision != expected_authorization_revision {229+ return invalid(230+ "subscription capability",231+ "capability authorization revision is stale",232+ );233+ }234+ if now_unix < self.claims.issued_at_unix || now_unix > self.claims.expires_at_unix {235+ return invalid(236+ "subscription capability",237+ "capability is not currently valid",238+ );239+ }240+ Ok(())241+ }242+243+ pub fn encode(&self) -> String {244+ let mut bytes = Vec::with_capacity(CAPABILITY_LEN);245+ bytes.extend_from_slice(&self.claims.encode());246+ bytes.extend_from_slice(&self.signature);247+ format!(248+ "{SUBSCRIPTION_CAPABILITY_TOKEN_PREFIX}{}",249+ URL_SAFE_NO_PAD.encode(bytes)250+ )251+ }252+253+ pub fn decode(token: &str) -> Result<Self> {254+ if token.len() > MAX_SUBSCRIPTION_CAPABILITY_TOKEN_LEN255+ || token.contains('=')256+ || token.bytes().any(|byte| byte.is_ascii_whitespace())257+ {258+ return invalid(259+ "subscription capability",260+ "capability token is oversized or non-canonical",261+ );262+ }263+ let encoded = token264+ .strip_prefix(SUBSCRIPTION_CAPABILITY_TOKEN_PREFIX)265+ .ok_or(ProtocolError::InvalidMessage {266+ message_type: "subscription capability",267+ reason: "capability token prefix is invalid",268+ })?;269+ let bytes = URL_SAFE_NO_PAD270+ .decode(encoded)271+ .map_err(|_| ProtocolError::InvalidMessage {272+ message_type: "subscription capability",273+ reason: "capability token is not canonical base64url",274+ })?;275+ if bytes.len() != CAPABILITY_LEN {276+ return Err(ProtocolError::InvalidMessageLength {277+ actual: bytes.len(),278+ maximum: CAPABILITY_LEN,279+ });280+ }281+ if bytes[0..4] != SUBSCRIPTION_MAGIC {282+ return invalid("subscription capability", "capability magic is invalid");283+ }284+ let version = u16::from_be_bytes(bytes[4..6].try_into().expect("fixed capability"));285+ if version != SUBSCRIPTION_VERSION {286+ return Err(ProtocolError::UnsupportedVersion {287+ expected: SUBSCRIPTION_VERSION,288+ actual: version,289+ });290+ }291+ let claims = SubscriptionClaims::new(292+ bytes[8..40].try_into().expect("fixed capability"),293+ u64::from_be_bytes(bytes[40..48].try_into().expect("fixed capability")),294+ bytes[48..80].try_into().expect("fixed capability"),295+ SubscriptionTracks::new(bytes[7])?,296+ u64::from_be_bytes(bytes[80..88].try_into().expect("fixed capability")),297+ u64::from_be_bytes(bytes[88..96].try_into().expect("fixed capability")),298+ u64::from_be_bytes(bytes[96..104].try_into().expect("fixed capability")),299+ bytes[104..136].try_into().expect("fixed capability"),300+ bytes[6],301+ )?;302+ let capability = Self {303+ claims,304+ signature: bytes[CLAIMS_LEN..]305+ .try_into()306+ .expect("fixed capability signature"),307+ };308+ capability.verify()?;309+ if capability.encode() != token {310+ return invalid(311+ "subscription capability",312+ "capability encoding is not canonical",313+ );314+ }315+ Ok(capability)316+ }317+}318+319+fn invalid<T>(message_type: &'static str, reason: &'static str) -> Result<T> {320+ Err(ProtocolError::InvalidMessage {321+ message_type,322+ reason,323+ })324+}325+326+327+mod tests {328+ use super::*;329+330+ fn capability() -> SignedSubscriptionCapability {331+ let host = SigningKey::from_bytes(&[7; 32]);332+ let subscriber = SigningKey::from_bytes(&[9; 32]);333+ SignedSubscriptionCapability::issue(334+ SubscriptionClaims::new(335+ host.verifying_key().to_bytes(),336+ 42,337+ subscriber.verifying_key().to_bytes(),338+ SubscriptionTracks::VIDEO_H264,339+ 3,340+ 1_700_000_000,341+ 1_700_000_600,342+ [11; 32],343+ 1,344+ )345+ .unwrap(),346+ &[7; 32],347+ )348+ .unwrap()349+ }350+351+ 352+ fn endpoint_bound_capability_is_canonical_and_round_trips() {353+ let capability = capability();354+ let token = capability.encode();355+ assert!(token.starts_with(SUBSCRIPTION_CAPABILITY_TOKEN_PREFIX));356+ assert_eq!(357+ SignedSubscriptionCapability::decode(&token).unwrap(),358+ capability359+ );360+ capability361+ .verify_binding(362+ capability.claims.host_node_id,363+ 42,364+ capability.claims.subscriber_endpoint_id,365+ SubscriptionTracks::VIDEO_H264,366+ 3,367+ 1_700_000_300,368+ )369+ .unwrap();370+ }371+372+ 373+ fn wrong_host_subscriber_generation_revision_track_and_time_fail_closed() {374+ let capability = capability();375+ let valid = (376+ capability.claims.host_node_id,377+ 42,378+ capability.claims.subscriber_endpoint_id,379+ SubscriptionTracks::VIDEO_H264,380+ 3,381+ 1_700_000_300,382+ );383+ assert!(384+ capability385+ .verify_binding([1; 32], valid.1, valid.2, valid.3, valid.4, valid.5)386+ .is_err()387+ );388+ assert!(389+ capability390+ .verify_binding(valid.0, 43, valid.2, valid.3, valid.4, valid.5)391+ .is_err()392+ );393+ assert!(394+ capability395+ .verify_binding(valid.0, valid.1, [2; 32], valid.3, valid.4, valid.5)396+ .is_err()397+ );398+ assert!(399+ capability400+ .verify_binding(401+ valid.0,402+ valid.1,403+ valid.2,404+ SubscriptionTracks::AUDIO_OPUS,405+ valid.4,406+ valid.5407+ )408+ .is_err()409+ );410+ assert!(411+ capability412+ .verify_binding(valid.0, valid.1, valid.2, valid.3, 4, valid.5)413+ .is_err()414+ );415+ assert!(416+ capability417+ .verify_binding(valid.0, valid.1, valid.2, valid.3, valid.4, 1_700_001_000)418+ .is_err()419+ );420+ }421+422+ 423+ fn tampering_unknown_tracks_hops_and_noncanonical_tokens_fail_closed() {424+ let token = capability().encode();425+ let encoded = token426+ .strip_prefix(SUBSCRIPTION_CAPABILITY_TOKEN_PREFIX)427+ .unwrap();428+ let bytes = URL_SAFE_NO_PAD.decode(encoded).unwrap();429+ for (offset, value) in [(0, 0), (5, 2), (6, 2), (7, 0x80), (CLAIMS_LEN, 0)] {430+ let mut malformed = bytes.clone();431+ malformed[offset] = value;432+ let token = format!(433+ "{SUBSCRIPTION_CAPABILITY_TOKEN_PREFIX}{}",434+ URL_SAFE_NO_PAD.encode(malformed)435+ );436+ assert!(SignedSubscriptionCapability::decode(&token).is_err());437+ }438+ assert!(SignedSubscriptionCapability::decode(&(token + "=")).is_err());439+ }440+}@@ -0,0 +1,102 @@
1+# Multi-viewer relay design and issue #81 decision2+3+## Decision4+5+Production viewer-to-viewer media routing is **deferred**. The Phase 2 wire shape is adopted now: every control-v2 media generation has a host-certified ephemeral Ed25519 key, every media object is signed over canonical coordinates and a SHA-256 payload digest, and every subscriber receives a short-lived host-signed capability bound to its exact Iroh endpoint. Direct-host v2 media uses this contract immediately, so later relay delivery does not require replacing the media trust model.6+7+Deferral is deliberate rather than an undocumented follow-up. A production relay scheduler, alternate-publisher discovery, automatic failover, and relay UI are not enabled by this phase. Input, focus, authorization mutation, and host administration always terminate at Sigil.8+9+## Spike result10+11+`scripts/relay-spike-proof.sh` models two Portal roles over a bounded virtual five-minute media horizon. Portal A receives each host-authored object and re-serves it through a four-object-capacity relay queue. Portal B verifies the host certificate, final MoQ group/object coordinates, object signature, and payload digest before accepting the existing media envelope. The spike does not sleep, so CPU percentages are normalized against the represented 300 seconds rather than wall-clock runtime.12+13+Release-mode result on the development macOS arm64 machine on 2026-07-23:14+15+| Measurement | Result |16+|---|---:|17+| Video target | 1280x800 at 60 fps |18+| Audio target | Opus, 48 kHz stereo, 50 objects/s |19+| Virtual duration | 300 s |20+| Signed video objects | 18,000 |21+| Signed audio objects | 15,000 |22+| Relay queue high-water mark | 1 of 4 |23+| Host upload through Portal A | 1,187,934,000 bytes |24+| Direct host upload to two viewers | 2,375,868,000 bytes |25+| Modeled host-upload saving | 50% |26+| Ed25519 signing CPU / virtual realtime | 0.72% of one core |27+| Ed25519 verification CPU / virtual realtime | 1.14% of one core |28+| Local verification p95 | 160 us |29+| Added network latency | Not measured by the virtual spike |30+31+The automated spike proves canonical re-serving, bounded memory, upload accounting, and cryptographic behavior. It is not evidence for Iroh direct-path or relay-fallback network latency. The manual Phase 2 gate must record two physical Portal paths before production mesh routing can be reconsidered.32+33+## Resource threshold and authentication mode34+35+Ed25519 is the only accepted v2 catalog authentication mode. It remains the default while exact-target release-mode evidence stays at or below all of these thresholds:36+37+- Host signing uses at most 5% of one core over the represented 1280x800/60 plus Opus workload.38+- Portal verification uses at most 5% of one core over the same workload.39+- Local authentication adds at most 1 ms p95 per object before codec validation.40+- Signing overhead does not cause a source, MoQ, frontend, audio, or input queue to exceed its existing bound.41+42+A shared-MAC contingency may be evaluated only if an exact Bazzite host or supported Portal target crosses one of those thresholds. It must use a separately negotiated and diagnosed catalog mode. It must never be a silent downgrade, and its security consequence must be explicit: any viewer holding the shared key could forge relayed media. No shared-MAC mode is implemented or accepted now.43+44+## Connectivity model45+46+Sigil and every Portal retain normal Iroh endpoints. Iroh owns endpoint identity, encryption, direct-path discovery, NAT traversal, and relay fallback. A media relay is another authenticated Iroh delivery hop; it does not become Sigil and does not receive input or administration authority.47+48+The future attachment sequence is:49+50+1. Sigil admits Portal B over control v2 and issues a short-lived subscription capability for B's exact Iroh endpoint, media generation, authorization revision, tracks, expiry, nonce, and one relay hop.51+2. Sigil's catalog certifies the generation key and media authentication format.52+3. Sigil or a selected Portal A advertises an alternate publisher for a bounded group range. A production descriptor must identify the publisher endpoint, generation, track, group range, expiry, and host signature; raw peer identities must not enter the shared roster UI.53+4. Portal B connects to Portal A through Iroh and presents the capability. Portal A verifies the host signature and compares the authenticated downstream `remote_id()` with the capability subscriber.54+5. Portal B independently verifies the generation certificate and every object. Trust remains rooted in the enrolled Sigil host, not Portal A or the delivery path.55+56+Alternate-publisher descriptors are intentionally not emitted in Phase 2 because no production relay accepts them. The v2 catalog has an explicit version and authentication block, so a later strict extension can add a bounded alternate-publisher list without changing object bytes or treating an unknown mode as valid.57+58+## Trust boundaries59+60+- The persistent Sigil Iroh key signs only the domain-separated generation certificate and subscription capability.61+- A fresh ephemeral generation key signs canonical media-object headers. The header binds host-certified generation, track, MoQ group, object index, media flags, payload length, and SHA-256 digest.62+- The existing H.264 or Opus envelope remains the signed payload and is validated only after object authentication succeeds.63+- Subscription capabilities authorize tracks, not focus or input. They are endpoint-bound, revision-bound, expiring, nonce-bearing, and permit exactly one relay hop.64+- Capability tokens, host secrets, generation secrets, peer keys, and object payloads are excluded from diagnostics and logs.65+66+## Failure, failover, and abuse67+68+| Failure or abuse | Required behavior |69+|---|---|70+| Relay changes payload, coordinates, flags, or signature | Portal rejects the object before codec parsing and increments a local verification-failure counter. |71+| Relay withholds a group or stalls an object | Existing bounded MoQ deadlines enter recovery; withheld bytes are never counted as trusted media. |72+| Relay disappears | Portal discards relay-owned state and reconnects directly to Sigil using the same host-rooted object trust contract. |73+| Wrong or expired subscriber capability | Relay or host rejects attachment before publishing media. |74+| Wrong host or generation certificate | Portal rejects the catalog and terminates v2 media setup without falling back to unauthenticated v2 objects. |75+| Replay from another track/group/object | Canonical coordinate comparison rejects it even when the payload and signature were once valid. |76+| Slow or malicious relay | Its bounded queue may drop or disconnect that path; it cannot increase Sigil's producer history or another viewer's queue. |77+| Alternate publisher lies about availability | Timeout leads to direct-host recovery; advertisements never make absent media trusted. |78+79+Direct fallback is a transport fallback, not an authentication fallback. An unknown catalog mode, absent v2 certificate, malformed capability, or failed object signature is terminal for that path.80+81+## Adoption triggers82+83+Re-open production mesh routing only when all of these are available:84+85+- Exact-target measurements show direct Sigil upload is sustained above 70% of available uplink with at least three viewers, or product UAT establishes a lower measured saturation threshold.86+- Two physical Portals prove direct Iroh and relay-fallback paths, one-hop capability presentation, relay loss, and direct-host recovery with added end-to-end p95 below one 60 Hz frame period.87+- The relay queue remains bounded under a slow downstream and cannot retain a playable history.88+- Abuse tests cover tampering, withholding, wrong host/subscriber/generation/revision, expiry, replay, and relay churn.89+- Portal UX can truthfully disclose `direct_host` versus `viewer_relay` without exposing endpoint keys or topology details.90+91+Until those triggers pass, the adopted authentication format keeps fan-out relay-ready while all production viewers subscribe directly to Sigil.92+93+## Reproduction94+95+```bash96+./scripts/relay-spike-proof.sh \97+ --video 1280x800@60 \98+ --audio opus-48k-stereo \99+ --duration-seconds 300100+```101+102+The command must report Ed25519 mode, 50% modeled two-viewer host-upload savings, bounded relay queue depth, tamper/wrong-subscriber/expiry rejection, zero trusted withheld media, and successful direct-host verification after simulated relay loss.@@ -3,6 +3,38 @@ export const MAX_STREAM_RATE_SAMPLES = 240;
33 export const CADENCE_HITCH_25_MS = 25;44 export const CADENCE_HITCH_33_333_MS = 33.333;55 6+export function mediaAuthenticationPresentation(networkDiagnostics) {7+ const authentication = networkDiagnostics?.media_authentication;8+ const modes = new Set(['ed25519-v1', 'transport-authenticated-v1']);9+ const deliveryRoles = new Set(['direct_host', 'viewer_relay']);10+ if (!authentication11+ || !modes.has(authentication.mode)12+ || !deliveryRoles.has(authentication.delivery_role)13+ || typeof authentication.generation_certified !== 'boolean'14+ || !Number.isSafeInteger(authentication.verification_failures)15+ || authentication.verification_failures < 0) {16+ return {17+ mode: 'invalid',18+ generationState: 'unverified',19+ deliveryRole: 'unknown',20+ verificationFailures: null,21+ trusted: false,22+ };23+ }24+ const authenticated = authentication.mode === 'ed25519-v1';25+ return {26+ mode: authentication.mode,27+ generationState: authentication.generation_certified ? 'certified' : 'transport-bound',28+ deliveryRole: authentication.delivery_role,29+ verificationFailures: authentication.verification_failures,30+ trusted: authenticated31+ ? authentication.generation_certified && authentication.verification_failures === 032+ : authentication.delivery_role === 'direct_host'33+ && !authentication.generation_certified34+ && authentication.verification_failures === 0,35+ };36+}37+638 /**739 * A bounded, monotonic rolling-rate estimator. Callers supply their own clock840 * so the same implementation is deterministic in tests and uses@@ -8,8 +8,52 @@ import {
88 LatestFramePresenter,99 MAX_STREAM_RATE_SAMPLES,1010 RollingRateWindow,11+ mediaAuthenticationPresentation,1112 } from './stream-metrics.mjs';1213 14+test('media authentication presentation reports mode role certification and failures', () => {15+ assert.deepEqual(mediaAuthenticationPresentation({16+ media_authentication: {17+ mode: 'ed25519-v1',18+ generation_certified: true,19+ delivery_role: 'direct_host',20+ verification_failures: 0,21+ },22+ }), {23+ mode: 'ed25519-v1',24+ generationState: 'certified',25+ deliveryRole: 'direct_host',26+ verificationFailures: 0,27+ trusted: true,28+ });29+ assert.equal(mediaAuthenticationPresentation({30+ media_authentication: {31+ mode: 'ed25519-v1',32+ generation_certified: true,33+ delivery_role: 'viewer_relay',34+ verification_failures: 1,35+ },36+ }).trusted, false);37+});38+39+test('media authentication presentation fails closed on unknown or malformed diagnostics', () => {40+ for (const diagnostics of [41+ null,42+ {},43+ { media_authentication: { mode: 'shared-mac-v1', generation_certified: true, delivery_role: 'direct_host', verification_failures: 0 } },44+ { media_authentication: { mode: 'ed25519-v1', generation_certified: true, delivery_role: 'mesh', verification_failures: 0 } },45+ { media_authentication: { mode: 'ed25519-v1', generation_certified: true, delivery_role: 'direct_host', verification_failures: -1 } },46+ ]) {47+ assert.deepEqual(mediaAuthenticationPresentation(diagnostics), {48+ mode: 'invalid',49+ generationState: 'unverified',50+ deliveryRole: 'unknown',51+ verificationFailures: null,52+ trusted: false,53+ });54+ }55+});56+1357 test('cadence reports bounded interval percentiles and hitch counts', () => {1458 const cadence = new BoundedCadenceWindow(5000, 8);1559 let nowMs = 100;@@ -0,0 +1,79 @@
1+2+set -euo pipefail3+4+video="1280x800@60"5+audio="opus-48k-stereo"6+duration_seconds=3007+8+usage() {9+ cat <<'USAGE'10+Usage: relay-spike-proof.sh [options]11+12+ --video TARGET Fixed encoded target (required: 1280x800@60)13+ --audio FORMAT Fixed audio format (required: opus-48k-stereo)14+ --duration-seconds SECONDS Virtual benchmark horizon, 1..300 (default: 300)15+USAGE16+}17+18+while (($#)); do19+ case "$1" in20+ --video)21+ video=${2:?missing --video value}22+ shift 223+ ;;24+ --audio)25+ audio=${2:?missing --audio value}26+ shift 227+ ;;28+ --duration-seconds)29+ duration_seconds=${2:?missing --duration-seconds value}30+ shift 231+ ;;32+ --help|-h)33+ usage34+ exit 035+ ;;36+ *)37+ printf 'error: unknown argument: %s\n' "$1" >&238+ usage >&239+ exit 240+ ;;41+ esac42+done43+44+[[ "$video" == "1280x800@60" ]] || { printf 'error: unsupported video target\n' >&2; exit 2; }45+[[ "$audio" == "opus-48k-stereo" ]] || { printf 'error: unsupported audio format\n' >&2; exit 2; }46+[[ "$duration_seconds" =~ ^[0-9]+$ ]] || { printf 'error: duration must be an integer\n' >&2; exit 2; }47+((duration_seconds >= 1 && duration_seconds <= 300)) || {48+ printf 'error: duration must be within 1..300 seconds\n' >&249+ exit 250+}51+52+repo_root=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)53+tmp_root=$(mktemp -d "${TMPDIR:-/tmp}/goq-relay-spike.XXXXXX")54+trap 'rm -rf "$tmp_root"' EXIT55+log="$tmp_root/relay-spike.log"56+57+# The automated spike advances a bounded virtual media horizon without sleeping.58+# Exact direct/relay-fallback Iroh path measurements remain the manual gate.59+(60+ cd "$repo_root"61+ source "${HOME}/.cargo/env"62+ cargo run --quiet --release -p sigil-host --bin sigil-relay-spike -- \63+ --video "$video" \64+ --audio "$audio" \65+ --duration-seconds "$duration_seconds"66+) | tee "$log"67+68+grep -Fxq 'relay_spike=ok' "$log"69+grep -Fxq 'authentication_mode=ed25519-v1' "$log"70+grep -Fxq 'decision=defer-production-mesh' "$log"71+grep -Fxq 'host_upload_savings_percent=50' "$log"72+grep -Fxq 'tamper_rejected=ok' "$log"73+grep -Fxq 'wrong_subscriber_rejected=ok' "$log"74+grep -Fxq 'expired_subscription_rejected=ok' "$log"75+grep -Fxq 'relay_loss_direct_fallback=ok' "$log"76+grep -Fxq 'withheld_media_trusted=0' "$log"77+grep -Eq '^maximum_relay_queue=[1-4]$' "$log"78+79+printf 'proof=ok\n'@@ -1,6 +1,6 @@
11 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};22 use std::sync::{Arc, Mutex as StdMutex};3-use std::time::{Duration, Instant};3+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};44 55 use iroh::{Endpoint, SecretKey, endpoint::presets};66 use serde::Serialize;@@ -28,8 +28,10 @@ use crate::media::input_delivery::{
2828 open_input_session, run_input_feedback, run_input_forwarder, v2_input_eligibility,2929 };3030 use crate::media::media_control::run_media_control_writer_v3;31-use crate::media::moq_receiver::open_upstream_moq_media;32-use crate::media::network_diagnostics::NetworkSessionDiagnostics;31+use crate::media::moq_receiver::{MoqAuthenticationExpectation, open_upstream_moq_media};32+use crate::media::network_diagnostics::{33+ MediaDeliveryRole, NetworkSessionDiagnostics, lock_network_diagnostics,34+};3335 use crate::media::session_control::{3436 SESSION_CONTROL_COMMAND_CAPACITY, install_initial_snapshot, run_session_control,3537 };@@ -216,24 +218,51 @@ pub(crate) async fn connect_client(
216218 NegotiatedMediaStream::V2(negotiation) => Some(negotiation.initial_snapshot.clone()),217219 NegotiatedMediaStream::V1(_) => None,218220 };221+ let moq_authentication = match &media_negotiation {222+ NegotiatedMediaStream::V2(negotiation) => Some(MoqAuthenticationExpectation {223+ expected_host: *host_node_id.as_bytes(),224+ subscription_capability: negotiation.media_subscription_capability.clone(),225+ now_unix: SystemTime::now()226+ .duration_since(UNIX_EPOCH)227+ .map_err(|_| "System clock is before the Unix epoch".to_string())?228+ .as_secs(),229+ }),230+ NegotiatedMediaStream::V1(_) => None,231+ };219232 let media_generation = next_media_generation(&state.client_media_generation)?;233+ let network_diagnostics = Arc::new(StdMutex::new(NetworkSessionDiagnostics::new(234+ Instant::now(),235+ false,236+ )));237+ if moq_authentication.is_some() {238+ lock_network_diagnostics(&network_diagnostics)239+ .configure_authenticated_media(MediaDeliveryRole::DirectHost);240+ }241+ let media_verification_failures =242+ lock_network_diagnostics(&network_diagnostics).media_verification_failure_counter();220243 let (upstream_moq_media, frame_connection_for_stats) = if media_transport221244 == MediaTransport::UpstreamMoq222245 {223- let (receiver, diagnostics_connection) =224- match open_upstream_moq_media(&endpoint, &addr, media_session_id).await {225- Ok(media) => media,226- Err(error) => {227- // CONTROL already authenticated and owns the host's228- // one-client lease. A post-auth MoQ failure is229- // terminal, and must explicitly release that lease;230- // it must never fall through to a legacy media ALPN.231- frame_conn.close(1_u32.into(), b"upstream MoQ setup failed");232- let _ =233- tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close()).await;234- return Err(error);235- }236- };246+ let (receiver, diagnostics_connection) = match open_upstream_moq_media(247+ &endpoint,248+ &addr,249+ media_session_id,250+ moq_authentication,251+ media_verification_failures,252+ )253+ .await254+ {255+ Ok(media) => media,256+ Err(error) => {257+ // CONTROL already authenticated and owns the host's258+ // one-client lease. A post-auth MoQ failure is259+ // terminal, and must explicitly release that lease;260+ // it must never fall through to a legacy media ALPN.261+ frame_conn.close(1_u32.into(), b"upstream MoQ setup failed");262+ let _ = tokio::time::timeout(CLIENT_ENDPOINT_CLOSE_TIMEOUT, endpoint.close()).await;263+ return Err(error);264+ }265+ };237266 (Some(receiver), diagnostics_connection)238267 } else {239268 (None, frame_conn.clone())@@ -267,10 +296,6 @@ pub(crate) async fn connect_client(
267296 }268297 };269298 270- let mut network_diagnostics = Arc::new(StdMutex::new(NetworkSessionDiagnostics::new(271- Instant::now(),272- false,273- )));274299 let mut input_connection_for_stats = None;275300 let (input_availability, v1_input_forwarder) = if control_protocol == ControlProtocol::V1 {276301 let InputSession {@@ -280,10 +305,8 @@ pub(crate) async fn connect_client(
280305 capabilities: input_capabilities,281306 availability: input_availability,282307 } = open_input_session(&endpoint, &addr, handshake_nonce, media_session_id, grants).await?;283- network_diagnostics = Arc::new(StdMutex::new(NetworkSessionDiagnostics::new(284- Instant::now(),285- input_availability.input_ack,286- )));308+ lock_network_diagnostics(&network_diagnostics)309+ .set_input_ack_negotiated(input_availability.input_ack);287310 input_connection_for_stats = Some(input_connection.clone());288311 if input_availability.pointer_position_feedback || input_availability.input_ack {289312 tokio::spawn(run_input_feedback(@@ -1,10 +1,14 @@
11 use std::time::Duration;22 33 use moq_net::{BroadcastConsumer, Error as MoqError, Track, TrackConsumer};4-use sigil_protocol::{GoqCatalogDocument, MAX_MOQ_CATALOG_BYTES};4+use sigil_protocol::{5+ GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES,6+ SignedMediaGenerationCertificate,7+};58 69 710 pub(crate) enum MoqCatalogMode {11+ GoqV2Authenticated,812 GoqV1,913 AbsentStaticTrackCompat,1014 }@@ -12,6 +16,7 @@ pub(crate) enum MoqCatalogMode {
1216 impl MoqCatalogMode {1317 pub(crate) const fn label(self) -> &'static str {1418 match self {19+ Self::GoqV2Authenticated => "goq-v2-authenticated",1520 Self::GoqV1 => "goq-v1",1621 Self::AbsentStaticTrackCompat => "absent-static-track-compat",1722 }@@ -21,6 +26,7 @@ impl MoqCatalogMode {
2126 pub(crate) struct MoqCatalogSelection {2227 pub(crate) track: TrackConsumer,2328 pub(crate) mode: MoqCatalogMode,29+ pub(crate) certificate: Option<SignedMediaGenerationCertificate>,2430 }2531 2632 fn is_track_not_found(error: &MoqError) -> bool {@@ -39,6 +45,7 @@ fn subscribe_static_video_track(
3945 Ok(MoqCatalogSelection {4046 track,4147 mode: MoqCatalogMode::AbsentStaticTrackCompat,48+ certificate: None,4249 })4350 }4451 @@ -58,13 +65,68 @@ pub(crate) async fn subscribe_goq_video_track(
5865 }5966 };6067 61- let mut catalog_track = catalog_track;68+ let snapshot = match read_catalog_snapshot(catalog_track, timeout).await {69+ Err(error) if error == "catalog-not-found" => {70+ return subscribe_static_video_track(broadcast);71+ }72+ result => result?,73+ };74+ let document: GoqCatalogDocument = serde_json::from_slice(&snapshot)75+ .map_err(|error| format!("Failed to decode Goq catalog snapshot: {error}"))?;76+ document77+ .validate()78+ .map_err(|error| format!("Invalid Goq catalog: {error}"))?;79+ let track = broadcast80+ .subscribe_track(&Track::new(document.goq.video.track.name))81+ .map_err(|error| {82+ format!("Failed to subscribe to catalog-selected Goq video track: {error}")83+ })?;84+ Ok(MoqCatalogSelection {85+ track,86+ mode: MoqCatalogMode::GoqV1,87+ certificate: None,88+ })89+}90+91+pub(crate) async fn subscribe_authenticated_goq_video_track(92+ broadcast: &BroadcastConsumer,93+ timeout: Duration,94+ expected_host: [u8; 32],95+ expected_generation: u64,96+ now_unix: u64,97+) -> Result<MoqCatalogSelection, String> {98+ let catalog_track = broadcast99+ .subscribe_track(&hang::Catalog::default_track())100+ .map_err(|error| format!("Authenticated v2 media requires catalog.json: {error}"))?;101+ let snapshot = read_catalog_snapshot(catalog_track, timeout).await?;102+ let document: GoqCatalogDocumentV2 = serde_json::from_slice(&snapshot)103+ .map_err(|error| format!("Failed to decode authenticated Goq catalog: {error}"))?;104+ let certificate = document105+ .validate()106+ .map_err(|error| format!("Invalid authenticated Goq catalog: {error}"))?;107+ certificate108+ .verify_binding(expected_host, expected_generation, now_unix)109+ .map_err(|error| format!("Invalid media generation certificate: {error}"))?;110+ let track = broadcast111+ .subscribe_track(&Track::new(document.goq.video.track.name))112+ .map_err(|error| {113+ format!("Failed to subscribe to authenticated Goq video track: {error}")114+ })?;115+ Ok(MoqCatalogSelection {116+ track,117+ mode: MoqCatalogMode::GoqV2Authenticated,118+ certificate: Some(certificate),119+ })120+}121+122+async fn read_catalog_snapshot(123+ mut catalog_track: TrackConsumer,124+ timeout: Duration,125+) -> Result<bytes::Bytes, String> {62126 let deadline = tokio::time::Instant::now() + timeout;63127 let mut group = match tokio::time::timeout_at(deadline, catalog_track.next_group()).await {64128 Err(_) => return Err("Timed out waiting for Goq catalog snapshot".to_string()),65- Ok(Err(error)) if is_track_not_found(&error) => {66- return subscribe_static_video_track(broadcast);67- }129+ Ok(Err(error)) if is_track_not_found(&error) => return Err("catalog-not-found".to_string()),68130 Ok(Err(error)) => return Err(format!("Failed to read Goq catalog snapshot: {error}")),69131 Ok(Ok(None)) => return Err("Goq catalog track ended before its snapshot".to_string()),70132 Ok(Ok(Some(group))) => group,@@ -89,31 +151,36 @@ pub(crate) async fn subscribe_goq_video_track(
89151 "Goq catalog snapshot exceeds {MAX_MOQ_CATALOG_BYTES} bytes"90152 ));91153 }92- let snapshot = tokio::time::timeout_at(deadline, frame.read_all())154+ tokio::time::timeout_at(deadline, frame.read_all())93155 .await94156 .map_err(|_| "Timed out reading Goq catalog frame".to_string())?95- .map_err(|error| format!("Failed to read Goq catalog frame: {error}"))?;96- let document: GoqCatalogDocument = serde_json::from_slice(&snapshot)97- .map_err(|error| format!("Failed to decode Goq catalog snapshot: {error}"))?;98- document99- .validate()100- .map_err(|error| format!("Invalid Goq catalog: {error}"))?;101- let track = broadcast102- .subscribe_track(&Track::new(document.goq.video.track.name))103- .map_err(|error| {104- format!("Failed to subscribe to catalog-selected Goq video track: {error}")105- })?;106- Ok(MoqCatalogSelection {107- track,108- mode: MoqCatalogMode::GoqV1,109- })157+ .map_err(|error| format!("Failed to read Goq catalog frame: {error}"))110158 }111159 112160 113161 mod tests {114162 use super::*;163+ use ed25519_dalek::SigningKey;115164 use moq_net::Broadcast;116165 166+ fn v2_document() -> (GoqCatalogDocumentV2, [u8; 32]) {167+ let host = SigningKey::from_bytes(&[7; 32]);168+ let generation = sigil_protocol::MediaGenerationSigningKey::from_bytes(&[9; 32]);169+ let certificate = generation170+ .certify(171+ host.verifying_key().to_bytes(),172+ &[7; 32],173+ 42,174+ 1_700_000_000,175+ 1_700_000_600,176+ )177+ .unwrap();178+ (179+ GoqCatalogDocumentV2::video_h264(42, &certificate).unwrap(),180+ host.verifying_key().to_bytes(),181+ )182+ }183+117184 fn publish_catalog(118185 broadcast: &mut moq_net::BroadcastProducer,119186 document: &GoqCatalogDocument,@@ -127,6 +194,19 @@ mod tests {
127194 track128195 }129196 197+ fn publish_catalog_v2(198+ broadcast: &mut moq_net::BroadcastProducer,199+ document: &GoqCatalogDocumentV2,200+ ) -> moq_net::TrackProducer {201+ let mut track = broadcast202+ .create_track(hang::Catalog::default_track())203+ .unwrap();204+ track205+ .write_frame(serde_json::to_vec(document).unwrap())206+ .unwrap();207+ track208+ }209+130210 131211 async fn catalog_selects_the_exact_goq_video_track() {132212 let mut broadcast = Broadcast::new().produce();@@ -140,6 +220,44 @@ mod tests {
140220 assert_eq!(selection.mode, MoqCatalogMode::GoqV1);141221 }142222 223+ 224+ async fn authenticated_catalog_verifies_host_generation_and_mode() {225+ let mut broadcast = Broadcast::new().produce();226+ let _video = broadcast227+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))228+ .unwrap();229+ let (document, host) = v2_document();230+ let _catalog = publish_catalog_v2(&mut broadcast, &document);231+ let selection = subscribe_authenticated_goq_video_track(232+ &broadcast.consume(),233+ Duration::from_millis(100),234+ host,235+ 42,236+ 1_700_000_300,237+ )238+ .await239+ .unwrap();240+ assert_eq!(selection.mode, MoqCatalogMode::GoqV2Authenticated);241+ assert_eq!(selection.certificate.unwrap().claims.generation_id, 42);242+243+ let mut wrong_host_broadcast = Broadcast::new().produce();244+ let _video = wrong_host_broadcast245+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))246+ .unwrap();247+ let _catalog = publish_catalog_v2(&mut wrong_host_broadcast, &document);248+ assert!(249+ subscribe_authenticated_goq_video_track(250+ &wrong_host_broadcast.consume(),251+ Duration::from_millis(100),252+ [1; 32],253+ 42,254+ 1_700_000_300,255+ )256+ .await257+ .is_err()258+ );259+ }260+143261 144262 async fn only_an_absent_catalog_uses_the_legacy_static_track() {145263 let mut broadcast = Broadcast::new().produce();@@ -1,4 +1,5 @@
1-use std::sync::atomic::Ordering;1+use std::sync::Arc;2+use std::sync::atomic::{AtomicU64, Ordering};23 use std::time::Duration;34 45 use iroh::Endpoint;@@ -9,16 +10,20 @@ use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};
910 1011 use sigil_protocol::MOQ_VIDEO_H264_TRACK;1112 use sigil_protocol::{12- FrameFlags, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,13+ AuthenticatedMediaObject, FrameFlags, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,1314 MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3, MediaCodec, MediaFrame,14- decode_media_frame_object, media_moq_broadcast_name,15+ MediaObjectCoordinates, MediaTrack, SignedMediaGenerationCertificate,16+ SignedSubscriptionCapability, SubscriptionTracks, decode_media_frame_object,17+ media_moq_broadcast_name,1518 };1619 use tauri::{AppHandle, Manager};1720 1821 use crate::commands::state::AppState;1922 use crate::media::audio_delivery::cancel_audio_generation;2023 use crate::media::frame_channel::{take_generation_owned, take_generation_owned_triple};21-use crate::media::moq_catalog::subscribe_goq_video_track;24+use crate::media::moq_catalog::{25+ subscribe_authenticated_goq_video_track, subscribe_goq_video_track,26+};2227 2328 use crate::media::object_receiver::media_object_frame;2429 use crate::media::transport::CLIENT_ENDPOINT_CLOSE_TIMEOUT;@@ -154,6 +159,8 @@ pub(crate) struct MoqMediaReceiver {
154159 last_frame_sequence: Option<u64>,155160 waiting_for_keyframe: bool,156161 pending_group_recovery: Option<MoqGroupRecovery>,162+ certificate: Option<SignedMediaGenerationCertificate>,163+ verification_failures: Option<Arc<AtomicU64>>,157164 }158165 159166 impl MoqMediaReceiver {@@ -162,6 +169,8 @@ impl MoqMediaReceiver {
162169 session: MoqSession,163170 broadcast: BroadcastConsumer,164171 track: TrackConsumer,172+ certificate: Option<SignedMediaGenerationCertificate>,173+ verification_failures: Option<Arc<AtomicU64>>,165174 ) -> Self {166175 Self {167176 _lifetime: Some(MoqMediaLifetime {@@ -175,6 +184,8 @@ impl MoqMediaReceiver {
175184 last_frame_sequence: None,176185 waiting_for_keyframe: true,177186 pending_group_recovery: None,187+ certificate,188+ verification_failures,178189 }179190 }180191 @@ -192,6 +203,27 @@ impl MoqMediaReceiver {
192203 last_frame_sequence: None,193204 waiting_for_keyframe: true,194205 pending_group_recovery: None,206+ certificate: None,207+ verification_failures: None,208+ }209+ }210+211+ 212+ fn authenticated_for_test(213+ track: TrackConsumer,214+ certificate: SignedMediaGenerationCertificate,215+ verification_failures: Arc<AtomicU64>,216+ ) -> Self {217+ Self {218+ _lifetime: None,219+ track,220+ current_group: None,221+ last_group_sequence: None,222+ last_frame_sequence: None,223+ waiting_for_keyframe: true,224+ pending_group_recovery: None,225+ certificate: Some(certificate),226+ verification_failures: Some(verification_failures),195227 }196228 }197229 @@ -328,7 +360,47 @@ impl MoqMediaReceiver {
328360 Ok(next_group_bytes) => next_group_bytes,329361 Err(error) => return Ok(Some(MoqMediaReadOutcome::Malformed(error))),330362 };331- let frame = match decode_media_frame_object(&object) {363+ let authenticated_coordinates = if self.certificate.is_some() {364+ match AuthenticatedMediaObject::coordinates(&object) {365+ Ok(coordinates) => Some(coordinates),366+ Err(error) => {367+ record_verification_failure(&self.verification_failures);368+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(369+ "Invalid authenticated MoQ media header in group {} object {}: {error}",370+ cursor.sequence, cursor.object_count371+ ))));372+ }373+ }374+ } else {375+ None376+ };377+ let payload = if let (Some(certificate), Some(coordinates)) =378+ (&self.certificate, authenticated_coordinates)379+ {380+ let expected_object_id = u32::try_from(cursor.object_count).map_err(|_| {381+ "Authenticated MoQ object index exceeded the protocol range".to_string()382+ })?;383+ let expected = MediaObjectCoordinates {384+ generation_id: certificate.claims.generation_id,385+ track: MediaTrack::VideoH264,386+ group_id: cursor.sequence,387+ object_id: expected_object_id,388+ flags: coordinates.flags,389+ };390+ match AuthenticatedMediaObject::verify(&object, certificate, expected) {391+ Ok(payload) => payload,392+ Err(error) => {393+ record_verification_failure(&self.verification_failures);394+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(395+ "Authenticated upstream MoQ media verification failed in group {} object {}: {error}",396+ cursor.sequence, cursor.object_count397+ ))));398+ }399+ }400+ } else {401+ object.as_ref()402+ };403+ let frame = match decode_media_frame_object(payload) {332404 Ok(frame) => frame,333405 Err(error) => {334406 return Ok(Some(MoqMediaReadOutcome::Malformed(format!(@@ -337,6 +409,15 @@ impl MoqMediaReceiver {
337409 ))));338410 }339411 };412+ if authenticated_coordinates.is_some_and(|coordinates| {413+ coordinates.flags != u16::from(frame.header.flags.bits())414+ }) {415+ record_verification_failure(&self.verification_failures);416+ return Ok(Some(MoqMediaReadOutcome::Malformed(format!(417+ "Authenticated upstream MoQ flags do not match the media payload in group {} object {}",418+ cursor.sequence, cursor.object_count419+ ))));420+ }340421 341422 let first_object = cursor.object_count == 0;342423 let frame_contiguous = match validate_moq_group_frame(@@ -464,6 +545,8 @@ pub(crate) async fn open_upstream_moq_media(
464545 endpoint: &Endpoint,465546 address: &iroh::EndpointAddr,466547 session_id: u64,548+ authentication: Option<MoqAuthenticationExpectation>,549+ verification_failures: Arc<AtomicU64>,467550 ) -> Result<(MoqMediaReceiver, iroh::endpoint::Connection), String> {468551 let broadcast_name = media_moq_broadcast_name(session_id)469552 .map_err(|error| format!("Invalid MoQ media session name: {error}"))?;@@ -491,18 +574,79 @@ pub(crate) async fn open_upstream_moq_media(
491574 .map_err(|error| {492575 format!("Failed to subscribe to upstream MoQ broadcast {broadcast_name}: {error}")493576 })?;494- let catalog = subscribe_goq_video_track(&broadcast, CLIENT_MOQ_SUBSCRIBE_TIMEOUT).await?;577+ let catalog = if let Some(authentication) = authentication {578+ let capability =579+ SignedSubscriptionCapability::decode(&authentication.subscription_capability)580+ .map_err(|error| format!("Invalid subscription capability: {error}"))?;581+ capability582+ .verify_binding(583+ authentication.expected_host,584+ session_id,585+ *endpoint.id().as_bytes(),586+ SubscriptionTracks::VIDEO_H264,587+ 1,588+ authentication.now_unix,589+ )590+ .map_err(|error| format!("Subscription capability rejected: {error}"))?;591+ subscribe_authenticated_goq_video_track(592+ &broadcast,593+ CLIENT_MOQ_SUBSCRIBE_TIMEOUT,594+ authentication.expected_host,595+ session_id,596+ authentication.now_unix,597+ )598+ .await?599+ } else {600+ subscribe_goq_video_track(&broadcast, CLIENT_MOQ_SUBSCRIBE_TIMEOUT).await?601+ };495602 eprintln!("[client] moq catalog: {}", catalog.mode.label());496603 Ok((497- MoqMediaReceiver::new(moq, session, broadcast, catalog.track),604+ MoqMediaReceiver::new(605+ moq,606+ session,607+ broadcast,608+ catalog.track,609+ catalog.certificate,610+ Some(verification_failures),611+ ),498612 diagnostics_connection,499613 ))500614 }501615 616+fn record_verification_failure(counter: &Option<Arc<AtomicU64>>) {617+ if let Some(counter) = counter {618+ counter.fetch_add(1, Ordering::Relaxed);619+ }620+}621+622+pub(crate) struct MoqAuthenticationExpectation {623+ pub(crate) expected_host: [u8; 32],624+ pub(crate) subscription_capability: String,625+ pub(crate) now_unix: u64,626+}627+502628 503629 mod tests {504630 use super::*;505631 632+ fn authentication_fixture() -> (633+ sigil_protocol::MediaGenerationSigningKey,634+ SignedMediaGenerationCertificate,635+ ) {636+ let host = ed25519_dalek::SigningKey::from_bytes(&[7; 32]);637+ let generation = sigil_protocol::MediaGenerationSigningKey::from_bytes(&[9; 32]);638+ let certificate = generation639+ .certify(640+ host.verifying_key().to_bytes(),641+ &[7; 32],642+ 42,643+ 1_700_000_000,644+ 1_700_000_600,645+ )646+ .unwrap();647+ (generation, certificate)648+ }649+506650 507651 fn upstream_moq_group_ids_detect_only_real_transport_gaps() {508652 assert!(!classify_moq_group_sequence(None, 41).unwrap());@@ -512,6 +656,66 @@ mod tests {
512656 assert!(classify_moq_group_sequence(Some(44), 43).is_err());513657 }514658 659+ 660+ async fn authenticated_receiver_verifies_before_codec_validation_and_counts_tampering() {661+ let mut producer = Track::new(MOQ_VIDEO_H264_TRACK).produce();662+ let (generation, certificate) = authentication_fixture();663+ let failures = Arc::new(AtomicU64::new(0));664+ let mut receiver = MoqMediaReceiver::authenticated_for_test(665+ producer.consume(),666+ certificate,667+ Arc::clone(&failures),668+ );669+ let frame = media_object_frame(1, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));670+ let payload = sigil_protocol::encode_media_frame_object(&frame).unwrap();671+ let mut group = producer.append_group().unwrap();672+ let object = generation673+ .authenticate(674+ MediaObjectCoordinates {675+ generation_id: 42,676+ track: MediaTrack::VideoH264,677+ group_id: group.sequence,678+ object_id: 0,679+ flags: u16::from(frame.header.flags.bits()),680+ },681+ &payload,682+ )683+ .unwrap();684+ group.write_frame(object.into_bytes()).unwrap();685+ group.finish().unwrap();686+ assert!(matches!(687+ receiver.next().await.unwrap(),688+ Some(MoqMediaReadOutcome::Frame { .. })689+ ));690+ assert_eq!(failures.load(Ordering::Relaxed), 0);691+692+ let mut tampered_group = producer.append_group().unwrap();693+ let tampered_frame =694+ media_object_frame(2, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));695+ let payload = sigil_protocol::encode_media_frame_object(&tampered_frame).unwrap();696+ let mut object = generation697+ .authenticate(698+ MediaObjectCoordinates {699+ generation_id: 42,700+ track: MediaTrack::VideoH264,701+ group_id: tampered_group.sequence,702+ object_id: 0,703+ flags: u16::from(tampered_frame.header.flags.bits()),704+ },705+ &payload,706+ )707+ .unwrap()708+ .into_bytes();709+ *object.last_mut().unwrap() ^= 1;710+ tampered_group.write_frame(object).unwrap();711+ tampered_group.finish().unwrap();712+ assert!(matches!(713+ receiver.next().await.unwrap(),714+ Some(MoqMediaReadOutcome::Malformed(_))715+ ));716+ assert_eq!(failures.load(Ordering::Relaxed), 1);717+ }718+515719 516720 fn upstream_moq_group_gap_keyframe_is_the_recovery_barrier() {517721 let keyframe = media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));@@ -1,5 +1,7 @@
11 use std::{22 collections::VecDeque,3+ sync::Arc,4+ sync::atomic::{AtomicU64, Ordering},35 sync::{Mutex as StdMutex, MutexGuard as StdMutexGuard},46 time::{Duration, Instant},57 };@@ -46,6 +48,52 @@ pub(crate) enum NetworkLeg {
4648 Audio,4749 }4850 51+52+53+pub(crate) enum MediaDeliveryRole {54+ 55+ DirectHost,56+ ViewerRelay,57+}58+59+60+struct MediaAuthenticationDiagnostics {61+ mode: &'static str,62+ generation_certified: bool,63+ delivery_role: MediaDeliveryRole,64+ verification_failures: Arc<AtomicU64>,65+}66+67+impl Default for MediaAuthenticationDiagnostics {68+ fn default() -> Self {69+ Self {70+ mode: "transport-authenticated-v1",71+ generation_certified: false,72+ delivery_role: MediaDeliveryRole::DirectHost,73+ verification_failures: Arc::new(AtomicU64::new(0)),74+ }75+ }76+}77+78+impl MediaAuthenticationDiagnostics {79+ fn snapshot(&self) -> MediaAuthenticationSnapshot {80+ MediaAuthenticationSnapshot {81+ mode: self.mode,82+ generation_certified: self.generation_certified,83+ delivery_role: self.delivery_role,84+ verification_failures: self.verification_failures.load(Ordering::Relaxed),85+ }86+ }87+}88+89+90+pub(crate) struct MediaAuthenticationSnapshot {91+ mode: &'static str,92+ generation_certified: bool,93+ delivery_role: MediaDeliveryRole,94+ verification_failures: u64,95+}96+4997 5098 struct BoundedLatencyHistogram {5199 buckets: Box<[u64]>,@@ -487,6 +535,7 @@ pub(crate) struct NetworkSessionDiagnostics {
487535 input: LegDiagnostics,488536 audio: Option<LegDiagnostics>,489537 input_ack: InputAckDiagnostics,538+ media_authentication: MediaAuthenticationDiagnostics,490539 }491540 492541 impl NetworkSessionDiagnostics {@@ -497,9 +546,25 @@ impl NetworkSessionDiagnostics {
497546 input: LegDiagnostics::default(),498547 audio: None,499548 input_ack: InputAckDiagnostics::new(input_ack_negotiated),549+ media_authentication: MediaAuthenticationDiagnostics::default(),500550 }501551 }502552 553+ pub(crate) fn configure_authenticated_media(&mut self, delivery_role: MediaDeliveryRole) {554+ self.media_authentication.mode = sigil_protocol::MediaAuthMode::Ed25519.label();555+ self.media_authentication.generation_certified = true;556+ self.media_authentication.delivery_role = delivery_role;557+ }558+559+ pub(crate) fn media_verification_failure_counter(&self) -> Arc<AtomicU64> {560+ Arc::clone(&self.media_authentication.verification_failures)561+ }562+563+ pub(crate) fn set_input_ack_negotiated(&mut self, negotiated: bool) {564+ debug_assert_eq!(self.input_ack.sent_total, 0);565+ self.input_ack = InputAckDiagnostics::new(negotiated);566+ }567+503568 pub(crate) fn observe_connection(&mut self, leg: NetworkLeg, connection: &Connection) {504569 let diagnostics = match leg {505570 NetworkLeg::Media => &mut self.media,@@ -540,6 +605,7 @@ impl NetworkSessionDiagnostics {
540605 input: self.input.snapshot(),541606 audio: self.audio.as_ref().map(LegDiagnostics::snapshot),542607 input_ack: self.input_ack.snapshot(),608+ media_authentication: self.media_authentication.snapshot(),543609 }544610 }545611 }@@ -552,6 +618,7 @@ pub(crate) struct NetworkDiagnosticsSnapshot {
552618 input: NetworkLegSnapshot,553619 audio: Option<NetworkLegSnapshot>,554620 input_ack: InputAckSnapshot,621+ media_authentication: MediaAuthenticationSnapshot,555622 }556623 557624 @@ -733,4 +800,26 @@ mod tests {
733800 );734801 }735802 }803+804+ 805+ fn authenticated_media_diagnostics_are_truthful_and_expose_no_key_material() {806+ let now = Instant::now();807+ let mut diagnostics = NetworkSessionDiagnostics::new(now, false);808+ diagnostics.configure_authenticated_media(MediaDeliveryRole::ViewerRelay);809+ diagnostics810+ .media_verification_failure_counter()811+ .fetch_add(2, Ordering::Relaxed);812+ let json = serde_json::to_value(diagnostics.snapshot(now)).unwrap();813+ assert_eq!(json["media_authentication"]["mode"], "ed25519-v1");814+ assert_eq!(json["media_authentication"]["generation_certified"], true);815+ assert_eq!(816+ json["media_authentication"]["delivery_role"],817+ "viewer_relay"818+ );819+ assert_eq!(json["media_authentication"]["verification_failures"], 2);820+ let serialized = json.to_string();821+ assert!(!serialized.contains("certificate"));822+ assert!(!serialized.contains("capability"));823+ assert!(!serialized.contains("public_key"));824+ }736825 }@@ -21,6 +21,7 @@ pub(crate) struct NegotiatedV2Stream {
2121 pub(crate) session_id: u64,2222 pub(crate) pointer_surface_dimensions: Option<PointerSurfaceDimensions>,2323 pub(crate) initial_snapshot: SessionSnapshotV2,24+ pub(crate) media_subscription_capability: String,2425 }2526 2627 pub(crate) enum NegotiatedMediaStream {@@ -172,6 +173,9 @@ async fn negotiate_v2(
172173 initial_snapshot: response173174 .snapshot174175 .ok_or_else(|| "Host omitted initial control v2 snapshot".to_string())?,176+ media_subscription_capability: response177+ .media_subscription_capability178+ .ok_or_else(|| "Host omitted control v2 media subscription capability".to_string())?,175179 })176180 }177181