msa-agent/internal/runtime/executor.go

package runtime

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"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"
	// ClassNodeAuth is reported when the agent cannot establish its own session against the local
	// SimAdmin. It is distinct from ClassUpstream because the phone's remedy differs: "the node is
	// misconfigured" calls for fixing the agent's password, while "the node errored" calls for
	// looking at SimAdmin itself.
	ClassNodeAuth = "node_auth_failed"
)

// 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
	auth    *NodeAuth
}

// 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
	// NodePassword is the credential the agent uses to log into the local SimAdmin itself. Empty
	// means the node is expected to be unprotected; a protected node then surfaces as
	// ClassNodeAuth rather than forwarding anonymous requests that will fail.
	NodePassword string
}

// 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 {
		// From the policy, not a constant here. The relay's deadline is compared against this number
		// when the policy is validated, and a comparison against a constant inside a binary is not one
		// anything can check.
		timeout = time.Duration(pol.Limits.AgentUpstreamTimeoutSeconds) * 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
	}
	auth, err := NewNodeAuth(client, target, AuthConfig{Password: cfg.NodePassword, Clock: clock})
	if err != nil {
		return nil, err
	}
	return &Executor{
		client:  client,
		target:  target,
		pol:     pol,
		replay:  replay,
		audit:   cfg.Audit,
		clock:   clock,
		limiter: newRateLimiter(perMinute, time.Minute),
		auth:    auth,
	}, 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)
	}
	// The phone's own deadline, checked before anything is executed.
	//
	// The relay answers 504 with "it may or may not have executed it; check the node before retrying",
	// which is honest — but it is only honest while it stays true. A batch is processed serially, so
	// the eighth envelope of a full mailbox can begin up to eight upstream timeouts after it was
	// signed; by then the phone has been told the request failed and the user may already have retried.
	// Executing it at that point turns an honest "unknown" into a duplicate SMS.
	//
	// The relay's deadline is the authority on when that happened, because the relay is the party that
	// starts the clock. Its value is therefore carried in the policy and asserted to be no larger than
	// the clock-skew window, which is what makes the freshness check above unable to mask this one.
	if age := int64(now.Unix()) - int64(envelope.TS); age > int64(e.pol.Limits.RequestTTLSeconds) {
		return Result{}, e.refuse(envelope, inner, ClassExpired,
			fmt.Errorf("%w: signed %ds ago, deadline is %ds", ErrExpired, age, e.pol.Limits.RequestTTLSeconds))
	}
	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
		switch {
		case errors.Is(err, ErrResponseTooLarge):
			class = ClassTooLarge
		case errors.Is(err, ErrAuthNoPassword), errors.Is(err, ErrAuthRejected), errors.Is(err, ErrAuthUnavailable):
			class = ClassNodeAuth
		}
		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.
//
// Authentication is established here, after every phone-facing gate: probing /api/auth/status on a
// request the policy will reject anyway would let a relay-fed caller use rejected commands as a
// login-attempt amplifier against a local lockout.
func (e *Executor) forward(ctx context.Context, method, path string, body []byte) (Result, error) {
	if !strings.HasPrefix(path, "/api/") {
		return Result{}, ClassLoopbackErr
	}
	if err := e.auth.Ensure(ctx); err != nil {
		return Result{}, err
	}
	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")
	}
	// The agent's own session, held from a login the phone cannot trigger or see. A node password
	// never traverses the relay in either direction.
	if cookie := e.auth.Cookie(); cookie != "" {
		request.Header.Set("Cookie", cookie)
	}
	response, err := e.client.Do(request)
	if err != nil {
		return Result{}, fmt.Errorf("local SimAdmin unreachable: %w", err)
	}
	defer response.Body.Close()
	if response.StatusCode == http.StatusUnauthorized || response.StatusCode == http.StatusForbidden {
		// The session died between Ensure and this call (SimAdmin restarted, cookie expired).
		// Drop it so the next command re-probes and re-logs in rather than 401ing until restart.
		// This command still answers with the node's refusal; retrying it here would turn every
		// genuine 401 from a node endpoint into a login attempt on the phone's behalf.
		e.auth.Invalidate()
	}
	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}
}

// record appends one audit entry, and reports a failure to the process log.
//
// The write error is not propagated to the caller: the audit log is evidence, not a control, and a
// full disk must not become a remote denial of service — a request that already succeeded would start
// failing because the log could not be appended to, which is the wrong thing to break.
//
// But it must not vanish either. Silently losing audit entries is the failure mode where an operator
// believes they have a record and does not, and the whole point of the log is to be able to answer
// "who asked for what" afterwards. So the error goes to the agent's own log, where a full disk shows
// up as a repeating line rather than as an absence.
func (e *Executor) record(envelope wire.Envelope, method, path string, status int, outcome string) {
	if e.audit == nil {
		return
	}
	if err := e.audit.Write(Entry{
		At:      e.clock(),
		Node:    envelope.Node,
		Req:     envelope.Req,
		App:     envelope.App,
		Method:  method,
		Path:    path,
		Status:  status,
		Outcome: outcome,
	}); err != nil {
		slog.Warn("audit entry could not be written", "error", err, "req", envelope.Req, "outcome", outcome)
	}
}

// rateLimiter is a sliding-window counter over the dangerous writes this node has performed.
//
// The property is "at most N dangerous writes in any 60 seconds", and the implementation has to be
// that. It used to be a fixed window — a counter reset on a calendar minute — which is a different
// property: spending the budget just before the boundary and again just after puts 2N writes inside a
// few milliseconds. The comment above the old code argued for a fixed window *because* "a bucket's
// refill rate makes it harder to state and to test", so the code contradicted its own stated reason;
// a token bucket would not have satisfied the property either.
//
// The relay's per-device budget had the same defect and was fixed the same way (relay/src/budget.js).
// Keeping the two in step matters more than either: they bound the same operation from two sides, and
// a node whose limiter is twice as permissive as the relay's is a discrepancy nobody would notice
// until an incident review.
//
// Memory is bounded by the limit — at most N timestamps — where the fixed window held one integer.
type rateLimiter struct {
	mu     sync.Mutex
	limit  int
	window time.Duration
	// accepted holds the timestamps of admitted writes, ascending. Anything older than the window is
	// dropped on the next call, so the slice never exceeds limit entries.
	accepted []time.Time
}

// TestDangerousWriteLimitDoesNotResetOnACalendarBoundary is the test for the property above: it spends
// the budget one second before a calendar-minute boundary and asserts nothing is admitted just after
// it. Written the obvious way — spend the budget, advance a minute, spend it again — the test passes
// against a fixed window too, because both implementations agree once everything has aged out.

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

// Allow reports whether a dangerous write may proceed, and records it if so.
//
// Unlike the relay's budget, this one does not return a retry hint, and that is a property of the wire
// format rather than an omission: the agent answers inside `InnerResponse`, which carries a status, an
// outcome class and a body, and the phone renders the class with a fixed sentence. There is nowhere to
// put a wait without changing the envelope, and a hint that no layer can deliver is a number nobody
// reads. The relay can return one because HTTP has a header for it.
func (r *rateLimiter) Allow(now time.Time) bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	cutoff := now.Add(-r.window)
	live := r.accepted[:0]
	for _, stamp := range r.accepted {
		if stamp.After(cutoff) {
			live = append(live, stamp)
		}
	}
	r.accepted = live
	if len(r.accepted) >= r.limit {
		return false
	}
	r.accepted = append(r.accepted, now)
	return true
}