feat(addons): ask the Authority on every host operation call

The broker copied the allowed prefixes when an add-on started, so a
change of grants needed a restart of the add-on. Take an Authority
instead and ask it on each call: a revoke or a deny applies to the next
call, and a call that already runs finishes. A refused call answers the
same text whatever the reason; the reason goes to the log.

The supervisor opens the host endpoint when the add-on holds an
operation or its manifest declares one, so an add-on with an optional
operation and no grant starts degraded and a later grant works at once.
The broker now exists whenever the add-on manager does. The manager,
the apps list and the tool filters read the same Authority, which is the
startup policy until another one is set. Admission also tells a denied
operation from a missing grant.

Assisted-by: Claude:claude-sonnet-5-5 go
This commit is contained in:
Ettore Di Giacinto committed 2026-11-03 21:33:36 +00:00
1 parent fdc3c78358
commit c605b26fc7
16 files changed
+417 -70

No files matched your search

+90 -24
View File
@@ -10,6 +10,7 @@ import (
"net"
"net/http"
"path/filepath"
"slices"
"strings"
"time"
@@ -47,8 +48,10 @@ func (a *Application) AddonToolProvider() *mcpTools.AddonToolProvider {
p := mcpTools.NewAddonToolProvider(addontools.NewToolAdapter(a.addonGateway), a.AddonOwnershipMode())
if db := a.AuthDB(); db != nil {
has := addonFeatureCheck(db)
hostOps := a.applicationConfig.AddonHostOps
p = p.WithHidden(func(actor contract.Actor) map[string]bool { return addonsWithoutFeature(hostOps, has, actor) })
authority := a.addonAuthority()
p = p.WithHidden(func(actor contract.Actor) map[string]bool {
return addonsWithoutFeature(authority, a.installedAddonIDs(), has, actor)
})
}
return p
}
@@ -57,19 +60,57 @@ func (a *Application) AddonToolProvider() *mcpTools.AddonToolProvider {
// when this process offers none.
func (a *Application) AddonHostOps() *addonruntime.HostOps { return a.addonHostOps }
// addonAuthority is what every reader of add-on host operation grants asks:
// the broker on each call, the manager at admission, and the tool filters. It
// is the Authority set with SetAddonAuthority, and until one is set, the
// startup policy of the deployment, which never changes.
func (a *Application) addonAuthority() hostops.Authority {
a.addonAuthorityOnce.Do(func() {
if a.hostAuthority == nil {
var floor map[string][]string
if a.applicationConfig != nil {
floor = a.applicationConfig.AddonHostOps
}
a.hostAuthority = hostops.NewStaticAuthority(floor)
}
})
return a.hostAuthority
}
// SetAddonAuthority sets the Authority of the add-on host operations. It must
// be called once, before the add-on manager starts; a later call is ignored,
// so a grant never changes the rule by which it is checked.
func (a *Application) SetAddonAuthority(auth hostops.Authority) {
a.addonAuthorityOnce.Do(func() { a.hostAuthority = auth })
}
// installedAddonIDs lists the add-ons the owner knows, or none in a process
// without the add-on manager.
func (a *Application) installedAddonIDs() []string {
if a.addonManager == nil {
return nil
}
ids, err := a.addonManager.InstalledIDs(context.Background())
if err != nil {
return nil
}
return ids
}
// newAddonHostOps builds the host operation broker for the supervisor of the
// owner. A frontend without the add-on manager, or a host whose setting names
// no add-on, builds nothing, so the channel fails closed. The setting is a
// host setting read at startup; no manifest or request can add to it.
// owner. A frontend without the add-on manager builds nothing, so the channel
// fails closed. The broker exists whenever the manager does, even when no
// grant exists yet: a grant made later applies to the next call, with no
// restart. What an add-on may call is the Authority; no manifest or request can
// add to it.
func (a *Application) newAddonHostOps(m *addons.Manager, limits contract.Limits) *addonruntime.HostOps {
allowed := a.applicationConfig.AddonHostOps
if m == nil || len(allowed) == 0 {
if m == nil {
return nil
}
in := hostops.NewInFlight()
broker := hostops.New(in, addonFeatureCheck(a.authDB), limits)
broker.SetAdminCheck(addonAdminCheck(a.authDB))
return &addonruntime.HostOps{Broker: broker, InFlight: in, Allowed: allowed}
return &addonruntime.HostOps{Broker: broker, InFlight: in, Authority: a.addonAuthority()}
}
// RegisterAddonEngineJobs registers the engine job operations with the services
@@ -575,7 +616,7 @@ func (a *Application) StudioAddonTools() studio.AddonTools {
if db := a.AuthDB(); db != nil {
has = addonFeatureCheck(db)
}
return &studioAddons{adapter: addontools.NewChatToolAdapter(a.addonGateway), mode: a.AddonOwnershipMode(), hostOps: a.applicationConfig.AddonHostOps, has: has}
return &studioAddons{adapter: addontools.NewChatToolAdapter(a.addonGateway), mode: a.AddonOwnershipMode(), hostOps: a.addonAuthority(), ids: a.installedAddonIDs, has: has}
}
// lateStudioAddons asks for the broker at each call: the chats service starts
@@ -659,9 +700,11 @@ func (l lateStudioAddons) Call(ctx context.Context, user, conv, name string, arg
type studioAddons struct {
adapter *addontools.ToolAdapter
mode string
// hostOps maps an add-on ID to the host operation prefixes it may call. An
// add-on that may call the workspace operations needs the Files workspace.
hostOps map[string][]string
// hostOps says which host operations an add-on may call. An add-on that may
// call the workspace operations needs the Files workspace.
hostOps hostops.Authority
// ids lists the installed add-ons, for the rules that name add-ons.
ids func() []string
// has answers whether a user holds a host feature. Nil offers every tool.
has func(userID, feature string) bool
}
@@ -673,26 +716,41 @@ var hostOpFeatures = map[string]string{"finetune.": auth.FeatureFineTuning, "qua
// addonsWithoutFeature names the add-ons whose host operations need a feature
// the actor does not hold. Local-owner mode has no accounts and hides none.
func addonsWithoutFeature(hostOps map[string][]string, has func(userID, feature string) bool, actor contract.Actor) map[string]bool {
if has == nil || actor.Mode == auth.AddonModeLocalOwner {
func addonsWithoutFeature(authority hostops.Authority, ids []string, has func(userID, feature string) bool, actor contract.Actor) map[string]bool {
if has == nil || actor.Mode == auth.AddonModeLocalOwner || authority == nil {
return nil
}
denied := map[string]bool{}
for addon, prefixes := range hostOps {
for _, p := range prefixes {
if f, ok := hostOpFeatures[p]; ok && !has(actor.UserID, f) {
denied[addon] = true
for _, addon := range ids {
for _, p := range heldOperations(authority, addon) {
for family, f := range hostOpFeatures {
if strings.HasPrefix(p, family) && !has(actor.UserID, f) {
denied[addon] = true
}
}
}
}
return denied
}
// heldOperations lists the operation prefixes and names an add-on holds now.
func heldOperations(authority hostops.Authority, addonID string) []string {
if authority == nil {
return nil
}
set := authority.Snapshot(addonID)
return append(slices.Clone(set.Prefixes), set.Operations...)
}
// filterByFeature drops the tools of the add-ons whose host operations need a
// feature the actor does not hold, so a model is not offered a tool that can
// only answer "no permission". Local-owner mode has no accounts and keeps all.
func filterByFeature(tools []studio.AddonTool, hostOps map[string][]string, has func(userID, feature string) bool, actor contract.Actor) []studio.AddonTool {
denied := addonsWithoutFeature(hostOps, has, actor)
func filterByFeature(tools []studio.AddonTool, hostOps hostops.Authority, has func(userID, feature string) bool, actor contract.Actor) []studio.AddonTool {
ids := make([]string, 0, len(tools))
for _, t := range tools {
ids = append(ids, t.AddonID)
}
denied := addonsWithoutFeature(hostOps, ids, has, actor)
if len(denied) == 0 {
return tools
}
@@ -705,9 +763,17 @@ func filterByFeature(tools []studio.AddonTool, hostOps map[string][]string, has
return out
}
// installed lists the installed add-ons, or none when no list is wired.
func (s *studioAddons) installed() []string {
if s.ids == nil {
return nil
}
return s.ids()
}
// needsWorkspace says whether the add-on works on the files of the user.
func (s *studioAddons) needsWorkspace(addonID string) bool {
return hasWorkspaceOperation(s.hostOps[addonID])
return hasWorkspaceOperation(heldOperations(s.hostOps, addonID))
}
func hasWorkspaceOperation(operations []string) bool {
@@ -825,7 +891,7 @@ func (s *studioAddons) AddonSkills(ctx context.Context, user string) ([]studio.A
if err != nil {
return nil, err
}
denied := addonsWithoutFeature(s.hostOps, s.has, actor)
denied := addonsWithoutFeature(s.hostOps, s.installed(), s.has, actor)
out := make([]studio.AddonSkillSet, 0, len(list))
for _, a := range list {
if denied[a.AddonID] {
@@ -852,7 +918,7 @@ func (s *studioAddons) AddonContextFiles(ctx context.Context, user string) ([]st
if err != nil {
return nil, err
}
denied := addonsWithoutFeature(s.hostOps, s.has, actor)
denied := addonsWithoutFeature(s.hostOps, s.installed(), s.has, actor)
out := make([]studio.AddonContextSet, 0, len(list))
for _, a := range list {
if denied[a.AddonID] {
@@ -899,7 +965,7 @@ func (s *studioAddons) AddonShadowers(ctx context.Context, user string) ([]studi
if err != nil {
return nil, err
}
denied := addonsWithoutFeature(s.hostOps, s.has, actor)
denied := addonsWithoutFeature(s.hostOps, s.installed(), s.has, actor)
out := make([]studio.AddonShadower, 0, len(list))
for _, a := range list {
if denied[a.AddonID] {
+2 -1
View File
@@ -2,6 +2,7 @@
package application
import (
"github.com/mudler/LocalAI/core/services/addons/hostops"
"slices"
. "github.com/onsi/ginkgo/v2"
@@ -13,7 +14,7 @@ import (
)
var _ = Describe("add-on tools offered to a chat by the features of the person", func() {
hostOps := map[string][]string{"fine-tuning": {"finetune."}, "quantization": {"quantize."}, "sandbox": {"workspace."}}
hostOps := hostops.NewStaticAuthority(map[string][]string{"fine-tuning": {"finetune."}, "quantization": {"quantize."}, "sandbox": {"workspace."}})
all := []studio.AddonTool{{Name: "addon__fine-tuning__fine-tune-list", AddonID: "fine-tuning"}, {Name: "addon__quantization__quantize-list", AddonID: "quantization"},
{Name: "addon__sandbox__bash", AddonID: "sandbox"}, {Name: "addon__tables__tables-create", AddonID: "tables"}}
names := func(in []studio.AddonTool) []string {
+4
View File
@@ -114,6 +114,9 @@ type Application struct {
backendLibrary *backendlib.Installer
addonManager *addons.Manager
addonHostOps *addonruntime.HostOps
// hostAuthority answers what each add-on may call; see addonAuthority.
hostAuthority hostops.Authority
addonAuthorityOnce sync.Once
// addonTransfers holds the handles of connector file transfers; see transfers.
addonTransfers *hostops.Transfers
addonTransfersOnce sync.Once
@@ -1009,6 +1012,7 @@ func (a *Application) startAddons() error {
}
}
m.SetHostOps(a.applicationConfig.AddonHostOps)
m.SetAuthority(a.addonAuthority())
if c.InstanceID == c.OwnerID {
// Data recovery sits beside the manager: an upgrade of an add-on that
// has data goes through it, and its journals are settled before any
+1 -1
View File
@@ -82,7 +82,7 @@ func (a *Application) resolveExecutionBindingWithTools(ctx context.Context, acto
// Workspace-dependent tools cannot be selected without the actual workspace.
eligible := defs[:0:0]
for _, d := range defs {
needsWorkspace := hasWorkspaceOperation(a.applicationConfig.AddonHostOps[d.AddonID])
needsWorkspace := hasWorkspaceOperation(heldOperations(a.addonAuthority(), d.AddonID))
if needsWorkspace && !a.StudioFiles().Available() {
continue
}
+2 -1
View File
@@ -9,6 +9,7 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/mudler/LocalAI/core/services/addons/hostops"
"net"
"net/http"
"net/http/httptest"
@@ -446,7 +447,7 @@ type ordinaryBindingStore struct{ store.Store }
var _ = Describe("workspace operation declarations", func() {
DescribeTable("retains the workspace dot prefix semantics", func(operations []string, want bool) {
Expect(hasWorkspaceOperation(operations)).To(Equal(want))
adapter := &studioAddons{hostOps: map[string][]string{"sample": operations}}
adapter := &studioAddons{hostOps: hostops.NewStaticAuthority(map[string][]string{"sample": operations})}
Expect(adapter.needsWorkspace("sample")).To(Equal(want))
Expect(adapter.needsWorkspace("other")).To(BeFalse())
},
+1 -1
View File
@@ -126,7 +126,7 @@ var _ = Describe("the studio add-ons", func() {
})
It("makes an add-on that needs the workspace unavailable while files are off", func() {
s := &studioAddons{mode: auth.AddonModeNamedUser, hostOps: map[string][]string{"sandbox": {"workspace."}, "office": {"workspace."}, "fine-tuning": {"finetune."}}}
s := &studioAddons{mode: auth.AddonModeNamedUser, hostOps: hostops.NewStaticAuthority(map[string][]string{"sandbox": {"workspace."}, "office": {"workspace."}, "fine-tuning": {"finetune."}})}
ctx := context.Background()
for _, id := range []string{"sandbox", "office"} {
Expect(s.Unavailable(ctx, "alice", id, false)).To(ContainSubstring("workspace"), "%s with files off", id)
+8 -1
View File
@@ -18,7 +18,7 @@ func (e *HostRequirementsRefusal) Error() string {
return fmt.Sprintf("required host operations unavailable: %s; configure host policy and restart", strings.Join(e.Summary.MissingRequired, ", "))
}
func (m *Manager) requests(r packages.PackageReceipt) hostops.RequestSummary {
return hostops.Requests(r.Manifest.Capabilities.HostOperations, hostops.SupportedOperations(), m.hostOps[r.Manifest.ID])
return hostops.RequestsFor(r.Manifest.Capabilities.HostOperations, hostops.SupportedOperations(), m.hostAuthority(), r.Manifest.ID)
}
func (m *Manager) admit(r packages.PackageReceipt) error {
s := m.requests(r)
@@ -53,3 +53,10 @@ func copyRequestSummary(s *hostops.RequestSummary) *hostops.RequestSummary {
out.MissingOptional = slices.Clone(s.MissingOptional)
return &out
}
// held lists what an add-on holds now as operation prefixes and names, the
// form the apps list reads to decide which host controls to draw.
func held(a hostops.Authority, id string) []string {
set := a.Snapshot(id)
return append(slices.Clone(set.Prefixes), set.Operations...)
}
+38 -6
View File
@@ -17,14 +17,35 @@ type RequestSummary struct {
Granted []string `json:"granted,omitempty"`
MissingRequired []string `json:"missingRequired,omitempty"`
MissingOptional []string `json:"missingOptional,omitempty"`
RequiredCount int `json:"requiredCount"`
OptionalCount int `json:"optionalCount"`
GrantedCount int `json:"grantedCount"`
// DeniedRequired and DeniedOptional are the missing operations that a deny
// rule forbids. They are also listed in the Missing lists. An operation
// that is denied can never be approved, so a caller reads this to tell
// "ask the administrator to approve" from "blocked by policy".
DeniedRequired []string `json:"deniedRequired,omitempty"`
DeniedOptional []string `json:"deniedOptional,omitempty"`
RequiredCount int `json:"requiredCount"`
OptionalCount int `json:"optionalCount"`
GrantedCount int `json:"grantedCount"`
}
// Requests is side-effect free. Registry membership is exact; coverage uses the
// broker's literal nonempty prefix rule, not a regexp or a request-as-grant.
func Requests(req *contract.HostOperations, registered, grants []string) RequestSummary {
return RequestsFor(req, registered, NewStaticAuthority(map[string][]string{"": grants}), "")
}
// RequestsFor is Requests against an Authority: an operation counts as granted
// when the authority allows it to the add-on now.
func RequestsFor(req *contract.HostOperations, registered []string, auth Authority, addonID string) RequestSummary {
return requestsWith(req, registered, func(op string) (bool, Reason) {
if auth == nil {
return false, ReasonNotGranted
}
return auth.Allows(addonID, op)
})
}
func requestsWith(req *contract.HostOperations, registered []string, allows func(op string) (bool, Reason)) RequestSummary {
s := RequestSummary{}
if req == nil {
return s
@@ -35,12 +56,23 @@ func Requests(req *contract.HostOperations, registered, grants []string) Request
slices.Sort(s.Optional)
for i, names := range [][]string{s.Required, s.Optional} {
for _, name := range names {
if slices.Contains(registered, name) && hasPrefix(grants, name) {
ok, why := false, ReasonNotGranted
if slices.Contains(registered, name) {
ok, why = allows(name)
}
switch {
case ok:
s.Granted = append(s.Granted, name)
} else if i == 0 {
case i == 0:
s.MissingRequired = append(s.MissingRequired, name)
} else {
if why.Denied() {
s.DeniedRequired = append(s.DeniedRequired, name)
}
default:
s.MissingOptional = append(s.MissingOptional, name)
if why.Denied() {
s.DeniedOptional = append(s.DeniedOptional, name)
}
}
}
}
+23 -6
View File
@@ -8,6 +8,7 @@ import (
"errors"
"fmt"
"github.com/mudler/LocalAI/core/services/addons/modelsprobe"
"github.com/mudler/xlog"
"io"
"net/http"
"strings"
@@ -157,8 +158,10 @@ func refuse(w http.ResponseWriter, status int, code, msg string) {
// Handler serves the channel of one add-on incarnation. credential is the
// secret that incarnation holds; allowed lists the operation prefixes the
// host setting gives the add-on. Whatever the check, the default is refuse.
// It is a fixed list: use HandlerForRuntime with an Authority for grants that
// change while the add-on runs.
func (b *Broker) Handler(addonID, credential string, allowed []string) http.Handler {
return b.HandlerForRuntime(addonID, credential, allowed, 0, "")
return b.HandlerForRuntime(addonID, credential, NewStaticAuthority(map[string][]string{addonID: allowed}), 0, "")
}
// RuntimeEndpoint owns callback admission and draining for one runtime. Stop
@@ -174,8 +177,10 @@ type RuntimeEndpoint struct {
type endpointKey struct{}
func (b *Broker) EndpointForRuntime(addonID, credential string, allowed []string, generation uint64, incarnation string) *RuntimeEndpoint {
return &RuntimeEndpoint{handler: b.HandlerForRuntime(addonID, credential, allowed, generation, incarnation)}
// EndpointForRuntime serves one runtime. auth is asked on every call, so a
// revoke or a deny applies to the next call; a nil auth refuses every call.
func (b *Broker) EndpointForRuntime(addonID, credential string, auth Authority, generation uint64, incarnation string) *RuntimeEndpoint {
return &RuntimeEndpoint{handler: b.HandlerForRuntime(addonID, credential, auth, generation, incarnation)}
}
func (e *RuntimeEndpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -204,8 +209,11 @@ func (e *RuntimeEndpoint) admit() bool {
return !e.closed
}
// HandlerForRuntime authenticates callbacks for exactly one runtime.
func (b *Broker) HandlerForRuntime(addonID, credential string, allowed []string, generation uint64, incarnation string) http.Handler {
// HandlerForRuntime authenticates callbacks for exactly one runtime. The
// authority is asked for every call, after the credential check and before
// anything else of the operation runs. A call that already runs is not
// interrupted by a later revoke.
func (b *Broker) HandlerForRuntime(addonID, credential string, auth Authority, generation uint64, incarnation string) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
refuse(w, http.StatusMethodNotAllowed, contract.HostBadRequest, "POST only")
@@ -234,7 +242,16 @@ func (b *Broker) HandlerForRuntime(addonID, credential string, allowed []string,
refuse(w, http.StatusForbidden, contract.HostScope, "observation callbacks denied")
return
}
if !hasPrefix(allowed, req.Op) {
// The add-on must not learn whether an operation is denied or only not
// granted, so the body is the same. The reason stays on the host.
if auth == nil {
refuse(w, http.StatusForbidden, contract.HostUnknownOp, "operation not allowed")
return
}
if ok, why := auth.Allows(addonID, req.Op); !ok {
if why.Denied() || why == ReasonIdentityMismatch {
xlog.Info("add-on host operation refused", "addon", addonID, "op", req.Op, "reason", string(why))
}
refuse(w, http.StatusForbidden, contract.HostUnknownOp, "operation not allowed")
return
}
@@ -0,0 +1,134 @@
// SPDX-License-Identifier: MIT
package hostops
import (
"context"
"encoding/json"
"net/http"
"sync"
"time"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
// liveAuthority is a test Authority whose grants change while it is in use.
type liveAuthority struct {
mu sync.Mutex
grants map[string]Reason
}
func (l *liveAuthority) set(op string, why Reason) {
l.mu.Lock()
defer l.mu.Unlock()
l.grants[op] = why
}
func (l *liveAuthority) Allows(_, op string) (bool, Reason) {
l.mu.Lock()
defer l.mu.Unlock()
why, ok := l.grants[op]
if !ok {
return false, ReasonNotGranted
}
return why == ReasonNone, why
}
func (l *liveAuthority) Snapshot(string) EffectiveSet { return EffectiveSet{} }
func (l *liveAuthority) Entitled(string) bool { return true }
var _ = Describe("Broker authority", func() {
newLive := func(auth Authority) *fixture {
f := newFixture(always)
f.h = f.b.HandlerForRuntime("fine-tuning", "secret", auth, 0, "")
end := f.inflight.Begin(env("fine-tuning", "op1", "read", time.Now().Add(time.Minute)))
DeferCleanup(end)
return f
}
It("applies a revoke to the next call", func() {
auth := &liveAuthority{grants: map[string]Reason{"finetune.get": ReasonNone}}
f := newLive(auth)
code, _ := f.call("secret", "op1", "finetune.get", `{}`)
Expect(code).To(Equal(http.StatusOK))
auth.set("finetune.get", ReasonDeniedByAdmin)
code, out := f.call("secret", "op1", "finetune.get", `{}`)
Expect(code).To(Equal(http.StatusForbidden))
Expect(out.Error.Code).To(Equal("unknown_op"))
auth.set("finetune.get", ReasonNone)
code, _ = f.call("secret", "op1", "finetune.get", `{}`)
Expect(code).To(Equal(http.StatusOK), "a later grant applies with no restart")
})
It("answers the same to a denied and a not granted operation", func() {
auth := &liveAuthority{grants: map[string]Reason{"finetune.get": ReasonDeniedByPolicyFile, "finetune.list": ReasonIdentityMismatch}}
f := newLive(auth)
codeA, a := f.call("secret", "op1", "finetune.get", `{}`)
codeB, b := f.call("secret", "op1", "finetune.list", `{}`)
codeC, c := f.call("secret", "op1", "finetune.start", `{}`)
Expect([]int{codeA, codeB, codeC}).To(Equal([]int{403, 403, 403}))
Expect(a).To(Equal(b))
Expect(b).To(Equal(c))
})
It("lets a running call finish after the revoke", func() {
auth := &liveAuthority{grants: map[string]Reason{"finetune.get": ReasonNone}}
f := newLive(auth)
started, release := make(chan struct{}), make(chan struct{})
b := New(f.inflight, always, f.b.limits)
Expect(b.Register(Spec{Op: "finetune.get", Scope: "read", Feature: "fine_tuning", Handler: func(context.Context, Call) (json.RawMessage, error) {
close(started)
<-release
return json.RawMessage(`{"done":true}`), nil
}})).To(Succeed())
f.b, f.h = b, b.HandlerForRuntime("fine-tuning", "secret", auth, 0, "")
done := make(chan int, 1)
go func() {
code, _ := f.call("secret", "op1", "finetune.get", `{}`)
done <- code
}()
Eventually(started).Should(BeClosed())
auth.set("finetune.get", ReasonDeniedByAdmin)
close(release)
Eventually(done).Should(Receive(Equal(http.StatusOK)))
code, _ := f.call("secret", "op1", "finetune.get", `{}`)
Expect(code).To(Equal(http.StatusForbidden))
})
It("refuses everything without an authority", func() {
f := newFixture(always)
f.h = f.b.HandlerForRuntime("fine-tuning", "secret", nil, 0, "")
code, _ := f.call("secret", "op1", "finetune.get", `{}`)
Expect(code).To(Equal(http.StatusForbidden))
})
It("is safe under concurrent calls and changes", func() {
auth := &liveAuthority{grants: map[string]Reason{"finetune.get": ReasonNone}}
f := newLive(auth)
b := New(f.inflight, always, f.b.limits)
Expect(b.Register(Spec{Op: "finetune.get", Scope: "read", Feature: "fine_tuning", Handler: func(context.Context, Call) (json.RawMessage, error) {
return json.RawMessage(`{}`), nil
}})).To(Succeed())
f.b, f.h = b, b.HandlerForRuntime("fine-tuning", "secret", auth, 0, "")
var wg sync.WaitGroup
for i := 0; i < 8; i++ {
wg.Add(2)
go func() {
defer wg.Done()
for j := 0; j < 20; j++ {
f.call("secret", "op1", "finetune.get", `{}`)
}
}()
go func(i int) {
defer wg.Done()
for j := 0; j < 20; j++ {
if (i+j)%2 == 0 {
auth.set("finetune.get", ReasonNone)
} else {
auth.set("finetune.get", ReasonDeniedByAdmin)
}
}
}(i)
}
wg.Wait()
})
})
@@ -35,7 +35,7 @@ var _ = Describe("Observation callback boundary", func() {
return json.RawMessage(`{}`), nil
}})).To(Succeed())
}
f.h = f.b.HandlerForRuntime("fine-tuning", "secret", []string{"workspace.", "finetune.", "connector."}, 1, "inc")
f.h = f.b.HandlerForRuntime("fine-tuning", "secret", NewStaticAuthority(map[string][]string{"fine-tuning": []string{"workspace.", "finetune.", "connector."}}), 1, "inc")
e := env("fine-tuning", "status", "read", time.Now().Add(time.Minute))
end, denied, err := f.inflight.BeginObservation(e, Binding{}, "inc", "request")
Expect(err).NotTo(HaveOccurred())
@@ -88,7 +88,7 @@ var _ = Describe("Observation identity checks", func() {
end, denied, err := f.inflight.BeginObservation(e, Binding{}, "inc", "q")
Expect(err).NotTo(HaveOccurred())
DeferCleanup(end)
f.h = f.b.HandlerForRuntime("fine-tuning", "secret", []string{"finetune."}, 2, "replacement")
f.h = f.b.HandlerForRuntime("fine-tuning", "secret", NewStaticAuthority(map[string][]string{"fine-tuning": []string{"finetune."}}), 2, "replacement")
code, _ := f.call("secret", "status", "finetune.get", `{}`)
Expect(code).To(Equal(403))
Expect(denied()).To(BeTrue())
@@ -167,7 +167,7 @@ var _ = Describe("Callback endpoint retirement", func() {
end, _, err := f.BeginObservation(env("fine-tuning", "status", "read", time.Now().Add(time.Minute)), Binding{}, "inc", "q")
Expect(err).NotTo(HaveOccurred())
end()
h := b.EndpointForRuntime("fine-tuning", "secret", []string{"connector."}, 1, "inc")
h := b.EndpointForRuntime("fine-tuning", "secret", NewStaticAuthority(map[string][]string{"fine-tuning": []string{"connector."}}), 1, "inc")
body := &callbackBodyBarrier{Reader: strings.NewReader(`{"operationId":"status","op":"connector.inbound","arguments":{}}`), read: make(chan struct{}), resume: make(chan struct{})}
req := httptest.NewRequest(http.MethodPost, "/host", body)
req.Header.Set(contract.HostCredentialHeader, "secret")
@@ -184,7 +184,7 @@ var _ = Describe("Callback endpoint retirement", func() {
default:
}
Expect(f.callbackDenied("fine-tuning", "status", 1, "inc")).To(BeTrue())
other := &fixture{b: b, inflight: f, h: b.HandlerForRuntime("other", "other-secret", []string{"workspace."}, 1, "other-inc")}
other := &fixture{b: b, inflight: f, h: b.HandlerForRuntime("other", "other-secret", NewStaticAuthority(map[string][]string{"other": []string{"workspace."}}), 1, "other-inc")}
defer f.BeginRuntime(env("other", "ordinary", "read", time.Now().Add(time.Minute)), Binding{}, false, "other-inc")()
code, _ := other.call("other-secret", "ordinary", "workspace.export", `{}`)
Expect(code).To(Equal(http.StatusOK))
@@ -213,7 +213,7 @@ var _ = Describe("Decoded callback dispatch retirement", func() {
calls.Add(1)
return json.RawMessage(`{}`), nil
}})).To(Succeed())
h := b.EndpointForRuntime("fine-tuning", "secret", []string{"workspace."}, 1, "inc")
h := b.EndpointForRuntime("fine-tuning", "secret", NewStaticAuthority(map[string][]string{"fine-tuning": []string{"workspace."}}), 1, "inc")
defer f.BeginRuntime(env("fine-tuning", "ordinary", "read", time.Now().Add(time.Minute)), Binding{}, false, "inc")()
fixture := &fixture{h: h}
result := make(chan int, 1)
@@ -243,7 +243,7 @@ var _ = Describe("Decoded callback dispatch retirement", func() {
h.Stop() // No lifecycle lock may be held across the handler or Drain.
return json.RawMessage(`{}`), nil
}})).To(Succeed())
h = b.EndpointForRuntime("fine-tuning", "secret", []string{"connector."}, 1, "inc")
h = b.EndpointForRuntime("fine-tuning", "secret", NewStaticAuthority(map[string][]string{"fine-tuning": []string{"connector."}}), 1, "inc")
fixture := &fixture{h: h}
done := make(chan struct{})
go func() { fixture.call("secret", "ordinary", "connector.inbound", `{}`); close(done) }()
+32 -1
View File
@@ -14,6 +14,7 @@ import (
"sort"
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
@@ -21,6 +22,7 @@ import (
"github.com/mudler/LocalAI/core/services/addons/apps"
"github.com/mudler/LocalAI/core/services/addons/catalog"
"github.com/mudler/LocalAI/core/services/addons/hostops"
"github.com/mudler/LocalAI/core/services/addons/packages"
"github.com/mudler/LocalAI/core/services/addons/recovery"
addonruntime "github.com/mudler/LocalAI/core/services/addons/runtime"
@@ -158,6 +160,33 @@ type Manager struct {
// add-on. The apps list reads it to decide which host controls it draws.
hostOps map[string][]string
hostOpsSet bool
// authority answers what each add-on may call now. Nil means the startup
// snapshot in hostOps. It is read on every admission and every listing,
// so a grant or a revoke applies without a restart.
authority atomic.Pointer[hostops.Authority]
floor atomic.Pointer[hostops.Authority]
}
// SetAuthority sets the Authority that admission and the apps list ask. The
// first call wins; set it before attaching the runtime. Without it the
// startup snapshot of SetHostOps is the authority.
func (m *Manager) SetAuthority(a hostops.Authority) {
if a == nil {
return
}
m.authority.CompareAndSwap(nil, &a)
}
// hostAuthority is the Authority in force: the one set, or the startup
// snapshot of host operation prefixes.
func (m *Manager) hostAuthority() hostops.Authority {
if a := m.authority.Load(); a != nil {
return *a
}
if f := m.floor.Load(); f != nil {
return *f
}
return hostops.NewStaticAuthority(nil)
}
// SetHostOps sets the host operation grants of the add-ons (the host setting,
@@ -175,6 +204,8 @@ func (m *Manager) SetHostOps(ops map[string][]string) {
for id, prefixes := range ops {
m.hostOps[id] = slices.Clone(prefixes)
}
floor := hostops.NewStaticAuthority(m.hostOps)
m.floor.Store(&floor)
}
// SkillSync copies the skills of an add-on package into the chats' skill layer
@@ -1231,7 +1262,7 @@ func (m *Manager) Apps(ctx context.Context, actor contract.Actor) ([]apps.App, e
if !ok {
continue
}
snap := apps.Snapshot{AddonID: id, InstalledAt: st.InstalledAt, Installed: st.Installed, Enabled: st.DesiredEnabled, Generation: st.Generation, State: addonruntime.StateStopped, Receipt: receipt, HostOps: m.hostOps[id]}
snap := apps.Snapshot{AddonID: id, InstalledAt: st.InstalledAt, Installed: st.Installed, Enabled: st.DesiredEnabled, Generation: st.Generation, State: addonruntime.StateStopped, Receipt: receipt, HostOps: held(m.hostAuthority(), id)}
if st.DesiredEnabled && m.runtime != nil {
snap.State = m.runtime.State(id, st.Generation)
}
@@ -68,7 +68,7 @@ var _ = Describe("Host-only observation read", func() {
calls.Add(1)
return json.RawMessage(`{}`), nil
}})).To(Succeed())
callback = b.HandlerForRuntime("sample", "secret", []string{"connector."}, 1, "inc")
callback = b.HandlerForRuntime("sample", "secret", hostops.NewStaticAuthority(map[string][]string{"sample": []string{"connector."}}), 1, "inc")
_, err := c.ReadObservation(context.Background(), env(), "q")
Expect(err).To(HaveOccurred())
Expect(code.Load()).To(Equal(int32(403)))
@@ -115,7 +115,7 @@ var _ = Describe("Host-only observation read", func() {
Expect(err).NotTo(HaveOccurred())
})
Expect(b.Register(hostops.Spec{Op: "workspace.export", Scope: "read", Feature: "chats", Handler: func(context.Context, hostops.Call) (json.RawMessage, error) { return json.RawMessage(`{}`), nil }})).To(Succeed())
callback = b.HandlerForRuntime("sample", "secret", []string{"workspace."}, 1, "inc")
callback = b.HandlerForRuntime("sample", "secret", hostops.NewStaticAuthority(map[string][]string{"sample": []string{"workspace."}}), 1, "inc")
done := make(chan error, 1)
go func() { _, err := c.ReadObservation(context.Background(), env(), "q"); done <- err }()
<-entered
+25 -18
View File
@@ -72,8 +72,9 @@ type Config struct {
DataRoot string
Limits contract.Limits
ReadyTimeout, StableHealth time.Duration
// HostOps, when set, gives each add-on that its Allowed map names a
// host operation endpoint. Nil means no add-on gets one.
// HostOps, when set, gives each add-on that holds a host operation, or
// declares one in its manifest, a host operation endpoint. Nil means no
// add-on gets one.
HostOps *HostOps
// Output receives each line an add-on writes to its standard output
// (stream "stdout") or error ("stderr"). Nil sends the lines to the LocalAI
@@ -81,21 +82,28 @@ type Config struct {
Output func(addonID, stream, line string)
}
// HostOps wires the host operation channel into the supervisor. Allowed maps
// an add-on ID to the operation prefixes it may call; an add-on that is not a
// key gets no endpoint and no credential. InFlight records the envelope of
// each request while the add-on serves it.
// HostOps wires the host operation channel into the supervisor. Authority
// decides, on every call, what an add-on may run; it is not copied at start, so
// a grant or a revoke applies without a restart. An add-on gets an endpoint and
// a credential when it holds an operation or its manifest declares one: a
// declared operation that is not granted yet is refused by the endpoint, and a
// later grant works at once. An add-on with neither gets none. InFlight records
// the envelope of each request while the add-on serves it.
type HostOps struct {
Broker *hostops.Broker
InFlight *hostops.InFlight
Allowed map[string][]string
Broker *hostops.Broker
InFlight *hostops.InFlight
Authority hostops.Authority
}
func (h *HostOps) prefixes(id string) []string {
if h == nil || h.Broker == nil || h.InFlight == nil {
return nil
// wants reports whether the add-on gets a host operation endpoint.
func (h *HostOps) wants(id string, declared *contract.HostOperations) bool {
if h == nil || h.Broker == nil || h.InFlight == nil || h.Authority == nil {
return false
}
return h.Allowed[id]
if declared != nil && len(declared.Required)+len(declared.Optional) > 0 {
return true
}
return h.Authority.Entitled(id)
}
type Registration struct {
@@ -321,9 +329,8 @@ func (s *Supervisor) launch(ctx context.Context, pkg packages.PackageReceipt, ge
// failed launch, and by serveHostOps when the process ends.
var hostListener net.Listener
var hostCredential string
hostPrefixes := s.config.HostOps.prefixes(pkg.Manifest.ID)
hostReady := false
if len(hostPrefixes) > 0 {
if s.config.HostOps.wants(pkg.Manifest.ID, pkg.Manifest.Capabilities.HostOperations) {
hostListener, e = net.ListenTCP("tcp4", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)})
if e != nil {
return nil, ErrRuntime
@@ -432,7 +439,7 @@ func (s *Supervisor) launch(ctx context.Context, pkg packages.PackageReceipt, ge
return nil, ErrRuntime
}
c.inflight = s.config.HostOps.InFlight
s.serveHostOps(p, hostListener, pkg.Manifest.ID, hostCredential, hostPrefixes)
s.serveHostOps(p, hostListener, pkg.Manifest.ID, hostCredential)
}
ident, identityErr := identity(cmd.Process.Pid)
ident.Incarnation = r.Incarnation
@@ -494,8 +501,8 @@ func (s *Supervisor) launch(ctx context.Context, pkg packages.PackageReceipt, ge
// serveHostOps serves the host operation channel of one incarnation and
// stops admission when the process ends. Cleanup drains active requests before
// releasing correlations; a handler ignoring cancellation cannot block terminate.
func (s *Supervisor) serveHostOps(p *process, l net.Listener, id, credential string, prefixes []string) {
endpoint := s.config.HostOps.Broker.EndpointForRuntime(id, credential, prefixes, p.registration.Generation, p.registration.Incarnation)
func (s *Supervisor) serveHostOps(p *process, l net.Listener, id, credential string) {
endpoint := s.config.HostOps.Broker.EndpointForRuntime(id, credential, s.config.HostOps.Authority, p.registration.Generation, p.registration.Incarnation)
srv := &http.Server{Handler: endpoint, ReadHeaderTimeout: 5 * time.Second, MaxHeaderBytes: 16 << 10}
go func() { _ = srv.Serve(l) }()
go func() {
@@ -681,6 +681,7 @@ var _ = Describe("Addon data directory", func() {
entries, _ := os.ReadDir(root)
Expect(entries).To(BeEmpty())
})
var hostOpsSupervisorWith func(hostops.Authority) (*Supervisor, *hostops.InFlight)
hostOpsSeen := func(p packages.PackageReceipt) (endpoint, credential string) {
var b []byte
Eventually(func() error {
@@ -692,6 +693,9 @@ var _ = Describe("Addon data directory", func() {
return parts[0], parts[1]
}
hostOpsSupervisor := func(allowed map[string][]string) (*Supervisor, *hostops.InFlight) {
return hostOpsSupervisorWith(hostops.NewStaticAuthority(allowed))
}
hostOpsSupervisorWith = func(auth hostops.Authority) (*Supervisor, *hostops.InFlight) {
l := contract.DefaultLimits()
l.ReadTimeout = 300 * time.Millisecond
l.ShutdownTimeout = 100 * time.Millisecond
@@ -700,7 +704,7 @@ var _ = Describe("Addon data directory", func() {
Expect(b.Register(hostops.Spec{Op: "finetune.get", Scope: "read", Feature: "fine_tuning", Handler: func(context.Context, hostops.Call) (json.RawMessage, error) {
return json.RawMessage(`{"ok":true}`), nil
}})).To(Succeed())
s, e := NewSupervisor(Config{StateRoot: GinkgoT().TempDir(), Limits: l, ReadyTimeout: 2 * time.Second, StableHealth: time.Second, HostOps: &HostOps{Broker: b, InFlight: in, Allowed: allowed}})
s, e := NewSupervisor(Config{StateRoot: GinkgoT().TempDir(), Limits: l, ReadyTimeout: 2 * time.Second, StableHealth: time.Second, HostOps: &HostOps{Broker: b, InFlight: in, Authority: auth}})
Expect(e).NotTo(HaveOccurred())
DeferCleanup(func() { Expect(s.Close(context.Background())).To(Succeed()) })
return s, in
@@ -735,6 +739,37 @@ var _ = Describe("Addon data directory", func() {
Expect(call(credential)).To(Equal(http.StatusForbidden))
})
It("opens an endpoint for a declared operation that is not granted, and a later grant works with no restart", func() {
auth := &switchAuthority{}
s, in := hostOpsSupervisorWith(auth)
p := withID("sample")
p.Manifest.Capabilities.HostOperations = &contract.HostOperations{Optional: []string{"finetune.get"}}
raw, e := json.Marshal(p)
Expect(e).NotTo(HaveOccurred())
Expect(os.Chmod(filepath.Join(p.Directory, "receipt.json"), 0600)).To(Succeed())
Expect(os.WriteFile(filepath.Join(p.Directory, "receipt.json"), raw, 0400)).To(Succeed())
inst, e := s.Start(context.Background(), p, 1)
Expect(e).NotTo(HaveOccurred())
endpoint, credential := hostOpsSeen(p)
Expect(endpoint).To(HavePrefix("http://127.0.0.1:"))
call := func() int {
body := strings.NewReader(`{"operationId":"op1","op":"finetune.get","arguments":{}}`)
req, _ := http.NewRequest("POST", endpoint+"/host", body)
req.Header.Set(contract.HostCredentialHeader, credential)
resp, e := http.DefaultClient.Do(req)
Expect(e).NotTo(HaveOccurred())
_ = resp.Body.Close()
return resp.StatusCode
}
end := in.BeginRuntime(contract.Envelope{Request: contract.Request{AddonID: "sample", OperationID: "op1", Generation: 1, Action: "a", Arguments: json.RawMessage(`{}`)}, Authority: contract.Authority{Actor: contract.Actor{UserID: "u", CredentialID: "c", Mode: "named-user"}, Scope: "read", ExpiresAt: time.Now().Add(time.Minute)}}, hostops.Binding{}, false, inst.Incarnation)
defer end()
Expect(call()).To(Equal(http.StatusForbidden))
auth.grant.Store(true)
Expect(call()).To(Equal(http.StatusOK))
auth.grant.Store(false)
Expect(call()).To(Equal(http.StatusForbidden))
})
It("gives an add-on that the host setting does not name no host endpoint", func() {
s, _ := hostOpsSupervisor(map[string][]string{"other": {"finetune."}})
p := withID("sample")
@@ -768,3 +803,15 @@ var _ = Describe("Addon data directory", func() {
}, 2*time.Second).Should(HaveOccurred())
})
})
// switchAuthority grants finetune.get to every add-on while grant is set.
type switchAuthority struct{ grant atomic.Bool }
func (a *switchAuthority) Allows(_, op string) (bool, hostops.Reason) {
if a.grant.Load() && op == "finetune.get" {
return true, hostops.ReasonNone
}
return false, hostops.ReasonNotGranted
}
func (a *switchAuthority) Snapshot(string) hostops.EffectiveSet { return hostops.EffectiveSet{} }
func (a *switchAuthority) Entitled(string) bool { return a.grant.Load() }
+1 -1
View File
@@ -143,7 +143,7 @@ func startAddonFrom(id, packageDir string) *world {
sup, err := addonruntime.NewSupervisor(addonruntime.Config{
StateRoot: filepath.Join(root, "state"), DataRoot: dataRoot, Limits: limits,
ReadyTimeout: 20 * time.Second, StableHealth: time.Second,
HostOps: &addonruntime.HostOps{Broker: broker, InFlight: inflight, Allowed: map[string][]string{id: {"workspace."}}},
HostOps: &addonruntime.HostOps{Broker: broker, InFlight: inflight, Authority: hostops.NewStaticAuthority(map[string][]string{id: {"workspace."}})},
})
Expect(err).NotTo(HaveOccurred())
DeferCleanup(func() { _ = sup.Close(context.Background()) })