diff --git a/cmd/tunnel.go b/cmd/tunnel.go new file mode 100644 index 0000000..94c8f11 --- /dev/null +++ b/cmd/tunnel.go @@ -0,0 +1,311 @@ +package cmd + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "os/signal" + "path/filepath" + "regexp" + "strings" + "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") + 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), + 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", "", + "Local URL that receives tunneled requests (default: running listener or http://127.0.0.1:4000)") + 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, 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( + 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..f18a154 --- /dev/null +++ b/cmd/tunnel_test.go @@ -0,0 +1,157 @@ +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) + } +} + +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) + } +}