msa-agent/cmd/msa-agent/main.go

// Command msa-agent runs on a SimAdmin node and lets a paired iPhone reach it through a public
// relay without any inbound path.
//
// It opens no listening socket. Every network action it takes is either an outbound HTTPS request to
// the relay or a loopback request to the SimAdmin on the same machine, which is the property that
// makes "no port mapping, no VPN" true rather than merely advertised.
package main

import (
	"context"
	"crypto/ed25519"
	"encoding/base64"
	"errors"
	"flag"
	"fmt"
	"log/slog"
	"os"
	"os/signal"
	"syscall"
	"time"

	"fun.chick/msa-agent/internal/agent"
	"fun.chick/msa-agent/internal/config"
	"fun.chick/msa-agent/internal/policy"
	"fun.chick/msa-agent/internal/runtime"
	"fun.chick/msa-agent/internal/transport"
)

// version is stamped in by the Makefile.
var version = "dev"

// Backoff bounds. The lower bound is what makes a relay outage survivable for the relay's operator,
// and the ceiling is what keeps a long outage from turning into a 30-second-per-retry queue.
const (
	minBackoff = time.Second
	maxBackoff = 60 * time.Second
)

func main() {
	var (
		configPath = flag.String("config", "/etc/msa-agent/config.json", "path to the agent config file")
		check      = flag.Bool("check", false, "validate the config and policy, then exit")
		once       = flag.Bool("once", false, "run a single poll cycle and exit")
	)
	flag.Parse()

	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))
	slog.SetDefault(logger)

	if err := run(*configPath, *check, *once); err != nil {
		// One line, then exit non-zero. systemd restarts it, and the journal keeps the reason.
		logger.Error("msa-agent exiting", "error", err)
		os.Exit(1)
	}
}

func run(configPath string, checkOnly, once bool) error {
	cfg, err := config.Load(configPath)
	if err != nil {
		return err
	}
	pol, err := policy.Load()
	if err != nil {
		return err
	}
	sign, err := decodeKey(cfg.SignKey, ed25519.PrivateKeySize, "signKey")
	if err != nil {
		return err
	}
	kx, err := decodeKey(cfg.KxKey, 32, "kxKey")
	if err != nil {
		return err
	}
	devices, err := deviceSet(cfg.Devices)
	if err != nil {
		return err
	}

	// The policy digest is logged at every startup and sent in the handshake. An operator comparing
	// one line across a fleet learns whether every node enforces the same surface; without it the
	// only way to know is to log into each box.
	slog.Info("msa-agent starting",
		"version", version,
		"node", cfg.NodeID,
		"localPort", cfg.LocalPort,
		"policy", pol.SHA256(),
		"devices", devices.Len())

	if checkOnly {
		slog.Info("configuration is valid", "relay", cfg.RelayURL)
		return nil
	}

	audit, err := runtime.OpenAudit(cfg.AuditPath)
	if err != nil {
		return err
	}
	defer audit.Close()

	exec, err := runtime.NewExecutor(pol, runtime.Config{
		Port:         cfg.LocalPort,
		Audit:        audit,
		NodePassword: cfg.NodePassword,
	})
	if err != nil {
		return err
	}
	node, err := agent.New(cfg.NodeID, agent.Keys{Sign: sign, Kx: kx}, devices, pol, exec)
	if err != nil {
		return err
	}
	client, err := transport.New(transport.Config{
		BaseURL:   cfg.RelayURL,
		NodeID:    cfg.NodeID,
		Token:     cfg.AgentToken,
		Signer:    sign,
		UserAgent: "msa-agent/" + version,
	})
	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	if once {
		return cycle(ctx, node, client, pol)
	}
	return loop(ctx, node, client, pol)
}

// loop is the agent's whole life: handshake, then poll forever, re-handshaking whenever the relay
// says the session is gone and backing off when the relay is unreachable.
//
// Errors are classified here rather than handled per call site, because the three situations look
// identical from inside a function and need three different responses: retry now, re-authenticate,
// or wait.
func loop(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy) error {
	backoff := minBackoff
	for {
		if ctx.Err() != nil {
			return nil
		}
		err := handshakeAndPoll(ctx, node, client, pol)
		switch {
		case err == nil:
			backoff = minBackoff
		case errors.Is(err, transport.ErrNoWork):
			backoff = minBackoff
		case errors.Is(err, transport.ErrUnauthorized), errors.Is(err, transport.ErrSessionExpired):
			// Re-handshake on the next iteration. The wait is what stops a bad token from becoming
			// a tight loop against the relay.
			slog.Warn("relay rejected the agent session, re-authenticating", "error", err)
			backoff = minBackoff * 5
		case errors.Is(err, transport.ErrRateLimited):
			// The relay said how long to wait; doubling its own backoff on top of that would be
			// the agent overriding an explicit instruction.
			backoff = client.RetryAfter()
		default:
			slog.Warn("poll cycle failed", "error", err, "retryIn", backoff)
			backoff = next(backoff)
		}
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(backoff):
		}
	}
}

func next(current time.Duration) time.Duration {
	doubled := current * 2
	if doubled > maxBackoff {
		return maxBackoff
	}
	return doubled
}

func handshakeAndPoll(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy) error {
	if _, err := client.Hello(ctx, version, pol.SHA256()); err != nil {
		return err
	}
	return cycle(ctx, node, client, pol)
}

// cycle drains one batch. Each request is handled and answered before the next is fetched, so a
// slow upstream request delays the queue rather than buffering it in memory.
func cycle(ctx context.Context, node *agent.Agent, client *transport.Client, pol *policy.Policy) error {
	batch, err := client.Poll(ctx)
	if err != nil {
		return err
	}
	if len(batch.Envelopes) == 0 {
		return transport.ErrNoWork
	}
	for _, raw := range batch.Envelopes {
		response, err := node.Handle(ctx, raw)
		if err != nil {
			// Addressed to another node, or malformed beyond answering. Logged, not fatal: one bad
			// envelope must not stop the batch.
			slog.Warn("envelope dropped", "error", err)
			continue
		}
		if err := client.Result(ctx, response); err != nil {
			return err
		}
	}
	return nil
}

func decodeKey(value string, size int, field string) ([]byte, error) {
	raw, err := base64.StdEncoding.Strict().DecodeString(value)
	if err != nil {
		return nil, fmt.Errorf("config: %s is not valid base64: %w", field, err)
	}
	if len(raw) != size {
		return nil, fmt.Errorf("config: %s must be %d bytes, got %d", field, size, len(raw))
	}
	return raw, nil
}

func deviceSet(devices []config.Device) (agent.StaticDevices, error) {
	set := make(agent.StaticDevices, len(devices))
	for _, device := range devices {
		public, err := decodeKey(device.SignKey, ed25519.PublicKeySize, fmt.Sprintf("device %s signKey", device.ID))
		if err != nil {
			return nil, err
		}
		set[device.ID] = ed25519.PublicKey(public)
	}
	return set, nil
}