feat(ops): M7 CP4 — migrate bin/call.sh to an AMI operational command (M-k)

bin/ami_call.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+"""Low-level AMI transport for bin/call.sh (the migrated operational command).
+
+Subcommands (all read the AMI secret from a FILE, never argv/env value):
+
+  list      Print active channels, one per line:
+              CHANNEL\tEXTENSION\tCALLERIDNUM\tSTATE
+  originate Place one outbound call (Async). Exits 0 only if AMI queued it.
+  hangup    Hang up ONE named channel.
+
+The AMI secret path (a mounted k8s Secret in production) is passed with
+--secret-file; the value is read here and never printed. This replaces the ESL
+password handling in the FreeSWITCH-era bin/call.sh.
+"""
+
+import argparse
+import socket
+import sys
+import uuid
+
+
+def read_secret(path):
+    with open(path, "r") as f:
+        return f.read().strip()
+
+
+def ami_txn(host, port, username, secret, action_lines, terminator):
+    """Login, send one action, read until `terminator` appears, logoff. Returns text."""
+    login = "Action: Login\r\nUsername: %s\r\nSecret: %s\r\n\r\n" % (username, secret)
+    action = "".join("%s\r\n" % l for l in action_lines) + "\r\n"
+    logoff = "Action: Logoff\r\n\r\n"
+    buf = bytearray()
+    with socket.create_connection((host, port), timeout=6) as s:
+        s.settimeout(6)
+        s.sendall((login + action + logoff).encode("utf-8"))
+        try:
+            while terminator.encode("utf-8") not in buf and b"Response: Goodbye\r\n" not in buf:
+                chunk = s.recv(65536)
+                if not chunk:
+                    break
+                buf.extend(chunk)
+        except socket.timeout:
+            pass
+    return buf.decode("utf-8", "replace")
+
+
+def parse_events(text):
+    """Split an AMI stream into a list of dict blocks (keyed by header name)."""
+    blocks = []
+    for raw in text.split("\r\n\r\n"):
+        raw = raw.strip("\r\n")
+        if not raw:
+            continue
+        d = {}
+        for line in raw.split("\r\n"):
+            if ":" in line:
+                k, v = line.split(":", 1)
+                d[k.strip()] = v.strip()
+        if d:
+            blocks.append(d)
+    return blocks
+
+
+def cmd_list(args):
+    secret = read_secret(args.secret_file)
+    aid = uuid.uuid4().hex[:8]
+    text = ami_txn(args.host, args.port, args.username, secret,
+                   ["Action: CoreShowChannels", "ActionID: %s" % aid],
+                   "CoreShowChannelsComplete")
+    for b in parse_events(text):
+        if b.get("Event") == "CoreShowChannel":
+            sys.stdout.write("%s\t%s\t%s\t%s\n" % (
+                b.get("Channel", ""),
+                b.get("Extension", ""),
+                b.get("CallerIDNum", ""),
+                b.get("ChannelStateDesc", ""),
+            ))
+    return 0
+
+
+def cmd_originate(args):
+    secret = read_secret(args.secret_file)
+    aid = uuid.uuid4().hex[:8]
+    lines = [
+        "Action: Originate",
+        "ActionID: %s" % aid,
+        "Channel: %s" % args.channel,
+        "Context: %s" % args.context,
+        "Exten: %s" % args.exten,
+        "Priority: %d" % args.priority,
+        "Timeout: %d" % args.timeout,
+        "Async: true",
+    ]
+    if args.callerid:
+        lines.append("CallerID: %s" % args.callerid)
+    text = ami_txn(args.host, args.port, args.username, secret, lines, "successfully queued")
+    sys.stdout.write(text)
+    # The Login reply also carries "Success"; require the Originate-specific
+    # queued message so a failed Originate cannot false-pass.
+    return 0 if "successfully queued" in text else 2
+
+
+def cmd_hangup(args):
+    secret = read_secret(args.secret_file)
+    aid = uuid.uuid4().hex[:8]
+    text = ami_txn(args.host, args.port, args.username, secret,
+                   ["Action: Hangup", "ActionID: %s" % aid, "Channel: %s" % args.channel],
+                   "Response:")
+    sys.stdout.write(text)
+    return 0
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--host", default="127.0.0.1")
+    ap.add_argument("--port", type=int, default=5038)
+    ap.add_argument("--username", default="operator")
+    ap.add_argument("--secret-file", required=True)
+    sub = ap.add_subparsers(dest="cmd", required=True)
+
+    sub.add_parser("list")
+
+    o = sub.add_parser("originate")
+    o.add_argument("--channel", required=True)
+    o.add_argument("--context", default="default")
+    o.add_argument("--exten", default="s")
+    o.add_argument("--priority", type=int, default=1)
+    o.add_argument("--callerid", default="")
+    o.add_argument("--timeout", type=int, default=30000)
+
+    h = sub.add_parser("hangup")
+    h.add_argument("--channel", required=True)
+
+    args = ap.parse_args()
+    if args.cmd == "list":
+        sys.exit(cmd_list(args))
+    if args.cmd == "originate":
+        sys.exit(cmd_originate(args))
+    if args.cmd == "hangup":
+        sys.exit(cmd_hangup(args))
+
+
+if __name__ == "__main__":
+    main()

bin/call.sh

@@ -0,0 +1,126 @@
+#!/usr/bin/env bash
+# Place an OUTBOUND call through rustisk and drop the callee into the agents'
+# bridge — the AMI-native successor to the FreeSWITCH/ESL bin/call.sh (M-k).
+#
+#   bin/call.sh 2000                     # ring dest 2000 through the carrier
+#   bin/call.sh 2000 --dry-run           # show the Originate, place NO call
+#   bin/call.sh --hangup 2000            # hang up the call(s) to dest 2000
+#   bin/call.sh --hangup-all --force     # GUARDED hupall: tear down EVERY channel
+#
+# vs the ESL original this is AMI-native (Action: Originate / CoreShowChannels /
+# Hangup), reads its secret from a mounted k8s Secret file (never argv), refuses
+# to stack a duplicate concurrent call to the same destination, and turns the old
+# unconditional `hupall` foot-gun into an EXPLICIT, DOUBLY-GUARDED operation
+# (--hangup-all AND --force both required).
+#
+# Config via env (all optional):
+#   AMI_HOST (127.0.0.1)  AMI_PORT (5038)  AMI_USERNAME (operator)
+#   AMI_SECRET_FILE (/run/secrets/rustisk-ami/secret)  — the mounted k8s Secret
+#   CALL_ENDPOINT (carrier)  CALL_CONTEXT (default)  CALL_CALLERID ("")
+#   CALL_TIMEOUT_MS (30000)
+set -euo pipefail
+
+HERE="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
+AMI_PY="$HERE/ami_call.py"
+
+AMI_HOST="${AMI_HOST:-127.0.0.1}"
+AMI_PORT="${AMI_PORT:-5038}"
+AMI_USERNAME="${AMI_USERNAME:-operator}"
+AMI_SECRET_FILE="${AMI_SECRET_FILE:-/run/secrets/rustisk-ami/secret}"
+CALL_ENDPOINT="${CALL_ENDPOINT:-carrier}"
+CALL_CONTEXT="${CALL_CONTEXT:-default}"
+CALL_CALLERID="${CALL_CALLERID:-}"
+CALL_TIMEOUT_MS="${CALL_TIMEOUT_MS:-30000}"
+
+DEST=""
+DRY=0
+HANGUP=0
+HANGUP_ALL=0
+FORCE=0
+for a in "$@"; do
+  case "$a" in
+    --dry-run)    DRY=1 ;;
+    --hangup)     HANGUP=1 ;;
+    --hangup-all) HANGUP_ALL=1 ;;
+    --force)      FORCE=1 ;;
+    --*)          echo "usage: $0 DEST [--dry-run] | --hangup DEST | --hangup-all --force" >&2; exit 2 ;;
+    *)            DEST="$a" ;;
+  esac
+done
+
+die() { echo "$*" >&2; exit 1; }
+
+ami() { python3 "$AMI_PY" --host "$AMI_HOST" --port "$AMI_PORT" \
+                          --username "$AMI_USERNAME" --secret-file "$AMI_SECRET_FILE" "$@"; }
+
+[ -r "$AMI_SECRET_FILE" ] || die "AMI secret file not readable: $AMI_SECRET_FILE (mount the k8s Secret)"
+
+channel_prefix="PJSIP/${CALL_ENDPOINT}-"
+
+# --- --hangup-all: the GUARDED hupall (both --hangup-all AND --force) ---------
+if [ "$HANGUP_ALL" = 1 ]; then
+  if [ "$FORCE" != 1 ]; then
+    echo "REFUSING to hang up EVERY channel." >&2
+    echo "This is the old hupall foot-gun. Re-run with an explicit confirmation:" >&2
+    echo "  $0 --hangup-all --force" >&2
+    exit 3
+  fi
+  n=0
+  while IFS=$'\t' read -r chan _exten _cid _state; do
+    [ -n "$chan" ] || continue
+    ami hangup --channel "$chan" >/dev/null || true
+    echo "hung up $chan"
+    n=$((n + 1))
+  done < <(ami list)
+  echo "hupall complete: $n channel(s) hung up"
+  exit 0
+fi
+
+# --- --hangup DEST: tear down only the call(s) to DEST ------------------------
+if [ "$HANGUP" = 1 ]; then
+  [ -n "$DEST" ] || die "usage: $0 --hangup DEST   (to hupall EVERY channel use --hangup-all --force)"
+  n=0
+  while IFS=$'\t' read -r chan exten _cid _state; do
+    [ -n "$chan" ] || continue
+    if [ "$exten" = "$DEST" ] && [[ "$chan" == "$channel_prefix"* ]]; then
+      ami hangup --channel "$chan" >/dev/null || true
+      echo "hung up $chan (dest $DEST)"
+      n=$((n + 1))
+    fi
+  done < <(ami list)
+  if [ "$n" = 0 ]; then echo "no active call to $DEST"; fi
+  exit 0
+fi
+
+# --- place a call ------------------------------------------------------------
+[ -n "$DEST" ] || die "usage: $0 DEST [--dry-run] | --hangup DEST | --hangup-all --force"
+
+CHANNEL="PJSIP/${CALL_ENDPOINT}"
+
+# Duplicate-call guard: refuse to stack a second concurrent call to the same
+# destination (matched on the active channel's Extension + endpoint prefix).
+if ami list | awk -F'\t' -v d="$DEST" -v p="$channel_prefix" \
+      '$2==d && index($1,p)==1 {found=1} END{exit found?0:1}'; then
+  echo "a call to $DEST is already up -- refusing to place another"
+  echo "(use '$0 --hangup $DEST' to clear it)"
+  exit 0
+fi
+
+if [ "$DRY" = 1 ]; then
+  echo "DRY RUN — would AMI Originate:"
+  echo "  Channel:  $CHANNEL"
+  echo "  Context:  $CALL_CONTEXT"
+  echo "  Exten:    $DEST"
+  echo "  CallerID: ${CALL_CALLERID:-<default>}"
+  echo "  (no call placed)"
+  exit 0
+fi
+
+echo ">> ringing $DEST via $CHANNEL (context $CALL_CONTEXT)"
+if ami originate --channel "$CHANNEL" --context "$CALL_CONTEXT" --exten "$DEST" \
+                 --priority 1 --callerid "$CALL_CALLERID" --timeout "$CALL_TIMEOUT_MS" >/dev/null; then
+  echo ">> Originate queued for $DEST"
+  echo ">> hang up with: $0 --hangup $DEST"
+else
+  die ">> Originate failed"
+fi

tests/cp4-call-command/carrier.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""Minimal offline carrier for the CP2 from-identity harness (stdlib only).
+
+Captures the outbound INVITE's From header RECEIVER-SIDE and answers the call
+(100 -> 200 with SDP -> consumes ACK/BYE) so the leg completes cleanly. The
+harness asserts the captured From carries the endpoint's configured
+from_user@from_domain — never a rustisk TX log.
+"""
+
+import argparse
+import re
+import socket
+import sys
+
+
+SIP_PORT = 5060
+
+
+def log(msg):
+    sys.stderr.write("[carrier_from] " + 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 append_capture(path, line):
+    with open(path, "a") as f:
+        f.write(line + "\n")
+        f.flush()
+    log(line)
+
+
+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 40000 RTP/AVP 0\r\na=rtpmap:0 PCMU/8000\r\n"
+    ) % (own, own)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--caller", required=True)
+    ap.add_argument("--capture", required=True)
+    args = ap.parse_args()
+
+    own = own_ip_toward(args.caller)
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind(("0.0.0.0", SIP_PORT))
+    append_capture(args.capture, "READY own=%s" % own)
+
+    while True:
+        try:
+            sock.settimeout(0.5)
+            data, src = sock.recvfrom(8192)
+        except socket.timeout:
+            continue
+        except OSError:
+            break
+        text = data.decode("utf-8", "replace")
+        first = text.split("\r\n", 1)[0]
+        if first.startswith("INVITE "):
+            frm = get_header(text, "From") or "?"
+            # Extract the bare URI (sip:user@domain) from the From value.
+            m = re.search(r'<([^>]+)>', frm)
+            uri = m.group(1) if m else frm
+            append_capture(args.capture, "INVITE-FROM own=%s from_uri=%s raw_from=%s" % (own, uri, frm))
+            sock.sendto(build_response(text, 100, "Trying", own), src)
+            sock.sendto(build_response(text, 200, "OK", own, to_tag="cp2from200", sdp=carrier_sdp(own)), src)
+        elif first.startswith("BYE "):
+            append_capture(args.capture, "BYE own=%s" % own)
+            sock.sendto(build_response(text, 200, "OK", own), src)
+        # ACK: nothing to answer.
+
+
+if __name__ == "__main__":
+    main()