265 lines
9.3 KiB
Python
265 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Run one explicit, non-retrying SIP probe through the current ECS host."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
from .aliyun_host import AliyunCLI, CloudError, plan
|
|
except ImportError:
|
|
from aliyun_host import AliyunCLI, CloudError, plan
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_CONFIG = ROOT / ".local/agent-call-persistent/aliyun-real.json"
|
|
DEFAULT_OUTPUT_DIR = ROOT / ".local/agent-call-persistent"
|
|
ANSI = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
|
|
SIP_STATUS = re.compile(r"SIP/2\.0\s+([1-6]\d{2})(?:\s+([^\r\n]+))?")
|
|
TERMINAL = {"Finished", "Success", "Failed", "Stopped", "PartialFailed"}
|
|
|
|
|
|
def validate_number(number: str) -> str:
|
|
if not re.fullmatch(r"[0-9]{3,32}", number):
|
|
raise ValueError(
|
|
"number must contain only 3-32 digits; the original value is preserved"
|
|
)
|
|
return number
|
|
|
|
|
|
def clean(text: str) -> str:
|
|
return ANSI.sub("", text)
|
|
|
|
|
|
def parse_int(value: str) -> int | None:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def remote_command(number: str, target: str, wait_seconds: int) -> str:
|
|
pattern = (
|
|
f"{target}|INVITE sip:|ACK sip:|BYE sip:|"
|
|
r"SIP/2.0 (100|180|183|200|4[0-9][0-9]|5[0-9][0-9]|6[0-9][0-9])|"
|
|
r"Reason:|m=audio|a=rtpmap|Called|Executing.*outbound|"
|
|
r"Everyone is busy|Hangup|Spawn extension"
|
|
)
|
|
return f"""set +e
|
|
docker exec agent-call-asterisk asterisk -rx 'pjsip set logger on' >/dev/null
|
|
start=$(date -u +%Y-%m-%dT%H:%M:%SZ)
|
|
echo "CALL_START $start number={number} target={target} codec=PCMA logger=on"
|
|
docker exec agent-call-asterisk asterisk -rx 'channel originate Local/{target}@outbound application Wait {wait_seconds}'
|
|
rc=$?
|
|
echo "ORIGINATE_RC $rc"
|
|
sleep {wait_seconds + 2}
|
|
echo "CALL_END $(date -u +%Y-%m-%dT%H:%M:%SZ)"
|
|
echo '== channels =='
|
|
docker exec agent-call-asterisk asterisk -rx 'core show channels count'
|
|
echo '== filtered SIP/Asterisk evidence =='
|
|
docker logs --since "$start" agent-call-asterisk 2>&1 | grep -E '{pattern}' || true
|
|
"""
|
|
|
|
|
|
def output_text(row: dict[str, Any]) -> str:
|
|
value = row.get("Output", "")
|
|
if not isinstance(value, str):
|
|
return ""
|
|
if "CALL_START" in value:
|
|
return value
|
|
try:
|
|
decoded = base64.b64decode(value, validate=True).decode("utf-8", "replace")
|
|
except (ValueError, UnicodeDecodeError):
|
|
return value
|
|
return decoded if "CALL_START" in decoded else value
|
|
|
|
|
|
def summarize(text: str, target: str) -> dict[str, Any]:
|
|
text = clean(text)
|
|
marker = text.find(f"INVITE sip:{target}")
|
|
relevant = text[marker:] if marker >= 0 else text
|
|
responses = []
|
|
for match in SIP_STATUS.finditer(relevant):
|
|
code = parse_int(match.group(1))
|
|
if code is not None:
|
|
responses.append({"code": code, "phrase": (match.group(2) or "").strip()})
|
|
final = next((item for item in responses if item["code"] >= 200), None)
|
|
reason_match = re.search(r"Reason:\s*([^\r\n]+)", relevant, re.IGNORECASE)
|
|
origin_match = re.search(r"ORIGINATE_RC\s+(-?\d+)", text)
|
|
channels_match = re.search(r"(\d+) active channels", text)
|
|
if final is None:
|
|
result = "provisional" if responses else "no_response"
|
|
else:
|
|
result = "connected" if final["code"] < 300 else "rejected"
|
|
return {
|
|
"result": result,
|
|
"origin_rc": parse_int(origin_match.group(1)) if origin_match else None,
|
|
"responses": responses,
|
|
"final_response": final,
|
|
"reason": reason_match.group(1).strip() if reason_match else None,
|
|
"active_channels": parse_int(channels_match.group(1))
|
|
if channels_match
|
|
else None,
|
|
}
|
|
|
|
|
|
def first_invocation_result(data: dict[str, Any]) -> dict[str, Any] | None:
|
|
rows = (
|
|
data.get("Invocation", {})
|
|
.get("InvocationResults", {})
|
|
.get("InvocationResult", [])
|
|
)
|
|
if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
|
|
return None
|
|
return rows[0]
|
|
|
|
|
|
def save_evidence(path: Path, evidence: dict[str, Any]) -> None:
|
|
if path.is_symlink():
|
|
raise ValueError("output path must not be a symlink")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(
|
|
json.dumps(evidence, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
def run(args: argparse.Namespace) -> tuple[int, Path, dict[str, Any]]:
|
|
number = validate_number(args.number)
|
|
config_path = Path(args.config).expanduser()
|
|
try:
|
|
cfg = json.loads(config_path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise ValueError(f"config file is unavailable: {config_path}") from exc
|
|
api = AliyunCLI(cfg.get("profile"))
|
|
current = plan(cfg, api)
|
|
if current.get("action") != "reuse" or current.get("instance_status") != "Running":
|
|
raise CloudError(
|
|
"fixed-EIP host is not a running reusable instance; no call sent"
|
|
)
|
|
instance_id = current.get("instance_id")
|
|
if not isinstance(instance_id, str) or not instance_id:
|
|
raise CloudError("reusable instance ID is unavailable; no call sent")
|
|
target = "7089" + number
|
|
command_name = (
|
|
f"agent-call-sip-probe-{number}-{time.strftime('%Y%m%d%H%M%S', time.gmtime())}"
|
|
)
|
|
run_result = api(
|
|
"ecs",
|
|
"RunCommand",
|
|
**{
|
|
"InstanceId.1": instance_id,
|
|
"Type": "RunShellScript",
|
|
"Name": command_name,
|
|
"CommandContent": remote_command(number, target, args.wait_seconds),
|
|
"ContentEncoding": "PlainText",
|
|
"Timeout": max(30, args.wait_seconds + 12),
|
|
"RepeatMode": "Once",
|
|
"KeepCommand": "false",
|
|
},
|
|
)
|
|
command_id = run_result.get("CommandId")
|
|
invoke_id = run_result.get("InvokeId")
|
|
if not isinstance(command_id, str) or not isinstance(invoke_id, str):
|
|
raise CloudError(
|
|
"RunCommand returned no command/invocation ID; inspect cloud state before retry"
|
|
)
|
|
|
|
result: dict[str, Any] | None = None
|
|
row: dict[str, Any] | None = None
|
|
deadline = time.monotonic() + args.timeout_seconds
|
|
while time.monotonic() < deadline:
|
|
result = api(
|
|
"ecs",
|
|
"DescribeInvocationResults",
|
|
InvokeId=invoke_id,
|
|
ContentEncoding="PlainText",
|
|
MaxResults=10,
|
|
)
|
|
candidate = first_invocation_result(result)
|
|
if candidate is not None:
|
|
row = candidate
|
|
status = candidate.get("InvokeRecordStatus") or candidate.get(
|
|
"InvocationStatus"
|
|
)
|
|
if status in TERMINAL:
|
|
break
|
|
time.sleep(args.poll_seconds)
|
|
if row is None:
|
|
raise CloudError(
|
|
f"invocation {invoke_id} did not finish before timeout; do not redial automatically"
|
|
)
|
|
status = row.get("InvokeRecordStatus") or row.get("InvocationStatus")
|
|
if status not in TERMINAL:
|
|
raise CloudError(
|
|
f"invocation {invoke_id} did not finish before timeout; do not redial automatically"
|
|
)
|
|
|
|
raw_output = output_text(row)
|
|
probe = summarize(raw_output, target)
|
|
stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
|
|
output_path = (
|
|
Path(args.output)
|
|
if args.output
|
|
else DEFAULT_OUTPUT_DIR / f"sip-probe-{number}-{stamp}.json"
|
|
)
|
|
evidence = {
|
|
"instance_id": instance_id,
|
|
"public_ip": current.get("public_ip"),
|
|
"number": number,
|
|
"target": target,
|
|
"dial_prefix": "7089",
|
|
"codec": "PCMA",
|
|
"logger": "on",
|
|
"command_id": command_id,
|
|
"invoke_id": invoke_id,
|
|
"invocation_status": row.get("InvocationStatus"),
|
|
"invoke_record_status": row.get("InvokeRecordStatus"),
|
|
"exit_code": row.get("ExitCode"),
|
|
"probe": probe,
|
|
"raw_output": clean(raw_output),
|
|
}
|
|
save_evidence(output_path, evidence)
|
|
summary = {key: value for key, value in evidence.items() if key != "raw_output"}
|
|
summary["evidence_file"] = str(output_path)
|
|
return (
|
|
0 if row.get("ExitCode") == 0 and probe["result"] != "no_response" else 1,
|
|
output_path,
|
|
summary,
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--number", required=True, help="original callee number; one call, no retry"
|
|
)
|
|
parser.add_argument("--config", default=str(DEFAULT_CONFIG))
|
|
parser.add_argument("--output", help="0600 evidence JSON path")
|
|
parser.add_argument("--wait-seconds", type=int, default=5)
|
|
parser.add_argument("--poll-seconds", type=float, default=1.0)
|
|
parser.add_argument("--timeout-seconds", type=int, default=60)
|
|
args = parser.parse_args()
|
|
if not 3 <= args.wait_seconds <= 30:
|
|
parser.error("--wait-seconds must be 3-30")
|
|
if args.poll_seconds <= 0 or args.timeout_seconds <= 0:
|
|
parser.error("poll and timeout must be positive")
|
|
try:
|
|
status, _, summary = run(args)
|
|
except (CloudError, OSError, ValueError, json.JSONDecodeError) as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
return status
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|