#![allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::panic,
clippy::indexing_slicing,
clippy::cast_possible_truncation
)]
#![allow(clippy::similar_names)]
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use sipx_audio::g711;
use sipx_call::{Call, DialOptions, answer, dial};
use sipx_sip::build::ResponseBuilder;
use sipx_sip::{HeaderName, Host, HostName, Method, StatusCode, Uri};
use sipx_transport::{Config as TransportConfig, Handle, Incoming, Target, bind};
use sipx_ua::push::PushService;
use sipx_ua::{Config, GruuKind, InstanceId, UserAgent};
use tokio::sync::mpsc::Receiver;
use tokio::sync::{Mutex, oneshot};
const AOR: &str = "sip:alice@sipx.test";
const DELIVERY_BOUND: Duration = Duration::from_secs(10);
const PURR: &str = "opaque-purr-1";
fn loopback() -> IpAddr {
"127.0.0.1".parse().expect("loopback")
}
async fn local_endpoint() -> (Handle, Receiver<Incoming>) {
bind(TransportConfig::new("127.0.0.1:0".parse().expect("valid")))
.await
.expect("binds")
}
fn aor_uri() -> Uri {
Uri::parse(Bytes::from_static(AOR.as_bytes())).expect("a URI")
}
fn ua_config(contact: String, registrar: Target) -> Config {
Config::new(
format!("<{AOR}>"),
contact,
Uri::sip(Host::Name(HostName::new("sipx.test").expect("valid"))),
registrar,
)
}
fn header(request: &Incoming, name: &HeaderName) -> String {
request
.request
.headers
.value(name)
.map(|raw| String::from_utf8_lossy(&raw).into_owned())
.unwrap_or_default()
}
fn instance_of(contact: &str) -> Option<String> {
let start = contact.find("+sip.instance=\"<")? + "+sip.instance=\"<".len();
Some(contact.get(start..)?.split('>').next()?.to_owned())
}
fn clip(milliseconds: usize) -> Vec<i16> {
(0..milliseconds * 8)
.map(|i| {
let t = f64::from(u32::try_from(i).unwrap_or(0)) / 8000.0;
let envelope = (t * 50.0).min(1.0);
let value = (t * 440.0 * std::f64::consts::TAU).sin() * 12_000.0 * envelope;
i16::try_from(value.round() as i32).unwrap_or(0)
})
.collect()
}
fn peak_of(samples: &[i16]) -> i32 {
samples
.iter()
.map(|sample| i32::from(sample.abs()))
.max()
.unwrap_or(0)
}
async fn carried(from: &Call, to: &Call, samples: &[i16]) -> Vec<i16> {
let (_played, heard) = tokio::join!(
from.media().play(samples, 160),
to.media().record_at_least(samples.len(), DELIVERY_BOUND)
);
heard
}
fn saw_an_invite(arriving: &mut Receiver<Incoming>) -> bool {
let mut seen = false;
while let Ok(request) = arriving.try_recv() {
seen |= request.request.method == Method::Invite;
}
seen
}
fn heard(who: &str, played: &[i16], recorded: &[i16]) {
let loudness = peak_of(played);
assert!(
loudness > 8000,
"the clip is too quiet for equality with it to mean audio arrived: peak {loudness}"
);
assert_eq!(recorded.len(), played.len(), "{who} heard a short clip");
assert_eq!(
g711::ulaw_encode_all(played),
g711::ulaw_encode_all(recorded),
"{who} heard something other than what was played"
);
}
#[derive(Clone, Debug)]
struct Binding {
instance: String,
contact: String,
flow: SocketAddr,
}
fn route(bindings: &[Binding], request_uri: &Uri) -> Vec<SocketAddr> {
match sipx_sip::gruu::gr_value(request_uri) {
Some(instance) => bindings
.iter()
.filter(|binding| binding.instance.as_bytes() == instance)
.map(|binding| binding.flow)
.collect(),
None => bindings.iter().map(|binding| binding.flow).collect(),
}
}
async fn registrar() -> (Target, Arc<Mutex<Vec<Binding>>>) {
let (handle, mut incoming) = local_endpoint().await;
let target = Target::udp(handle.local_addr());
let bindings: Arc<Mutex<Vec<Binding>>> = Arc::new(Mutex::new(Vec::new()));
let held = Arc::clone(&bindings);
tokio::spawn(async move {
while let Some(request) = incoming.recv().await {
if request.request.method != Method::Register {
continue;
}
let contact = header(&request, &HeaderName::Contact);
let Some(instance) = instance_of(&contact) else {
continue;
};
let snapshot = {
let mut bindings = held.lock().await;
bindings.retain(|binding| binding.instance != instance);
bindings.push(Binding {
instance,
contact,
flow: request.source,
});
bindings.clone()
};
let _ = handle
.respond(&request.key, registered(&request, &snapshot))
.await;
}
});
(target, bindings)
}
fn registered(request: &Incoming, bindings: &[Binding]) -> sipx_sip::Response {
let mut builder =
ResponseBuilder::to_request(&request.request, StatusCode::new(200).expect("valid"), "OK")
.expect("builds");
for binding in bindings {
let value = format!(
"{};pub-gruu=\"{AOR};gr={}\";expires=3600",
binding.contact, binding.instance
);
builder = builder
.header(HeaderName::Contact, Bytes::from(value))
.expect("valid");
}
builder.build()
}
async fn instance(registrar: Target) -> (UserAgent, Receiver<Incoming>) {
let (endpoint, arriving) = local_endpoint().await;
let contact = format!("<sip:alice@{}>", endpoint.local_addr());
let mut agent = UserAgent::new(
endpoint,
ua_config(contact, registrar).with_gruu(InstanceId::generate(), GruuKind::Public),
);
agent.register().await.expect("the instance registers");
(agent, arriving)
}
async fn called_at(
caller: &Handle,
gruu: &Uri,
flow: SocketAddr,
mine: &UserAgent,
other: &UserAgent,
arriving: &mut Receiver<Incoming>,
) -> (Call, Call) {
let dialing = tokio::spawn({
let (endpoint, to) = (caller.clone(), gruu.clone());
async move {
dial(
&endpoint,
Target::udp(flow),
&to,
&DialOptions::new("<sip:bob@sipx.test>", loopback()),
)
.await
}
});
let incoming = arriving
.recv()
.await
.expect("the INVITE reached the instance the GRUU names");
assert_eq!(incoming.request.method, Method::Invite);
assert!(
mine.sent_to_our_gruu(&incoming.request),
"the instance did not recognise the call as addressed to its own GRUU"
);
assert!(
!other.sent_to_our_gruu(&incoming.request),
"the other instance would have claimed a call addressed to this one's GRUU"
);
assert!(
!mine
.answer(&incoming)
.await
.expect("the UA admits its own GRUU"),
"the UA consumed an INVITE for its own GRUU instead of leaving it to the call layer"
);
let callee = answer(mine.endpoint(), &incoming, loopback())
.await
.expect("the instance answers the call placed at its GRUU");
let caller = dialing
.await
.expect("the dialling task")
.expect("the call connects");
(caller, callee)
}
async fn audio_passes(caller: &Call, callee: &Call, callee_name: &str) {
let from_caller = clip(200);
let from_callee: Vec<i16> = from_caller.iter().map(|sample| -sample).collect();
heard(
callee_name,
&from_caller,
&carried(caller, callee, &from_caller).await,
);
heard(
"the caller",
&from_callee,
&carried(callee, caller, &from_callee).await,
);
}
#[tokio::test(flavor = "multi_thread")]
#[allow(
clippy::too_many_lines,
reason = "the two contacts and their independent calls are one reachability vector"
)]
async fn each_of_two_registrations_of_an_address_of_record_is_called_individually() {
let (registrar_target, bindings) = registrar().await;
let (one, mut arriving_at_one) = instance(registrar_target.clone()).await;
let (two, mut arriving_at_two) = instance(registrar_target).await;
let bindings = bindings.lock().await.clone();
assert_eq!(
bindings.len(),
2,
"both instances must be currently registered for the same AOR: {bindings:?}"
);
let gruu_of_one = one
.gruus()
.public()
.expect("the registrar issued a public GRUU to the first instance")
.clone();
let gruu_of_two = two
.gruus()
.public()
.expect("the registrar issued a public GRUU to the second instance")
.clone();
assert_ne!(
gruu_of_one.to_string(),
gruu_of_two.to_string(),
"two instances of one AOR were issued the same GRUU, which names neither"
);
let fan_out = route(&bindings, &aor_uri());
assert_eq!(
fan_out.len(),
2,
"the AOR must reach every registration; if it reached one, the clause would be vacuous"
);
let (flow_of_one, flow_of_two) = (
route(&bindings, &gruu_of_one),
route(&bindings, &gruu_of_two),
);
assert_eq!(
flow_of_one,
vec![one.endpoint().local_addr()],
"the GRUU must resolve to the one binding whose instance it names (RFC 5627 §5.2)"
);
assert_eq!(flow_of_two, vec![two.endpoint().local_addr()]);
let (caller_endpoint, _caller_incoming) = local_endpoint().await;
let misdirected = tokio::spawn({
let endpoint = caller_endpoint.clone();
let to = gruu_of_one.clone();
let flow = two.endpoint().local_addr();
async move {
dial(
&endpoint,
Target::udp(flow),
&to,
&DialOptions::new("<sip:bob@sipx.test>", loopback()),
)
.await
}
});
let wrong_invitation = arriving_at_two
.recv()
.await
.expect("the deliberately misrouted INVITE reached the other instance");
assert_eq!(wrong_invitation.request.method, Method::Invite);
assert!(
!two.sent_to_our_gruu(&wrong_invitation.request),
"the second instance took the first instance's GRUU for its own"
);
if !two
.answer(&wrong_invitation)
.await
.expect("the UA decides whether it owns the invitation")
{
let wrongly_answered = answer(two.endpoint(), &wrong_invitation, loopback())
.await
.expect("the wrong instance answers when the UA leaves the INVITE unclaimed");
let wrong_caller = misdirected
.await
.expect("the dialling task")
.expect("the wrong instance established the call");
audio_passes(&wrong_caller, &wrongly_answered, "the wrong instance").await;
panic!(
"an INVITE for the first instance's GRUU was answered by the second and carried audio"
);
}
match misdirected.await.expect("the dialling task") {
Err(sipx_call::Error::Rejected { status: 404, .. }) => {}
other => panic!("a GRUU misrouted to another instance was not refused 404: {other:?}"),
}
let (caller_one, callee_one) = called_at(
&caller_endpoint,
&gruu_of_one,
flow_of_one[0],
&one,
&two,
&mut arriving_at_one,
)
.await;
audio_passes(&caller_one, &callee_one, "the first instance").await;
assert!(
!saw_an_invite(&mut arriving_at_two),
"a call placed at the first instance's GRUU reached the second registration too"
);
let (caller_two, callee_two) = called_at(
&caller_endpoint,
&gruu_of_two,
flow_of_two[0],
&two,
&one,
&mut arriving_at_two,
)
.await;
audio_passes(&caller_two, &callee_two, "the second instance").await;
assert!(
!saw_an_invite(&mut arriving_at_one),
"a call placed at the second instance's GRUU reached the first registration too"
);
}
struct Doorbell;
impl PushService for Doorbell {
fn provider(&self) -> &'static str {
"webpush"
}
fn prid(&self) -> &'static str {
"c1a5b3e7d9f2"
}
fn param(&self) -> Option<&str> {
Some("7f3ad0")
}
}
type Timeline = Arc<Mutex<Vec<&'static str>>>;
async fn push_registrar() -> (Target, Timeline, oneshot::Receiver<SocketAddr>) {
let (handle, mut incoming) = local_endpoint().await;
let target = Target::udp(handle.local_addr());
let timeline: Timeline = Arc::new(Mutex::new(Vec::new()));
let recorder = Arc::clone(&timeline);
let (release, released) = oneshot::channel();
tokio::spawn(async move {
let mut release = Some(release);
while let Some(request) = incoming.recv().await {
if request.request.method != Method::Register {
continue;
}
recorder.lock().await.push("register");
let contact = header(&request, &HeaderName::Contact);
let response = ResponseBuilder::to_request(
&request.request,
StatusCode::new(200).expect("valid"),
"OK",
)
.expect("builds")
.header(
HeaderName::Contact,
Bytes::from(format!("{contact};expires=600")),
)
.expect("valid")
.header(
HeaderName::Other(Bytes::from_static(b"Feature-Caps")),
Bytes::from(format!(
"*;+sip.pns=\"webpush\";+sip.pnsreg=\"120\";+sip.pnspurr=\"{PURR}\""
)),
)
.expect("valid")
.build();
let _ = handle.respond(&request.key, response).await;
if let Some(release) = release.take() {
let _ = release.send(request.source);
}
}
});
(target, timeline, released)
}
#[tokio::test(flavor = "multi_thread")]
async fn a_push_wakes_a_client_that_held_no_connection_into_an_answered_call() {
let (client, mut arriving) = local_endpoint().await;
let contact = format!("<sip:alice@{}>", client.local_addr());
let (registrar_target, timeline, released) = push_registrar().await;
let device = Doorbell.device().expect("valid push parameters");
let mut agent = UserAgent::new(
client,
ua_config(contact, registrar_target).with_push(device),
);
assert!(arriving.try_recv().is_err());
let (caller_endpoint, _caller_incoming) = local_endpoint().await;
let held = tokio::spawn(async move {
let flow = released.await.expect("the binding was refreshed");
dial(
&caller_endpoint,
Target::udp(flow),
&aor_uri(),
&DialOptions::new("<sip:bob@sipx.test>", loopback()),
)
.await
});
timeline.lock().await.push("push");
let pending = agent.woken().await.expect("the binding is refreshed");
assert_eq!(
pending.purr.as_deref(),
Some(PURR),
"§8.2's PURR names the binding the request will be released down"
);
let incoming = arriving
.recv()
.await
.expect("the INVITE the push was sent for");
assert_eq!(incoming.request.method, Method::Invite);
timeline.lock().await.push("invite");
let callee = answer(agent.endpoint(), &incoming, loopback())
.await
.expect("the woken client answers the call");
let caller = held
.await
.expect("the held call's task")
.expect("the call connects");
timeline.lock().await.push("answered");
audio_passes(&caller, &callee, "the woken client").await;
assert_eq!(
*timeline.lock().await,
["push", "register", "invite", "answered"],
"§4.1.3's order is push, then the binding-refresh REGISTER, then the request — and the \
clause is that the request is answered"
);
}