netbird/proxy/internal/appsec/client.go

// Package appsec implements the CrowdSec AppSec (WAF) side of the remediation
// component protocol: each inspected HTTP request is mirrored to the Security
// Engine's AppSec endpoint, which replies with an allow / ban / captcha verdict
// for that request.
//
// This is a separate endpoint from the LAPI decision stream used by the
// crowdsec package: LAPI answers "is this IP known bad", AppSec answers "is
// this request an attack". The two are configured and enabled independently.
package appsec

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/netip"
	"net/url"
	"strings"
	"sync"
	"time"

	log "github.com/sirupsen/logrus"

	"github.com/netbirdio/netbird/proxy/internal/netutil"
	"github.com/netbirdio/netbird/proxy/internal/restrict"
)

// Header names the AppSec component reads off the mirrored request. IP, URI and
// Verb are mandatory: the engine answers 500 when any of them is missing.
const (
	headerAPIKey        = "X-Crowdsec-Appsec-Api-Key" //nolint:gosec // G101: a header name, not a credential
	headerIP            = "X-Crowdsec-Appsec-Ip"
	headerURI           = "X-Crowdsec-Appsec-Uri"
	headerVerb          = "X-Crowdsec-Appsec-Verb"
	headerHost          = "X-Crowdsec-Appsec-Host"
	headerUserAgent     = "X-Crowdsec-Appsec-User-Agent"
	headerHTTPVersion   = "X-Crowdsec-Appsec-Http-Version"
	headerTransactionID = "X-Crowdsec-Appsec-Transaction-Id"
)

// headerPrefix covers every protocol header. Any client-supplied header in this
// namespace is dropped before forwarding so a caller cannot influence the
// engine's view of its own address, or replay an API key.
const headerPrefix = "X-Crowdsec-Appsec-"

// Remediation actions the engine can return.
const (
	actionAllow   = "allow"
	actionBan     = "ban"
	actionCaptcha = "captcha"
)

const (
	// DefaultTimeout matches the 200ms budget CrowdSec's remediation component
	// spec sets for the blocking AppSec call.
	DefaultTimeout = 200 * time.Millisecond
	// MinTimeout and MaxTimeout bound the configured inspection timeout.
	// Inspection is synchronous, so the upper bound is what keeps a
	// mis-set value from parking every request to an inspected service on a
	// slow engine; the lower bound keeps the call from timing out before the
	// engine can realistically answer. Mirrors the per-middleware bounds the
	// proxy already applies to in-path calls.
	MinTimeout = 10 * time.Millisecond
	MaxTimeout = 5 * time.Second
	// DefaultMaxBodyBytes caps the request body mirrored to the engine.
	// Requests with a larger body are inspected on headers and URI only.
	DefaultMaxBodyBytes int64 = 64 << 10
	// DefaultMaxConcurrent bounds inspections in flight toward the engine. The
	// point is to fail fast instead of parking a goroutine per request for the
	// whole timeout once the engine is saturated: a slow engine otherwise turns
	// a traffic burst into a pile of waiters that all time out anyway. Sized so
	// a healthy engine (single-digit milliseconds per call) never reaches it.
	DefaultMaxConcurrent = 256
	// MaxConcurrentLimit is the ceiling for that bound.
	MaxConcurrentLimit = 4096
	// MaxBodyBytesLimit is the ceiling for that cap. A single request can hold
	// this much in memory; the shared Budget is what bounds the total across
	// concurrent requests. Matches the proxy-wide body-capture ceiling.
	MaxBodyBytesLimit int64 = 8 << 20
	// maxResponseBytes bounds how much of a verdict response is read. The
	// engine answers with a two-field JSON object, so anything beyond this is
	// not a response we can act on.
	maxResponseBytes int64 = 4 << 10
)

// Reasons the request body was not mirrored. Reported so an access-log reader
// can distinguish "inspected and clean" from "never inspected", and so an
// oversize opt-out is visible rather than silent.
const (
	BypassOversize = "oversize"
	BypassUpgrade  = "upgrade"
	BypassDisabled = "disabled"
	BypassBudget   = "budget_exhausted"
)

// ErrUnavailable reports that the engine could not produce a verdict: the call
// failed, timed out, or the engine rejected it (401 bad key, 500 malformed).
// Distinguished from a block verdict so the caller can apply the per-service
// mode: enforce fails closed, observe allows.
var ErrUnavailable = errors.New("appsec engine unavailable")

// Config configures a Client.
type Config struct {
	// URL is the AppSec endpoint, e.g. http://127.0.0.1:7422/.
	URL string
	// APIKey is the CrowdSec bouncer API key. The AppSec component validates it
	// against LAPI, so the same key used for the decision stream works here.
	APIKey string
	// Timeout bounds a single inspection call. Zero means DefaultTimeout.
	Timeout time.Duration
	// MaxBodyBytes caps the mirrored request body. Zero means
	// DefaultMaxBodyBytes; negative disables body forwarding entirely.
	MaxBodyBytes int64
	// MaxConcurrent bounds inspections in flight toward the engine. Zero means
	// DefaultMaxConcurrent; negative disables the bound.
	MaxConcurrent int
	// Budget bounds the total body buffering in flight across all inspected
	// requests. Nil disables that ceiling, which leaves the worst case at
	// MaxBodyBytes times the concurrent request count; callers serving
	// untrusted traffic should share the proxy-wide capture budget here.
	Budget Budget
	Logger *log.Entry
}

// Budget is the shared allowance for in-flight body buffering. Acquire reports
// whether n bytes could be reserved; every successful Acquire is matched by a
// Release of the same n. Satisfied by the proxy's capture budget, so AppSec and
// the middleware body tap draw down one pool rather than two independent ones.
type Budget interface {
	Acquire(n int64) bool
	Release(n int64)
}

// Client mirrors HTTP requests to a CrowdSec AppSec endpoint. It holds no
// per-service state and is safe for concurrent use.
type Client struct {
	url          string
	apiKey       string
	maxBodyBytes int64
	// sem bounds in-flight inspections. Nil when the bound is disabled.
	sem    chan struct{}
	budget Budget
	http   *http.Client
	logger *log.Entry
}

// New validates the config and returns a Client. The endpoint is not contacted
// here: the engine may come up after the proxy.
func New(cfg Config) (*Client, error) {
	if cfg.URL == "" {
		return nil, errors.New("appsec url is empty")
	}
	if cfg.APIKey == "" {
		return nil, errors.New("appsec api key is empty")
	}
	parsed, err := url.Parse(cfg.URL)
	if err != nil {
		return nil, fmt.Errorf("parse appsec url: %w", err)
	}
	if parsed.Scheme != "http" && parsed.Scheme != "https" {
		return nil, fmt.Errorf("appsec url scheme %q is not http(s)", parsed.Scheme)
	}
	if parsed.Host == "" {
		return nil, errors.New("appsec url has no host")
	}

	logger := cfg.Logger
	if logger == nil {
		logger = log.NewEntry(log.StandardLogger())
	}

	timeout := cfg.Timeout
	switch {
	case timeout <= 0:
		timeout = DefaultTimeout
	case timeout < MinTimeout:
		logger.Warnf("appsec timeout %s is below the minimum, using %s", timeout, MinTimeout)
		timeout = MinTimeout
	case timeout > MaxTimeout:
		logger.Warnf("appsec timeout %s exceeds the maximum, using %s", timeout, MaxTimeout)
		timeout = MaxTimeout
	}

	// A negative cap is meaningful: forward no body at all.
	maxBody := cfg.MaxBodyBytes
	switch {
	case maxBody == 0:
		maxBody = DefaultMaxBodyBytes
	case maxBody > MaxBodyBytesLimit:
		logger.Warnf("appsec max body %d exceeds the maximum, using %d", maxBody, MaxBodyBytesLimit)
		maxBody = MaxBodyBytesLimit
	}

	maxConcurrent := cfg.MaxConcurrent
	switch {
	case maxConcurrent == 0:
		maxConcurrent = DefaultMaxConcurrent
	case maxConcurrent > MaxConcurrentLimit:
		logger.Warnf("appsec max concurrent %d exceeds the maximum, using %d", maxConcurrent, MaxConcurrentLimit)
		maxConcurrent = MaxConcurrentLimit
	}
	var sem chan struct{}
	if maxConcurrent > 0 {
		sem = make(chan struct{}, maxConcurrent)
	}

	return &Client{
		url:          cfg.URL,
		apiKey:       cfg.APIKey,
		maxBodyBytes: maxBody,
		sem:          sem,
		budget:       cfg.Budget,
		logger:       logger,
		http: &http.Client{
			Timeout: timeout,
			Transport: &http.Transport{
				MaxIdleConns:        100,
				MaxIdleConnsPerHost: 32,
				IdleConnTimeout:     90 * time.Second,
			},
		},
	}, nil
}

// Request is one inspection request.
type Request struct {
	// HTTP is the in-flight client request. Inspect buffers and restores its
	// body, so the request stays forwardable afterwards.
	HTTP *http.Request
	// ClientIP is the resolved client address (after trusted-proxy handling).
	ClientIP netip.Addr
	// TransactionID correlates the engine's alert with the proxy's access log
	// entry. Empty lets the engine generate its own UUID.
	TransactionID string
	// RedactBodyFields lists form fields whose values are replaced before the
	// body is mirrored. Used to keep credentials submitted to the proxy's own
	// login form out of the engine while still inspecting the rest.
	RedactBodyFields []string
	// RedactHeaders, RedactCookies and RedactQueryParams name the credentials
	// the proxy already withholds from backends: the header-auth values, its
	// session cookie, and the OIDC session token. The engine logs and alerts on
	// what it inspects, so mirroring them there would reintroduce the leak the
	// upstream strippers exist to prevent. Only the values are replaced, so the
	// surrounding headers, cookies and query stay inspectable.
	RedactHeaders     []string
	RedactCookies     []string
	RedactQueryParams []string
}

// Result is the outcome of an inspection.
type Result struct {
	Verdict restrict.Verdict
	// BodyBypass names why the request body was not mirrored, empty when it
	// was (or when the request had none). The engine still saw the headers and
	// URI, so this is a coverage note, not a failure.
	BodyBypass string
	// Release returns the buffered body's budget reservation. Never nil, so it
	// is always safe to defer. It must run only once the request has been
	// served, not when Inspect returns: the buffer stays alive as r.Body for
	// the backend to read, so releasing earlier would let the budget admit
	// buffering that is still resident.
	Release func()
}

// noopRelease is the Release for inspections that reserved no budget.
func noopRelease() {}

// Inspect mirrors r to the AppSec engine and returns its verdict. A nil error
// with restrict.Allow means the request passed. On failure it returns
// DenyAppSecUnavailable wrapped with ErrUnavailable; the caller decides whether
// that blocks, based on the per-service mode.
func (c *Client) Inspect(ctx context.Context, req Request) (Result, error) {
	if c == nil {
		return Result{Verdict: restrict.DenyAppSecUnavailable, Release: noopRelease}, ErrUnavailable
	}
	if req.HTTP == nil {
		return Result{Verdict: restrict.DenyAppSecUnavailable, Release: noopRelease}, fmt.Errorf("%w: nil request", ErrUnavailable)
	}

	// release is carried out to the caller rather than deferred here: the
	// buffered body outlives this call as r.Body.
	if !c.acquireSlot() {
		// Deny rather than wave through: a flood must not be a way to switch
		// inspection off. Enforce blocks, observe logs and allows, exactly as
		// for an unreachable engine.
		return Result{Verdict: restrict.DenyAppSecUnavailable, Release: noopRelease},
			fmt.Errorf("%w: %d inspections already in flight", ErrUnavailable, cap(c.sem))
	}
	defer c.releaseSlot()

	body, bypass, release, err := c.readBody(req)
	if err != nil {
		return Result{Verdict: restrict.DenyAppSecUnavailable, Release: release}, fmt.Errorf("%w: read body: %w", ErrUnavailable, err)
	}

	outbound, err := c.buildRequest(ctx, req, body)
	if err != nil {
		return Result{Verdict: restrict.DenyAppSecUnavailable, BodyBypass: bypass, Release: release}, fmt.Errorf("%w: %w", ErrUnavailable, err)
	}

	resp, err := c.http.Do(outbound)
	if err != nil {
		return Result{Verdict: restrict.DenyAppSecUnavailable, BodyBypass: bypass, Release: release}, fmt.Errorf("%w: %w", ErrUnavailable, err)
	}
	defer func() {
		// Drain before closing. net/http only returns a connection to the idle
		// pool once its body is read to EOF; closing with bytes outstanding
		// discards it. Every verdict carries a JSON body, so skipping this
		// would cost a fresh handshake per inspected request, inside the
		// timeout budget.
		if _, err := io.Copy(io.Discard, io.LimitReader(resp.Body, maxResponseBytes)); err != nil {
			c.logger.Tracef("drain appsec response body: %v", err)
		}
		if err := resp.Body.Close(); err != nil {
			c.logger.Tracef("close appsec response body: %v", err)
		}
	}()

	verdict, err := c.verdict(resp)
	return Result{Verdict: verdict, BodyBypass: bypass, Release: release}, err
}

// acquireSlot takes an in-flight slot without blocking, reporting false when
// the engine is already at capacity.
func (c *Client) acquireSlot() bool {
	if c.sem == nil {
		return true
	}
	select {
	case c.sem <- struct{}{}:
		return true
	default:
		return false
	}
}

// releaseSlot returns the slot. Scoped to the engine call, not the request: the
// buffered body outlives the call but the engine's attention does not.
func (c *Client) releaseSlot() {
	if c.sem == nil {
		return
	}
	select {
	case <-c.sem:
	default:
	}
}

// readBody buffers the body so it can be mirrored, always restoring it on the
// original request. Returns nil when there is no body to forward: no body at
// all, an upgrade request, or a body over the cap. A login form is forwarded
// with its credential values redacted rather than suppressed.
// release is never nil; the caller invokes it once the request has been served.
func (c *Client) readBody(req Request) (body []byte, bypass string, release func(), err error) {
	r := req.HTTP
	if r.Body == nil || r.Body == http.NoBody {
		return nil, "", noopRelease, nil
	}
	if c.maxBodyBytes < 0 {
		return nil, BypassDisabled, noopRelease, nil
	}
	// A genuine upgrade request carries no body to inspect (net/http hands us
	// http.NoBody, caught above); the hijacked stream is reached through
	// Hijacker, never r.Body. The test has to be the forwarder's own, because a
	// looser one would skip inspection for requests the forwarder still
	// delivers to the backend with their body intact.
	if netutil.IsUpgradeRequest(r.Header) {
		return nil, BypassUpgrade, noopRelease, nil
	}
	// A Content-Length over the cap is known to be too large before reading.
	if r.ContentLength > c.maxBodyBytes {
		return nil, BypassOversize, noopRelease, nil
	}

	// Reserve the whole cap rather than the eventual length: the reservation
	// has to be made before the body is read, and until then the only bound
	// known is the cap. Skipping inspection when the pool is drained keeps a
	// burst of large bodies from being an out-of-memory lever; the bypass is
	// recorded so the gap in coverage is visible.
	release = noopRelease
	if c.budget != nil {
		if !c.budget.Acquire(c.maxBodyBytes) {
			c.logger.Debugf("appsec buffer budget exhausted, inspecting headers and URI only")
			return nil, BypassBudget, noopRelease, nil
		}
		var once sync.Once
		release = func() { once.Do(func() { c.budget.Release(c.maxBodyBytes) }) }
	}

	buffered, oversize, err := bufferBody(r, c.maxBodyBytes)
	if err != nil {
		// bufferBody restored r.Body from the bytes it did read, so the
		// reservation stays held until the caller releases it.
		return nil, "", release, err
	}
	// An oversize body was only partially read: a truncated prefix changes the
	// engine's verdict in both directions, so inspect headers and URI only.
	if oversize {
		return nil, BypassOversize, release, nil
	}
	return redactFormFields(r.Header.Get("Content-Type"), buffered, req.RedactBodyFields), "", release, nil
}

// buildRequest assembles the mirrored request. Per the protocol it is a GET
// when there is no body and a POST otherwise; bytes.Reader gives the outbound
// request an accurate Content-Length, which the engine relies on to read the
// body at all.
func (c *Client) buildRequest(ctx context.Context, req Request, body []byte) (*http.Request, error) {
	method := http.MethodGet
	var payload io.Reader
	if len(body) > 0 {
		method = http.MethodPost
		payload = bytes.NewReader(body)
	}

	outbound, err := http.NewRequestWithContext(ctx, method, c.url, payload)
	if err != nil {
		return nil, fmt.Errorf("build appsec request: %w", err)
	}

	r := req.HTTP
	copyInspectableHeaders(outbound.Header, r.Header)
	redactSecrets(outbound.Header, req)

	outbound.Header.Set(headerAPIKey, c.apiKey)
	outbound.Header.Set(headerIP, req.ClientIP.Unmap().String())
	outbound.Header.Set(headerURI, mirroredURI(r.URL, req.RedactQueryParams))
	outbound.Header.Set(headerVerb, r.Method)
	outbound.Header.Set(headerHost, r.Host)
	if ua := r.UserAgent(); ua != "" {
		outbound.Header.Set(headerUserAgent, ua)
	}
	outbound.Header.Set(headerHTTPVersion, httpVersion(r))
	if req.TransactionID != "" {
		outbound.Header.Set(headerTransactionID, req.TransactionID)
	}
	return outbound, nil
}

// verdict maps the engine's response to a restrict.Verdict. 200 is a pass and
// 401/500 are engine-side failures; every other status carries a remediation in
// the body. The blocked status code is operator-configurable
// (blocked_http_code), so the action field decides, not the status.
func (c *Client) verdict(resp *http.Response) (restrict.Verdict, error) {
	switch resp.StatusCode {
	case http.StatusUnauthorized:
		return restrict.DenyAppSecUnavailable, fmt.Errorf("%w: rejected api key", ErrUnavailable)
	case http.StatusInternalServerError:
		return restrict.DenyAppSecUnavailable, fmt.Errorf("%w: engine error", ErrUnavailable)
	}

	// Every status, 200 included, has to carry a decodable remediation. Taking a
	// bare 200 as a pass would mean a URL pointing at anything that answers 200
	// (a health endpoint, a load balancer's default page) silently allows every
	// request while the service reports itself as enforcing.

	var decoded struct {
		Action string `json:"action"`
	}
	if err := json.NewDecoder(io.LimitReader(resp.Body, maxResponseBytes)).Decode(&decoded); err != nil {
		// Every remediation carries a decodable action, so a response without
		// one is not a verdict: most often the URL points at something that is
		// not the AppSec endpoint, which answers 404 with HTML. Reported as
		// unavailable rather than a ban so the access log names the real fault
		// instead of sending an operator hunting for a rule that never fired.
		// Enforce still blocks either way; only the recorded reason differs.
		return restrict.DenyAppSecUnavailable, fmt.Errorf("%w: undecodable response (status %d): %w", ErrUnavailable, resp.StatusCode, err)
	}

	switch decoded.Action {
	case actionAllow:
		return restrict.Allow, nil
	case actionCaptcha:
		return restrict.DenyAppSecCaptcha, nil
	case actionBan:
		return restrict.DenyAppSecBan, nil
	case "":
		// Decodable JSON without a remediation is not a verdict either: the
		// endpoint answered, but not as the engine. Same reasoning as an
		// undecodable body, and the same reason to point at configuration.
		return restrict.DenyAppSecUnavailable,
			fmt.Errorf("%w: response carried no remediation (status %d)", ErrUnavailable, resp.StatusCode)
	default:
		// A remediation we do not implement still means the engine flagged the
		// request, so deny.
		c.logger.Debugf("unknown appsec action %q (status %d), treating as ban", decoded.Action, resp.StatusCode)
		return restrict.DenyAppSecBan, nil
	}
}

// copyInspectableHeaders copies the client's headers, which are what the WAF
// rules actually match on, dropping hop-by-hop headers that describe the
// proxy-to-engine connection rather than the client request, and any header in
// the AppSec protocol namespace.
func copyInspectableHeaders(dst, src http.Header) {
	for name, values := range src {
		if hopByHopHeaders[http.CanonicalHeaderKey(name)] {
			continue
		}
		if strings.HasPrefix(http.CanonicalHeaderKey(name), headerPrefix) {
			continue
		}
		dst[http.CanonicalHeaderKey(name)] = append([]string(nil), values...)
	}
	// Content-Length describes the mirrored payload, not the client's: net/http
	// sets it from the body we actually attach. Content-Type is kept either way
	// so rules matching on it still fire when the body was not forwarded.
	dst.Del("Content-Length")
}

// redactSecrets replaces the credential values the proxy withholds from
// backends, so the mirrored copy does not carry them either.
func redactSecrets(dst http.Header, req Request) {
	for _, name := range req.RedactHeaders {
		// Presence, not Get: a header whose first value is empty still carries
		// its later values to the engine, while the upstream strip deletes the
		// name outright. Set collapses every value into the placeholder.
		if len(dst.Values(name)) > 0 {
			dst.Set(name, redactedPlaceholder)
		}
	}
	// Every Cookie line, not just the first: a client may send several, and Get
	// would leave the session cookie in any later one mirrored in the clear.
	if cookies := dst.Values("Cookie"); len(cookies) > 0 {
		redacted := make([]string, len(cookies))
		for i, cookie := range cookies {
			redacted[i] = redactCookieHeader(cookie, req.RedactCookies)
		}
		dst["Cookie"] = redacted
	}
}

// mirroredURI renders the request target for the URI header, with the named
// query parameter values replaced.
func mirroredURI(u *url.URL, redactParams []string) string {
	uri := u.RequestURI()
	if u.RawQuery == "" || len(redactParams) == 0 {
		return uri
	}
	redacted := redactQuery(u.RawQuery, redactParams)
	if redacted == u.RawQuery {
		return uri
	}
	// RequestURI is path + "?" + RawQuery; swap only the query part so the
	// path keeps its original encoding.
	return strings.TrimSuffix(uri, u.RawQuery) + redacted
}

var hopByHopHeaders = map[string]bool{
	"Connection":          true,
	"Keep-Alive":          true,
	"Proxy-Authenticate":  true,
	"Proxy-Authorization": true,
	"Proxy-Connection":    true,
	"Te":                  true,
	"Trailer":             true,
	"Transfer-Encoding":   true,
	"Upgrade":             true,
}

// httpVersion renders the two-digit form the engine parses ("11", "20").
func httpVersion(r *http.Request) string {
	major, minor := r.ProtoMajor, r.ProtoMinor
	if major < 0 || major > 9 || minor < 0 || minor > 9 {
		return ""
	}
	return fmt.Sprintf("%d%d", major, minor)
}