fix(exec): proxy Cloudflare asset upload sessions

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
 `<subject>@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.

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

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"})

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
 		}

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