Revoke a removed node's certificate, and require --api-client with a mirror's mutual TLS
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
fe5b7f88f0
commit
9b61bbc983
16 files changed
+268
-32
No files matched your search
@@ -145,6 +145,13 @@ APIs change between minors without a deprecation period.
|
||||
**Behind a reverse proxy, set it**, or the pages and scripts name the
|
||||
address the proxy reached and every client counts as the proxy. `/setup`
|
||||
scripts are served `no-store`.
|
||||
- **Removing a node revokes its certificate.** `ctl node remove` only
|
||||
dropped the node from the inventory; its certificate, valid for a year,
|
||||
still connected and was given work. Now the removal is recorded in the
|
||||
store and the agent port refuses any certificate issued for that node
|
||||
before it — onboarding the node again issues one that works. A mirror
|
||||
serving its management API with `--tls-ca` now requires `--api-client`
|
||||
(the control plane's name), since the same authority signs every node.
|
||||
|
||||
### Everything that can be changed on a running cluster, from the console
|
||||
|
||||
|
||||
@@ -65,9 +65,13 @@ hand it certificates:
|
||||
llmgw mirror serve --dir /srv/models --base http://models.internal:8081 \
|
||||
--tls-ca /etc/llmgw/ca.pem \
|
||||
--tls-cert /etc/llmgw/mirror.pem \
|
||||
--tls-key /etc/llmgw/mirror-key.pem
|
||||
--tls-key /etc/llmgw/mirror-key.pem \
|
||||
--api-client control-plane
|
||||
```
|
||||
|
||||
`--api-client` is required with `--tls-ca`: it names the one certificate the
|
||||
API answers, because the CA signs every node's certificate as well.
|
||||
|
||||
That is the same CA the agents already use, so the certificates come from the
|
||||
authority you have:
|
||||
|
||||
|
||||
+8
-5
@@ -269,17 +269,20 @@ it can start downloads and delete models, and only the control plane needs it.
|
||||
The split is what lets you lock the second one down without re-certifying every
|
||||
agent that fetches from the first.
|
||||
|
||||
Give the management port the cluster's own CA and it takes client certificates:
|
||||
Give the management port the cluster's own CA and it takes client certificates —
|
||||
the control plane's, named with `--api-client`, and no other. The same CA signs
|
||||
every node's certificate, so the name is required: without it any node, a
|
||||
removed one's included, could delete models.
|
||||
|
||||
```
|
||||
llmgw mirror serve --dir /srv/models --base http://models.internal:8081 \
|
||||
--tls-ca /etc/llmgw/ca.pem \
|
||||
--tls-cert /etc/llmgw/mirror.pem --tls-key /etc/llmgw/mirror-key.pem
|
||||
--tls-cert /etc/llmgw/mirror.pem --tls-key /etc/llmgw/mirror-key.pem \
|
||||
--api-client control-plane
|
||||
```
|
||||
|
||||
Without those three flags the management port is open to anyone who can reach
|
||||
it, and the mirror says so in its own log at every start rather than letting you
|
||||
find out later.
|
||||
Without those flags the management port answers this machine only, and the
|
||||
addresses given with `--api-allow`.
|
||||
|
||||
### The layout
|
||||
|
||||
|
||||
@@ -281,6 +281,13 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
if len(clients) > 0 && !tlsc.Enabled() {
|
||||
return fmt.Errorf("--api-client names a certificate, and needs --tls-ca, --tls-cert and --tls-key")
|
||||
}
|
||||
if tlsc.Enabled() && len(clients) == 0 {
|
||||
// The same root signs every node's certificate, so without a
|
||||
// name any node — a removed one's included — could delete
|
||||
// models here.
|
||||
return fmt.Errorf("with --tls-ca, name the control plane's certificate with --api-client: " +
|
||||
"the authority signs every node's certificate too")
|
||||
}
|
||||
|
||||
out := cmd.OutOrStdout()
|
||||
log := slog.Default()
|
||||
@@ -315,13 +322,7 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
go func() { errs <- fmt.Errorf("serving the API: %w", serveAPI(admin, tlsc)) }()
|
||||
switch {
|
||||
case tlsc.Enabled():
|
||||
fmt.Fprintf(out, "management API on %s, mutual TLS against %s\n", apiListen, tlsc.CA)
|
||||
if len(clients) == 0 {
|
||||
// The same root signs every node's certificate, so without a
|
||||
// name any node could delete models here.
|
||||
log.Warn("the management API takes any certificate from " + tlsc.CA +
|
||||
", every node's included; name the control plane's with --api-client")
|
||||
}
|
||||
fmt.Fprintf(out, "management API on %s, mutual TLS for %s\n", apiListen, strings.Join(clients, ", "))
|
||||
case len(allowed) > 0:
|
||||
fmt.Fprintf(out, "management API on %s, for %s and this machine only\n", apiListen, strings.Join(allow, ", "))
|
||||
default:
|
||||
@@ -346,7 +347,7 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
c.Flags().StringArrayVar(&allow, "api-allow", nil,
|
||||
"an address or network the management API answers, besides this machine; repeatable (default: this machine only)")
|
||||
c.Flags().StringArrayVar(&clients, "api-client", nil,
|
||||
"with mutual TLS, the common name of a certificate the management API answers; repeatable (default: any from --tls-ca)")
|
||||
"with mutual TLS, the common name of a certificate the management API answers; repeatable; required with --tls-ca")
|
||||
c.Flags().BoolVar(&allowPrivate, "allow-private-sources", false,
|
||||
"let downloads reach loopback, private and link-local addresses")
|
||||
c.Flags().StringVar(&tlsc.CA, "tls-ca", "", "CA bundle that management clients must present a certificate from")
|
||||
|
||||
@@ -236,6 +236,10 @@ func newServerCmd() *cobra.Command {
|
||||
}
|
||||
adminAPI = api
|
||||
api.SetReleases(cfg.Releases)
|
||||
if st := rt.Store(); st != nil {
|
||||
// Removing a node revokes its certificate.
|
||||
api.SetRevoker(st.RevokeNode)
|
||||
}
|
||||
// Catalog changes need somewhere to persist, so this is wired
|
||||
// only when there is a store. Without one, the endpoints say so
|
||||
// rather than accepting a model that vanishes on restart.
|
||||
@@ -615,6 +619,8 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Live
|
||||
return urls, err
|
||||
})
|
||||
go svc.WatchControlPlanes(ctx, 30*time.Second)
|
||||
// A removed node's certificate is refused.
|
||||
svc.SetRevocations(st.NodeRevokedAt)
|
||||
}
|
||||
fwd.set(svc)
|
||||
// Every InfiniBand fabric mapped every five minutes, through one node.
|
||||
|
||||
@@ -47,6 +47,8 @@ type API struct {
|
||||
serve *modelserve.Server
|
||||
// history is the Prometheus the console's charts read from.
|
||||
history History
|
||||
// revoke records a removed node, so its certificate is refused.
|
||||
revoke func(ctx context.Context, nodeID string, at time.Time) error
|
||||
|
||||
models Models
|
||||
capacity Capacity
|
||||
@@ -58,6 +60,11 @@ type API struct {
|
||||
// New builds the API over a controller.
|
||||
func New(ctl *control.Controller) *API { return &API{ctl: ctl} }
|
||||
|
||||
// SetRevoker has removing a node revoke its certificate.
|
||||
func (a *API) SetRevoker(f func(ctx context.Context, nodeID string, at time.Time) error) {
|
||||
a.revoke = f
|
||||
}
|
||||
|
||||
// Register mounts the endpoints on a mux under /api/v1.
|
||||
func (a *API) Register(mux *http.ServeMux) {
|
||||
mux.HandleFunc("GET /api/v1/summary", a.summary)
|
||||
@@ -1098,6 +1105,14 @@ func (a *API) removeNode(w http.ResponseWriter, r *http.Request) {
|
||||
"%s is %s: stop its agent (systemctl stop llmgw-agent) and remove it once it is lost, a minute later", id, n.State))
|
||||
return
|
||||
}
|
||||
// Revoked first: a node removed but still holding a certificate the
|
||||
// agent port accepts could connect again and be given work.
|
||||
if a.revoke != nil {
|
||||
if err := a.revoke(r.Context(), id, time.Now()); err != nil {
|
||||
writeError(w, http.StatusInternalServerError, "revoking the node's certificate: "+err.Error())
|
||||
return
|
||||
}
|
||||
}
|
||||
inv.Remove(id)
|
||||
a.ctl.Reconciler().Notify()
|
||||
writeJSON(w, http.StatusOK, map[string]string{"id": id, "removed": "true"})
|
||||
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -707,3 +708,30 @@ func TestOnlyANodeThatIsGoneCanBeRemoved(t *testing.T) {
|
||||
t.Errorf("removing it again gave %d, expected 404", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRemovingANodeRevokesItsCertificate(t *testing.T) {
|
||||
a, ctl := setup(t, testModels(), gpuNode("n1", 8), gpuNode("n2", 8))
|
||||
var revoked []string
|
||||
a.SetRevoker(func(_ context.Context, id string, _ time.Time) error {
|
||||
revoked = append(revoked, id)
|
||||
return nil
|
||||
})
|
||||
ctl.Inventory().SetState("n1", inventory.Lost)
|
||||
if code := call(t, a, "DELETE", "/api/v1/nodes/n1", "", nil); code != 200 {
|
||||
t.Fatalf("a lost node gave %d, expected 200", code)
|
||||
}
|
||||
if !slices.Equal(revoked, []string{"n1"}) {
|
||||
t.Errorf("revoked %v, expected n1", revoked)
|
||||
}
|
||||
|
||||
// Not revoked, not removed: a node left listed is one the operator can
|
||||
// see and try again on.
|
||||
a.SetRevoker(func(context.Context, string, time.Time) error { return errors.New("store down") })
|
||||
ctl.Inventory().SetState("n2", inventory.Lost)
|
||||
if code := call(t, a, "DELETE", "/api/v1/nodes/n2", "", nil); code != 500 {
|
||||
t.Errorf("a failed revocation gave %d, expected 500", code)
|
||||
}
|
||||
if _, ok := ctl.Inventory().Get("n2"); !ok {
|
||||
t.Error("the node was removed though its certificate was not revoked")
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,9 @@ import (
|
||||
"context"
|
||||
"crypto/x509"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/tlsconf"
|
||||
)
|
||||
|
||||
// peerKey carries the verified client identity through to the stream handler.
|
||||
@@ -21,6 +24,9 @@ type PeerIdentity struct {
|
||||
// Verified is false when the connection carried no client certificate,
|
||||
// which only happens when TLS is switched off.
|
||||
Verified bool
|
||||
// IssuedAt is when the certificate was issued, for revocation: one issued
|
||||
// for a node before it was removed is refused.
|
||||
IssuedAt time.Time
|
||||
}
|
||||
|
||||
// WithPeerIdentity records a verified identity on the context.
|
||||
@@ -53,5 +59,5 @@ func AuthMiddleware(next http.Handler) http.Handler {
|
||||
}
|
||||
|
||||
func identityOf(cert *x509.Certificate) PeerIdentity {
|
||||
return PeerIdentity{CommonName: cert.Subject.CommonName, Verified: true}
|
||||
return PeerIdentity{CommonName: cert.Subject.CommonName, Verified: true, IssuedAt: tlsconf.IssuedAt(cert)}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package agentsvc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"connectrpc.com/connect"
|
||||
)
|
||||
|
||||
func TestARemovedNodesCertificateIsRefused(t *testing.T) {
|
||||
removed := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC)
|
||||
s := newTestService()
|
||||
s.requireIdentity = true
|
||||
s.SetRevocations(func(_ context.Context, id string) (time.Time, error) {
|
||||
if id == "gpu01" {
|
||||
return removed, nil
|
||||
}
|
||||
return time.Time{}, nil
|
||||
})
|
||||
peer := func(cn string, issued time.Time) context.Context {
|
||||
return WithPeerIdentity(context.Background(), PeerIdentity{CommonName: cn, Verified: true, IssuedAt: issued})
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
ctx context.Context
|
||||
node string
|
||||
refuse connect.Code
|
||||
}{
|
||||
{"issued before the removal", peer("gpu01", removed.Add(-time.Minute)), "gpu01", connect.CodePermissionDenied},
|
||||
{"issued at the removal", peer("gpu01", removed), "gpu01", connect.CodePermissionDenied},
|
||||
{"issued after, by onboarding again", peer("gpu01", removed.Add(time.Second)), "gpu01", 0},
|
||||
{"another node", peer("gpu02", removed.Add(-time.Hour)), "gpu02", 0},
|
||||
{"another node's name", peer("gpu02", removed.Add(time.Hour)), "gpu01", connect.CodePermissionDenied},
|
||||
{"no certificate", context.Background(), "gpu02", connect.CodeUnauthenticated},
|
||||
}
|
||||
for _, c := range cases {
|
||||
err := s.admit(c.ctx, c.node)
|
||||
if c.refuse == 0 {
|
||||
if err != nil {
|
||||
t.Errorf("%s: refused: %v", c.name, err)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if connect.CodeOf(err) != c.refuse {
|
||||
t.Errorf("%s: got %v, expected %v", c.name, err, c.refuse)
|
||||
}
|
||||
}
|
||||
|
||||
// Not knowing is not "not revoked".
|
||||
s.SetRevocations(func(context.Context, string) (time.Time, error) { return time.Time{}, errors.New("store down") })
|
||||
if err := s.admit(peer("gpu02", removed), "gpu02"); connect.CodeOf(err) != connect.CodeUnavailable {
|
||||
t.Errorf("with the store down: %v, expected unavailable", err)
|
||||
}
|
||||
}
|
||||
@@ -49,6 +49,9 @@ type Service struct {
|
||||
// requireIdentity refuses agents that present no certificate. It is off
|
||||
// only for the insecure development path.
|
||||
requireIdentity bool
|
||||
// revokedAt is when a node was removed, whose certificates issued before
|
||||
// then are refused; nil checks nothing.
|
||||
revokedAt func(ctx context.Context, nodeID string) (time.Time, error)
|
||||
|
||||
mu sync.RWMutex
|
||||
conns map[string]*conn
|
||||
@@ -137,6 +140,47 @@ func New(inv *inventory.Store, status StatusSink, requireIdentity bool, log *slo
|
||||
}
|
||||
}
|
||||
|
||||
// admit decides whether the peer on ctx may be nodeID.
|
||||
func (s *Service) admit(ctx context.Context, nodeID string) error {
|
||||
// The certificate decides who this is, not the message.
|
||||
//
|
||||
// Without this check an agent could register as any node in the fleet and
|
||||
// receive that node's shards — which is the same exposure the stream had
|
||||
// before TLS, just with encryption on top. mTLS is only worth having if the
|
||||
// identity it proves is the one actually used.
|
||||
if peer := peerFrom(ctx); peer.Verified {
|
||||
if peer.CommonName != nodeID {
|
||||
s.log.Warn("rejecting an agent claiming a node id its certificate does not authorize",
|
||||
"certificate_cn", peer.CommonName, "claimed", nodeID)
|
||||
return connect.NewError(connect.CodePermissionDenied,
|
||||
fmt.Errorf("certificate authorizes node %q, not %q", peer.CommonName, nodeID))
|
||||
}
|
||||
if s.revokedAt != nil {
|
||||
at, err := s.revokedAt(ctx, nodeID)
|
||||
if err != nil {
|
||||
// Unknown is not "not revoked": the agent retries.
|
||||
return connect.NewError(connect.CodeUnavailable, fmt.Errorf("checking revocation: %w", err))
|
||||
}
|
||||
if !at.IsZero() && !peer.IssuedAt.After(at) {
|
||||
s.log.Warn("rejecting a removed node's certificate", "node", nodeID,
|
||||
"issued", peer.IssuedAt, "removed", at)
|
||||
return connect.NewError(connect.CodePermissionDenied, fmt.Errorf(
|
||||
"node %q was removed from the cluster, and its certificate with it; onboard it again", nodeID))
|
||||
}
|
||||
}
|
||||
} else if s.requireIdentity {
|
||||
return connect.NewError(connect.CodeUnauthenticated,
|
||||
errors.New("no verified client certificate"))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetRevocations has agents refused whose certificate was issued before
|
||||
// their node was removed.
|
||||
func (s *Service) SetRevocations(f func(ctx context.Context, nodeID string) (time.Time, error)) {
|
||||
s.revokedAt = f
|
||||
}
|
||||
|
||||
// Handler returns the HTTP path and handler to mount.
|
||||
//
|
||||
// The handler is wrapped so the verified client identity reaches the stream:
|
||||
@@ -182,22 +226,8 @@ func (s *Service) Connect(
|
||||
return connect.NewError(connect.CodeInvalidArgument, errors.New("node_id is required"))
|
||||
}
|
||||
|
||||
// The certificate decides who this is, not the message.
|
||||
//
|
||||
// Without this check an agent could register as any node in the fleet and
|
||||
// receive that node's shards — which is the same exposure the stream had
|
||||
// before TLS, just with encryption on top. mTLS is only worth having if the
|
||||
// identity it proves is the one actually used.
|
||||
if peer := peerFrom(ctx); peer.Verified {
|
||||
if peer.CommonName != nodeID {
|
||||
s.log.Warn("rejecting an agent claiming a node id its certificate does not authorize",
|
||||
"certificate_cn", peer.CommonName, "claimed", nodeID)
|
||||
return connect.NewError(connect.CodePermissionDenied,
|
||||
fmt.Errorf("certificate authorizes node %q, not %q", peer.CommonName, nodeID))
|
||||
}
|
||||
} else if s.requireIdentity {
|
||||
return connect.NewError(connect.CodeUnauthenticated,
|
||||
errors.New("no verified client certificate"))
|
||||
if err := s.admit(ctx, nodeID); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
node := NodeFromRegister(reg)
|
||||
|
||||
@@ -52,6 +52,8 @@ var copyOrder = []string{
|
||||
"cluster_secrets",
|
||||
"node_engines",
|
||||
"control_planes",
|
||||
// Or a removed node's certificate would work again in the new store.
|
||||
"revoked_nodes",
|
||||
}
|
||||
|
||||
// notCopied are tables deliberately left behind.
|
||||
|
||||
@@ -0,0 +1,8 @@
|
||||
-- Ported from the SQLite migration of the same number.
|
||||
-- Nodes removed from the cluster, and when. A certificate issued for one
|
||||
-- before that moment is refused when it connects: removing a node revokes
|
||||
-- the certificate it was given, and onboarding it again issues a new one.
|
||||
CREATE TABLE revoked_nodes (
|
||||
node_id TEXT PRIMARY KEY,
|
||||
revoked_at BIGINT NOT NULL
|
||||
);
|
||||
@@ -0,0 +1,7 @@
|
||||
-- Nodes removed from the cluster, and when. A certificate issued for one
|
||||
-- before that moment is refused when it connects: removing a node revokes
|
||||
-- the certificate it was given, and onboarding it again issues a new one.
|
||||
CREATE TABLE revoked_nodes (
|
||||
node_id TEXT PRIMARY KEY,
|
||||
revoked_at INTEGER NOT NULL
|
||||
);
|
||||
@@ -0,0 +1,29 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"time"
|
||||
)
|
||||
|
||||
// RevokeNode records that a node was removed at a moment: every certificate
|
||||
// issued for it until then is refused from now on.
|
||||
func (s *DB) RevokeNode(ctx context.Context, nodeID string, at time.Time) error {
|
||||
_, err := s.db.ExecContext(ctx, `INSERT INTO revoked_nodes (node_id, revoked_at) VALUES (?, ?)
|
||||
ON CONFLICT (node_id) DO UPDATE SET revoked_at = excluded.revoked_at`, nodeID, millis(at))
|
||||
return err
|
||||
}
|
||||
|
||||
// NodeRevokedAt is when a node was last removed; zero if it never was.
|
||||
func (s *DB) NodeRevokedAt(ctx context.Context, nodeID string) (time.Time, error) {
|
||||
var at int64
|
||||
err := s.db.QueryRowContext(ctx, `SELECT revoked_at FROM revoked_nodes WHERE node_id = ?`, nodeID).Scan(&at)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return time.Time{}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return time.Time{}, err
|
||||
}
|
||||
return fromMillis(at), nil
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestARemovedNodeIsRememberedWithItsLastRemoval(t *testing.T) {
|
||||
db := capacityStore(t)
|
||||
ctx := t.Context()
|
||||
if at, err := db.NodeRevokedAt(ctx, "gpu01"); err != nil || !at.IsZero() {
|
||||
t.Fatalf("never removed: %v, %v", at, err)
|
||||
}
|
||||
first := time.UnixMilli(1_700_000_000_000)
|
||||
if err := db.RevokeNode(ctx, "gpu01", first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
second := first.Add(time.Hour)
|
||||
if err := db.RevokeNode(ctx, "gpu01", second); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if at, err := db.NodeRevokedAt(ctx, "gpu01"); err != nil || !at.Equal(second) {
|
||||
t.Errorf("got %v, %v; expected the second removal", at, err)
|
||||
}
|
||||
}
|
||||
@@ -48,6 +48,14 @@ const CAValidity = 10 * 365 * 24 * time.Hour
|
||||
// enough to be calm, short enough that a leaked key is not permanent.
|
||||
const LeafValidity = 365 * 24 * time.Hour
|
||||
|
||||
// Backdate is how far before its issue a certificate is valid from, to
|
||||
// tolerate clock skew across the fleet.
|
||||
const Backdate = time.Hour
|
||||
|
||||
// IssuedAt is when this authority issued a certificate: its NotBefore, less
|
||||
// the backdating. What a node's revocation is compared against.
|
||||
func IssuedAt(cert *x509.Certificate) time.Time { return cert.NotBefore.Add(Backdate) }
|
||||
|
||||
// NewCA creates an authority.
|
||||
func NewCA(commonName string) (*CA, error) {
|
||||
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
|
||||
@@ -156,7 +164,7 @@ func (ca *CA) issue(commonName string, hosts []string, usage x509.ExtKeyUsage) (
|
||||
tmpl := &x509.Certificate{
|
||||
SerialNumber: serial,
|
||||
Subject: pkix.Name{CommonName: commonName},
|
||||
NotBefore: time.Now().Add(-time.Hour),
|
||||
NotBefore: time.Now().Add(-Backdate),
|
||||
NotAfter: time.Now().Add(LeafValidity),
|
||||
KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
|
||||
ExtKeyUsage: []x509.ExtKeyUsage{usage},
|
||||
|
||||
Reference in new issue
Block a user