1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
//! SIP server implementation
pub mod actions;
use crate::server::connection::ConnectionId;
use anyhow::Result;
use std::net::SocketAddr;
use std::sync::{Arc, LazyLock};
use tokio::net::UdpSocket;
use tokio::sync::mpsc;
use crate::llm::action_helper::call_llm;
use crate::llm::ollama_client::OllamaClient;
use crate::logging::emit::Log;
use crate::protocol::Event;
use crate::server::SipProtocol;
use crate::state::app_state::AppState;
use actions::{
SIP_ACK_EVENT, SIP_BYE_EVENT, SIP_CANCEL_EVENT, SIP_INVITE_EVENT, SIP_OPTIONS_EVENT,
SIP_REGISTER_EVENT,
};
/// SIP server that handles VoIP signaling
pub struct SipServer;
impl SipServer {
/// Spawn SIP server with integrated LLM actions
pub async fn spawn_with_llm_actions(
listen_addr: SocketAddr,
llm_client: OllamaClient,
app_state: Arc<AppState>,
status_tx: mpsc::UnboundedSender<String>,
server_id: crate::state::ServerId,
) -> Result<SocketAddr> {
let socket = Arc::new(UdpSocket::bind(listen_addr).await?);
let local_addr = socket.local_addr()?;
Log::new(Some(&status_tx)).info(format!("SIP server listening on {}", local_addr));
let protocol = Arc::new(SipProtocol::new());
let task_registrar = app_state.clone();
let accept_handle = tokio::spawn(async move {
let mut buffer = vec![0u8; 65535]; // Max UDP packet size
let log = Log::new(Some(&status_tx));
loop {
match socket.recv_from(&mut buffer).await {
Ok((n, peer_addr)) => {
let data = buffer[..n].to_vec();
let connection_id =
ConnectionId::new(app_state.get_next_unified_id().await);
// Add connection to ServerInstance
use crate::state::server::{
ConnectionState as ServerConnectionState, ConnectionStatus,
ProtocolConnectionInfo,
};
let now = crate::utils::clock::Instant::now();
let conn_state = ServerConnectionState {
id: connection_id,
remote_addr: peer_addr,
local_addr,
bytes_sent: 0,
bytes_received: n as u64,
packets_sent: 0,
packets_received: 1,
last_activity: now,
status: ConnectionStatus::Active,
status_changed_at: now,
protocol_info: ProtocolConnectionInfo::empty(),
};
app_state
.add_connection_to_server(server_id, conn_state)
.await;
let _ = status_tx.send("__UPDATE_UI__".to_string());
// Summary + full payload FileOnly: the sip_* event templates
// render the equivalent line to the TUI.
log.debug(format!("SIP received {} bytes from {}", n, peer_addr));
if let Ok(text) = String::from_utf8(data.clone()) {
let preview = crate::utils::truncate_for_log(&text, 200);
log.trace(format!("SIP message: {}", preview));
}
let llm_clone = llm_client.clone();
let state_clone = app_state.clone();
let status_clone = status_tx.clone();
let socket_clone = socket.clone();
let protocol_clone = protocol.clone();
// Tracked, not detached: stop_server must abort this task too.
let task_owner = app_state.clone();
task_owner
.spawn_server_task(server_id, async move {
let log = Log::new(Some(&status_clone));
// Parse SIP message
let sip_message = match Self::parse_sip_message(&data) {
Ok(msg) => msg,
Err(e) => {
log.warn(format!(
"SIP failed to parse message from {}: {}",
peer_addr, e
));
return;
}
};
// Request summary FileOnly: the sip_* event template renders
// the equivalent line to the TUI.
log.debug(format!(
"SIP {} request from {} (Call-ID: {})",
sip_message.method, peer_addr, sip_message.call_id
));
// Create event based on SIP method
let event = Self::create_event(
&sip_message,
peer_addr,
local_addr,
connection_id,
);
// Call LLM with SIP event
match call_llm(
&llm_clone,
&state_clone,
server_id,
Some(connection_id),
&event,
protocol_clone.as_ref(),
)
.await
{
Ok(execution_result) => {
// ACK takes no response at all (RFC 3261 §17), so the
// whole reply path is skipped for it - including the
// no-answer refusal below. Answering an ACK is a protocol
// violation whatever the model said.
if sip_message.method == "ACK" {
log.debug(format!(
"SIP ACK from {} (Call-ID {}) decision=model_answer; \
ACK takes no response",
peer_addr, sip_message.call_id
));
return;
}
// Pick the action that is actually a SIP response. The model
// may legitimately return a common action (set_memory,
// schedule_task, ...) alongside - or ahead of - its reply, and
// taking `raw_actions.first()` blindly turned that into a
// fabricated status line while the real answer was ignored.
let response_action = execution_result
.raw_actions
.iter()
.find(|a| is_sip_response_action(a));
let Some(action) = response_action else {
// The model was reachable and produced no SIP response.
// That is not an accept: on REGISTER or INVITE, staying
// silent leaves the UAC retransmitting for 32 seconds and
// tells the operator nothing, so refuse explicitly and
// keep the decision distinguishable from a backend
// failure (radius' decision= shape).
log.warn(format!(
"SIP {} from {} (Call-ID {}) decision=model_silent; \
refusing with 500 (the model returned no SIP \
response action)",
sip_message.method, peer_addr, sip_message.call_id
));
let refusal = Self::build_sip_response(
&sip_message,
&serde_json::json!({
"status_code": 500,
"reason_phrase": "Server Internal Error"
}),
);
if let Err(e) =
socket_clone.send_to(&refusal, peer_addr).await
{
log.error(format!(
"SIP failed to send 500 to {}: {}",
peer_addr, e
));
}
return;
};
// An explicit non-2xx is the model refusing, which must stay
// separable in the log from the silence above and from a
// backend outage below.
let decision = match action
.get("status_code")
.and_then(|v| v.as_u64())
{
Some(code) if !(200..300).contains(&code) => {
"model_reject"
}
Some(_) => "model_answer",
// build_sip_response answers 500 for this; naming it
// here keeps "the model forgot a field" apart from
// "the model chose 500".
None => "fail_closed_malformed_action",
};
log.debug(format!(
"SIP {} from {} (Call-ID {}) decision={}",
sip_message.method,
peer_addr,
sip_message.call_id,
decision
));
// Build SIP response from action JSON
let response =
Self::build_sip_response(&sip_message, action);
// Send SIP response
match socket_clone.send_to(&response, peer_addr).await {
Ok(sent) => {
// Summary FileOnly: the send action template
// already reports the send to the TUI.
log.debug(format!(
"SIP sent {} byte response to {}",
sent, peer_addr
));
}
Err(e) => {
log.error(format!(
"SIP failed to send response: {}",
e
));
}
}
// Media flow: SIP is signaling only, but when built with
// the `rtp` feature and the accept action carries an
// `rtp_audio` description, we honour the SDP we just
// negotiated by streaming real RTP to the caller's
// advertised media address. This is what makes a SIP INVITE
// result in RTP actually arriving, rather than a 200 OK with
// an SDP that points at nothing.
{
// No `unwrap_or(200)`: a missing status_code was answered
// 500 above, and treating it as 200 here would stream
// media for a call that was never accepted.
let status_code = action
.get("status_code")
.and_then(|v| v.as_u64())
.unwrap_or(0);
if sip_message.method == "INVITE" && status_code == 200
{
if let Some(rtp_audio) = action.get("rtp_audio") {
Self::stream_invite_media(
sip_message.body.as_deref(),
peer_addr,
rtp_audio.clone(),
status_clone.clone(),
state_clone.clone(),
server_id,
)
.await;
}
}
}
}
Err(e) => {
// Non-fatal: a 503 (wire fallback) is built and sent
// below for everything except ACK, so this is WARN.
log.warn(format!(
"SIP LLM error for {} from {} (Call-ID {}): {}",
sip_message.method, peer_addr, sip_message.call_id, e
));
// ACK is the one method that must never be answered
// (RFC 3261 §17: ACK is not a transaction that takes a
// response), so silence is correct there and only there.
if sip_message.method == "ACK" {
log.debug("SIP ACK needs no response; nothing sent");
return;
}
// Everything else gets a 503 Service Unavailable built
// from the request, so the Via/From/To/Call-ID/CSeq match
// and the client's transaction actually completes. Left
// silent, the UAC retransmits on timers A/E and only gives
// up after Timer B/F (32s).
if crate::llm::is_overload_error(&e) {
log.warn(format!(
"SIP 503 to {} (Call-ID {}): LLM capacity exhausted",
peer_addr, sip_message.call_id
));
}
let response = Self::build_sip_response(
&sip_message,
&serde_json::json!({
"status_code": 503,
"reason_phrase": "Service Unavailable",
"retry_after": 5
}),
);
match socket_clone.send_to(&response, peer_addr).await {
Ok(sent) => {
log.debug(format!(
"SIP 503 Service Unavailable to {} ({} bytes)",
peer_addr, sent
));
}
Err(send_err) => {
log.error(format!(
"SIP failed to send 503 to {}: {}",
peer_addr, send_err
));
}
}
}
}
})
.await;
}
Err(e) => {
log.error(format!("SIP recv error: {}", e));
}
}
}
});
// Register the accept loop so stop_server can abort it and release the port.
task_registrar
.register_server_task(server_id, accept_handle)
.await;
Ok(local_addr)
}
/// Parse SIP message from bytes
fn parse_sip_message(data: &[u8]) -> Result<SipMessage> {
let text = String::from_utf8(data.to_vec())?;
let lines: Vec<&str> = text.lines().collect();
if lines.is_empty() {
anyhow::bail!("Empty SIP message");
}
// Parse request line (e.g., "REGISTER sip:example.com SIP/2.0")
let request_line: Vec<&str> = lines[0].split_whitespace().collect();
if request_line.len() < 3 {
anyhow::bail!("Invalid SIP request line");
}
let method = request_line[0].to_string();
let request_uri = request_line[1].to_string();
// Parse headers
let mut call_id = String::new();
let mut from = String::new();
let mut to = String::new();
let mut via = Vec::new();
let mut cseq = String::new();
let mut contact = None;
let mut expires = None;
let mut content_type = None;
let mut content_length = 0;
let mut i = 1;
while i < lines.len() {
let line = lines[i];
if line.is_empty() {
// End of headers
i += 1;
break;
}
if let Some(colon_pos) = line.find(':') {
let (header_name, header_value) = line.split_at(colon_pos);
let header_value = header_value[1..].trim();
match header_name.to_lowercase().as_str() {
"call-id" => call_id = header_value.to_string(),
"from" | "f" => from = header_value.to_string(),
"to" | "t" => to = header_value.to_string(),
"via" | "v" => via.push(header_value.to_string()),
"cseq" => cseq = header_value.to_string(),
"contact" | "m" => contact = Some(header_value.to_string()),
"expires" => expires = header_value.parse().ok(),
"content-type" | "c" => content_type = Some(header_value.to_string()),
"content-length" | "l" => content_length = header_value.parse().unwrap_or(0),
_ => {}
}
}
i += 1;
}
// Parse body (SDP)
let mut body = String::new();
if content_length > 0 && i < lines.len() {
body = lines[i..].join("\r\n");
}
Ok(SipMessage {
method,
request_uri,
call_id,
from,
to,
via,
cseq,
contact,
expires,
content_type,
body: if body.is_empty() { None } else { Some(body) },
})
}
/// Create event from SIP message
fn create_event(
sip_message: &SipMessage,
peer_addr: SocketAddr,
local_addr: SocketAddr,
connection_id: ConnectionId,
) -> Event {
let base_data = serde_json::json!({
"peer_addr": peer_addr.to_string(),
"local_addr": local_addr.to_string(),
"connection_id": connection_id.to_string(),
"call_id": sip_message.call_id,
"from": sip_message.from,
"to": sip_message.to,
"cseq": sip_message.cseq,
});
match sip_message.method.as_str() {
"REGISTER" => {
let mut data = base_data;
if let Some(contact) = &sip_message.contact {
data["contact"] = serde_json::json!(contact);
}
if let Some(expires) = sip_message.expires {
data["expires"] = serde_json::json!(expires);
}
Event {
event_type: &SIP_REGISTER_EVENT,
data,
}
}
"INVITE" => {
let mut data = base_data;
if let Some(body) = &sip_message.body {
data["sdp"] = serde_json::json!(body);
}
Event {
event_type: &SIP_INVITE_EVENT,
data,
}
}
"BYE" => Event {
event_type: &SIP_BYE_EVENT,
data: base_data,
},
"ACK" => Event {
event_type: &SIP_ACK_EVENT,
data: base_data,
},
"OPTIONS" => {
let mut data = base_data;
data["answer_with"] = serde_json::json!(actions::OPTIONS_ANSWER_WITH);
Event {
event_type: &SIP_OPTIONS_EVENT,
data,
}
}
"CANCEL" => Event {
event_type: &SIP_CANCEL_EVENT,
data: base_data,
},
_ => {
// Unknown method, treat as OPTIONS event
Event {
event_type: &SIP_OPTIONS_EVENT,
data: base_data,
}
}
}
}
/// Build SIP response from action JSON
fn build_sip_response(request: &SipMessage, response_action: &serde_json::Value) -> Vec<u8> {
// A non-object action used to `.expect(...)` here. That is a panic reachable from the
// wire - the model's own output, inside a `tokio::spawn` that swallows the panic - so
// the peer got silence, the log got nothing, and the server went on reporting Running.
// An empty object routes into the missing-status_code branch below, which is the same
// fail-closed 500 a malformed action deserves.
static EMPTY: LazyLock<serde_json::Map<String, serde_json::Value>> =
LazyLock::new(serde_json::Map::new);
let response_data = match response_action.as_object() {
Some(obj) => obj,
None => {
tracing::warn!(
"SIP action was not a JSON object ({}); answering 500",
crate::utils::truncate_for_log(&response_action.to_string(), 120)
);
&EMPTY
}
};
// A missing `status_code` is a malformed action, not a decision, so it must not
// become 200. On REGISTER a defaulted 200 is an accepted registration and on INVITE
// an accepted call - granted because a field was forgotten. SIP has no default
// status, so the honest answer is the server's own 500; the model's actual choices
// (401, 403, 486) are unaffected because they name the field.
let (status_code, default_reason) =
match response_data.get("status_code").and_then(|v| v.as_u64()) {
Some(code) => (code as u16, actions::default_reason_phrase(code as u16)),
None => {
tracing::warn!(
"SIP action carried no status_code; answering 500 rather than defaulting \
to 200, which would accept the request"
);
(500u16, "Server Internal Error")
}
};
let reason_phrase = response_data
.get("reason_phrase")
.and_then(|v| v.as_str())
.unwrap_or(default_reason);
// Build status line
let mut response = format!("SIP/2.0 {} {}\r\n", status_code, reason_phrase);
// Add Via headers (copy from request)
for via in &request.via {
response.push_str(&format!("Via: {}\r\n", via));
}
// Add From header (copy from request)
response.push_str(&format!("From: {}\r\n", request.from));
// Add To header (copy from request, add tag if not present)
let to_header = if request.to.contains(";tag=") {
request.to.clone()
} else {
// Generate a tag for the response
format!("{};tag={}", request.to, Self::generate_tag())
};
response.push_str(&format!("To: {}\r\n", to_header));
// Add Call-ID header (copy from request)
response.push_str(&format!("Call-ID: {}\r\n", request.call_id));
// Add CSeq header (copy from request)
response.push_str(&format!("CSeq: {}\r\n", request.cseq));
// Add Contact header for successful REGISTER
if request.method == "REGISTER" && status_code == 200 {
if let Some(contact) = &request.contact {
response.push_str(&format!("Contact: {}\r\n", contact));
}
}
// Add Expires header for REGISTER responses
if request.method == "REGISTER" {
let expires = response_data
.get("expires")
.and_then(|v| v.as_u64())
.unwrap_or(3600);
response.push_str(&format!("Expires: {}\r\n", expires));
}
// Add Retry-After when the action asks for one. RFC 3261 §20.33: a 503 SHOULD
// carry it, so the caller retries after a bounded wait instead of failing over
// to another proxy or giving up on the URI entirely.
if let Some(retry_after) = response_data.get("retry_after").and_then(|v| v.as_u64()) {
response.push_str(&format!("Retry-After: {}\r\n", retry_after));
}
// Add Allow header for OPTIONS responses
if request.method == "OPTIONS" {
if let Some(allow_methods) = response_data.get("allow_methods") {
if let Some(methods) = allow_methods.as_array() {
let methods_str: Vec<String> = methods
.iter()
.filter_map(|v| v.as_str())
.map(String::from)
.collect();
response.push_str(&format!("Allow: {}\r\n", methods_str.join(", ")));
}
}
}
// Add body (SDP) for successful INVITE responses
let body = if request.method == "INVITE" && status_code == 200 {
response_data
.get("sdp")
.and_then(|v| v.as_str())
.map(String::from)
} else {
None
};
// Add Content-Length and body
if let Some(body_text) = &body {
response.push_str("Content-Type: application/sdp\r\n");
response.push_str(&format!("Content-Length: {}\r\n", body_text.len()));
response.push_str("\r\n");
response.push_str(body_text);
} else {
response.push_str("Content-Length: 0\r\n");
response.push_str("\r\n");
}
response.into_bytes()
}
/// Generate a random tag for SIP responses
fn generate_tag() -> String {
use rand::Rng;
let tag: u32 = rand::thread_rng().gen();
format!("{:x}", tag)
}
/// Stream RTP audio to the media address the caller advertised in its INVITE SDP.
///
/// Only compiled with the `rtp` feature. The `rtp_audio` value is the same structured media
/// description the RTP protocol understands (`content`, `tone_hz`, `payload_type`,
/// `duration_ms`) — the model describes what the call carries, and the shared media engine
/// (`crate::server::rtp::media`) owns the samples and RTP framing. A fresh ephemeral UDP
/// socket is used because RTP must not share the SIP signaling port.
async fn stream_invite_media(
caller_sdp: Option<&str>,
caller_addr: SocketAddr,
rtp_audio: serde_json::Value,
status_tx: mpsc::UnboundedSender<String>,
app_state: Arc<AppState>,
server_id: crate::state::ServerId,
) {
use crate::server::rtp::media::{self, AudioCodec, RtpPacketizer};
let Some((ip, port)) = caller_sdp.and_then(parse_sdp_audio_target) else {
Log::new(Some(&status_tx)).warn(
"SIP INVITE had rtp_audio but no parseable m=audio target in the caller's SDP",
);
return;
};
// Refuse to stream anywhere but the source of the INVITE. SIP runs on UDP, so the
// source address is trivially spoofable and the SDP body names its own destination: an
// attacker who sends one 400-byte INVITE claiming `c=IN IP4 <victim>` would otherwise
// have NetGet blast seconds of G.711 at a third party it never heard from. Sending only
// to the address that actually spoke to us makes the media path unamplifiable, and the
// legitimate case is unaffected because a real UAC advertises its own address.
if ip != caller_addr.ip() {
Log::new(Some(&status_tx)).warn(format!(
"SIP INVITE decision=refused_media_redirect: SDP asks for media at {} but the \
INVITE came from {}; refusing to stream RTP to a third party",
ip,
caller_addr.ip()
));
return;
}
let target = std::net::SocketAddr::new(ip, port);
let codec = rtp_audio
.get("payload_type")
.and_then(|v| v.as_str())
.map(AudioCodec::parse)
.unwrap_or(Ok(AudioCodec::Pcmu))
.unwrap_or(AudioCodec::Pcmu);
let content = media::parse_audio_content(&rtp_audio)
.unwrap_or(media::AudioContent::Tone { hz: 440.0 });
let duration_ms = rtp_audio
.get("duration_ms")
.and_then(|v| v.as_u64())
.unwrap_or(2000);
// Tracked, not detached: stop_server must abort this task too.
let task_owner = app_state.clone();
task_owner
.spawn_server_task(server_id, async move {
let log = Log::new(Some(&status_tx));
let socket = match tokio::net::UdpSocket::bind(std::net::SocketAddr::from((
[0u8, 0, 0, 0],
0,
)))
.await
{
Ok(s) => s,
Err(e) => {
log.error(format!("SIP RTP socket bind failed: {}", e));
return;
}
};
let payload = match media::synthesize(codec, &content, duration_ms) {
Ok(p) => p,
Err(e) => {
log.warn(format!("SIP RTP synthesis: {}", e));
return;
}
};
let mut packetizer =
RtpPacketizer::new(rand::random(), codec.payload_type(), None, None);
let packets = packetizer.packetize(&payload, media::G711_SAMPLES_PER_FRAME);
let mut sent = 0u64;
for pkt in &packets {
if socket.send_to(pkt, target).await.is_err() {
break;
}
sent += 1;
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
}
log.info(format!(
"SIP call media: {} RTP {} packet(s) to {}",
sent,
codec.rtpmap_name(),
target
));
})
.await;
}
}
/// True if this action is one of the six SIP response verbs, i.e. something
/// [`SipServer::build_sip_response`] can turn into a status line.
///
/// The reply path selects on this rather than taking the first action the model returned: a
/// common action (`set_memory`, `schedule_task`, …) returned alongside the reply is executed by
/// the shared executor and must not be mistaken for one.
fn is_sip_response_action(action: &serde_json::Value) -> bool {
matches!(
action.get("type").and_then(|v| v.as_str()),
Some("sip_register" | "sip_invite" | "sip_bye" | "sip_ack" | "sip_options" | "sip_cancel")
)
}
/// Extract the media target (connection IP + audio port) from an SDP offer.
///
/// Reads the `c=IN IP4 <ip>` connection line and the first `m=audio <port> ...` line. Returns
/// None if either is missing or unparseable.
fn parse_sdp_audio_target(sdp: &str) -> Option<(std::net::IpAddr, u16)> {
let mut ip: Option<std::net::IpAddr> = None;
let mut port: Option<u16> = None;
for line in sdp.lines() {
let line = line.trim();
if let Some(rest) = line.strip_prefix("c=IN IP4 ") {
ip = rest
.trim()
.split('/')
.next()
.and_then(|s| s.trim().parse().ok());
} else if let Some(rest) = line.strip_prefix("m=audio ") {
port = rest.split_whitespace().next().and_then(|s| s.parse().ok());
}
}
Some((ip?, port?))
}
/// Parsed SIP message
#[derive(Debug, Clone)]
struct SipMessage {
method: String,
#[allow(dead_code)]
request_uri: String,
call_id: String,
from: String,
to: String,
via: Vec<String>,
cseq: String,
contact: Option<String>,
expires: Option<u32>,
#[allow(dead_code)]
content_type: Option<String>,
body: Option<String>,
}