Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
611fddd236 | ||
|
|
9853936e0b | ||
|
|
d167686cf1 | ||
|
|
97dc49cbeb | ||
|
|
3a26277c32 |
No files matched your search
@@ -20,6 +20,49 @@ is only released once its **exit criterion** is met, not when its code lands.
|
||||
Until v1.0.0 the minor version is the breaking one: configuration formats and Go
|
||||
APIs change between minors without a deprecation period.
|
||||
|
||||
## v0.10.3 — 2026-09-30
|
||||
|
||||
Security fixes from a review of v0.10.2. Three of them were confirmed on the
|
||||
live cluster before being fixed.
|
||||
|
||||
### Engines demand a key only the router holds
|
||||
|
||||
Engines listened on every address of their node with no key. A request from a
|
||||
neighbouring node reached the model directly, past the gateway's
|
||||
authentication, quotas and accounting. Each replica's engine is now started
|
||||
with a key derived from the replica id and a secret in the store. The router,
|
||||
its metrics scrape, the agent's readiness check and `ctl bench` send it.
|
||||
llama.cpp and vLLM receive it through their environment, not their command
|
||||
line. Engines started before this keep running without a key until their
|
||||
replica is next started.
|
||||
|
||||
### The mirror's management API answers only who it was told to
|
||||
|
||||
Without `--api-allow`, the management API answered anyone who could reach it,
|
||||
including every node. It now answers this machine only; the control plane is
|
||||
let in with `--api-allow` or mutual TLS. **Upgrading a mirror set up by hand:
|
||||
add `--api-allow <control plane address>` to its unit.** Downloads started
|
||||
through the API go to http and https only, and never to loopback, private or
|
||||
link-local addresses (`--allow-private-sources` to permit that). Model,
|
||||
version and file names can no longer be `..` or start with a dot.
|
||||
|
||||
### The admin listener refuses what a browser page could send
|
||||
|
||||
The admin API has no login and trusts whoever reaches it. A page open in an
|
||||
operator's browser, reaching it over an ssh tunnel, could post a model
|
||||
definition. The listener now requires Host to be an address or localhost,
|
||||
refuses cross-site requests, and takes changes only as `application/json`.
|
||||
A profile's `env` may no longer set variables that make an engine run other
|
||||
code (`LD_*`, `PATH`, `PYTHON*` start-up variables, `BASH_ENV` and similar).
|
||||
They are refused where a model is defined and dropped again before an engine
|
||||
starts.
|
||||
|
||||
### Fixes
|
||||
|
||||
- The agent installs only engine versions that look like one, whatever the
|
||||
control plane sent; a version of `../x` could have installed outside
|
||||
`/opt/llmgw/engines`.
|
||||
|
||||
## v0.10.2 — 2026-09-29
|
||||
|
||||
- After `server init`'s screen closes, only the administrator's password and
|
||||
|
||||
@@ -2525,7 +2525,11 @@ type StartShard struct {
|
||||
Fabric string `protobuf:"bytes,14,opt,name=fabric,proto3" json:"fabric,omitempty"`
|
||||
// profile_fp identifies the version of the profile, so the agent can say
|
||||
// which one a running shard was started from. The agent never interprets it.
|
||||
ProfileFp string `protobuf:"bytes,15,opt,name=profile_fp,json=profileFp,proto3" json:"profile_fp,omitempty"`
|
||||
ProfileFp string `protobuf:"bytes,15,opt,name=profile_fp,json=profileFp,proto3" json:"profile_fp,omitempty"`
|
||||
// api_key is what the engine demands of every request, derived by the
|
||||
// control plane from the replica id. The router sends it; nothing else
|
||||
// that can reach the node's ports has it.
|
||||
ApiKey string `protobuf:"bytes,16,opt,name=api_key,json=apiKey,proto3" json:"api_key,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -2665,6 +2669,13 @@ func (x *StartShard) GetProfileFp() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *StartShard) GetApiKey() string {
|
||||
if x != nil {
|
||||
return x.ApiKey
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type EngineProfile struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
|
||||
@@ -3693,7 +3704,7 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
|
||||
"\n" +
|
||||
"bytes_done\x18\x02 \x01(\x03R\tbytesDone\x12\x1f\n" +
|
||||
"\vbytes_total\x18\x03 \x01(\x03R\n" +
|
||||
"bytesTotal\"\xd9\x03\n" +
|
||||
"bytesTotal\"\xf2\x03\n" +
|
||||
"\n" +
|
||||
"StartShard\x12\x1d\n" +
|
||||
"\n" +
|
||||
@@ -3716,7 +3727,8 @@ const file_llmgw_v1_agent_proto_rawDesc = "" +
|
||||
"\x04pool\x18\r \x01(\tR\x04pool\x12\x16\n" +
|
||||
"\x06fabric\x18\x0e \x01(\tR\x06fabric\x12\x1d\n" +
|
||||
"\n" +
|
||||
"profile_fp\x18\x0f \x01(\tR\tprofileFp\"\xfd\x02\n" +
|
||||
"profile_fp\x18\x0f \x01(\tR\tprofileFp\x12\x17\n" +
|
||||
"\aapi_key\x18\x10 \x01(\tR\x06apiKey\"\xfd\x02\n" +
|
||||
"\rEngineProfile\x12\x12\n" +
|
||||
"\x04name\x18\x01 \x01(\tR\x04name\x12\x16\n" +
|
||||
"\x06engine\x18\x02 \x01(\tR\x06engine\x12\x1b\n" +
|
||||
|
||||
@@ -524,6 +524,10 @@ message StartShard {
|
||||
// profile_fp identifies the version of the profile, so the agent can say
|
||||
// which one a running shard was started from. The agent never interprets it.
|
||||
string profile_fp = 15;
|
||||
// api_key is what the engine demands of every request, derived by the
|
||||
// control plane from the replica id. The router sends it; nothing else
|
||||
// that can reach the node's ports has it.
|
||||
string api_key = 16;
|
||||
}
|
||||
|
||||
message EngineProfile {
|
||||
|
||||
@@ -71,6 +71,7 @@ rather than repeated.
|
||||
| `lease_ttl` | `time.Duration` | [Control](#control) |
|
||||
| `id` | `string` | [Endpoint](#endpoint) |
|
||||
| `url` | `string` | [Endpoint](#endpoint) |
|
||||
| `api_key` | `string` | [Endpoint](#endpoint) |
|
||||
| `models` | `[]string` | [Endpoint](#endpoint) |
|
||||
| `engine` | `string` | [Endpoint](#endpoint) |
|
||||
| `pool` | `string` | [Endpoint](#endpoint) |
|
||||
@@ -566,6 +567,13 @@ reading the alert. Zero takes the default.
|
||||
|
||||
`string`
|
||||
|
||||
### `api_key`
|
||||
|
||||
`string`
|
||||
|
||||
APIKey is sent to the engine as a bearer token, for an engine started
|
||||
with one.
|
||||
|
||||
### `models`
|
||||
|
||||
`[]string`
|
||||
|
||||
@@ -70,6 +70,13 @@ type ShardSpec struct {
|
||||
HTTPPort int
|
||||
Profile Profile
|
||||
|
||||
// APIKey is what the engine demands of every request. Engines listen on
|
||||
// all of a node's addresses; without a key anybody who can reach the port
|
||||
// talks to the model directly, past the gateway's authentication, quotas
|
||||
// and accounting. Given to the engine in its environment rather than on
|
||||
// its command line, which every user on the node can read.
|
||||
APIKey string `json:"api_key,omitempty"`
|
||||
|
||||
// Pool and Fabric are control plane concepts the agent cannot derive. They
|
||||
// are carried so the agent can report them back after a restart; Fabric
|
||||
// also says, with an "nvl:" prefix, that the replica is inside a
|
||||
@@ -249,6 +256,11 @@ func baseEnv(s ShardSpec) map[string]string {
|
||||
// Profile overrides win: they are the escape hatch for the per-cluster
|
||||
// tuning no adapter can guess.
|
||||
for k, v := range s.Profile.Env {
|
||||
// Checked again here, not only where the model was defined: this is
|
||||
// the process about to run as root, whatever sent the profile.
|
||||
if bad, _ := UnsafeEnv(k); bad {
|
||||
continue
|
||||
}
|
||||
env[k] = v
|
||||
}
|
||||
return env
|
||||
|
||||
@@ -120,8 +120,12 @@ func (l *LlamaCpp) Commands(s ShardSpec) ([]Command, error) {
|
||||
}
|
||||
args = append(args, p.Args...)
|
||||
|
||||
env := baseEnv(s)
|
||||
if s.APIKey != "" {
|
||||
env["LLAMA_API_KEY"] = s.APIKey
|
||||
}
|
||||
return []Command{{
|
||||
Name: "llama-server", Path: bin, Args: args, Env: materializeEnv(baseEnv(s)),
|
||||
Name: "llama-server", Path: bin, Args: args, Env: materializeEnv(env),
|
||||
}}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -245,3 +245,19 @@ func TestAMissingPathIsLeftToTheEngine(t *testing.T) {
|
||||
t.Errorf("expected it passed through, got %q %v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLlamaCppIsGivenItsKeyOutOfSightOfTheProcessList(t *testing.T) {
|
||||
// The command line is readable by every user on the node; the
|
||||
// environment of a root process is not.
|
||||
cmds, err := (&LlamaCpp{}).Commands(ShardSpec{HTTPPort: 8080, APIKey: "k1",
|
||||
Profile: Profile{ModelPath: "/m.gguf"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if slices.Contains(cmds[0].Args, "k1") || slices.Contains(cmds[0].Args, "--api-key") {
|
||||
t.Errorf("the key is on the command line: %v", cmds[0].Args)
|
||||
}
|
||||
if !slices.Contains(cmds[0].Env, "LLAMA_API_KEY=k1") {
|
||||
t.Error("llama-server was not given its key")
|
||||
}
|
||||
}
|
||||
@@ -96,6 +96,11 @@ func (g *SGLang) serverCommand(s ShardSpec) (Command, error) {
|
||||
)
|
||||
}
|
||||
args = append(args, p.Args...)
|
||||
// SGLang reads its key only from the command line, so on a node with
|
||||
// other users it is visible to them in the process list.
|
||||
if s.APIKey != "" && s.Leader() {
|
||||
args = append(args, "--api-key", s.APIKey)
|
||||
}
|
||||
|
||||
return Command{Name: "sglang", Path: python, Args: args, Env: materializeEnv(baseEnv(s))}, nil
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package engines
|
||||
|
||||
import "strings"
|
||||
|
||||
// UnsafeEnv reports whether a profile may not set an environment variable,
|
||||
// and why.
|
||||
//
|
||||
// A profile's env is the escape hatch for tuning no adapter can guess —
|
||||
// CUDA_*, NCCL_*, VLLM_*, HF_* — and it reaches an engine the agent starts as
|
||||
// root. Some variables are not tuning but a way to run other code in that
|
||||
// process: the dynamic loader's LD_PRELOAD, Python's PYTHONSTARTUP, a shell's
|
||||
// BASH_ENV. Set from a model definition, which an admin API or a config file
|
||||
// supplies, they would turn "add a model" into "run this as root on every
|
||||
// node". They are refused where a model is defined and again where an engine
|
||||
// starts.
|
||||
func UnsafeEnv(key string) (bool, string) {
|
||||
k := strings.ToUpper(key)
|
||||
switch {
|
||||
case strings.HasPrefix(k, "LD_"), strings.HasPrefix(k, "DYLD_"):
|
||||
return true, "it changes what the dynamic loader loads"
|
||||
case k == "PATH":
|
||||
return true, "it changes which binaries run"
|
||||
case strings.HasPrefix(k, "PYTHON") && k != "PYTHONUNBUFFERED" && k != "PYTHONHASHSEED":
|
||||
return true, "it changes what Python imports or runs at start"
|
||||
case k == "BASH_ENV" || k == "ENV" || k == "PERL5OPT" || k == "PERL5LIB" ||
|
||||
k == "NODE_OPTIONS" || k == "RUBYOPT" || k == "GCONV_PATH" || k == "OPENSSL_CONF":
|
||||
return true, "it makes an interpreter or library load code"
|
||||
}
|
||||
return false, ""
|
||||
}
|
||||
@@ -150,6 +150,9 @@ func (v *VLLM) serverCommand(s ShardSpec) (Command, error) {
|
||||
// Profile arguments last, so an operator can override anything above.
|
||||
args = append(args, p.Args...)
|
||||
|
||||
if s.APIKey != "" {
|
||||
env["VLLM_API_KEY"] = s.APIKey
|
||||
}
|
||||
return Command{Name: "vllm", Path: bin, Args: args, Env: materializeEnv(env)}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -77,6 +78,10 @@ type Installer struct {
|
||||
PyPIBase string
|
||||
}
|
||||
|
||||
// validVersion is what a version may look like: "b10991", "0.29.0",
|
||||
// "0.11.0+cu128". No separators, and not starting with a dot.
|
||||
var validVersion = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._+-]{0,63}$`)
|
||||
|
||||
// DefaultRoot is where engines are installed when nothing says otherwise.
|
||||
const DefaultRoot = "/opt/llmgw/engines"
|
||||
|
||||
@@ -114,6 +119,15 @@ func (in *Installer) Converge(ctx context.Context, wants []Want) []Status {
|
||||
|
||||
func (in *Installer) one(ctx context.Context, w Want) Status {
|
||||
st := Status{Engine: w.Engine, Version: w.Version}
|
||||
// Checked here, on the node, whatever the control plane checked: the
|
||||
// version names a directory under the install root and is passed to a
|
||||
// package manager, and "../x" would put an install — and the RemoveAll
|
||||
// that replaces one — outside it, as root.
|
||||
if !validVersion.MatchString(w.Version) {
|
||||
st.State = Unsupported
|
||||
st.Detail = fmt.Sprintf("%q is not a version this agent will install", w.Version)
|
||||
return st
|
||||
}
|
||||
recipe, ok := Recipes[w.Engine]
|
||||
if !ok {
|
||||
st.State = Unsupported
|
||||
|
||||
@@ -330,3 +330,15 @@ func TestAPackageManagerBusyWithItsOwnUpdatesIsWaitedFor(t *testing.T) {
|
||||
t.Error("the wait was not reported")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAVersionThatIsAPathIsRefused(t *testing.T) {
|
||||
// The version names a directory under the install root; "../" would put
|
||||
// an install, and the RemoveAll that replaces one, outside it, as root.
|
||||
in := &Installer{Root: t.TempDir(), Host: ubuntu}
|
||||
for _, v := range []string{"../../etc", "..", ".hidden", "b1/../../x", "", "b1 && rm -rf /"} {
|
||||
got := in.Converge(t.Context(), []Want{{Engine: "llamacpp", Version: v}})
|
||||
if got[0].State != Unsupported {
|
||||
t.Errorf("version %q: %+v", v, got[0])
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -53,8 +53,8 @@ func NewHTTPHealth(timeout time.Duration, opts ...HTTPHealthOption) *HTTPHealth
|
||||
}
|
||||
|
||||
// Ready reports whether the engine at baseURL is serving model.
|
||||
func (h *HTTPHealth) Ready(ctx context.Context, baseURL, model string) error {
|
||||
if err := h.get(ctx, baseURL+h.healthPath, nil); err != nil {
|
||||
func (h *HTTPHealth) Ready(ctx context.Context, baseURL, model, apiKey string) error {
|
||||
if err := h.get(ctx, baseURL+h.healthPath, apiKey, nil); err != nil {
|
||||
return fmt.Errorf("health: %w", err)
|
||||
}
|
||||
if !h.requireName || model == "" {
|
||||
@@ -66,7 +66,7 @@ func (h *HTTPHealth) Ready(ctx context.Context, baseURL, model string) error {
|
||||
ID string `json:"id"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := h.get(ctx, baseURL+h.modelsPath, &list); err != nil {
|
||||
if err := h.get(ctx, baseURL+h.modelsPath, apiKey, &list); err != nil {
|
||||
return fmt.Errorf("models: %w", err)
|
||||
}
|
||||
for _, d := range list.Data {
|
||||
@@ -77,11 +77,14 @@ func (h *HTTPHealth) Ready(ctx context.Context, baseURL, model string) error {
|
||||
return fmt.Errorf("engine is up but does not serve %q yet", model)
|
||||
}
|
||||
|
||||
func (h *HTTPHealth) get(ctx context.Context, url string, out any) error {
|
||||
func (h *HTTPHealth) get(ctx context.Context, url, apiKey string, out any) error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if apiKey != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+apiKey)
|
||||
}
|
||||
resp, err := h.client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
|
||||
@@ -39,7 +39,7 @@ func engineStub(t *testing.T, healthy bool, models ...string) *httptest.Server {
|
||||
|
||||
func TestReadyWhenServingModel(t *testing.T) {
|
||||
srv := engineStub(t, true, "llama-3.3-70b")
|
||||
if err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "llama-3.3-70b"); err != nil {
|
||||
if err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "llama-3.3-70b", ""); err != nil {
|
||||
t.Errorf("a serving engine was reported unready: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -48,7 +48,7 @@ func TestNotReadyWhileLoading(t *testing.T) {
|
||||
// The case that matters: the process is up but the weights are not loaded.
|
||||
// Routing on /health alone would send traffic into a loading model.
|
||||
srv := engineStub(t, false, "llama-3.3-70b")
|
||||
err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "llama-3.3-70b")
|
||||
err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "llama-3.3-70b", "")
|
||||
if err == nil {
|
||||
t.Fatal("an unhealthy engine was reported ready")
|
||||
}
|
||||
@@ -61,7 +61,7 @@ func TestNotReadyWhenModelMissing(t *testing.T) {
|
||||
// /health passes but the model is not listed: the engine is up and serving
|
||||
// something else, or still registering.
|
||||
srv := engineStub(t, true, "some-other-model")
|
||||
err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "llama-3.3-70b")
|
||||
err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "llama-3.3-70b", "")
|
||||
if err == nil {
|
||||
t.Fatal("a model that is not served was reported ready")
|
||||
}
|
||||
@@ -73,14 +73,33 @@ func TestNotReadyWhenModelMissing(t *testing.T) {
|
||||
func TestModelCheckCanBeSkipped(t *testing.T) {
|
||||
srv := engineStub(t, true)
|
||||
h := NewHTTPHealth(time.Second, WithoutModelCheck())
|
||||
if err := h.Ready(t.Context(), srv.URL, "anything"); err != nil {
|
||||
if err := h.Ready(t.Context(), srv.URL, "anything", ""); err != nil {
|
||||
t.Errorf("the model check was not skipped: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnreachableEngine(t *testing.T) {
|
||||
h := NewHTTPHealth(200 * time.Millisecond)
|
||||
if err := h.Ready(t.Context(), "http://127.0.0.1:1", "m"); err == nil {
|
||||
if err := h.Ready(t.Context(), "http://127.0.0.1:1", "m", ""); err == nil {
|
||||
t.Error("an unreachable engine was reported ready")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheHealthCheckSendsTheEnginesKey(t *testing.T) {
|
||||
// An engine started with a key refuses /v1/models without it, and a
|
||||
// shard that is serving would read as never ready.
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Header.Get("Authorization") != "Bearer k1" {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(`{"data":[{"id":"m"}]}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
if err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "m", "k1"); err != nil {
|
||||
t.Errorf("with the key: %v", err)
|
||||
}
|
||||
if err := NewHTTPHealth(time.Second).Ready(t.Context(), srv.URL, "m", ""); err == nil {
|
||||
t.Error("without the key the engine refused, and that read as ready")
|
||||
}
|
||||
}
|
||||
@@ -79,7 +79,7 @@ func (c *Config) normalize() {
|
||||
// supervisor can be tested without an HTTP engine.
|
||||
type HealthChecker interface {
|
||||
// Ready reports whether the shard at baseURL serves model.
|
||||
Ready(ctx context.Context, baseURL, model string) error
|
||||
Ready(ctx context.Context, baseURL, model, apiKey string) error
|
||||
}
|
||||
|
||||
// Shard is one supervised engine process.
|
||||
@@ -411,7 +411,7 @@ func (s *Supervisor) WaitReady(ctx context.Context, sh *Shard, baseURL, model st
|
||||
last := strings.Join(sh.Logs(10), "\n")
|
||||
return fmt.Errorf("shard %s not ready within %v; last output:\n%s", sh.ID, timeout, last)
|
||||
case <-t.C:
|
||||
if err := s.health.Ready(ctx, baseURL, model); err != nil {
|
||||
if err := s.health.Ready(ctx, baseURL, model, sh.Spec.APIKey); err != nil {
|
||||
continue
|
||||
}
|
||||
sh.setState(Ready, nil)
|
||||
|
||||
@@ -21,7 +21,7 @@ import (
|
||||
// scriptedHealth reports ready once the flag is set.
|
||||
type scriptedHealth struct{ ready atomic.Bool }
|
||||
|
||||
func (h *scriptedHealth) Ready(context.Context, string, string) error {
|
||||
func (h *scriptedHealth) Ready(context.Context, string, string, string) error {
|
||||
if h.ready.Load() {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -225,6 +225,7 @@ func runStep(ctx context.Context, client *http.Client, cfg Config, concurrency i
|
||||
)
|
||||
|
||||
sampler := newSampler(cfg.MetricsURL, cfg.Engine)
|
||||
sampler.apiKey = cfg.APIKey
|
||||
sampler.start(ctx)
|
||||
|
||||
start := time.Now()
|
||||
|
||||
@@ -19,6 +19,7 @@ import (
|
||||
// Optional, because a run pointed at the gateway has no single engine to read.
|
||||
type sampler struct {
|
||||
url string
|
||||
apiKey string
|
||||
engine string
|
||||
client *http.Client
|
||||
cancel context.CancelFunc
|
||||
@@ -65,6 +66,9 @@ func (s *sampler) sample(ctx context.Context) {
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
if s.apiKey != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+s.apiKey)
|
||||
}
|
||||
resp, err := s.client.Do(req)
|
||||
if err != nil {
|
||||
return
|
||||
|
||||
@@ -41,6 +41,11 @@ func newBenchCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
|
||||
cfg.MetricsURL = metrics
|
||||
cfg.Engine = engine
|
||||
cfg.APIKey = key
|
||||
// A replica's engine demands its own key. Asked for rather than
|
||||
// typed: the admin API knows which replica answers at the target.
|
||||
if cfg.APIKey == "" {
|
||||
cfg.APIKey = replicaKey(cmd, client, target)
|
||||
}
|
||||
if len(steps) > 0 {
|
||||
cfg.Concurrency = steps
|
||||
}
|
||||
@@ -227,3 +232,18 @@ func ms(d time.Duration) string {
|
||||
return d.Round(10 * time.Millisecond).String()
|
||||
}
|
||||
}
|
||||
|
||||
// replicaKey is the engine key of the replica at target, or "" when target is
|
||||
// not a replica (the gateway, say) or the admin API cannot be asked.
|
||||
func replicaKey(cmd *cobra.Command, client func() *ctlClient, target string) string {
|
||||
var reps []adminapi.ReplicaView
|
||||
if err := client().get(cmd.Context(), "/api/v1/replicas", &reps); err != nil {
|
||||
return ""
|
||||
}
|
||||
for _, rep := range reps {
|
||||
if sameEndpoint(rep.Endpoint, target) {
|
||||
return rep.EngineKey
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
+3
-1
@@ -67,7 +67,9 @@ func (c *ctlClient) do(ctx context.Context, method, path string, body, out any)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if body != nil {
|
||||
// On every change, body or not: the admin listener refuses anything else,
|
||||
// because a browser can send a bodiless POST across origins unasked.
|
||||
if method != http.MethodGet {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
// Who made the change, kept with a stored model so a catalog that grew
|
||||
|
||||
+26
-13
@@ -233,6 +233,7 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
var listen, apiListen, base string
|
||||
var tlsc tlsconf.Config
|
||||
var allow []string
|
||||
var allowPrivate bool
|
||||
|
||||
c := &cobra.Command{
|
||||
Use: "serve",
|
||||
@@ -262,6 +263,9 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
out := cmd.OutOrStdout()
|
||||
log := slog.Default()
|
||||
mgr := mirror.NewManager(*dir, log)
|
||||
if allowPrivate {
|
||||
mgr.AllowPrivateSources()
|
||||
}
|
||||
api := mirror.NewAPI(mgr, strings.TrimSuffix(base, "/"))
|
||||
|
||||
// http.Dir refuses paths that escape the root, which is the whole
|
||||
@@ -278,7 +282,7 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
}
|
||||
admin := &http.Server{
|
||||
Addr: apiListen,
|
||||
Handler: onlyFrom(allowed, api.Handler()),
|
||||
Handler: apiGate(tlsc.Enabled(), allowed, api.Handler()),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
}
|
||||
|
||||
@@ -293,11 +297,11 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
case len(allowed) > 0:
|
||||
fmt.Fprintf(out, "management API on %s, for %s and this machine only\n", apiListen, strings.Join(allow, ", "))
|
||||
default:
|
||||
// Said plainly rather than left to be discovered. An API that
|
||||
// can delete models and start downloads, open to anyone who can
|
||||
// reach the port, is a choice the operator should have made
|
||||
// rather than inherited from a default.
|
||||
fmt.Fprintf(out, "management API on %s, UNAUTHENTICATED — keep it on a trusted network\n", apiListen)
|
||||
// This machine only, which was once "anyone": an API that can
|
||||
// delete models and start downloads, open to every node that
|
||||
// could reach the port, is a choice the operator should make
|
||||
// with --api-allow rather than inherit.
|
||||
fmt.Fprintf(out, "management API on %s, for this machine only (--api-allow to add the control plane)\n", apiListen)
|
||||
}
|
||||
fmt.Fprintf(out, "catalog sources should point at %s\n", base)
|
||||
|
||||
@@ -312,7 +316,9 @@ func newMirrorServeCmd(dir *string) *cobra.Command {
|
||||
c.Flags().StringVar(&base, "base", "http://localhost:8081",
|
||||
"the address nodes use to reach this mirror, as it should appear in catalog sources")
|
||||
c.Flags().StringArrayVar(&allow, "api-allow", nil,
|
||||
"an address or network the management API answers, besides this machine; repeatable (default: anyone)")
|
||||
"an address or network the management API answers, besides this machine; repeatable (default: this machine only)")
|
||||
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")
|
||||
c.Flags().StringVar(&tlsc.Cert, "tls-cert", "", "this mirror's certificate")
|
||||
c.Flags().StringVar(&tlsc.Key, "tls-key", "", "this mirror's private key")
|
||||
@@ -435,14 +441,21 @@ func parseAllow(list []string) ([]netip.Prefix, error) {
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// onlyFrom answers only this machine and the addresses allowed, when any
|
||||
// are: the management API can delete models and start downloads, and a
|
||||
// mirror on its own machine is reached over the network by one control plane
|
||||
// and no one else. Nothing allowed is the old behaviour, anyone.
|
||||
func onlyFrom(allowed []netip.Prefix, next http.Handler) http.Handler {
|
||||
if len(allowed) == 0 {
|
||||
// apiGate decides who reaches the management API. With mutual TLS the
|
||||
// certificate decides, narrowed by --api-allow when given. Without it, only
|
||||
// this machine and the addresses allowed: nothing allowed used to mean anyone,
|
||||
// and every node that could reach the port could delete models.
|
||||
func apiGate(mtls bool, allowed []netip.Prefix, next http.Handler) http.Handler {
|
||||
if mtls && len(allowed) == 0 {
|
||||
return next
|
||||
}
|
||||
return onlyFrom(allowed, next)
|
||||
}
|
||||
|
||||
// onlyFrom answers only this machine and the addresses allowed: the management
|
||||
// API can delete models and start downloads, and a mirror on its own machine
|
||||
// is reached over the network by one control plane and no one else.
|
||||
func onlyFrom(allowed []netip.Prefix, next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
host, _, err := net.SplitHostPort(r.RemoteAddr)
|
||||
a, perr := netip.ParseAddr(host)
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/netip"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTheMirrorAPIAnswersOnlyThisMachineUnlessTold(t *testing.T) {
|
||||
ok := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {})
|
||||
call := func(h http.Handler, from string) int {
|
||||
r := httptest.NewRequest("DELETE", "/v1/models/m/v", nil)
|
||||
r.RemoteAddr = from + ":5000"
|
||||
w := httptest.NewRecorder()
|
||||
h.ServeHTTP(w, r)
|
||||
return w.Code
|
||||
}
|
||||
cp := []netip.Prefix{netip.MustParsePrefix("10.0.0.140/32")}
|
||||
|
||||
// Nothing allowed and no TLS used to mean anyone: every node could delete
|
||||
// models.
|
||||
if got := call(apiGate(false, nil, ok), "10.0.0.141"); got != http.StatusForbidden {
|
||||
t.Errorf("a node reached an unconfigured mirror API: %d", got)
|
||||
}
|
||||
if got := call(apiGate(false, nil, ok), "127.0.0.1"); got != http.StatusOK {
|
||||
t.Errorf("the mirror's own machine was refused: %d", got)
|
||||
}
|
||||
if got := call(apiGate(false, cp, ok), "10.0.0.140"); got != http.StatusOK {
|
||||
t.Errorf("the allowed control plane was refused: %d", got)
|
||||
}
|
||||
if got := call(apiGate(false, cp, ok), "10.0.0.141"); got != http.StatusForbidden {
|
||||
t.Errorf("a node outside the allowlist got through: %d", got)
|
||||
}
|
||||
// With mutual TLS the certificate decides.
|
||||
if got := call(apiGate(true, nil, ok), "10.0.0.141"); got != http.StatusOK {
|
||||
t.Errorf("mTLS without an allowlist refused a client: %d", got)
|
||||
}
|
||||
}
|
||||
+15
-1
@@ -246,7 +246,7 @@ func newServerCmd() *cobra.Command {
|
||||
// enforces its own per-phase deadlines.
|
||||
}
|
||||
adminSrv := &http.Server{
|
||||
Handler: adminMux,
|
||||
Handler: router.AdminGuard(adminMux),
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
}
|
||||
|
||||
@@ -385,6 +385,20 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Live
|
||||
ccfg.Reconciler.PortRange = [2]int{cfg.Control.PortRange[0], cfg.Control.PortRange[1]}
|
||||
}
|
||||
|
||||
// Each replica's engine demands a key only the router has. The secret it
|
||||
// is derived from lives in the store, so every control plane against it —
|
||||
// and this one after a restart — computes the same key for a replica.
|
||||
if h, ok := sink.(interface{ Store() *storepkg.DB }); ok && h.Store() != nil {
|
||||
secret, err := h.Store().ClusterSecret(ctx, "engine-key")
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("reading the engine key secret: %w", err)
|
||||
}
|
||||
ccfg.Reconciler.EngineKey = control.EngineKeys([]byte(secret))
|
||||
} else {
|
||||
log.Warn("engines start without a key: there is no store to keep the secret " +
|
||||
"they are derived from, so anyone who can reach a node's ports reaches its models")
|
||||
}
|
||||
|
||||
// The service needs the inventory and the status sink, both of which the
|
||||
// controller owns; the controller needs the service as its dispatcher. The
|
||||
// cycle is broken by giving the controller a dispatcher that forwards to
|
||||
|
||||
@@ -404,8 +404,11 @@ type Scoring struct {
|
||||
|
||||
// Endpoint is one engine replica. Static in phase 1.
|
||||
type Endpoint struct {
|
||||
ID string `koanf:"id"`
|
||||
URL string `koanf:"url"`
|
||||
ID string `koanf:"id"`
|
||||
URL string `koanf:"url"`
|
||||
// APIKey is sent to the engine as a bearer token, for an engine started
|
||||
// with one.
|
||||
APIKey string `koanf:"api_key"`
|
||||
Models []string `koanf:"models"`
|
||||
// Engine names the inference engine, which decides what its metrics are
|
||||
// called. Empty means vLLM's names.
|
||||
|
||||
@@ -238,9 +238,13 @@ type ModelView struct {
|
||||
|
||||
// ReplicaView is a replica as `ctl replicas` shows it.
|
||||
type ReplicaView struct {
|
||||
ID string `json:"id"`
|
||||
Model string `json:"model"`
|
||||
Profile string `json:"profile"`
|
||||
ID string `json:"id"`
|
||||
// EngineKey is the key this replica's engine demands, for `ctl bench`
|
||||
// and anyone else holding the admin listener who needs to reach an engine
|
||||
// directly. The client-facing pages never render it.
|
||||
EngineKey string `json:"engine_key,omitempty"`
|
||||
Model string `json:"model"`
|
||||
Profile string `json:"profile"`
|
||||
// Stale is true when the profile has been edited since this replica was
|
||||
// placed, so it is running a definition the catalog no longer holds.
|
||||
//
|
||||
@@ -471,9 +475,10 @@ func (a *API) Replicas(model string) []ReplicaView {
|
||||
ID: rep.ID, Model: rep.Model, Profile: rep.Profile,
|
||||
State: rep.State.String(), Pool: rep.Pool, Fabric: rep.Fabric,
|
||||
Endpoint: rep.Endpoint, Err: rep.Err,
|
||||
Stale: a.profileChanged(rep.Model, rep.Profile, rep.ProfileFP),
|
||||
AgeSec: int64(time.Since(rep.CreatedAt).Seconds()),
|
||||
IdleSec: -1,
|
||||
EngineKey: a.ctl.EngineKey(rep.ID),
|
||||
Stale: a.profileChanged(rep.Model, rep.Profile, rep.ProfileFP),
|
||||
AgeSec: int64(time.Since(rep.CreatedAt).Seconds()),
|
||||
IdleSec: -1,
|
||||
}
|
||||
if rep.LastUsedAt.IsZero() {
|
||||
v.IdleSec = -1
|
||||
|
||||
@@ -191,6 +191,7 @@ func StartShardFromCommand(cmd reconciler.ShardCommand) *llmgwv1.StartShard {
|
||||
Pool: cmd.Pool,
|
||||
Fabric: cmd.Fabric,
|
||||
ProfileFp: cmd.ProfileFP,
|
||||
ApiKey: cmd.Spec.APIKey,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -211,6 +212,7 @@ func ShardSpecFromStart(m *llmgwv1.StartShard) (engines.ShardSpec, time.Duration
|
||||
Pool: m.GetPool(),
|
||||
Fabric: m.GetFabric(),
|
||||
ProfileFP: m.GetProfileFp(),
|
||||
APIKey: m.GetApiKey(),
|
||||
Profile: engines.Profile{
|
||||
Name: p.GetName(), Engine: p.GetEngine(), GPUClass: p.GetGpuClass(),
|
||||
TP: int(p.GetTp()), PP: int(p.GetPp()),
|
||||
|
||||
@@ -302,6 +302,13 @@ func (m Model) validate() error {
|
||||
}
|
||||
seen[p.Name] = struct{}{}
|
||||
|
||||
for k := range p.Env {
|
||||
if bad, why := engines.UnsafeEnv(k); bad {
|
||||
return fmt.Errorf("model %q profile %q: env %s is not allowed: %s",
|
||||
m.Name, p.Name, k, why)
|
||||
}
|
||||
}
|
||||
|
||||
if p.GPUClass == "" && !cpuCapable(p.Engine) {
|
||||
return fmt.Errorf("model %q profile %q: gpu_class is required for engine %q",
|
||||
m.Name, p.Name, p.Engine)
|
||||
|
||||
@@ -302,3 +302,22 @@ func TestSimulatedNeedsEveryProfileToBeSimulated(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAProfileCannotSetWhatRunsCodeInTheEngine(t *testing.T) {
|
||||
// A profile's env reaches an engine started as root. LD_PRELOAD from a
|
||||
// model definition would make "add a model" into "run this everywhere".
|
||||
for _, k := range []string{"LD_PRELOAD", "ld_library_path", "PYTHONSTARTUP", "PATH", "BASH_ENV"} {
|
||||
m := Model{Name: "m", Source: "http://x/m.gguf", Profiles: []Profile{{
|
||||
Name: "cpu", Engine: "llamacpp", Env: map[string]string{k: "/tmp/x"},
|
||||
}}}
|
||||
if _, err := New([]Model{m}); err == nil || !strings.Contains(err.Error(), "not allowed") {
|
||||
t.Errorf("%s: %v", k, err)
|
||||
}
|
||||
}
|
||||
ok := Model{Name: "m", Source: "http://x/m.gguf", Profiles: []Profile{{
|
||||
Name: "cpu", Engine: "llamacpp", Env: map[string]string{"OMP_NUM_THREADS": "4", "PYTHONUNBUFFERED": "1"},
|
||||
}}}
|
||||
if _, err := New([]Model{ok}); err != nil {
|
||||
t.Errorf("ordinary tuning refused: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,9 @@ package control
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
@@ -304,6 +307,7 @@ func (c *Controller) endpoints() []endpoint.Info {
|
||||
ID: rep.ID,
|
||||
URL: strings.TrimRight(rep.Endpoint, "/"),
|
||||
Models: names,
|
||||
APIKey: c.engineKey(rep.ID),
|
||||
Engine: prof.Engine,
|
||||
Pool: rep.Pool,
|
||||
Weight: rep.Weight,
|
||||
@@ -321,3 +325,29 @@ func fingerprint(infos []endpoint.Info) string {
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// EngineKey is the key a replica's engine demands, empty when engines run
|
||||
// without one.
|
||||
func (c *Controller) EngineKey(replicaID string) string { return c.engineKey(replicaID) }
|
||||
|
||||
func (c *Controller) engineKey(replicaID string) string {
|
||||
if c.cfg.Reconciler.EngineKey == nil {
|
||||
return ""
|
||||
}
|
||||
return c.cfg.Reconciler.EngineKey(replicaID)
|
||||
}
|
||||
|
||||
// EngineKeys derives each replica's engine key from one secret.
|
||||
//
|
||||
// HMAC rather than a stored random key per replica: the router needs the key
|
||||
// for every replica it sends to, including ones adopted after a restart that
|
||||
// this process never started, and a second control plane has to agree. A
|
||||
// function of a shared secret and the replica id gives all three without a
|
||||
// table to keep in step. A key taken from one node opens that replica only.
|
||||
func EngineKeys(secret []byte) func(replicaID string) string {
|
||||
return func(replicaID string) string {
|
||||
m := hmac.New(sha256.New, secret)
|
||||
m.Write([]byte("llmgw engine key\x00" + replicaID))
|
||||
return "llmgw-" + hex.EncodeToString(m.Sum(nil))[:40]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
package control
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestEngineKeysAreTheSameForTheSameReplicaAndNoOther(t *testing.T) {
|
||||
// A control plane that restarts, or a second one, must compute the key a
|
||||
// running engine was started with; a key taken from one node must open
|
||||
// that replica only.
|
||||
a, b := EngineKeys([]byte("s")), EngineKeys([]byte("s"))
|
||||
if a("r1") != b("r1") {
|
||||
t.Error("the same secret gave two keys for one replica")
|
||||
}
|
||||
if a("r1") == a("r2") {
|
||||
t.Error("two replicas share a key")
|
||||
}
|
||||
if a("r1") == EngineKeys([]byte("other"))("r1") {
|
||||
t.Error("the key does not depend on the secret")
|
||||
}
|
||||
}
|
||||
@@ -69,6 +69,11 @@ type Config struct {
|
||||
PortRange [2]int
|
||||
// MgmtInterface is the ethernet interface used for NCCL bootstrap.
|
||||
MgmtInterface string
|
||||
// EngineKey gives each replica the key its engine demands. Derived from
|
||||
// the replica id rather than stored, so a control plane that restarts —
|
||||
// or a second one — computes the same key for a replica it adopts. Nil
|
||||
// starts engines without one.
|
||||
EngineKey func(replicaID string) string
|
||||
// EvictionGrace is how long a replica must have gone unasked-for before it
|
||||
// may be stopped to make room for another model.
|
||||
//
|
||||
@@ -883,6 +888,7 @@ func (r *Reconciler) startReplica(ctx context.Context, m catalog.Model, pl *plac
|
||||
MgmtInterface: r.cfg.MgmtInterface,
|
||||
HTTPPort: s.HTTPPort,
|
||||
Profile: espec,
|
||||
APIKey: r.engineKey(id),
|
||||
},
|
||||
StartupTimeout: prof.StartupTimeout,
|
||||
Pool: pl.Pool,
|
||||
@@ -1380,3 +1386,10 @@ func (r *Reconciler) profileMemory(model, profile string) int {
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (r *Reconciler) engineKey(replicaID string) string {
|
||||
if r.cfg.EngineKey == nil {
|
||||
return ""
|
||||
}
|
||||
return r.cfg.EngineKey(replicaID)
|
||||
}
|
||||
@@ -1721,3 +1721,23 @@ func TestAnAgentTooOldToReportTheProfileVersionIsNotGuessedAt(t *testing.T) {
|
||||
t.Errorf("an unreported fingerprint should stay empty, got %q", fp)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnEngineIsStartedWithItsReplicasKey(t *testing.T) {
|
||||
// Engines listen on every address a node has. Started without a key, any
|
||||
// machine that could reach the port reached the model past the gateway's
|
||||
// authentication, quotas and accounting — checked from a neighbouring node
|
||||
// on the live cluster.
|
||||
h := newHarness(t, testModel(1), gpuNode("1", 8))
|
||||
h.r.cfg.EngineKey = func(id string) string { return "key-for-" + id }
|
||||
|
||||
h.r.Reconcile(t.Context())
|
||||
started := h.disp.startedShards()
|
||||
if len(started) == 0 {
|
||||
t.Fatal("nothing was started")
|
||||
}
|
||||
for _, c := range started {
|
||||
if c.Spec.APIKey != "key-for-"+c.ReplicaID {
|
||||
t.Errorf("shard %s started with key %q", c.ShardID, c.Spec.APIKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
+14
-6
@@ -7,7 +7,6 @@ import (
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -29,13 +28,22 @@ type Added struct {
|
||||
// the layout — a version that can be overwritten takes away the ability to
|
||||
// change the catalog back and still have the old weights.
|
||||
func Add(ctx context.Context, root, source string, e Entry, progress func(done, total int64)) (Added, error) {
|
||||
return addWith(ctx, http.DefaultClient, root, source, e, progress)
|
||||
}
|
||||
|
||||
// addWith is Add through a given client. The management API passes one that
|
||||
// will not reach private addresses; `llmgw mirror add`, run by somebody on
|
||||
// the mirror, passes the ordinary one.
|
||||
func addWith(ctx context.Context, client *http.Client, root, source string, e Entry,
|
||||
progress func(done, total int64),
|
||||
) (Added, error) {
|
||||
if e.Model == "" || e.Version == "" || e.File == "" {
|
||||
return Added{}, fmt.Errorf("a model, a version and a filename are all required")
|
||||
}
|
||||
if strings.ContainsAny(e.Model+e.Version+e.File, `/\`) {
|
||||
// Each is one path element. A slash would put the file somewhere the
|
||||
// layout does not describe, and List would not find it again.
|
||||
return Added{}, fmt.Errorf("model, version and filename are single names, not paths")
|
||||
for _, n := range []string{e.Model, e.Version, e.File} {
|
||||
if err := ValidName(n); err != nil {
|
||||
return Added{}, err
|
||||
}
|
||||
}
|
||||
|
||||
final := e.Path(root)
|
||||
@@ -50,7 +58,7 @@ func Add(ctx context.Context, root, source string, e Entry, progress func(done,
|
||||
if err != nil {
|
||||
return Added{}, err
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return Added{}, fmt.Errorf("fetching %s: %w", source, err)
|
||||
}
|
||||
|
||||
@@ -130,8 +130,8 @@ func (a *API) status(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
func (a *API) remove(w http.ResponseWriter, r *http.Request) {
|
||||
model, version := r.PathValue("model"), r.PathValue("version")
|
||||
if strings.ContainsAny(model+version, `/\`) {
|
||||
fail(w, http.StatusBadRequest, errors.New("model and version are single names, not paths"))
|
||||
if err := errors.Join(ValidName(model), ValidName(version)); err != nil {
|
||||
fail(w, http.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
@@ -17,7 +17,10 @@ import (
|
||||
func testAPI(t *testing.T) (*API, string) {
|
||||
t.Helper()
|
||||
root := t.TempDir()
|
||||
a := NewAPI(NewManager(root, discardLog()), "http://mirror:8081")
|
||||
m := NewManager(root, discardLog())
|
||||
// The upstreams in these tests are httptest servers on loopback.
|
||||
m.AllowPrivateSources()
|
||||
a := NewAPI(m, "http://mirror:8081")
|
||||
|
||||
// A download outlives the request that started it — that is the point of
|
||||
// the manager. So the temp directory must not be torn down while one is
|
||||
@@ -434,3 +437,33 @@ func TestRemovingAVersionIsRefusedWhileAnyShardIsArriving(t *testing.T) {
|
||||
t.Errorf("expected 409 while a shard is arriving, got %d", code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheAPIWillNotFetchFromThePrivateNetwork(t *testing.T) {
|
||||
// The file server hands whatever was fetched to anyone, so a fetch that
|
||||
// could reach 169.254.169.254 or a database on the LAN was a way to read
|
||||
// them through the mirror.
|
||||
m := NewManager(t.TempDir(), discardLog())
|
||||
for _, src := range []string{
|
||||
"http://127.0.0.1:1/x.gguf", "http://169.254.169.254/latest/meta-data",
|
||||
"http://10.0.0.140:9090/api/v1/nodes", "http://[::1]:1/x",
|
||||
} {
|
||||
st, err := m.Start(src, Entry{Model: "m", Version: "v", File: "x.gguf"})
|
||||
if err != nil {
|
||||
t.Fatalf("%s: %v", src, err)
|
||||
}
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for st.State == Fetching && time.Now().Before(deadline) {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
st, _ = m.Status("m", "v")
|
||||
}
|
||||
if st.State != Failed || !strings.Contains(st.Err, "refusing to fetch") {
|
||||
t.Errorf("%s: %+v", src, st)
|
||||
}
|
||||
m.Forget("m", "v")
|
||||
}
|
||||
for _, src := range []string{"file:///etc/passwd", "gopher://x/", "not a url"} {
|
||||
if _, err := m.Start(src, Entry{Model: "m", Version: "v", File: "x"}); err == nil {
|
||||
t.Errorf("%s was accepted as a source", src)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -72,8 +73,18 @@ type Manager struct {
|
||||
// until the first had finished. A fifteen-shard model had to be pulled one
|
||||
// shard at a time, each waiting on the last.
|
||||
running map[string]*Status
|
||||
|
||||
// client fetches sources. It will not reach private addresses unless
|
||||
// AllowPrivateSources was called.
|
||||
client *http.Client
|
||||
}
|
||||
|
||||
// AllowPrivateSources lets downloads reach loopback, private and link-local
|
||||
// addresses — a model server on the same network, say. Off by default: the
|
||||
// file server hands whatever was fetched to anyone, so a fetch that can reach
|
||||
// the private network is a way to read it.
|
||||
func (m *Manager) AllowPrivateSources() { m.client = fetchClient(true) }
|
||||
|
||||
// runKey identifies one download.
|
||||
func runKey(e Entry) string { return e.Model + "/" + e.Version + "/" + e.File }
|
||||
|
||||
@@ -82,7 +93,7 @@ func NewManager(root string, log *slog.Logger) *Manager {
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
return &Manager{root: root, log: log, running: map[string]*Status{}}
|
||||
return &Manager{root: root, log: log, running: map[string]*Status{}, client: fetchClient(false)}
|
||||
}
|
||||
|
||||
// ErrAlreadyFetching means a download for that model and version is running.
|
||||
@@ -97,6 +108,9 @@ func (m *Manager) Start(source string, e Entry) (Status, error) {
|
||||
if source == "" {
|
||||
return Status{}, fmt.Errorf("a source URL is required")
|
||||
}
|
||||
if err := validSource(source); err != nil {
|
||||
return Status{}, err
|
||||
}
|
||||
key := runKey(e)
|
||||
|
||||
m.mu.Lock()
|
||||
@@ -125,7 +139,7 @@ func (m *Manager) run(source string, e Entry, key string) {
|
||||
// minutes to hours and the thing that asked for it is long gone.
|
||||
ctx := context.Background()
|
||||
|
||||
added, err := Add(ctx, m.root, source, e, func(done, total int64) {
|
||||
added, err := addWith(ctx, m.client, m.root, source, e, func(done, total int64) {
|
||||
m.mu.Lock()
|
||||
if st, ok := m.running[key]; ok {
|
||||
st.BytesDone, st.BytesTotal = done, total
|
||||
|
||||
@@ -37,6 +37,21 @@ type Entry struct {
|
||||
Bytes int64
|
||||
}
|
||||
|
||||
// ValidName checks one element of an entry — a model, version or file name.
|
||||
//
|
||||
// Each is one path element, and a name that is not one writes somewhere the
|
||||
// layout does not describe. Rejecting slashes was not enough: ".." is a
|
||||
// single element too, and a model called ".." put its file beside the mirror
|
||||
// root rather than inside it. A name starting with a dot is refused as well,
|
||||
// because that is where the mirror keeps its own bookkeeping.
|
||||
func ValidName(n string) error {
|
||||
if n == "" || n[0] == '.' || strings.ContainsAny(n, "/\\\x00") {
|
||||
return fmt.Errorf("%q is not a usable name: it must be a single path element "+
|
||||
"that does not start with a dot", n)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Path is where this entry lives under a root.
|
||||
func (e Entry) Path(root string) string {
|
||||
return filepath.Join(root, e.Model, e.Version, e.File)
|
||||
|
||||
@@ -162,3 +162,18 @@ func TestProgressIsReportedForALongDownload(t *testing.T) {
|
||||
t.Error("a download reported no progress at all")
|
||||
}
|
||||
}
|
||||
|
||||
func TestANameThatLeavesTheMirrorIsRefused(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
for _, e := range []Entry{
|
||||
{Model: "..", Version: "v", File: "f.gguf"},
|
||||
{Model: "m", Version: "..", File: "f.gguf"},
|
||||
{Model: "m", Version: "v", File: ".f.sha256"},
|
||||
{Model: "a/b", Version: "v", File: "f.gguf"},
|
||||
} {
|
||||
if _, err := Add(t.Context(), root, "http://127.0.0.1:1/x", e, nil); err == nil ||
|
||||
!strings.Contains(err.Error(), "not a usable name") {
|
||||
t.Errorf("%+v: %v", e, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
package mirror
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"net/url"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
// errPrivateSource is a fetch refused because it would reach this machine or
|
||||
// the network behind it rather than the internet.
|
||||
var errPrivateSource = errors.New("refusing to fetch from a loopback, private or link-local address")
|
||||
|
||||
// fetchClient downloads sources for the management API.
|
||||
//
|
||||
// The API fetches whatever URL it is given and the file server then hands the
|
||||
// result to anyone. Without this, "fetch http://169.254.169.254/..." was a way
|
||||
// to read a cloud's instance metadata — or anything else on the network the
|
||||
// mirror sits in — through the mirror. The check is made on the address
|
||||
// actually dialled, after resolution and on every redirect, so a name that
|
||||
// resolves somewhere public and then somewhere private is caught too.
|
||||
func fetchClient(allowPrivate bool) *http.Client {
|
||||
d := &net.Dialer{Timeout: 30 * time.Second, KeepAlive: 30 * time.Second}
|
||||
if !allowPrivate {
|
||||
d.Control = func(_, address string, _ syscall.RawConn) error {
|
||||
host, _, err := net.SplitHostPort(address)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
a, err := netip.ParseAddr(host)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if blockedAddr(a.Unmap()) {
|
||||
return fmt.Errorf("%w (%s)", errPrivateSource, a)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
return &http.Client{
|
||||
Transport: &http.Transport{
|
||||
Proxy: http.ProxyFromEnvironment,
|
||||
DialContext: d.DialContext,
|
||||
TLSHandshakeTimeout: 30 * time.Second,
|
||||
ResponseHeaderTimeout: 2 * time.Minute,
|
||||
ForceAttemptHTTP2: true,
|
||||
},
|
||||
// No overall timeout: a model is tens of gigabytes. The dial, the
|
||||
// handshake and the first byte are bounded above; after that a stall
|
||||
// is the context's to end.
|
||||
CheckRedirect: func(req *http.Request, via []*http.Request) error {
|
||||
if len(via) >= 10 {
|
||||
return errors.New("too many redirects")
|
||||
}
|
||||
return checkScheme(req.URL)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func blockedAddr(a netip.Addr) bool {
|
||||
cgnat := netip.MustParsePrefix("100.64.0.0/10")
|
||||
return a.IsLoopback() || a.IsPrivate() || a.IsLinkLocalUnicast() ||
|
||||
a.IsLinkLocalMulticast() || a.IsMulticast() || a.IsUnspecified() || cgnat.Contains(a)
|
||||
}
|
||||
|
||||
// checkScheme allows the two schemes a model is published over. file:// and
|
||||
// the rest would read this machine's own disk.
|
||||
func checkScheme(u *url.URL) error {
|
||||
if u.Scheme != "http" && u.Scheme != "https" {
|
||||
return fmt.Errorf("%q is not an http or https source", u.Redacted())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// validSource checks a source before a download is started for it.
|
||||
func validSource(source string) error {
|
||||
u, err := url.Parse(source)
|
||||
if err != nil || u.Host == "" {
|
||||
return fmt.Errorf("%q is not a URL", source)
|
||||
}
|
||||
return checkScheme(u)
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package router
|
||||
|
||||
import (
|
||||
"net"
|
||||
"net/http"
|
||||
"net/netip"
|
||||
"net/url"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// AdminGuard keeps a browser from being used against the admin listener.
|
||||
//
|
||||
// The listener has no login: reaching it is meant to mean holding the machine,
|
||||
// so it binds to loopback and an operator reaches it over an ssh tunnel. But a
|
||||
// browser on that operator's machine reaches it too, from any page it has
|
||||
// open. A page could POST a model whose profile runs as root on every node,
|
||||
// or — by pointing its own domain at 127.0.0.1 — read pprof's heap, which
|
||||
// holds API keys in the clear. Three checks close that without touching what
|
||||
// ctl and curl do:
|
||||
//
|
||||
// - Host must be an address or localhost. A page that rebinds its own name
|
||||
// to 127.0.0.1 still sends that name.
|
||||
// - A request carrying Origin or Sec-Fetch-Site from another site is
|
||||
// refused. Browsers send them; ctl and curl do not.
|
||||
// - Anything but GET and HEAD must be application/json. A page can send a
|
||||
// form or text/plain anywhere without asking; JSON across origins needs a
|
||||
// preflight this listener never grants.
|
||||
func AdminGuard(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if !addressHost(r.Host) {
|
||||
http.Error(w, "the admin listener answers to an address or localhost, not a name", http.StatusMisdirectedRequest)
|
||||
return
|
||||
}
|
||||
if crossSite(r) {
|
||||
http.Error(w, "cross-site request refused", http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
if r.Method != http.MethodGet && r.Method != http.MethodHead {
|
||||
ct := r.Header.Get("Content-Type")
|
||||
if mt, _, _ := strings.Cut(ct, ";"); strings.TrimSpace(strings.ToLower(mt)) != "application/json" {
|
||||
http.Error(w, "a change to the admin API must be sent as application/json", http.StatusUnsupportedMediaType)
|
||||
return
|
||||
}
|
||||
}
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
func addressHost(hostport string) bool {
|
||||
host := hostport
|
||||
if h, _, err := net.SplitHostPort(hostport); err == nil {
|
||||
host = h
|
||||
}
|
||||
host = strings.Trim(host, "[]")
|
||||
if strings.EqualFold(host, "localhost") {
|
||||
return true
|
||||
}
|
||||
_, err := netip.ParseAddr(host)
|
||||
return err == nil
|
||||
}
|
||||
|
||||
func crossSite(r *http.Request) bool {
|
||||
switch r.Header.Get("Sec-Fetch-Site") {
|
||||
case "", "same-origin", "none":
|
||||
default:
|
||||
return true
|
||||
}
|
||||
if o := r.Header.Get("Origin"); o != "" {
|
||||
u, err := url.Parse(o)
|
||||
if err != nil || !strings.EqualFold(u.Host, r.Host) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package router
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTheAdminListenerRefusesWhatABrowserPageCouldSend(t *testing.T) {
|
||||
h := AdminGuard(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {}))
|
||||
do := func(method, host, ct string, hdr map[string]string) int {
|
||||
r := httptest.NewRequest(method, "/api/v1/models", strings.NewReader("{}"))
|
||||
r.Host = host
|
||||
if ct != "" {
|
||||
r.Header.Set("Content-Type", ct)
|
||||
}
|
||||
for k, v := range hdr {
|
||||
r.Header.Set(k, v)
|
||||
}
|
||||
w := httptest.NewRecorder()
|
||||
h.ServeHTTP(w, r)
|
||||
return w.Code
|
||||
}
|
||||
|
||||
// What ctl and curl send.
|
||||
if got := do("POST", "127.0.0.1:9090", "application/json", nil); got != 200 {
|
||||
t.Errorf("ctl's request was refused: %d", got)
|
||||
}
|
||||
if got := do("GET", "localhost:9090", "", nil); got != 200 {
|
||||
t.Errorf("a GET through an ssh tunnel was refused: %d", got)
|
||||
}
|
||||
if got := do("DELETE", "[::1]:9090", "application/json", nil); got != 200 {
|
||||
t.Errorf("IPv6 loopback refused: %d", got)
|
||||
}
|
||||
|
||||
// A page posting a form, or text/plain, needs no preflight.
|
||||
if got := do("POST", "127.0.0.1:9090", "text/plain", nil); got != http.StatusUnsupportedMediaType {
|
||||
t.Errorf("text/plain POST accepted: %d", got)
|
||||
}
|
||||
if got := do("POST", "127.0.0.1:9090", "", nil); got != http.StatusUnsupportedMediaType {
|
||||
t.Errorf("a bodiless POST accepted: %d", got)
|
||||
}
|
||||
// DNS rebinding: the page's own name, now resolving to 127.0.0.1.
|
||||
if got := do("GET", "evil.example:9090", "", nil); got != http.StatusMisdirectedRequest {
|
||||
t.Errorf("a rebound name was answered: %d", got)
|
||||
}
|
||||
// And anything a browser marks as coming from elsewhere.
|
||||
if got := do("POST", "127.0.0.1:9090", "application/json",
|
||||
map[string]string{"Sec-Fetch-Site": "cross-site"}); got != http.StatusForbidden {
|
||||
t.Errorf("cross-site fetch accepted: %d", got)
|
||||
}
|
||||
if got := do("POST", "127.0.0.1:9090", "application/json",
|
||||
map[string]string{"Origin": "https://evil.example"}); got != http.StatusForbidden {
|
||||
t.Errorf("foreign Origin accepted: %d", got)
|
||||
}
|
||||
}
|
||||
@@ -22,6 +22,9 @@ type Info struct {
|
||||
ID string
|
||||
// URL is the engine's root address (e.g. http://gpu-a-01:8000), no trailing slash.
|
||||
URL string
|
||||
// APIKey is what the engine demands of every request, sent as a bearer
|
||||
// token. Empty for an engine that asks for none.
|
||||
APIKey string
|
||||
// Models lists the model names this replica serves.
|
||||
Models []string
|
||||
// Engine names the inference engine serving here, which decides what its
|
||||
|
||||
@@ -148,6 +148,8 @@ type Request struct {
|
||||
Body []byte
|
||||
// Header carries headers to forward (authorization is deliberately not).
|
||||
Header http.Header
|
||||
// APIKey is the engine's own key, sent in place of the client's.
|
||||
APIKey string
|
||||
// Stream says whether an SSE response is expected.
|
||||
Stream bool
|
||||
// SuppressUsageChunk drops the final usage-only chunk from the client while
|
||||
@@ -218,6 +220,9 @@ func (p *Proxy) Do(ctx context.Context, w http.ResponseWriter, r Request) (Outco
|
||||
}
|
||||
req.Header = cloneForUpstream(r.Header)
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if r.APIKey != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+r.APIKey)
|
||||
}
|
||||
req.ContentLength = int64(len(r.Body))
|
||||
|
||||
// A non-streaming request waits for the whole completion before a single
|
||||
@@ -518,8 +523,7 @@ func copyHeaders(dst, src http.Header) {
|
||||
|
||||
// cloneForUpstream forwards only what the engine needs. The client's
|
||||
// Authorization header is deliberately dropped: it authenticates the caller to
|
||||
// llmgw, not llmgw to the engine, and engines are reached over a trusted
|
||||
// network (plan §14).
|
||||
// llmgw, not llmgw to the engine. llmgw's own key for the engine is set by Do.
|
||||
func cloneForUpstream(h http.Header) http.Header {
|
||||
out := make(http.Header, 4)
|
||||
for _, k := range []string{"Accept", "User-Agent", "Traceparent", "Tracestate"} {
|
||||
|
||||
@@ -355,3 +355,22 @@ func TestAStreamingRequestKeepsTheShortLivenessTimeout(t *testing.T) {
|
||||
t.Errorf("took %s; the streaming liveness timeout is not being applied", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheEngineGetsLlmgwsKeyNotTheClients(t *testing.T) {
|
||||
var got string
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
got = r.Header.Get("Authorization")
|
||||
io.WriteString(w, `{}`)
|
||||
}))
|
||||
defer up.Close()
|
||||
|
||||
client := http.Header{"Authorization": {"Bearer sk-the-callers-own"}}
|
||||
_, err := testProxy().Do(t.Context(), httptest.NewRecorder(),
|
||||
Request{URL: up.URL, Body: []byte(`{}`), Header: client, APIKey: "llmgw-engine"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got != "Bearer llmgw-engine" {
|
||||
t.Errorf("the engine was sent %q", got)
|
||||
}
|
||||
}
|
||||
@@ -373,6 +373,7 @@ func toInfos(eps []config.Endpoint) []endpoint.Info {
|
||||
out = append(out, endpoint.Info{
|
||||
ID: e.ID,
|
||||
URL: strings.TrimRight(e.URL, "/"),
|
||||
APIKey: e.APIKey,
|
||||
Models: e.Models,
|
||||
Engine: e.Engine,
|
||||
Pool: e.Pool,
|
||||
@@ -673,6 +674,7 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa
|
||||
|
||||
out, perr := r.px.Do(ctx, w, proxy.Request{
|
||||
URL: dec.Endpoint.URL + req.Path,
|
||||
APIKey: dec.Endpoint.APIKey,
|
||||
Body: req.UpstreamBody,
|
||||
Header: req.Header,
|
||||
Stream: req.Stream,
|
||||
|
||||
@@ -157,7 +157,7 @@ func (p *Poller) loop(ctx context.Context, in endpoint.Info) {
|
||||
}
|
||||
|
||||
func (p *Poller) once(ctx context.Context, in endpoint.Info, reported *string) {
|
||||
stats, err := p.fetch(ctx, in.URL, p.mappingFor(in))
|
||||
stats, err := p.fetch(ctx, in.URL, in.APIKey, p.mappingFor(in))
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
@@ -201,7 +201,7 @@ func (p *Poller) report(in endpoint.Info, err error, reported *string) {
|
||||
*reported = "unreachable"
|
||||
}
|
||||
|
||||
func (p *Poller) fetch(ctx context.Context, base string, mapping Mapping) (endpoint.Stats, error) {
|
||||
func (p *Poller) fetch(ctx context.Context, base, apiKey string, mapping Mapping) (endpoint.Stats, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, p.cfg.Timeout)
|
||||
defer cancel()
|
||||
|
||||
@@ -209,6 +209,10 @@ func (p *Poller) fetch(ctx context.Context, base string, mapping Mapping) (endpo
|
||||
if err != nil {
|
||||
return endpoint.Stats{}, err
|
||||
}
|
||||
// llama.cpp guards /metrics with its key; vLLM does not, and ignores one.
|
||||
if apiKey != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+apiKey)
|
||||
}
|
||||
resp, err := p.http.Do(req)
|
||||
if err != nil {
|
||||
return endpoint.Stats{}, err
|
||||
|
||||
@@ -38,6 +38,10 @@ var copyOrder = []string{
|
||||
"threads",
|
||||
"thread_messages",
|
||||
"memory_notes",
|
||||
// Copied, not regenerated: every running engine was started with a key
|
||||
// derived from this, and a new secret would lock the router out of all
|
||||
// of them at the moment of the move.
|
||||
"cluster_secrets",
|
||||
}
|
||||
|
||||
// notCopied are tables deliberately left behind.
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
-- Secrets the control planes share and nobody types: generated on first use,
|
||||
-- read by every control plane against this store afterwards.
|
||||
--
|
||||
-- The first is the one each replica's engine key is derived from. It has to be
|
||||
-- the same for every control plane — the router sends a replica the key it was
|
||||
-- started with, whichever control plane started it — and the same after a
|
||||
-- restart, when running replicas are adopted rather than restarted.
|
||||
CREATE TABLE cluster_secrets (
|
||||
name TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL,
|
||||
created_at BIGINT NOT NULL
|
||||
);
|
||||
@@ -0,0 +1,12 @@
|
||||
-- Secrets the control planes share and nobody types: generated on first use,
|
||||
-- read by every control plane against this store afterwards.
|
||||
--
|
||||
-- The first is the one each replica's engine key is derived from. It has to be
|
||||
-- the same for every control plane — the router sends a replica the key it was
|
||||
-- started with, whichever control plane started it — and the same after a
|
||||
-- restart, when running replicas are adopted rather than restarted.
|
||||
CREATE TABLE cluster_secrets (
|
||||
name TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL,
|
||||
created_at BIGINT NOT NULL
|
||||
);
|
||||
@@ -0,0 +1,31 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ClusterSecret returns the named secret, creating a random one the first time
|
||||
// it is asked for.
|
||||
//
|
||||
// Insert-if-absent and then read, so two control planes asking at once end up
|
||||
// with the same value: the loser's insert changes nothing and it reads the
|
||||
// winner's.
|
||||
func (s *DB) ClusterSecret(ctx context.Context, name string) (string, error) {
|
||||
b := make([]byte, 32)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if _, err := s.db.ExecContext(ctx,
|
||||
`INSERT INTO cluster_secrets (name, value, created_at) VALUES (?, ?, ?)
|
||||
ON CONFLICT (name) DO NOTHING`,
|
||||
name, hex.EncodeToString(b), time.Now().UnixMilli()); err != nil {
|
||||
return "", err
|
||||
}
|
||||
var v string
|
||||
err := s.db.QueryRowContext(ctx,
|
||||
`SELECT value FROM cluster_secrets WHERE name = ?`, name).Scan(&v)
|
||||
return v, err
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestAClusterSecretIsMadeOnceAndKept(t *testing.T) {
|
||||
db, err := Open(t.Context(), filepath.Join(t.TempDir(), "s.db"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
a, err := db.ClusterSecret(t.Context(), "engine-key")
|
||||
if err != nil || len(a) != 64 {
|
||||
t.Fatalf("%q %v", a, err)
|
||||
}
|
||||
b, err := db.ClusterSecret(t.Context(), "engine-key")
|
||||
if err != nil || a != b {
|
||||
t.Errorf("a second read made a new secret: %q then %q", a, b)
|
||||
}
|
||||
if c, _ := db.ClusterSecret(t.Context(), "other"); c == a {
|
||||
t.Error("two names share a secret")
|
||||
}
|
||||
}
|
||||
@@ -303,7 +303,7 @@ func TestThePageRendersThroughTheRealClient(t *testing.T) {
|
||||
|
||||
log := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||
srv := httptest.NewServer(
|
||||
mirror.NewAPI(mirror.NewManager(dir, log), "http://mirror:8081").Handler())
|
||||
mirror.NewAPI(privateOK(mirror.NewManager(dir, log)), "http://mirror:8081").Handler())
|
||||
defer srv.Close()
|
||||
|
||||
c, err := mirror.NewClient(srv.URL, tlsconf.Config{}, 0)
|
||||
@@ -648,3 +648,10 @@ func TestAModelsFamilyIsHowItsNameStarts(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// privateOK lets a test's manager fetch from the httptest upstreams on
|
||||
// loopback, which a real one refuses.
|
||||
func privateOK(m *mirror.Manager) *mirror.Manager {
|
||||
m.AllowPrivateSources()
|
||||
return m
|
||||
}
|
||||
Reference in new issue
Block a user