wechat_protocol/calls/client/
incoming.rs

1//! Source-selected personal voice acceptance. The embedding owner authenticates
2//! and retains the live carrier and grants the broker winner before `accept`.
3
4use super::*;
5use crate::calls::incoming::{CancelNotification, InviteNotification};
6
7#[derive(Clone, Copy, Debug, PartialEq, Eq)]
8pub enum IncomingPhase {
9    Invited,
10    Acknowledged,
11    Joining,
12    Joined,
13    Ending,
14    TerminalPending,
15    Uncertain,
16    Finished,
17}
18struct Invite {
19    room: VoiceRoom,
20    caller: String,
21    caller_openid: Vec<u8>,
22    callee_openid: Vec<u8>,
23    caller_member: Option<u16>,
24}
25
26/// One incoming direct voice call. ACK is ringing setup, not acceptance. Only
27/// the embedding broker's winning owner may request Join. These methods do not
28/// verify carrier freshness, consume account sync, retry or create history.
29pub struct NativeIncomingSession {
30    base: NativePersonalSession,
31    invite: Invite,
32    phase: IncomingPhase,
33    ack_spent: bool,
34    join_spent: bool,
35    reject_spent: bool,
36}
37impl NativeIncomingSession {
38    /// Current remote microphone state from the authenticated AV snapshot.
39    pub fn peer_microphone_muted(&self) -> Option<bool> {
40        (self.phase == IncomingPhase::Joined)
41            .then(|| self.base.peer_microphone_muted())
42            .flatten()
43    }
44    /// Call only after a current account-owned live notification has been
45    /// retained and a listening controller has reserved the global call slot.
46    #[allow(clippy::too_many_arguments)]
47    pub fn authenticate(
48        client: &impl PersonalTransport,
49        state: &mut AccountState,
50        authority: AppAuthority,
51        host_task: u32,
52        device32: [u8; 32],
53        mut parameters: VoiceParameters,
54        notification: &InviteNotification,
55        expected_caller: &str,
56    ) -> Result<Self> {
57        ensure!(
58            notification.unknown_fields.is_empty(),
59            "Unknown incoming invitation fields"
60        );
61        ensure!(
62            notification.room_type == Some(2)
63                && notification.app_id.as_deref() == Some("wechat")
64                && notification.sub_app_id.as_deref() == Some("voip-2p"),
65            "Unsupported incoming call family"
66        );
67        ensure!(
68            notification.callee_username.as_deref() == Some(state.own_username.as_str())
69                && notification.caller_username.as_deref() == Some(expected_caller)
70                && expected_caller != state.own_username
71                && !expected_caller.is_empty()
72                && expected_caller.len() <= 128,
73            "Incoming account or caller mismatch"
74        );
75        ensure!(
76            notification.usernames.len() <= 2
77                && notification
78                    .usernames
79                    .iter()
80                    .all(|name| name == expected_caller || name == &state.own_username),
81            "Unsupported incoming participants"
82        );
83        let room = VoiceRoom {
84            room: notification
85                .room_id
86                .ok_or_else(|| anyhow::anyhow!("Missing incoming room"))?,
87            room_key: notification
88                .room_key
89                .ok_or_else(|| anyhow::anyhow!("Missing incoming room key"))?,
90            own_member: notification
91                .member_id
92                .ok_or_else(|| anyhow::anyhow!("Missing incoming member"))?,
93        };
94        ensure!(
95            room.room != 0 && room.room_key != 0 && u16::try_from(room.own_member).is_ok(),
96            "Invalid incoming room identity"
97        );
98        let caller_openid = notification
99            .caller_openid
100            .as_deref()
101            .ok_or_else(|| anyhow::anyhow!("Missing incoming caller identity"))?
102            .as_bytes()
103            .to_vec();
104        let callee_openid = notification
105            .callee_openid
106            .as_deref()
107            .ok_or_else(|| anyhow::anyhow!("Missing incoming recipient identity"))?
108            .as_bytes()
109            .to_vec();
110        ensure!(
111            !caller_openid.is_empty()
112                && caller_openid.len() <= 128
113                && !callee_openid.is_empty()
114                && callee_openid.len() <= 128
115                && caller_openid != callee_openid,
116            "Invalid incoming app identities"
117        );
118        let owner = notification.wechat_group_id.as_deref().unwrap_or("");
119        ensure!(owner.len() <= 256, "Incoming owner group too large");
120        let group = fields(&parameters.group)?;
121        ensure!(
122            number(&group, 5, None)? == 1
123                && number(&group, 2, None)? == 0
124                && data(&group, 4)?.is_empty(),
125            "Incoming session requires fresh selected group mode"
126        );
127        parameters.group = bind_owner_group(&parameters.group, owner.as_bytes())?;
128        let caller_member = notification
129            .caller_member_id
130            .map(u16::try_from)
131            .transpose()?;
132        if let Some(member) = caller_member {
133            ensure!(
134                u32::from(member) != room.own_member,
135                "Incoming member collision"
136            );
137        }
138        let mut base = NativePersonalSession::authenticate(
139            client, state, authority, host_task, device32, parameters,
140        )?;
141        base.control = Some(PersonalControl {
142            identity: room,
143            peer: expected_caller.to_owned(),
144            origin: ControlOrigin::Incoming,
145            peer_joined: None,
146            terminal: None,
147            invitation_pending: false,
148            pending_recipients: vec![],
149        });
150        base.phase = PersonalPhase::Ringing;
151        Ok(Self {
152            base,
153            invite: Invite {
154                room,
155                caller: expected_caller.to_owned(),
156                caller_openid,
157                callee_openid,
158                caller_member,
159            },
160            phase: IncomingPhase::Invited,
161            ack_spent: false,
162            join_spent: false,
163            reject_spent: false,
164        })
165    }
166    pub fn phase(&self) -> IncomingPhase {
167        self.phase
168    }
169    /// Latest authenticated local room/member binding. This is private call
170    /// routing data and must not be exposed as a public event or diagnostic.
171    pub fn room(&self) -> VoiceRoom {
172        self.invite.room
173    }
174    /// Perform singleton ringing ACK8069. A successful ACK may be offered to
175    /// controllers, but it grants neither a media stream nor permission to Join.
176    pub fn ringing_ack(
177        &mut self,
178        client: &impl PersonalTransport,
179        state: &mut AccountState,
180        current: AppAuthority,
181        tasks: TaskIds,
182    ) -> Result<()> {
183        self.base.check(state, current)?;
184        ensure!(
185            self.phase == IncomingPhase::Invited && !self.ack_spent,
186            "Incoming ACK already spent"
187        );
188        self.base.reserve(tasks)?;
189        self.ack_spent = true;
190        self.phase = IncomingPhase::Uncertain;
191        let result = self.ringing_ack_once(client, state, tasks);
192        if result
193            .as_ref()
194            .is_err_and(|error| error.is::<CallRpcNotSubmitted>())
195        {
196            // Preserve the source phase for a separately authenticated Cancel;
197            // the spent action is not retried with a new task.
198            self.phase = IncomingPhase::Invited;
199        }
200        result
201    }
202    fn ringing_ack_once(
203        &mut self,
204        client: &impl PersonalTransport,
205        state: &mut AccountState,
206        tasks: TaskIds,
207    ) -> Result<()> {
208        let request = personal::acknowledge_voice(
209            &self.base.parameters.inputs(),
210            self.invite.room,
211            tasks.sdk,
212        )?;
213        let body = self.base.host(
214            client,
215            state,
216            CallHostRoute::RingingAck,
217            request.body(),
218            tasks.host,
219        )?;
220        let outer = fields(&body)?;
221        only(&outer, &[1, 2])?;
222        ensure!(
223            p::optional_data(&outer, 2)?.is_none_or(|auth| auth.is_empty()),
224            "Unexpected incoming auth rotation"
225        );
226        let ack = fields(data(&outer, 1)?)?;
227        only(&ack, &[1, 2, 3, 4, 5, 6, 7, 8, 10])?;
228        ensure!(
229            number(&ack, 1, None)? == self.invite.room.room
230                && number(&ack, 2, None)? == self.invite.room.room_key,
231            "Incoming ACK room mismatch"
232        );
233        let member = u16::try_from(number(&ack, 3, None)?)?;
234        let live_id = number(&ack, 4, Some(0))?;
235        for tag in [5, 6, 7, 8, 10] {
236            p::optional_data(&ack, tag)?;
237        }
238        self.invite.room.own_member = u32::from(member);
239        self.base
240            .control
241            .as_mut()
242            .expect("incoming control")
243            .identity = self.invite.room;
244        let group = fields(&self.base.parameters.group)?;
245        let cloud_group = std::str::from_utf8(data(&group, 4)?)?;
246        self.base.parameters.group =
247            replace_group(&self.base.parameters.group, live_id, cloud_group)?;
248        // ACK6 is MemList(seq1/member2); it is not LiveSync item10(seq3/member4).
249        let (sequence, members) = if let Some(raw) = p::optional_data(&ack, 6)? {
250            let list = fields(raw)?;
251            only(&list, &[1, 2, 3])?;
252            p::optional_data(&list, 3)?;
253            let sequence = u32::try_from(number(&list, 1, Some(0))?)?;
254            let members = if sequence == 0 {
255                BTreeMap::new()
256            } else {
257                parse_roster_entries(
258                    repeated(&list, 2, 16)?,
259                    &self.base.own_username,
260                    &self.invite.caller,
261                )?
262            };
263            (sequence, members)
264        } else {
265            (0, BTreeMap::new())
266        };
267        for (name, entry) in &members {
268            let expected = if name == &self.base.own_username {
269                &self.invite.callee_openid
270            } else {
271                &self.invite.caller_openid
272            };
273            ensure!(
274                &entry.openid == expected,
275                "Incoming ACK app identity mismatch"
276            );
277            if name == &self.base.own_username {
278                ensure!(entry.member == member, "Incoming ACK own member mismatch");
279            } else if let Some(expected) = self.invite.caller_member {
280                ensure!(
281                    entry.member == expected,
282                    "Incoming ACK caller member mismatch"
283                );
284            }
285        }
286        if sequence != 0 {
287            if let Some(action) = termination::classify(
288                termination::Origin::Callee,
289                3,
290                members
291                    .get(&self.base.own_username)
292                    .map(|entry| entry.status),
293                members.get(&self.invite.caller).map(|entry| entry.status),
294                false,
295            )? {
296                ensure!(
297                    matches!(action, RemoteTermination::LocalStop(_)),
298                    "Unsupported incoming ACK terminal operation"
299                );
300                self.base
301                    .control
302                    .as_mut()
303                    .expect("incoming control")
304                    .terminal = Some(action);
305                self.base.phase = PersonalPhase::Finished;
306                self.phase = IncomingPhase::Finished;
307                return Ok(());
308            }
309        }
310        if self.invite.caller_member.is_none() {
311            self.invite.caller_member = members.get(&self.invite.caller).map(|entry| entry.member);
312        }
313        ensure!(
314            self.invite
315                .caller_member
316                .is_none_or(|caller| caller != member)
317                && live_id != 0,
318            "Incomplete incoming ACK control context"
319        );
320        let strategy = p::optional_data(&ack, 7)?.unwrap_or_default();
321        ensure!(
322            number(&fields(strategy)?, 14, Some(0))? & (1 << 36) == 0,
323            "Alternate incoming sync unsupported"
324        );
325        // Original entries below bind identities only. They do not fabricate an
326        // accepted roster when the initial ACK sequence is absent/stale.
327        let mut original_members = BTreeMap::from([(
328            self.base.own_username.clone(),
329            Member {
330                member,
331                status: 0,
332                openid: self.invite.callee_openid.clone(),
333            },
334        )]);
335        if let Some(caller_member) = self.invite.caller_member {
336            original_members.insert(
337                self.invite.caller.clone(),
338                Member {
339                    member: caller_member,
340                    status: 0,
341                    openid: self.invite.caller_openid.clone(),
342                },
343            );
344        }
345        self.base.active = Some(ActiveRoom {
346            identity: self.invite.room,
347            peer: self.invite.caller.clone(),
348            live_id,
349            cookies: p::optional_data(&ack, 5)?.unwrap_or_default().to_vec(),
350            strategy: Zeroizing::new(strategy.to_vec()),
351            members,
352            original_members,
353            unbound_peer_openid: self
354                .invite
355                .caller_member
356                .is_none()
357                .then(|| self.invite.caller_openid.clone()),
358            member_sequence: sequence,
359            cap_sequence: 0,
360            engine: vec![],
361            channel: vec![],
362            av_sequence: 0,
363            av: BTreeMap::new(),
364            media_taken: false,
365        });
366        self.base.phase = PersonalPhase::Ringing;
367        self.phase = IncomingPhase::Acknowledged;
368        Ok(())
369    }
370    /// The broker winner accepts with actual callee Join3361. Timestamp8 is the
371    /// source-selected literal0; this does not reuse a caller Invite response.
372    pub fn accept(
373        &mut self,
374        client: &impl PersonalTransport,
375        state: &mut AccountState,
376        current: AppAuthority,
377        tasks: TaskIds,
378    ) -> Result<()> {
379        self.base.check(state, current)?;
380        ensure!(
381            self.phase == IncomingPhase::Acknowledged && !self.join_spent,
382            "Incoming call not acknowledged or Join spent"
383        );
384        self.base.reserve(tasks)?;
385        self.join_spent = true;
386        self.phase = IncomingPhase::Joining;
387        let result = self.accept_once(client, state, current, tasks);
388        if let Err(error) = &result {
389            if error.is::<CallRpcNotSubmitted>() {
390                self.phase = IncomingPhase::Acknowledged;
391                self.base.phase = PersonalPhase::Ringing;
392            } else if !error.is::<UnsupportedCallMedia>() {
393                self.phase = IncomingPhase::Uncertain;
394                self.base.phase = PersonalPhase::Uncertain;
395            }
396        }
397        result
398    }
399    fn accept_once(
400        &mut self,
401        client: &impl PersonalTransport,
402        state: &mut AccountState,
403        current: AppAuthority,
404        tasks: TaskIds,
405    ) -> Result<()> {
406        let request = personal::join(
407            &self.base.parameters.inputs(),
408            self.invite.room,
409            0,
410            tasks.sdk,
411        )?;
412        let raw = self.transfer(
413            client,
414            state,
415            current,
416            CallOperation::Join,
417            request.body.as_deref().expect("prepared Join"),
418            tasks.host,
419        )?;
420        let joined = fields(&raw)?;
421        only(
422            &joined,
423            &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 16, 17, 18],
424        )?;
425        ensure!(
426            number(&joined, 1, None)? == self.invite.room.room
427                && number(&joined, 2, None)? == self.invite.room.room_key,
428            "Incoming Join room mismatch"
429        );
430        ensure!(
431            matches!(number(&joined, 8, Some(2))?, 2) && number(&joined, 18, Some(0))? == 0,
432            "Unsupported incoming Join mode"
433        );
434        // Tags5/6 are opaque compatibility fields in this selected message.
435        if let Some(value) = p::optional(&joined, 5)? {
436            ensure!(
437                matches!(value, Value::Varint(_)),
438                "Invalid incoming opaque field"
439            );
440        }
441        repeated(&joined, 6, 16)?;
442        let own_member = u16::try_from(number(&joined, 3, None)?)?;
443        let room = VoiceRoom {
444            room: self.invite.room.room,
445            room_key: self.invite.room.room_key,
446            own_member: u32::from(own_member),
447        };
448        let raw_members = data(&joined, 13)?;
449        let members = roster(
450            raw_members,
451            &self.base.own_username,
452            &self.invite.caller,
453            own_member,
454        )?;
455        ensure!(
456            members[&self.invite.caller].status == 1
457                && members[&self.invite.caller].openid == self.invite.caller_openid
458                && members[&self.base.own_username].openid == self.invite.callee_openid
459                && self
460                    .invite
461                    .caller_member
462                    .is_none_or(|member| member == members[&self.invite.caller].member),
463            "Incoming Join participant mismatch"
464        );
465        let control = self.base.control.as_mut().expect("incoming control");
466        control.identity = room;
467        control.peer_joined = Some(true);
468        self.invite.room = room;
469        self.base.phase = PersonalPhase::Joined;
470        self.phase = IncomingPhase::Joined;
471        let strategy = data(&joined, 4)?;
472        let parsed_strategy = fields(strategy)?;
473        ensure!(
474            number(&parsed_strategy, 14, None)? & (1 << 36) == 0,
475            "Alternate incoming sync unsupported"
476        );
477        let live_id = number(&joined, 11, None)?;
478        ensure!(live_id != 0, "Missing incoming live identity");
479        let old_group = fields(&self.base.parameters.group)?;
480        let cloud_group = std::str::from_utf8(data(&old_group, 4)?)?;
481        self.base.parameters.group =
482            replace_group(&self.base.parameters.group, live_id, cloud_group)?;
483        let (cap_sequence, engine, channel) = cap(data(&parsed_strategy, 13)?)?;
484        let (av_sequence, av) = av(
485            data(&joined, 9)?,
486            own_member,
487            members[&self.invite.caller].member,
488        )?;
489        let member_sequence = u32::try_from(number(&fields(raw_members)?, 1, Some(0))?)?;
490        if let Some(acked) = self.base.active.as_ref() {
491            ensure!(
492                member_sequence >= acked.member_sequence,
493                "Incoming Join roster sequence regressed"
494            );
495            if member_sequence == acked.member_sequence && member_sequence != 0 {
496                ensure!(
497                    members
498                        .iter()
499                        .all(|(name, entry)| acked
500                            .members
501                            .get(name)
502                            .is_some_and(|old| old.member == entry.member
503                                && old.openid == entry.openid
504                                && old.status == entry.status)),
505                    "Incoming Join changed a stale roster"
506                );
507            }
508        }
509
510        self.base.active = Some(ActiveRoom {
511            identity: room,
512            peer: self.invite.caller.clone(),
513            live_id,
514            cookies: data(&joined, 12)?.to_vec(),
515            strategy: Zeroizing::new(strategy.to_vec()),
516            original_members: members.clone(),
517            unbound_peer_openid: None,
518            members,
519            member_sequence,
520            cap_sequence,
521            engine,
522            channel,
523            av_sequence,
524            av,
525            media_taken: false,
526        });
527        let _ = self.base.media_ready(state, current)?;
528        Ok(())
529    }
530    fn transfer(
531        &mut self,
532        client: &impl PersonalTransport,
533        state: &mut AccountState,
534        current: AppAuthority,
535        operation: CallOperation,
536        inner: &[u8],
537        host_task: u32,
538    ) -> Result<Vec<u8>> {
539        let parameters = AppFrameParameters {
540            sdk_version: self.base.parameters.sdk_version()?,
541            cloud_device_id: &self.base.parameters.cloud_device_id,
542            host_client_version: state.device.client_version as u32,
543            app_config_version: self.base.parameters.app_config_version,
544        };
545        let body =
546            app_frame::prepare_transfer(&self.base.app, current, operation, inner, &parameters)?;
547        let reply = self
548            .base
549            .host(client, state, CallHostRoute::Transfer, &body, host_task)?;
550        app_frame::decode_transfer_reply(&self.base.app, current, operation, &reply)
551    }
552    /// Decline only the source ACKed-callee state. Pre-ACK busy/other-device
553    /// paths have separate operations and are not substituted here.
554    pub fn reject(
555        &mut self,
556        client: &impl PersonalTransport,
557        state: &mut AccountState,
558        current: AppAuthority,
559        tasks: TaskIds,
560    ) -> Result<()> {
561        self.base.check(state, current)?;
562        ensure!(
563            self.phase == IncomingPhase::Acknowledged && !self.reject_spent,
564            "Incoming rejection unavailable"
565        );
566        self.base.reserve(tasks)?;
567        self.reject_spent = true;
568        self.phase = IncomingPhase::Ending;
569        let result: Result<()> = (|| {
570            let request =
571                personal::reject(&self.base.parameters.inputs(), self.invite.room, tasks.sdk)?;
572            let raw = self.transfer(
573                client,
574                state,
575                current,
576                CallOperation::Reject,
577                request.body.as_deref().expect("prepared Reject"),
578                tasks.host,
579            )?;
580            let reply = fields(&raw)?;
581            only(&reply, &[1, 2, 3])?;
582            if let Some(members) = p::optional_data(&reply, 3)? {
583                fields(members)?;
584            }
585            ensure!(
586                number(&reply, 1, None)? == self.invite.room.room
587                    && number(&reply, 2, None)? == self.invite.room.room_key,
588                "Incoming Reject room mismatch"
589            );
590            Ok(())
591        })();
592        let unsent = result
593            .as_ref()
594            .is_err_and(|error| error.is::<CallRpcNotSubmitted>());
595        self.phase = if result.is_ok() {
596            IncomingPhase::Finished
597        } else if unsent {
598            IncomingPhase::Acknowledged
599        } else {
600            IncomingPhase::Uncertain
601        };
602        self.base.phase = if result.is_ok() {
603            PersonalPhase::Finished
604        } else if unsent {
605            PersonalPhase::Ringing
606        } else {
607            PersonalPhase::Uncertain
608        };
609        result
610    }
611    pub fn poll(
612        &mut self,
613        client: &impl PersonalTransport,
614        state: &mut AccountState,
615        current: AppAuthority,
616        tasks: TaskIds,
617    ) -> Result<SyncProgress> {
618        ensure!(
619            matches!(
620                self.phase,
621                IncomingPhase::Acknowledged | IncomingPhase::Joined
622            ),
623            "Incoming sync unavailable"
624        );
625        let result = self.base.poll(client, state, current, tasks);
626        if result.is_ok() && self.invite.caller_member.is_none() {
627            self.invite.caller_member = self
628                .base
629                .active
630                .as_ref()
631                .and_then(|active| active.original_members.get(&self.invite.caller))
632                .map(|member| member.member);
633        }
634        if self.base.phase == PersonalPhase::Uncertain {
635            self.phase = IncomingPhase::Uncertain;
636        }
637        if let Ok(SyncProgress::Terminal(action)) = result.as_ref() {
638            self.phase = if matches!(action, RemoteTermination::LocalStop(_)) {
639                IncomingPhase::Finished
640            } else {
641                IncomingPhase::TerminalPending
642            };
643        }
644        result
645    }
646    /// Source-classified remote cleanup, separate from local user hangup.
647    pub fn finish_remote(
648        &mut self,
649        client: &impl PersonalTransport,
650        state: &mut AccountState,
651        current: AppAuthority,
652        tasks: TaskIds,
653        talk_success: bool,
654    ) -> Result<()> {
655        ensure!(
656            self.phase == IncomingPhase::TerminalPending,
657            "No incoming remote teardown"
658        );
659        self.phase = IncomingPhase::Ending;
660        let result = self
661            .base
662            .finish_remote(client, state, current, tasks, talk_success);
663        self.phase = if result.is_ok() {
664            IncomingPhase::Finished
665        } else {
666            IncomingPhase::Uncertain
667        };
668        result
669    }
670    pub fn media_ready(&self, state: &AccountState, current: AppAuthority) -> Result<bool> {
671        ensure!(
672            self.phase == IncomingPhase::Joined,
673            "Incoming media unavailable"
674        );
675        self.base.media_ready(state, current)
676    }
677    pub fn take_media(
678        &mut self,
679        state: &AccountState,
680        current: AppAuthority,
681        connection_index: usize,
682        udp_index: usize,
683    ) -> Result<NegotiatedMedia> {
684        ensure!(
685            self.phase == IncomingPhase::Joined,
686            "Incoming media unavailable"
687        );
688        self.base
689            .take_media(state, current, connection_index, udp_index)
690    }
691    pub fn exit(
692        &mut self,
693        client: &impl PersonalTransport,
694        state: &mut AccountState,
695        current: AppAuthority,
696        tasks: TaskIds,
697        talk_success: bool,
698    ) -> Result<()> {
699        ensure!(
700            self.phase == IncomingPhase::Joined,
701            "Incoming exit unavailable"
702        );
703        self.phase = IncomingPhase::Ending;
704        let result = self.base.exit(client, state, current, tasks, talk_success);
705        self.phase = if result.is_ok() {
706            IncomingPhase::Finished
707        } else {
708            IncomingPhase::Uncertain
709        };
710        result
711    }
712    /// Only a separately authenticated, matching cancellation while invited or
713    /// ACKed retires this callee. Joined calls need their own terminal evidence.
714    pub fn retire_cancel(
715        &mut self,
716        state: &AccountState,
717        current: AppAuthority,
718        cancel: &CancelNotification,
719    ) -> Result<bool> {
720        self.base.check(state, current)?;
721        if !matches!(
722            self.phase,
723            IncomingPhase::Invited | IncomingPhase::Acknowledged
724        ) {
725            return Ok(false);
726        }
727        ensure!(
728            cancel.unknown_fields.is_empty(),
729            "Unknown incoming cancellation fields"
730        );
731        if cancel.room_id != Some(self.invite.room.room)
732            || cancel.room_key != Some(self.invite.room.room_key)
733            || cancel.member_id != Some(self.invite.room.own_member)
734        {
735            return Ok(false);
736        }
737        self.phase = IncomingPhase::Finished;
738        self.base.phase = PersonalPhase::Finished;
739        Ok(true)
740    }
741}
742fn bind_owner_group(raw: &[u8], owner: &[u8]) -> Result<Vec<u8>> {
743    let values = fields(raw)?;
744    let old = p::optional_data(&values, 6)?.unwrap_or_default();
745    ensure!(
746        old.is_empty() || old == owner,
747        "Incoming owner-group mismatch"
748    );
749    let mut encoded = Vec::new();
750    for field in values {
751        if field.number == 6 {
752            continue;
753        }
754        encoded.push(match field.value {
755            Value::Varint(value) => p::uint(field.number, value),
756            Value::Bytes(value) => p::bytes(field.number, &value),
757            _ => bail!("Unsupported incoming group wire type"),
758        });
759    }
760    encoded.push(p::bytes(6, owner));
761    p::message(&encoded)
762}
763
764#[cfg(test)]
765mod tests {
766    use super::super::tests::{
767        auth_reply, authority, av_status, capabilities_for_codec, exit_reply, host_reply, member,
768        parameters, state, FakeTransport,
769    };
770    use super::*;
771    fn notification() -> InviteNotification {
772        InviteNotification::decode(
773            &p::message(&[
774                p::uint(1, 7),
775                p::uint(2, 9),
776                p::uint(3, 0),
777                p::uint(4, 2),
778                p::bytes(5, b"peer"),
779                p::bytes(6, b"own"),
780                p::bytes(7, b"peer"),
781                p::bytes(8, b"own"),
782                p::bytes(9, b"own"),
783                p::bytes(9, b"peer"),
784                p::bytes(10, b"wechat"),
785                p::bytes(11, b"voip-2p"),
786                p::uint(12, 1),
787                p::bytes(13, b"owner-group"),
788            ])
789            .unwrap(),
790        )
791        .unwrap()
792    }
793    fn ack_reply() -> Vec<u8> {
794        host_reply(&p::bytes(
795            1,
796            &p::message(&[
797                p::uint(1, 7),
798                p::uint(2, 9),
799                p::uint(3, 0),
800                p::uint(4, 99),
801                p::bytes(5, b"ack-cursor"),
802                p::bytes(
803                    6,
804                    &p::message(&[
805                        p::uint(1, 1),
806                        p::bytes(2, &member("own", 0, 3)),
807                        p::bytes(2, &member("peer", 1, 1)),
808                    ])
809                    .unwrap(),
810                ),
811                p::bytes(7, &p::uint(14, 0)),
812            ])
813            .unwrap(),
814        ))
815    }
816    fn wrapped_reply(command: u32, body: &[u8]) -> Vec<u8> {
817        let frame = app_frame::TaskFrame {
818            command,
819            integers: BTreeMap::from([
820                (4, 0),
821                (7, 0),
822                (8, 0),
823                (21, 0),
824                (22, 17),
825                (23, 0),
826                (25, 0),
827                (26, 0),
828            ]),
829            strings: BTreeMap::new(),
830            payload: p::message(&[p::bytes(1, &[8, 0]), p::uint(2, 17), p::bytes(3, body)])
831                .unwrap(),
832        }
833        .encode()
834        .unwrap();
835        host_reply(&p::message(&[p::uint(1, 0), p::bytes(2, &frame)]).unwrap())
836    }
837    fn join_reply(codec: u64, forged_openid: bool) -> Vec<u8> {
838        let (engine, channel) = capabilities_for_codec(codec);
839        let cap = p::message(&[
840            p::uint(1, 1),
841            p::bytes(
842                2,
843                &p::message(&[p::bytes(1, &engine), p::bytes(2, &channel)]).unwrap(),
844            ),
845        ])
846        .unwrap();
847        let address = p::message(&[p::bytes(1, b"127.0.0.1"), p::uint(2, 6000)]).unwrap();
848        let strategy = p::message(&[
849            p::bytes(10, &[7; 32]),
850            p::bytes(13, &cap),
851            p::uint(14, 0),
852            p::bytes(
853                15,
854                &p::message(&[p::uint(1, 1), p::bytes(4, &address)]).unwrap(),
855            ),
856        ])
857        .unwrap();
858        let mut peer = member("peer", 1, 1);
859        if forged_openid {
860            let mut values = fields(&peer).unwrap();
861            values.retain(|field| field.number != 1);
862            peer = p::message(
863                &values
864                    .into_iter()
865                    .map(|field| match field.value {
866                        Value::Varint(value) => p::uint(field.number, value),
867                        Value::Bytes(value) => p::bytes(field.number, &value),
868                        _ => unreachable!(),
869                    })
870                    .chain([p::bytes(1, b"wrong")])
871                    .collect::<Vec<_>>(),
872            )
873            .unwrap();
874        }
875        let members = p::message(&[
876            p::uint(1, 2),
877            p::bytes(2, &member("own", 0, 1)),
878            p::bytes(2, &peer),
879        ])
880        .unwrap();
881        // This is actual callee JoinRsp: Strategy4, AV9, live11/cookies12,
882        // MemList13. Caller JoinClientRsp uses different numbered boundaries.
883        wrapped_reply(
884            3361,
885            &p::message(&[
886                p::uint(1, 7),
887                p::uint(2, 9),
888                p::uint(3, 0),
889                p::bytes(4, &strategy),
890                p::bytes(6, b"opaque-one"),
891                p::bytes(6, b"opaque-two"),
892                p::uint(5, 0),
893                p::bytes(6, &[]),
894                p::uint(8, 2),
895                p::bytes(9, &av_status(1, 1)),
896                p::uint(11, 99),
897                p::bytes(12, b"joined-cursor"),
898                p::bytes(13, &members),
899            ])
900            .unwrap(),
901        )
902    }
903    fn transport(replies: Vec<(CallHostRoute, Vec<u8>)>) -> FakeTransport {
904        FakeTransport(std::cell::RefCell::new(replies.into()))
905    }
906    fn session(transport: &FakeTransport, state: &mut AccountState) -> NativeIncomingSession {
907        NativeIncomingSession::authenticate(
908            transport,
909            state,
910            authority(),
911            1,
912            [2; 32],
913            parameters(),
914            &notification(),
915            "peer",
916        )
917        .unwrap()
918    }
919    fn tasks(number: u32) -> TaskIds {
920        TaskIds {
921            sdk: number,
922            host: number + 1,
923        }
924    }
925    struct Unsent<'a> {
926        inner: &'a FakeTransport,
927        route: CallHostRoute,
928        attempts: std::cell::Cell<usize>,
929    }
930    impl PersonalTransport for Unsent<'_> {
931        fn call_rpc(
932            &self,
933            state: &mut AccountState,
934            route: CallHostRoute,
935            request: &[u8],
936        ) -> Result<Vec<u8>> {
937            if route == self.route {
938                self.attempts.set(self.attempts.get() + 1);
939                return Err(CallRpcNotSubmitted.into());
940            }
941            self.inner.call_rpc(state, route, request)
942        }
943    }
944    #[test]
945    fn known_unsent_callee_actions_keep_cancel_phase_but_are_never_retried() {
946        for operation in 0..3 {
947            let transport = transport(if operation == 0 {
948                vec![(CallHostRoute::AppAuth, auth_reply())]
949            } else {
950                vec![
951                    (CallHostRoute::AppAuth, auth_reply()),
952                    (CallHostRoute::RingingAck, ack_reply()),
953                ]
954            });
955            let mut state = state();
956            let mut session = session(&transport, &mut state);
957            if operation != 0 {
958                session
959                    .ringing_ack(&transport, &mut state, authority(), tasks(1))
960                    .unwrap();
961            }
962            let unsent = Unsent {
963                inner: &transport,
964                route: if operation == 0 {
965                    CallHostRoute::RingingAck
966                } else {
967                    CallHostRoute::Transfer
968                },
969                attempts: std::cell::Cell::new(0),
970            };
971            for task in [2, 3] {
972                let result = match operation {
973                    0 => session.ringing_ack(&unsent, &mut state, authority(), tasks(task)),
974                    1 => session.accept(&unsent, &mut state, authority(), tasks(task)),
975                    _ => session.reject(&unsent, &mut state, authority(), tasks(task)),
976                };
977                assert!(result.is_err());
978                if task == 2 {
979                    assert!(result.unwrap_err().is::<CallRpcNotSubmitted>());
980                }
981            }
982            assert_eq!(unsent.attempts.get(), 1);
983            assert_eq!(
984                session.phase(),
985                if operation == 0 {
986                    IncomingPhase::Invited
987                } else {
988                    IncomingPhase::Acknowledged
989                }
990            );
991            let cancel = CancelNotification::decode(
992                &p::message(&[p::uint(1, 7), p::uint(2, 9), p::uint(3, 0)]).unwrap(),
993            )
994            .unwrap();
995            assert!(session.retire_cancel(&state, authority(), &cancel).unwrap());
996            assert_eq!(session.phase(), IncomingPhase::Finished);
997            assert!(transport.0.borrow().is_empty());
998        }
999    }
1000    #[test]
1001    fn incoming_reject_uses_its_own_operation_and_cannot_be_replayed() {
1002        let reply = wrapped_reply(
1003            3095,
1004            &p::message(&[p::uint(1, 7), p::uint(2, 9), p::bytes(3, &[])]).unwrap(),
1005        );
1006        let transport = transport(vec![
1007            (CallHostRoute::AppAuth, auth_reply()),
1008            (CallHostRoute::RingingAck, ack_reply()),
1009            (CallHostRoute::Transfer, reply),
1010        ]);
1011        let mut state = state();
1012        let mut session = session(&transport, &mut state);
1013        assert!(session
1014            .reject(&transport, &mut state, authority(), tasks(1))
1015            .is_err());
1016        session
1017            .ringing_ack(&transport, &mut state, authority(), tasks(1))
1018            .unwrap();
1019        session
1020            .reject(&transport, &mut state, authority(), tasks(2))
1021            .unwrap();
1022        assert_eq!(session.phase(), IncomingPhase::Finished);
1023        assert!(session
1024            .reject(&transport, &mut state, authority(), tasks(3))
1025            .is_err());
1026        assert!(transport.0.borrow().is_empty());
1027    }
1028    #[test]
1029    fn incoming_known_unselected_media_keeps_joined_control_for_one_exit() {
1030        let transport = transport(vec![
1031            (CallHostRoute::AppAuth, auth_reply()),
1032            (CallHostRoute::RingingAck, ack_reply()),
1033            (CallHostRoute::Transfer, join_reply(0, false)),
1034            (CallHostRoute::Transfer, exit_reply()),
1035        ]);
1036        let mut state = state();
1037        let mut session = session(&transport, &mut state);
1038        session
1039            .ringing_ack(&transport, &mut state, authority(), tasks(1))
1040            .unwrap();
1041        assert!(session
1042            .accept(&transport, &mut state, authority(), tasks(2))
1043            .unwrap_err()
1044            .is::<UnsupportedCallMedia>());
1045        assert_eq!(session.phase(), IncomingPhase::Joined);
1046        assert!(session.take_media(&state, authority(), 0, 0).is_err());
1047        session
1048            .exit(&transport, &mut state, authority(), tasks(3), false)
1049            .unwrap();
1050        assert!(transport.0.borrow().is_empty());
1051    }
1052    #[test]
1053    fn incoming_participant_mismatch_is_uncertain_and_cannot_be_canceled_as_caller() {
1054        let transport = transport(vec![
1055            (CallHostRoute::AppAuth, auth_reply()),
1056            (CallHostRoute::RingingAck, ack_reply()),
1057            (CallHostRoute::Transfer, join_reply(17, true)),
1058        ]);
1059        let mut state = state();
1060        let mut session = session(&transport, &mut state);
1061        session
1062            .ringing_ack(&transport, &mut state, authority(), tasks(1))
1063            .unwrap();
1064        assert!(session
1065            .accept(&transport, &mut state, authority(), tasks(2))
1066            .is_err());
1067        assert_eq!(session.phase(), IncomingPhase::Uncertain);
1068        assert!(session
1069            .exit(&transport, &mut state, authority(), tasks(3), false)
1070            .is_err());
1071        assert!(transport.0.borrow().is_empty());
1072    }
1073    #[test]
1074    fn incoming_builders_have_source_selected_ack_and_reject_tags() {
1075        let parameters = parameters();
1076        let room = VoiceRoom {
1077            room: 7,
1078            room_key: 9,
1079            own_member: 0,
1080        };
1081        let request = personal::acknowledge_voice(&parameters.inputs(), room, 1).unwrap();
1082        assert_eq!(request.operation, personal::HostOperation::RingingAck);
1083        let wrapper = fields(request.body()).unwrap();
1084        assert_eq!(
1085            wrapper.iter().map(|field| field.number).collect::<Vec<_>>(),
1086            vec![1, 3]
1087        );
1088        let ack = fields(data(&wrapper, 1).unwrap()).unwrap();
1089        assert_eq!(
1090            ack.iter().map(|field| field.number).collect::<Vec<_>>(),
1091            vec![1, 2, 3, 4, 5, 6, 7, 9, 11, 12, 13, 14, 15]
1092        );
1093        assert_eq!(number(&ack, 4, None).unwrap(), 1);
1094        assert_eq!(number(&ack, 13, None).unwrap(), 1);
1095        let reject = personal::reject(&parameters.inputs(), room, 2).unwrap();
1096        assert_eq!(reject.command_id, Some(3095));
1097        let body = fields(reject.body.as_deref().unwrap()).unwrap();
1098        assert_eq!(
1099            body.iter().map(|field| field.number).collect::<Vec<_>>(),
1100            vec![1, 2, 3, 4, 5, 6]
1101        );
1102        assert_eq!(number(&body, 6, None).unwrap(), 1);
1103        let join = personal::join(&parameters.inputs(), room, 0, 3).unwrap();
1104        assert_eq!(
1105            number(&fields(join.body.as_deref().unwrap()).unwrap(), 8, None).unwrap(),
1106            0
1107        );
1108    }
1109
1110    #[test]
1111    fn acked_callee_polls_without_join_and_only_newer_full_roster_handles_elsewhere() {
1112        use super::super::tests::roster_sync;
1113        let transport = transport(vec![
1114            (CallHostRoute::AppAuth, auth_reply()),
1115            (CallHostRoute::RingingAck, ack_reply()),
1116            (
1117                CallHostRoute::LiveSync,
1118                roster_sync(1, Some(1), Some(1), false),
1119            ),
1120            (
1121                CallHostRoute::LiveSync,
1122                roster_sync(2, Some(1), Some(1), true),
1123            ),
1124        ]);
1125        let mut state = state();
1126        let mut session = session(&transport, &mut state);
1127        session
1128            .ringing_ack(&transport, &mut state, authority(), tasks(1))
1129            .unwrap();
1130        assert_eq!(
1131            session
1132                .poll(&transport, &mut state, authority(), tasks(2))
1133                .unwrap(),
1134            SyncProgress::Waiting
1135        );
1136        assert_eq!(session.phase(), IncomingPhase::Acknowledged);
1137        assert_eq!(session.base.active.as_ref().unwrap().member_sequence, 1);
1138        assert_eq!(
1139            session
1140                .poll(&transport, &mut state, authority(), tasks(3))
1141                .unwrap(),
1142            SyncProgress::Terminal(RemoteTermination::LocalStop(
1143                LocalStopReason::HandledElsewhere { sdk_end: 13 }
1144            ))
1145        );
1146        assert_eq!(session.phase(), IncomingPhase::Finished);
1147        assert!(transport.0.borrow().is_empty());
1148        // A zero-sequence ACK may omit both the roster and caller member.
1149        // Bind once from the next authenticated roster, using the invitation's
1150        // expected app identity; a different identity must not establish it.
1151        for wrong_identity in [false, true] {
1152            let empty_ack = host_reply(&p::bytes(
1153                1,
1154                &p::message(&[
1155                    p::uint(1, 7),
1156                    p::uint(2, 9),
1157                    p::uint(3, 0),
1158                    p::uint(4, 99),
1159                    p::bytes(5, b"ack-cursor"),
1160                    p::bytes(6, &p::message(&[p::uint(1, 0), p::bytes(3, b"")]).unwrap()),
1161                ])
1162                .unwrap(),
1163            ));
1164            let wire = self::transport(vec![
1165                (CallHostRoute::AppAuth, auth_reply()),
1166                (CallHostRoute::RingingAck, empty_ack),
1167                (
1168                    CallHostRoute::LiveSync,
1169                    roster_sync(1, Some(3), Some(1), false),
1170                ),
1171            ]);
1172            let mut next_state = super::super::tests::state();
1173            let mut next = self::session(&wire, &mut next_state);
1174            next.invite.caller_member = None;
1175            if wrong_identity {
1176                next.invite.caller_openid = b"different-peer".to_vec();
1177            }
1178            next.ringing_ack(&wire, &mut next_state, authority(), tasks(1))
1179                .unwrap();
1180            assert!(!next
1181                .base
1182                .active
1183                .as_ref()
1184                .unwrap()
1185                .original_members
1186                .contains_key("peer"));
1187            let update = next.poll(&wire, &mut next_state, authority(), tasks(2));
1188            if wrong_identity {
1189                assert!(update.is_err());
1190                assert!(next.invite.caller_member.is_none());
1191            } else {
1192                assert_eq!(update.unwrap(), SyncProgress::Waiting);
1193                assert_eq!(next.invite.caller_member, Some(1));
1194                assert!(next
1195                    .base
1196                    .active
1197                    .as_ref()
1198                    .unwrap()
1199                    .unbound_peer_openid
1200                    .is_none());
1201            }
1202        }
1203    }
1204    #[test]
1205    fn joined_callee_remote_hangup_requests_exit_even_after_peer_is_not_joined() {
1206        use super::super::tests::{recorded_exit_reason, roster_sync, RecordingTransport};
1207        let transport = RecordingTransport {
1208            inner: transport(vec![
1209                (CallHostRoute::AppAuth, auth_reply()),
1210                (CallHostRoute::RingingAck, ack_reply()),
1211                (CallHostRoute::Transfer, join_reply(17, false)),
1212                (
1213                    CallHostRoute::LiveSync,
1214                    roster_sync(3, Some(1), Some(2), false),
1215                ),
1216                (
1217                    CallHostRoute::LiveSync,
1218                    roster_sync(4, Some(1), Some(3), false),
1219                ),
1220                (
1221                    CallHostRoute::LiveSync,
1222                    roster_sync(5, Some(1), Some(1), false),
1223                ),
1224                (
1225                    CallHostRoute::LiveSync,
1226                    roster_sync(6, Some(1), Some(4), false),
1227                ),
1228                (CallHostRoute::Transfer, exit_reply()),
1229            ]),
1230            requests: Default::default(),
1231        };
1232        let mut state = state();
1233        let mut session = NativeIncomingSession::authenticate(
1234            &transport,
1235            &mut state,
1236            authority(),
1237            1,
1238            [2; 32],
1239            parameters(),
1240            &notification(),
1241            "peer",
1242        )
1243        .unwrap();
1244        session
1245            .ringing_ack(&transport, &mut state, authority(), tasks(1))
1246            .unwrap();
1247        session
1248            .accept(&transport, &mut state, authority(), tasks(2))
1249            .unwrap();
1250        for task in [3, 4] {
1251            assert_eq!(
1252                session
1253                    .poll(&transport, &mut state, authority(), tasks(task))
1254                    .unwrap(),
1255                SyncProgress::Waiting
1256            );
1257            assert_eq!(session.phase(), IncomingPhase::Joined);
1258            assert_eq!(session.base.phase(), PersonalPhase::Joined);
1259            assert!(!session.media_ready(&state, authority()).unwrap());
1260        }
1261        assert_eq!(
1262            session
1263                .poll(&transport, &mut state, authority(), tasks(5))
1264                .unwrap(),
1265            SyncProgress::MediaReady
1266        );
1267        assert_eq!(session.phase(), IncomingPhase::Joined);
1268        assert!(matches!(
1269            session
1270                .poll(&transport, &mut state, authority(), tasks(6))
1271                .unwrap(),
1272            SyncProgress::Terminal(RemoteTermination::Exit(RemoteExitReason::LastPeerLeft))
1273        ));
1274        session
1275            .finish_remote(&transport, &mut state, authority(), tasks(7), false)
1276            .unwrap();
1277        assert_eq!(session.phase(), IncomingPhase::Finished);
1278        assert_eq!(
1279            recorded_exit_reason(&transport.requests.borrow().last().unwrap().1),
1280            5
1281        );
1282    }
1283}
wechat_protocol/calls/client/incoming.rs

wechat_protocol/calls/client/
incoming.rs
