Repository: avenga/couper
Pull request #618 (2022-11-07 13:35:25)

Fix seq panic
Fix a possible issue(panic) with nested and/or chained sequences with depth > 1.

---
<details>
    <summary>Reviewer checklist</summary>
    <ul>
        <li>Read PR description: a summary about the changes is required</li>
        <li>Changelog updated</li>
        <li>Documentation: docs/{Reference, Cli, ...}, Docker and cli help/usage</li>
        <li>Pulled branch, manually tested</li>
        <li>Verified requirements are met</li>
        <li>Reviewed the code</li>
        <li>Reviewed the related tests</li>
    </ul>
</details>



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index b101f584..99eabe57 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@ Unreleased changes are available as `avenga/couper:edge` container.
 
 * **Fixed**
   * CVE-2021-3538 related to our `request_id_format` option if switched to `uuid4`: replaced the underlying package to `github.com/google/uuid` ([#611](https://github.com/avenga/couper/pull/611))
+  * Possible panic for nested endpoint sequences ([#618](https://github.com/avenga/couper/pull/618))
 
 ---
 
diff --git a/config/runtime/endpoint.go b/config/runtime/endpoint.go
index d6b7ae8e..30d07f72 100644
--- a/config/runtime/endpoint.go
+++ b/config/runtime/endpoint.go
@@ -146,7 +146,7 @@ func newEndpointOptions(confCtx *hcl.EvalContext, endpointConf *config.Endpoint,
 		blockBodies = append(blockBodies, requestConf.Backend, requestConf.HCLBody())
 	}
 
-	sequences, requests, proxies := newSequences(allProxies, allRequests, endpointConf.Sequences...)
+	sequences, requests, proxies := resolveDependencies(allProxies, allRequests, endpointConf.Sequences...)
 
 	// TODO: redirect
 	if endpointConf.Response == nil && len(proxies)+len(requests)+len(sequences) == 0 { // && redirect == nil
@@ -191,20 +191,20 @@ func newEndpointOptions(confCtx *hcl.EvalContext, endpointConf *config.Endpoint,
 	}, nil
 }
 
-// newSequences lookups any request related dependency and sort them into a sequence.
+// resolveDependencies lookups any request related dependency and sort them into a sequence.
 // Also return left-overs for parallel usage.
-func newSequences(proxies map[string]*producer.Proxy, requests map[string]*producer.Request,
-	items ...*sequence.Item) (producer.Sequences, producer.Requests, producer.Proxies) {
+func resolveDependencies(proxies map[string]*producer.Proxy, requests map[string]*producer.Request,
+	items ...*sequence.Item) (producer.Parallel, producer.Requests, producer.Proxies) {
 
 	allDeps := sequence.Dependencies(items)
 
 	var reqs producer.Requests
 	var ps producer.Proxies
-	var seqs producer.Sequences
+	var seqs producer.Parallel
 
 	// read from prepared config sequences
 	for _, seq := range items {
-		seqs = append(seqs, newSequence(seq, proxies, requests))
+		seqs = append(seqs, newRoundtrip(seq, proxies, requests))
 	}
 
 proxyLeftovers:
@@ -234,7 +234,7 @@ reqLeftovers:
 	return seqs, reqs, ps
 }
 
-func newSequence(seq *sequence.Item,
+func newRoundtrip(seq *sequence.Item,
 	proxies map[string]*producer.Proxy,
 	requests map[string]*producer.Request) producer.Roundtrip {
 
@@ -243,14 +243,14 @@ func newSequence(seq *sequence.Item,
 
 	var previous []string
 	if len(deps) > 1 { // more deps per item can be parallelized
-		var seqs producer.Sequences
+		var seqs producer.Parallel
 		for _, d := range deps {
-			seqs = append(seqs, newSequence(d, proxies, requests))
+			seqs = append(seqs, newRoundtrip(d, proxies, requests))
 			previous = append(previous, d.Name)
 		}
 		rt = seqs
 	} else if len(deps) == 1 {
-		rt = newSequence(deps[0], proxies, requests)
+		rt = newRoundtrip(deps[0], proxies, requests)
 		previous = append(previous, deps[0].Name)
 	}
 
diff --git a/handler/endpoint.go b/handler/endpoint.go
index 004847eb..ae3d20d3 100644
--- a/handler/endpoint.go
+++ b/handler/endpoint.go
@@ -45,7 +45,7 @@ type EndpointOptions struct {
 	Proxies   producer.Roundtrip
 	Redirect  *producer.Redirect
 	Requests  producer.Roundtrip
-	Sequences producer.Sequences
+	Sequences producer.Roundtrip
 	Response  *producer.Response
 }
 
@@ -128,6 +128,8 @@ func (e *Endpoint) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 		if !ok {
 			err = errors.Server
 		}
+	} else {
+		err = errors.Server.Message("missing client response")
 	}
 
 	if handled := e.handleError(rw, req, err); handled {
@@ -215,23 +217,18 @@ func (e *Endpoint) produce(req *http.Request) (producer.ResultMap, error) {
 			continue
 		}
 
-		resultCh := make(chan *producer.Result, trip.Len())
-		go func(rt producer.Roundtrip, rc chan *producer.Result) {
-			rt.Produce(outreq, resultCh)
-			close(rc)
-		}(trip, resultCh)
-		tripCh <- resultCh
+		tripCh <- trip.Produce(outreq)
 	}
 	close(tripCh)
 
 	for resultCh := range tripCh {
-		e.readResults(req.Context(), resultCh, results)
+		e.readResults(resultCh, results)
 	}
 
 	var err error // TODO: prefer default resp err
 	// TODO: additionally log all panic error types
 	for _, r := range results {
-		if r.Err != nil {
+		if r != nil && r.Err != nil {
 			err = r.Err
 			break
 		}
@@ -240,26 +237,19 @@ func (e *Endpoint) produce(req *http.Request) (producer.ResultMap, error) {
 	return results, err
 }
 
-func (e *Endpoint) readResults(ctx context.Context, requestResults producer.Results, beresps producer.ResultMap) {
+func (e *Endpoint) readResults(requestResults producer.Results, beresps producer.ResultMap) {
 	i := 0
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case r, more := <-requestResults:
-			if !more {
-				return
-			}
-
-			i++
-			name := r.RoundTripName
+	for r := range requestResults {
+		i++
+		var name string
+		if r != nil {
+			name = r.RoundTripName
+		}
 
-			// fallback
-			if name == "" { // panic case
-				name = strconv.Itoa(i)
-			}
-			beresps[name] = r
+		if name == "" {
+			name = strconv.Itoa(i)
 		}
+		beresps[name] = r
 	}
 }
 
diff --git a/handler/producer/produce.go b/handler/producer/produce.go
index 64a17f69..7e949732 100644
--- a/handler/producer/produce.go
+++ b/handler/producer/produce.go
@@ -7,11 +7,11 @@ import (
 var (
 	_ Roundtrip = Proxies{}
 	_ Roundtrip = Requests{}
-	_ Roundtrip = Sequences{}
+	_ Roundtrip = Parallel{}
 	_ Roundtrip = Sequence{}
 )
 
 type Roundtrip interface {
-	Produce(req *http.Request, results chan<- *Result)
+	Produce(req *http.Request) chan *Result
 	Len() int
 }
diff --git a/handler/producer/proxy.go b/handler/producer/proxy.go
index 1d0b066a..f04dbbaf 100644
--- a/handler/producer/proxy.go
+++ b/handler/producer/proxy.go
@@ -24,7 +24,7 @@ type Proxy struct {
 
 type Proxies []*Proxy
 
-func (pr Proxies) Produce(clientReq *http.Request, results chan<- *Result) {
+func (pr Proxies) Produce(clientReq *http.Request) chan *Result {
 	var currentName string // at least pre roundtrip
 	wg := &sync.WaitGroup{}
 	ctx := clientReq.Context()
@@ -33,6 +33,9 @@ func (pr Proxies) Produce(clientReq *http.Request, results chan<- *Result) {
 		ctx, rootSpan = telemetry.NewSpanFromContext(ctx, "proxies", trace.WithSpanKind(trace.SpanKindProducer))
 	}
 
+	results := make(chan *Result, pr.Len())
+	defer close(results)
+
 	defer func() {
 		if rp := recover(); rp != nil {
 			results <- &Result{
@@ -83,6 +86,8 @@ func (pr Proxies) Produce(clientReq *http.Request, results chan<- *Result) {
 	}
 
 	wg.Wait()
+
+	return results
 }
 
 func (pr Proxies) Len() int {
diff --git a/handler/producer/request.go b/handler/producer/request.go
index 98c533b5..6df7e8f3 100644
--- a/handler/producer/request.go
+++ b/handler/producer/request.go
@@ -30,7 +30,7 @@ type Request struct {
 // Requests represents the producer <Requests> object.
 type Requests []*Request
 
-func (r Requests) Produce(req *http.Request, results chan<- *Result) {
+func (r Requests) Produce(req *http.Request) chan *Result {
 	var currentName string // at least pre roundtrip
 	wg := &sync.WaitGroup{}
 	ctx := req.Context()
@@ -39,6 +39,9 @@ func (r Requests) Produce(req *http.Request, results chan<- *Result) {
 		ctx, rootSpan = telemetry.NewSpanFromContext(ctx, "requests", trace.WithSpanKind(trace.SpanKindProducer))
 	}
 
+	results := make(chan *Result, r.Len())
+	defer close(results)
+
 	defer func() {
 		if rp := recover(); rp != nil {
 			results <- &Result{
@@ -134,6 +137,8 @@ func (r Requests) Produce(req *http.Request, results chan<- *Result) {
 	}
 
 	wg.Wait()
+
+	return results
 }
 
 func (r Requests) Len() int {
diff --git a/handler/producer/sequence.go b/handler/producer/sequence.go
index aa4a50f2..701276c8 100644
--- a/handler/producer/sequence.go
+++ b/handler/producer/sequence.go
@@ -1,111 +1,102 @@
 package producer
 
 import (
+	"context"
 	"net/http"
 
-	"github.com/hashicorp/hcl/v2"
 	"go.opentelemetry.io/otel/trace"
 
 	"github.com/avenga/couper/errors"
 	"github.com/avenga/couper/telemetry"
 )
 
-type SequenceItem struct {
-	Backend http.RoundTripper
-	Context hcl.Body
-	Name    string // label
-}
-
-// Sequence represents a list of serialized items.
+// Sequence holds a list of items which get executed sequentially.
 type Sequence []Roundtrip
 
-// Sequences holds several list of serialized items.
-type Sequences []Roundtrip
-
-func (seqs Sequences) Produce(req *http.Request, results chan<- *Result) {
-	var rootSpan trace.Span
-	ctx := req.Context()
-	if len(seqs) > 0 {
-		ctx, rootSpan = telemetry.NewSpanFromContext(ctx, "sequences", trace.WithSpanKind(trace.SpanKindProducer))
-	}
-
-	resultsCh := make(chan *Result, seqs.Len())
+// Parallel holds a list of items which get executed in parallel.
+type Parallel []Roundtrip
 
-	for _, s := range seqs {
-		outreq := req.WithContext(ctx)
-		go s.Produce(outreq, resultsCh)
-	}
+func (seqs Parallel) Produce(req *http.Request) chan *Result {
+	return pipe(req, seqs, "parallel")
+}
 
-	for i := 0; i < seqs.Len(); i++ {
-		results <- <-resultsCh
-	}
+func (seqs Parallel) Len() int {
+	return len(seqs)
+}
 
-	if rootSpan != nil {
-		rootSpan.End()
-	}
+func (s Sequence) Produce(req *http.Request) chan *Result {
+	return pipe(req, s, "sequence")
 }
 
-func (seqs Sequences) Len() int {
-	return len(seqs)
+func (s Sequence) Len() int {
+	var sum int
+	for _, t := range s {
+		sum += t.Len()
+	}
+	return sum
 }
 
-func (s Sequence) Produce(req *http.Request, results chan<- *Result) {
+// pipe calls the Roundtrip Interface on each given item and distinguishes between parallelism and trace kind.
+// The returned channel will be closed if this chain part has been ended.
+func pipe(req *http.Request, rt []Roundtrip, kind string) chan *Result {
 	var rootSpan trace.Span
-
 	ctx := req.Context()
-	l := s.Len()
-	if l > 0 {
-		ctx, rootSpan = telemetry.NewSpanFromContext(ctx, "sequence", trace.WithSpanKind(trace.SpanKindProducer))
+	if len(rt) > 0 {
+		ctx, rootSpan = telemetry.NewSpanFromContext(ctx, kind, trace.WithSpanKind(trace.SpanKindProducer))
+		defer rootSpan.End()
 	}
-	defer func() {
-		if rootSpan != nil {
-			rootSpan.End()
-		}
-	}()
 
-	result := make(chan *Result, l)
+	result := make(chan *Result, len(rt))
+	var allResults []chan *Result
 
-	var lastResult *Result
-	var lastBeresps []*http.Response
-	var moreEntries bool
-	for _, seq := range s {
-		outCtx := ctx
+	for _, srt := range rt {
+		rch := make(chan *Result, srt.Len())
+		allResults = append(allResults, rch)
 
-		outreq := req.WithContext(outCtx)
+		switch kind {
+		case "parallel": // execute each sequence branch in parallel
+			go pipeResult(ctx, req, rch, srt)
+		case "sequence": // one by one
+			pipeResult(ctx, req, rch, srt)
+		}
+	}
 
-		seq.Produce(outreq, result)
+	// Since the sequence gets resolved in order just the last item matters.
+	for _, rch := range allResults {
+		var last *Result
+		var err error
 
-		select {
-		case <-outreq.Context().Done():
-			return
-		case lastResult, moreEntries = <-result:
-			if !moreEntries {
-				return
+		for last = range rch {
+			// drain
+			if last.Err != nil {
+				err = last.Err
+				// drain must be continued (pipeResult)
 			}
 		}
 
-		if lastResult.Err != nil {
-			if _, ok := lastResult.Err.(*errors.Error); !ok {
-				lastResult.Err = errors.Sequence.With(lastResult.Err)
-			}
-			results <- lastResult
-			return
+		if err != nil {
+			result <- &Result{Err: err}
+		} else if last == nil {
+			result <- &Result{Err: errors.Sequence.Message("no result")}
+		} else {
+			result <- last
 		}
-
-		lastBeresps = append(lastBeresps, lastResult.Beresp)
-	}
-
-	if lastResult == nil {
-		results <- &Result{Err: errors.Sequence.With(errors.New().Message("no result"))}
 	}
 
-	results <- lastResult
+	close(result)
+	return result
 }
 
-func (s Sequence) Len() int {
-	var sum int
-	for _, t := range s {
-		sum += t.Len()
+func pipeResult(ctx context.Context, req *http.Request, results chan *Result, rt Roundtrip) {
+	outreq := req.WithContext(ctx)
+	defer close(results)
+
+	for r := range rt.Produce(outreq) {
+		select {
+		case <-ctx.Done():
+			results <- &Result{Err: ctx.Err()}
+			return
+		case results <- r:
+		}
 	}
-	return sum
 }
diff --git a/handler/transport/token_request.go b/handler/transport/token_request.go
index 71280431..5244606e 100644
--- a/handler/transport/token_request.go
+++ b/handler/transport/token_request.go
@@ -79,13 +79,11 @@ func (t *TokenRequest) readToken() string {
 }
 
 func (t *TokenRequest) requestToken(req *http.Request) (string, int64, error) {
-	results := make(chan *producer.Result, 1)
 	ctx := context.WithValue(req.Context(), request.Wildcard, nil)           // disable handling this
 	ctx = context.WithValue(ctx, request.BufferOptions, eval.BufferResponse) // always read out a possible token
 	ctx = context.WithValue(ctx, request.TokenRequest, t.config.Name)        // set the name for variable mapping purposes
 	outreq, _ := http.NewRequestWithContext(ctx, req.Method, "", nil)
-	t.reqProducer.Produce(outreq, results)
-	result := <-results
+	result := <-t.reqProducer.Produce(outreq)
 	if result.Err != nil {
 		return "", 0, fmt.Errorf("token request failed") // don't propagate token request roundtrip error
 	}


Test patch:
diff --git a/handler/endpoint_test.go b/handler/endpoint_test.go
index 6745d2ab..01cbe7ce 100644
--- a/handler/endpoint_test.go
+++ b/handler/endpoint_test.go
@@ -379,21 +379,27 @@ func TestEndpoint_RoundTripContext_Null_Eval(t *testing.T) {
 	}
 }
 
+var _ producer.Roundtrip = &mockProducerResult{}
+
 type mockProducerResult struct {
 	rt http.RoundTripper
 }
 
-func (m *mockProducerResult) Produce(r *http.Request, results chan<- *producer.Result) {
+func (m *mockProducerResult) Produce(r *http.Request) chan *producer.Result {
+	result := make(chan *producer.Result, 1)
+	defer close(result)
+
 	if m == nil || m.rt == nil {
-		return
+		return result
 	}
 
 	res, err := m.rt.RoundTrip(r)
-	results <- &producer.Result{
+	result <- &producer.Result{
 		RoundTripName: "default",
 		Beresp:        res,
 		Err:           err,
 	}
+	return result
 }
 
 func (m *mockProducerResult) Len() int {
diff --git a/handler/producer/request_test.go b/handler/producer/request_test.go
index b5498a42..9c4cdc49 100644
--- a/handler/producer/request_test.go
+++ b/handler/producer/request_test.go
@@ -107,9 +107,7 @@ func Test_ProduceExpectedStatus(t *testing.T) {
 				outreq := clientRequest.WithContext(ctx)
 				outreq.Header.Set("X-Status", strconv.Itoa(tt.reflectStatus))
 
-				resultCh := make(chan *producer.Result, 1)
-				rt.Produce(outreq, resultCh)
-				result := <-resultCh
+				result := <-rt.Produce(outreq)
 
 				if !errors.Equals(tt.expectedErr, result.Err) {
 					t.Fatalf("expected error: %v, got %v", tt.expectedErr, result.Err)
diff --git a/server/http_endpoints_test.go b/server/http_endpoints_test.go
index 99c7e861..b522ebf9 100644
--- a/server/http_endpoints_test.go
+++ b/server/http_endpoints_test.go
@@ -567,7 +567,7 @@ func TestHTTPServer_NoGzipForSmallContent(t *testing.T) {
 	}
 }
 
-func TestEndpointSequence(t *testing.T) {
+func TestEndpoint_Sequence(t *testing.T) {
 	client := test.NewHTTPClient()
 	helper := test.New(t)
 
@@ -673,7 +673,7 @@ func TestEndpointSequence(t *testing.T) {
 
 }
 
-func TestEndpointSequenceClientCancel(t *testing.T) {
+func TestEndpoint_Sequence_ClientCancel(t *testing.T) {
 	client := test.NewHTTPClient()
 	helper := test.New(t)
 
@@ -715,18 +715,19 @@ func TestEndpointSequenceClientCancel(t *testing.T) {
 
 		path := entry.Data["request"].(logging.Fields)["path"]
 
-		if strings.Contains(entry.Message, context.Canceled.Error()) {
-			ctxCanceledSeen = true
-			if path != "/" {
-				t.Errorf("expected '/' to fail")
-			}
-		}
-
-		if entry.Message == "" && entry.Data["status"] == 200 {
-			statusOKseen = true
-			if path != "/reflect" {
-				t.Errorf("expected '/reflect' to be ok")
+		switch path {
+		case "/":
+			isCancelErr := strings.Contains(entry.Message, context.Canceled.Error())
+			hasStatusZero := entry.Data["status"] == 0
+			ctxCanceledSeen = isCancelErr && hasStatusZero && entry.Level == logrus.ErrorLevel
+		case "/reflect":
+			request := entry.Data["request"].(logging.Fields)
+			if request["name"] != "resolve_first" {
+				continue
 			}
+			isCancelErr := strings.Contains(entry.Message, context.Canceled.Error())
+			hasStatusOK := entry.Data["status"] == http.StatusOK
+			statusOKseen = !isCancelErr && hasStatusOK && entry.Level == logrus.InfoLevel
 		}
 	}
 
@@ -735,7 +736,7 @@ func TestEndpointSequenceClientCancel(t *testing.T) {
 	}
 }
 
-func TestEndpointSequenceBackendTimeout(t *testing.T) {
+func TestEndpoint_Sequence_BackendTimeout(t *testing.T) {
 	client := test.NewHTTPClient()
 	helper := test.New(t)
 
@@ -775,25 +776,47 @@ func TestEndpointSequenceBackendTimeout(t *testing.T) {
 		}
 
 		path := entry.Data["request"].(logging.Fields)["path"]
-		if entry.Message == "backend error: anonymous_3_23: deadline exceeded" {
-			ctxDeadlineSeen = true
-			if path != "/" {
-				t.Errorf("expected '/' to fail")
-			}
-		}
 
-		if entry.Message == "" && entry.Data["status"] == 200 {
-			statusOKseen = true
-			if path != "/reflect" {
-				t.Errorf("expected '/reflect' to be ok")
-			}
+		switch path {
+		case "/":
+			isDeadlineErr := entry.Message == "backend error: anonymous_3_23: deadline exceeded"
+			hasStatusZero := entry.Data["status"] == 0
+			ctxDeadlineSeen = isDeadlineErr && hasStatusZero && entry.Level == logrus.ErrorLevel
+		case "/reflect":
+			hasStatusOK := entry.Data["status"] == http.StatusOK
+			statusOKseen = hasStatusOK && entry.Level == logrus.InfoLevel
 		}
 	}
 
 	if !ctxDeadlineSeen || !statusOKseen {
 		t.Errorf("Expected one sucessful and one failed backend request")
 	}
+}
+
+func TestEndpoint_Sequence_NestedDefaultRequest(t *testing.T) {
+	client := test.NewHTTPClient()
+	helper := test.New(t)
+
+	shutdown, _ := newCouper(filepath.Join(testdataPath, "19_couper.hcl"), helper)
+	defer shutdown()
+
+	req, err := http.NewRequest(http.MethodGet, "http://domain.local:8080/", nil)
+	helper.Must(err)
+
+	res, err := client.Do(req)
+	helper.Must(err)
+
+	if res.StatusCode != http.StatusOK {
+		t.Errorf("Expected StatusOK, got: %d", res.StatusCode)
+	}
 
+	b, err := io.ReadAll(res.Body)
+	helper.Must(err)
+
+	exp := `{"data":[{"features":1},{"features":2}]}`
+	if !bytes.Equal([]byte(exp), b) {
+		t.Errorf("expected %v, got %v", exp, string(b))
+	}
 }
 
 func TestEndpointCyclicSequence(t *testing.T) {
diff --git a/server/testdata/endpoints/19_couper.hcl b/server/testdata/endpoints/19_couper.hcl
new file mode 100644
index 00000000..a077a89a
--- /dev/null
+++ b/server/testdata/endpoints/19_couper.hcl
@@ -0,0 +1,72 @@
+server { # sequences
+  hosts = ["*:8080"]
+
+  api {
+    endpoint "/" {
+      request "r1" {
+        url = "/res1"
+        backend = "src"
+      }
+
+      request "r2" {
+        url = "/res2"
+        backend = "src"
+      }
+
+      request { # default
+        url = "/"
+        backend = "dest"
+        json_body = {
+          data = [
+            backend_responses.r1.json_body,
+            backend_responses.r2.json_body
+          ]
+        }
+      }
+    }
+  }
+}
+
+definitions {
+  backend "src" {
+    origin = "http://localhost:8081"
+  }
+
+  backend "dest" {
+    origin = "http://localhost:8082"
+  }
+}
+
+server "src" {
+  hosts = ["*:8081"]
+
+  api {
+    endpoint "/res1" {
+      response {
+        json_body = {
+          features = 1
+        }
+      }
+    }
+
+    endpoint "/res2" {
+      response {
+        json_body = {
+          features = 2
+        }
+      }
+    }
+  }
+}
+
+server "dest" {
+  hosts = ["*:8082"]
+
+  api {
+    endpoint "/" {
+      response {
+        json_body = request.json_body
+      }
+    }
+  }
+}