From 1dee661f0796734b84b37909e3d6bec4c7e047c1 Mon Sep 17 00:00:00 2001 From: Codex <019feeeb-6504-74ec-bfdc-da5259f73fc0@agents.realmroot.dev> Date: Tue, 11 Aug 2026 11:25:23 -0400 Subject: [PATCH] fix(exec): proxy Cloudflare asset upload sessions --- README.md | 5 +- internal/execution/broker.go | 160 ++++++++++++++++++++++++++---- internal/execution/broker_test.go | 69 +++++++++++++ internal/execution/run.go | 9 +- specs/cli.feature | 1 + 5 files changed, 217 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index 82c3d76..3db147f 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,10 @@ GitHub execution supports REST, GraphQL, and Git Smart HTTP. Git commits made through `realmroot exec github -- git ...` use the stable Agent name and `@agents.realmroot.dev` email without changing repository or global Git configuration. Cloudflare execution redirects Wrangler's API base to the -Cloudflare Resource Server and retains its native subcommands. +Cloudflare Resource Server and retains its native subcommands. When Cloudflare +returns a short-lived asset-upload credential, the broker keeps it in memory +and accepts it only for that account's asset-upload path during the same exec +session. `exec` consumes the exact active credential binding created by `realmroot agent request`; it never opens approval, requests access, or expands scopes. diff --git a/internal/execution/broker.go b/internal/execution/broker.go index 30957c0..c24ed7e 100644 --- a/internal/execution/broker.go +++ b/internal/execution/broker.go @@ -4,6 +4,8 @@ import ( "context" "crypto/rand" "encoding/hex" + "encoding/json" + "errors" "fmt" "io" "net" @@ -12,6 +14,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" restish "github.com/saltbo/restish/v2" @@ -33,6 +36,13 @@ type Broker struct { listener net.Listener temporaryDirectory string sessionToken string + cloudflareSession cloudflareAssetSession + cloudflareMu sync.RWMutex +} + +type cloudflareAssetSession struct { + account string + token string } func NewBroker(resource, reference string, scopes []string, source credentialSource, client *http.Client) (*Broker, error) { @@ -63,6 +73,20 @@ func (b *Broker) StartTCP(mapPath func(*http.Request) (string, error), authorize return "http://" + listener.Addr().String(), nil } +func (b *Broker) StartCloudflareAPIBase(providerBase string) (string, error) { + provider, err := url.ParseRequestURI(providerBase) + if err != nil || provider.Scheme == "" || provider.Host == "" || provider.RawQuery != "" || provider.Fragment != "" { + return "", errors.New("invalid Cloudflare provider API base URL") + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + return "", err + } + b.listener = listener + b.serve(http.HandlerFunc(b.cloudflareHandler(strings.TrimSuffix(providerBase, "/")))) + return "http://" + listener.Addr().String(), nil +} + func (b *Broker) StartGitHubSocket() (string, error) { directory, err := os.MkdirTemp("", "realmroot-exec-gh-*") if err != nil { @@ -133,40 +157,140 @@ func (b *Broker) handler(mapPath func(*http.Request) (string, error), authorize http.Error(response, "invalid upstream target", http.StatusBadGateway) return } - upstream, err := http.NewRequestWithContext(request.Context(), request.Method, target.String(), request.Body) + result, err := b.realmrootRequest(request, target.String()) if err != nil { - http.Error(response, "invalid upstream request", http.StatusBadGateway) + http.Error(response, err.Error(), http.StatusBadGateway) return } - upstream.Header = forwardedHeaders(request.Header) - err = b.auth.Authenticate(request.Context(), upstream, restish.AuthContext{ - APIName: "exec", ProfileName: "default", BaseURL: b.resource, CacheKey: "exec:" + b.reference, - Params: map[string]string{"source": "realmroot", "reference": b.reference, "scopes": strings.Join(b.scopes, " ")}, - TokenStore: b.store, HTTPClient: b.client, Stderr: io.Discard, - }) - if err != nil { - http.Error(response, err.Error(), http.StatusBadGateway) + defer result.Body.Close() + writeResponse(response, result) + } +} + +func (b *Broker) cloudflareHandler(providerBase string) http.HandlerFunc { + return func(response http.ResponseWriter, request *http.Request) { + if !strings.HasPrefix(request.URL.Path, "/client/v4/") { + http.Error(response, "Wrangler request is outside Cloudflare API v4", http.StatusBadRequest) return } - result, err := b.client.Do(upstream) + path := strings.TrimPrefix(request.URL.RequestURI(), "/client/v4") + authorization := request.Header.Get("Authorization") + if authorization == "Bearer "+b.sessionToken { + result, err := b.realmrootRequest(request, b.resource+path) + if err != nil { + http.Error(response, err.Error(), http.StatusBadGateway) + return + } + defer result.Body.Close() + account, capturesAssetSession := cloudflareAssetSessionAccount(request.URL.Path) + if capturesAssetSession && result.StatusCode >= 200 && result.StatusCode < 300 { + body, err := io.ReadAll(io.LimitReader(result.Body, 1<<20+1)) + if err != nil || len(body) > 1<<20 { + http.Error(response, "invalid Cloudflare asset upload session response", http.StatusBadGateway) + return + } + var document struct { + Result struct { + JWT string `json:"jwt"` + } `json:"result"` + } + if json.Unmarshal(body, &document) != nil || document.Result.JWT == "" { + http.Error(response, "invalid Cloudflare asset upload session response", http.StatusBadGateway) + return + } + b.cloudflareMu.Lock() + b.cloudflareSession = cloudflareAssetSession{account: account, token: document.Result.JWT} + b.cloudflareMu.Unlock() + writeResponseBytes(response, result, body) + return + } + writeResponse(response, result) + return + } + + account, isAssetUpload := cloudflareAssetUploadAccount(request.URL.Path) + b.cloudflareMu.RLock() + session := b.cloudflareSession + b.cloudflareMu.RUnlock() + if !isAssetUpload || account != session.account || authorization != "Bearer "+session.token { + http.Error(response, "unauthorized", http.StatusUnauthorized) + return + } + result, err := b.providerRequest(request, providerBase+path, authorization) if err != nil { http.Error(response, err.Error(), http.StatusBadGateway) return } defer result.Body.Close() - for name, values := range result.Header { - for _, value := range values { - response.Header().Add(name, value) - } + writeResponse(response, result) + } +} + +func (b *Broker) realmrootRequest(request *http.Request, target string) (*http.Response, error) { + upstream, err := http.NewRequestWithContext(request.Context(), request.Method, target, request.Body) + if err != nil { + return nil, errors.New("invalid upstream request") + } + upstream.Header = forwardedHeaders(request.Header) + if err := b.auth.Authenticate(request.Context(), upstream, restish.AuthContext{ + APIName: "exec", ProfileName: "default", BaseURL: b.resource, CacheKey: "exec:" + b.reference, + Params: map[string]string{"source": "realmroot", "reference": b.reference, "scopes": strings.Join(b.scopes, " ")}, + TokenStore: b.store, HTTPClient: b.client, Stderr: io.Discard, + }); err != nil { + return nil, err + } + return b.client.Do(upstream) +} + +func (b *Broker) providerRequest(request *http.Request, target, authorization string) (*http.Response, error) { + upstream, err := http.NewRequestWithContext(request.Context(), request.Method, target, request.Body) + if err != nil { + return nil, errors.New("invalid provider request") + } + upstream.Header = forwardedHeaders(request.Header) + upstream.Header.Set("Authorization", authorization) + return b.client.Do(upstream) +} + +func cloudflareAssetSessionAccount(path string) (string, bool) { + parts := strings.Split(strings.Trim(path, "/"), "/") + if len(parts) == 8 && parts[0] == "client" && parts[1] == "v4" && parts[2] == "accounts" && parts[4] == "workers" && parts[5] == "scripts" && parts[7] == "assets-upload-session" { + return parts[3], parts[3] != "" && parts[6] != "" + } + return "", false +} + +func cloudflareAssetUploadAccount(path string) (string, bool) { + parts := strings.Split(strings.Trim(path, "/"), "/") + if (len(parts) == 7 || len(parts) == 8) && parts[0] == "client" && parts[1] == "v4" && parts[2] == "accounts" && parts[4] == "workers" && parts[5] == "assets" && parts[6] == "upload" { + return parts[3], parts[3] != "" && (len(parts) == 7 || parts[7] != "") + } + return "", false +} + +func writeResponse(response http.ResponseWriter, result *http.Response) { + copyResponseHeaders(response, result) + response.WriteHeader(result.StatusCode) + _, _ = io.Copy(response, result.Body) +} + +func writeResponseBytes(response http.ResponseWriter, result *http.Response, body []byte) { + copyResponseHeaders(response, result) + response.WriteHeader(result.StatusCode) + _, _ = response.Write(body) +} + +func copyResponseHeaders(response http.ResponseWriter, result *http.Response) { + for name, values := range result.Header { + for _, value := range values { + response.Header().Add(name, value) } - response.WriteHeader(result.StatusCode) - _, _ = io.Copy(response, result.Body) } } func forwardedHeaders(source http.Header) http.Header { result := source.Clone() - for _, name := range []string{"Authorization", "Dpop", "Host", "Connection", "Proxy-Connection", "Keep-Alive", "Transfer-Encoding", "Upgrade", "Cookie"} { + for _, name := range []string{"Authorization", "Dpop", "Host", "Connection", "Proxy-Connection", "Keep-Alive", "Transfer-Encoding", "Upgrade", "Cookie", "Accept-Encoding"} { result.Del(name) } return result diff --git a/internal/execution/broker_test.go b/internal/execution/broker_test.go index 50243c6..34b018a 100644 --- a/internal/execution/broker_test.go +++ b/internal/execution/broker_test.go @@ -1,6 +1,7 @@ package execution import ( + "compress/gzip" "context" "io" "net/http" @@ -53,6 +54,74 @@ func TestNativeResourceToolUsesEphemeralDPoPBroker(t *testing.T) { } } +func TestCloudflareBrokerUsesCapturedAssetUploadCredentialOnlyForItsSession(t *testing.T) { + t.Parallel() + var adapterRequests, providerRequests int + adapter := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + adapterRequests++ + if request.Header.Get("Authorization") != "DPoP resource-token" || request.Header.Get("DPoP") == "" { + t.Fatalf("adapter request did not use the Agent credential") + } + response.Header().Set("Content-Type", "application/json") + response.Header().Set("Content-Encoding", "gzip") + compressed := gzip.NewWriter(response) + _, _ = compressed.Write([]byte(`{"result":{"jwt":"asset-session"}}`)) + _ = compressed.Close() + })) + defer adapter.Close() + provider := httptest.NewServer(http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) { + providerRequests++ + if request.URL.Path != "/client/v4/accounts/account-1/workers/assets/upload" || request.URL.RawQuery != "base64=true" { + t.Fatalf("provider target = %s", request.URL.String()) + } + if request.Header.Get("Authorization") != "Bearer asset-session" { + t.Fatalf("provider auth = %q", request.Header.Get("Authorization")) + } + response.WriteHeader(http.StatusNoContent) + })) + defer provider.Close() + source := &fakeSource{resource: adapter.URL} + broker, err := NewBroker(adapter.URL, "rrcs_reference", []string{"workers-scripts.write"}, source, adapter.Client()) + if err != nil { + t.Fatal(err) + } + defer broker.Close() + base, err := broker.StartCloudflareAPIBase(provider.URL + "/client/v4") + if err != nil { + t.Fatal(err) + } + + sessionRequest, _ := http.NewRequest(http.MethodPost, base+"/client/v4/accounts/account-1/workers/scripts/wallet/assets-upload-session", nil) + sessionRequest.Header.Set("Authorization", "Bearer "+broker.SessionToken()) + sessionResponse, err := http.DefaultClient.Do(sessionRequest) + if err != nil { + t.Fatal(err) + } + sessionResponse.Body.Close() + + uploadRequest, _ := http.NewRequest(http.MethodPost, base+"/client/v4/accounts/account-1/workers/assets/upload?base64=true", strings.NewReader("asset")) + uploadRequest.Header.Set("Authorization", "Bearer asset-session") + uploadResponse, err := http.DefaultClient.Do(uploadRequest) + if err != nil { + t.Fatal(err) + } + uploadResponse.Body.Close() + if uploadResponse.StatusCode != http.StatusNoContent || adapterRequests != 1 || providerRequests != 1 { + t.Fatalf("status = %d, adapter requests = %d, provider requests = %d", uploadResponse.StatusCode, adapterRequests, providerRequests) + } + + wrongPath, _ := http.NewRequest(http.MethodGet, base+"/client/v4/accounts/account-1/workers/scripts", nil) + wrongPath.Header.Set("Authorization", "Bearer asset-session") + wrongResponse, err := http.DefaultClient.Do(wrongPath) + if err != nil { + t.Fatal(err) + } + wrongResponse.Body.Close() + if wrongResponse.StatusCode != http.StatusUnauthorized || adapterRequests != 1 || providerRequests != 1 { + t.Fatalf("wrong-path status = %d, adapter requests = %d, provider requests = %d", wrongResponse.StatusCode, adapterRequests, providerRequests) + } +} + func TestNativeResourceToolRejectsUnadvertisedExecutables(t *testing.T) { t.Parallel() _, _, err := selectIntegration([]catalog.ToolIntegration{{ID: "git", Executables: []string{"git"}, Protocol: "git-smart-http"}}, []string{"curl"}) diff --git a/internal/execution/run.go b/internal/execution/run.go index 94d75b4..e9a9874 100644 --- a/internal/execution/run.go +++ b/internal/execution/run.go @@ -48,14 +48,7 @@ func (r *Runner) Run(ctx context.Context, server catalog.ResourceServer, integra environment := cleanEnvironment(os.Environ(), providerCredentialNames(integration.ID)) switch integration.Protocol { case "cloudflare-api-base": - base, err := broker.StartTCP(func(request *http.Request) (string, error) { - if !strings.HasPrefix(request.URL.Path, "/client/v4") { - return "", errors.New("Wrangler request is outside Cloudflare API v4") - } - return strings.TrimPrefix(request.URL.RequestURI(), "/client/v4"), nil - }, func(request *http.Request) bool { - return request.Header.Get("Authorization") == "Bearer "+broker.SessionToken() - }) + base, err := broker.StartCloudflareAPIBase("https://api.cloudflare.com/client/v4") if err != nil { return err } diff --git a/specs/cli.feature b/specs/cli.feature index 8dfe716..3e08a62 100644 --- a/specs/cli.feature +++ b/specs/cli.feature @@ -60,3 +60,4 @@ Feature: Realmroot Toolbox command line When it runs Wrangler through "realmroot exec cloudflare" Then Wrangler API traffic is routed through the Cloudflare Resource Server And existing Cloudflare credentials are removed from the child environment + And Cloudflare asset-upload credentials remain process-local and are accepted only for their matching upload session