feat(14-01): guarded fetchguard client replayed through a tide upstream fake
- fetchguard.NewClient with Do, Send, Get and PostJSON over a capped, never-redirecting transport - WithTransport: code-only context seam for offline replay; Result gains Header - tide UpstreamSidecar, LoadUpstream, UpstreamPath and the asserting UpstreamFake
This commit is contained in:
Jakub Zych
305
modules/tide/upstream.go
Normal file
305
modules/tide/upstream.go
Normal file
@@ -0,0 +1,305 @@
|
||||
package tide
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/goccy/go-yaml"
|
||||
)
|
||||
|
||||
// UpstreamSidecar is the upstream HTTP traffic a backend sent to outside
|
||||
// services while one parity fixture was recorded. It lives next to the
|
||||
// fixture as <fixture>.upstream.yaml (see UpstreamPath). Exchanges are
|
||||
// consumed in recorded order across the whole flow.
|
||||
type UpstreamSidecar struct {
|
||||
Version int `yaml:"version"`
|
||||
Exchanges []UpstreamExchange `yaml:"exchanges"`
|
||||
}
|
||||
|
||||
// UpstreamExchange is one outbound request and the vendor's answer.
|
||||
type UpstreamExchange struct {
|
||||
Request UpstreamRequest `yaml:"request"`
|
||||
Response UpstreamResponse `yaml:"response"`
|
||||
}
|
||||
|
||||
// UpstreamRequest is the request the backend sent upstream. URL is absolute
|
||||
// (scheme, host, path and query). Headers holds only the compared headers
|
||||
// (UpstreamCompareHeaders); credential values are {{name}} placeholders.
|
||||
type UpstreamRequest struct {
|
||||
Method string `yaml:"method"`
|
||||
URL string `yaml:"url"`
|
||||
Headers map[string]string `yaml:"headers,omitempty"`
|
||||
Body string `yaml:"body,omitempty"`
|
||||
}
|
||||
|
||||
// UpstreamResponse is the recorded vendor answer the fake replays.
|
||||
type UpstreamResponse struct {
|
||||
Status int `yaml:"status"`
|
||||
Headers map[string]string `yaml:"headers,omitempty"`
|
||||
Body string `yaml:"body,omitempty"`
|
||||
}
|
||||
|
||||
// UpstreamCompareHeaders are the request headers the fake asserts. A header
|
||||
// is compared when either the recorded or the sent request carries it.
|
||||
var UpstreamCompareHeaders = []string{
|
||||
"User-Agent",
|
||||
"Accept",
|
||||
"Content-Type",
|
||||
"Authorization",
|
||||
"X-Api-Key",
|
||||
"Anthropic-Version",
|
||||
"Anthropic-Beta",
|
||||
}
|
||||
|
||||
// UpstreamPath maps a fixture path to its sidecar path: x.yaml becomes
|
||||
// x.upstream.yaml.
|
||||
func UpstreamPath(fixturePath string) string {
|
||||
for _, ext := range []string{".yaml", ".yml"} {
|
||||
if strings.HasSuffix(fixturePath, ext) {
|
||||
return strings.TrimSuffix(fixturePath, ext) + ".upstream.yaml"
|
||||
}
|
||||
}
|
||||
return fixturePath + ".upstream.yaml"
|
||||
}
|
||||
|
||||
// LoadUpstream reads a version-1 sidecar, rejecting unknown fields. A missing
|
||||
// file returns an error wrapping fs.ErrNotExist.
|
||||
func LoadUpstream(path string) (UpstreamSidecar, error) {
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return UpstreamSidecar{}, fmt.Errorf("tide: read upstream %s: %w", path, err)
|
||||
}
|
||||
var s UpstreamSidecar
|
||||
dec := yaml.NewDecoder(bytes.NewReader(raw), yaml.DisallowUnknownField())
|
||||
if err := dec.Decode(&s); err != nil {
|
||||
return UpstreamSidecar{}, fmt.Errorf("tide: parse upstream %s: %w", path, err)
|
||||
}
|
||||
if err := validateUpstream(s); err != nil {
|
||||
return UpstreamSidecar{}, fmt.Errorf("tide: upstream %s: %w", path, err)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func validateUpstream(s UpstreamSidecar) error {
|
||||
if s.Version != CurrentVersion {
|
||||
return fmt.Errorf("version %d, want %d", s.Version, CurrentVersion)
|
||||
}
|
||||
for i, ex := range s.Exchanges {
|
||||
if strings.TrimSpace(ex.Request.Method) == "" {
|
||||
return fmt.Errorf("exchange %d: request method is required", i)
|
||||
}
|
||||
u, err := url.Parse(ex.Request.URL)
|
||||
if err != nil || u.Scheme == "" || u.Host == "" {
|
||||
return fmt.Errorf("exchange %d: request url %q must be absolute", i, ex.Request.URL)
|
||||
}
|
||||
if ex.Response.Status < 100 || ex.Response.Status > 999 {
|
||||
return fmt.Errorf("exchange %d: response status %d out of range", i, ex.Response.Status)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpstreamFake is an http.RoundTripper that answers from a sidecar without
|
||||
// dialing. Each request takes the next recorded exchange and is asserted
|
||||
// against it: method, scheme, host, path, query (order-insensitive), the
|
||||
// UpstreamCompareHeaders with {{name}} placeholders expanded from the store,
|
||||
// and the body (JSON semantically, otherwise byte for byte). A mismatch fails
|
||||
// the request and is remembered for Verify.
|
||||
type UpstreamFake struct {
|
||||
mu sync.Mutex
|
||||
store *Store
|
||||
exchanges []UpstreamExchange
|
||||
next int
|
||||
errs []error
|
||||
}
|
||||
|
||||
// NewUpstreamFake returns a fake that replays s. store resolves {{name}}
|
||||
// placeholders in the recorded requests and responses; it may be nil when the
|
||||
// sidecar has none.
|
||||
func NewUpstreamFake(s UpstreamSidecar, store *Store) *UpstreamFake {
|
||||
return &UpstreamFake{store: store, exchanges: slices.Clone(s.Exchanges)}
|
||||
}
|
||||
|
||||
// RoundTrip implements http.RoundTripper.
|
||||
func (f *UpstreamFake) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
var body []byte
|
||||
if req.Body != nil {
|
||||
var err error
|
||||
body, err = io.ReadAll(io.LimitReader(req.Body, DefaultMaxBody+1))
|
||||
_ = req.Body.Close()
|
||||
if err != nil {
|
||||
return nil, f.fail(fmt.Errorf("tide: upstream %s %s: read body: %w", req.Method, req.URL, err))
|
||||
}
|
||||
}
|
||||
f.mu.Lock()
|
||||
if f.next >= len(f.exchanges) {
|
||||
f.mu.Unlock()
|
||||
return nil, f.fail(fmt.Errorf("tide: upstream extra request %s %s", req.Method, redactURL(req.URL)))
|
||||
}
|
||||
idx := f.next
|
||||
ex := f.exchanges[idx]
|
||||
f.next++
|
||||
f.mu.Unlock()
|
||||
|
||||
if problems := f.compare(ex.Request, req, body); len(problems) > 0 {
|
||||
return nil, f.fail(fmt.Errorf("tide: upstream exchange %d %s %s: %s",
|
||||
idx, req.Method, redactURL(req.URL), strings.Join(problems, "; ")))
|
||||
}
|
||||
resp, err := f.response(ex.Response, req)
|
||||
if err != nil {
|
||||
return nil, f.fail(fmt.Errorf("tide: upstream exchange %d: %w", idx, err))
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// Verify reports every mismatch, every extra request and every recorded
|
||||
// exchange that was never requested. It returns nil when the backend sent
|
||||
// exactly the recorded requests.
|
||||
func (f *UpstreamFake) Verify() error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
errs := slices.Clone(f.errs)
|
||||
for i := f.next; i < len(f.exchanges); i++ {
|
||||
r := f.exchanges[i].Request
|
||||
errs = append(errs, fmt.Errorf("tide: upstream exchange %d unconsumed: %s %s", i, r.Method, r.URL))
|
||||
}
|
||||
return errors.Join(errs...)
|
||||
}
|
||||
|
||||
func (f *UpstreamFake) fail(err error) error {
|
||||
f.mu.Lock()
|
||||
f.errs = append(f.errs, err)
|
||||
f.mu.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
func (f *UpstreamFake) compare(want UpstreamRequest, got *http.Request, body []byte) []string {
|
||||
var problems []string
|
||||
if !strings.EqualFold(want.Method, got.Method) {
|
||||
problems = append(problems, fmt.Sprintf("method: want %s, got %s", want.Method, got.Method))
|
||||
}
|
||||
rawURL, err := f.store.Expand(want.URL)
|
||||
if err != nil {
|
||||
return append(problems, "url: "+err.Error())
|
||||
}
|
||||
wu, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return append(problems, "url: "+err.Error())
|
||||
}
|
||||
gu := got.URL
|
||||
if !strings.EqualFold(wu.Scheme, gu.Scheme) {
|
||||
problems = append(problems, fmt.Sprintf("scheme: want %s, got %s", wu.Scheme, gu.Scheme))
|
||||
}
|
||||
if !strings.EqualFold(wu.Host, gu.Host) {
|
||||
problems = append(problems, fmt.Sprintf("host: want %s, got %s", wu.Host, gu.Host))
|
||||
}
|
||||
if wu.EscapedPath() != gu.EscapedPath() && wu.Path != gu.Path {
|
||||
problems = append(problems, fmt.Sprintf("path: want %s, got %s", wu.EscapedPath(), gu.EscapedPath()))
|
||||
}
|
||||
problems = append(problems, compareQuery(wu.Query(), gu.Query())...)
|
||||
|
||||
for _, name := range UpstreamCompareHeaders {
|
||||
wv, err := f.store.Expand(headerValue(want.Headers, name))
|
||||
if err != nil {
|
||||
problems = append(problems, "header "+name+": "+err.Error())
|
||||
continue
|
||||
}
|
||||
gv := got.Header.Get(name)
|
||||
if wv == gv {
|
||||
continue
|
||||
}
|
||||
if isCredentialHeader(name) {
|
||||
problems = append(problems, fmt.Sprintf("header %s: value differs", name))
|
||||
continue
|
||||
}
|
||||
problems = append(problems, fmt.Sprintf("header %s: want %q, got %q", name, wv, gv))
|
||||
}
|
||||
|
||||
wantBody, err := f.store.Expand(want.Body)
|
||||
if err != nil {
|
||||
return append(problems, "body: "+err.Error())
|
||||
}
|
||||
jsonBody := isJSONContentType(want.Headers) || isJSONContentType(map[string]string{"Content-Type": got.Header.Get("Content-Type")})
|
||||
switch {
|
||||
case jsonBody && (len(wantBody) > 0 || len(body) > 0):
|
||||
if diffs := diffJSON([]byte(wantBody), body); len(diffs) > 0 {
|
||||
d := diffs[0]
|
||||
problems = append(problems, fmt.Sprintf("body %s: want %s, got %s", d.Path, d.Expected, d.Actual))
|
||||
}
|
||||
case wantBody != string(body):
|
||||
problems = append(problems, fmt.Sprintf("body: want %d bytes, got %d bytes", len(wantBody), len(body)))
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
func compareQuery(want, got url.Values) []string {
|
||||
var problems []string
|
||||
keys := make([]string, 0, len(want)+len(got))
|
||||
for k := range want {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
for k := range got {
|
||||
if _, ok := want[k]; !ok {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
}
|
||||
slices.Sort(keys)
|
||||
for _, k := range keys {
|
||||
wv, gv := slices.Clone(want[k]), slices.Clone(got[k])
|
||||
slices.Sort(wv)
|
||||
slices.Sort(gv)
|
||||
if !slices.Equal(wv, gv) {
|
||||
problems = append(problems, fmt.Sprintf("query %s: want %q, got %q", k, wv, gv))
|
||||
}
|
||||
}
|
||||
return problems
|
||||
}
|
||||
|
||||
func (f *UpstreamFake) response(r UpstreamResponse, req *http.Request) (*http.Response, error) {
|
||||
h := http.Header{}
|
||||
for k, v := range r.Headers {
|
||||
ev, err := f.store.Expand(v)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("response header %s: %w", k, err)
|
||||
}
|
||||
h.Set(k, ev)
|
||||
}
|
||||
body, err := f.store.Expand(r.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("response body: %w", err)
|
||||
}
|
||||
return &http.Response{
|
||||
Status: fmt.Sprintf("%d %s", r.Status, http.StatusText(r.Status)),
|
||||
StatusCode: r.Status,
|
||||
Proto: "HTTP/1.1",
|
||||
ProtoMajor: 1,
|
||||
ProtoMinor: 1,
|
||||
Header: h,
|
||||
Body: io.NopCloser(strings.NewReader(body)),
|
||||
ContentLength: int64(len(body)),
|
||||
Request: req,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func isCredentialHeader(name string) bool {
|
||||
return strings.EqualFold(name, "Authorization") || strings.EqualFold(name, "X-Api-Key")
|
||||
}
|
||||
|
||||
// redactURL drops the query, which may carry a vendor token.
|
||||
func redactURL(u *url.URL) string {
|
||||
if u == nil {
|
||||
return ""
|
||||
}
|
||||
c := *u
|
||||
c.RawQuery = ""
|
||||
c.User = nil
|
||||
return c.String()
|
||||
}
|
||||
Reference in New Issue
Block a user