From 2f5e0a3fa167bb125ed020cdf9376eef0a4b5ba5 Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Sun, 3 May 2026 11:22:09 +0800 Subject: [PATCH] fix: route openclaw gateway through bridge --- internal/acp/gateway.go | 4 +- internal/acp/http_handler.go | 78 +++++++++++++ internal/acp/orchestrator.go | 176 +++++++++++++++++++++++++++--- internal/acp/routing_test.go | 17 +++ internal/acp/web_contract_test.go | 98 +++++++++++++++-- 5 files changed, 347 insertions(+), 26 deletions(-) diff --git a/internal/acp/gateway.go b/internal/acp/gateway.go index 9b8702d..776eab3 100644 --- a/internal/acp/gateway.go +++ b/internal/acp/gateway.go @@ -197,12 +197,12 @@ func ensureProductionGatewayConnected( gatewayruntime.ConnectRequest{ RuntimeID: "xworkmate-bridge-openclaw", Mode: "openclaw", - ClientID: "xworkmate-bridge", + ClientID: "openclaw-macos", Locale: "en_US", UserAgent: "xworkmate-bridge", Endpoint: gatewayruntime.Endpoint{Host: "127.0.0.1", Port: 18789, TLS: false}, PackageInfo: gatewayruntime.PackageInfo{AppName: "XWorkmate Bridge", PackageName: "xworkmate-bridge", Version: "bridge", BuildNumber: "0"}, - DeviceInfo: gatewayruntime.DeviceInfo{Platform: "linux", DeviceFamily: "bridge"}, + DeviceInfo: gatewayruntime.DeviceInfo{Platform: "macos", DeviceFamily: "Mac", ModelIdentifier: "Mac14,5"}, Identity: newBridgeGatewayIdentity(), }, ) diff --git a/internal/acp/http_handler.go b/internal/acp/http_handler.go index 7358375..4d3591b 100644 --- a/internal/acp/http_handler.go +++ b/internal/acp/http_handler.go @@ -41,6 +41,14 @@ func (s *Server) Handler() http.Handler { case "/acp": s.HandleWebSocket(w, r) default: + if r.URL.Path == "/gateway/openclaw" { + s.HandleOpenClawGatewayRPC(w, r) + return + } + if strings.HasPrefix(r.URL.Path, "/acp-server/") { + s.HandleDisabledProviderDirectPath(w, r) + return + } http.NotFound(w, r) } }) @@ -97,6 +105,41 @@ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request) { } func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) { + s.handleRPCWithTransform(w, r, nil) +} + +func (s *Server) HandleOpenClawGatewayRPC(w http.ResponseWriter, r *http.Request) { + s.handleRPCWithTransform(w, r, forceOpenClawGatewayRequest) +} + +func (s *Server) HandleDisabledProviderDirectPath(w http.ResponseWriter, r *http.Request) { + shared.ApplyCORS(w, r, s.allowedOrigins) + if r.Method == http.MethodOptions { + w.WriteHeader(http.StatusNoContent) + return + } + if !s.authorized(r) { + shared.WriteJSONError(w, nil, http.StatusUnauthorized, -32001, "missing bearer authorization") + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusGone) + _ = json.NewEncoder(w).Encode(map[string]any{ + "jsonrpc": "2.0", + "error": map[string]any{ + "code": -32004, + "message": "PROVIDER_DIRECT_PATH_DISABLED: use /acp/rpc provider catalog and routing", + }, + "type": "res", + "ok": false, + }) +} + +func (s *Server) handleRPCWithTransform( + w http.ResponseWriter, + r *http.Request, + transform func(shared.RPCRequest) (shared.RPCRequest, *shared.RPCError), +) { shared.ApplyCORS(w, r, s.allowedOrigins) if r.Method == http.MethodOptions { w.WriteHeader(http.StatusNoContent) @@ -128,6 +171,15 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) { return } request.Params = injectInboundAuthorizationHeader(request.Params, r.Header.Get("Authorization")) + if transform != nil { + transformed, rpcErr := transform(request) + if rpcErr != nil { + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(shared.ErrorEnvelope(request.ID, rpcErr.Code, rpcErr.Message)) + return + } + request = transformed + } accept := strings.ToLower(r.Header.Get("Accept")) stream := strings.Contains(accept, "text/event-stream") @@ -182,6 +234,32 @@ func (s *Server) HandleRPC(w http.ResponseWriter, r *http.Request) { _ = json.NewEncoder(w).Encode(shared.ResultEnvelope(request.ID, response)) } +func forceOpenClawGatewayRequest(request shared.RPCRequest) (shared.RPCRequest, *shared.RPCError) { + method := strings.TrimSpace(request.Method) + switch method { + case "session.start", "session.message", "session.cancel", "session.close": + default: + return request, &shared.RPCError{Code: -32601, Message: "OPENCLAW_GATEWAY_METHOD_NOT_ALLOWED: " + method} + } + params := shared.AsMap(request.Params) + if params == nil { + params = map[string]any{} + } + routing := shared.AsMap(params["routing"]) + if routing == nil { + routing = map[string]any{} + } + routing["routingMode"] = "explicit" + routing["explicitExecutionTarget"] = "gateway" + routing["preferredGatewayProviderId"] = "openclaw" + delete(routing, "explicitProviderId") + params["routing"] = routing + params["requestedExecutionTarget"] = "gateway" + params["executionTarget"] = "gateway" + request.Params = params + return request, nil +} + func (s *Server) authorized(r *http.Request) bool { if s == nil { return false diff --git a/internal/acp/orchestrator.go b/internal/acp/orchestrator.go index ba5b526..7e7520d 100644 --- a/internal/acp/orchestrator.go +++ b/internal/acp/orchestrator.go @@ -2,7 +2,11 @@ package acp import ( "context" + "crypto/sha256" + "encoding/base64" "fmt" + "os" + "path/filepath" "strings" "time" @@ -323,7 +327,7 @@ func buildArtifactRecord(sess *session, result map[string]any, output string) Ar if record.RemoteWorkspaceRefKind == "" && record.RemoteWorkingDirectory != "" { record.RemoteWorkspaceRefKind = "remotePath" } - record.Artifacts = extractArtifactPayloads(result) + record.Artifacts = extractArtifactPayloads(result, record.RemoteWorkingDirectory) sess.mu.Lock() sess.artifacts = record sess.control.UpdatedAt = record.UpdatedAt @@ -331,26 +335,168 @@ func buildArtifactRecord(sess *session, result map[string]any, output string) Ar return record } -func extractArtifactPayloads(result map[string]any) []map[string]any { - rawArtifacts := result["artifacts"] - items, ok := rawArtifacts.([]any) - if !ok { - if typed, ok := rawArtifacts.([]map[string]any); ok { - copied := make([]map[string]any, 0, len(typed)) - copied = append(copied, typed...) - return copied +func extractArtifactPayloads(result map[string]any, remoteWorkingDirectory string) []map[string]any { + artifacts := make([]map[string]any, 0) + for _, key := range []string{"artifacts", "files", "attachments"} { + rawArtifacts := result[key] + items, ok := rawArtifacts.([]any) + if !ok { + if typed, ok := rawArtifacts.([]map[string]any); ok { + for _, item := range typed { + if artifact := normalizeArtifactPayload(item, remoteWorkingDirectory); len(artifact) > 0 { + artifacts = append(artifacts, artifact) + } + } + } + continue + } + for _, item := range items { + if mapped := shared.AsMap(item); len(mapped) > 0 { + if artifact := normalizeArtifactPayload(mapped, remoteWorkingDirectory); len(artifact) > 0 { + artifacts = append(artifacts, artifact) + } + } } - return nil } - artifacts := make([]map[string]any, 0, len(items)) - for _, item := range items { - if mapped := shared.AsMap(item); len(mapped) > 0 { - artifacts = append(artifacts, mapped) - } + if len(artifacts) == 0 { + artifacts = append(artifacts, collectDirectoryArtifacts(remoteWorkingDirectory)...) } return artifacts } +func normalizeArtifactPayload(item map[string]any, remoteWorkingDirectory string) map[string]any { + artifact := make(map[string]any, len(item)+4) + for key, value := range item { + artifact[key] = value + } + relativePath := strings.TrimSpace(shared.StringArg(artifact, "relativePath", "")) + if relativePath == "" { + relativePath = strings.TrimSpace(shared.StringArg(artifact, "path", "")) + } + if relativePath == "" { + relativePath = strings.TrimSpace(shared.StringArg(artifact, "name", "")) + } + relativePath = safeArtifactRelativePath(remoteWorkingDirectory, relativePath) + if relativePath == "" { + return nil + } + artifact["relativePath"] = relativePath + if strings.TrimSpace(shared.StringArg(artifact, "label", "")) == "" { + artifact["label"] = filepath.Base(relativePath) + } + if strings.TrimSpace(shared.StringArg(artifact, "contentType", "")) == "" { + artifact["contentType"] = artifactContentType(relativePath) + } + if strings.TrimSpace(shared.StringArg(artifact, "content", "")) == "" { + if filled := readArtifactFile(remoteWorkingDirectory, relativePath); len(filled) > 0 { + for key, value := range filled { + artifact[key] = value + } + } + } + return artifact +} + +func collectDirectoryArtifacts(root string) []map[string]any { + root = strings.TrimSpace(root) + if root == "" { + return nil + } + info, err := os.Stat(root) + if err != nil || !info.IsDir() { + return nil + } + artifacts := make([]map[string]any, 0) + _ = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { + if err != nil || len(artifacts) >= 64 { + return nil + } + name := d.Name() + if d.IsDir() { + if name == ".git" || name == ".dart_tool" || name == "build" || name == "node_modules" { + return filepath.SkipDir + } + return nil + } + relativePath, err := filepath.Rel(root, path) + if err != nil { + return nil + } + relativePath = filepath.ToSlash(relativePath) + if artifact := readArtifactFile(root, relativePath); len(artifact) > 0 { + artifact["relativePath"] = relativePath + artifact["label"] = filepath.Base(relativePath) + artifact["contentType"] = artifactContentType(relativePath) + artifacts = append(artifacts, artifact) + } + return nil + }) + return artifacts +} + +func readArtifactFile(root string, relativePath string) map[string]any { + root = strings.TrimSpace(root) + relativePath = safeArtifactRelativePath(root, relativePath) + if root == "" || relativePath == "" { + return nil + } + target := filepath.Clean(filepath.Join(root, filepath.FromSlash(relativePath))) + rootClean := filepath.Clean(root) + if target != rootClean && !strings.HasPrefix(target, rootClean+string(os.PathSeparator)) { + return nil + } + info, err := os.Stat(target) + if err != nil || info.IsDir() || info.Size() > 10*1024*1024 { + return nil + } + content, err := os.ReadFile(target) + if err != nil { + return nil + } + sum := sha256.Sum256(content) + return map[string]any{ + "encoding": "base64", + "content": base64.StdEncoding.EncodeToString(content), + "sizeBytes": len(content), + "sha256": fmt.Sprintf("%x", sum[:]), + } +} + +func safeArtifactRelativePath(root string, rawPath string) string { + path := strings.TrimSpace(rawPath) + if path == "" || strings.Contains(path, "\x00") { + return "" + } + path = filepath.ToSlash(path) + if strings.TrimSpace(root) != "" && filepath.IsAbs(path) { + rel, err := filepath.Rel(root, path) + if err != nil { + return "" + } + path = filepath.ToSlash(rel) + } + path = filepath.Clean(filepath.FromSlash(path)) + if path == "." || filepath.IsAbs(path) || strings.HasPrefix(path, ".."+string(os.PathSeparator)) || path == ".." { + return "" + } + return filepath.ToSlash(path) +} + +func artifactContentType(relativePath string) string { + switch strings.ToLower(filepath.Ext(relativePath)) { + case ".pdf": + return "application/pdf" + case ".pptx": + return "application/vnd.openxmlformats-officedocument.presentationml.presentation" + case ".docx": + return "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + case ".txt", ".md": + return "text/plain" + default: + return "application/octet-stream" + } +} + func (s *Server) getOrCreateSession(sessionID, threadID string) *session { s.mu.Lock() defer s.mu.Unlock() diff --git a/internal/acp/routing_test.go b/internal/acp/routing_test.go index 3d8846d..7e22f8a 100644 --- a/internal/acp/routing_test.go +++ b/internal/acp/routing_test.go @@ -489,6 +489,13 @@ func TestExecuteSessionTaskGatewayAutoConnectsLocalOpenClaw(t *testing.T) { if gateway.SessionStartCount() != 1 { t.Fatalf("expected one session.start request, got %d", gateway.SessionStartCount()) } + client := gateway.LastConnectClient() + if got := client["id"]; got != "openclaw-macos" { + t.Fatalf("expected OpenClaw-compatible client id, got %#v", client) + } + if got := strings.TrimSpace(shared.StringArg(client, "modelIdentifier", "")); got == "" { + t.Fatalf("expected non-empty modelIdentifier, got %#v", client) + } } func TestExecuteSessionTaskDefaultsExplicitGatewayToOpenClaw(t *testing.T) { @@ -521,6 +528,7 @@ type acpFakeOpenClawGateway struct { listener net.Listener connectCount atomic.Int32 sessionStartCount atomic.Int32 + lastConnectClient atomic.Value } func newAcpFakeOpenClawGateway(t *testing.T) *acpFakeOpenClawGateway { @@ -563,6 +571,7 @@ func newAcpFakeOpenClawGateway(t *testing.T) *acpFakeOpenClawGateway { switch strings.TrimSpace(shared.StringArg(frame, "method", "")) { case "connect": fake.connectCount.Add(1) + fake.lastConnectClient.Store(shared.AsMap(shared.AsMap(frame["params"])["client"])) _ = conn.WriteJSON(map[string]any{ "type": "res", "id": id, @@ -619,6 +628,14 @@ func (f *acpFakeOpenClawGateway) SessionStartCount() int { return int(f.sessionStartCount.Load()) } +func (f *acpFakeOpenClawGateway) LastConnectClient() map[string]any { + value := f.lastConnectClient.Load() + if value == nil { + return nil + } + return value.(map[string]any) +} + func (f *acpFakeOpenClawGateway) Close() { _ = f.server.Close() } diff --git a/internal/acp/web_contract_test.go b/internal/acp/web_contract_test.go index 837419d..d321ba6 100644 --- a/internal/acp/web_contract_test.go +++ b/internal/acp/web_contract_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "path/filepath" "reflect" "strings" "testing" @@ -61,32 +62,111 @@ func TestHTTPHandlerRootAndPingExposeRuntimeVersionInfo(t *testing.T) { } func TestHTTPHandlerRejectsLegacyACPCodexPath(t *testing.T) { - t.Setenv("BRIDGE_AUTH_TOKEN", "") + t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token") t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml") server := NewServer() handler := server.Handler() recorder := httptest.NewRecorder() - request := httptest.NewRequest(http.MethodGet, "http://127.0.0.1/acp-server/codex", nil) + request := httptest.NewRequest(http.MethodPost, "http://127.0.0.1/acp-server/codex", nil) + request.Header.Set("Authorization", "Bearer bridge-test-token") handler.ServeHTTP(recorder, request) - if recorder.Code != http.StatusNotFound { - t.Fatalf("expected 404, got %d", recorder.Code) + if recorder.Code != http.StatusGone { + t.Fatalf("expected 410, got %d", recorder.Code) + } + if !strings.Contains(recorder.Body.String(), "PROVIDER_DIRECT_PATH_DISABLED") { + t.Fatalf("expected disabled provider path error, got %q", recorder.Body.String()) } } -func TestHTTPHandlerRejectsGatewayOpenClawPublicAlias(t *testing.T) { - t.Setenv("BRIDGE_AUTH_TOKEN", "") +func TestHTTPHandlerProviderDirectPathRequiresAuthorization(t *testing.T) { + t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token") t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml") server := NewServer() handler := server.Handler() recorder := httptest.NewRecorder() - request := httptest.NewRequest(http.MethodPost, "http://127.0.0.1/gateway/openclaw", nil) + request := httptest.NewRequest(http.MethodPost, "http://127.0.0.1/acp-server/hermes", nil) handler.ServeHTTP(recorder, request) - if recorder.Code != http.StatusNotFound { - t.Fatalf("expected 404, got %d", recorder.Code) + if recorder.Code != http.StatusUnauthorized { + t.Fatalf("expected 401, got %d", recorder.Code) + } +} + +func TestHTTPHandlerGatewayOpenClawRequiresAuthorization(t *testing.T) { + t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token") + t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml") + server := NewServer() + handler := server.Handler() + + recorder := httptest.NewRecorder() + request := httptest.NewRequest( + http.MethodPost, + "http://127.0.0.1/gateway/openclaw", + strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"session.start","params":{"sessionId":"test"}}`), + ) + request.Header.Set("Content-Type", "application/json") + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusUnauthorized { + t.Fatalf("expected 401, got %d", recorder.Code) + } +} + +func TestHTTPHandlerGatewayOpenClawRejectsNonSessionMethods(t *testing.T) { + t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token") + t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml") + server := NewServer() + handler := server.Handler() + + recorder := httptest.NewRecorder() + request := httptest.NewRequest( + http.MethodPost, + "http://127.0.0.1/gateway/openclaw", + strings.NewReader(`{"jsonrpc":"2.0","id":1,"method":"acp.capabilities","params":{}}`), + ) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Authorization", "Bearer bridge-test-token") + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("expected JSON-RPC 200, got %d", recorder.Code) + } + if !strings.Contains(recorder.Body.String(), "OPENCLAW_GATEWAY_METHOD_NOT_ALLOWED") { + t.Fatalf("expected method allowlist error, got %q", recorder.Body.String()) + } +} + +func TestHTTPHandlerGatewayOpenClawForcesGatewayRouting(t *testing.T) { + gateway := newAcpFakeOpenClawGateway(t) + defer gateway.Close() + + t.Setenv("GATEWAY_RPC_URL", gateway.URL()) + t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token") + t.Setenv("BRIDGE_CONFIG_PATH", filepath.Join(t.TempDir(), "missing-config.yaml")) + server := NewServer() + handler := server.Handler() + + recorder := httptest.NewRecorder() + request := httptest.NewRequest( + http.MethodPost, + "http://127.0.0.1/gateway/openclaw", + strings.NewReader(`{"jsonrpc":"2.0","id":"task-1","method":"session.start","params":{"sessionId":"s1","threadId":"t1","taskPrompt":"Reply pong","workingDirectory":"`+t.TempDir()+`"}}`), + ) + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Authorization", "Bearer bridge-test-token") + handler.ServeHTTP(recorder, request) + + if recorder.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", recorder.Code, recorder.Body.String()) + } + if !strings.Contains(recorder.Body.String(), `"resolvedGatewayProviderId":"openclaw"`) { + t.Fatalf("expected forced OpenClaw gateway result, got %q", recorder.Body.String()) + } + if gateway.SessionStartCount() != 1 { + t.Fatalf("expected one OpenClaw session.start, got %d", gateway.SessionStartCount()) } }