msa-agent/internal/runtime/executor.go

package runtime

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"

	"fun.chick/msa-agent/internal/egress"
	"fun.chick/msa-agent/internal/policy"
	"fun.chick/msa-agent/internal/wire"
)

// Refusal is classified, and the classification is part of the contract with the operator. A node
// with a drifted clock and a node under attack produce the same outward symptom; collapsing both
// into "denied" guarantees the operator will respond to the wrong one.
type Refusal struct {
	Class string
	Path  string
	Err   error
}

func (r Refusal) Error() string { return fmt.Sprintf("%s: %v", r.Class, r.Err) }

// Refusal classes, mirrored by the relay's ProblemDetails codes so one string is searchable across
// the agent log, the relay log and the phone.
const (
	ClassBadEnvelope = "bad_envelope"
	ClassSignature   = "bad_signature"
	ClassExpired     = "expired"
	ClassReplay      = "replay"
	ClassPolicy      = "policy_denied"
	ClassNoConfirm   = "confirmation_missing"
	ClassBadConfirm  = "confirmation_mismatch"
	ClassRateLimited = "rate_limited"
	ClassTooLarge    = "payload_too_large"
	ClassUpstream    = "upstream_error"
	ClassLoopback    = "egress_violation"
)

// Executor turns a decrypted InnerRequest into one loopback HTTP call.
//
// It holds the only HTTP client in the process, built by egress.NewClient, so there is no path from
// a request body to a socket that is not the loopback one.
type Executor struct {
	client  *http.Client
	target  egress.Target
	pol     *policy.Policy
	replay  *ReplayCache
	audit   *Audit
	clock   func() time.Time
	limiter *rateLimiter
}

// Config is what the agent needs to build an Executor. Port is a number and nothing else: no host,
// by the same reasoning as egress.Target.
type Config struct {
	Port            int
	Timeout         time.Duration
	Audit           *Audit
	Clock           func() time.Time
	DangerousPerMin int
}

// NewExecutor wires the policy, the loopback client and the replay cache together.
//
// It fails closed: an unusable policy or an unbuildable client is a startup error, because an agent
// that starts with a permissive fallback would be indistinguishable from one that is working.
func NewExecutor(pol *policy.Policy, cfg Config) (*Executor, error) {
	if pol == nil {
		return nil, errors.New("runtime: policy is nil")
	}
	target, err := egress.NewTarget(cfg.Port)
	if err != nil {
		return nil, err
	}
	timeout := cfg.Timeout
	if timeout <= 0 {
		timeout = 20 * time.Second
	}
	client, err := egress.NewClient(target, timeout)
	if err != nil {
		return nil, err
	}
	// The cache ttl is the skew window, not an independent knob. An id older than the widest
	// acceptable timestamp can never be accepted again, so keeping it costs memory and buys nothing.
	replay, err := NewReplayCache(pol.Limits.ReplayCacheSize,
		time.Duration(pol.Limits.ClockSkewSeconds)*time.Second)
	if err != nil {
		return nil, err
	}
	clock := cfg.Clock
	if clock == nil {
		clock = time.Now
	}
	perMinute := cfg.DangerousPerMin
	if perMinute <= 0 {
		perMinute = pol.Limits.DangerousWritesPerMinute
	}
	return &Executor{
		client:  client,
		target:  target,
		pol:     pol,
		replay:  replay,
		audit:   cfg.Audit,
		clock:   clock,
		limiter: newRateLimiter(perMinute, time.Minute),
	}, nil
}

// Result is what comes back from the local SimAdmin, or why it did not.
type Result struct {
	Status int
	Body   []byte
}

// Execute validates, then performs, exactly one request.
//
// Order matters and is not cosmetic. The replay and freshness gates run before the policy decision
// so a denied request still burns its request id: otherwise an attacker could probe the allowlist
// by resubmitting one signed request with a dozen paths. The intent is checked after the policy
// decision because the expected digest depends on the path that survived it.
func (e *Executor) Execute(ctx context.Context, envelope wire.Envelope, inner wire.InnerRequest) (Result, *Refusal) {
	now := e.clock()
	if err := Freshness(envelope.TS, uint64(now.Unix()), e.pol.Limits.ClockSkewSeconds); err != nil {
		return Result{}, e.refuse(envelope, inner, ClassExpired, err)
	}
	if e.replay.Seen(envelope.Req, now) {
		return Result{}, e.refuse(envelope, inner, ClassReplay, fmt.Errorf("%w: %s", ErrReplay, envelope.Req))
	}

	method := strings.ToUpper(strings.TrimSpace(inner.Method))
	allowed, dangerous, reason := e.pol.Decision(method, inner.Path)
	if !allowed {
		return Result{}, e.refuse(envelope, inner, ClassPolicy, errors.New(reason))
	}
	if len(inner.Body) > e.pol.Limits.MaxRequestBytes {
		return Result{}, e.refuse(envelope, inner, ClassTooLarge,
			fmt.Errorf("body is %d bytes, limit is %d", len(inner.Body), e.pol.Limits.MaxRequestBytes))
	}

	if dangerous {
		// Ordering inside the dangerous path: rate limit first, then intent. A caller who can send
		// unlimited attempts can also brute-force a 32-byte digest in principle, and the limit is
		// the cheaper of the two checks to apply.
		if !e.limiter.Allow(now) {
			return Result{}, e.refuse(envelope, inner, ClassRateLimited, errors.New("too many dangerous writes"))
		}
		want, err := wire.IntentDigest(method, inner.Path, inner.Body)
		if err != nil {
			return Result{}, e.refuse(envelope, inner, ClassBadConfirm, err)
		}
		if len(inner.Intent) == 0 {
			return Result{}, e.refuse(envelope, inner, ClassNoConfirm, errors.New("dangerous write carries no confirmation intent"))
		}
		if !policy.RequiresConfirm(inner.Intent, want) {
			return Result{}, e.refuse(envelope, inner, ClassBadConfirm, errors.New("confirmation intent does not match the request being sent"))
		}
	}

	result, err := e.forward(ctx, method, inner.Path, inner.Body)
	if err != nil {
		class := ClassUpstream
		if errors.Is(err, ErrResponseTooLarge) {
			class = ClassTooLarge
		}
		return Result{}, e.refuse(envelope, inner, class, err)
	}
	e.record(envelope, method, inner.Path, result.Status, "ok")
	return result, nil
}

// forward performs the loopback call. The origin comes from egress.Target, which has no host field,
// and the path has already survived policy.Decision, so nothing here is assembled from parts that
// a caller could steer off the machine.
func (e *Executor) forward(ctx context.Context, method, path string, body []byte) (Result, error) {
	if !strings.HasPrefix(path, "/api/") {
		return Result{}, ClassLoopbackErr
	}
	var reader io.Reader
	if len(body) > 0 {
		reader = strings.NewReader(string(body))
	}
	request, err := http.NewRequestWithContext(ctx, method, e.target.Origin()+path, reader)
	if err != nil {
		return Result{}, fmt.Errorf("%w: %v", ClassLoopbackErr, err)
	}
	if len(body) > 0 {
		request.Header.Set("Content-Type", "application/json")
	}
	// No credential is attached here. The agent logs into the local SimAdmin itself on startup and
	// holds the session; a phone cannot use this path to present a node password it does not have.
	response, err := e.client.Do(request)
	if err != nil {
		return Result{}, fmt.Errorf("local SimAdmin unreachable: %w", err)
	}
	defer response.Body.Close()
	limited := io.LimitReader(response.Body, int64(e.pol.Limits.MaxResponseBytes)+1)
	payload, err := io.ReadAll(limited)
	if err != nil {
		return Result{}, fmt.Errorf("reading local response: %w", err)
	}
	if len(payload) > e.pol.Limits.MaxResponseBytes {
		return Result{}, fmt.Errorf("%w: %s answered with more than %d bytes",
			ErrResponseTooLarge, path, e.pol.Limits.MaxResponseBytes)
	}
	return Result{Status: response.StatusCode, Body: payload}, nil
}

// ClassLoopbackErr is the named error used when a request could not even be constructed.
var ClassLoopbackErr = errors.New("request could not be built for the loopback target")

// ErrResponseTooLarge is returned when the local SimAdmin answers with more than the policy allows.
//
// The response is dropped rather than truncated. A truncated body is not a smaller answer, it is a
// broken one: JSON cut off at an arbitrary byte offset parses as nothing, and the phone would show
// a decoding failure that points at the app instead of at the node that answered with 3 MB.
var ErrResponseTooLarge = errors.New("local response exceeds the policy limit")

func (e *Executor) refuse(envelope wire.Envelope, inner wire.InnerRequest, class string, err error) *Refusal {
	e.record(envelope, strings.ToUpper(strings.TrimSpace(inner.Method)), inner.Path, 0, class)
	return &Refusal{Class: class, Path: inner.Path, Err: err}
}

func (e *Executor) record(envelope wire.Envelope, method, path string, status int, outcome string) {
	if e.audit == nil {
		return
	}
	e.audit.Write(Entry{
		At:      e.clock(),
		Node:    envelope.Node,
		Req:     envelope.Req,
		App:     envelope.App,
		Method:  method,
		Path:    path,
		Status:  status,
		Outcome: outcome,
	})
}

// rateLimiter is a fixed-window counter. Deliberately not a token bucket: the property wanted is
// "at most N dangerous writes in any 60 seconds", and a bucket's refill rate makes that harder to
// state and to test.
type rateLimiter struct {
	mu       sync.Mutex
	limit    int
	window   time.Duration
	start    time.Time
	consumed int
}

func newRateLimiter(limit int, window time.Duration) *rateLimiter {
	return &rateLimiter{limit: limit, window: window}
}

func (r *rateLimiter) Allow(now time.Time) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.start.IsZero() || now.Sub(r.start) >= r.window {
		r.start = now
		r.consumed = 0
	}
	if r.consumed >= r.limit {
		return false
	}
	r.consumed++
	return true
}