feat(ami): M7 CP3 — AMI Originate waits for answer before running the app

crates/asterisk-ami/src/actions.rs

@@ -663,7 +663,7 @@ fn handle_originate(
         .get_header("Priority")
         .and_then(|p| p.parse::<i32>().ok())
         .unwrap_or(1);
-    let _timeout = action
+    let timeout_ms = action
         .get_header("Timeout")
         .and_then(|t| t.parse::<u64>().ok())
         .unwrap_or(30000);
@@ -802,6 +802,63 @@ fn handle_originate(
             ch.state = call_channel.state;
         }
 
+        // CP3: a PJSIP origination must NOT run its dialplan app/exten until the
+        // far end ANSWERS (200 OK). driver.call() only PUTS THE INVITE ON THE
+        // WIRE; the answer arrives asynchronously (handle_response's 2xx path
+        // flips the store channel to Up). Running the app now would emit SIP/RTP
+        // before answer. Wait up to the Originate Timeout for Up; on
+        // hangup/rejection/timeout, abandon the leg (CANCEL a still-pending
+        // INVITE) WITHOUT running the app, and report failure.
+        if tech.eq_ignore_ascii_case("PJSIP") {
+            let deadline = std::time::Instant::now()
+                + std::time::Duration::from_millis(timeout_ms.max(1));
+            let mut answered = false;
+            loop {
+                let (state, hung) = {
+                    let ch = store_chan.lock();
+                    (ch.state, ch.check_hangup())
+                };
+                if state == asterisk_types::ChannelState::Up {
+                    answered = true;
+                    break;
+                }
+                // A rejection/failure softhangs-up the leg (or drops it to Down/
+                // Busy) before answer — stop waiting and do not run the app.
+                if hung
+                    || state == asterisk_types::ChannelState::Down
+                    || state == asterisk_types::ChannelState::Busy
+                {
+                    break;
+                }
+                if std::time::Instant::now() >= deadline {
+                    break;
+                }
+                tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+            }
+            if !answered {
+                warn!(
+                    "Originate: PJSIP channel {} not answered within {}ms; abandoning before app",
+                    chan_name, timeout_ms
+                );
+                // Put the correct request on the wire for the unanswered leg (a
+                // CANCEL for a still-pending INVITE) and release its resources.
+                if let Some(handler) = asterisk_sip::get_global_event_handler() {
+                    handler.cancel_or_bye_outbound_leg(&chan_name).await;
+                } else {
+                    let _ = driver.hangup(&mut call_channel).await;
+                }
+                release_originate_leg(&tech, &chan_name, &chan_uid);
+                crate::event_bus::publish_event(
+                    crate::protocol::AmiEvent::new("OriginateResponse", 0x02)
+                        .with_header("Response", "Failure")
+                        .with_header("Reason", "3") // no answer
+                        .with_header("Channel", &chan_name)
+                        .with_header("Uniqueid", &chan_uid),
+                );
+                return;
+            }
+        }
+
         // Create a tokio::sync::Mutex copy for execution on ;1
         let pbx_channel = {
             let guard = store_chan.lock();

tests/cp3-originate-wait/ami_originate.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Trigger one rustisk outbound Dial via a single authenticated AMI Originate.
+
+    ami_originate.py HOST PORT ENDPOINT ACTION_ID
+
+Sends `Originate Channel: PJSIP/<ENDPOINT>` asynchronously. rustisk resolves the
+endpoint's contact (live registrar binding preferred over static config) and
+sends the INVITE to it. Prints the AMI response; exits nonzero if the action was
+not queued.
+"""
+
+import socket
+import sys
+
+
+def main():
+    if len(sys.argv) != 5:
+        raise SystemExit("usage: ami_originate.py HOST PORT ENDPOINT ACTION_ID")
+    host, port, endpoint, action_id = sys.argv[1], int(sys.argv[2]), sys.argv[3], sys.argv[4]
+
+    login = (
+        "Action: Login\r\n"
+        "Username: cp3\r\n"
+        "Secret: cp3-local-only\r\n"
+        "\r\n"
+    )
+    originate = (
+        "Action: Originate\r\n"
+        "ActionID: %s\r\n"
+        "Channel: PJSIP/%s\r\n"
+        "Context: default\r\n"
+        "Exten: s\r\n"
+        "Priority: 1\r\n"
+        "Timeout: 8000\r\n"
+        "Async: true\r\n"
+        "\r\n"
+    ) % (action_id, endpoint)
+    logoff = "Action: Logoff\r\n\r\n"
+
+    payload = (login + originate + logoff).encode("utf-8")
+    response = bytearray()
+    with socket.create_connection((host, port), timeout=4) as mgr:
+        mgr.settimeout(4)
+        mgr.sendall(payload)
+        try:
+            while b"Response: Goodbye\r\n" not in response:
+                chunk = mgr.recv(65536)
+                if not chunk:
+                    break
+                response.extend(chunk)
+        except socket.timeout:
+            pass
+
+    text = response.decode("utf-8", "replace")
+    sys.stdout.write(text)
+    # Require the Originate-specific queued message. A bare "Success" is NOT
+    # sufficient: the Login reply also carries "Success", so matching it would
+    # green-light a session whose Originate actually failed.
+    if "successfully queued" not in text:
+        sys.stderr.write("Originate not queued\n")
+        sys.exit(2)
+
+
+if __name__ == "__main__":
+    main()

tests/cp3-originate-wait/carrier_delay.py

@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""Offline carrier that DELAYS its 200 OK, for the CP3 wait-for-answer harness.
+
+Proves RECEIVER-SIDE that an AMI Originate does not run the dialplan app until
+the far end ANSWERS: the carrier holds its 200 for --answer-delay seconds and
+tags every captured datagram `phase=pre` (before it sent the 200) or
+`phase=post`. A wait-for-answer-correct rustisk sends NOTHING but the INVITE in
+the pre-answer window (no ACK/BYE/CANCEL, no RTP) and only after the 200 does the
+app run — so the ACK, the app's DTMF RTP, and the BYE are all `phase=post`.
+
+RED (app run immediately, before answer): the app runs and tears the unanswered
+leg down before the delayed 200; the 200 is never ACKed and no post-answer BYE
+appears.
+
+Captures SIP on :5060 and RTP on :40000 (the SDP-answered media port).
+"""
+
+import argparse
+import re
+import select
+import socket
+import sys
+import time
+
+SIP_PORT = 5060
+RTP_PORT = 40000
+
+
+def log(msg):
+    sys.stderr.write("[carrier_delay] " + msg + "\n")
+    sys.stderr.flush()
+
+
+def own_ip_toward(peer_ip):
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    try:
+        s.connect((peer_ip, 15060))
+        return s.getsockname()[0]
+    finally:
+        s.close()
+
+
+def get_headers(text, name):
+    out = []
+    for line in text.split("\r\n"):
+        if line == "":
+            break
+        if ":" in line:
+            hn, hv = line.split(":", 1)
+            if hn.strip().lower() == name.lower():
+                out.append(hv.strip())
+    return out
+
+
+def get_header(text, name):
+    v = get_headers(text, name)
+    return v[0] if v else None
+
+
+def cseq_of(text):
+    cs = get_header(text, "CSeq")
+    return cs.strip() if cs else None
+
+
+def build_response(req_text, code, reason, own, to_tag=None, sdp=None):
+    vias = get_headers(req_text, "Via")
+    frm = get_header(req_text, "From") or ""
+    to = get_header(req_text, "To") or ""
+    call_id = get_header(req_text, "Call-ID") or ""
+    cseq = get_header(req_text, "CSeq") or ""
+    if to_tag and "tag=" not in to:
+        to = to + ";tag=%s" % to_tag
+    lines = ["SIP/2.0 %d %s" % (code, reason)]
+    for v in vias:
+        lines.append("Via: %s" % v)
+    lines.append("From: %s" % frm)
+    lines.append("To: %s" % to)
+    lines.append("Call-ID: %s" % call_id)
+    lines.append("CSeq: %s" % cseq)
+    lines.append("Contact: <sip:carrier@%s:%d>" % (own, SIP_PORT))
+    body = sdp or ""
+    if body:
+        lines.append("Content-Type: application/sdp")
+    lines.append("Content-Length: %d" % len(body))
+    lines.append("")
+    lines.append(body)
+    return ("\r\n".join(lines)).encode("utf-8")
+
+
+def carrier_sdp(own):
+    return (
+        "v=0\r\no=carrier 0 0 IN IP4 %s\r\ns=-\r\nc=IN IP4 %s\r\n"
+        "t=0 0\r\nm=audio %d RTP/AVP 0 101\r\na=rtpmap:0 PCMU/8000\r\n"
+        "a=rtpmap:101 telephone-event/8000\r\na=fmtp:101 0-16\r\n"
+    ) % (own, own, RTP_PORT)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--caller", required=True)
+    ap.add_argument("--capture", required=True)
+    ap.add_argument("--answer-delay", type=float, default=3.0)
+    args = ap.parse_args()
+
+    own = own_ip_toward(args.caller)
+    t0 = time.time()
+
+    def cap(line):
+        with open(args.capture, "a") as f:
+            f.write(line + "\n")
+            f.flush()
+        log(line)
+
+    sip = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sip.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sip.bind(("0.0.0.0", SIP_PORT))
+    rtp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    rtp.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    rtp.bind(("0.0.0.0", RTP_PORT))
+
+    cap("READY own=%s answer_delay=%.2f rel=%.3f" % (own, args.answer_delay, 0.0))
+
+    answered = False            # global (single-call harness)
+    pending_200 = None          # (send_at, invite_text, src)
+
+    def phase():
+        return "post" if answered else "pre"
+
+    while True:
+        # Fire the delayed 200 if due.
+        if pending_200 is not None and time.time() >= pending_200[0]:
+            send_at, inv, src = pending_200
+            pending_200 = None
+            ok = build_response(inv, 200, "OK", own, to_tag="cp3del200", sdp=carrier_sdp(own))
+            sip.sendto(ok, src)
+            answered = True
+            cap("SENT-200 own=%s rel=%.3f" % (own, time.time() - t0))
+
+        timeout = 0.2
+        if pending_200 is not None:
+            timeout = max(0.01, min(0.2, pending_200[0] - time.time()))
+        r, _, _ = select.select([sip, rtp], [], [], timeout)
+        for s in r:
+            try:
+                data, src = s.recvfrom(8192)
+            except OSError:
+                continue
+            rel = time.time() - t0
+            if s is rtp:
+                cap("RTP phase=%s own=%s src=%s:%d bytes=%d rel=%.3f" % (
+                    phase(), own, src[0], src[1], len(data), rel))
+                continue
+            text = data.decode("utf-8", "replace")
+            first = text.split("\r\n", 1)[0]
+            if first.startswith("INVITE "):
+                cap("INVITE phase=%s own=%s src=%s:%d cseq=%s rel=%.3f" % (
+                    phase(), own, src[0], src[1], cseq_of(text), rel))
+                # 100 Trying immediately; 200 OK is DELAYED.
+                sip.sendto(build_response(text, 100, "Trying", own), src)
+                if pending_200 is None and not answered:
+                    pending_200 = (time.time() + args.answer_delay, text, src)
+            elif first.startswith("ACK "):
+                cap("ACK phase=%s own=%s src=%s:%d cseq=%s rel=%.3f" % (
+                    phase(), own, src[0], src[1], cseq_of(text), rel))
+            elif first.startswith("BYE "):
+                cap("BYE phase=%s own=%s src=%s:%d cseq=%s rel=%.3f" % (
+                    phase(), own, src[0], src[1], cseq_of(text), rel))
+                sip.sendto(build_response(text, 200, "OK", own), src)
+            elif first.startswith("CANCEL "):
+                cap("CANCEL phase=%s own=%s src=%s:%d cseq=%s rel=%.3f" % (
+                    phase(), own, src[0], src[1], cseq_of(text), rel))
+                sip.sendto(build_response(text, 200, "OK", own), src)
+                # RFC 3261: also 487 the INVITE and cancel the pending 200.
+                if pending_200 is not None:
+                    _, inv, isrc = pending_200
+                    pending_200 = None
+                    sip.sendto(build_response(inv, 487, "Request Terminated", own, to_tag="cp3del487"), isrc)
+
+
+if __name__ == "__main__":
+    main()

tests/cp3-originate-wait/config/asterisk.conf.tmpl

@@ -0,0 +1,4 @@
+[directories]
+astetcdir = @CONFIG_DIR@
+astrundir = @RUN_DIR@
+astincludedir = @RUN_DIR@/include

tests/cp3-originate-wait/config/extensions.conf

@@ -0,0 +1,5 @@
+[default]
+exten => s,1,Answer()
+ same => n,SendDTMF(9876)
+ same => n,Wait(1)
+ same => n,Hangup()

tests/cp3-originate-wait/config/manager.conf

@@ -0,0 +1,9 @@
+[general]
+enabled = yes
+bindaddr = 0.0.0.0
+port = 15038
+
+[cp3]
+secret = cp3-local-only
+read = all
+write = system

tests/cp3-originate-wait/config/pjsip.conf.tmpl

@@ -0,0 +1,20 @@
+; CP3 wait-for-answer harness. The carrier DELAYS its 200; the Originate must
+; not run the [default] app until that 200 arrives.
+[transport-udp]
+type = transport
+protocol = udp
+bind = 0.0.0.0:15060
+
+[carrier]
+type = endpoint
+context = default
+disallow = all
+allow = ulaw
+direct_media = no
+rtp_symmetric = yes
+dtmf_mode = rfc4733
+aors = carrier_aor
+
+[carrier_aor]
+type = aor
+contact = sip:carrier@@CORE_S@:5060

tests/cp3-originate-wait/config/rtp.conf

@@ -0,0 +1,3 @@
+[general]
+rtpstart = 31000
+rtpend = 31040