Archive/sipcord-bridge
1
0
Fork 0
mirror of https://github.com/coral/sipcord-bridge.git synced 2026-04-12 12:32:32 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

ye

Browse source
This commit is contained in:
coral 2026-03-20 16:40:47 -07:00
parent 4abaef773b
commit 7ff46977f5
21 changed files with 2142 additions and 2011 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
README.md
View file

@ -2,19 +2,31 @@
This is a slice of the code that powers [SIPcord](https://sipcord.net/) that you can use to self host something similar. It's not the full SIPcord package but rather the core functionality used in SIPcord with ways to build your own backend adapter. SIPcord itself uses this as a component of the full build so the code is the same that runs on the public bridges.
## Help!
This means you have to build the call routing backend yourself. I am including a `static-router` backend which you can use to map extensions in a TOML file like this
```toml
[extensions]
1000 = { guild = 123456789012345620, channel = 987654321012345620 }
2000 = { guild = 123456789012345620, channel = 111222333444555620 }
```
but if you want more fancy routing you have to build it. You can easily use sipcord-bridge as a library and provide your own routers by implementing the `Backend` trait.
I am providing 0 support for this, my goal is to run [sipcord.net](https://sipcord.net/), not support self hosting. If you want to run this self hosted, feel free to use this code but do not ask me for support.
This was written a mix between myself and claude, sure, some of it's big slop but the parts I care about are not.
## I have a feature request!
### Can you help me set this up?
**No.** I am not providing support for this as my goal is to run [sipcord.net](https://sipcord.net/), not support self hosting. If you want to run this self hosted, feel free to use this code but you are on your own here.
### I have a feature request!
**PR's welcome**. No really, feel free to implement it and contribute.
## Acknowledgements
### Acknowledgements
- Thanks to [dusthillguy](https://www.youtube.com/watch?v=IK1ydvw3xkU) for letting me use the song *"Joona Kouvolalainen buttermilk"* as hold music and distribute it.
- Thanks to chrischrome for hosting bridge-use1
- Thanks to [wberg](https://wberg.com/) for hosting `bridge-eu1`
- Thanks to [chrischrome](https://litenet.tel/) for hosting `bridge-use1`
## License
### License
GPLv3
Code is AGPLv3
Dusthillguy track is whatever dusthillguy wishe

4
pjsua/build.rs
View file

@ -7,7 +7,7 @@
//! cmake build (used in Docker to separate the slow C build into its own layer).
use std::env;
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use std::process::Command;
fn main() {
@ -214,7 +214,7 @@ fn main() {
}
/// Build pjproject from source and return include paths.
fn build_from_source(out_dir: &PathBuf) -> Vec<PathBuf> {
fn build_from_source(out_dir: &Path) -> Vec<PathBuf> {
let manifest_dir = PathBuf::from(env::var("CARGO_MANIFEST_DIR").unwrap());
let pjproject_src = manifest_dir.join("pjproject");

171
sipcord-bridge/src/call/mod.rs
View file

@ -40,6 +40,9 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, trace, warn};
use udptl::AsyncUdptlSocket;
/// Type alias for fax session entries stored in the DashMap.
type FaxSessionEntry = (Arc<tokio::sync::Mutex<FaxSession>>, CancellationToken);
/// Ring buffer capacity for Discord→SIP audio (i16 mono @ 16kHz).
/// 3200 samples = 200ms of audio, enough for timing jitter.
const DISCORD_TO_SIP_RING_BUFFER_SIZE: usize = 3200;
@ -107,7 +110,7 @@ struct BridgeContext {
sip_calls: Arc<DashMap<CallId, SipCallInfo>>,
/// Active fax sessions keyed by SIP call ID.
/// Each entry holds the session and a cancellation token for the T.38 processing task.
fax_sessions: Arc<DashMap<CallId, (Arc<tokio::sync::Mutex<FaxSession>>, CancellationToken)>>,
fax_sessions: Arc<DashMap<CallId, FaxSessionEntry>>,
discord_event_tx: Sender<DiscordEvent>,
sip_cmd_tx: Sender<SipCommand>,
sound_manager: Arc<SoundManager>,
@ -127,7 +130,7 @@ pub struct BridgeCoordinator {
sip_calls: Arc<DashMap<CallId, SipCallInfo>>,
/// Active fax sessions keyed by SIP call ID.
/// Each entry holds the session and a cancellation token for the T.38 processing task.
fax_sessions: Arc<DashMap<CallId, (Arc<tokio::sync::Mutex<FaxSession>>, CancellationToken)>>,
fax_sessions: Arc<DashMap<CallId, FaxSessionEntry>>,
/// Stores outbound call requests by tracking_id so the answered handler can retrieve them.
/// Entries are cleaned on answer/fail and periodically swept for stale entries.
outbound_requests: Arc<DashMap<String, OutboundCallRequest>>,
@ -221,28 +224,28 @@ impl BridgeCoordinator {
);
// Check for config-based extension sounds (easter eggs)
if let Ok(ext_num) = extension.parse::<u32>() {
if let Some(sound_name) = sound_manager.get_extension_sound(ext_num) {
info!(
"Extension {} maps to sound '{}' (call {})",
ext_num, sound_name, call_id
);
if let Ok(ext_num) = extension.parse::<u32>()
&& let Some(sound_name) = sound_manager.get_extension_sound(ext_num)
{
info!(
"Extension {} maps to sound '{}' (call {})",
ext_num, sound_name, call_id
);
let sound_manager = sound_manager.clone();
let sip_cmd_tx = sip_cmd_tx.clone();
let sound_name = sound_name.to_string();
let sound_manager = sound_manager.clone();
let sip_cmd_tx = sip_cmd_tx.clone();
let sound_name = sound_name.to_string();
tokio::spawn(async move {
play_extension_sound_and_hangup(
call_id,
&sound_name,
&sound_manager,
&sip_cmd_tx,
)
.await;
});
continue;
}
tokio::spawn(async move {
play_extension_sound_and_hangup(
call_id,
&sound_name,
&sound_manager,
&sip_cmd_tx,
)
.await;
});
continue;
}
// Track this call
@ -325,34 +328,34 @@ impl BridgeCoordinator {
backend.on_call_ended(&sip_call_id_str).await;
});
if let Some((_, call_info)) = sip_calls.remove(&call_id) {
if let Some(channel_id) = call_info.channel_id {
let should_destroy = {
if let Some(mut bridge) = bridges.get_mut(&channel_id) {
bridge.sip_calls.remove(&call_id);
info!(
"Removed call {} from bridge for channel {} ({} callers remaining)",
call_id,
channel_id,
bridge.sip_calls.len()
);
bridge.sip_calls.is_empty()
} else {
false
}
};
if should_destroy {
if let Some((_, call_info)) = sip_calls.remove(&call_id)
&& let Some(channel_id) = call_info.channel_id
{
let should_destroy = {
if let Some(mut bridge) = bridges.get_mut(&channel_id) {
bridge.sip_calls.remove(&call_id);
info!(
"Last caller left, destroying bridge for channel {}",
channel_id
"Removed call {} from bridge for channel {} ({} callers remaining)",
call_id,
channel_id,
bridge.sip_calls.len()
);
cleanup_channel_port(channel_id);
teardown_channel_ring_buffers(channel_id);
bridge.sip_calls.is_empty()
} else {
false
}
};
if let Some((_, bridge)) = bridges.remove(&channel_id) {
bridge.discord_connection.disconnect().await;
}
if should_destroy {
info!(
"Last caller left, destroying bridge for channel {}",
channel_id
);
cleanup_channel_port(channel_id);
teardown_channel_ring_buffers(channel_id);
if let Some((_, bridge)) = bridges.remove(&channel_id) {
bridge.discord_connection.disconnect().await;
}
}
}
@ -366,16 +369,15 @@ impl BridgeCoordinator {
// If no audio was ever received, report no_audio to the coordinator
// so the Discord embed can show a diagnostic message
if rx_count == 0 {
if let Some(call_info) = sip_calls.get(&call_id) {
if let Some(ref tracking_id) = call_info.tracking_id {
info!(
"Call {} had zero RTP packets, reporting no_audio (tracking_id={})",
call_id, tracking_id
);
backend_for_sip.report_call_status(tracking_id, "no_audio");
}
}
if rx_count == 0
&& let Some(call_info) = sip_calls.get(&call_id)
&& let Some(ref tracking_id) = call_info.tracking_id
{
info!(
"Call {} had zero RTP packets, reporting no_audio (tracking_id={})",
call_id, tracking_id
);
backend_for_sip.report_call_status(tracking_id, "no_audio");
}
let _ = sip_cmd_tx.send(SipCommand::Hangup { call_id });
@ -1621,26 +1623,26 @@ async fn play_extension_sound_and_hangup(
}
// Check if this is a streaming sound (large file)
if sound_manager.is_streaming(sound_name) {
if let Some(config) = sound_manager.get_streaming(sound_name) {
info!(
"Starting streaming playback '{}' from {} for call {}",
sound_name,
config.path.display(),
call_id
);
if sound_manager.is_streaming(sound_name)
&& let Some(config) = sound_manager.get_streaming(sound_name)
{
info!(
"Starting streaming playback '{}' from {} for call {}",
sound_name,
config.path.display(),
call_id
);
// Use the new port-based streaming approach
// The audio thread handles timing and the hangup happens automatically when done
let _ = sip_cmd_tx.send(SipCommand::StartStreaming {
call_id,
path: config.path.clone(),
});
// Use the new port-based streaming approach
// The audio thread handles timing and the hangup happens automatically when done
let _ = sip_cmd_tx.send(SipCommand::StartStreaming {
call_id,
path: config.path.clone(),
});
// Don't hangup here - the streaming player will hangup when done
// or when the call ends (detected via CALL_CONF_PORTS check)
return;
}
// Don't hangup here - the streaming player will hangup when done
// or when the call ends (detected via CALL_CONF_PORTS check)
return;
}
// Preloaded sound - play all at once
@ -1805,17 +1807,16 @@ async fn process_fax_audio(
}
let tx_available = tx_producer.slots();
let to_write = tx_generated.min(tx_available);
if to_write > 0 {
if let Ok(mut chunk) = tx_producer.write_chunk(to_write) {
let (first, second) = chunk.as_mut_slices();
let first_len = first.len().min(to_write);
first[..first_len].copy_from_slice(&tx_buf[..first_len]);
if first_len < to_write {
second[..to_write - first_len]
.copy_from_slice(&tx_buf[first_len..to_write]);
}
chunk.commit_all();
if to_write > 0
&& let Ok(mut chunk) = tx_producer.write_chunk(to_write)
{
let (first, second) = chunk.as_mut_slices();
let first_len = first.len().min(to_write);
first[..first_len].copy_from_slice(&tx_buf[..first_len]);
if first_len < to_write {
second[..to_write - first_len].copy_from_slice(&tx_buf[first_len..to_write]);
}
chunk.commit_all();
}
} else {
tx_silent_frames += 1;

102
sipcord-bridge/src/fax/audio_port.rs
View file

@ -256,44 +256,50 @@ unsafe extern "C" fn fax_port_get_frame(
return pj_constants__PJ_SUCCESS as pj_status_t;
}
let call_id_ldata = (*this_port).port_data.ldata;
let call_id_ldata = unsafe { (*this_port).port_data.ldata };
if let Some(consumer_entry) = get_fax_tx_consumers().get(&call_id_ldata) {
if let Some(mut consumer) = consumer_entry.try_lock() {
let available = consumer.slots();
if available >= SAMPLES_PER_FRAME {
if let Ok(chunk) = consumer.read_chunk(SAMPLES_PER_FRAME) {
let (first, second) = chunk.as_slices();
let buf = (*frame).buf as *mut i16;
let out = std::slice::from_raw_parts_mut(buf, SAMPLES_PER_FRAME);
out[..first.len()].copy_from_slice(first);
if !second.is_empty() {
out[first.len()..first.len() + second.len()].copy_from_slice(second);
}
chunk.commit_all();
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
(*frame).size = SAMPLES_PER_FRAME * 2;
return pj_constants__PJ_SUCCESS as pj_status_t;
}
if let Some(consumer_entry) = get_fax_tx_consumers().get(&call_id_ldata)
&& let Some(mut consumer) = consumer_entry.try_lock()
{
let available = consumer.slots();
if available >= SAMPLES_PER_FRAME
&& let Ok(chunk) = consumer.read_chunk(SAMPLES_PER_FRAME)
{
let (first, second) = chunk.as_slices();
let out = unsafe {
let buf = (*frame).buf as *mut i16;
std::slice::from_raw_parts_mut(buf, SAMPLES_PER_FRAME)
};
out[..first.len()].copy_from_slice(first);
if !second.is_empty() {
out[first.len()..first.len() + second.len()].copy_from_slice(second);
}
chunk.commit_all();
unsafe {
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
(*frame).size = SAMPLES_PER_FRAME * 2;
}
return pj_constants__PJ_SUCCESS as pj_status_t;
}
}
// No TX audio available — return silence audio frame (not NONE).
// Returning FRAME_TYPE_NONE can cause PJSIP's conference bridge to
// exclude this port from the audio mix, breaking the TX path.
let buf = (*frame).buf as *mut i16;
let out = std::slice::from_raw_parts_mut(buf, SAMPLES_PER_FRAME);
out.fill(0);
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
(*frame).size = SAMPLES_PER_FRAME * 2;
unsafe {
let buf = (*frame).buf as *mut i16;
let out = std::slice::from_raw_parts_mut(buf, SAMPLES_PER_FRAME);
out.fill(0);
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
(*frame).size = SAMPLES_PER_FRAME * 2;
}
pj_constants__PJ_SUCCESS as pj_status_t
}
/// on_destroy callback — no-op since cleanup is done in remove_fax_audio_port().
/// Required by PJSIP to avoid "on_destroy() not found" warning.
unsafe extern "C" fn fax_port_on_destroy(_this_port: *mut pjmedia_port) -> pj_status_t {
pj_constants__PJ_SUCCESS as pj_status_t
pj_constants__PJ_SUCCESS as pj_status_t // no unsafe ops needed
}
/// put_frame callback — captures SIP audio and pushes to RX ring buffer for SpanDSP.
@ -306,36 +312,40 @@ unsafe extern "C" fn fax_port_put_frame(
}
// Only process audio frames with data
if (*frame).type_ != pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO || (*frame).size == 0 {
if unsafe {
(*frame).type_ != pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO || (*frame).size == 0
} {
return pj_constants__PJ_SUCCESS as pj_status_t;
}
let call_id_ldata = (*this_port).port_data.ldata;
let call_id_ldata = unsafe { (*this_port).port_data.ldata };
// View frame buffer as i16 slice
let num_samples = (*frame).size / 2;
let frame_buf = (*frame).buf as *const i16;
let samples = std::slice::from_raw_parts(frame_buf, num_samples);
let samples = unsafe {
let num_samples = (*frame).size / 2;
let frame_buf = (*frame).buf as *const i16;
std::slice::from_raw_parts(frame_buf, num_samples)
};
// Push to RX ring buffer
if let Some(producer_entry) = get_fax_rx_producers().get(&call_id_ldata) {
if let Some(mut producer) = producer_entry.try_lock() {
let available = producer.slots();
if available >= samples.len() {
if let Ok(mut chunk) = producer.write_chunk(samples.len()) {
let (first, second) = chunk.as_mut_slices();
let first_len = first.len().min(samples.len());
first[..first_len].copy_from_slice(&samples[..first_len]);
if first_len < samples.len() {
second[..samples.len() - first_len].copy_from_slice(&samples[first_len..]);
}
chunk.commit_all();
}
} else {
// Buffer full — fax processing is falling behind. Track the drop.
if let Some(counter) = get_fax_rx_drop_counts().get(&call_id_ldata) {
counter.fetch_add(1, Ordering::Relaxed);
if let Some(producer_entry) = get_fax_rx_producers().get(&call_id_ldata)
&& let Some(mut producer) = producer_entry.try_lock()
{
let available = producer.slots();
if available >= samples.len() {
if let Ok(mut chunk) = producer.write_chunk(samples.len()) {
let (first, second) = chunk.as_mut_slices();
let first_len = first.len().min(samples.len());
first[..first_len].copy_from_slice(&samples[..first_len]);
if first_len < samples.len() {
second[..samples.len() - first_len].copy_from_slice(&samples[first_len..]);
}
chunk.commit_all();
}
} else {
// Buffer full — fax processing is falling behind. Track the drop.
if let Some(counter) = get_fax_rx_drop_counts().get(&call_id_ldata) {
counter.fetch_add(1, Ordering::Relaxed);
}
}
}

29
sipcord-bridge/src/fax/spandsp.rs
View file

@ -115,12 +115,14 @@ unsafe fn configure_log_state(log_state: *mut spandsp_sys::logging_state_t) {
return;
}
let log_level = LogLevel::Flow as i32 | LogShowFlags::TAG.bits();
spandsp_sys::span_log_set_level(log_state, log_level);
spandsp_sys::span_log_set_message_handler(
log_state,
Some(spandsp_log_handler),
std::ptr::null_mut(),
);
unsafe {
spandsp_sys::span_log_set_level(log_state, log_level);
spandsp_sys::span_log_set_message_handler(
log_state,
Some(spandsp_log_handler),
std::ptr::null_mut(),
);
}
}
/// Check fax reception completion status from callback state.
@ -467,8 +469,11 @@ unsafe extern "C" fn tx_packet_handler(
if user_data.is_null() || buf.is_null() || len <= 0 {
return -1;
}
let state = &*(user_data as *const TxCallbackState);
let data = std::slice::from_raw_parts(buf, len as usize);
let (state, data) = unsafe {
let state = &*(user_data as *const TxCallbackState);
let data = std::slice::from_raw_parts(buf, len as usize);
(state, data)
};
debug!("SpanDSP TX IFP: {}B (count={})", len, count);
// Send the packet `count` times as SpanDSP requests.
// For indicator packets (CNG, CED, DIS), count is typically 3 — these
@ -491,7 +496,7 @@ unsafe extern "C" fn tx_packet_handler(
/// Phase B handler: called when T.30 negotiation starts.
unsafe extern "C" fn phase_b_handler(user_data: *mut std::ffi::c_void, result: i32) -> i32 {
if !user_data.is_null() {
let state = &mut *(user_data as *mut FaxCallbackState);
let state = unsafe { &mut *(user_data as *mut FaxCallbackState) };
state.negotiation_started = true;
info!(
"SpanDSP phase B: fax negotiation started (result={})",
@ -504,7 +509,7 @@ unsafe extern "C" fn phase_b_handler(user_data: *mut std::ffi::c_void, result: i
/// Phase D handler: called when a page is received.
unsafe extern "C" fn phase_d_handler(user_data: *mut std::ffi::c_void, result: i32) -> i32 {
if !user_data.is_null() {
let state = &mut *(user_data as *mut FaxCallbackState);
let state = unsafe { &mut *(user_data as *mut FaxCallbackState) };
state.pages_received += 1;
info!(
"SpanDSP phase D: page {} received (result={})",
@ -517,7 +522,7 @@ unsafe extern "C" fn phase_d_handler(user_data: *mut std::ffi::c_void, result: i
/// Phase E handler: called when fax reception completes (success or failure).
unsafe extern "C" fn phase_e_handler(user_data: *mut std::ffi::c_void, completion_code: i32) {
if !user_data.is_null() {
let state = &mut *(user_data as *mut FaxCallbackState);
let state = unsafe { &mut *(user_data as *mut FaxCallbackState) };
state.completion_code = completion_code;
state.completed = true;
@ -550,7 +555,7 @@ unsafe extern "C" fn spandsp_log_handler(
if text.is_null() {
return;
}
let msg = std::ffi::CStr::from_ptr(text).to_string_lossy();
let msg = unsafe { std::ffi::CStr::from_ptr(text) }.to_string_lossy();
let msg = msg.trim_end(); // SpanDSP messages often have trailing newlines
match level {

10
sipcord-bridge/src/fax/tiff_decoder.rs
View file

@ -941,11 +941,11 @@ fn decode_group4(data: &[u8], width: u32, height: u32) -> Result<Vec<Vec<u16>>>
for _ in 0..height {
// Check for EOFB (End Of Facsimile Block): two consecutive EOL codes
if let Some(v) = reader.peek(12) {
if v == 0x001 {
// Possible EOFB — check for second EOL
break;
}
if let Some(v) = reader.peek(12)
&& v == 0x001
{
// Possible EOFB — check for second EOL
break;
}
let line = match decode_line_2d(&mut reader, &reference, w) {

14
sipcord-bridge/src/services/registrar.rs
View file

@ -46,6 +46,12 @@ pub struct Registrar {
discord_to_sip: DashMap<String, String>,
}
impl Default for Registrar {
fn default() -> Self {
Self::new()
}
}
impl Registrar {
pub fn new() -> Self {
Self {
@ -68,10 +74,10 @@ impl Registrar {
.find(|r| r.source_addr == reg.source_addr && r.contact_uri == reg.contact_uri)
{
// If discord_username changed, remove the old reverse mapping
if existing.discord_username != reg.discord_username {
if let Some(ref old_du) = existing.discord_username {
self.discord_to_sip.remove(old_du);
}
if existing.discord_username != reg.discord_username
&& let Some(ref old_du) = existing.discord_username
{
self.discord_to_sip.remove(old_du);
}
existing.expires_at = reg.expires_at;

23
sipcord-bridge/src/transport/discord/mod.rs
View file

@ -1112,19 +1112,18 @@ impl VoiceEventHandler for VoiceReceiver {
// CRITICAL: Skip our own SSRC to prevent feedback loop
// When we send audio to Discord, it comes back in VoiceTick.
// If we don't filter it out, we get: SIP -> Discord -> SIP -> Discord -> ...
if let Some(ref map) = ssrc_map {
if let Some(&user_id) = map.get(ssrc) {
if user_id == self.bot_user_id {
skipped_self = true;
if should_log {
trace!(
"VoiceTick: Skipping bot's own SSRC {} to prevent feedback",
ssrc
);
}
continue;
}
if let Some(ref map) = ssrc_map
&& let Some(&user_id) = map.get(ssrc)
&& user_id == self.bot_user_id
{
skipped_self = true;
if should_log {
trace!(
"VoiceTick: Skipping bot's own SSRC {} to prevent feedback",
ssrc
);
}
continue;
}
if let Some(ref decoded) = voice_data.decoded_voice {

70
sipcord-bridge/src/transport/sip/audio_thread.rs
View file

@ -230,27 +230,27 @@ pub fn stop_audio_thread() {
// If the thread is blocked on a conference bridge lock, we don't want
// shutdown to hang indefinitely. The 2s force-exit timer in main.rs
// is a final backstop, but this avoids relying on a hard process exit.
if let Some(handle_storage) = AUDIO_THREAD_HANDLE.get() {
if let Some(handle) = handle_storage.lock().take() {
tracing::debug!("Joining audio thread (2s timeout)...");
let (done_tx, done_rx) = std::sync::mpsc::channel();
let join_thread = std::thread::spawn(move || {
let result = handle.join();
let _ = done_tx.send(result);
});
match done_rx.recv_timeout(std::time::Duration::from_secs(2)) {
Ok(Ok(())) => {
tracing::debug!("Audio thread joined successfully");
}
Ok(Err(e)) => {
tracing::error!("Audio thread panicked: {:?}", e);
}
Err(_) => {
tracing::warn!("Audio thread join timed out after 2s, detaching");
// Detach the join thread — the audio thread will be
// cleaned up by process exit
drop(join_thread);
}
if let Some(handle_storage) = AUDIO_THREAD_HANDLE.get()
&& let Some(handle) = handle_storage.lock().take()
{
tracing::debug!("Joining audio thread (2s timeout)...");
let (done_tx, done_rx) = std::sync::mpsc::channel();
let join_thread = std::thread::spawn(move || {
let result = handle.join();
let _ = done_tx.send(result);
});
match done_rx.recv_timeout(std::time::Duration::from_secs(2)) {
Ok(Ok(())) => {
tracing::debug!("Audio thread joined successfully");
}
Ok(Err(e)) => {
tracing::error!("Audio thread panicked: {:?}", e);
}
Err(_) => {
tracing::warn!("Audio thread join timed out after 2s, detaching");
// Detach the join thread — the audio thread will be
// cleaned up by process exit
drop(join_thread);
}
}
}
@ -331,15 +331,15 @@ fn process_pending_pjsua_ops() {
PendingPjsuaOp::Hangup { call_id } => Some(*call_id),
PendingPjsuaOp::ConnectFaxPort { call_id, .. } => Some(*call_id),
};
if let Some(cid) = call_id {
if !is_call_valid(cid) {
tracing::warn!("Skipping stale op for dead call {}: {:?}", cid, op);
// For ConnectFaxPort, signal failure so the caller doesn't hang
if let PendingPjsuaOp::ConnectFaxPort { done_tx, .. } = op {
let _ = done_tx.send(false);
}
continue;
if let Some(cid) = call_id
&& !is_call_valid(cid)
{
tracing::warn!("Skipping stale op for dead call {}: {:?}", cid, op);
// For ConnectFaxPort, signal failure so the caller doesn't hang
if let PendingPjsuaOp::ConnectFaxPort { done_tx, .. } = op {
let _ = done_tx.send(false);
}
continue;
}
count += 1;
match op {
@ -559,7 +559,7 @@ unsafe fn process_audio_frame(
timestamp: pj_timestamp { u64_: *timestamp },
bit_info: 0,
};
pjmedia_port_get_frame(master_port, &mut clock_frame);
unsafe { pjmedia_port_get_frame(master_port, &mut clock_frame) };
// Now drain the SIP->Discord buffers that were filled by channel_port_put_frame callbacks
// during the conference tick above.
@ -718,11 +718,11 @@ pub fn check_rtp_inactivity() {
}
}
if let Some(sender_lock) = TIMEOUT_EVENT_TX.get() {
if let Some(ref tx) = *sender_lock.lock() {
for (call_id, rx_count) in timed_out_calls {
let _ = tx.send(super::SipEvent::CallTimeout { call_id, rx_count });
}
if let Some(sender_lock) = TIMEOUT_EVENT_TX.get()
&& let Some(ref tx) = *sender_lock.lock()
{
for (call_id, rx_count) in timed_out_calls {
let _ = tx.send(super::SipEvent::CallTimeout { call_id, rx_count });
}
}
}

2198
sipcord-bridge/src/transport/sip/callbacks.rs
View file

File diff suppressed because it is too large Load diff

154
sipcord-bridge/src/transport/sip/channel_audio.rs
View file

@ -84,10 +84,12 @@ pub unsafe extern "C" fn channel_port_get_frame(
return -1; // PJ_EINVAL
}
let channel_id = Snowflake::new((*this_port).port_data.ldata as u64);
let channel_id = unsafe { Snowflake::new((*this_port).port_data.ldata as u64) };
if *channel_id == 0 {
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_NONE;
(*frame).size = 0;
unsafe {
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_NONE;
(*frame).size = 0;
}
return pj_constants__PJ_SUCCESS as pj_status_t;
}
@ -169,10 +171,10 @@ pub unsafe extern "C" fn channel_port_get_frame(
}
if samples_len > 0 {
let samples = std::slice::from_raw_parts(samples_ptr, samples_len);
super::ffi::frame_utils::fill_audio_frame(frame, samples);
let samples = unsafe { std::slice::from_raw_parts(samples_ptr, samples_len) };
unsafe { super::ffi::frame_utils::fill_audio_frame(frame, samples) };
} else {
super::ffi::frame_utils::fill_silence_frame(frame);
unsafe { super::ffi::frame_utils::fill_silence_frame(frame) };
}
pj_constants__PJ_SUCCESS as pj_status_t
@ -186,51 +188,51 @@ fn get_samples_from_buffer(channel_id: Snowflake, buf: &mut [i16; SAMPLES_PER_FR
static DRAIN_COUNT: AtomicU64 = AtomicU64::new(0);
static UNDERRUN_COUNT: AtomicU64 = AtomicU64::new(0);
if let Some(consumer_entry) = get_discord_to_sip_consumers().get(&channel_id) {
if let Some(mut consumer) = consumer_entry.try_lock() {
let available = consumer.slots();
if available >= SAMPLES_PER_FRAME {
let count = DRAIN_COUNT.fetch_add(1, Ordering::Relaxed);
if count.is_multiple_of(250) {
tracing::debug!(
"Discord->SIP drain: channel={}, available={}, draining {}",
channel_id,
available,
SAMPLES_PER_FRAME
);
}
if let Ok(chunk) = consumer.read_chunk(SAMPLES_PER_FRAME) {
let (first, second) = chunk.as_slices();
buf[..first.len()].copy_from_slice(first);
if !second.is_empty() {
buf[first.len()..first.len() + second.len()].copy_from_slice(second);
}
chunk.commit_all();
}
return SAMPLES_PER_FRAME;
} else if available > 0 {
// Partial buffer - drain what we have, zero-fill the rest
let underruns = UNDERRUN_COUNT.fetch_add(1, Ordering::Relaxed) + 1;
if underruns <= 10 || underruns.is_multiple_of(100) {
tracing::warn!(
"BUFFER UNDERRUN (Discord->SIP): channel={}, only {} available (need {}), total: {}",
channel_id,
available,
SAMPLES_PER_FRAME,
underruns
);
}
buf[available..].fill(0);
if let Ok(chunk) = consumer.read_chunk(available) {
let (first, second) = chunk.as_slices();
buf[..first.len()].copy_from_slice(first);
if !second.is_empty() {
buf[first.len()..first.len() + second.len()].copy_from_slice(second);
}
chunk.commit_all();
}
return available;
if let Some(consumer_entry) = get_discord_to_sip_consumers().get(&channel_id)
&& let Some(mut consumer) = consumer_entry.try_lock()
{
let available = consumer.slots();
if available >= SAMPLES_PER_FRAME {
let count = DRAIN_COUNT.fetch_add(1, Ordering::Relaxed);
if count.is_multiple_of(250) {
tracing::debug!(
"Discord->SIP drain: channel={}, available={}, draining {}",
channel_id,
available,
SAMPLES_PER_FRAME
);
}
if let Ok(chunk) = consumer.read_chunk(SAMPLES_PER_FRAME) {
let (first, second) = chunk.as_slices();
buf[..first.len()].copy_from_slice(first);
if !second.is_empty() {
buf[first.len()..first.len() + second.len()].copy_from_slice(second);
}
chunk.commit_all();
}
return SAMPLES_PER_FRAME;
} else if available > 0 {
// Partial buffer - drain what we have, zero-fill the rest
let underruns = UNDERRUN_COUNT.fetch_add(1, Ordering::Relaxed) + 1;
if underruns <= 10 || underruns.is_multiple_of(100) {
tracing::warn!(
"BUFFER UNDERRUN (Discord->SIP): channel={}, only {} available (need {}), total: {}",
channel_id,
available,
SAMPLES_PER_FRAME,
underruns
);
}
buf[available..].fill(0);
if let Ok(chunk) = consumer.read_chunk(available) {
let (first, second) = chunk.as_slices();
buf[..first.len()].copy_from_slice(first);
if !second.is_empty() {
buf[first.len()..first.len() + second.len()].copy_from_slice(second);
}
chunk.commit_all();
}
return available;
}
}
@ -254,11 +256,13 @@ pub unsafe extern "C" fn channel_port_put_frame(
}
// Only process audio frames with data
if (*frame).type_ != pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO || (*frame).size == 0 {
if unsafe {
(*frame).type_ != pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO || (*frame).size == 0
} {
return pj_constants__PJ_SUCCESS as pj_status_t;
}
let channel_id = Snowflake::new((*this_port).port_data.ldata as u64);
let channel_id = unsafe { Snowflake::new((*this_port).port_data.ldata as u64) };
if *channel_id == 0 {
return pj_constants__PJ_SUCCESS as pj_status_t;
}
@ -270,16 +274,18 @@ pub unsafe extern "C" fn channel_port_put_frame(
call_count,
this_port,
channel_id,
(*frame).size
unsafe { (*frame).size }
);
} else if call_count == 10 {
tracing::trace!("channel_port_put_frame: suppressing further per-call logs");
}
// View frame buffer as i16 slice (zero-copy)
let num_samples = (*frame).size / 2;
let frame_buf = (*frame).buf as *const i16;
let samples = std::slice::from_raw_parts(frame_buf, num_samples);
let samples = unsafe {
let num_samples = (*frame).size / 2;
let frame_buf = (*frame).buf as *const i16;
std::slice::from_raw_parts(frame_buf, num_samples)
};
// Store in the SIP->Discord buffer for this channel
let buffers = CHANNEL_AUDIO_IN.get_or_init(DashMap::new);
@ -324,7 +330,7 @@ pub unsafe extern "C" fn channel_port_put_frame(
/// Custom on_destroy callback for channel buffer ports
pub unsafe extern "C" fn channel_port_on_destroy(this_port: *mut pjmedia_port) -> pj_status_t {
if !this_port.is_null() {
// Remove from reverse mapping
// Remove from reverse mapping (no unsafe ops needed here, just pointer-to-usize cast)
let port_key = this_port as usize;
if let Some(mapping) = PORT_TO_CHANNEL.get() {
mapping.lock().remove(&port_key);
@ -348,10 +354,12 @@ unsafe fn connect_call_to_channel(
) {
// Connect this call to other calls in the same channel
for &(other_call_id, other_conf_port) in other_calls {
let status1 =
pjmedia_conf_connect_port(conf, *conf_port as u32, *other_conf_port as u32, 0);
let status2 =
pjmedia_conf_connect_port(conf, *other_conf_port as u32, *conf_port as u32, 0);
let (status1, status2) = unsafe {
(
pjmedia_conf_connect_port(conf, *conf_port as u32, *other_conf_port as u32, 0),
pjmedia_conf_connect_port(conf, *other_conf_port as u32, *conf_port as u32, 0),
)
};
if status1 == pj_constants__PJ_SUCCESS as i32 && status2 == pj_constants__PJ_SUCCESS as i32
{
@ -377,10 +385,14 @@ unsafe fn connect_call_to_channel(
// Connect call to channel's conference port bidirectionally
if let Some(channel_slot) = get_or_create_channel_port(channel_id) {
// Channel port -> call (Discord audio reaches this call)
let status1 = pjmedia_conf_connect_port(conf, *channel_slot as u32, *conf_port as u32, 0);
// Call -> channel port (SIP audio goes to channel for Discord)
let status2 = pjmedia_conf_connect_port(conf, *conf_port as u32, *channel_slot as u32, 0);
let (status1, status2) = unsafe {
(
// Channel port -> call (Discord audio reaches this call)
pjmedia_conf_connect_port(conf, *channel_slot as u32, *conf_port as u32, 0),
// Call -> channel port (SIP audio goes to channel for Discord)
pjmedia_conf_connect_port(conf, *conf_port as u32, *channel_slot as u32, 0),
)
};
if status1 != pj_constants__PJ_SUCCESS as i32 {
tracing::warn!(
@ -429,8 +441,10 @@ unsafe fn disconnect_call_from_channel(
// Disconnect from other calls in the channel (both directions)
for &other_call_id in remaining_calls {
if let Some(other_conf_port) = conf_ports.get(&other_call_id).map(|r| *r) {
pjmedia_conf_disconnect_port(conf, *conf_port as u32, *other_conf_port as u32);
pjmedia_conf_disconnect_port(conf, *other_conf_port as u32, *conf_port as u32);
unsafe {
pjmedia_conf_disconnect_port(conf, *conf_port as u32, *other_conf_port as u32);
pjmedia_conf_disconnect_port(conf, *other_conf_port as u32, *conf_port as u32);
}
tracing::debug!(
"Disconnected call {} from call {} in channel {}",
call_id,
@ -442,8 +456,10 @@ unsafe fn disconnect_call_from_channel(
// Disconnect from channel port (both directions)
if let Some(channel_slot) = get_channel_slot(channel_id) {
pjmedia_conf_disconnect_port(conf, *channel_slot as u32, *conf_port as u32);
pjmedia_conf_disconnect_port(conf, *conf_port as u32, *channel_slot as u32);
unsafe {
pjmedia_conf_disconnect_port(conf, *channel_slot as u32, *conf_port as u32);
pjmedia_conf_disconnect_port(conf, *conf_port as u32, *channel_slot as u32);
}
tracing::debug!(
"Disconnected channel {} slot {} <-> call {} (port {}) bidirectionally",
channel_id,

14
sipcord-bridge/src/transport/sip/ffi/direct_player.rs
View file

@ -11,6 +11,10 @@ use std::collections::HashMap;
/// Custom get_frame callback for direct player ports
/// Returns samples from the player's buffer, advancing position each call
///
/// # Safety
/// Called by the pjmedia conference bridge. `this_port` and `frame` must be
/// valid, non-null pointers to pjmedia structures owned by pjsua.
pub unsafe extern "C" fn direct_player_get_frame(
this_port: *mut pjmedia_port,
frame: *mut pjmedia_frame,
@ -45,13 +49,13 @@ pub unsafe extern "C" fn direct_player_get_frame(
if let Some((buffer, pos)) = state.get_mut(&port_key) {
if *pos < buffer.len() {
let end = (*pos + SAMPLES_PER_FRAME).min(buffer.len());
super::frame_utils::fill_audio_frame(frame, &buffer[*pos..end]);
unsafe { super::frame_utils::fill_audio_frame(frame, &buffer[*pos..end]) };
*pos = end;
} else {
super::frame_utils::fill_silence_frame(frame); // Playback complete
unsafe { super::frame_utils::fill_silence_frame(frame) }; // Playback complete
}
} else {
super::frame_utils::fill_silence_frame(frame);
unsafe { super::frame_utils::fill_silence_frame(frame) };
}
}
@ -59,6 +63,10 @@ pub unsafe extern "C" fn direct_player_get_frame(
}
/// Custom on_destroy callback for direct player ports
///
/// # Safety
/// Called by pjmedia when the port is being destroyed. `this_port` must be
/// a valid pointer to a pjmedia_port that was previously created by this module.
pub unsafe extern "C" fn direct_player_on_destroy(this_port: *mut pjmedia_port) -> pj_status_t {
if !this_port.is_null() {
let port_key = this_port as usize;

81
sipcord-bridge/src/transport/sip/ffi/frame_utils.rs
View file

@ -24,7 +24,7 @@ pub unsafe fn get_conference_bridge() -> Option<*mut pjmedia_conf> {
if master_port.is_null() {
return None;
}
let conf = (*master_port).port_data.pdata as *mut pjmedia_conf;
let conf = unsafe { (*master_port).port_data.pdata as *mut pjmedia_conf };
if conf.is_null() {
return None;
}
@ -38,18 +38,20 @@ pub unsafe fn get_conference_bridge() -> Option<*mut pjmedia_conf> {
/// `frame` must be a valid, non-null pointer to a pjmedia_frame with a buffer
/// large enough for SAMPLES_PER_FRAME i16 samples.
pub unsafe fn fill_audio_frame(frame: *mut pjmedia_frame, samples: &[i16]) {
let frame_buf = (*frame).buf as *mut i16;
std::ptr::copy_nonoverlapping(samples.as_ptr(), frame_buf, samples.len());
// Pad with silence if we got fewer samples than a full frame
if samples.len() < SAMPLES_PER_FRAME {
std::ptr::write_bytes(
frame_buf.add(samples.len()),
0,
SAMPLES_PER_FRAME - samples.len(),
);
unsafe {
let frame_buf = (*frame).buf as *mut i16;
std::ptr::copy_nonoverlapping(samples.as_ptr(), frame_buf, samples.len());
// Pad with silence if we got fewer samples than a full frame
if samples.len() < SAMPLES_PER_FRAME {
std::ptr::write_bytes(
frame_buf.add(samples.len()),
0,
SAMPLES_PER_FRAME - samples.len(),
);
}
(*frame).size = (SAMPLES_PER_FRAME * 2) as pj_size_t;
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
}
(*frame).size = (SAMPLES_PER_FRAME * 2) as pj_size_t;
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
}
/// Fill a pjmedia_frame with silence.
@ -58,10 +60,12 @@ pub unsafe fn fill_audio_frame(frame: *mut pjmedia_frame, samples: &[i16]) {
/// `frame` must be a valid, non-null pointer to a pjmedia_frame with a buffer
/// large enough for SAMPLES_PER_FRAME i16 samples.
pub unsafe fn fill_silence_frame(frame: *mut pjmedia_frame) {
let frame_buf = (*frame).buf as *mut u8;
std::ptr::write_bytes(frame_buf, 0, SAMPLES_PER_FRAME * 2);
(*frame).size = (SAMPLES_PER_FRAME * 2) as pj_size_t;
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
unsafe {
let frame_buf = (*frame).buf as *mut u8;
std::ptr::write_bytes(frame_buf, 0, SAMPLES_PER_FRAME * 2);
(*frame).size = (SAMPLES_PER_FRAME * 2) as pj_size_t;
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
}
}
/// No-op put_frame callback for ports that only produce audio.
@ -119,14 +123,14 @@ pub unsafe fn create_and_connect_port(
) -> Result<ConfPortGuard> {
// Get or create the memory pool
let pool = pool.get_or_init(|| {
let p = pjsua_pool_create(pool_name.as_ptr() as *const _, 4096, 4096);
let p = unsafe { pjsua_pool_create(pool_name.as_ptr() as *const _, 4096, 4096) };
Mutex::new(SendablePool(p))
});
let pool_ptr = pool.lock().0;
// Allocate pjmedia_port structure
let port_size = std::mem::size_of::<pjmedia_port>();
let port = pj_pool_alloc(pool_ptr, port_size) as *mut pjmedia_port;
let port = unsafe { pj_pool_alloc(pool_ptr, port_size) as *mut pjmedia_port };
if port.is_null() {
anyhow::bail!(
"Failed to allocate {} port for call {}",
@ -134,7 +138,7 @@ pub unsafe fn create_and_connect_port(
call_id
);
}
std::ptr::write_bytes(port as *mut u8, 0, port_size);
unsafe { std::ptr::write_bytes(port as *mut u8, 0, port_size) };
// Create port name
let port_name = format!("{}{}", name_prefix, call_id);
@ -142,38 +146,41 @@ pub unsafe fn create_and_connect_port(
.map_err(|e| anyhow::anyhow!("Invalid port name: {}", e))?;
// Initialize port info
pjmedia_port_info_init(
&mut (*port).info,
&pj_str(port_name_cstr.as_ptr() as *mut _),
signature,
CONF_SAMPLE_RATE,
CONF_CHANNELS,
16,
SAMPLES_PER_FRAME as u32,
);
unsafe {
pjmedia_port_info_init(
&mut (*port).info,
&pj_str(port_name_cstr.as_ptr() as *mut _),
signature,
CONF_SAMPLE_RATE,
CONF_CHANNELS,
16,
SAMPLES_PER_FRAME as u32,
);
// Set callbacks
(*port).get_frame = Some(callbacks.get_frame);
(*port).put_frame = Some(callbacks.put_frame);
(*port).on_destroy = callbacks.on_destroy;
// Set callbacks
(*port).get_frame = Some(callbacks.get_frame);
(*port).put_frame = Some(callbacks.put_frame);
(*port).on_destroy = callbacks.on_destroy;
}
// Add to conference
let mut player_slot: i32 = 0;
let status = pjsua_conf_add_port(pool_ptr, port, &mut player_slot);
let status = unsafe { pjsua_conf_add_port(pool_ptr, port, &mut player_slot) };
if status != pj_constants__PJ_SUCCESS as i32 {
anyhow::bail!("Failed to add {} port to conf: {}", name_prefix, status);
}
// Connect player port to the target call's port
let conf = get_conference_bridge();
let conf = unsafe { get_conference_bridge() };
let Some(conf) = conf else {
pjsua_conf_remove_port(player_slot);
unsafe { pjsua_conf_remove_port(player_slot) };
anyhow::bail!("Failed to get conference bridge for {} port", name_prefix);
};
let status = pjmedia_conf_connect_port(conf, player_slot as u32, *call_conf_port as u32, 0);
let status =
unsafe { pjmedia_conf_connect_port(conf, player_slot as u32, *call_conf_port as u32, 0) };
if status != pj_constants__PJ_SUCCESS as i32 {
pjsua_conf_remove_port(player_slot);
unsafe { pjsua_conf_remove_port(player_slot) };
anyhow::bail!("Failed to connect {} port to call: {}", name_prefix, status);
}

10
sipcord-bridge/src/transport/sip/ffi/init.rs
View file

@ -161,7 +161,7 @@ unsafe extern "C" fn pjsip_log_callback(level: c_int, data: *const c_char, _len:
return;
}
let c_str = std::ffi::CStr::from_ptr(data);
let c_str = unsafe { std::ffi::CStr::from_ptr(data) };
let msg = c_str.to_string_lossy();
let msg = msg.trim_end();
@ -375,10 +375,10 @@ pub fn init_pjsua(config: &SipConfig, tls_config: Option<&TlsConfig>) -> Result<
tracing::info!("TCP transport created on port {}", config.port);
// Create TLS transport if configured (skip gracefully if certs missing)
if let Some(tls) = tls_config {
if !create_tls_transport(tls, &config.public_host)? {
tracing::warn!("TLS transport not created - running without TLS");
}
if let Some(tls) = tls_config
&& !create_tls_transport(tls, &config.public_host)?
{
tracing::warn!("TLS transport not created - running without TLS");
}
// Start pjsua

24
sipcord-bridge/src/transport/sip/ffi/looping_player.rs
View file

@ -38,6 +38,10 @@ pub struct LoopingPlayerState {
/// Custom get_frame callback for looping player ports
/// Returns samples from the player's buffer, looping back to start when reaching end
///
/// # Safety
/// Called by the pjmedia conference bridge. `this_port` and `frame` must be
/// valid, non-null pointers to pjmedia structures owned by pjsua.
pub unsafe extern "C" fn looping_player_get_frame(
this_port: *mut pjmedia_port,
frame: *mut pjmedia_frame,
@ -73,7 +77,9 @@ pub unsafe extern "C" fn looping_player_get_frame(
if player_data.is_active.load(Ordering::SeqCst) && !player_data.samples.is_empty() {
let pos = player_data.position;
let end = (pos + SAMPLES_PER_FRAME).min(player_data.samples.len());
super::frame_utils::fill_audio_frame(frame, &player_data.samples[pos..end]);
unsafe {
super::frame_utils::fill_audio_frame(frame, &player_data.samples[pos..end])
};
// Advance position, loop back if at end
player_data.position = if end >= player_data.samples.len() {
@ -82,10 +88,10 @@ pub unsafe extern "C" fn looping_player_get_frame(
end
};
} else {
super::frame_utils::fill_silence_frame(frame);
unsafe { super::frame_utils::fill_silence_frame(frame) };
}
} else {
super::frame_utils::fill_silence_frame(frame);
unsafe { super::frame_utils::fill_silence_frame(frame) };
}
}
@ -93,6 +99,10 @@ pub unsafe extern "C" fn looping_player_get_frame(
}
/// Custom on_destroy callback for looping player ports
///
/// # Safety
/// Called by pjmedia when the port is being destroyed. `this_port` must be
/// a valid pointer to a pjmedia_port that was previously created by this module.
pub unsafe extern "C" fn looping_player_on_destroy(this_port: *mut pjmedia_port) -> pj_status_t {
if !this_port.is_null() {
let port_key = this_port as usize;
@ -194,10 +204,10 @@ pub fn stop_loop(call_id: CallId) {
if let Some(state) = state {
// Mark as inactive (get_frame will return silence)
if let Some(data) = LOOPING_PLAYER_DATA.get() {
if let Some(player_data) = data.lock().get(&state.port_key) {
player_data.is_active.store(false, Ordering::SeqCst);
}
if let Some(data) = LOOPING_PLAYER_DATA.get()
&& let Some(player_data) = data.lock().get(&state.port_key)
{
player_data.is_active.store(false, Ordering::SeqCst);
}
// Remove from conference

4
sipcord-bridge/src/transport/sip/ffi/streaming_player.rs
View file

@ -118,9 +118,9 @@ pub unsafe extern "C" fn streaming_get_frame(
// Fill frame buffer
if !samples.is_empty() {
super::frame_utils::fill_audio_frame(frame, &samples);
unsafe { super::frame_utils::fill_audio_frame(frame, &samples) };
} else {
super::frame_utils::fill_silence_frame(frame);
unsafe { super::frame_utils::fill_silence_frame(frame) };
}
pj_constants__PJ_SUCCESS as pj_status_t

36
sipcord-bridge/src/transport/sip/ffi/test_tone.rs
View file

@ -76,7 +76,7 @@ pub unsafe extern "C" fn test_tone_get_frame(
}
if tone_state.finished {
super::frame_utils::fill_silence_frame(frame);
unsafe { super::frame_utils::fill_silence_frame(frame) };
} else {
// Copy from precomputed LUT with wraparound (two memcpy calls max)
let lut = tone_lut();
@ -85,27 +85,29 @@ pub unsafe extern "C" fn test_tone_get_frame(
tone_state.phase += SAMPLES_PER_FRAME as u64;
let first_chunk = (lut_len - phase).min(SAMPLES_PER_FRAME);
let frame_buf = (*frame).buf as *mut i16;
std::ptr::copy_nonoverlapping(
lut[phase..phase + first_chunk].as_ptr(),
frame_buf,
first_chunk,
);
if first_chunk < SAMPLES_PER_FRAME {
let remaining = SAMPLES_PER_FRAME - first_chunk;
unsafe {
let frame_buf = (*frame).buf as *mut i16;
std::ptr::copy_nonoverlapping(
lut.as_ptr(),
frame_buf.add(first_chunk),
remaining,
lut[phase..phase + first_chunk].as_ptr(),
frame_buf,
first_chunk,
);
}
(*frame).size = (SAMPLES_PER_FRAME * 2) as pj_size_t;
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
if first_chunk < SAMPLES_PER_FRAME {
let remaining = SAMPLES_PER_FRAME - first_chunk;
std::ptr::copy_nonoverlapping(
lut.as_ptr(),
frame_buf.add(first_chunk),
remaining,
);
}
(*frame).size = (SAMPLES_PER_FRAME * 2) as pj_size_t;
(*frame).type_ = pjmedia_frame_type_PJMEDIA_FRAME_TYPE_AUDIO;
}
}
} else {
super::frame_utils::fill_silence_frame(frame);
unsafe { super::frame_utils::fill_silence_frame(frame) };
}
}

2
sipcord-bridge/src/transport/sip/ffi/utils.rs
View file

@ -11,7 +11,7 @@ pub unsafe fn pj_str_to_string(s: &pj_str_t) -> String {
return String::new();
}
let slice = std::slice::from_raw_parts(s.ptr as *const u8, s.slen as usize);
let slice = unsafe { std::slice::from_raw_parts(s.ptr as *const u8, s.slen as usize) };
String::from_utf8_lossy(slice).to_string()
}

5
sipcord-bridge/src/transport/sip/mod.rs
View file

@ -539,10 +539,9 @@ fn make_outbound_call(sip_uri: &str, caller_display_name: Option<&str>) -> Resul
== ::pjsua::pj_constants__PJ_SUCCESS as i32
{
let ai = acc_info.assume_init();
let uri_str = std::ffi::CStr::from_ptr(ai.acc_uri.ptr)
std::ffi::CStr::from_ptr(ai.acc_uri.ptr)
.to_string_lossy()
.into_owned();
uri_str
.into_owned()
} else {
"sip:sipcord@localhost".to_string()
};

407
sipcord-bridge/src/transport/sip/nat.rs
View file

@ -79,12 +79,11 @@ fn sanitize_sdp_missing_rtpmap(sdp: &str) -> Option<String> {
let mut rtpmap_pts: std::collections::HashSet<u32> = std::collections::HashSet::new();
for line in section_lines {
// a=rtpmap:96 opus/48000/2
if let Some(rest) = line.strip_prefix("a=rtpmap:") {
if let Some(pt_str) = rest.split_whitespace().next() {
if let Ok(pt) = pt_str.parse::<u32>() {
rtpmap_pts.insert(pt);
}
}
if let Some(rest) = line.strip_prefix("a=rtpmap:")
&& let Some(pt_str) = rest.split_whitespace().next()
&& let Ok(pt) = pt_str.parse::<u32>()
{
rtpmap_pts.insert(pt);
}
}
@ -94,11 +93,12 @@ fn sanitize_sdp_missing_rtpmap(sdp: &str) -> Option<String> {
let mut stripped_pts: Vec<u32> = Vec::new();
for pt_str in payload_types {
if let Ok(pt) = pt_str.parse::<u32>() {
if pt >= 96 && !rtpmap_pts.contains(&pt) {
stripped_pts.push(pt);
continue;
}
if let Ok(pt) = pt_str.parse::<u32>()
&& pt >= 96
&& !rtpmap_pts.contains(&pt)
{
stripped_pts.push(pt);
continue;
}
kept_pts.push(pt_str);
}
@ -134,14 +134,12 @@ fn sanitize_sdp_missing_rtpmap(sdp: &str) -> Option<String> {
// Copy section attribute lines, stripping a=fmtp: for removed PTs
let stripped_set: std::collections::HashSet<u32> = stripped_pts.into_iter().collect();
for line in section_lines {
if let Some(rest) = line.strip_prefix("a=fmtp:") {
if let Some(pt_str) = rest.split_whitespace().next() {
if let Ok(pt) = pt_str.parse::<u32>() {
if stripped_set.contains(&pt) {
continue; // skip fmtp for stripped PT
}
}
}
if let Some(rest) = line.strip_prefix("a=fmtp:")
&& let Some(pt_str) = rest.split_whitespace().next()
&& let Ok(pt) = pt_str.parse::<u32>()
&& stripped_set.contains(&pt)
{
continue; // skip fmtp for stripped PT
}
result_lines.push(line.to_string());
}
@ -164,14 +162,14 @@ fn is_rfc1918(ip: Ipv4Addr) -> bool {
///
/// Returns `None` if transport info is invalid or the address is not IPv4.
unsafe fn extract_dst_ipv4(tdata: *const pjsip_tx_data) -> Option<Ipv4Addr> {
if (*tdata).tp_info.transport.is_null() || (*tdata).tp_info.dst_addr_len <= 0 {
if unsafe { (*tdata).tp_info.transport.is_null() || (*tdata).tp_info.dst_addr_len <= 0 } {
return None;
}
let dst_addr = &(*tdata).tp_info.dst_addr;
let dst_addr = unsafe { &(*tdata).tp_info.dst_addr };
// PJ_AF_INET is typically 2 (same as AF_INET on most systems)
if dst_addr.addr.sa_family == 2 {
let addr_in = &dst_addr.ipv4;
if unsafe { dst_addr.addr.sa_family } == 2 {
let addr_in = unsafe { &dst_addr.ipv4 };
let ip_bytes = addr_in.sin_addr.s_addr.to_ne_bytes();
Some(Ipv4Addr::new(
ip_bytes[0],
@ -195,32 +193,33 @@ unsafe fn rewrite_contact_host(
new_host: &str,
new_port: u16,
) -> bool {
let contact_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut())
as *mut pjsip_contact_hdr;
let contact_hdr =
unsafe { pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut()) }
as *mut pjsip_contact_hdr;
if contact_hdr.is_null() {
return false;
}
let uri = (*contact_hdr).uri;
let uri = unsafe { (*contact_hdr).uri };
if uri.is_null() {
return false;
}
// Unwrap via vtable to handle pjsip_name_addr wrapping
let uri_vptr = (*(uri as *const pjsip_uri)).vptr;
let uri_vptr = unsafe { (*(uri as *const pjsip_uri)).vptr };
if uri_vptr.is_null() {
return false;
}
let get_uri_fn = match (*uri_vptr).p_get_uri {
let get_uri_fn = match unsafe { (*uri_vptr).p_get_uri } {
Some(f) => f,
None => return false,
};
let sip_uri_raw = get_uri_fn(uri as *mut std::os::raw::c_void);
let sip_uri_raw = unsafe { get_uri_fn(uri as *mut std::os::raw::c_void) };
if sip_uri_raw.is_null() {
return false;
}
let sip_uri = sip_uri_raw as *mut pjsip_sip_uri;
if (*sip_uri).host.ptr.is_null() || (*sip_uri).host.slen <= 0 {
if unsafe { (*sip_uri).host.ptr.is_null() || (*sip_uri).host.slen <= 0 } {
return false;
}
@ -228,15 +227,17 @@ unsafe fn rewrite_contact_host(
return false;
};
let host_len = new_host.len();
let pool_str = pj_pool_alloc(pool, host_len + 1) as *mut c_char;
let pool_str = unsafe { pj_pool_alloc(pool, host_len + 1) } as *mut c_char;
if pool_str.is_null() {
return false;
}
ptr::copy_nonoverlapping(host_cstr.as_ptr(), pool_str, host_len + 1);
(*sip_uri).host.ptr = pool_str;
(*sip_uri).host.slen = host_len as i64;
(*sip_uri).port = new_port as i32;
unsafe {
ptr::copy_nonoverlapping(host_cstr.as_ptr(), pool_str, host_len + 1);
(*sip_uri).host.ptr = pool_str;
(*sip_uri).host.slen = host_len as i64;
(*sip_uri).port = new_port as i32;
}
true
}
@ -251,12 +252,13 @@ unsafe fn rewrite_sdp_body(
old_ip: &str,
new_ip: &str,
) -> bool {
let body = (*msg).body;
if body.is_null() || (*body).len == 0 || (*body).data.is_null() {
let body = unsafe { (*msg).body };
if body.is_null() || unsafe { (*body).len == 0 || (*body).data.is_null() } {
return false;
}
let body_slice = std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize);
let body_slice =
unsafe { std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize) };
let Ok(sdp_str) = std::str::from_utf8(body_slice) else {
return false;
};
@ -284,14 +286,16 @@ unsafe fn rewrite_sdp_body(
}
let new_len = new_sdp.len();
let new_body_ptr = pj_pool_alloc(pool, new_len) as *mut u8;
let new_body_ptr = unsafe { pj_pool_alloc(pool, new_len) } as *mut u8;
if new_body_ptr.is_null() {
return false;
}
ptr::copy_nonoverlapping(new_sdp.as_ptr(), new_body_ptr, new_len);
(*body).data = new_body_ptr as *mut _;
(*body).len = new_len as u32;
unsafe {
ptr::copy_nonoverlapping(new_sdp.as_ptr(), new_body_ptr, new_len);
(*body).data = new_body_ptr as *mut _;
(*body).len = new_len as u32;
}
true
}
@ -310,7 +314,7 @@ unsafe fn rewrite_local_network_tdata(tdata: *mut pjsip_tx_data, direction: &str
return false;
}
let Some(dst_ip) = extract_dst_ipv4(tdata) else {
let Some(dst_ip) = (unsafe { extract_dst_ipv4(tdata) }) else {
return false;
};
@ -318,7 +322,7 @@ unsafe fn rewrite_local_network_tdata(tdata: *mut pjsip_tx_data, direction: &str
return false;
}
let msg = (*tdata).msg;
let msg = unsafe { (*tdata).msg };
if msg.is_null() {
return false;
}
@ -326,7 +330,7 @@ unsafe fn rewrite_local_network_tdata(tdata: *mut pjsip_tx_data, direction: &str
let mut changed = false;
// Rewrite Contact header
if rewrite_contact_host((*tdata).pool, msg, local_host, *port) {
if unsafe { rewrite_contact_host((*tdata).pool, msg, local_host, *port) } {
tracing::debug!(
"Rewrote {} Contact header for local client {}: host -> {}:{}",
direction,
@ -338,17 +342,17 @@ unsafe fn rewrite_local_network_tdata(tdata: *mut pjsip_tx_data, direction: &str
}
// Rewrite SDP body if we have an RTP public IP to replace
if let Some(public_ip) = rtp_public_ip {
if rewrite_sdp_body((*tdata).pool, msg, public_ip, local_host) {
tracing::debug!(
"Rewrote {} SDP for local client {}: {} -> {}",
direction,
dst_ip,
public_ip,
local_host
);
changed = true;
}
if let Some(public_ip) = rtp_public_ip
&& unsafe { rewrite_sdp_body((*tdata).pool, msg, public_ip, local_host) }
{
tracing::debug!(
"Rewrote {} SDP for local client {}: {} -> {}",
direction,
dst_ip,
public_ip,
local_host
);
changed = true;
}
changed
@ -370,42 +374,43 @@ unsafe fn rewrite_private_contact_for_external(tdata: *mut pjsip_tx_data, direct
return false;
}
let msg = (*tdata).msg;
let msg = unsafe { (*tdata).msg };
if msg.is_null() {
return false;
}
// Find Contact header
let contact_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut())
as *mut pjsip_contact_hdr;
let contact_hdr =
unsafe { pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut()) }
as *mut pjsip_contact_hdr;
if contact_hdr.is_null() {
return false;
}
let uri = (*contact_hdr).uri;
let uri = unsafe { (*contact_hdr).uri };
if uri.is_null() {
return false;
}
// Unwrap via vtable to handle pjsip_name_addr wrapping
let uri_vptr = (*(uri as *const pjsip_uri)).vptr;
let uri_vptr = unsafe { (*(uri as *const pjsip_uri)).vptr };
if uri_vptr.is_null() {
return false;
}
let get_uri_fn = match (*uri_vptr).p_get_uri {
let get_uri_fn = match unsafe { (*uri_vptr).p_get_uri } {
Some(f) => f,
None => return false,
};
let sip_uri_raw = get_uri_fn(uri as *mut std::os::raw::c_void);
let sip_uri_raw = unsafe { get_uri_fn(uri as *mut std::os::raw::c_void) };
if sip_uri_raw.is_null() {
return false;
}
let sip_uri = sip_uri_raw as *mut pjsip_sip_uri;
if (*sip_uri).host.ptr.is_null() || (*sip_uri).host.slen <= 0 {
if unsafe { (*sip_uri).host.ptr.is_null() || (*sip_uri).host.slen <= 0 } {
return false;
}
let host = pj_str_to_string(&(*sip_uri).host);
let host = unsafe { pj_str_to_string(&(*sip_uri).host) };
// Only rewrite if Contact host is a private (RFC 1918) IP
let contact_ip: Ipv4Addr = match host.parse() {
@ -418,14 +423,14 @@ unsafe fn rewrite_private_contact_for_external(tdata: *mut pjsip_tx_data, direct
}
// Skip if destination is also private (local-network rewrite handles that)
if let Some(dst_ip) = extract_dst_ipv4(tdata) {
if is_rfc1918(dst_ip) {
return false;
}
if let Some(dst_ip) = unsafe { extract_dst_ipv4(tdata) }
&& is_rfc1918(dst_ip)
{
return false;
}
// Rewrite Contact to public host
if rewrite_contact_host((*tdata).pool, msg, public_host, *port) {
if unsafe { rewrite_contact_host((*tdata).pool, msg, public_host, *port) } {
tracing::debug!(
"Rewrote {} Contact for external client: {} -> {}:{}",
direction,
@ -448,15 +453,17 @@ unsafe fn rewrite_private_contact_for_external(tdata: *mut pjsip_tx_data, direct
/// 2. Public-host rewrite: for external clients, replace private Contact IPs
/// with the public hostname so they can route BYE back to us
pub unsafe extern "C" fn on_tx_response_cb(tdata: *mut pjsip_tx_data) -> pj_status_t {
let local_rewrite = rewrite_local_network_tdata(tdata, "response");
let public_rewrite = rewrite_private_contact_for_external(tdata, "response");
let local_rewrite = unsafe { rewrite_local_network_tdata(tdata, "response") };
let public_rewrite = unsafe { rewrite_private_contact_for_external(tdata, "response") };
// If we modified headers, the buffer was already serialized by mod-msg-print
// (priority 8, before our module at priority 32). Invalidate and re-encode
// so the changes actually reach the wire.
if local_rewrite || public_rewrite {
pjsip_tx_data_invalidate_msg(tdata);
pjsip_tx_data_encode(tdata);
unsafe {
pjsip_tx_data_invalidate_msg(tdata);
pjsip_tx_data_encode(tdata);
}
}
pj_constants__PJ_SUCCESS as pj_status_t
@ -465,15 +472,17 @@ pub unsafe extern "C" fn on_tx_response_cb(tdata: *mut pjsip_tx_data) -> pj_stat
/// Callback to rewrite Contact header and SDP body in outgoing requests.
/// Same dual-rewrite logic as the response path.
pub unsafe extern "C" fn on_tx_request_cb(tdata: *mut pjsip_tx_data) -> pj_status_t {
let local_rewrite = rewrite_local_network_tdata(tdata, "request");
let public_rewrite = rewrite_private_contact_for_external(tdata, "request");
let local_rewrite = unsafe { rewrite_local_network_tdata(tdata, "request") };
let public_rewrite = unsafe { rewrite_private_contact_for_external(tdata, "request") };
// If we modified headers, the buffer was already serialized by mod-msg-print
// (priority 8, before our module at priority 32). Invalidate and re-encode
// so the changes actually reach the wire.
if local_rewrite || public_rewrite {
pjsip_tx_data_invalidate_msg(tdata);
pjsip_tx_data_encode(tdata);
unsafe {
pjsip_tx_data_invalidate_msg(tdata);
pjsip_tx_data_encode(tdata);
}
}
pj_constants__PJ_SUCCESS as pj_status_t
@ -492,38 +501,37 @@ pub unsafe extern "C" fn on_rx_request_nat_fixup_cb(rdata: *mut pjsip_rx_data) -
return pj_constants__PJ_FALSE as pj_bool_t;
}
let msg = (*rdata).msg_info.msg;
let msg = unsafe { (*rdata).msg_info.msg };
if msg.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Only process requests (safety check)
if (*msg).type_ != pjsip_msg_type_e_PJSIP_REQUEST_MSG {
if unsafe { (*msg).type_ } != pjsip_msg_type_e_PJSIP_REQUEST_MSG {
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Only process INVITE and re-INVITE (they carry SDP with media addresses)
let method_id = (*msg).line.req.method.id;
let method_id = unsafe { (*msg).line.req.method.id };
if method_id != pjsip_method_e_PJSIP_INVITE_METHOD {
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Check if there's a body (SDP)
let body = (*msg).body;
if body.is_null() || (*body).len == 0 || (*body).data.is_null() {
let body = unsafe { (*msg).body };
if body.is_null() || unsafe { (*body).len == 0 || (*body).data.is_null() } {
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Extract source IP from packet info
let src_name = &(*rdata).pkt_info.src_name;
let src_name = unsafe { &(*rdata).pkt_info.src_name };
let name_len = src_name
.iter()
.position(|&c| c == 0)
.unwrap_or(src_name.len());
let src_ip_str = match std::str::from_utf8(std::slice::from_raw_parts(
src_name.as_ptr() as *const u8,
name_len,
)) {
let src_ip_str = match std::str::from_utf8(unsafe {
std::slice::from_raw_parts(src_name.as_ptr() as *const u8, name_len)
}) {
Ok(s) => s,
Err(_) => return pj_constants__PJ_FALSE as pj_bool_t,
};
@ -533,87 +541,89 @@ pub unsafe extern "C" fn on_rx_request_nat_fixup_cb(rdata: *mut pjsip_rx_data) -
};
// Parse SDP to find c= line IP and check if it's a private address
let body_slice = std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize);
let body_slice =
unsafe { std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize) };
let sdp_str = match std::str::from_utf8(body_slice) {
Ok(s) => s,
Err(_) => return pj_constants__PJ_FALSE as pj_bool_t,
};
// Find any connection address in the SDP that needs NAT fixup.
// Check ALL c= lines (session-level and per-media) — if any contain a
// Check ALL c= lines (session-level and per-media) -- if any contain a
// private IP different from the packet source, rewrite the SDP.
let mut needs_rewrite = false;
let mut private_ip_str: Option<&str> = None;
for line in sdp_str.lines() {
if let Some(addr_str) = line.strip_prefix("c=IN IP4 ") {
let addr_str = addr_str.trim();
if let Ok(sdp_ip) = addr_str.parse::<Ipv4Addr>() {
if is_rfc1918(sdp_ip) && sdp_ip != src_ip {
needs_rewrite = true;
private_ip_str = Some(addr_str);
break;
}
if let Ok(sdp_ip) = addr_str.parse::<Ipv4Addr>()
&& is_rfc1918(sdp_ip)
&& sdp_ip != src_ip
{
needs_rewrite = true;
private_ip_str = Some(addr_str);
break;
}
}
}
if needs_rewrite {
if let Some(private_ip) = private_ip_str {
let pool = (*rdata).tp_info.pool;
if !pool.is_null() {
if rewrite_sdp_body(pool, msg, private_ip, src_ip_str) {
tracing::debug!(
"NAT fixup (INVITE): SDP rewritten {} -> {} (from {}:{})",
private_ip,
src_ip_str,
src_ip_str,
(*rdata).pkt_info.src_port
);
}
}
if needs_rewrite && let Some(private_ip) = private_ip_str {
let pool = unsafe { (*rdata).tp_info.pool };
if !pool.is_null() && unsafe { rewrite_sdp_body(pool, msg, private_ip, src_ip_str) } {
tracing::debug!(
"NAT fixup (INVITE): SDP rewritten {} -> {} (from {}:{})",
private_ip,
src_ip_str,
src_ip_str,
unsafe { (*rdata).pkt_info.src_port }
);
}
}
// Also rewrite Contact header if present and has private IP
let contact_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut())
as *mut pjsip_contact_hdr;
let contact_hdr =
unsafe { pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut()) }
as *mut pjsip_contact_hdr;
if !contact_hdr.is_null() {
let uri = (*contact_hdr).uri;
let uri = unsafe { (*contact_hdr).uri };
if !uri.is_null() {
let uri_vptr = (*(uri as *const pjsip_uri)).vptr;
if !uri_vptr.is_null() {
if let Some(get_uri_fn) = (*uri_vptr).p_get_uri {
let sip_uri_raw = get_uri_fn(uri as *mut std::os::raw::c_void);
if !sip_uri_raw.is_null() {
let sip_uri = sip_uri_raw as *mut pjsip_sip_uri;
let contact_host = pj_str_to_string(&(*sip_uri).host);
if let Ok(contact_ip) = contact_host.parse::<Ipv4Addr>() {
if is_rfc1918(contact_ip) && contact_ip != src_ip {
let src_port = (*rdata).pkt_info.src_port as u16;
let pool = (*rdata).tp_info.pool;
if !pool.is_null() {
if let Ok(new_host_cstr) = CString::new(src_ip_str) {
let host_len = src_ip_str.len();
let pool_str =
pj_pool_alloc(pool, host_len + 1) as *mut c_char;
if !pool_str.is_null() {
ptr::copy_nonoverlapping(
new_host_cstr.as_ptr(),
pool_str,
host_len + 1,
);
(*sip_uri).host.ptr = pool_str;
(*sip_uri).host.slen = host_len as i64;
(*sip_uri).port = src_port as i32;
tracing::debug!(
"NAT fixup (INVITE): Contact rewritten {} -> {}:{}",
contact_host,
src_ip_str,
src_port
);
}
}
let uri_vptr = unsafe { (*(uri as *const pjsip_uri)).vptr };
if !uri_vptr.is_null()
&& let Some(get_uri_fn) = unsafe { (*uri_vptr).p_get_uri }
{
let sip_uri_raw = unsafe { get_uri_fn(uri as *mut std::os::raw::c_void) };
if !sip_uri_raw.is_null() {
let sip_uri = sip_uri_raw as *mut pjsip_sip_uri;
let contact_host = unsafe { pj_str_to_string(&(*sip_uri).host) };
if let Ok(contact_ip) = contact_host.parse::<Ipv4Addr>()
&& is_rfc1918(contact_ip)
&& contact_ip != src_ip
{
let src_port = unsafe { (*rdata).pkt_info.src_port } as u16;
let pool = unsafe { (*rdata).tp_info.pool };
if !pool.is_null()
&& let Ok(new_host_cstr) = CString::new(src_ip_str)
{
let host_len = src_ip_str.len();
let pool_str =
unsafe { pj_pool_alloc(pool, host_len + 1) } as *mut c_char;
if !pool_str.is_null() {
unsafe {
ptr::copy_nonoverlapping(
new_host_cstr.as_ptr(),
pool_str,
host_len + 1,
);
(*sip_uri).host.ptr = pool_str;
(*sip_uri).host.slen = host_len as i64;
(*sip_uri).port = src_port as i32;
}
tracing::debug!(
"NAT fixup (INVITE): Contact rewritten {} -> {}:{}",
contact_host,
src_ip_str,
src_port
);
}
}
}
@ -624,26 +634,28 @@ pub unsafe extern "C" fn on_rx_request_nat_fixup_cb(rdata: *mut pjsip_rx_data) -
// Sanitize SDP: strip dynamic payload types (96+) that lack a=rtpmap attributes.
// Without this, PJSIP's SDP validator rejects these INVITEs with EMISSINGRTPMAP.
let body = (*msg).body;
if !body.is_null() && (*body).len > 0 && !(*body).data.is_null() {
let body = unsafe { (*msg).body };
if !body.is_null() && unsafe { (*body).len > 0 && !(*body).data.is_null() } {
let body_slice =
std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize);
if let Ok(sdp_str) = std::str::from_utf8(body_slice) {
if let Some(sanitized) = sanitize_sdp_missing_rtpmap(sdp_str) {
let pool = (*rdata).tp_info.pool;
if !pool.is_null() {
let new_len = sanitized.len();
let new_body_ptr = pj_pool_alloc(pool, new_len) as *mut u8;
if !new_body_ptr.is_null() {
unsafe { std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize) };
if let Ok(sdp_str) = std::str::from_utf8(body_slice)
&& let Some(sanitized) = sanitize_sdp_missing_rtpmap(sdp_str)
{
let pool = unsafe { (*rdata).tp_info.pool };
if !pool.is_null() {
let new_len = sanitized.len();
let new_body_ptr = unsafe { pj_pool_alloc(pool, new_len) } as *mut u8;
if !new_body_ptr.is_null() {
unsafe {
ptr::copy_nonoverlapping(sanitized.as_ptr(), new_body_ptr, new_len);
(*body).data = new_body_ptr as *mut _;
(*body).len = new_len as u32;
tracing::debug!(
"SDP sanitized: stripped orphan dynamic payload types (from {}:{})",
src_ip_str,
(*rdata).pkt_info.src_port
);
}
tracing::debug!(
"SDP sanitized: stripped orphan dynamic payload types (from {}:{})",
src_ip_str,
unsafe { (*rdata).pkt_info.src_port }
);
}
}
}
@ -670,27 +682,26 @@ pub unsafe extern "C" fn on_rx_response_nat_fixup_cb(rdata: *mut pjsip_rx_data)
return pj_constants__PJ_FALSE as pj_bool_t;
}
let msg = (*rdata).msg_info.msg;
let msg = unsafe { (*rdata).msg_info.msg };
if msg.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Only process 1xx/2xx responses (provisional and success)
let status_code = (*msg).line.status.code;
let status_code = unsafe { (*msg).line.status.code };
if !(100..300).contains(&status_code) {
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Extract source IP from pkt_info.src_name (null-terminated char array)
let src_name = &(*rdata).pkt_info.src_name;
let src_name = unsafe { &(*rdata).pkt_info.src_name };
let name_len = src_name
.iter()
.position(|&c| c == 0)
.unwrap_or(src_name.len());
let src_ip_str = match std::str::from_utf8(std::slice::from_raw_parts(
src_name.as_ptr() as *const u8,
name_len,
)) {
let src_ip_str = match std::str::from_utf8(unsafe {
std::slice::from_raw_parts(src_name.as_ptr() as *const u8, name_len)
}) {
Ok(s) => s,
Err(_) => return pj_constants__PJ_FALSE as pj_bool_t,
};
@ -698,11 +709,12 @@ pub unsafe extern "C" fn on_rx_response_nat_fixup_cb(rdata: *mut pjsip_rx_data)
Ok(ip) => ip,
Err(_) => return pj_constants__PJ_FALSE as pj_bool_t, // IPv6 or invalid
};
let src_port = (*rdata).pkt_info.src_port as u16;
let src_port = unsafe { (*rdata).pkt_info.src_port } as u16;
// Find Contact header in the response
let contact_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut())
as *mut pjsip_contact_hdr;
let contact_hdr =
unsafe { pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut()) }
as *mut pjsip_contact_hdr;
if contact_hdr.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
@ -711,26 +723,26 @@ pub unsafe extern "C" fn on_rx_response_nat_fixup_cb(rdata: *mut pjsip_rx_data)
// The rx path requires vtable-based URI unwrapping (p_get_uri) because
// the Contact URI may be wrapped in a pjsip_name_addr, unlike the tx
// path where we can cast directly.
let uri = (*contact_hdr).uri;
let uri = unsafe { (*contact_hdr).uri };
if uri.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
let uri_vptr = (*(uri as *const pjsip_uri)).vptr;
let uri_vptr = unsafe { (*(uri as *const pjsip_uri)).vptr };
if uri_vptr.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
let get_uri_fn = match (*uri_vptr).p_get_uri {
let get_uri_fn = match unsafe { (*uri_vptr).p_get_uri } {
Some(f) => f,
None => return pj_constants__PJ_FALSE as pj_bool_t,
};
let sip_uri_raw = get_uri_fn(uri as *mut std::os::raw::c_void);
let sip_uri_raw = unsafe { get_uri_fn(uri as *mut std::os::raw::c_void) };
if sip_uri_raw.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
let sip_uri = sip_uri_raw as *mut pjsip_sip_uri;
// Parse Contact host as IPv4
let contact_host = pj_str_to_string(&(*sip_uri).host);
let contact_host = unsafe { pj_str_to_string(&(*sip_uri).host) };
let contact_ip: Ipv4Addr = match contact_host.parse() {
Ok(ip) => ip,
Err(_) => return pj_constants__PJ_FALSE as pj_bool_t, // Hostname, skip
@ -753,12 +765,14 @@ pub unsafe extern "C" fn on_rx_response_nat_fixup_cb(rdata: *mut pjsip_rx_data)
);
// Rewrite Contact URI host to the public source IP
let pool = (*rdata).tp_info.pool;
if !pool.is_null() {
if let Ok(new_host_cstr) = CString::new(src_ip_str) {
let host_len = src_ip_str.len();
let pool_str = pj_pool_alloc(pool, host_len + 1) as *mut c_char;
if !pool_str.is_null() {
let pool = unsafe { (*rdata).tp_info.pool };
if !pool.is_null()
&& let Ok(new_host_cstr) = CString::new(src_ip_str)
{
let host_len = src_ip_str.len();
let pool_str = unsafe { pj_pool_alloc(pool, host_len + 1) } as *mut c_char;
if !pool_str.is_null() {
unsafe {
ptr::copy_nonoverlapping(new_host_cstr.as_ptr(), pool_str, host_len + 1);
(*sip_uri).host.ptr = pool_str;
(*sip_uri).host.slen = host_len as i64;
@ -768,27 +782,28 @@ pub unsafe extern "C" fn on_rx_response_nat_fixup_cb(rdata: *mut pjsip_rx_data)
}
// Rewrite SDP body: replace private IP with public source IP.
// Parse the SDP c= line directly to get the actual media IP — it may differ
// Parse the SDP c= line directly to get the actual media IP -- it may differ
// from the Contact header IP (e.g., dual-homed phone or double NAT).
let body = (*msg).body;
if !body.is_null() && (*body).len > 0 && !(*body).data.is_null() {
let body = unsafe { (*msg).body };
if !body.is_null() && unsafe { (*body).len > 0 && !(*body).data.is_null() } {
let body_slice =
std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize);
unsafe { std::slice::from_raw_parts((*body).data as *const u8, (*body).len as usize) };
if let Ok(sdp_str) = std::str::from_utf8(body_slice) {
for line in sdp_str.lines() {
if let Some(addr_str) = line.strip_prefix("c=IN IP4 ") {
let addr_str = addr_str.trim();
if let Ok(sdp_ip) = addr_str.parse::<Ipv4Addr>() {
if is_rfc1918(sdp_ip) && sdp_ip != src_ip {
if rewrite_sdp_body(pool, msg, addr_str, src_ip_str) {
tracing::debug!(
"NAT fixup: SDP rewritten {} -> {}",
addr_str,
src_ip_str
);
}
break;
if let Ok(sdp_ip) = addr_str.parse::<Ipv4Addr>()
&& is_rfc1918(sdp_ip)
&& sdp_ip != src_ip
{
if unsafe { rewrite_sdp_body(pool, msg, addr_str, src_ip_str) } {
tracing::debug!(
"NAT fixup: SDP rewritten {} -> {}",
addr_str,
src_ip_str
);
}
break;
}
}
}

769
sipcord-bridge/src/transport/sip/register_handler.rs
View file

@ -72,92 +72,101 @@ pub fn set_register_module_ptr(ptr: *mut pjsip_module) {
/// Initialize a pjsip_hdr as a list head (equivalent to pj_list_init C macro).
#[inline]
unsafe fn pj_list_init_hdr(hdr: *mut pjsip_hdr) {
(*hdr).next = hdr as *mut _;
(*hdr).prev = hdr as *mut _;
unsafe {
(*hdr).next = hdr as *mut _;
(*hdr).prev = hdr as *mut _;
}
}
/// Send a simple stateless SIP response (no custom headers).
unsafe fn send_simple_response(rdata: *mut pjsip_rx_data, status_code: u16, reason: &str) {
let endpt = pjsua_get_pjsip_endpt();
if !endpt.is_null() {
let reason_cstr = CString::new(reason).unwrap();
let reason_pj = pj_str(reason_cstr.as_ptr() as *mut c_char);
pjsip_endpt_respond_stateless(
endpt,
rdata,
status_code.into(),
&reason_pj,
ptr::null(),
ptr::null(),
);
unsafe {
let endpt = pjsua_get_pjsip_endpt();
if !endpt.is_null() {
let reason_cstr = CString::new(reason).unwrap();
let reason_pj = pj_str(reason_cstr.as_ptr() as *mut c_char);
pjsip_endpt_respond_stateless(
endpt,
rdata,
status_code.into(),
&reason_pj,
ptr::null(),
ptr::null(),
);
}
}
}
/// Send a stateless 200 OK with an Expires header.
unsafe fn send_register_ok(rdata: *mut pjsip_rx_data, expires: u32) {
let endpt = pjsua_get_pjsip_endpt();
if endpt.is_null() {
return;
}
let expires_str = format!("{}", expires);
let hdr_name = CString::new("Expires").unwrap();
let hdr_value = CString::new(expires_str).unwrap();
let pool = pjsua_pool_create(c"register_ok".as_ptr(), 512, 512);
if !pool.is_null() {
let name = pj_str(hdr_name.as_ptr() as *mut c_char);
let value = pj_str(hdr_value.as_ptr() as *mut c_char);
let hdr = pjsip_generic_string_hdr_create(pool, &name, &value);
if !hdr.is_null() {
let hdr_list = pj_pool_alloc(pool, std::mem::size_of::<pjsip_hdr>()) as *mut pjsip_hdr;
if !hdr_list.is_null() {
pj_list_init_hdr(hdr_list);
pj_list_insert_before(hdr_list as *mut pj_list_type, hdr as *mut pj_list_type);
let status = pjsip_endpt_respond_stateless(
endpt,
rdata,
200,
ptr::null(),
hdr_list,
ptr::null(),
);
if status != pj_constants__PJ_SUCCESS as i32 {
tracing::warn!("Failed to respond 200 OK to REGISTER: {}", status);
}
// Release pool — pjsip_endpt_respond_stateless clones what it
// needs into rdata's pool, so our header pool can be freed now.
pj_pool_release(pool);
return;
}
unsafe {
let endpt = pjsua_get_pjsip_endpt();
if endpt.is_null() {
return;
}
// Header creation failed — release the pool before falling through
pj_pool_release(pool);
}
// Fallback: respond without Expires header
let status =
pjsip_endpt_respond_stateless(endpt, rdata, 200, ptr::null(), ptr::null(), ptr::null());
if status != pj_constants__PJ_SUCCESS as i32 {
tracing::warn!("Failed to respond 200 OK to REGISTER: {}", status);
let expires_str = format!("{}", expires);
let hdr_name = CString::new("Expires").unwrap();
let hdr_value = CString::new(expires_str).unwrap();
let pool = pjsua_pool_create(c"register_ok".as_ptr(), 512, 512);
if !pool.is_null() {
let name = pj_str(hdr_name.as_ptr() as *mut c_char);
let value = pj_str(hdr_value.as_ptr() as *mut c_char);
let hdr = pjsip_generic_string_hdr_create(pool, &name, &value);
if !hdr.is_null() {
let hdr_list =
pj_pool_alloc(pool, std::mem::size_of::<pjsip_hdr>()) as *mut pjsip_hdr;
if !hdr_list.is_null() {
pj_list_init_hdr(hdr_list);
pj_list_insert_before(hdr_list as *mut pj_list_type, hdr as *mut pj_list_type);
let status = pjsip_endpt_respond_stateless(
endpt,
rdata,
200,
ptr::null(),
hdr_list,
ptr::null(),
);
if status != pj_constants__PJ_SUCCESS as i32 {
tracing::warn!("Failed to respond 200 OK to REGISTER: {}", status);
}
// Release pool — pjsip_endpt_respond_stateless clones what it
// needs into rdata's pool, so our header pool can be freed now.
pj_pool_release(pool);
return;
}
}
// Header creation failed — release the pool before falling through
pj_pool_release(pool);
}
// Fallback: respond without Expires header
let status =
pjsip_endpt_respond_stateless(endpt, rdata, 200, ptr::null(), ptr::null(), ptr::null());
if status != pj_constants__PJ_SUCCESS as i32 {
tracing::warn!("Failed to respond 200 OK to REGISTER: {}", status);
}
}
}
/// Detect transport type (UDP/TCP/TLS) from the incoming request.
unsafe fn detect_transport(rdata: *mut pjsip_rx_data) -> crate::services::registrar::SipTransport {
if !(*rdata).tp_info.transport.is_null() {
let tp_type = (*(*rdata).tp_info.transport).key.type_ as u32;
if tp_type == pjsip_transport_type_e_PJSIP_TRANSPORT_TLS {
crate::services::registrar::SipTransport::Tls
} else if tp_type == pjsip_transport_type_e_PJSIP_TRANSPORT_TCP {
crate::services::registrar::SipTransport::Tcp
unsafe {
if !(*rdata).tp_info.transport.is_null() {
let tp_type = (*(*rdata).tp_info.transport).key.type_ as u32;
if tp_type == pjsip_transport_type_e_PJSIP_TRANSPORT_TLS {
crate::services::registrar::SipTransport::Tls
} else if tp_type == pjsip_transport_type_e_PJSIP_TRANSPORT_TCP {
crate::services::registrar::SipTransport::Tcp
} else {
crate::services::registrar::SipTransport::Udp
}
} else {
crate::services::registrar::SipTransport::Udp
}
} else {
crate::services::registrar::SipTransport::Udp
}
}
@ -168,37 +177,39 @@ unsafe fn create_register_tsx(
rdata: *mut pjsip_rx_data,
expires: u32,
) -> Option<PendingRegisterTsx> {
let endpt = pjsua_get_pjsip_endpt();
let module_ptr = REGISTER_MODULE_PTR.load(Ordering::Acquire);
unsafe {
let endpt = pjsua_get_pjsip_endpt();
let module_ptr = REGISTER_MODULE_PTR.load(Ordering::Acquire);
if endpt.is_null() || module_ptr.is_null() {
return None;
if endpt.is_null() || module_ptr.is_null() {
return None;
}
// Create UAS transaction
let mut tsx: *mut pjsip_transaction = ptr::null_mut();
let status = pjsip_tsx_create_uas2(module_ptr, rdata, ptr::null_mut(), &mut tsx);
if status != pj_constants__PJ_SUCCESS as i32 || tsx.is_null() {
return None;
}
// Feed the request to the transaction (starts Timer F, stores headers)
pjsip_tsx_recv_msg(tsx, rdata);
// Pre-build a 200 OK response while rdata is still valid.
// The status code / reason will be modified before sending if auth fails.
let mut tdata: *mut pjsip_tx_data = ptr::null_mut();
let status = pjsip_endpt_create_response(endpt, rdata, 200, ptr::null(), &mut tdata);
if status != pj_constants__PJ_SUCCESS as i32 || tdata.is_null() {
pjsip_tsx_terminate(tsx, 500);
return None;
}
Some(PendingRegisterTsx {
tsx: SendableTsx(tsx),
tdata: SendableTdata(tdata),
expires,
})
}
// Create UAS transaction
let mut tsx: *mut pjsip_transaction = ptr::null_mut();
let status = pjsip_tsx_create_uas2(module_ptr, rdata, ptr::null_mut(), &mut tsx);
if status != pj_constants__PJ_SUCCESS as i32 || tsx.is_null() {
return None;
}
// Feed the request to the transaction (starts Timer F, stores headers)
pjsip_tsx_recv_msg(tsx, rdata);
// Pre-build a 200 OK response while rdata is still valid.
// The status code / reason will be modified before sending if auth fails.
let mut tdata: *mut pjsip_tx_data = ptr::null_mut();
let status = pjsip_endpt_create_response(endpt, rdata, 200, ptr::null(), &mut tdata);
if status != pj_constants__PJ_SUCCESS as i32 || tdata.is_null() {
pjsip_tsx_terminate(tsx, 500);
return None;
}
Some(PendingRegisterTsx {
tsx: SendableTsx(tsx),
tdata: SendableTdata(tdata),
expires,
})
}
// Main callback
@ -216,61 +227,63 @@ unsafe fn create_register_tsx(
/// b. Cache hit + mismatch -> immediate 403 Forbidden (stateless)
/// c. Cache miss -> defer via UAS transaction, verify via API, respond later
pub unsafe extern "C" fn on_rx_request_cb(rdata: *mut pjsip_rx_data) -> pj_bool_t {
if rdata.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
unsafe {
if rdata.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
let msg = (*rdata).msg_info.msg;
if msg.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
let msg = (*rdata).msg_info.msg;
if msg.is_null() {
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Check if this is a REGISTER request
let method_id = (*msg).line.req.method.id;
if method_id != pjsip_method_e_PJSIP_REGISTER_METHOD {
// Not REGISTER, let other modules handle it
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Check if this is a REGISTER request
let method_id = (*msg).line.req.method.id;
if method_id != pjsip_method_e_PJSIP_REGISTER_METHOD {
// Not REGISTER, let other modules handle it
return pj_constants__PJ_FALSE as pj_bool_t;
}
// Extract source IP for logging and ban checking
let source_ip = extract_source_ip(rdata);
let ip_str = source_ip
.map(|ip| ip.to_string())
.unwrap_or_else(|| "unknown".to_string());
// Extract source IP for logging and ban checking
let source_ip = extract_source_ip(rdata);
let ip_str = source_ip
.map(|ip| ip.to_string())
.unwrap_or_else(|| "unknown".to_string());
// Extract source port
let source_port = (*rdata).pkt_info.src_port as u16;
// Extract source port
let source_port = (*rdata).pkt_info.src_port as u16;
// Ban checks: skip if banning disabled or IP is whitelisted
if let Some(ip) = source_ip {
if let Some(ban_mgr) = crate::services::ban::global() {
if ban_mgr.is_enabled() && !ban_mgr.is_whitelisted(&ip) {
// Check if IP is banned
let result = ban_mgr.check_banned(&ip);
if result.is_banned {
tracing::debug!("Rejecting REGISTER from banned IP {}", ip);
send_simple_response(rdata, 403, "Forbidden");
return pj_constants__PJ_TRUE as pj_bool_t;
}
// Ban checks: skip if banning disabled or IP is whitelisted
if let Some(ip) = source_ip
&& let Some(ban_mgr) = crate::services::ban::global()
&& ban_mgr.is_enabled()
&& !ban_mgr.is_whitelisted(&ip)
{
// Check if IP is banned
let result = ban_mgr.check_banned(&ip);
if result.is_banned {
tracing::debug!("Rejecting REGISTER from banned IP {}", ip);
send_simple_response(rdata, 403, "Forbidden");
return pj_constants__PJ_TRUE as pj_bool_t;
}
}
}
// Check User-Agent for SIPVicious scanners - instant permaban
if let Some(user_agent) = extract_user_agent(rdata) {
if is_sipvicious_scanner(&user_agent) {
// Check User-Agent for SIPVicious scanners - instant permaban
if let Some(user_agent) = extract_user_agent(rdata)
&& is_sipvicious_scanner(&user_agent)
{
if let Some(ip) = source_ip {
if let Some(ban_mgr) = crate::services::ban::global() {
if ban_mgr.is_enabled() && !ban_mgr.is_whitelisted(&ip) {
let result =
ban_mgr.record_permanent_ban(ip, "sipvicious_scanner_register");
if result.should_log {
tracing::warn!(
"PERMABAN IP {} - SIPVicious scanner detected in REGISTER: User-Agent='{}'",
ip,
user_agent
);
}
if let Some(ban_mgr) = crate::services::ban::global()
&& ban_mgr.is_enabled()
&& !ban_mgr.is_whitelisted(&ip)
{
let result = ban_mgr.record_permanent_ban(ip, "sipvicious_scanner_register");
if result.should_log {
tracing::warn!(
"PERMABAN IP {} - SIPVicious scanner detected in REGISTER: User-Agent='{}'",
ip,
user_agent
);
}
}
} else {
@ -282,29 +295,30 @@ pub unsafe extern "C" fn on_rx_request_cb(rdata: *mut pjsip_rx_data) -> pj_bool_
send_simple_response(rdata, 403, "Forbidden");
return pj_constants__PJ_TRUE as pj_bool_t;
}
}
// Rate limit REGISTER requests
if let Some(ip) = source_ip {
if let Some(ban_mgr) = crate::services::ban::global() {
if ban_mgr.is_enabled() && !ban_mgr.is_whitelisted(&ip) && ban_mgr.record_register(ip) {
tracing::debug!("Rejecting REGISTER from {} - rate limit exceeded", ip);
send_simple_response(rdata, 429, "Too Many Requests");
return pj_constants__PJ_TRUE as pj_bool_t;
}
// Rate limit REGISTER requests
if let Some(ip) = source_ip
&& let Some(ban_mgr) = crate::services::ban::global()
&& ban_mgr.is_enabled()
&& !ban_mgr.is_whitelisted(&ip)
&& ban_mgr.record_register(ip)
{
tracing::debug!("Rejecting REGISTER from {} - rate limit exceeded", ip);
send_simple_response(rdata, 429, "Too Many Requests");
return pj_constants__PJ_TRUE as pj_bool_t;
}
}
// Try to extract Digest auth params from Authorization header
let digest_params = extract_digest_auth_from_rdata(rdata);
// Try to extract Digest auth params from Authorization header
let digest_params = extract_digest_auth_from_rdata(rdata);
if let Some(mut params) = digest_params {
// Has auth - fill in REGISTER method
params.method = "REGISTER".to_string();
if let Some(mut params) = digest_params {
// Has auth - fill in REGISTER method
params.method = "REGISTER".to_string();
// Check auth failure cooldown before processing
if let Some(cache) = crate::services::auth_cache::AuthCache::global() {
if cache.is_in_cooldown(&params.username) {
// Check auth failure cooldown before processing
if let Some(cache) = crate::services::auth_cache::AuthCache::global()
&& cache.is_in_cooldown(&params.username)
{
tracing::debug!(
"Rejecting REGISTER from {} (user={}) - auth cooldown active",
ip_str,
@ -313,71 +327,26 @@ pub unsafe extern "C" fn on_rx_request_cb(rdata: *mut pjsip_rx_data) -> pj_bool_
send_simple_response(rdata, 429, "Too Many Requests");
return pj_constants__PJ_TRUE as pj_bool_t;
}
}
// Extract fields needed for all code paths
let contact_uri = extract_contact_uri(rdata);
let expires = extract_expires(rdata);
let source_addr = source_ip.map(|ip| SocketAddr::new(ip, source_port));
let transport = detect_transport(rdata);
// Extract fields needed for all code paths
let contact_uri = extract_contact_uri(rdata);
let expires = extract_expires(rdata);
let source_addr = source_ip.map(|ip| SocketAddr::new(ip, source_port));
let transport = detect_transport(rdata);
// Auth cache verification
if let Some(cache) = crate::services::auth_cache::AuthCache::global() {
use crate::services::auth_cache::VerifyResult;
match cache.check(&params) {
VerifyResult::Verified => {
// Cache hit, auth OK — fast-path 200 OK
tracing::debug!(
"REGISTER auth OK (cached): user={} from {}",
params.username,
ip_str
);
send_register_ok(rdata, expires);
// Send to async handler for registrar update
if let Some(tx) = REGISTER_EVENT_TX.get() {
let _ = tx.try_send(RegisterRequest {
digest_auth: params,
contact_uri: contact_uri.unwrap_or_default(),
source_addr,
transport,
expires,
pending_tsx: None,
});
}
return pj_constants__PJ_TRUE as pj_bool_t;
}
VerifyResult::Mismatch => {
// Wrong password (cached HA1 didn't match) — 403
tracing::debug!(
"REGISTER auth mismatch (cached): user={} from {}",
params.username,
ip_str
);
send_simple_response(rdata, 403, "Forbidden");
// Send to async so API can re-verify (cache may be stale
// after a password change) and update failure counts
if let Some(tx) = REGISTER_EVENT_TX.get() {
let _ = tx.try_send(RegisterRequest {
digest_auth: params,
contact_uri: contact_uri.unwrap_or_default(),
source_addr,
transport,
expires,
pending_tsx: None,
});
}
return pj_constants__PJ_TRUE as pj_bool_t;
}
VerifyResult::Miss => {
// No cached HA1 — need API round-trip.
// Create a UAS transaction so we can respond after the
// async handler completes, without blocking pjsip.
tracing::debug!(
"REGISTER cache miss: user={} from {}, deferring to API",
params.username,
ip_str
);
if let Some(pending) = create_register_tsx(rdata, expires) {
// Auth cache verification
if let Some(cache) = crate::services::auth_cache::AuthCache::global() {
use crate::services::auth_cache::VerifyResult;
match cache.check(&params) {
VerifyResult::Verified => {
// Cache hit, auth OK — fast-path 200 OK
tracing::debug!(
"REGISTER auth OK (cached): user={} from {}",
params.username,
ip_str
);
send_register_ok(rdata, expires);
// Send to async handler for registrar update
if let Some(tx) = REGISTER_EVENT_TX.get() {
let _ = tx.try_send(RegisterRequest {
digest_auth: params,
@ -385,105 +354,151 @@ pub unsafe extern "C" fn on_rx_request_cb(rdata: *mut pjsip_rx_data) -> pj_bool_
source_addr,
transport,
expires,
pending_tsx: Some(pending),
pending_tsx: None,
});
}
return pj_constants__PJ_TRUE as pj_bool_t;
}
// Transaction creation failed — fall through to stateless
// 200 OK below (same behaviour as before this change).
tracing::warn!(
"Failed to create tsx for deferred REGISTER, falling back to stateless 200"
VerifyResult::Mismatch => {
// Wrong password (cached HA1 didn't match) — 403
tracing::debug!(
"REGISTER auth mismatch (cached): user={} from {}",
params.username,
ip_str
);
send_simple_response(rdata, 403, "Forbidden");
// Send to async so API can re-verify (cache may be stale
// after a password change) and update failure counts
if let Some(tx) = REGISTER_EVENT_TX.get() {
let _ = tx.try_send(RegisterRequest {
digest_auth: params,
contact_uri: contact_uri.unwrap_or_default(),
source_addr,
transport,
expires,
pending_tsx: None,
});
}
return pj_constants__PJ_TRUE as pj_bool_t;
}
VerifyResult::Miss => {
// No cached HA1 — need API round-trip.
// Create a UAS transaction so we can respond after the
// async handler completes, without blocking pjsip.
tracing::debug!(
"REGISTER cache miss: user={} from {}, deferring to API",
params.username,
ip_str
);
if let Some(pending) = create_register_tsx(rdata, expires) {
if let Some(tx) = REGISTER_EVENT_TX.get() {
let _ = tx.try_send(RegisterRequest {
digest_auth: params,
contact_uri: contact_uri.unwrap_or_default(),
source_addr,
transport,
expires,
pending_tsx: Some(pending),
});
}
return pj_constants__PJ_TRUE as pj_bool_t;
}
// Transaction creation failed — fall through to stateless
// 200 OK below (same behaviour as before this change).
tracing::warn!(
"Failed to create tsx for deferred REGISTER, falling back to stateless 200"
);
}
}
}
// Default path: stateless 200 OK + async verification
// (non-sipcord builds, auth cache unavailable, or tsx creation failed)
tracing::debug!(
"REGISTER with auth from {} (user={}), responding 200 OK (stateless)",
ip_str,
params.username
);
if let Some(tx) = REGISTER_EVENT_TX.get() {
let _ = tx.try_send(RegisterRequest {
digest_auth: params,
contact_uri: contact_uri.unwrap_or_default(),
source_addr,
transport,
expires,
pending_tsx: None,
});
}
send_register_ok(rdata, expires);
} else {
// No Authorization header - send 401 challenge
tracing::debug!(
"REGISTER without auth from {}, sending 401 challenge",
ip_str
);
let endpt = pjsua_get_pjsip_endpt();
if endpt.is_null() {
tracing::error!("Failed to get PJSIP endpoint for REGISTER 401 response");
return pj_constants__PJ_TRUE as pj_bool_t;
}
// Generate a cryptographically random nonce
let nonce = {
let bytes: [u8; 16] = rand::random();
bytes
.iter()
.map(|b| format!("{:02x}", b))
.collect::<String>()
};
let www_auth = format!(
"Digest realm=\"{}\", nonce=\"{}\", algorithm=MD5, qop=\"auth\"",
SIP_REALM, nonce
);
// Create WWW-Authenticate header
let hdr_name = CString::new("WWW-Authenticate").unwrap();
let hdr_value = CString::new(www_auth).unwrap();
let pool = pjsua_pool_create(c"register_401".as_ptr(), 512, 512);
if pool.is_null() {
tracing::error!("Failed to create pool for REGISTER 401 response");
return pj_constants__PJ_TRUE as pj_bool_t;
}
let name = pj_str(hdr_name.as_ptr() as *mut c_char);
let value = pj_str(hdr_value.as_ptr() as *mut c_char);
let hdr = pjsip_generic_string_hdr_create(pool, &name, &value);
if !hdr.is_null() {
let hdr_list =
pj_pool_alloc(pool, std::mem::size_of::<pjsip_hdr>()) as *mut pjsip_hdr;
if !hdr_list.is_null() {
pj_list_init_hdr(hdr_list);
pj_list_insert_before(hdr_list as *mut pj_list_type, hdr as *mut pj_list_type);
let status = pjsip_endpt_respond_stateless(
endpt,
rdata,
401,
ptr::null(),
hdr_list,
ptr::null(),
);
if status != pj_constants__PJ_SUCCESS as i32 {
tracing::warn!("Failed to respond 401 to REGISTER: {}", status);
}
}
}
// Release pool — pjsip_endpt_respond_stateless clones headers internally
pj_pool_release(pool);
}
// Default path: stateless 200 OK + async verification
// (non-sipcord builds, auth cache unavailable, or tsx creation failed)
tracing::debug!(
"REGISTER with auth from {} (user={}), responding 200 OK (stateless)",
ip_str,
params.username
);
if let Some(tx) = REGISTER_EVENT_TX.get() {
let _ = tx.try_send(RegisterRequest {
digest_auth: params,
contact_uri: contact_uri.unwrap_or_default(),
source_addr,
transport,
expires,
pending_tsx: None,
});
}
send_register_ok(rdata, expires);
} else {
// No Authorization header - send 401 challenge
tracing::debug!(
"REGISTER without auth from {}, sending 401 challenge",
ip_str
);
let endpt = pjsua_get_pjsip_endpt();
if endpt.is_null() {
tracing::error!("Failed to get PJSIP endpoint for REGISTER 401 response");
return pj_constants__PJ_TRUE as pj_bool_t;
}
// Generate a cryptographically random nonce
let nonce = {
let bytes: [u8; 16] = rand::random();
bytes
.iter()
.map(|b| format!("{:02x}", b))
.collect::<String>()
};
let www_auth = format!(
"Digest realm=\"{}\", nonce=\"{}\", algorithm=MD5, qop=\"auth\"",
SIP_REALM, nonce
);
// Create WWW-Authenticate header
let hdr_name = CString::new("WWW-Authenticate").unwrap();
let hdr_value = CString::new(www_auth).unwrap();
let pool = pjsua_pool_create(c"register_401".as_ptr(), 512, 512);
if pool.is_null() {
tracing::error!("Failed to create pool for REGISTER 401 response");
return pj_constants__PJ_TRUE as pj_bool_t;
}
let name = pj_str(hdr_name.as_ptr() as *mut c_char);
let value = pj_str(hdr_value.as_ptr() as *mut c_char);
let hdr = pjsip_generic_string_hdr_create(pool, &name, &value);
if !hdr.is_null() {
let hdr_list = pj_pool_alloc(pool, std::mem::size_of::<pjsip_hdr>()) as *mut pjsip_hdr;
if !hdr_list.is_null() {
pj_list_init_hdr(hdr_list);
pj_list_insert_before(hdr_list as *mut pj_list_type, hdr as *mut pj_list_type);
let status = pjsip_endpt_respond_stateless(
endpt,
rdata,
401,
ptr::null(),
hdr_list,
ptr::null(),
);
if status != pj_constants__PJ_SUCCESS as i32 {
tracing::warn!("Failed to respond 401 to REGISTER: {}", status);
}
}
}
// Release pool — pjsip_endpt_respond_stateless clones headers internally
pj_pool_release(pool);
// Return TRUE to indicate we handled this request
pj_constants__PJ_TRUE as pj_bool_t
}
// Return TRUE to indicate we handled this request
pj_constants__PJ_TRUE as pj_bool_t
}
// Extraction helpers
@ -494,56 +509,58 @@ unsafe fn extract_contact_uri(rdata: *mut pjsip_rx_data) -> Option<String> {
return None;
}
let msg = (*rdata).msg_info.msg;
if msg.is_null() {
return None;
unsafe {
let msg = (*rdata).msg_info.msg;
if msg.is_null() {
return None;
}
let contact_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut())
as *const pjsip_contact_hdr;
if contact_hdr.is_null() {
return None;
}
let uri = (*contact_hdr).uri;
if uri.is_null() {
return None;
}
// The Contact header URI is typically a pjsip_name_addr wrapping a pjsip_sip_uri.
// We must unwrap it via the vtable's p_get_uri (equivalent to pjsip_uri_get_uri()
// which is an inline C function not available through FFI).
let uri_vptr = (*(uri as *const pjsip_uri)).vptr;
if uri_vptr.is_null() {
return None;
}
let get_uri_fn = (*uri_vptr).p_get_uri?;
let sip_uri_raw = get_uri_fn(uri as *mut std::os::raw::c_void);
if sip_uri_raw.is_null() {
return None;
}
let sip_uri = sip_uri_raw as *const pjsip_sip_uri;
if (*sip_uri).host.ptr.is_null() || (*sip_uri).host.slen <= 0 {
return None;
}
let host = pj_str_to_string(&(*sip_uri).host);
let port = (*sip_uri).port;
let user = if !(*sip_uri).user.ptr.is_null() && (*sip_uri).user.slen > 0 {
Some(pj_str_to_string(&(*sip_uri).user))
} else {
None
};
let uri_str = match (user, port) {
(Some(u), p) if p > 0 => format!("sip:{}@{}:{}", u, host, p),
(Some(u), _) => format!("sip:{}@{}", u, host),
(None, p) if p > 0 => format!("sip:{}:{}", host, p),
(None, _) => format!("sip:{}", host),
};
Some(uri_str)
}
let contact_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_CONTACT, ptr::null_mut())
as *const pjsip_contact_hdr;
if contact_hdr.is_null() {
return None;
}
let uri = (*contact_hdr).uri;
if uri.is_null() {
return None;
}
// The Contact header URI is typically a pjsip_name_addr wrapping a pjsip_sip_uri.
// We must unwrap it via the vtable's p_get_uri (equivalent to pjsip_uri_get_uri()
// which is an inline C function not available through FFI).
let uri_vptr = (*(uri as *const pjsip_uri)).vptr;
if uri_vptr.is_null() {
return None;
}
let get_uri_fn = (*uri_vptr).p_get_uri?;
let sip_uri_raw = get_uri_fn(uri as *mut std::os::raw::c_void);
if sip_uri_raw.is_null() {
return None;
}
let sip_uri = sip_uri_raw as *const pjsip_sip_uri;
if (*sip_uri).host.ptr.is_null() || (*sip_uri).host.slen <= 0 {
return None;
}
let host = pj_str_to_string(&(*sip_uri).host);
let port = (*sip_uri).port;
let user = if !(*sip_uri).user.ptr.is_null() && (*sip_uri).user.slen > 0 {
Some(pj_str_to_string(&(*sip_uri).user))
} else {
None
};
let uri_str = match (user, port) {
(Some(u), p) if p > 0 => format!("sip:{}@{}:{}", u, host, p),
(Some(u), _) => format!("sip:{}@{}", u, host),
(None, p) if p > 0 => format!("sip:{}:{}", host, p),
(None, _) => format!("sip:{}", host),
};
Some(uri_str)
}
/// Extract Expires value from REGISTER request (header or Contact param)
@ -552,21 +569,23 @@ unsafe fn extract_expires(rdata: *mut pjsip_rx_data) -> u32 {
return 3600;
}
let msg = (*rdata).msg_info.msg;
if msg.is_null() {
return 3600;
unsafe {
let msg = (*rdata).msg_info.msg;
if msg.is_null() {
return 3600;
}
// Try Expires header first
let expires_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_EXPIRES, ptr::null_mut())
as *const pjsip_expires_hdr;
if !expires_hdr.is_null() {
return (*expires_hdr).ivalue as u32;
}
// Default
3600
}
// Try Expires header first
let expires_hdr = pjsip_msg_find_hdr(msg, pjsip_hdr_e_PJSIP_H_EXPIRES, ptr::null_mut())
as *const pjsip_expires_hdr;
if !expires_hdr.is_null() {
return (*expires_hdr).ivalue as u32;
}
// Default
3600
}
// Types