Assemble isolated Agent session and recording worker

2026-09-30 05:01:26 +08:00
parent 2b09ba30ca
commit 99f8f5d721
3 changed files with 245 additions and 0 deletions
@@ -0,0 +1,123 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"log"
+	"maps"
+	"net"
+	"net/http"
+	"os"
+	"reflect"
+	"slices"
+	"strings"
+	"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/config"
+	"git.ipao.vip/rogee/go-sip/internal/rpc"
+	"github.com/google/uuid"
+)
+
+// newCurrentAgentServer binds the authenticated Agent session, task controls
+// and per-call Mock recording delivery. Serving the returned server requires
+// a separately verified mutual-TLS listener and a pinned local D connection.
+func newCurrentAgentServer(ctx context.Context, settings config.AgentEnvironment, scenario approvedMockScenario, appliedSIP map[string]int64, dispatcher agentpb.AgentControlServiceClient) (*rpc.Server, error) {
+	if ctx == nil || ctx.Err() != nil || settings.AgentID == "" || settings.CellID == "" || settings.SessionPath == "" ||
+		settings.RecoveryRoot == "" || len(settings.PeerFingerprints) == 0 || len(appliedSIP) == 0 ||
+		scenario.MaxWAVBytes <= 44 || len(scenario.Script.Turns) == 0 || strings.TrimSpace(scenario.ReasonMessage) == "" {
+		return nil, errors.New("current Agent requires an active process, explicit Mock media and deployment identity")
+	}
+	if dispatcher == nil {
+		return nil, errors.New("current Agent requires a pinned Dispatcher transport")
+	}
+	value := reflect.ValueOf(dispatcher)
+	switch value.Kind() {
+	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice:
+		if value.IsNil() {
+			return nil, errors.New("current Agent requires a pinned Dispatcher transport")
+		}
+	}
+	root, err := os.Stat(settings.RecoveryRoot)
+	if err != nil || !root.IsDir() || root.Mode().Perm() != 0700 {
+		return nil, errors.New("current Agent requires an existing private 0700 recovery directory")
+	}
+	for trunk, revision := range appliedSIP {
+		if strings.TrimSpace(trunk) == "" || revision <= 0 {
+			return nil, errors.New("current Agent requires explicit applied Mock SIP revisions")
+		}
+	}
+	loaded := maps.Clone(appliedSIP)
+	pins := maps.Clone(settings.PeerFingerprints)
+	scenario.InboundPCM16 = bytes.Clone(scenario.InboundPCM16)
+	scenario.Script.OpeningPCM16 = bytes.Clone(scenario.Script.OpeningPCM16)
+	scenario.Script.Turns = slices.Clone(scenario.Script.Turns)
+	for index := range scenario.Script.Turns {
+		scenario.Script.Turns[index].ReplyPCM16 = bytes.Clone(scenario.Script.Turns[index].ReplyPCM16)
+	}
+	uploadHTTP := localMockHTTPClient()
+	var handler *rpc.Server
+	worker := &rpc.ApprovedCallWorker{
+		Lifecycle: ctx,
+		Calls:     &agent.TaskCalls{},
+		Prepare: func(execution rpc.ApprovedExecution) (func(context.Context) error, error) {
+			if handler == nil {
+				return nil, errors.New("Agent session is unavailable")
+			}
+			delivery := &agent.RecordingDelivery{
+				Call: agent.RecordingClient{
+					Client: dispatcher, DispatcherID: execution.DispatcherID, TenantID: execution.TenantID,
+					SourceEventID: execution.SourceEventID, Session: func(context.Context) (*agentpb.RequestMeta, error) { return handler.ActiveSessionMeta() },
+				},
+				Recovery: &agent.RecordingRecovery{
+					Root:   settings.RecoveryRoot,
+					Upload: agent.UploadClient{HTTPClient: uploadHTTP, AllowInsecureHTTP: true},
+				},
+			}
+			mock := &rpc.ApprovedRecordedMockCall{
+				InboundPCM16: scenario.InboundPCM16, Script: scenario.Script,
+				MaxWAVBytes: scenario.MaxWAVBytes, ExpectedRecording: scenario.ExpectedRecording,
+				Outcome: scenario.Outcome, ReasonMessage: scenario.ReasonMessage,
+				ReportTimeout: 15 * time.Minute, Delivery: delivery,
+			}
+			return mock.Prepare(execution)
+		},
+		OnFailure: func(execution rpc.ApprovedExecution, cause error) error {
+			// The runner already tried to report termination and persisted any
+			// failed upload. Never invent a second result or retry an unknown PUT.
+			log.Printf("Agent Mock call requires inspection: event_id=%q task_id=%q cause_type=%T", execution.SourceEventID, execution.TaskID, cause)
+			return nil
+		},
+	}
+	handler, err = rpc.NewApprovedAgentServer(rpc.ServerOptions{
+		Mode: "mock", StatePath: settings.SessionPath,
+		Status:                      &agentpb.AgentStatus{AgentId: settings.AgentID, CellId: settings.CellID, BootId: uuid.NewString(), ProtocolVersion: "agent.v1"},
+		LoadedSIP:                   func(context.Context) (map[string]int64, error) { return maps.Clone(loaded), nil },
+		PeerCertificateFingerprints: pins, RequirePeerCertificate: true,
+	}, worker)
+	if err != nil {
+		return nil, err
+	}
+	return handler, nil
+}
+
+// The isolated Mock uploader may reach localhost only, even if a grant or
+// redirect unexpectedly names a real OSS endpoint. It never logs signed URLs.
+func localMockHTTPClient() *http.Client {
+	transport := http.DefaultTransport.(*http.Transport).Clone()
+	transport.Proxy = nil
+	transport.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) {
+		host, _, err := net.SplitHostPort(address)
+		if err != nil {
+			return nil, errors.New("Mock upload target is not local")
+		}
+		ip := net.ParseIP(host)
+		if !strings.EqualFold(host, "localhost") && (ip == nil || !ip.IsLoopback()) {
+			return nil, errors.New("Mock upload target is not local")
+		}
+		return (&net.Dialer{}).DialContext(ctx, network, address)
+	}
+	return &http.Client{Transport: transport, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
+}
@@ -0,0 +1,121 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
+	"git.ipao.vip/rogee/go-sip/internal/ai"
+	"git.ipao.vip/rogee/go-sip/internal/config"
+)
+
+type isolatedAgentRecordingClient struct {
+	agentpb.AgentControlServiceClient
+}
+
+func currentAgentSetupFixture(t *testing.T) (config.AgentEnvironment, approvedMockScenario) {
+	t.Helper()
+	root := t.TempDir()
+	if err := os.Chmod(root, 0700); err != nil {
+		t.Fatal(err)
+	}
+	settings := config.AgentEnvironment{
+		AgentID: "agent-mock", CellID: "cell-mock", Listen: "127.0.0.1:0",
+		SessionPath: filepath.Join(root, "session.json"), RecoveryRoot: root,
+		DispatcherEndpoint: "127.0.0.1:39443", DispatcherServerName: "dispatcher.local",
+		CAFile: filepath.Join(root, "ca.pem"), CertFile: filepath.Join(root, "agent.pem"),
+		KeyFile: filepath.Join(root, "agent.key"), PeerFingerprints: map[string]struct{}{strings.Repeat("a", 64): {}},
+	}
+	scenario := approvedMockScenario{
+		InboundPCM16: bytes.Repeat([]byte{1, 0}, 1600),
+		Script:       ai.ApprovedMockScript{Turns: []ai.ApprovedMockTurn{{Transcript: "synthetic ASR fixture"}}},
+		MaxWAVBytes:  4096, ExpectedRecording: true, Outcome: "answered", ReasonMessage: "isolated Mock answered",
+	}
+	return settings, scenario
+}
+
+func TestNewCurrentAgentServerBindsMockCallsToOneSessionAndRecoveryRoot(t *testing.T) {
+	settings, scenario := currentAgentSetupFixture(t)
+	server, err := newCurrentAgentServer(context.Background(), settings, scenario, map[string]int64{"trunk-mock": 8}, &isolatedAgentRecordingClient{})
+	if err != nil || server == nil {
+		t.Fatalf("valid isolated Agent could not be assembled: %v", err)
+	}
+	if _, err := os.Stat(settings.SessionPath); !os.IsNotExist(err) {
+		t.Fatalf("server assembly opened durable session before activation: %v", err)
+	}
+}
+
+func TestNewCurrentAgentServerRefusesUnsafeAdaptersBeforeResources(t *testing.T) {
+	settings, scenario := currentAgentSetupFixture(t)
+	for _, tc := range []struct {
+		name   string
+		change func(*config.AgentEnvironment, *approvedMockScenario, *map[string]int64, *agentpb.AgentControlServiceClient)
+	}{
+		{"missing Agent identity", func(c *config.AgentEnvironment, _ *approvedMockScenario, _ *map[string]int64, _ *agentpb.AgentControlServiceClient) {
+			c.AgentID = ""
+		}},
+		{"missing SIP evidence", func(_ *config.AgentEnvironment, _ *approvedMockScenario, sip *map[string]int64, _ *agentpb.AgentControlServiceClient) {
+			*sip = nil
+		}},
+		{"unknown SIP revision", func(_ *config.AgentEnvironment, _ *approvedMockScenario, sip *map[string]int64, _ *agentpb.AgentControlServiceClient) {
+			(*sip)["trunk-mock"] = 0
+		}},
+		{"missing recovery", func(c *config.AgentEnvironment, _ *approvedMockScenario, _ *map[string]int64, _ *agentpb.AgentControlServiceClient) {
+			c.RecoveryRoot = filepath.Join(t.TempDir(), "missing")
+		}},
+		{"missing scenario", func(_ *config.AgentEnvironment, s *approvedMockScenario, _ *map[string]int64, _ *agentpb.AgentControlServiceClient) {
+			s.MaxWAVBytes = 0
+		}},
+		{"missing Dispatcher transport", func(_ *config.AgentEnvironment, _ *approvedMockScenario, _ *map[string]int64, client *agentpb.AgentControlServiceClient) {
+			*client = nil
+		}},
+		{"typed-nil Dispatcher transport", func(_ *config.AgentEnvironment, _ *approvedMockScenario, _ *map[string]int64, client *agentpb.AgentControlServiceClient) {
+			var typed *isolatedAgentRecordingClient
+			*client = typed
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg, media := settings, scenario
+			loaded := map[string]int64{"trunk-mock": 8}
+			var client agentpb.AgentControlServiceClient = &isolatedAgentRecordingClient{}
+			tc.change(&cfg, &media, &loaded, &client)
+			if server, err := newCurrentAgentServer(context.Background(), cfg, media, loaded, client); err == nil || server != nil {
+				t.Fatalf("unsafe current Agent assembly was accepted: %v", err)
+			}
+			if _, err := os.Stat(settings.SessionPath); !os.IsNotExist(err) {
+				t.Fatalf("rejected Agent assembly wrote durable session: %v", err)
+			}
+		})
+	}
+}
+
+func TestLocalMockHTTPClientRefusesExternalAndRedirectedTargets(t *testing.T) {
+	client := localMockHTTPClient()
+	if response, err := client.Get("https://oss.example.invalid/approved"); err == nil || response != nil || !strings.Contains(err.Error(), "not local") {
+		t.Fatalf("Mock uploader contacted an external target: %v", err)
+	}
+	local := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/redirect" {
+			http.Redirect(w, r, "https://oss.example.invalid/approved", http.StatusFound)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer local.Close()
+	response, err := client.Get(local.URL + "/ok")
+	if err != nil || response.StatusCode != http.StatusOK {
+		t.Fatalf("local Mock endpoint was refused: response=%v err=%v", response, err)
+	}
+	_ = response.Body.Close()
+	response, err = client.Get(local.URL + "/redirect")
+	if err != nil || response.StatusCode != http.StatusFound {
+		t.Fatalf("Mock uploader followed an external redirect: response=%v err=%v", response, err)
+	}
+	_ = response.Body.Close()
+}