|
|
|
@@ -2,6 +2,7 @@ package link
|
|
|
|
|
|
|
|
|
|
import (
|
|
|
|
|
"context"
|
|
|
|
|
"encoding/base64"
|
|
|
|
|
"encoding/json"
|
|
|
|
|
"errors"
|
|
|
|
|
"fmt"
|
|
|
|
@@ -46,13 +47,24 @@ type EnrolRequest struct {
|
|
|
|
|
ServingKey string `json:"serving_key,omitempty"`
|
|
|
|
|
|
|
|
|
|
Profile map[string]any `json:"profile,omitempty"`
|
|
|
|
|
|
|
|
|
|
// Proof is this node's identity key signing EnrolProof over this request: that the presenter
|
|
|
|
|
// holds the private half of PublicKey. The mesh asks for it before letting an enrolment finish
|
|
|
|
|
// on a token this key already spent (novox/hq issue 083).
|
|
|
|
|
Proof []byte `json:"proof,omitempty"`
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// EnrolReply is what the mesh says back.
|
|
|
|
|
type EnrolReply struct {
|
|
|
|
|
Accepted bool `json:"accepted"`
|
|
|
|
|
Node string `json:"node,omitempty"`
|
|
|
|
|
Queue string `json:"queue,omitempty"`
|
|
|
|
|
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
|
|
|
|
@@ -65,9 +77,46 @@ type EnrolReply struct {
|
|
|
|
|
Refusal string `json:"refusal,omitempty"`
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// EnrolProof is what a node signs with its identity key when it enrols: the token and every key it
|
|
|
|
|
// presents, so a proof cannot be moved to another request. The mesh builds the same bytes.
|
|
|
|
|
func EnrolProof(secret string, public []byte, overlay, sealing, serving string) []byte {
|
|
|
|
|
return []byte("novox-mesh-enrol\x00" + secret + "\x00" + base64.StdEncoding.EncodeToString(public) +
|
|
|
|
|
"\x00" + overlay + "\x00" + sealing + "\x00" + serving)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
// 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
|
|
|
|
@@ -75,7 +124,7 @@ var ErrRefused = errors.New("the mesh refused this enrolment")
|
|
|
|
|
// 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,
|
|
|
|
|
overlayKey, sealingKey, servingKey string, profile map[string]any, proof []byte,
|
|
|
|
|
timeout time.Duration) (EnrolReply, error) {
|
|
|
|
|
|
|
|
|
|
config, err := PinnedConfig(pin)
|
|
|
|
@@ -122,24 +171,36 @@ func Enrol(ctx context.Context, address, pin, node, secret string, public []byte
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
request := EnrolRequest{Node: node, Secret: secret, PublicKey: public,
|
|
|
|
|
OverlayKey: overlayKey, SealingKey: sealingKey, ServingKey: servingKey, Profile: profile}
|
|
|
|
|
OverlayKey: overlayKey, SealingKey: sealingKey, ServingKey: servingKey, Profile: profile,
|
|
|
|
|
Proof: proof}
|
|
|
|
|
body, err := json.Marshal(request)
|
|
|
|
|
if err != nil {
|
|
|
|
|
return EnrolReply{}, err
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
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 EnrolReply{}, fmt.Errorf("cannot publish to the %s exchange: %w", Exchange, 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
|
|
|
|
@@ -170,10 +231,28 @@ func Enrol(ctx context.Context, address, pin, node, secret string, public []byte
|
|
|
|
|
if err := json.Unmarshal(delivery.Body, &reply); err != nil {
|
|
|
|
|
return EnrolReply{}, fmt.Errorf("the mesh's answer could not be read: %w", err)
|
|
|
|
|
}
|
|
|
|
|
if !reply.Accepted {
|
|
|
|
|
return reply, fmt.Errorf("%w: %s", ErrRefused, reply.Refusal)
|
|
|
|
|
again, err := answered(reply, time.Since(began))
|
|
|
|
|
if err != nil {
|
|
|
|
|
return reply, err
|
|
|
|
|
}
|
|
|
|
|
return reply, nil
|
|
|
|
|
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)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|