jschoubben
bbbc860188
`declare` sends a node a signed declaration; `serve` now also consumes reports. Signed over the exact bytes published, which is what the node verifies. Anything re-encoding in between would sign one thing and check another, and a difference in key order alone would have a node refuse a declaration that was genuinely the mesh's. Sent to the node's queue directly rather than through the exchange: a declaration is for one node, and routing by name through a shared exchange means a binding per node that nothing removes when a node is retired. Enrolment now issues the node its own broker password, replacing the token's secret, and tells it the broker address, the fingerprint and the signing key -- so a node can reconnect after a restart without a person and a new token, which is what makes disconnection ordinary rather than a crisis. A report is a statement, not a write. What a node says it applied is its own account of its own machine, kept as a copy for recovery rather than as a source.
54 lines
1.6 KiB
Go
54 lines
1.6 KiB
Go
package link
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
// Signer is whatever holds the control plane's signing key.
|
|
type Signer interface {
|
|
Sign(ctx context.Context, message []byte) ([]byte, error)
|
|
}
|
|
|
|
// Declare sends a node what it should be, signed.
|
|
//
|
|
// The signature is over the declaration exactly as it is published — the same bytes the node
|
|
// verifies. Anything that re-encoded between here and there would produce a signature over
|
|
// something else, and the node would refuse a declaration that was genuinely the mesh's.
|
|
//
|
|
// Published to the node's own queue, which its account alone may read.
|
|
func Declare(ctx context.Context, channel *amqp.Channel, signer Signer, node string,
|
|
declaration []byte, timeout time.Duration) error {
|
|
|
|
if !json.Valid(declaration) {
|
|
return fmt.Errorf("refusing to send %s something that is not a declaration", node)
|
|
}
|
|
|
|
signature, err := signer.Sign(ctx, declaration)
|
|
if err != nil {
|
|
return fmt.Errorf("cannot sign a declaration for %s: %w", node, err)
|
|
}
|
|
|
|
body, err := json.Marshal(Signed{Declaration: declaration, Signature: signature})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
publish, cancel := context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
|
|
// Published to the queue directly rather than through the exchange: a declaration is for one
|
|
// node, and routing it by name through a shared exchange would mean a binding per node that
|
|
// nothing removes when a node is retired.
|
|
return channel.PublishWithContext(publish, "", QueueFor(node), false, false,
|
|
amqp.Publishing{
|
|
ContentType: "application/json",
|
|
DeliveryMode: amqp.Persistent,
|
|
Body: body,
|
|
})
|
|
}
|