From 780ab246cf3bca357a75c3bfa71c0f10403a5af2 Mon Sep 17 00:00:00 2001 From: Caio Pizzol <33255434+caiopizzol@users.noreply.github.com> Date: Wed, 19 Aug 2026 07:55:50 -0300 Subject: [PATCH 1/2] feat: add managed Cloudflare tunnel Keep tunnel setup with webhook destinations because the destination URL is the resource changed during local testing. Restore the original URL on shutdown so repeated tests do not leave project configuration pointing at an expired tunnel. --- cmd/tunnel.go | 275 +++++++++++++++++++++++++++++++++++++++++++++ cmd/tunnel_test.go | 132 ++++++++++++++++++++++ 2 files changed, 407 insertions(+) create mode 100644 cmd/tunnel.go create mode 100644 cmd/tunnel_test.go diff --git a/cmd/tunnel.go b/cmd/tunnel.go new file mode 100644 index 0000000..0e01a68 --- /dev/null +++ b/cmd/tunnel.go @@ -0,0 +1,275 @@ +package cmd + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "os/signal" + "regexp" + "syscall" + "time" + + "github.com/amp-labs/cli/flags" + "github.com/amp-labs/cli/logger" + "github.com/amp-labs/cli/request" + "github.com/spf13/cobra" +) + +const ( + defaultTunnelTarget = "http://127.0.0.1:4000" + tunnelStartTimeout = 30 * time.Second + tunnelStopTimeout = 10 * time.Second +) + +var ( + errCloudflaredNotFound = errors.New("cloudflared was not found in PATH") + errTunnelURLTimeout = errors.New("timed out waiting for the Cloudflare tunnel URL") + errTunnelStopped = errors.New("cloudflared stopped before the tunnel was ready") + cloudflareURLPattern = regexp.MustCompile(`https://[a-z0-9-]+\.trycloudflare\.com`) + tunnelTarget string + tunnelCommand = &cobra.Command{ + Use: "tunnel ", + Short: "Start a local tunnel for a webhook destination", + Long: "Start a Cloudflare tunnel, point one webhook destination to it, and restore the destination when stopped.", + Args: cobra.ExactArgs(1), + Hidden: true, + RunE: runTunnelCommand, + } +) + +type runningTunnel struct { + publicURL string + done <-chan error + stop context.CancelFunc +} + +type tunnelStarter func(context.Context, string) (*runningTunnel, error) + +func init() { + tunnelCommand.Flags().StringVar(&tunnelTarget, "target", defaultTunnelTarget, + "Local URL that receives tunneled requests") + rootCmd.AddCommand(tunnelCommand) +} + +func runTunnelCommand(cmd *cobra.Command, args []string) error { + ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer stop() + + projectId := flags.GetProjectOrFail() + apiKey := flags.GetAPIKey() + client := request.NewAPIClient(projectId, &apiKey) + + destination, err := getTunnelDestination(ctx, client, args[0]) + if err != nil { + return err + } + + return runDestinationTunnel(ctx, client, destination, tunnelTarget, startCloudflareTunnel) +} + +func getTunnelDestination( + ctx context.Context, client *request.APIClient, identifier string, +) (Destination, error) { + destinations, err := client.ListDestinations(ctx) + if err != nil { + return Destination{}, fmt.Errorf("list destinations: %w", err) + } + + nameMap, idMap := buildDestinationMaps(destinations) + + return resolveDestination(identifier, nameMap, idMap) +} + +func runDestinationTunnel( + ctx context.Context, + client *request.APIClient, + destination Destination, + targetURL string, + start tunnelStarter, +) error { + tunnel, err := start(ctx, targetURL) + if err != nil { + return err + } + + if tunnel.stop != nil { + defer tunnel.stop() + } + + temporaryURL, err := mergeURLs(tunnel.publicURL, destination.URL) + if err != nil { + return err + } + + err = ctx.Err() + if err != nil { + return err + } + + updateCtx, cancelUpdate := context.WithTimeout(context.WithoutCancel(ctx), tunnelStopTimeout) + err = setDestinationURL(updateCtx, client, destination.Id, temporaryURL) + + cancelUpdate() + + if err != nil { + return err + } + + logger.Infof("Tunnel ready: %s", temporaryURL) + logger.Info("Press Ctrl+C to stop and restore the destination") + + var tunnelErr error + + select { + case <-ctx.Done(): + case tunnelErr = <-tunnel.done: + } + + restoreCtx, cancelRestore := context.WithTimeout(context.WithoutCancel(ctx), tunnelStopTimeout) + defer cancelRestore() + + err = setDestinationURL(restoreCtx, client, destination.Id, destination.URL) + if err != nil { + return fmt.Errorf("restore destination URL: %w", err) + } + + logger.Infof("Restored destination: %s", destination.URL) + + if ctx.Err() == nil { + if tunnelErr == nil { + tunnelErr = errTunnelStopped + } + + return fmt.Errorf("cloudflared stopped: %w", tunnelErr) + } + + return nil +} + +func setDestinationURL( + ctx context.Context, client *request.APIClient, destinationId string, destinationURL string, +) error { + _, err := client.PatchDestination(ctx, destinationId, &request.PatchDestination{ + Destination: map[string]any{ + "metadata": map[string]any{ + "url": destinationURL, + }, + }, + UpdateMask: []string{"metadata.url"}, + }) + if err != nil { + return fmt.Errorf("update destination URL: %w", err) + } + + return nil +} + +func startCloudflareTunnel(ctx context.Context, targetURL string) (*runningTunnel, error) { + cloudflaredPath, err := exec.LookPath("cloudflared") + if err != nil { + return nil, errCloudflaredNotFound + } + + processCtx, stop := context.WithCancel(ctx) + // cloudflaredPath is resolved from PATH, and targetURL is passed as one argument without a shell. + //nolint:gosec + command := exec.CommandContext(processCtx, cloudflaredPath, + "tunnel", "--url", targetURL, "--no-autoupdate") + command.Stdout = io.Discard + + stderr, err := command.StderrPipe() + if err != nil { + stop() + + return nil, fmt.Errorf("read cloudflared output: %w", err) + } + + err = command.Start() + if err != nil { + stop() + + return nil, fmt.Errorf("start cloudflared: %w", err) + } + + lines := scanLines(stderr) + done := make(chan error, 1) + + go func() { + done <- command.Wait() + }() + + publicURL, err := waitForCloudflareTunnelURL(processCtx, lines, done) + if err != nil { + stop() + + return nil, err + } + + go discardTunnelOutput(lines) + + return &runningTunnel{ + publicURL: publicURL, + done: done, + stop: stop, + }, nil +} + +func scanLines(reader io.Reader) <-chan string { + lines := make(chan string) + + go func() { + defer close(lines) + + scanner := bufio.NewScanner(reader) + for scanner.Scan() { + lines <- scanner.Text() + } + }() + + return lines +} + +func waitForCloudflareTunnelURL( + ctx context.Context, lines <-chan string, done <-chan error, +) (string, error) { + timer := time.NewTimer(tunnelStartTimeout) + defer timer.Stop() + + for { + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-timer.C: + return "", errTunnelURLTimeout + case err := <-done: + if err == nil { + return "", errTunnelStopped + } + + return "", fmt.Errorf("%w: %w", errTunnelStopped, err) + case line, ok := <-lines: + if !ok { + lines = nil + + continue + } + + if publicURL := cloudflareTunnelURL(line); publicURL != "" { + return publicURL, nil + } + } + } +} + +func cloudflareTunnelURL(line string) string { + return cloudflareURLPattern.FindString(line) +} + +func discardTunnelOutput(lines <-chan string) { + for range lines { + } +} diff --git a/cmd/tunnel_test.go b/cmd/tunnel_test.go new file mode 100644 index 0000000..23784d2 --- /dev/null +++ b/cmd/tunnel_test.go @@ -0,0 +1,132 @@ +package cmd + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "reflect" + "sync" + "testing" + + "github.com/amp-labs/cli/request" +) + +func TestRunDestinationTunnelRestoresOriginalURL(t *testing.T) { + t.Parallel() + + var ( + patchesMu sync.Mutex + patches []*request.PatchDestination + ) + + updated := make(chan struct{}, 1) + + server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodPatch || + req.URL.Path != "/v1/projects/test-project/destinations/destination-id" { + http.NotFound(writer, req) + + return + } + + var patch request.PatchDestination + + err := json.NewDecoder(req.Body).Decode(&patch) + if err != nil { + t.Errorf("decode patch: %v", err) + + return + } + + patchesMu.Lock() + + patches = append(patches, &patch) + patchCount := len(patches) + patchesMu.Unlock() + + writer.Header().Set("Content-Type", "application/json") + _, _ = writer.Write([]byte(`{"id":"destination-id","name":"local-webhook","type":"webhook"}`)) + + if patchCount == 1 { + updated <- struct{}{} + } + })) + t.Cleanup(server.Close) + + apiKey := "test-key" + client := &request.APIClient{ + Root: server.URL + "/v1", + ProjectId: "test-project", + APIKey: &apiKey, + Client: request.NewRequestClient(), + } + destination := Destination{ + Id: "destination-id", + Name: "local-webhook", + URL: "https://original.example.com/webhook?source=test", + Type: "webhook", + } + + ctx, cancel := context.WithCancel(t.Context()) + result := make(chan error, 1) + tunnelDone := make(chan error) + + go func() { + result <- runDestinationTunnel(ctx, client, destination, "http://127.0.0.1:4000", + func(_ context.Context, targetURL string) (*runningTunnel, error) { + if targetURL != "http://127.0.0.1:4000" { + t.Errorf("target URL = %q", targetURL) + } + + return &runningTunnel{ + publicURL: "https://temporary.trycloudflare.com", + done: tunnelDone, + }, nil + }) + }() + + <-updated + cancel() + + err := <-result + if err != nil { + t.Fatalf("runDestinationTunnel() error = %v", err) + } + + patchesMu.Lock() + defer patchesMu.Unlock() + + if len(patches) != 2 { + t.Fatalf("patch count = %d, want 2", len(patches)) + } + + wantURLs := []string{ + "https://temporary.trycloudflare.com/webhook?source=test", + "https://original.example.com/webhook?source=test", + } + + for index, patch := range patches { + if !reflect.DeepEqual(patch.UpdateMask, []string{"metadata.url"}) { + t.Errorf("patch %d update mask = %#v", index, patch.UpdateMask) + } + + metadata, ok := patch.Destination["metadata"].(map[string]any) + if !ok { + t.Fatalf("patch %d metadata = %#v", index, patch.Destination["metadata"]) + } + + if metadata["url"] != wantURLs[index] { + t.Errorf("patch %d URL = %q, want %q", index, metadata["url"], wantURLs[index]) + } + } +} + +func TestCloudflareTunnelURL(t *testing.T) { + t.Parallel() + + line := "Your quick Tunnel has been created at https://small-test.trycloudflare.com" + if got := cloudflareTunnelURL(line); got != "https://small-test.trycloudflare.com" { + t.Fatalf("cloudflareTunnelURL() = %q", got) + } +} From d97c5988de91d4b1afcc9db76f542acd4b57105c Mon Sep 17 00:00:00 2001 From: Caio Pizzol <33255434+caiopizzol@users.noreply.github.com> Date: Wed, 19 Aug 2026 12:42:54 -0300 Subject: [PATCH 2/2] feat(tunnel): discover local listener --- cmd/tunnel.go | 54 ++++++++++++++++++++++++++++++++++++++-------- cmd/tunnel_test.go | 25 +++++++++++++++++++++ 2 files changed, 70 insertions(+), 9 deletions(-) diff --git a/cmd/tunnel.go b/cmd/tunnel.go index 0e01a68..94c8f11 100644 --- a/cmd/tunnel.go +++ b/cmd/tunnel.go @@ -9,7 +9,9 @@ import ( "os" "os/exec" "os/signal" + "path/filepath" "regexp" + "strings" "syscall" "time" @@ -27,17 +29,17 @@ const ( var ( errCloudflaredNotFound = errors.New("cloudflared was not found in PATH") + errListenerPortMissing = errors.New("listener port is empty") errTunnelURLTimeout = errors.New("timed out waiting for the Cloudflare tunnel URL") errTunnelStopped = errors.New("cloudflared stopped before the tunnel was ready") cloudflareURLPattern = regexp.MustCompile(`https://[a-z0-9-]+\.trycloudflare\.com`) tunnelTarget string tunnelCommand = &cobra.Command{ - Use: "tunnel ", - Short: "Start a local tunnel for a webhook destination", - Long: "Start a Cloudflare tunnel, point one webhook destination to it, and restore the destination when stopped.", - Args: cobra.ExactArgs(1), - Hidden: true, - RunE: runTunnelCommand, + Use: "tunnel ", + Short: "Start a local tunnel for a webhook destination", + Long: "Start a Cloudflare tunnel, point one webhook destination to it, and restore the destination when stopped.", + Args: cobra.ExactArgs(1), + RunE: runTunnelCommand, } ) @@ -50,8 +52,8 @@ type runningTunnel struct { type tunnelStarter func(context.Context, string) (*runningTunnel, error) func init() { - tunnelCommand.Flags().StringVar(&tunnelTarget, "target", defaultTunnelTarget, - "Local URL that receives tunneled requests") + tunnelCommand.Flags().StringVar(&tunnelTarget, "target", "", + "Local URL that receives tunneled requests (default: running listener or http://127.0.0.1:4000)") rootCmd.AddCommand(tunnelCommand) } @@ -68,7 +70,41 @@ func runTunnelCommand(cmd *cobra.Command, args []string) error { return err } - return runDestinationTunnel(ctx, client, destination, tunnelTarget, startCloudflareTunnel) + return runDestinationTunnel( + ctx, client, destination, resolveTunnelTarget(tunnelTarget, readSavedListenerPort), startCloudflareTunnel, + ) +} + +func resolveTunnelTarget(explicitTarget string, readPort func() (string, error)) string { + if explicitTarget != "" { + return explicitTarget + } + + port, err := readPort() + if err != nil { + return defaultTunnelTarget + } + + return "http://127.0.0.1:" + port +} + +func readSavedListenerPort() (string, error) { + dir, err := os.UserCacheDir() + if err != nil { + return "", fmt.Errorf("get user cache directory: %w", err) + } + + data, err := os.ReadFile(filepath.Join(dir, "ampersand", "webhook-port")) + if err != nil { + return "", fmt.Errorf("read listener port: %w", err) + } + + port := strings.TrimSpace(string(data)) + if port == "" { + return "", errListenerPortMissing + } + + return port, nil } func getTunnelDestination( diff --git a/cmd/tunnel_test.go b/cmd/tunnel_test.go index 23784d2..f18a154 100644 --- a/cmd/tunnel_test.go +++ b/cmd/tunnel_test.go @@ -130,3 +130,28 @@ func TestCloudflareTunnelURL(t *testing.T) { t.Fatalf("cloudflareTunnelURL() = %q", got) } } + +func TestResolveTunnelTarget(t *testing.T) { + t.Parallel() + + if got := resolveTunnelTarget("", func() (string, error) { + return "", errListenerPortMissing + }); got != defaultTunnelTarget { + t.Fatalf("resolveTunnelTarget() without listener = %q", got) + } + + if got := resolveTunnelTarget("", func() (string, error) { + return "54321", nil + }); got != "http://127.0.0.1:54321" { + t.Fatalf("resolveTunnelTarget() = %q", got) + } + + const explicit = "http://127.0.0.1:9876" + if got := resolveTunnelTarget(explicit, func() (string, error) { + t.Fatal("read listener port for explicit target") + + return "", nil + }); got != explicit { + t.Fatalf("resolveTunnelTarget(explicit) = %q", got) + } +}