2026-08-18 16:27:04 +08:00
4 changed files with 288 additions and 102 deletions
@@ -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/线路能力验证。
@@ -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)
-	}
-	data, err := io.ReadAll(io.LimitReader(res.Body, 32<<20+1))
-	if err != nil {
+	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
 		}
-	if len(data) > 32<<20 {
+	}
+	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")
 			}
-	if bytes.HasPrefix(data, []byte("RIFF")) {
-		return readWAV(bytes.NewReader(data))
+			_, _ = 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 fmt.Errorf("encode TTS command: %w", err)
+	}
+	if err := wsutil.WriteClientText(w, data); err != nil {
+		return fmt.Errorf("write TTS WebSocket: %w", err)
+	}
+	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
+		}
+	}
+}
+
+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)
+	}
+	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 {
@@ -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)
-}
@@ -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