1use std::collections::BTreeSet;23use serde::{Deserialize, Serialize};4use tokio::io::{AsyncRead, AsyncWrite};56use crate::framing::{read_json, write_json};7use crate::{8 Capability, ControllerSlot, KeyframeRequestReasonV3, MAX_INVITATION_TOKEN_LEN,9 PROTOCOL_VERSION_V2, PointerSurfaceDimensions, ProtocolError, Result, SignedInvitation,10 SignedSubscriptionCapability,11};1213const MAX_AGENT_LEN: usize = 128;14const MAX_REJECTION_MESSAGE_LEN: usize = 512;15const MAX_CAPABILITIES: usize = 16;16const MAX_PRESENCE_ID_LEN: usize = 32;17const MAX_BROADCAST_NAME_LEN: usize = 128;1819pub const MAX_SESSION_ROSTER_VIEWERS: usize = 8;20pub const MAX_CONTROL_V2_MESSAGE_LEN: usize = 32 * 1024;21222324pub struct ClientHelloV2 {25 pub version: u16,26 pub agent: String,27 pub nonce: [u8; 16],28 pub capabilities: Vec<Capability>,29 30 pub invitation: Option<String>,31}3233impl ClientHelloV2 {34 pub fn new(agent: impl Into<String>, nonce: [u8; 16], capabilities: Vec<Capability>) -> Self {35 Self {36 version: PROTOCOL_VERSION_V2,37 agent: agent.into(),38 nonce,39 capabilities,40 invitation: None,41 }42 }4344 pub fn with_invitation(mut self, invitation: impl Into<String>) -> Self {45 self.invitation = Some(invitation.into());46 self47 }4849 pub fn validate(&self) -> Result<()> {50 validate_version(self.version)?;51 if self.agent.is_empty() || self.agent.len() > MAX_AGENT_LEN {52 return invalid("client hello v2", "agent must contain 1..=128 UTF-8 bytes");53 }54 validate_capabilities(&self.capabilities)?;55 if let Some(invitation) = &self.invitation {56 if invitation.len() > MAX_INVITATION_TOKEN_LEN {57 return invalid(58 "client hello v2",59 "invitation exceeds the bounded token length",60 );61 }62 SignedInvitation::decode(invitation)?;63 }64 Ok(())65 }66}67686970pub struct ViewerPresenceId(String);7172impl ViewerPresenceId {73 pub fn new(value: impl Into<String>) -> Result<Self> {74 let value = value.into();75 if value.is_empty()76 || value.len() > MAX_PRESENCE_ID_LEN77 || !value78 .bytes()79 .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')80 {81 return invalid(82 "viewer presence id",83 "presence id must contain 1..=32 ASCII letters, digits, or hyphens",84 );85 }86 Ok(Self(value))87 }8889 pub fn as_str(&self) -> &str {90 &self.091 }9293 pub fn validate(&self) -> Result<()> {94 Self::new(self.0.clone()).map(|_| ())95 }96}979899100pub struct ViewerPresenceV2 {101 pub presence_id: ViewerPresenceId,102 pub session_id: u64,103 pub input_capable: bool,104 pub you: bool,105}106107impl ViewerPresenceV2 {108 pub fn validate(&self) -> Result<()> {109 self.presence_id.validate()?;110 if self.session_id == 0 {111 return invalid("viewer presence", "session id must be non-zero");112 }113 Ok(())114 }115}116117118119pub enum FocusTransitionReasonV2 {120 Initial,121 Requested,122 HandoffRequested,123 HandoffApproved,124 HandoffDenied,125 ProposalExpired,126 Preempted,127 ActivationExpired,128 Released,129 Disconnected,130 Replaced,131 Revoked,132}133134135136pub enum FocusStateV2 {137 Vacant {138 slot: ControllerSlot,139 },140 Held {141 slot: ControllerSlot,142 holder: ViewerPresenceId,143 session_id: u64,144 focus_generation: u64,145 },146 Neutralizing {147 slot: ControllerSlot,148 former_holder: ViewerPresenceId,149 former_session_id: u64,150 former_focus_generation: u64,151 transition_id: u64,152 },153}154155impl FocusStateV2 {156 pub fn slot(&self) -> ControllerSlot {157 match self {158 Self::Vacant { slot } | Self::Held { slot, .. } | Self::Neutralizing { slot, .. } => {159 *slot160 }161 }162 }163164 pub fn validate(&self) -> Result<()> {165 self.slot().validate()?;166 match self {167 Self::Vacant { .. } => Ok(()),168 Self::Held {169 holder,170 session_id,171 focus_generation,172 ..173 } => {174 holder.validate()?;175 if *session_id == 0 || *focus_generation == 0 {176 return invalid(177 "focus state",178 "held focus requires non-zero session and focus generations",179 );180 }181 Ok(())182 }183 Self::Neutralizing {184 former_holder,185 former_session_id,186 former_focus_generation,187 transition_id,188 ..189 } => {190 former_holder.validate()?;191 if *former_session_id == 0 || *former_focus_generation == 0 || *transition_id == 0 {192 return invalid(193 "focus state",194 "neutralizing focus requires non-zero generations and transition id",195 );196 }197 Ok(())198 }199 }200 }201}202203204205pub struct FocusProposalV2 {206 pub proposal_id: u64,207 pub slot: ControllerSlot,208 pub requester: ViewerPresenceId,209 pub requester_session_id: u64,210 pub holder: ViewerPresenceId,211 pub holder_session_id: u64,212 pub expires_at_unix_ms: u64,213}214215impl FocusProposalV2 {216 pub fn validate(&self) -> Result<()> {217 if self.proposal_id == 0218 || self.requester_session_id == 0219 || self.holder_session_id == 0220 || self.expires_at_unix_ms == 0221 {222 return invalid(223 "focus proposal",224 "proposal, session, and expiry identifiers must be non-zero",225 );226 }227 self.slot.validate()?;228 self.requester.validate()?;229 self.holder.validate()?;230 if self.requester == self.holder && self.requester_session_id == self.holder_session_id {231 return invalid("focus proposal", "holder cannot request focus from itself");232 }233 Ok(())234 }235}236237238239pub struct MediaGenerationDescriptorV2 {240 pub generation_id: u64,241 pub broadcast_name: String,242}243244impl MediaGenerationDescriptorV2 {245 pub fn validate(&self) -> Result<()> {246 if self.generation_id == 0 {247 return invalid("media generation", "generation id must be non-zero");248 }249 if self.broadcast_name.is_empty()250 || self.broadcast_name.len() > MAX_BROADCAST_NAME_LEN251 || !self.broadcast_name.is_ascii()252 {253 return invalid(254 "media generation",255 "broadcast name must contain 1..=128 ASCII bytes",256 );257 }258 Ok(())259 }260}261262263264pub struct SessionSnapshotV2 {265 pub revision: u64,266 pub self_presence_id: ViewerPresenceId,267 pub viewers: Vec<ViewerPresenceV2>,268 pub focus: FocusStateV2,269 270 pub focus_proposal: Option<FocusProposalV2>,271 pub self_is_focus_owner: bool,272 pub transition_reason: FocusTransitionReasonV2,273 pub media: MediaGenerationDescriptorV2,274}275276impl SessionSnapshotV2 {277 pub fn validate(&self) -> Result<()> {278 if self.revision == 0 {279 return invalid("session snapshot", "revision must be non-zero");280 }281 self.self_presence_id.validate()?;282 if self.viewers.is_empty() || self.viewers.len() > MAX_SESSION_ROSTER_VIEWERS {283 return invalid("session snapshot", "roster must contain 1..=8 viewers");284 }285 let mut identities = BTreeSet::new();286 let mut self_entries = 0;287 for viewer in &self.viewers {288 viewer.validate()?;289 if !identities.insert(viewer.presence_id.clone()) {290 return invalid("session snapshot", "roster contains a duplicate identity");291 }292 if viewer.you {293 self_entries += 1;294 if viewer.presence_id != self.self_presence_id {295 return invalid(296 "session snapshot",297 "the you marker does not match self identity",298 );299 }300 }301 }302 if self_entries != 1 || !identities.contains(&self.self_presence_id) {303 return invalid(304 "session snapshot",305 "roster requires exactly one matching you entry",306 );307 }308 self.focus.validate()?;309 if let FocusStateV2::Held {310 holder, session_id, ..311 } = &self.focus312 && !self313 .viewers314 .iter()315 .any(|viewer| &viewer.presence_id == holder && viewer.session_id == *session_id)316 {317 return invalid("session snapshot", "focus holder is absent from the roster");318 }319 if let Some(proposal) = &self.focus_proposal {320 proposal.validate()?;321 let FocusStateV2::Held {322 slot,323 holder,324 session_id,325 ..326 } = &self.focus327 else {328 return invalid("session snapshot", "focus proposals require a held slot");329 };330 if *slot != proposal.slot331 || holder != &proposal.holder332 || *session_id != proposal.holder_session_id333 {334 return invalid(335 "session snapshot",336 "focus proposal does not match the authoritative holder",337 );338 }339 if !self.viewers.iter().any(|viewer| {340 viewer.presence_id == proposal.requester341 && viewer.session_id == proposal.requester_session_id342 && viewer.input_capable343 }) {344 return invalid(345 "session snapshot",346 "focus proposal requester is absent or not input-capable",347 );348 }349 }350 self.media.validate()351 }352353 pub fn self_viewer(&self) -> &ViewerPresenceV2 {354 self.viewers355 .iter()356 .find(|viewer| viewer.presence_id == self.self_presence_id)357 .expect("validated snapshots contain the self viewer")358 }359360 pub fn self_focus_generation(&self) -> Option<u64> {361 match &self.focus {362 FocusStateV2::Held {363 holder,364 session_id,365 focus_generation,366 ..367 } if holder == &self.self_presence_id368 && *session_id == self.self_viewer().session_id =>369 {370 Some(*focus_generation)371 }372 _ => None,373 }374 }375}376377378379pub enum FocusCommandActionV2 {380 Request,381 Approve,382 Deny,383 Release,384 Preempt,385}386387388389pub struct FocusCommandV2 {390 pub request_id: u64,391 pub action: FocusCommandActionV2,392 pub slot: ControllerSlot,393 pub expected_revision: u64,394 395 pub expected_focus_generation: Option<u64>,396 397 pub expected_proposal_id: Option<u64>,398}399400impl FocusCommandV2 {401 pub fn validate(&self) -> Result<()> {402 if self.request_id == 0 || self.expected_revision == 0 {403 return invalid(404 "focus command",405 "request id and expected revision must be non-zero",406 );407 }408 self.slot.validate()?;409 let focus_generation = self.expected_focus_generation.filter(|value| *value != 0);410 let proposal_id = self.expected_proposal_id.filter(|value| *value != 0);411 match self.action {412 FocusCommandActionV2::Request413 if focus_generation.is_some() || proposal_id.is_some() =>414 {415 invalid("focus command", "focus requests forbid expected ownership")416 }417 FocusCommandActionV2::Approve | FocusCommandActionV2::Deny418 if focus_generation.is_none() || proposal_id.is_none() =>419 {420 invalid(421 "focus command",422 "handoff decisions require focus generation and proposal id",423 )424 }425 FocusCommandActionV2::Release | FocusCommandActionV2::Preempt426 if focus_generation.is_none() || proposal_id.is_some() =>427 {428 invalid(429 "focus command",430 "release and preempt require only a focus generation",431 )432 }433 _ => Ok(()),434 }435 }436}437438439440pub struct FocusCommandResultV2 {441 pub request_id: u64,442 pub accepted: bool,443 pub revision: u64,444 445 pub message: Option<String>,446}447448impl FocusCommandResultV2 {449 pub fn validate(&self) -> Result<()> {450 if self.request_id == 0 || self.revision == 0 {451 return invalid(452 "focus command result",453 "request id and revision must be non-zero",454 );455 }456 if let Some(message) = &self.message457 && (message.is_empty() || message.len() > MAX_REJECTION_MESSAGE_LEN)458 {459 return invalid(460 "focus command result",461 "message must contain 1..=512 UTF-8 bytes",462 );463 }464 if self.accepted && self.message.is_some() {465 return invalid(466 "focus command result",467 "accepted results must not carry a message",468 );469 }470 if !self.accepted && self.message.is_none() {471 return invalid("focus command result", "rejected results require a message");472 }473 Ok(())474 }475}476477478479pub enum ClientControlEnvelopeV2 {480 Focus {481 command: FocusCommandV2,482 },483 Keyframe {484 request_id: u64,485 486 last_sequence: Option<u64>,487 reason: ControlKeyframeReasonV2,488 },489}490491492493pub enum ControlKeyframeReasonV2 {494 Join,495 DecoderReset,496 TransportGap,497 DeliveryTimeout,498 FrontendBackpressure,499}500501impl From<ControlKeyframeReasonV2> for KeyframeRequestReasonV3 {502 fn from(value: ControlKeyframeReasonV2) -> Self {503 match value {504 ControlKeyframeReasonV2::Join => Self::Join,505 ControlKeyframeReasonV2::DecoderReset => Self::DecoderReset,506 ControlKeyframeReasonV2::TransportGap => Self::TransportGap,507 ControlKeyframeReasonV2::DeliveryTimeout => Self::DeliveryTimeout,508 ControlKeyframeReasonV2::FrontendBackpressure => Self::FrontendBackpressure,509 }510 }511}512513impl From<KeyframeRequestReasonV3> for ControlKeyframeReasonV2 {514 fn from(value: KeyframeRequestReasonV3) -> Self {515 match value {516 KeyframeRequestReasonV3::Join => Self::Join,517 KeyframeRequestReasonV3::DecoderReset => Self::DecoderReset,518 KeyframeRequestReasonV3::TransportGap => Self::TransportGap,519 KeyframeRequestReasonV3::DeliveryTimeout => Self::DeliveryTimeout,520 KeyframeRequestReasonV3::FrontendBackpressure => Self::FrontendBackpressure,521 }522 }523}524525impl ClientControlEnvelopeV2 {526 pub fn validate(&self) -> Result<()> {527 match self {528 Self::Focus { command } => command.validate(),529 Self::Keyframe { request_id, .. } if *request_id == 0 => {530 invalid("client control envelope v2", "request id must be non-zero")531 }532 Self::Keyframe { .. } => Ok(()),533 }534 }535}536537538539pub enum ServerControlEnvelopeV2 {540 Snapshot { snapshot: SessionSnapshotV2 },541 FocusResult { result: FocusCommandResultV2 },542}543544impl ServerControlEnvelopeV2 {545 pub fn validate(&self) -> Result<()> {546 match self {547 Self::Snapshot { snapshot } => snapshot.validate(),548 Self::FocusResult { result } => result.validate(),549 }550 }551}552553554555pub struct HostHelloV2 {556 pub version: u16,557 pub accepted: bool,558 pub session_id: Option<u64>,559 pub capabilities: Vec<Capability>,560 pub message: Option<String>,561 562 pub pointer_surface_dimensions: Option<PointerSurfaceDimensions>,563 564 pub snapshot: Option<SessionSnapshotV2>,565 566 pub media_subscription_capability: Option<String>,567 /// Authorization epoch the subscription capability was minted against.568 /// A client cannot know its own enrollment revision, so the host states it569 /// here and `validate` refuses a hello whose capability disagrees.570 571 pub media_authorization_revision: Option<u64>,572}573574impl HostHelloV2 {575 pub fn accepted(576 session_id: u64,577 capabilities: Vec<Capability>,578 snapshot: SessionSnapshotV2,579 ) -> Self {580 Self {581 version: PROTOCOL_VERSION_V2,582 accepted: true,583 session_id: Some(session_id),584 capabilities,585 message: None,586 pointer_surface_dimensions: None,587 snapshot: Some(snapshot),588 media_subscription_capability: None,589 media_authorization_revision: None,590 }591 }592593 pub fn rejected(message: impl Into<String>) -> Self {594 Self {595 version: PROTOCOL_VERSION_V2,596 accepted: false,597 session_id: None,598 capabilities: Vec::new(),599 message: Some(message.into()),600 pointer_surface_dimensions: None,601 snapshot: None,602 media_subscription_capability: None,603 media_authorization_revision: None,604 }605 }606607 pub fn with_pointer_surface_dimensions(mut self, dimensions: PointerSurfaceDimensions) -> Self {608 self.pointer_surface_dimensions = Some(dimensions);609 self610 }611612 pub fn with_media_subscription_capability(613 mut self,614 capability: impl Into<String>,615 authorization_revision: u64,616 ) -> Self {617 self.media_subscription_capability = Some(capability.into());618 self.media_authorization_revision = Some(authorization_revision);619 self620 }621622 pub fn validate(&self) -> Result<()> {623 validate_version(self.version)?;624 validate_capabilities(&self.capabilities)?;625 if self.accepted {626 let session_id =627 self.session_id628 .filter(|id| *id != 0)629 .ok_or(ProtocolError::InvalidMessage {630 message_type: "host hello v2",631 reason: "accepted responses require a non-zero session id",632 })?;633 if self.message.is_some() {634 return invalid("host hello v2", "accepted responses forbid a message");635 }636 let snapshot = self637 .snapshot638 .as_ref()639 .ok_or(ProtocolError::InvalidMessage {640 message_type: "host hello v2",641 reason: "accepted responses require an initial snapshot",642 })?;643 snapshot.validate()?;644 if snapshot.self_viewer().session_id != session_id {645 return invalid(646 "host hello v2",647 "snapshot session does not match host hello",648 );649 }650 match (651 &self.media_subscription_capability,652 self.media_authorization_revision,653 ) {654 (Some(capability), Some(authorization_revision)) => {655 let capability = SignedSubscriptionCapability::decode(capability)?;656 if capability.claims.media_generation_id != snapshot.media.generation_id {657 return invalid(658 "host hello v2",659 "subscription capability generation does not match the snapshot",660 );661 }662 if authorization_revision == 0663 || capability.claims.authorization_revision != authorization_revision664 {665 return invalid(666 "host hello v2",667 "subscription capability revision does not match the advertised authorization revision",668 );669 }670 }671 (None, None) => {}672 _ => {673 return invalid(674 "host hello v2",675 "a subscription capability and its authorization revision must be advertised together",676 );677 }678 }679 } else if self.session_id.is_some()680 || self.snapshot.is_some()681 || self.media_subscription_capability.is_some()682 || self.media_authorization_revision.is_some()683 || self.message.as_ref().is_none_or(|message| {684 message.is_empty() || message.len() > MAX_REJECTION_MESSAGE_LEN685 })686 {687 return invalid(688 "host hello v2",689 "rejections require a bounded message and forbid session state",690 );691 }692 if let Some(dimensions) = self.pointer_surface_dimensions {693 dimensions.validate()?;694 if !self.accepted || !self.capabilities.contains(&Capability::VideoH264) {695 return invalid(696 "host hello v2",697 "pointer dimensions require accepted H.264 media",698 );699 }700 }701 Ok(())702 }703}704705macro_rules! json_io {706 ($read:ident, $write:ident, $ty:ty) => {707 pub async fn $read<R>(reader: &mut R) -> Result<Option<$ty>>708 where709 R: AsyncRead + Unpin,710 {711 let value: Option<$ty> = read_json(reader, MAX_CONTROL_V2_MESSAGE_LEN).await?;712 if let Some(value) = &value {713 value.validate()?;714 }715 Ok(value)716 }717718 pub async fn $write<W>(writer: &mut W, value: &$ty) -> Result<()>719 where720 W: AsyncWrite + Unpin,721 {722 value.validate()?;723 write_json(writer, value, MAX_CONTROL_V2_MESSAGE_LEN).await724 }725 };726}727728json_io!(read_client_hello_v2, write_client_hello_v2, ClientHelloV2);729json_io!(read_host_hello_v2, write_host_hello_v2, HostHelloV2);730json_io!(731 read_client_control_v2,732 write_client_control_v2,733 ClientControlEnvelopeV2734);735json_io!(736 read_server_control_v2,737 write_server_control_v2,738 ServerControlEnvelopeV2739);740741fn validate_version(version: u16) -> Result<()> {742 if version != PROTOCOL_VERSION_V2 {743 return Err(ProtocolError::UnsupportedVersion {744 expected: PROTOCOL_VERSION_V2,745 actual: version,746 });747 }748 Ok(())749}750751fn validate_capabilities(capabilities: &[Capability]) -> Result<()> {752 if capabilities.len() > MAX_CAPABILITIES {753 return invalid("hello v2", "too many capabilities");754 }755 for (index, capability) in capabilities.iter().enumerate() {756 if capabilities[..index].contains(capability) {757 return invalid("hello v2", "duplicate capability");758 }759 }760 Ok(())761}762763fn invalid<T>(message_type: &'static str, reason: &'static str) -> Result<T> {764 Err(ProtocolError::InvalidMessage {765 message_type,766 reason,767 })768}769770771mod tests {772 use tokio::io::{AsyncWriteExt, duplex};773774 use super::*;775 use crate::media_moq_broadcast_name;776777 fn snapshot(revision: u64) -> SessionSnapshotV2 {778 let presence = ViewerPresenceId::new("viewer-0001").unwrap();779 SessionSnapshotV2 {780 revision,781 self_presence_id: presence.clone(),782 viewers: vec![ViewerPresenceV2 {783 presence_id: presence,784 session_id: 7,785 input_capable: true,786 you: true,787 }],788 focus: FocusStateV2::Vacant {789 slot: ControllerSlot::ZERO,790 },791 focus_proposal: None,792 self_is_focus_owner: false,793 transition_reason: FocusTransitionReasonV2::Initial,794 media: MediaGenerationDescriptorV2 {795 generation_id: 7,796 broadcast_name: media_moq_broadcast_name(7).unwrap(),797 },798 }799 }800801 802 fn control_v2_snapshot_json_is_a_golden_vector() {803 assert_eq!(804 serde_json::to_string(&snapshot(1)).unwrap(),805 r#"{"revision":1,"self_presence_id":"viewer-0001","viewers":[{"presence_id":"viewer-0001","session_id":7,"input_capable":true,"you":true}],"focus":{"state":"vacant","slot":0},"self_is_focus_owner":false,"transition_reason":"initial","media":{"generation_id":7,"broadcast_name":"sigil/session/7/video"}}"#806 );807 }808809 810 fn snapshots_reject_zero_revision_duplicate_identity_and_unsupported_slot() {811 assert!(snapshot(0).validate().is_err());812 let mut duplicate = snapshot(1);813 duplicate.viewers.push(duplicate.viewers[0].clone());814 assert!(duplicate.validate().is_err());815 assert!(serde_json::from_str::<ControllerSlot>("1").is_err());816 }817818 fn test_capability(819 host: &ed25519_dalek::SigningKey,820 generation_id: u64,821 authorization_revision: u64,822 ) -> String {823 let subscriber = ed25519_dalek::SigningKey::from_bytes(&[13; 32]);824 crate::SignedSubscriptionCapability::issue(825 crate::SubscriptionClaims::new(826 host.verifying_key().to_bytes(),827 generation_id,828 subscriber.verifying_key().to_bytes(),829 crate::SubscriptionTracks::VIDEO_H264,830 authorization_revision,831 1_000,832 1_600,833 [5; 32],834 1,835 )836 .unwrap(),837 &host.to_bytes(),838 )839 .unwrap()840 .encode()841 }842843 // Regression: enrolled hosts mint capabilities at the viewer's enrollment844 // revision, which is never 1 on a real appliance. The hello must carry that845 // revision so a client never has to guess it.846 847 fn accepted_hello_binds_the_capability_to_the_advertised_authorization_revision() {848 let host = ed25519_dalek::SigningKey::from_bytes(&[11; 32]);849 for revision in [1_u64, 2, 9, u64::MAX] {850 let token = test_capability(&host, 7, revision);851 HostHelloV2::accepted(7, vec![Capability::VideoH264], snapshot(1))852 .with_media_subscription_capability(token.clone(), revision)853 .validate()854 .expect("a hello whose revision matches its capability is valid");855856 let mismatched = HostHelloV2::accepted(7, vec![Capability::VideoH264], snapshot(1))857 .with_media_subscription_capability(token, revision.wrapping_add(1).max(1));858 assert!(859 mismatched.validate().is_err(),860 "revision {revision} must not validate against a mismatched advertisement"861 );862 }863 }864865 866 fn accepted_hello_rejects_a_capability_without_its_authorization_revision() {867 let host = ed25519_dalek::SigningKey::from_bytes(&[12; 32]);868 let mut orphaned = HostHelloV2::accepted(7, vec![Capability::VideoH264], snapshot(1));869 orphaned.media_subscription_capability = Some(test_capability(&host, 7, 4));870 assert!(orphaned.validate().is_err());871872 let mut revision_only = HostHelloV2::accepted(7, vec![Capability::VideoH264], snapshot(1));873 revision_only.media_authorization_revision = Some(4);874 assert!(revision_only.validate().is_err());875876 let mut zero = HostHelloV2::accepted(7, vec![Capability::VideoH264], snapshot(1))877 .with_media_subscription_capability(test_capability(&host, 7, 4), 4);878 zero.media_authorization_revision = Some(0);879 assert!(zero.validate().is_err());880881 let mut rejected = HostHelloV2::rejected("nope");882 rejected.media_authorization_revision = Some(4);883 assert!(rejected.validate().is_err());884 }885886 887 fn focus_commands_are_revision_and_generation_bound() {888 let request = FocusCommandV2 {889 request_id: 1,890 action: FocusCommandActionV2::Request,891 slot: ControllerSlot::ZERO,892 expected_revision: 1,893 expected_focus_generation: None,894 expected_proposal_id: None,895 };896 request.validate().unwrap();897 let mut release = request.clone();898 release.action = FocusCommandActionV2::Release;899 assert!(release.validate().is_err());900 release.expected_focus_generation = Some(2);901 release.validate().unwrap();902 }903904 905 async fn v2_hello_and_envelopes_round_trip_with_clean_eof() {906 let hello = ClientHelloV2::new("portal/0.1.0", [4; 16], vec![Capability::VideoH264]);907 let host = HostHelloV2::accepted(7, vec![Capability::VideoH264], snapshot(1));908 let command = ClientControlEnvelopeV2::Focus {909 command: FocusCommandV2 {910 request_id: 2,911 action: FocusCommandActionV2::Request,912 slot: ControllerSlot::ZERO,913 expected_revision: 1,914 expected_focus_generation: None,915 expected_proposal_id: None,916 },917 };918 let (mut sender, mut receiver) = duplex(4096);919 write_client_hello_v2(&mut sender, &hello).await.unwrap();920 write_host_hello_v2(&mut sender, &host).await.unwrap();921 write_client_control_v2(&mut sender, &command)922 .await923 .unwrap();924 sender.shutdown().await.unwrap();925 assert_eq!(926 read_client_hello_v2(&mut receiver).await.unwrap(),927 Some(hello)928 );929 assert_eq!(read_host_hello_v2(&mut receiver).await.unwrap(), Some(host));930 assert_eq!(931 read_client_control_v2(&mut receiver).await.unwrap(),932 Some(command)933 );934 assert_eq!(read_client_control_v2(&mut receiver).await.unwrap(), None);935 }936937 938 async fn peer_controlled_v2_length_is_bounded_before_allocation() {939 let (mut sender, mut receiver) = duplex(16);940 sender941 .write_all(&((MAX_CONTROL_V2_MESSAGE_LEN as u32) + 1).to_be_bytes())942 .await943 .unwrap();944 assert!(matches!(945 read_client_control_v2(&mut receiver).await,946 Err(ProtocolError::InvalidMessageLength { .. })947 ));948 }949}