diff --git a/internal/cli/mirror.go b/internal/cli/mirror.go index 113d4c7..8cda330 100644 --- a/internal/cli/mirror.go +++ b/internal/cli/mirror.go @@ -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) diff --git a/internal/cli/mirror_gate_test.go b/internal/cli/mirror_gate_test.go new file mode 100644 index 0000000..746d9ee --- /dev/null +++ b/internal/cli/mirror_gate_test.go @@ -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) + } +} diff --git a/internal/mirror/add.go b/internal/mirror/add.go index bf22f95..fbe10fe 100644 --- a/internal/mirror/add.go +++ b/internal/mirror/add.go @@ -28,6 +28,15 @@ 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") } @@ -49,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) } diff --git a/internal/mirror/api_test.go b/internal/mirror/api_test.go index 4252b04..dbf41d4 100644 --- a/internal/mirror/api_test.go +++ b/internal/mirror/api_test.go @@ -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) + } + } +} diff --git a/internal/mirror/manager.go b/internal/mirror/manager.go index ff4ee06..a69752c 100644 --- a/internal/mirror/manager.go +++ b/internal/mirror/manager.go @@ -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 diff --git a/internal/mirror/safefetch.go b/internal/mirror/safefetch.go new file mode 100644 index 0000000..bfa37ac --- /dev/null +++ b/internal/mirror/safefetch.go @@ -0,0 +1,87 @@ +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) +} + diff --git a/internal/web/admin_mirror_test.go b/internal/web/admin_mirror_test.go index efe6e5c..5e207b4 100644 --- a/internal/web/admin_mirror_test.go +++ b/internal/web/admin_mirror_test.go @@ -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 +}