spinifex/tests/integration/conformance.go

//go:build integration

package integration

import (
	"bytes"
	"fmt"
	"io"
	"maps"
	"net/http"
	"net/url"
	"slices"
	"sort"
	"strings"
	"sync"

	"github.com/mulgadc/spinifex/internal/awsmodel"
)

type conformanceFinding struct {
	service   awsmodel.Service
	operation string
	violation awsmodel.Violation
	count     int
}

type conformanceDiagnostic struct {
	service   awsmodel.Service
	operation string
	message   string
	count     int
}

// conformanceCollector is shared across the integration suite. Findings for
// services in the checked-in promotion policy become blocking in fail mode.
type conformanceCollector struct {
	mu                       sync.Mutex
	checked                  int
	checkedByService         map[awsmodel.Service]int
	errorChecked             int
	errorCheckedByService    map[awsmodel.Service]int
	errorUnmodelled          int
	errorUnmodelledByService map[awsmodel.Service]int
	findings                 map[string]*conformanceFinding
	diagnostics              map[string]*conformanceDiagnostic
}

func newConformanceCollector() *conformanceCollector {
	return &conformanceCollector{
		checkedByService:         make(map[awsmodel.Service]int),
		errorCheckedByService:    make(map[awsmodel.Service]int),
		errorUnmodelledByService: make(map[awsmodel.Service]int),
		findings:                 make(map[string]*conformanceFinding),
		diagnostics:              make(map[string]*conformanceDiagnostic),
	}
}

var suiteConformance = newConformanceCollector()

func (c *conformanceCollector) record(service awsmodel.Service, operation string, body []byte) {
	violations, err := awsmodel.ValidateResponse(service, operation, body)

	c.mu.Lock()
	defer c.mu.Unlock()
	c.checked++
	c.checkedByService[service]++
	if err != nil {
		key := string(service) + "\x00" + operation + "\x00" + err.Error()
		diagnostic := c.diagnostics[key]
		if diagnostic == nil {
			diagnostic = &conformanceDiagnostic{service: service, operation: operation, message: err.Error()}
			c.diagnostics[key] = diagnostic
		}
		diagnostic.count++
		return
	}
	c.recordViolationsLocked(service, operation, violations)
}

func (c *conformanceCollector) recordError(service awsmodel.Service, operation string, status int, body []byte) {
	var (
		violations []awsmodel.Violation
		modelled   bool
		err        error
	)
	if service == awsmodel.EC2 {
		violations, err = awsmodel.ValidateEC2ErrorResponse(status, body)
		modelled = true
	} else {
		violations, modelled, err = awsmodel.ValidateErrorResponse(service, operation, body)
	}

	c.mu.Lock()
	defer c.mu.Unlock()
	if err != nil {
		key := string(service) + "\x00" + operation + "\x00error\x00" + err.Error()
		diagnostic := c.diagnostics[key]
		if diagnostic == nil {
			diagnostic = &conformanceDiagnostic{service: service, operation: operation, message: err.Error()}
			c.diagnostics[key] = diagnostic
		}
		diagnostic.count++
		return
	}
	if !modelled {
		c.errorUnmodelled++
		c.errorUnmodelledByService[service]++
		return
	}
	c.errorChecked++
	c.errorCheckedByService[service]++
	c.recordViolationsLocked(service, operation, violations)
}

func (c *conformanceCollector) recordViolationsLocked(service awsmodel.Service, operation string, violations []awsmodel.Violation) {
	for _, violation := range violations {
		key := strings.Join([]string{string(service), operation, string(violation.Rule), violation.Path, violation.Message}, "\x00")
		finding := c.findings[key]
		if finding == nil {
			finding = &conformanceFinding{service: service, operation: operation, violation: violation}
			c.findings[key] = finding
		}
		finding.count++
	}
}

func (c *conformanceCollector) counts() (checked, violations, diagnostics int) {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, finding := range c.findings {
		violations += finding.count
	}
	for _, diagnostic := range c.diagnostics {
		diagnostics += diagnostic.count
	}
	return c.checked, violations, diagnostics
}

func (c *conformanceCollector) blocking(policy conformancePolicy, mode conformanceMode) int {
	if mode != conformanceModeFail {
		return 0
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.blockingLocked(policy)
}

func (c *conformanceCollector) blockingLocked(policy conformancePolicy) int {
	blocking := 0
	for _, finding := range c.findings {
		if policy.isPromoted(finding.service) {
			blocking += finding.count
		}
	}
	for _, diagnostic := range c.diagnostics {
		if policy.isPromoted(diagnostic.service) {
			blocking += diagnostic.count
		}
	}
	return blocking
}

func (c *conformanceCollector) report(policy conformancePolicy, mode conformanceMode) string {
	c.mu.Lock()
	defer c.mu.Unlock()

	violationCount := 0
	findings := make([]*conformanceFinding, 0, len(c.findings))
	for _, finding := range c.findings {
		violationCount += finding.count
		findings = append(findings, finding)
	}
	diagnosticCount := 0
	diagnostics := make([]*conformanceDiagnostic, 0, len(c.diagnostics))
	for _, diagnostic := range c.diagnostics {
		diagnosticCount += diagnostic.count
		diagnostics = append(diagnostics, diagnostic)
	}
	sort.Slice(findings, func(i, j int) bool {
		left, right := findings[i], findings[j]
		return fmt.Sprint(left.service, "\x00", left.operation, "\x00", left.violation.Path, "\x00", left.violation.Rule) <
			fmt.Sprint(right.service, "\x00", right.operation, "\x00", right.violation.Path, "\x00", right.violation.Rule)
	})
	sort.Slice(diagnostics, func(i, j int) bool {
		left, right := diagnostics[i], diagnostics[j]
		return fmt.Sprint(left.service, "\x00", left.operation, "\x00", left.message) <
			fmt.Sprint(right.service, "\x00", right.operation, "\x00", right.message)
	})

	var report strings.Builder
	blocking := 0
	if mode == conformanceModeFail {
		blocking = c.blockingLocked(policy)
	}
	promoted := make([]string, 0, len(policy.promoted))
	for _, service := range policy.services() {
		promoted = append(promoted, string(service))
	}
	fmt.Fprintf(&report, "AWS model conformance (%s): checked=%d errors_checked=%d errors_unmodelled=%d violations=%d decode_errors=%d blocking=%d promoted=%s\n",
		mode, c.checked, c.errorChecked, c.errorUnmodelled, violationCount, diagnosticCount, blocking, strings.Join(promoted, ","))

	checkedServices := make(map[awsmodel.Service]bool, len(c.checkedByService)+len(c.errorCheckedByService)+len(c.errorUnmodelledByService)+len(policy.promoted))
	for service := range c.checkedByService {
		checkedServices[service] = true
	}
	for service := range c.errorCheckedByService {
		checkedServices[service] = true
	}
	for service := range c.errorUnmodelledByService {
		checkedServices[service] = true
	}
	for service := range policy.promoted {
		checkedServices[service] = true
	}
	services := slices.Sorted(maps.Keys(checkedServices))
	for _, service := range services {
		fmt.Fprintf(&report, "CHECKED %s success=%d errors=%d unmodelled_errors=%d promoted=%t\n",
			service, c.checkedByService[service], c.errorCheckedByService[service], c.errorUnmodelledByService[service], policy.isPromoted(service))
	}

	for _, finding := range findings {
		severity := "WARN"
		if mode == conformanceModeFail && policy.isPromoted(finding.service) {
			severity = "FAIL"
		}
		fmt.Fprintf(&report, "%s %s %s %s %s count=%d: %s\n",
			severity,
			finding.service, finding.operation, finding.violation.Path, finding.violation.Rule, finding.count, finding.violation.Message)
	}
	for _, diagnostic := range diagnostics {
		severity := "WARN"
		if mode == conformanceModeFail && policy.isPromoted(diagnostic.service) {
			severity = "FAIL"
		}
		fmt.Fprintf(&report, "%s %s %s decode_error count=%d: %s\n",
			severity,
			diagnostic.service, diagnostic.operation, diagnostic.count, diagnostic.message)
	}
	return strings.TrimSuffix(report.String(), "\n")
}

func conformanceMiddleware(next http.Handler, collector *conformanceCollector) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		service, operation, ok := resolveConformanceRequest(r)
		if !ok {
			next.ServeHTTP(w, r)
			return
		}

		capture := &conformanceResponseWriter{ResponseWriter: w}
		next.ServeHTTP(capture, r)
		status := capture.status
		if status == 0 {
			status = http.StatusOK
		}
		if status >= 200 && status < 300 {
			collector.record(service, operation, capture.body.Bytes())
		} else if errorCodeConformanceService(service) {
			collector.recordError(service, operation, status, capture.body.Bytes())
		}
	})
}

func errorCodeConformanceService(service awsmodel.Service) bool {
	switch service {
	case awsmodel.EC2, awsmodel.IAM, awsmodel.STS, awsmodel.ECS, awsmodel.ElasticLoadBalancingV2:
		return true
	default:
		return false
	}
}

type conformanceResponseWriter struct {
	http.ResponseWriter
	status int
	body   bytes.Buffer
}

func (w *conformanceResponseWriter) WriteHeader(status int) {
	if w.status == 0 {
		w.status = status
	}
	w.ResponseWriter.WriteHeader(status)
}

func (w *conformanceResponseWriter) Write(body []byte) (int, error) {
	if w.status == 0 {
		w.status = http.StatusOK
	}
	w.body.Write(body)
	return w.ResponseWriter.Write(body)
}

func (w *conformanceResponseWriter) Flush() {
	_ = http.NewResponseController(w.ResponseWriter).Flush()
}

func (w *conformanceResponseWriter) Unwrap() http.ResponseWriter { return w.ResponseWriter }

func resolveConformanceRequest(r *http.Request) (awsmodel.Service, string, bool) {
	service, ok := conformanceService(r)
	if !ok {
		return "", "", false
	}

	var operation string
	if target := r.Header.Get("X-Amz-Target"); target != "" {
		if separator := strings.LastIndex(target, "."); separator >= 0 {
			operation = target[separator+1:]
		} else {
			operation = target
		}
	} else if service != awsmodel.S3 {
		operation = queryAction(r)
	}
	if operation == "" {
		return "", "", false
	}

	model, err := awsmodel.Load(service)
	if err != nil {
		return "", "", false
	}
	if _, modelled := model.Operation(operation); !modelled {
		return "", "", false
	}
	return service, operation, true
}

func conformanceService(r *http.Request) (awsmodel.Service, bool) {
	if target := r.Header.Get("X-Amz-Target"); target != "" {
		prefix := target
		if separator := strings.LastIndex(target, "."); separator >= 0 {
			prefix = target[:separator]
		}
		switch prefix {
		case "AmazonEC2ContainerServiceV20141113":
			return awsmodel.ECS, true
		case "AmazonEC2ContainerRegistry_V20150921":
			return awsmodel.ECR, true
		case "CertificateManager":
			return awsmodel.ACM, true
		}
	}

	authorization := r.Header.Get("Authorization")
	credentialAt := strings.Index(authorization, "Credential=")
	if credentialAt >= 0 {
		credential := authorization[credentialAt+len("Credential="):]
		if end := strings.IndexAny(credential, " ,"); end >= 0 {
			credential = credential[:end]
		}
		parts := strings.Split(credential, "/")
		if len(parts) >= 5 && parts[len(parts)-1] == "aws4_request" {
			return modelService(parts[len(parts)-2])
		}
	}

	// AssumeRoleWithWebIdentity is intentionally unsigned and is the only
	// anonymous modelled request in this harness.
	if queryAction(r) == "AssumeRoleWithWebIdentity" {
		return awsmodel.STS, true
	}
	return "", false
}

func modelService(signatureService string) (awsmodel.Service, bool) {
	switch signatureService {
	case "acm":
		return awsmodel.ACM, true
	case "ec2":
		return awsmodel.EC2, true
	case "ecr":
		return awsmodel.ECR, true
	case "ecs":
		return awsmodel.ECS, true
	case "elasticloadbalancing":
		return awsmodel.ElasticLoadBalancingV2, true
	case "iam":
		return awsmodel.IAM, true
	case "s3":
		return awsmodel.S3, true
	case "sts":
		return awsmodel.STS, true
	default:
		return "", false
	}
}

func queryAction(r *http.Request) string {
	if action := r.URL.Query().Get("Action"); action != "" {
		return action
	}
	if r.Body == nil {
		return ""
	}
	body, err := io.ReadAll(r.Body)
	if err != nil {
		return ""
	}
	r.Body = io.NopCloser(bytes.NewReader(body))
	values, err := url.ParseQuery(string(body))
	if err != nil {
		return ""
	}
	return values.Get("Action")
}