diff --git a/.gitignore b/.gitignore index 5b90e79..f8b0ec6 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,6 @@ go.work.sum # env file .env - +logs/ +recordings/ +*.wav diff --git a/README.md b/README.md index 0c19b97..5845b68 100644 --- a/README.md +++ b/README.md @@ -4,3 +4,35 @@ - [AI 外呼 SaaS 调研与架构指南](docs/research/ai-outbound-saas.md) - [公网模型外呼目标架构](docs/architecture/public-model-outbound.md) + +## SIP/RTP 单次呼出 MVP + +cmd/sip-demo 通过 UDP 向 61.132.228.221:5060 发起一通 IP 白名单、无鉴权呼叫: + +- 主叫固定为企业 code BD93205882,被叫按 7089 + SIP_TEST_NUMBER 生成; +- 协商 G.711 A-law/μ-law、20 ms RTP 帧,播放百练合成披露话术和 WAV/内置测试音; +- 上行连续三帧达到 VAD 阈值时立即停止下行播放,用于验证最小打断链路; +- 上下行分别录为 recordings/*-rx.wav、recordings/*-tx.wav; +- SIP 请求、响应和通话事件写入 logs/sip.log,号码在写入前脱敏。 + +真实号码、API Key、运行日志和录音均被 .gitignore 排除。仓库只保留[脱敏日志样本](docs/examples/sip-signaling.redacted.log)。 + +### 离线验证 + + go test ./... + go build ./cmd/sip-demo + +测试使用本地 HTTP/RTP 端点,不调用百练或 SIP 服务器。 + +### 经授权的内部联调 + +确认执行机出口 IP 已加入白名单、UDP SIP/RTP 可达,再仅通过环境提供号码和凭据: + + export SIP_TEST_NUMBER='<已授权内部测试号码>' + export BAILIAN_BASE_URL='https:///compatible-mode/v1' + export BAILIAN_API_KEY='' + go run ./cmd/sip-demo + +BAILIAN_BASE_URL 应指向支持 POST /audio/speech、PCM16 输出的百练兼容端点;若它已以 /audio/speech 结尾则直接使用。除 loopback 测试外只接受 HTTPS,并拒绝跨 host 或 HTTPS 降级重定向。Key 只进入 Bearer 请求头,不写日志。可用 -audio music.wav 追加单声道 PCM16 WAV;未指定时追加两秒测试音。NAT 环境通过 SIP_ADVERTISE_IP 或 -advertise-ip 指定 SDP/Via 公网 IP。 + +本 MVP 不做注册、鉴权、重拨、ASR/LLM、AEC 或抖动缓冲。打断依赖线路提供独立上行;若远端回声触发误打断,用 -vad-threshold 校准,正式方案仍需 AEC/线路能力验证。 diff --git a/cmd/sip-demo/main.go b/cmd/sip-demo/main.go new file mode 100644 index 0000000..66cf91b --- /dev/null +++ b/cmd/sip-demo/main.go @@ -0,0 +1,669 @@ +package main + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/binary" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "math" + "net" + "net/http" + "net/url" + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "sync" + "time" + + "github.com/emiago/sipgo" + "github.com/emiago/sipgo/sip" + "github.com/zaf/g711" +) + +const ( + defaultSIPServer = "61.132.228.221:5060" + caller = "BD93205882" + calleePrefix = "7089" + defaultTTS = "您好,这里是北京美珈科技有限公司的人工智能语音测试,本次通话将被录音。" +) + +type config struct { + server, number, audio, ttsText, ttsVoice, logPath, recordDir, localAddr, advertiseIP string + ringTimeout, maxDuration time.Duration + vadThreshold int +} + +func main() { + cfg := config{number: os.Getenv("SIP_TEST_NUMBER")} + flag.StringVar(&cfg.server, "sip-server", envOr("SIP_SERVER", defaultSIPServer), "SIP host:port") + flag.StringVar(&cfg.audio, "audio", "", "optional mono PCM16 WAV to play after TTS") + flag.StringVar(&cfg.ttsText, "tts-text", defaultTTS, "text synthesized before the call") + flag.StringVar(&cfg.ttsVoice, "tts-voice", "longxiaochun", "Bailian voice ID") + flag.StringVar(&cfg.logPath, "sip-log", "logs/sip.log", "redacted SIP log path") + flag.StringVar(&cfg.recordDir, "record-dir", "recordings", "untracked RX/TX WAV directory") + flag.StringVar(&cfg.localAddr, "sip-local", envOr("SIP_LOCAL_ADDR", "0.0.0.0:0"), "local SIP UDP address") + flag.StringVar(&cfg.advertiseIP, "advertise-ip", os.Getenv("SIP_ADVERTISE_IP"), "SDP/Via IP; auto-detected when empty") + flag.DurationVar(&cfg.ringTimeout, "ring-timeout", 45*time.Second, "maximum ringing time") + flag.DurationVar(&cfg.maxDuration, "max-duration", 60*time.Second, "maximum answered call duration") + flag.IntVar(&cfg.vadThreshold, "vad-threshold", 1200, "inbound PCM RMS needed to interrupt playback") + flag.Parse() + + if err := run(context.Background(), cfg); err != nil { + fmt.Fprintln(os.Stderr, "sip-demo:", err) + os.Exit(1) + } +} + +func run(ctx context.Context, cfg config) error { + return runWithACK(ctx, cfg, func(ctx context.Context, session *sipgo.DialogClientSession) error { + return session.Ack(ctx) + }) +} + +func runWithACK(ctx context.Context, cfg config, ack func(context.Context, *sipgo.DialogClientSession) error) error { + if err := validateNumber(cfg.number); err != nil { + return fmt.Errorf("SIP_TEST_NUMBER: %w", err) + } + if cfg.maxDuration <= 0 || cfg.ringTimeout <= 0 || cfg.vadThreshold <= 0 { + return errors.New("timeouts and vad-threshold must be positive") + } + host, portText, err := net.SplitHostPort(cfg.server) + if err != nil { + return fmt.Errorf("invalid SIP server: %w", err) + } + port, err := strconv.Atoi(portText) + if err != nil || port < 1 || port > 65535 { + return errors.New("invalid SIP server port") + } + + program, err := mediaProgram(ctx, cfg) + if err != nil { + return err + } + rtpConn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0}) + if err != nil { + return fmt.Errorf("listen RTP: %w", err) + } + defer rtpConn.Close() + advertiseIP := cfg.advertiseIP + if advertiseIP == "" { + advertiseIP, err = routeIP(cfg.server) + if err != nil { + return err + } + } + if net.ParseIP(advertiseIP) == nil { + return errors.New("advertise-ip must be an IP address") + } + + log, err := newSignalLog(cfg.logPath, cfg.number, caller) + if err != nil { + return err + } + defer log.Close() + sdp := offerSDP(advertiseIP, rtpConn.LocalAddr().(*net.UDPAddr).Port) + + ua, err := sipgo.NewUA(sipgo.WithUserAgent(caller), sipgo.WithUserAgentHostname(host)) + if err != nil { + return fmt.Errorf("create SIP UA: %w", err) + } + defer ua.Close() + client, err := sipgo.NewClient(ua, + sipgo.WithClientConnectionAddr(cfg.localAddr), + sipgo.WithClientHostname(advertiseIP), + sipgo.WithClientNAT(), + ) + if err != nil { + return fmt.Errorf("create SIP client: %w", err) + } + server, err := sipgo.NewServer(ua) + if err != nil { + return fmt.Errorf("create SIP server: %w", err) + } + dialogs := sipgo.NewDialogClientCache(client, sip.ContactHeader{Address: sip.Uri{User: caller}}) + remoteHangup := make(chan struct{}) + var hangupOnce sync.Once + server.OnBye(func(req *sip.Request, tx sip.ServerTransaction) { + log.message("<<<", req.String()) + if err := dialogs.ReadBye(req, tx); err != nil { + log.event("failed to process remote BYE: " + err.Error()) + _ = tx.Respond(sip.NewResponseFromRequest(req, sip.StatusCallTransactionDoesNotExists, "Call/Transaction Does Not Exist", nil)) + } + hangupOnce.Do(func() { close(remoteHangup) }) + }) + + callee := calleePrefix + cfg.number + recipient := sip.Uri{Scheme: "sip", User: callee, Host: host, Port: port} + ringCtx, cancelRing := context.WithTimeout(ctx, cfg.ringTimeout) + defer cancelRing() + session, err := dialogs.Invite(ringCtx, recipient, sdp, sip.NewHeader("Content-Type", "application/sdp")) + if err != nil { + return fmt.Errorf("send INVITE: %w", err) + } + defer session.Close() + log.message(">>>", session.InviteRequest.String()) + + err = session.WaitAnswer(ringCtx, sipgo.AnswerOptions{OnResponse: func(res *sip.Response) error { + log.message("<<<", res.String()) + return nil + }}) + if err != nil { + var responseErr *sipgo.ErrDialogResponse + if errors.As(err, &responseErr) && (responseErr.Res.StatusCode == sip.StatusUnauthorized || responseErr.Res.StatusCode == sip.StatusProxyAuthRequired) { + return errors.New("trunk requested authentication although this MVP is configured for IP whitelist only") + } + return fmt.Errorf("wait for answer: %w", err) + } + acked := false + var callErr error + if err := ack(ctx, session); err != nil { + callErr = fmt.Errorf("send ACK: %w", err) + } else { + acked = true + log.event(fmt.Sprintf(">>> ACK call-id=%s cseq=%d", session.InviteRequest.CallID().Value(), session.CSEQ())) + remoteRTP, payloadType, err := answerMedia(session.InviteResponse.Body()) + if err != nil { + callErr = err + } else { + callErr = runMedia(ctx, rtpConn, remoteRTP, payloadType, program, cfg.maxDuration, cfg.vadThreshold, remoteHangup, cfg.recordDir, log) + } + } + if session.LoadState() != sip.DialogStateEnded { + callErr = errors.Join(callErr, endAnsweredSession(session, log, acked)) + } + return callErr +} + +func endAnsweredSession(session *sipgo.DialogClientSession, log *signalLog, acked bool) error { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + log.event(fmt.Sprintf(">>> BYE call-id=%s cseq=%d", session.InviteRequest.CallID().Value(), session.CSEQ()+1)) + if acked { + if err := session.Bye(ctx); err != nil { + return fmt.Errorf("send BYE: %w", err) + } + return nil + } + + recipient := session.InviteRequest.Recipient + if contact := session.InviteResponse.Contact(); contact != nil { + recipient = contact.Address + } + response, err := session.Do(ctx, sip.NewRequest(sip.BYE, recipient)) + if err != nil { + return fmt.Errorf("send BYE after ACK failure: %w", err) + } + if response.StatusCode != sip.StatusOK { + return fmt.Errorf("send BYE after ACK failure: %w", &sipgo.ErrDialogResponse{Res: response}) + } + return nil +} + +func mediaProgram(ctx context.Context, cfg config) ([]int16, error) { + var out []int16 + if cfg.ttsText != "" { + base, key := os.Getenv("BAILIAN_BASE_URL"), os.Getenv("BAILIAN_API_KEY") + if base == "" || key == "" { + return nil, errors.New("BAILIAN_BASE_URL and BAILIAN_API_KEY are required when tts-text is set") + } + ttsCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + pcm, rate, err := synthesize(ttsCtx, http.DefaultClient, base, key, cfg.ttsVoice, cfg.ttsText) + cancel() + if err != nil { + return nil, fmt.Errorf("Bailian TTS: %w", err) + } + out = append(out, resample(pcm, rate, 8000)...) + } + if cfg.audio != "" { + f, err := os.Open(cfg.audio) + if err != nil { + return nil, fmt.Errorf("open audio: %w", err) + } + pcm, rate, err := readWAV(io.LimitReader(f, 64<<20)) + f.Close() + if err != nil { + return nil, fmt.Errorf("read audio: %w", err) + } + out = append(out, resample(pcm, rate, 8000)...) + } else { + out = append(out, tone(440, 2*time.Second)...) + } + return out, nil +} + +func synthesize(ctx context.Context, client *http.Client, base, key, voice, text string) ([]int16, int, error) { + u, err := url.Parse(base) + if err != nil || u.Host == "" || u.User != nil || (u.Scheme != "https" && !(u.Scheme == "http" && isLoopback(u.Hostname()))) { + return nil, 0, errors.New("invalid BAILIAN_BASE_URL") + } + if !strings.HasSuffix(strings.TrimRight(u.Path, "/"), "/audio/speech") { + u.Path = strings.TrimRight(u.Path, "/") + "/audio/speech" + } + body, _ := json.Marshal(map[string]any{ + "model": "cosyvoice-v3.5-plus", "voice": voice, "input": text, + "response_format": "pcm", "sample_rate": 24000, + }) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), bytes.NewReader(body)) + if err != nil { + return nil, 0, err + } + req.Header.Set("Authorization", "Bearer "+key) + req.Header.Set("Content-Type", "application/json") + safeClient := *client + previousRedirect := safeClient.CheckRedirect + safeClient.CheckRedirect = func(req *http.Request, via []*http.Request) error { + if req.URL.User != nil || req.URL.Host != u.Host || (u.Scheme == "https" && req.URL.Scheme != "https") || + (req.URL.Scheme != "https" && !(req.URL.Scheme == "http" && isLoopback(req.URL.Hostname()))) { + return errors.New("unsafe TTS redirect blocked") + } + if len(via) >= 10 { + return errors.New("stopped after 10 redirects") + } + if previousRedirect != nil { + return previousRedirect(req, via) + } + return nil + } + res, err := safeClient.Do(req) + if err != nil { + return nil, 0, err + } + defer res.Body.Close() + if res.StatusCode < 200 || res.StatusCode > 299 { + return nil, 0, fmt.Errorf("HTTP %d", res.StatusCode) + } + data, err := io.ReadAll(io.LimitReader(res.Body, 32<<20+1)) + if err != nil { + return nil, 0, err + } + if len(data) > 32<<20 { + return nil, 0, errors.New("audio response exceeds 32 MiB") + } + if bytes.HasPrefix(data, []byte("RIFF")) { + return readWAV(bytes.NewReader(data)) + } + if len(data)%2 != 0 { + return nil, 0, errors.New("TTS response is not PCM16") + } + return bytesToPCM(data), 24000, nil +} + +func isLoopback(host string) bool { + ip := net.ParseIP(host) + return strings.EqualFold(host, "localhost") || ip != nil && ip.IsLoopback() +} + +func runMedia(parent context.Context, conn *net.UDPConn, remote *net.UDPAddr, payloadType uint8, audio []int16, duration time.Duration, threshold int, remoteHangup <-chan struct{}, recordDir string, log *signalLog) error { + ctx, cancel := context.WithTimeout(parent, duration) + defer cancel() + interrupted := make(chan struct{}) + received := make(chan []int16, 1) + go receiveRTP(ctx, conn, remote, threshold, int(duration.Seconds()*8000), interrupted, received) + + var sent []int16 + seq := randomUint16() + timestamp, ssrc := randomUint32(), randomUint32() + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + playing := true + for offset := 0; offset < len(audio) && playing; offset += 160 { + end := min(offset+160, len(audio)) + frame := make([]int16, 160) + copy(frame, audio[offset:end]) + payload := encodeG711(pcmToBytes(frame), payloadType) + packet := marshalRTP(payloadType, seq, timestamp, ssrc, payload) + if _, err := conn.WriteToUDP(packet, remote); err != nil { + cancel() + <-received + return fmt.Errorf("send RTP: %w", err) + } + sent = append(sent, frame...) + seq++ + timestamp += 160 + select { + case <-ticker.C: + case <-interrupted: + log.event("playback interrupted by inbound voice") + playing = false + case <-remoteHangup: + playing = false + cancel() + case <-ctx.Done(): + playing = false + } + } + + select { + case <-remoteHangup: + case <-ctx.Done(): + } + cancel() + rx := <-received + if err := os.MkdirAll(recordDir, 0o700); err != nil { + return fmt.Errorf("create recording directory: %w", err) + } + stamp := time.Now().UTC().Format("20060102T150405Z") + if err := writeWAV(filepath.Join(recordDir, "call-"+stamp+"-rx.wav"), rx, 8000); err != nil { + return err + } + if err := writeWAV(filepath.Join(recordDir, "call-"+stamp+"-tx.wav"), sent, 8000); err != nil { + return err + } + log.event(fmt.Sprintf("media ended rx_samples=%d tx_samples=%d", len(rx), len(sent))) + return nil +} + +func receiveRTP(ctx context.Context, conn *net.UDPConn, remote *net.UDPAddr, threshold, maxSamples int, interrupted chan<- struct{}, done chan<- []int16) { + var pcm []int16 + voiceFrames := 0 + interruptedSent := false + buf := make([]byte, 2048) + for { + _ = conn.SetReadDeadline(time.Now().Add(50 * time.Millisecond)) + n, source, err := conn.ReadFromUDP(buf) + if err != nil { + if errors.Is(err, os.ErrDeadlineExceeded) { + select { + case <-ctx.Done(): + done <- pcm + return + default: + continue + } + } + done <- pcm + return + } + if !source.IP.Equal(remote.IP) || source.Port != remote.Port { + continue + } + payload, pt, ok := rtpPayload(buf[:n]) + if !ok || (pt != 8 && pt != 0) { + continue + } + frame := bytesToPCM(decodeG711(payload, pt)) + if remaining := maxSamples - len(pcm); remaining > 0 { + pcm = append(pcm, frame[:min(len(frame), remaining)]...) + } + if rms(frame) >= threshold { + voiceFrames++ + } else { + voiceFrames = 0 + } + if voiceFrames >= 3 && !interruptedSent { + close(interrupted) + interruptedSent = true + } + } +} + +func offerSDP(ip string, port int) []byte { + return []byte(fmt.Sprintf("v=0\r\no=- 0 0 IN IP4 %s\r\ns=magic-sonar\r\nc=IN IP4 %s\r\nt=0 0\r\nm=audio %d RTP/AVP 8 0\r\na=rtpmap:8 PCMA/8000\r\na=rtpmap:0 PCMU/8000\r\na=ptime:20\r\na=sendrecv\r\n", ip, ip, port)) +} + +func answerMedia(body []byte) (*net.UDPAddr, uint8, error) { + var host string + port := 0 + var payload uint8 = 255 + for _, raw := range strings.Split(strings.ReplaceAll(string(body), "\r\n", "\n"), "\n") { + fields := strings.Fields(raw) + if len(fields) == 3 && fields[0] == "c=IN" && fields[1] == "IP4" { + host = fields[2] + } + if len(fields) >= 4 && fields[0] == "m=audio" { + port, _ = strconv.Atoi(fields[1]) + for _, offered := range fields[3:] { + if offered == "8" { + payload = 8 + break + } + if offered == "0" { + payload = 0 + } + } + } + } + if net.ParseIP(host) == nil || port < 1 || port > 65535 || payload == 255 { + return nil, 0, errors.New("answer SDP lacks usable IPv4 G.711 media") + } + return &net.UDPAddr{IP: net.ParseIP(host), Port: port}, payload, nil +} + +func marshalRTP(pt uint8, seq uint16, timestamp, ssrc uint32, payload []byte) []byte { + p := make([]byte, 12+len(payload)) + p[0], p[1] = 0x80, pt&0x7f + binary.BigEndian.PutUint16(p[2:4], seq) + binary.BigEndian.PutUint32(p[4:8], timestamp) + binary.BigEndian.PutUint32(p[8:12], ssrc) + copy(p[12:], payload) + return p +} + +func rtpPayload(packet []byte) ([]byte, uint8, bool) { + if len(packet) < 12 || packet[0]>>6 != 2 { + return nil, 0, false + } + cc := int(packet[0] & 0x0f) + offset := 12 + cc*4 + if len(packet) < offset { + return nil, 0, false + } + if packet[0]&0x10 != 0 { + if len(packet) < offset+4 { + return nil, 0, false + } + offset += 4 + int(binary.BigEndian.Uint16(packet[offset+2:offset+4]))*4 + } + end := len(packet) + if packet[0]&0x20 != 0 { + padding := int(packet[len(packet)-1]) + if padding == 0 || padding > end-offset { + return nil, 0, false + } + end -= padding + } + if offset > end { + return nil, 0, false + } + return packet[offset:end], packet[1] & 0x7f, true +} + +func encodeG711(pcm []byte, pt uint8) []byte { + if pt == 0 { + return g711.EncodeUlaw(pcm) + } + return g711.EncodeAlaw(pcm) +} + +func decodeG711(data []byte, pt uint8) []byte { + if pt == 0 { + return g711.DecodeUlaw(data) + } + return g711.DecodeAlaw(data) +} + +func readWAV(r io.Reader) ([]int16, int, error) { + data, err := io.ReadAll(r) + if err != nil || len(data) < 12 || string(data[:4]) != "RIFF" || string(data[8:12]) != "WAVE" { + return nil, 0, errors.New("not a RIFF/WAVE file") + } + var rate int + var pcm []byte + for offset := 12; offset+8 <= len(data); { + size := int(binary.LittleEndian.Uint32(data[offset+4 : offset+8])) + start, end := offset+8, offset+8+size + if end > len(data) { + return nil, 0, errors.New("truncated WAV chunk") + } + switch string(data[offset : offset+4]) { + case "fmt ": + if size < 16 || binary.LittleEndian.Uint16(data[start:start+2]) != 1 || binary.LittleEndian.Uint16(data[start+2:start+4]) != 1 || binary.LittleEndian.Uint16(data[start+14:start+16]) != 16 { + return nil, 0, errors.New("WAV must be mono PCM16") + } + rate = int(binary.LittleEndian.Uint32(data[start+4 : start+8])) + case "data": + pcm = data[start:end] + } + offset = end + size%2 + } + if rate < 8000 || rate > 96000 || len(pcm) == 0 || len(pcm)%2 != 0 { + return nil, 0, errors.New("WAV lacks valid fmt/data chunks") + } + return bytesToPCM(pcm), rate, nil +} + +func writeWAV(path string, pcm []int16, rate int) error { + data := pcmToBytes(pcm) + var out bytes.Buffer + out.WriteString("RIFF") + _ = binary.Write(&out, binary.LittleEndian, uint32(36+len(data))) + out.WriteString("WAVEfmt ") + _ = binary.Write(&out, binary.LittleEndian, uint32(16)) + _ = binary.Write(&out, binary.LittleEndian, uint16(1)) + _ = binary.Write(&out, binary.LittleEndian, uint16(1)) + _ = binary.Write(&out, binary.LittleEndian, uint32(rate)) + _ = binary.Write(&out, binary.LittleEndian, uint32(rate*2)) + _ = binary.Write(&out, binary.LittleEndian, uint16(2)) + _ = binary.Write(&out, binary.LittleEndian, uint16(16)) + out.WriteString("data") + _ = binary.Write(&out, binary.LittleEndian, uint32(len(data))) + out.Write(data) + if err := os.WriteFile(path, out.Bytes(), 0o600); err != nil { + return fmt.Errorf("write recording: %w", err) + } + return nil +} + +func resample(in []int16, from, to int) []int16 { + if len(in) == 0 || from <= 0 || to <= 0 || from == to { + return append([]int16(nil), in...) + } + out := make([]int16, int(math.Ceil(float64(len(in))*float64(to)/float64(from)))) + for i := range out { + position := float64(i) * float64(from) / float64(to) + left := min(int(position), len(in)-1) + right := min(left+1, len(in)-1) + fraction := position - float64(left) + out[i] = int16(float64(in[left])*(1-fraction) + float64(in[right])*fraction) + } + return out +} + +func tone(hz float64, duration time.Duration) []int16 { + n := int(duration.Seconds() * 8000) + out := make([]int16, n) + for i := range out { + out[i] = int16(5000 * math.Sin(2*math.Pi*hz*float64(i)/8000)) + } + return out +} + +func rms(pcm []int16) int { + if len(pcm) == 0 { + return 0 + } + var sum float64 + for _, sample := range pcm { + sum += float64(sample) * float64(sample) + } + return int(math.Sqrt(sum / float64(len(pcm)))) +} + +func pcmToBytes(pcm []int16) []byte { + out := make([]byte, len(pcm)*2) + for i, sample := range pcm { + binary.LittleEndian.PutUint16(out[i*2:], uint16(sample)) + } + return out +} + +func bytesToPCM(data []byte) []int16 { + out := make([]int16, len(data)/2) + for i := range out { + out[i] = int16(binary.LittleEndian.Uint16(data[i*2:])) + } + return out +} + +type signalLog struct { + mu sync.Mutex + file *os.File + number, callee string + caller string +} + +func newSignalLog(path, number, callerID string) (*signalLog, error) { + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return nil, fmt.Errorf("create log directory: %w", err) + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600) + if err != nil { + return nil, fmt.Errorf("open SIP log: %w", err) + } + return &signalLog{file: f, number: number, callee: calleePrefix + number, caller: callerID}, nil +} + +func (l *signalLog) Close() error { return l.file.Close() } + +func (l *signalLog) message(direction, message string) { + l.write(direction + "\n" + l.redact(message)) +} + +func (l *signalLog) event(message string) { l.write(l.redact(message)) } + +func (l *signalLog) write(message string) { + l.mu.Lock() + defer l.mu.Unlock() + _, _ = fmt.Fprintf(l.file, "[%s] %s\n", time.Now().UTC().Format(time.RFC3339Nano), strings.TrimSpace(message)) +} + +func (l *signalLog) redact(message string) string { + message = strings.ReplaceAll(message, l.callee, calleePrefix+"*******"+l.number[len(l.number)-4:]) + message = strings.ReplaceAll(message, l.number, "*******"+l.number[len(l.number)-4:]) + message = strings.ReplaceAll(message, l.caller, "BD****"+l.caller[len(l.caller)-4:]) + return mobileNumber.ReplaceAllString(message, "*******$1") +} + +var mobileNumber = regexp.MustCompile("\\b1[3-9][0-9]{6}([0-9]{4})\\b") + +func validateNumber(number string) error { + if !regexp.MustCompile("^1[3-9][0-9]{9}$").MatchString(number) { + return errors.New("must be an authorized 11-digit mainland mobile number supplied only through the environment") + } + return nil +} + +func routeIP(server string) (string, error) { + conn, err := net.Dial("udp4", server) + if err != nil { + return "", fmt.Errorf("detect route IP: %w", err) + } + defer conn.Close() + return conn.LocalAddr().(*net.UDPAddr).IP.String(), nil +} + +func envOr(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +} + +func randomUint16() uint16 { + var b [2]byte + _, _ = rand.Read(b[:]) + return binary.BigEndian.Uint16(b[:]) +} + +func randomUint32() uint32 { + var b [4]byte + _, _ = rand.Read(b[:]) + return binary.BigEndian.Uint32(b[:]) +} diff --git a/cmd/sip-demo/main_test.go b/cmd/sip-demo/main_test.go new file mode 100644 index 0000000..b93152b --- /dev/null +++ b/cmd/sip-demo/main_test.go @@ -0,0 +1,365 @@ +package main + +import ( + "bytes" + "context" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "github.com/emiago/sipgo" + "github.com/emiago/sipgo/sip" +) + +func TestOfflineMediaAndSafety(t *testing.T) { + const testNumber = "138" + "0000" + "1234" + if err := validateNumber(testNumber); err != nil { + t.Fatal(err) + } + for _, number := range []string{"", "1380000123", "23800001234", "1380000123x"} { + if validateNumber(number) == nil { + t.Fatalf("accepted invalid number %q", number) + } + } + + addr, pt, err := answerMedia([]byte("v=0\r\nc=IN IP4 192.0.2.10\r\nm=audio 40000 RTP/AVP 0 8\r\n")) + if err != nil || addr.String() != "192.0.2.10:40000" || pt != 8 { + t.Fatalf("answerMedia = %v, %d, %v", addr, pt, err) + } + payload := []byte{1, 2, 3} + packet := marshalRTP(8, 7, 160, 42, payload) + got, gotPT, ok := rtpPayload(packet) + if !ok || gotPT != 8 || !bytes.Equal(got, payload) { + t.Fatalf("rtpPayload = %v, %d, %v", got, gotPT, ok) + } + + pcm := []int16{-2000, 0, 2000, 1000} + if got := bytesToPCM(pcmToBytes(pcm)); !equalPCM(got, pcm) { + t.Fatalf("PCM round trip = %v", got) + } + if got := resample(pcm, 8000, 16000); len(got) != 8 { + t.Fatalf("resampled length = %d", len(got)) + } + + dir := t.TempDir() + wav := filepath.Join(dir, "roundtrip.wav") + if err := writeWAV(wav, pcm, 8000); err != nil { + t.Fatal(err) + } + f, err := os.Open(wav) + if err != nil { + t.Fatal(err) + } + gotPCM, rate, err := readWAV(f) + f.Close() + if err != nil || rate != 8000 || !equalPCM(gotPCM, pcm) { + t.Fatalf("WAV round trip = %v, %d, %v", gotPCM, rate, err) + } + + logPath := filepath.Join(dir, "sip.log") + log, err := newSignalLog(logPath, testNumber, caller) + if err != nil { + t.Fatal(err) + } + log.message(">>>", "INVITE sip:"+calleePrefix+testNumber+"@example.invalid From: "+caller) + log.Close() + content, err := os.ReadFile(logPath) + if err != nil { + t.Fatal(err) + } + text := string(content) + if strings.Contains(text, testNumber) || strings.Contains(text, caller) || + !strings.Contains(text, "7089*******1234") || !strings.Contains(text, "BD****5882") { + t.Fatalf("unsafe redaction: %s", text) + } +} + +func TestBailianTTSContract(t *testing.T) { + const key = "test-key-that-must-not-leak" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/compatible-mode/v1/audio/speech" { + t.Errorf("path = %s", r.URL.Path) + } + if r.Header.Get("Authorization") != "Bearer "+key { + t.Error("missing bearer authorization") + } + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = w.Write(pcmToBytes([]int16{1, 2, 3})) + })) + defer server.Close() + + pcm, rate, err := synthesize(context.Background(), server.Client(), server.URL+"/compatible-mode/v1", key, "test-voice", "测试") + if err != nil || rate != 24000 || !equalPCM(pcm, []int16{1, 2, 3}) || strings.Contains(errString(err), key) { + t.Fatalf("synthesize = %v, %d, %v", pcm, rate, err) + } +} + +func TestBailianTTSCredentialsRejectPlaintextAndUnsafeRedirects(t *testing.T) { + const key = "redirect-key-that-must-not-leak" + plainRequests := 0 + plainClient := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + plainRequests++ + return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(""))}, nil + })} + if _, _, err := synthesize(context.Background(), plainClient, "http://192.0.2.1", key, "voice", "text"); err == nil || plainRequests != 0 { + t.Fatal("accepted non-loopback HTTP") + } + + targetHit := make(chan struct{}, 1) + target := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { + targetHit <- struct{}{} + })) + defer target.Close() + source := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL+"/audio/speech", http.StatusTemporaryRedirect) + })) + defer source.Close() + _, _, err := synthesize(context.Background(), source.Client(), source.URL, key, "voice", "text") + if err == nil || strings.Contains(err.Error(), key) { + t.Fatalf("cross-origin redirect error = %v", err) + } + select { + case <-targetHit: + t.Fatal("credential request reached redirect target") + default: + } + + downgradeRequests := 0 + downgradeClient := &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + downgradeRequests++ + if downgradeRequests == 1 { + return &http.Response{ + StatusCode: http.StatusTemporaryRedirect, + Header: http.Header{"Location": []string{"http://secure.example/audio/speech"}}, + Body: io.NopCloser(strings.NewReader("")), + Request: req, + }, nil + } + return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(""))}, nil + })} + _, _, err = synthesize(context.Background(), downgradeClient, "https://secure.example", key, "voice", "text") + if err == nil || downgradeRequests != 1 || strings.Contains(err.Error(), key) { + t.Fatalf("HTTPS downgrade error = %v", err) + } +} + +func TestInboundVoiceInterruptsPlayback(t *testing.T) { + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + sender, err := net.DialUDP("udp4", nil, conn.LocalAddr().(*net.UDPAddr)) + if err != nil { + t.Fatal(err) + } + defer sender.Close() + + ctx, cancel := context.WithCancel(context.Background()) + interrupted := make(chan struct{}) + done := make(chan []int16, 1) + go receiveRTP(ctx, conn, sender.LocalAddr().(*net.UDPAddr), 1000, 8000, interrupted, done) + frame := make([]int16, 160) + for i := range frame { + frame[i] = 4000 + } + payload := encodeG711(pcmToBytes(frame), 8) + for seq := uint16(0); seq < 3; seq++ { + _, _ = sender.Write(marshalRTP(8, seq, uint32(seq)*160, 1, payload)) + } + select { + case <-interrupted: + case <-time.After(time.Second): + t.Fatal("playback was not interrupted") + } + cancel() + select { + case pcm := <-done: + if len(pcm) != 480 { + t.Fatalf("recorded samples = %d", len(pcm)) + } + case <-time.After(time.Second): + t.Fatal("RTP receiver did not stop") + } +} + +func TestSingleCallAgainstLocalSIPAndRTP(t *testing.T) { + ackFailure := errors.New("injected ACK failure") + tests := []struct { + name string + ack func(context.Context, *sipgo.DialogClientSession) error + validSDP bool + byeStatus int + wantMethods string + wantErrors []string + }{ + {name: "normal", validSDP: true, byeStatus: sip.StatusOK, wantMethods: "INVITE,ACK,BYE"}, + { + name: "ACK failure still sends BYE", validSDP: true, byeStatus: sip.StatusOK, wantMethods: "INVITE,BYE", + ack: func(context.Context, *sipgo.DialogClientSession) error { return ackFailure }, + wantErrors: []string{"send ACK: injected ACK failure"}, + }, + { + name: "media and cleanup errors are preserved", validSDP: false, byeStatus: sip.StatusInternalServerError, wantMethods: "INVITE,ACK,BYE", + wantErrors: []string{"answer SDP lacks usable", "SIP/2.0 500"}, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + methods, logData, err := localCall(t, test.ack, test.validSDP, test.byeStatus) + if len(test.wantErrors) == 0 && err != nil { + t.Fatal(err) + } + for _, want := range test.wantErrors { + if err == nil || !strings.Contains(err.Error(), want) { + t.Fatalf("error %v does not contain %q", err, want) + } + } + if got := strings.Join(methods, ","); got != test.wantMethods { + t.Fatalf("SIP methods = %s, want %s", got, test.wantMethods) + } + if strings.Contains(string(logData), "138"+"0000"+"1234") || !strings.Contains(string(logData), "200 OK") { + t.Fatalf("unexpected SIP log: %s", logData) + } + }) + } +} + +func localCall(t *testing.T, ack func(context.Context, *sipgo.DialogClientSession) error, validSDP bool, byeStatus int) ([]string, []byte, error) { + t.Helper() + const testNumber = "138" + "0000" + "1234" + rtpConn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4zero, Port: 0}) + if err != nil { + t.Fatal(err) + } + defer rtpConn.Close() + sipConn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + if err != nil { + t.Fatal(err) + } + defer sipConn.Close() + + type serverResult struct { + methods []string + err error + } + serverDone := make(chan serverResult, 1) + go func() { + parser := sip.NewParser() + buf := make([]byte, 4096) + var methods []string + _ = sipConn.SetReadDeadline(time.Now().Add(3 * time.Second)) + for { + n, source, err := sipConn.ReadFromUDP(buf) + if err != nil { + serverDone <- serverResult{methods, err} + return + } + message, err := parser.ParseSIP(buf[:n]) + if err != nil { + serverDone <- serverResult{methods, err} + return + } + request, ok := message.(*sip.Request) + if !ok { + continue + } + methods = append(methods, request.Method.String()) + switch request.Method { + case sip.INVITE: + for _, status := range []int{sip.StatusRinging, sip.StatusOK} { + response := sip.NewResponseFromRequest(request, status, map[int]string{sip.StatusRinging: "Ringing", sip.StatusOK: "OK"}[status], nil) + response.To().Params.Add("tag", "offline-test") + if status == sip.StatusOK { + response.AppendHeader(&sip.ContactHeader{Address: sip.Uri{Scheme: "sip", User: "mock", Host: "127.0.0.1", Port: sipConn.LocalAddr().(*net.UDPAddr).Port}}) + response.AppendHeader(sip.NewHeader("Content-Type", "application/sdp")) + if validSDP { + response.SetBody([]byte("v=0\r\nc=IN IP4 127.0.0.1\r\nm=audio " + strconv.Itoa(rtpConn.LocalAddr().(*net.UDPAddr).Port) + " RTP/AVP 8\r\n")) + } else { + response.SetBody([]byte("v=0\r\n")) + } + } + if _, err := sipConn.WriteToUDP([]byte(response.String()), source); err != nil { + serverDone <- serverResult{methods, err} + return + } + if status == sip.StatusRinging { + time.Sleep(5 * time.Millisecond) + } + } + case sip.BYE: + reason := "OK" + if byeStatus != sip.StatusOK { + reason = "Server Error" + } + response := sip.NewResponseFromRequest(request, byeStatus, reason, nil) + _, err := sipConn.WriteToUDP([]byte(response.String()), source) + serverDone <- serverResult{methods, err} + return + } + } + }() + + dir := t.TempDir() + cfg := config{ + server: sipConn.LocalAddr().String(), + number: testNumber, + ttsText: "", + ttsVoice: "unused", + logPath: filepath.Join(dir, "sip.log"), + recordDir: filepath.Join(dir, "recordings"), + localAddr: "127.0.0.1:0", + advertiseIP: "127.0.0.1", + ringTimeout: time.Second, + maxDuration: 120 * time.Millisecond, + vadThreshold: 1000, + } + var runErr error + if ack == nil { + runErr = run(context.Background(), cfg) + } else { + runErr = runWithACK(context.Background(), cfg, ack) + } + result := <-serverDone + if result.err != nil { + t.Fatal(result.err) + } + logData, err := os.ReadFile(cfg.logPath) + if err != nil { + t.Fatal(err) + } + return result.methods, logData, runErr +} + +func equalPCM(a, b []int16) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +func errString(err error) string { + if err == nil { + return "" + } + return err.Error() +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} diff --git a/docs/examples/sip-signaling.redacted.log b/docs/examples/sip-signaling.redacted.log new file mode 100644 index 0000000..8070b86 --- /dev/null +++ b/docs/examples/sip-signaling.redacted.log @@ -0,0 +1,47 @@ +[2026-08-18T00:00:00Z] >>> +INVITE sip:7089*******1234@61.132.228.221:5060 SIP/2.0 +Via: SIP/2.0/UDP 192.0.2.10:5062;branch=z9hG4bK-sample;rport +From: "BD****5882" ;tag=sample-from +To: +Call-ID: 00000000-0000-0000-0000-000000000000 +CSeq: 1 INVITE +Contact: +Content-Type: application/sdp +Content-Length: 170 + +v=0 +o=- 0 0 IN IP4 192.0.2.10 +s=magic-sonar +c=IN IP4 192.0.2.10 +t=0 0 +m=audio 40000 RTP/AVP 8 0 +a=rtpmap:8 PCMA/8000 +a=rtpmap:0 PCMU/8000 +a=ptime:20 +a=sendrecv +[2026-08-18T00:00:01Z] <<< +SIP/2.0 180 Ringing +Via: SIP/2.0/UDP 192.0.2.10:5062;branch=z9hG4bK-sample;rport +From: "BD****5882" ;tag=sample-from +To: ;tag=sample-to +Call-ID: 00000000-0000-0000-0000-000000000000 +CSeq: 1 INVITE +Content-Length: 0 +[2026-08-18T00:00:02Z] <<< +SIP/2.0 200 OK +Via: SIP/2.0/UDP 192.0.2.10:5062;branch=z9hG4bK-sample;rport +From: "BD****5882" ;tag=sample-from +To: ;tag=sample-to +Call-ID: 00000000-0000-0000-0000-000000000000 +CSeq: 1 INVITE +Contact: +Content-Type: application/sdp +Content-Length: 73 + +v=0 +c=IN IP4 192.0.2.20 +m=audio 42000 RTP/AVP 8 +a=rtpmap:8 PCMA/8000 +[2026-08-18T00:00:02Z] >>> ACK call-id=00000000-0000-0000-0000-000000000000 cseq=1 +[2026-08-18T00:00:04Z] playback interrupted by inbound voice +[2026-08-18T00:00:10Z] >>> BYE call-id=00000000-0000-0000-0000-000000000000 cseq=2 diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..2c8b1a8 --- /dev/null +++ b/go.mod @@ -0,0 +1,18 @@ +module git.ipao.vip/rogee/magic-sonar + +go 1.23.0 + +require ( + github.com/emiago/sipgo v1.4.3 + github.com/zaf/g711 v1.4.0 +) + +require ( + github.com/gobwas/httphead v0.1.0 // indirect + github.com/gobwas/pool v0.2.1 // indirect + github.com/gobwas/ws v1.3.2 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/icholy/digest v1.1.0 // indirect + golang.org/x/sync v0.16.0 // indirect + golang.org/x/sys v0.24.0 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..ad3dcf6 --- /dev/null +++ b/go.sum @@ -0,0 +1,31 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/emiago/sipgo v1.4.3 h1:Ju1Ilp0LhTcSHCyp2jBYAQDQ8TAwuxQHzT1IhO6jSIY= +github.com/emiago/sipgo v1.4.3/go.mod h1:DuwAxBZhKMqIzQFPGZb1MVAGU6Wuxj64oTOhd5dx/FY= +github.com/gobwas/httphead v0.1.0 h1:exrUm0f4YX0L7EBwZHuCF4GDp8aJfVeBrlLQrs6NqWU= +github.com/gobwas/httphead v0.1.0/go.mod h1:O/RXo79gxV8G+RqlR/otEwx4Q36zl9rqC5u12GKvMCM= +github.com/gobwas/pool v0.2.1 h1:xfeeEhW7pwmX8nuLVlqbzVc7udMDrwetjEv+TZIz1og= +github.com/gobwas/pool v0.2.1/go.mod h1:q8bcK0KcYlCgd9e7WYLm9LpyS+YeLd8JVDW6WezmKEw= +github.com/gobwas/ws v1.3.2 h1:zlnbNHxumkRvfPWgfXu8RBwyNR1x8wh9cf5PTOCqs9Q= +github.com/gobwas/ws v1.3.2/go.mod h1:hRKAFb8wOxFROYNsT1bqfWnhX+b5MFeJM9r2ZSwg/KY= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/icholy/digest v1.1.0 h1:HfGg9Irj7i+IX1o1QAmPfIBNu/Q5A5Tu3n/MED9k9H4= +github.com/icholy/digest v1.1.0/go.mod h1:QNrsSGQ5v7v9cReDI0+eyjsXGUoRSUZQHeQ5C4XLa0Y= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/zaf/g711 v1.4.0 h1:XZYkjjiAg9QTBnHqEg37m2I9q3IIDv5JRYXs2N8ma7c= +github.com/zaf/g711 v1.4.0/go.mod h1:eCDXt3dSp/kYYAoooba7ukD/Q75jvAaS4WOMr0l1Roo= +golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw= +golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg= +golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gotest.tools/v3 v3.5.1 h1:EENdUnS3pdur5nybKYIh2Vfgc8IUNBjxDPSjtiJcOzU= +gotest.tools/v3 v3.5.1/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=