mesh-host/internal/link/enrol.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

// The wire format shared with the control plane, which defines it separately because this binary
// requires nothing present and does not import it. A test on each side asserts the field names.
const (
	Exchange = "mesh"
	KeyEnrol = "enrol"
)

// QueueFor is the queue this node consumes from — the only one its account may read.
func QueueFor(node string) string { return "node." + node }

// EnrolRequest is what this node says when joining.
type EnrolRequest struct {
	Node      string `json:"node"`
	Secret    string `json:"secret"`
	PublicKey []byte `json:"public_key"`

	// OverlayKey is the public half of this node's key on the private network — a different key
	// from PublicKey above, generated at the same moment and for a different purpose.
	//
	// Sent with enrolment because the overlay is the first declaration a node receives, and the
	// mesh cannot compose it without this. Asking for it afterwards would mean a node is enrolled
	// and unreachable for a round trip, which is the state everything else here works to avoid.
	OverlayKey string `json:"overlay_key,omitempty"`

	// SealingKey is the public half of the key secrets are sealed to. A third key, and the
	// reasoning is the same one twice over: the mesh must be able to send this node something
	// nothing else can read, and it must never be able to read it either.
	SealingKey string `json:"sealing_key,omitempty"`

	// ServingKey is the public half of the key this node serves TLS with on its internal name.
	// The mesh signs a certificate binding it; the private half never leaves the machine, so
	// there is nothing to seal and nothing that could be stolen from the mesh's copy.
	ServingKey string `json:"serving_key,omitempty"`

	Profile map[string]any `json:"profile,omitempty"`
}

// EnrolReply is what the mesh says back.
type EnrolReply struct {
	Accepted bool `json:"accepted"`

	// TryAgain is the mesh saying it cannot answer right now — its store is restarting, or the
	// token is held for a moment by another enrolment — and that nothing was spent. The same
	// request is asked again (novox/hq issue 083).
	TryAgain bool `json:"try_again,omitempty"`

	Node  string `json:"node,omitempty"`
	Queue string `json:"queue,omitempty"`

	// What this node keeps so it can come back on its own. Without these a restart would need a
	// person with a new token, which would make disconnection a crisis rather than the ordinary
	// situation novox/hq ADR 0004 says it is.
	Password    string `json:"password,omitempty"`
	Broker      string `json:"broker,omitempty"`
	Fingerprint string `json:"fingerprint,omitempty"`
	Signer      []byte `json:"signer,omitempty"`

	Refusal string `json:"refusal,omitempty"`
}

// ErrRefused is what a node gets when the mesh will not have it.
var ErrRefused = errors.New("the mesh refused this enrolment")

// EnrolPatience is how long a node keeps asking while the mesh says "try again" — as long as the
// mesh holds a token for the one enrolment presenting it, so a node asking the whole time is never
// held off by its own earlier attempt.
const EnrolPatience = 2 * time.Minute

// AskAgainAfter is the pause between asks while the mesh says "try again".
const AskAgainAfter = 3 * time.Second

// ErrNotNow is a mesh that said "try again" for longer than this node would keep asking.
var ErrNotNow = errors.New("the mesh could not answer this enrolment")

// answered decides what one reply means: done, ask again, or stop with an error. Separate from the
// broker so it can be held to that by a test.
func answered(reply EnrolReply, asking time.Duration) (again bool, err error) {
	switch {
	case reply.Accepted:
		return false, nil
	case reply.TryAgain && asking < EnrolPatience:
		return true, nil
	case reply.TryAgain:
		// Said with what to do. The mesh holds the token for this attempt's keys for as long as
		// this node kept asking, so a new attempt — with keys of its own — waits that out first.
		return false, fmt.Errorf("%w for %s: %s. The token was not spent: wait about %s and run "+
			"enrol again with it; if it is then refused, issue a new one",
			ErrNotNow, EnrolPatience, reply.Refusal, EnrolPatience)
	default:
		return false, fmt.Errorf("%w: %s", ErrRefused, reply.Refusal)
	}
}

// Enrol presents this node's key and its one-time secret, and waits to be told it is known.
//
// The broker has already authenticated this connection: the account was created when the token
// was issued and the secret is its password. So this is not how the node gets in — it is what it
// says once it is in, and the secret travels again because the control plane must not have to ask
// the broker who connected.
func Enrol(ctx context.Context, address, pin, node, secret string, public []byte,
	overlayKey, sealingKey, servingKey string, profile map[string]any,
	timeout time.Duration) (EnrolReply, error) {

	config, err := PinnedConfig(pin)
	if err != nil {
		return EnrolReply{}, err
	}

	// The account name is the node's, and the password is the token's secret. Escaped because a
	// name or secret containing a colon or an at-sign would otherwise change which host this
	// connects to — a credential silently redirecting a connection is the worst shape this could
	// take.
	dsn := fmt.Sprintf("amqps://%s:%s@%s/",
		url.QueryEscape(node), url.QueryEscape(secret), address)

	conn, err := amqp.DialConfig(dsn, amqp.Config{
		TLSClientConfig: config,
		Dial:            amqp.DefaultDial(timeout),
	})
	if err != nil {
		if errors.Is(err, ErrWrongCertificate) {
			return EnrolReply{}, err
		}
		// Not quoted back: the DSN carries the one-time secret.
		return EnrolReply{}, fmt.Errorf("cannot reach the broker at %s as %s: %w", address, node, err)
	}
	defer conn.Close()

	channel, err := conn.Channel()
	if err != nil {
		return EnrolReply{}, err
	}
	defer channel.Close()

	// This node's own queue, which its account is scoped to and nothing else may read.
	queue, err := channel.QueueDeclare(QueueFor(node), true, false, false, false, nil)
	if err != nil {
		return EnrolReply{}, fmt.Errorf(
			"cannot declare this node's queue %s: %w", QueueFor(node), err)
	}

	replies, err := channel.Consume(queue.Name, "", true, false, false, false, nil)
	if err != nil {
		return EnrolReply{}, err
	}

	request := EnrolRequest{Node: node, Secret: secret, PublicKey: public,
		OverlayKey: overlayKey, SealingKey: sealingKey, ServingKey: servingKey, Profile: profile}
	body, err := json.Marshal(request)
	if err != nil {
		return EnrolReply{}, err
	}

	// Asked, and asked again with the same request while the mesh says "try again": the keys
	// this node generated are the ones it keeps, so the same request is the same enrolment, and
	// the mesh holds the token for it (novox/hq issue 083).
	ask := func() (string, error) {
		correlation := fmt.Sprintf("%s-%d", node, time.Now().UnixNano())
		publish, cancel := context.WithTimeout(ctx, timeout)
		defer cancel()
		if err := channel.PublishWithContext(publish, Exchange, KeyEnrol, false, false,
			amqp.Publishing{
				ContentType:   "application/json",
				CorrelationId: correlation,
				ReplyTo:       queue.Name,
				Body:          body,
			}); err != nil {
			return "", fmt.Errorf("cannot publish to the %s exchange: %w", Exchange, err)
		}
		return correlation, nil
	}
	correlation, err := ask()
	if err != nil {
		return EnrolReply{}, err
	}
	began := time.Now()

	// Waited for rather than assumed. A published message that nothing answers means the control
	// plane is not running, and a node that carried on regardless would believe it had joined a
	// mesh that has never heard of it.
	deadline := time.NewTimer(timeout)
	defer deadline.Stop()
	closed := conn.NotifyClose(make(chan *amqp.Error, 1))

	for {
		select {
		case <-ctx.Done():
			return EnrolReply{}, ctx.Err()
		case reason := <-closed:
			return EnrolReply{}, fmt.Errorf("the broker closed the connection: %v", reason)
		case <-deadline.C:
			return EnrolReply{}, fmt.Errorf(
				"the broker accepted this node's connection and nothing answered within %s. The "+
					"mesh's broker is running and its control plane is not", timeout)
		case delivery, ok := <-replies:
			if !ok {
				return EnrolReply{}, errors.New("the broker stopped delivering")
			}
			// Anything else on this queue is not the answer to this question.
			if delivery.CorrelationId != correlation {
				continue
			}
			var reply EnrolReply
			if err := json.Unmarshal(delivery.Body, &reply); err != nil {
				return EnrolReply{}, fmt.Errorf("the mesh's answer could not be read: %w", err)
			}
			again, err := answered(reply, time.Since(began))
			if err != nil {
				return reply, err
			}
			if !again {
				return reply, nil
			}
			select {
			case <-ctx.Done():
				return EnrolReply{}, ctx.Err()
			case <-time.After(AskAgainAfter):
			}
			if correlation, err = ask(); err != nil {
				return EnrolReply{}, err
			}
			if !deadline.Stop() {
				select {
				case <-deadline.C:
				default:
				}
			}
			deadline.Reset(timeout)
		}
	}
}