rogee/magic-sonar
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: use Bailian CosyVoice websocket contract

Browse Source 
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
rogee
2026-08-18 16:15:32 +08:00
co-authored by multica-agent
parent 8f423d3381
commit fb30b9d958
4 changed files with 288 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+3 -3
View File
@@ -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://<endpoint>/compatible-mode/v1'
export BAILIAN_BASE_URL='wss://<WorkspaceId>.cn-beijing.maas.aliyuncs.com/api-ws/v1/inference'
export BAILIAN_API_KEY='<server-side-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/线路能力验证。
cmd/sip-demo/main.go
+158 -42
View File
@@ -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 {
cmd/sip-demo/main_test.go
+126 -56
View File
@@ -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)
}
go.mod
+1 -1
View File
@@ -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