go-sip/cmd/saas-server/integration_test.go

//go:build integration

package main

import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"encoding/json"
	"encoding/pem"
	"errors"
	"log/slog"
	"math/big"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"testing"
	"time"

	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
	"git.ipao.vip/rogee/go-sip/internal/agent"
	"git.ipao.vip/rogee/go-sip/internal/ai"
	"git.ipao.vip/rogee/go-sip/internal/configread"
	"git.ipao.vip/rogee/go-sip/internal/dispatcher"
	"git.ipao.vip/rogee/go-sip/internal/mq"
	"git.ipao.vip/rogee/go-sip/internal/oss"
	"git.ipao.vip/rogee/go-sip/internal/rpc"
	"git.ipao.vip/rogee/go-sip/internal/store"
	"github.com/gofiber/fiber/v3"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

func localCertificates(t *testing.T) ([]byte, []byte, []byte, []byte, []byte, *x509.Certificate, *x509.Certificate) {
	t.Helper()
	makeKey := func() *ecdsa.PrivateKey {
		k, e := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
		if e != nil {
			t.Fatal(e)
		}
		return k
	}
	caKey := makeKey()
	now := time.Now()
	ca := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "isolated-saas-test"}, NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour), IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign}
	caDER, e := x509.CreateCertificate(rand.Reader, ca, ca, &caKey.PublicKey, caKey)
	if e != nil {
		t.Fatal(e)
	}
	caPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER})
	leaf := func(serial int64, name string) ([]byte, []byte, *x509.Certificate) {
		key := makeKey()
		cert := &x509.Certificate{SerialNumber: big.NewInt(serial), Subject: pkix.Name{CommonName: name}, DNSNames: []string{name}, IPAddresses: []net.IP{net.ParseIP("127.0.0.1")}, NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour), KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth, x509.ExtKeyUsageClientAuth}}
		raw, err := x509.CreateCertificate(rand.Reader, cert, ca, &key.PublicKey, caKey)
		if err != nil {
			t.Fatal(err)
		}
		parsed, err := x509.ParseCertificate(raw)
		if err != nil {
			t.Fatal(err)
		}
		keyDER, err := x509.MarshalECPrivateKey(key)
		if err != nil {
			t.Fatal(err)
		}
		return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: raw}), pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), parsed
	}
	dc, dk, dl := leaf(2, "dispatcher.local")
	ac, ak, al := leaf(3, "agent.local")
	return caPEM, dc, dk, ac, ak, dl, al
}

func TestSaaSServerIsolatedHTTPSMQAgentMTLS(t *testing.T) {
	address := os.Getenv("SAAS_MOCK_TEST_BROKER_URL")
	if address == "" {
		t.Skip("requires dedicated isolated RabbitMQ vhost")
	}
	dir, id := fixture(t)
	// Synthetic test snapshots permit this Mock run at any time; no real configuration is read.
	task, err := os.ReadFile("../../contracts/schema/examples/config-read-task-asr.json")
	if err != nil {
		t.Fatal(err)
	}
	var fields map[string]any
	json.Unmarshal(task, &fields)
	fields["task_id"] = "1"
	weekly := map[string]any{}
	for _, day := range []string{"monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"} {
		weekly[day] = []map[string]string{{"start": "00:00", "end": "24:00"}}
	}
	fields["schedule"] = map[string]any{"time_zone": "Asia/Shanghai", "starts_at": "2000-01-01T00:00:00+08:00", "weekly_windows": weekly}
	task, _ = json.Marshal(fields)
	if err = os.WriteFile(filepath.Join(dir, "tasks/1.json"), task, 0600); err != nil {
		t.Fatal(err)
	}
	sipRaw, err := os.ReadFile(filepath.Join(dir, "sip.json"))
	if err != nil {
		t.Fatal(err)
	}
	fields = nil // JSON decoding into an existing map retains unrelated task fields.
	json.Unmarshal(sipRaw, &fields)
	for _, item := range fields["trunks"].([]any) {
		trunk := item.(map[string]any)
		trunk["transport"] = "udp"
		trunk["auth_mode"] = "ip"
		trunk["registration_required"] = false
		trunk["max_concurrent_calls"] = float64(2)
		trunk["schedule"] = map[string]any{"time_zone": "Asia/Shanghai", "weekly_windows": weekly}
	}
	sipRaw, _ = json.Marshal(fields)
	if err = os.WriteFile(filepath.Join(dir, "sip.json"), sipRaw, 0600); err != nil {
		t.Fatal(err)
	}
	before := map[string][]byte{}
	if err = filepath.WalkDir(dir, func(path string, d os.DirEntry, e error) error {
		if e != nil {
			return e
		}
		if !d.IsDir() {
			raw, e := os.ReadFile(path)
			if e != nil {
				return e
			}
			before[path] = raw
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	data, err := loadSnapshots(dir, id)
	if err != nil {
		t.Fatal(err)
	}
	data, err = testSnapshots(data)
	if err != nil {
		t.Fatal(err)
	}
	id = data.dispatcherID
	b, err := connectBroker(address, data)
	if err != nil {
		t.Fatal(err)
	}
	defer b.conn.Close()
	// The normal SaaS control queue already exists in the shared deployment.
	setup, err := b.conn.Channel()
	if err != nil {
		t.Fatal(err)
	}
	if _, err = setup.QueueDeclare(mq.ControlQueueName(id), true, false, false, false, nil); err != nil {
		t.Fatal(err)
	}
	if err = setup.QueueBind(mq.ControlQueueName(id), "d."+id+".control.in", mq.CommandsExchange, false, nil); err != nil {
		t.Fatal(err)
	}
	setup.Close()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	s := newService(data, "isolated-secret")
	s.feedbackDir = t.TempDir()
	s.publish = b.publish
	consumeDone := make(chan error, 1)
	go func() { consumeDone <- b.consume(ctx, s) }()
	defer func() {
		cancel()
		select {
		case e := <-consumeDone:
			if e != nil {
				t.Error(e)
			}
		case <-time.After(5 * time.Second):
			t.Error("consumer did not stop")
		}
	}()
	ca, dc, dk, ac, ak, dl, al := localCertificates(t)
	roots := x509.NewCertPool()
	roots.AppendCertsFromPEM(ca)
	pair, err := tls.X509KeyPair(dc, dk)
	if err != nil {
		t.Fatal(err)
	}
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	go s.app.Listener(tls.NewListener(listener, &tls.Config{Certificates: []tls.Certificate{pair}, MinVersion: tls.VersionTLS12}), fiber.ListenConfig{DisableStartupMessage: true})
	defer s.app.ShutdownWithTimeout(5 * time.Second)
	httpClient := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12}}, Timeout: 5 * time.Second}
	client, err := configread.NewClient("https://"+listener.Addr().String(), id, "isolated-secret", httpClient)
	if err != nil {
		t.Fatal(err)
	}
	db, err := store.Open(filepath.Join(t.TempDir(), "dispatcher.sqlite"))
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	coordinator := dispatcher.NewAgentCoordinator(time.Now)
	signer, err := oss.NewClient(oss.Config{Endpoint: "https://127.0.0.1:1", Region: "cn-test", Bucket: "mock-bucket", AccessKeyID: "test-id", AccessKeySecret: "test-secret", MaxAssetBytes: 4096})
	if err != nil {
		t.Fatal(err)
	}
	feedbackRPC := &rpc.RecordingServer{Store: db, OSS: signer, DispatcherID: id, TrustedFingerprints: map[string]struct{}{rpc.CertificateFingerprint(al): {}}, AuthorizeSession: coordinator.AuthorizeInboundMeta}
	serveRPC := func(cert, key []byte, handler agentpb.AgentControlServiceServer) string {
		cfg, e := rpc.NewServerTLSConfig(ca, cert, key)
		if e != nil {
			t.Fatal(e)
		}
		l, e := net.Listen("tcp", "127.0.0.1:0")
		if e != nil {
			t.Fatal(e)
		}
		server := grpc.NewServer(grpc.Creds(credentials.NewTLS(cfg)))
		agentpb.RegisterAgentControlServiceServer(server, handler)
		go server.Serve(l)
		t.Cleanup(server.Stop)
		return l.Addr().String()
	}
	connectRPC := func(target, name string, cert, key []byte) agentpb.AgentControlServiceClient {
		cfg, e := rpc.NewClientTLSConfig(ca, cert, key, name)
		if e != nil {
			t.Fatal(e)
		}
		conn, e := grpc.NewClient(target, grpc.WithTransportCredentials(credentials.NewTLS(cfg)))
		if e != nil {
			t.Fatal(e)
		}
		t.Cleanup(func() { conn.Close() })
		return agentpb.NewAgentControlServiceClient(conn)
	}
	agentToDispatcher := connectRPC(serveRPC(dc, dk, feedbackRPC), "dispatcher.local", ac, ak)
	workerFailure := make(chan error, 1)
	var handler *rpc.Server
	worker := &rpc.ApprovedCallWorker{Lifecycle: ctx, Calls: &agent.TaskCalls{}, Prepare: func(execution rpc.ApprovedExecution) (func(context.Context) error, error) {
		runner := &rpc.ApprovedRecordedMockCall{InboundPCM16: bytes.Repeat([]byte{1, 0}, 1600), Script: ai.ApprovedMockScript{Turns: []ai.ApprovedMockTurn{{Transcript: "synthetic ASR test"}}}, MaxWAVBytes: 4096, ExpectedRecording: false, Outcome: "answered", ReasonMessage: "isolated synthetic media", ReportTimeout: 5 * time.Second, Delivery: &agent.RecordingDelivery{Call: agent.RecordingClient{Client: agentToDispatcher, DispatcherID: execution.DispatcherID, TenantID: execution.TenantID, SourceEventID: execution.SourceEventID, Session: func(context.Context) (*agentpb.RequestMeta, error) { return handler.ActiveSessionMeta() }}}}
		return runner.Prepare(execution)
	}, OnFailure: func(_ rpc.ApprovedExecution, e error) error { workerFailure <- e; return nil }}
	handler, err = rpc.NewApprovedAgentServer(rpc.ServerOptions{Mode: "mock", StatePath: filepath.Join(t.TempDir(), "session.json"), ApprovedDispatcherID: id, Status: &agentpb.AgentStatus{AgentId: "agent-mock", CellId: "cell-mock", BootId: "boot-test", ProtocolVersion: "agent.v1"}, LoadedSIP: func(context.Context) (map[string]int64, error) { return map[string]int64{"trunk-mock": 8}, nil }, RequirePeerCertificate: true, PeerCertificateFingerprints: map[string]struct{}{rpc.CertificateFingerprint(dl): {}}}, worker)
	if err != nil {
		t.Fatal(err)
	}
	agentClient := connectRPC(serveRPC(ac, ak, handler), "agent.local", dc, dk)
	if err = coordinator.Register("agent-mock", agentClient); err != nil {
		t.Fatal(err)
	}
	activateCtx, stop := context.WithTimeout(ctx, 5*time.Second)
	defer stop()
	status, err := coordinator.Probe(activateCtx, "agent-mock", "cell-mock")
	if err != nil {
		t.Fatal(err)
	}
	if _, err = coordinator.Activate(activateCtx, id, "agent-mock", "cell-mock", status.BootId, "test-epoch", 1); err != nil {
		t.Fatal(err)
	}
	originator := &dispatcher.ApprovedOriginator{DispatcherID: id, Client: agentClient, Meta: func(ctx context.Context) (*agentpb.RequestMeta, error) {
		return coordinator.ApprovedMeta(ctx, "agent-mock")
	}}
	dispatcherBroker, err := mq.Open(address, id, 1)
	if err != nil {
		t.Fatal(err)
	}
	defer dispatcherBroker.Close()
	runtime := &dispatcher.Runtime{Broker: dispatcherBroker, Bootstrap: dispatcher.Bootstrap{DispatcherID: id, Client: client, Store: db, VerifySIP: originator.VerifySIP}, Execute: dispatcher.ExecuteController{DispatcherID: id, Store: db, Originator: originator, Publisher: dispatcherBroker, Now: time.Now}, Control: dispatcher.ControlController{DispatcherID: id, Store: db, Client: client, Agent: originator, Now: time.Now}, PollInterval: 20 * time.Millisecond, Logger: slog.Default()}
	runtimeDone := make(chan error, 1)
	go func() { runtimeDone <- runtime.Serve(ctx) }()
	defer func() {
		cancel()
		select {
		case e := <-runtimeDone:
			if e != nil && !errors.Is(e, context.Canceled) {
				t.Error(e)
			}
		case <-time.After(5 * time.Second):
			t.Error("Dispatcher did not stop")
		}
	}()
	sourceID := "numberID-integration/Original:001"
	// The first request may queue while bootstrap completes; no direct Agent call or forged feedback.
	code, body := request(t, s, "POST", "/test/calls", []byte(`{"event_id":"`+sourceID+`","task_id":"test-1","callee":"123456"}`))
	if code != 200 {
		t.Fatalf("publish %d %s", code, body)
	}
	until := time.NewTimer(15 * time.Second)
	defer until.Stop()
	for {
		records := s.events(sourceID)
		if len(records) >= 2 {
			seen := map[string]bool{}
			for _, record := range records {
				if record.Validation != "valid" {
					t.Fatalf("feedback: %+v", record)
				}
				seen[record.EventType] = true
			}
			if !seen["call.execute"] || !seen["call.execute.result"] {
				t.Fatalf("missing receipt or result: %+v", records)
			}
			break
		}
		select {
		case e := <-workerFailure:
			t.Fatal(e)
		case <-until.C:
			t.Fatalf("feedback timeout: %+v", records)
		case <-time.After(20 * time.Millisecond):
		}
	}
	for path, want := range before {
		got, e := os.ReadFile(path)
		if e != nil || !bytes.Equal(want, got) {
			t.Fatalf("input changed: %s %v", path, e)
		}
	}
	files, err := os.ReadDir(s.feedbackDir)
	if err != nil || len(files) < 2 {
		t.Fatalf("feedback evidence missing: %v %v", files, err)
	}
	t.Log("Fiber HTTPS → RabbitMQ → unchanged Dispatcher → Mock Agent mTLS → original-ID receipt/result verified; no SIP or real providers")
}