From 79013ae11e4440dff1a395584e06e7175f0403e3 Mon Sep 17 00:00:00 2001 From: Rogee Date: Tue, 18 Aug 2026 16:27:04 +0800 Subject: [PATCH] =?UTF-8?q?H-294:=20=E4=BF=AE=E5=A4=8D=E7=99=BE=E7=BB=83?= =?UTF-8?q?=20CosyVoice=20WebSocket=20=E5=A5=91=E7=BA=A6=20(#4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Rogee --- README.md | 6 +- cmd/sip-demo/main.go | 200 ++++++++++++++++++++++++++++++-------- cmd/sip-demo/main_test.go | 182 +++++++++++++++++++++++----------- go.mod | 2 +- 4 files changed, 288 insertions(+), 102 deletions(-) diff --git a/README.md b/README.md index 5845b68..9bd3cb9 100644 --- a/README.md +++ b/README.md @@ -22,17 +22,17 @@ cmd/sip-demo 通过 UDP 向 61.132.228.221:5060 发起一通 IP 白名单、无 go test ./... go build ./cmd/sip-demo -测试使用本地 HTTP/RTP 端点，不调用百练或 SIP 服务器。 +测试使用本地 WebSocket/RTP 端点，不调用百练或 SIP 服务器。 ### 经授权的内部联调 确认执行机出口 IP 已加入白名单、UDP SIP/RTP 可达，再仅通过环境提供号码和凭据： export SIP_TEST_NUMBER='<已授权内部测试号码>' - export BAILIAN_BASE_URL='https:///compatible-mode/v1' + export BAILIAN_BASE_URL='wss://.cn-beijing.maas.aliyuncs.com/api-ws/v1/inference' 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。 +BAILIAN_BASE_URL 必须是上述北京地域工作空间的 CosyVoice WebSocket 完整端点；程序不拼接、回退或重试其他路径。除 loopback 测试外只接受 WSS。Key 只进入 WebSocket 握手的 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 index 66cf91b..a0f24e0 100644 --- a/cmd/sip-demo/main.go +++ b/cmd/sip-demo/main.go @@ -5,6 +5,7 @@ import ( "context" "crypto/rand" "encoding/binary" + "encoding/hex" "encoding/json" "errors" "flag" @@ -24,6 +25,8 @@ import ( "github.com/emiago/sipgo" "github.com/emiago/sipgo/sip" + "github.com/gobwas/ws" + "github.com/gobwas/ws/wsutil" "github.com/zaf/g711" ) @@ -214,7 +217,7 @@ func mediaProgram(ctx context.Context, cfg config) ([]int16, error) { 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) + pcm, rate, err := synthesize(ttsCtx, base, key, cfg.ttsVoice, cfg.ttsText) cancel() if err != nil { return nil, fmt.Errorf("Bailian TTS: %w", err) @@ -238,61 +241,174 @@ func mediaProgram(ctx context.Context, cfg config) ([]int16, error) { return out, nil } -func synthesize(ctx context.Context, client *http.Client, base, key, voice, text string) ([]int16, int, error) { +func synthesize(ctx context.Context, 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()))) { + if err != nil || u.Host == "" || u.User != nil || u.Path != "/api-ws/v1/inference" || u.RawPath != "" || + u.RawQuery != "" || u.Fragment != "" || (u.Scheme != "wss" && !(u.Scheme == "ws" && 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)) + taskID, err := newTaskID() if err != nil { + return nil, 0, fmt.Errorf("create TTS task ID: %w", err) + } + status := 0 + dialer := ws.Dialer{ + Header: ws.HandshakeHeaderHTTP(http.Header{"Authorization": []string{"Bearer " + key}}), + OnStatusError: func(code int, _ []byte, _ io.Reader) { + status = code + }, + } + conn, buffered, _, err := dialer.Dial(ctx, u.String()) + if err != nil { + if status != 0 { + return nil, 0, fmt.Errorf("WebSocket handshake HTTP %d", status) + } + return nil, 0, fmt.Errorf("connect TTS WebSocket: %w", err) + } + defer conn.Close() + if deadline, ok := ctx.Deadline(); ok { + _ = conn.SetDeadline(deadline) + } + var stream io.ReadWriter = conn + if buffered != nil { + defer ws.PutReader(buffered) + stream = struct { + io.Reader + io.Writer + }{buffered, conn} + } + runTask := map[string]any{ + "header": map[string]any{"action": "run-task", "task_id": taskID, "streaming": "duplex"}, + "payload": map[string]any{ + "task_group": "audio", "task": "tts", "function": "SpeechSynthesizer", "model": "cosyvoice-v3.5-plus", + "parameters": map[string]any{"text_type": "PlainText", "voice": voice, "format": "pcm", "sample_rate": 24000}, + "input": map[string]any{}, + }, + } + if err := writeTTSCommand(stream, runTask); 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 { + if err := waitTTSEvent(stream, taskID, "task-started"); 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) + for _, command := range []map[string]any{ + {"header": map[string]any{"action": "continue-task", "task_id": taskID, "streaming": "duplex"}, "payload": map[string]any{"input": map[string]any{"text": text}}}, + {"header": map[string]any{"action": "finish-task", "task_id": taskID, "streaming": "duplex"}, "payload": map[string]any{"input": map[string]any{}}}, + } { + if err := writeTTSCommand(stream, command); err != nil { + return nil, 0, err + } } - data, err := io.ReadAll(io.LimitReader(res.Body, 32<<20+1)) + var audio bytes.Buffer + for { + data, op, err := wsutil.ReadServerData(stream) + if err != nil { + return nil, 0, fmt.Errorf("read TTS WebSocket: %w", err) + } + if op == ws.OpBinary { + if audio.Len()+len(data) > 32<<20 { + return nil, 0, errors.New("audio response exceeds 32 MiB") + } + _, _ = audio.Write(data) + continue + } + event, err := parseTTSEvent(data, taskID) + if err != nil { + return nil, 0, err + } + switch event.Header.Event { + case "task-finished": + data = audio.Bytes() + if len(data) == 0 { + return nil, 0, errors.New("TTS response contains no audio") + } + if len(data)%2 != 0 { + return nil, 0, errors.New("TTS response is not PCM16") + } + return bytesToPCM(data), 24000, nil + case "task-failed": + return nil, 0, ttsFailure(event) + } + } +} + +type ttsEvent struct { + Header struct { + Event string `json:"event"` + TaskID string `json:"task_id"` + ErrorCode string `json:"error_code"` + } `json:"header"` +} + +func writeTTSCommand(w io.Writer, command any) error { + data, err := json.Marshal(command) if err != nil { - return nil, 0, err + return fmt.Errorf("encode TTS command: %w", err) } - if len(data) > 32<<20 { - return nil, 0, errors.New("audio response exceeds 32 MiB") + if err := wsutil.WriteClientText(w, data); err != nil { + return fmt.Errorf("write TTS WebSocket: %w", err) } - if bytes.HasPrefix(data, []byte("RIFF")) { - return readWAV(bytes.NewReader(data)) + return nil +} + +func waitTTSEvent(stream io.ReadWriter, taskID, want string) error { + for { + data, op, err := wsutil.ReadServerData(stream) + if err != nil { + return fmt.Errorf("read TTS WebSocket: %w", err) + } + if op != ws.OpText { + return fmt.Errorf("expected TTS event %q", want) + } + event, err := parseTTSEvent(data, taskID) + if err != nil { + return err + } + if event.Header.Event == "task-failed" { + return ttsFailure(event) + } + if event.Header.Event == want { + return nil + } } - if len(data)%2 != 0 { - return nil, 0, errors.New("TTS response is not PCM16") +} + +func parseTTSEvent(data []byte, taskID string) (ttsEvent, error) { + var event ttsEvent + if err := json.Unmarshal(data, &event); err != nil { + return event, fmt.Errorf("decode TTS event: %w", err) } - return bytesToPCM(data), 24000, nil + if event.Header.Event == "" || event.Header.TaskID != taskID { + return event, errors.New("invalid TTS event") + } + return event, nil +} + +func ttsFailure(event ttsEvent) error { + if event.Header.ErrorCode == "" { + return errors.New("TTS task failed") + } + return fmt.Errorf("TTS task failed: %s", event.Header.ErrorCode) +} + +func newTaskID() (string, error) { + var raw [16]byte + if _, err := rand.Read(raw[:]); err != nil { + return "", err + } + raw[6] = raw[6]&0x0f | 0x40 + raw[8] = raw[8]&0x3f | 0x80 + var id [36]byte + hex.Encode(id[0:8], raw[0:4]) + id[8] = '-' + hex.Encode(id[9:13], raw[4:6]) + id[13] = '-' + hex.Encode(id[14:18], raw[6:8]) + id[18] = '-' + hex.Encode(id[19:23], raw[8:10]) + id[23] = '-' + hex.Encode(id[24:36], raw[10:16]) + return string(id[:]), nil } func isLoopback(host string) bool { diff --git a/cmd/sip-demo/main_test.go b/cmd/sip-demo/main_test.go index b93152b..545e9a3 100644 --- a/cmd/sip-demo/main_test.go +++ b/cmd/sip-demo/main_test.go @@ -3,8 +3,9 @@ package main import ( "bytes" "context" + "encoding/json" "errors" - "io" + "fmt" "net" "net/http" "net/http/httptest" @@ -12,11 +13,14 @@ import ( "path/filepath" "strconv" "strings" + "sync/atomic" "testing" "time" "github.com/emiago/sipgo" "github.com/emiago/sipgo/sip" + "github.com/gobwas/ws" + "github.com/gobwas/ws/wsutil" ) func TestOfflineMediaAndSafety(t *testing.T) { @@ -84,71 +88,143 @@ func TestOfflineMediaAndSafety(t *testing.T) { func TestBailianTTSContract(t *testing.T) { const key = "test-key-that-must-not-leak" + type command struct { + Header struct { + Action string `json:"action"` + TaskID string `json:"task_id"` + Streaming string `json:"streaming"` + } `json:"header"` + Payload struct { + TaskGroup string `json:"task_group"` + Task string `json:"task"` + Function string `json:"function"` + Model string `json:"model"` + Parameters struct { + TextType string `json:"text_type"` + Voice string `json:"voice"` + Format string `json:"format"` + SampleRate int `json:"sample_rate"` + } `json:"parameters"` + Input map[string]any `json:"input"` + } `json:"payload"` + } + serverErrors := make(chan error, 1) 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) + fail := func(format string, args ...any) { + select { + case serverErrors <- fmt.Errorf(format, args...): + default: + } + } + if r.URL.Path != "/api-ws/v1/inference" { + fail("path = %s", r.URL.Path) } if r.Header.Get("Authorization") != "Bearer "+key { - t.Error("missing bearer authorization") + fail("missing bearer authorization") } - w.Header().Set("Content-Type", "application/octet-stream") - _, _ = w.Write(pcmToBytes([]int16{1, 2, 3})) + conn, _, _, err := ws.UpgradeHTTP(r, w) + if err != nil { + fail("upgrade: %v", err) + return + } + defer conn.Close() + read := func() command { + var got command + data, op, err := wsutil.ReadClientData(conn) + if err != nil { + fail("read command: %v", err) + return got + } + if op != ws.OpText { + fail("command opcode = %v", op) + } + if err := json.Unmarshal(data, &got); err != nil { + fail("decode command: %v", err) + } + return got + } + writeEvent := func(event, taskID string) { + data, _ := json.Marshal(map[string]any{"header": map[string]any{"event": event, "task_id": taskID}, "payload": map[string]any{}}) + if err := wsutil.WriteServerText(conn, data); err != nil { + fail("write event: %v", err) + } + } + + run := read() + if run.Header.Action != "run-task" || run.Header.Streaming != "duplex" || + run.Payload.TaskGroup != "audio" || run.Payload.Task != "tts" || run.Payload.Function != "SpeechSynthesizer" || + run.Payload.Model != "cosyvoice-v3.5-plus" || run.Payload.Parameters.TextType != "PlainText" || + run.Payload.Parameters.Voice != "test-voice" || run.Payload.Parameters.Format != "pcm" || run.Payload.Parameters.SampleRate != 24000 { + fail("invalid run-task contract") + } + if !isUUID(run.Header.TaskID) { + fail("task_id is not a UUID") + } + writeEvent("task-started", run.Header.TaskID) + + continued := read() + if continued.Header.Action != "continue-task" || continued.Header.TaskID != run.Header.TaskID || + continued.Header.Streaming != "duplex" || continued.Payload.Input["text"] != "测试" { + fail("invalid continue-task contract") + } + finished := read() + if finished.Header.Action != "finish-task" || finished.Header.TaskID != run.Header.TaskID || finished.Header.Streaming != "duplex" { + fail("invalid finish-task contract") + } + if err := wsutil.WriteServerBinary(conn, pcmToBytes([]int16{1, 2, 3})); err != nil { + fail("write audio: %v", err) + } + writeEvent("task-finished", run.Header.TaskID) })) defer server.Close() - pcm, rate, err := synthesize(context.Background(), server.Client(), server.URL+"/compatible-mode/v1", key, "test-voice", "测试") + endpoint := "ws" + strings.TrimPrefix(server.URL, "http") + "/api-ws/v1/inference" + pcm, rate, err := synthesize(context.Background(), endpoint, 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") + case err := <-serverErrors: + t.Fatal(err) 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 +func TestBailianTTSEndpointMustBeExactAndSecure(t *testing.T) { + var hits atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { hits.Add(1) })) + defer server.Close() + loopback := "ws" + strings.TrimPrefix(server.URL, "http") + for _, endpoint := range []string{ + loopback + "/compatible-mode/v1", + loopback + "/api-ws/v1/inference/", + loopback + "/api-ws/v1/inference?path=other", + "ws://192.0.2.1/api-ws/v1/inference", + "https://example.invalid/compatible-mode/v1", + } { + _, _, err := synthesize(context.Background(), endpoint, "secret", "voice", "text") + if err == nil || strings.Contains(err.Error(), "secret") { + t.Fatalf("accepted endpoint %q: %v", endpoint, err) } - 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) } + if hits.Load() != 0 { + t.Fatalf("invalid endpoint received %d requests", hits.Load()) + } +} + +func isUUID(value string) bool { + if len(value) != 36 || value[8] != '-' || value[13] != '-' || value[18] != '-' || value[23] != '-' { + return false + } + for i, c := range value { + if i == 8 || i == 13 || i == 18 || i == 23 { + continue + } + if !strings.ContainsRune("0123456789abcdef", c) { + return false + } + } + return true } func TestInboundVoiceInterruptsPlayback(t *testing.T) { @@ -357,9 +433,3 @@ func errString(err error) string { } 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/go.mod b/go.mod index 2c8b1a8..a02afcb 100644 --- a/go.mod +++ b/go.mod @@ -4,13 +4,13 @@ go 1.23.0 require ( github.com/emiago/sipgo v1.4.3 + github.com/gobwas/ws v1.3.2 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