Update dependencies

vendor/tailscale.com/util/syspolicy/rsop/change_callbacks.go

@@ -0,0 +1,107 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package rsop
+
+import (
+	"reflect"
+	"slices"
+	"sync"
+	"time"
+
+	"tailscale.com/util/set"
+	"tailscale.com/util/syspolicy/internal/loggerx"
+	"tailscale.com/util/syspolicy/setting"
+)
+
+// Change represents a change from the Old to the New value of type T.
+type Change[T any] struct {
+	New, Old T
+}
+
+// PolicyChangeCallback is a function called whenever a policy changes.
+type PolicyChangeCallback func(*PolicyChange)
+
+// PolicyChange describes a policy change.
+type PolicyChange struct {
+	snapshots Change[*setting.Snapshot]
+}
+
+// New returns the [setting.Snapshot] after the change.
+func (c PolicyChange) New() *setting.Snapshot {
+	return c.snapshots.New
+}
+
+// Old returns the [setting.Snapshot] before the change.
+func (c PolicyChange) Old() *setting.Snapshot {
+	return c.snapshots.Old
+}
+
+// HasChanged reports whether a policy setting with the specified [setting.Key], has changed.
+func (c PolicyChange) HasChanged(key setting.Key) bool {
+	new, newErr := c.snapshots.New.GetErr(key)
+	old, oldErr := c.snapshots.Old.GetErr(key)
+	if newErr != nil && oldErr != nil {
+		return false
+	}
+	if newErr != nil || oldErr != nil {
+		return true
+	}
+	switch newVal := new.(type) {
+	case bool, uint64, string, setting.Visibility, setting.PreferenceOption, time.Duration:
+		return newVal != old
+	case []string:
+		oldVal, ok := old.([]string)
+		return !ok || !slices.Equal(newVal, oldVal)
+	default:
+		loggerx.Errorf("[unexpected] %q has an unsupported value type: %T", key, newVal)
+		return !reflect.DeepEqual(new, old)
+	}
+}
+
+// policyChangeCallbacks are the callbacks to invoke when the effective policy changes.
+// It is safe for concurrent use.
+type policyChangeCallbacks struct {
+	mu  sync.Mutex
+	cbs set.HandleSet[PolicyChangeCallback]
+}
+
+// Register adds the specified callback to be invoked whenever the policy changes.
+func (c *policyChangeCallbacks) Register(callback PolicyChangeCallback) (unregister func()) {
+	c.mu.Lock()
+	handle := c.cbs.Add(callback)
+	c.mu.Unlock()
+	return func() {
+		c.mu.Lock()
+		delete(c.cbs, handle)
+		c.mu.Unlock()
+	}
+}
+
+// Invoke calls the registered callback functions with the specified policy change info.
+func (c *policyChangeCallbacks) Invoke(snapshots Change[*setting.Snapshot]) {
+	var wg sync.WaitGroup
+	defer wg.Wait()
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	wg.Add(len(c.cbs))
+	change := &PolicyChange{snapshots: snapshots}
+	for _, cb := range c.cbs {
+		go func() {
+			defer wg.Done()
+			cb(change)
+		}()
+	}
+}
+
+// Close awaits the completion of active callbacks and prevents any further invocations.
+func (c *policyChangeCallbacks) Close() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.cbs != nil {
+		clear(c.cbs)
+		c.cbs = nil
+	}
+}

vendor/tailscale.com/util/syspolicy/rsop/resultant_policy.go

@@ -0,0 +1,456 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package rsop
+
+import (
+	"errors"
+	"fmt"
+	"slices"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"tailscale.com/util/syspolicy/internal"
+	"tailscale.com/util/syspolicy/internal/loggerx"
+	"tailscale.com/util/syspolicy/setting"
+
+	"tailscale.com/util/syspolicy/source"
+)
+
+// ErrPolicyClosed is returned by [Policy.Reload], [Policy.addSource],
+// [Policy.removeSource] and [Policy.replaceSource] if the policy has been closed.
+var ErrPolicyClosed = errors.New("effective policy closed")
+
+// The minimum and maximum wait times after detecting a policy change
+// before reloading the policy. This only affects policy reloads triggered
+// by a change in the underlying [source.Store] and does not impact
+// synchronous, caller-initiated reloads, such as when [Policy.Reload] is called.
+//
+// Policy changes occurring within [policyReloadMinDelay] of each other
+// will be batched together, resulting in a single policy reload
+// no later than [policyReloadMaxDelay] after the first detected change.
+// In other words, the effective policy will be reloaded no more often than once
+// every 5 seconds, but at most 15 seconds after an underlying [source.Store]
+// has issued a policy change callback.
+//
+// See [Policy.watchReload].
+var (
+	policyReloadMinDelay = 5 * time.Second
+	policyReloadMaxDelay = 15 * time.Second
+)
+
+// Policy provides access to the current effective [setting.Snapshot] for a given
+// scope and allows to reload it from the underlying [source.Store] list. It also allows to
+// subscribe and receive a callback whenever the effective [setting.Snapshot] is changed.
+//
+// It is safe for concurrent use.
+type Policy struct {
+	scope setting.PolicyScope
+
+	reloadCh chan reloadRequest // 1-buffered; written to when a policy reload is required
+	closeCh  chan struct{}      // closed to signal that the Policy is being closed
+	doneCh   chan struct{}      // closed by [Policy.closeInternal]
+
+	// effective is the most recent version of the [setting.Snapshot]
+	// containing policy settings merged from all applicable sources.
+	effective atomic.Pointer[setting.Snapshot]
+
+	changeCallbacks policyChangeCallbacks
+
+	mu             sync.Mutex
+	watcherStarted bool // whether [Policy.watchReload] was started
+	sources        source.ReadableSources
+	closing        bool // whether [Policy.Close] was called (even if we're still closing)
+}
+
+// newPolicy returns a new [Policy] for the specified [setting.PolicyScope]
+// that tracks changes and merges policy settings read from the specified sources.
+func newPolicy(scope setting.PolicyScope, sources ...*source.Source) (_ *Policy, err error) {
+	readableSources := make(source.ReadableSources, 0, len(sources))
+	defer func() {
+		if err != nil {
+			readableSources.Close()
+		}
+	}()
+	for _, s := range sources {
+		reader, err := s.Reader()
+		if err != nil {
+			return nil, fmt.Errorf("failed to get a store reader: %w", err)
+		}
+		session, err := reader.OpenSession()
+		if err != nil {
+			return nil, fmt.Errorf("failed to open a reading session: %w", err)
+		}
+		readableSources = append(readableSources, source.ReadableSource{Source: s, ReadingSession: session})
+	}
+
+	// Sort policy sources by their precedence from lower to higher.
+	// For example, {UserPolicy},{ProfilePolicy},{DevicePolicy}.
+	readableSources.StableSort()
+
+	p := &Policy{
+		scope:    scope,
+		sources:  readableSources,
+		reloadCh: make(chan reloadRequest, 1),
+		closeCh:  make(chan struct{}),
+		doneCh:   make(chan struct{}),
+	}
+	if _, err := p.reloadNow(false); err != nil {
+		p.Close()
+		return nil, err
+	}
+	p.startWatchReloadIfNeeded()
+	return p, nil
+}
+
+// IsValid reports whether p is in a valid state and has not been closed.
+//
+// Since p's state can be changed by other goroutines at any time, this should
+// only be used as an optimization.
+func (p *Policy) IsValid() bool {
+	select {
+	case <-p.closeCh:
+		return false
+	default:
+		return true
+	}
+}
+
+// Scope returns the [setting.PolicyScope] that this policy applies to.
+func (p *Policy) Scope() setting.PolicyScope {
+	return p.scope
+}
+
+// Get returns the effective [setting.Snapshot].
+func (p *Policy) Get() *setting.Snapshot {
+	return p.effective.Load()
+}
+
+// RegisterChangeCallback adds a function to be called whenever the effective
+// policy changes. The returned function can be used to unregister the callback.
+func (p *Policy) RegisterChangeCallback(callback PolicyChangeCallback) (unregister func()) {
+	return p.changeCallbacks.Register(callback)
+}
+
+// Reload synchronously re-reads policy settings from the underlying list of policy sources,
+// constructing a new merged [setting.Snapshot] even if the policy remains unchanged.
+// In most scenarios, there's no need to re-read the policy manually.
+// Instead, it is recommended to register a policy change callback, or to use
+// the most recent [setting.Snapshot] returned by the [Policy.Get] method.
+//
+// It must not be called with p.mu held.
+func (p *Policy) Reload() (*setting.Snapshot, error) {
+	return p.reload(true)
+}
+
+// reload is like Reload, but allows to specify whether to re-read policy settings
+// from unchanged policy sources.
+//
+// It must not be called with p.mu held.
+func (p *Policy) reload(force bool) (*setting.Snapshot, error) {
+	if !p.startWatchReloadIfNeeded() {
+		return p.Get(), nil
+	}
+
+	respCh := make(chan reloadResponse, 1)
+	select {
+	case p.reloadCh <- reloadRequest{force: force, respCh: respCh}:
+		// continue
+	case <-p.closeCh:
+		return nil, ErrPolicyClosed
+	}
+	select {
+	case resp := <-respCh:
+		return resp.policy, resp.err
+	case <-p.closeCh:
+		return nil, ErrPolicyClosed
+	}
+}
+
+// reloadAsync requests an asynchronous background policy reload.
+// The policy will be reloaded no later than in [policyReloadMaxDelay].
+//
+// It must not be called with p.mu held.
+func (p *Policy) reloadAsync() {
+	if !p.startWatchReloadIfNeeded() {
+		return
+	}
+	select {
+	case p.reloadCh <- reloadRequest{}:
+		// Sent.
+	default:
+		// A reload request is already en route.
+	}
+}
+
+// reloadNow loads and merges policies from all sources, updating the effective policy.
+// If the force parameter is true, it forcibly reloads policies
+// from the underlying policy store, even if no policy changes were detected.
+//
+// Except for the initial policy reload during the [Policy] creation,
+// this method should only be called from the [Policy.watchReload] goroutine.
+func (p *Policy) reloadNow(force bool) (*setting.Snapshot, error) {
+	new, err := p.readAndMerge(force)
+	if err != nil {
+		return nil, err
+	}
+	old := p.effective.Swap(new)
+	// A nil old value indicates the initial policy load rather than a policy change.
+	// Additionally, we should not invoke the policy change callbacks unless the
+	// policy items have actually changed.
+	if old != nil && !old.EqualItems(new) {
+		snapshots := Change[*setting.Snapshot]{New: new, Old: old}
+		p.changeCallbacks.Invoke(snapshots)
+	}
+	return new, nil
+}
+
+// Done returns a channel that is closed when the [Policy] is closed.
+func (p *Policy) Done() <-chan struct{} {
+	return p.doneCh
+}
+
+// readAndMerge reads and merges policy settings from all applicable sources,
+// returning a [setting.Snapshot] with the merged result.
+// If the force parameter is true, it re-reads policy settings from each source
+// even if no policy change was observed, and returns an error if the read
+// operation fails.
+func (p *Policy) readAndMerge(force bool) (*setting.Snapshot, error) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	// Start with an empty policy in the target scope.
+	effective := setting.NewSnapshot(nil, setting.SummaryWith(p.scope))
+	// Then merge policy settings from all sources.
+	// Policy sources with the highest precedence (e.g., the device policy) are merged last,
+	// overriding any conflicting policy settings with lower precedence.
+	for _, s := range p.sources {
+		var policy *setting.Snapshot
+		if force {
+			var err error
+			if policy, err = s.ReadSettings(); err != nil {
+				return nil, err
+			}
+		} else {
+			policy = s.GetSettings()
+		}
+		effective = setting.MergeSnapshots(effective, policy)
+	}
+	return effective, nil
+}
+
+// addSource adds the specified source to the list of sources used by p,
+// and triggers a synchronous policy refresh. It returns an error
+// if the source is not a valid source for this effective policy,
+// or if the effective policy is being closed,
+// or if policy refresh fails with an error.
+func (p *Policy) addSource(source *source.Source) error {
+	return p.applySourcesChange(source, nil)
+}
+
+// removeSource removes the specified source from the list of sources used by p,
+// and triggers a synchronous policy refresh. It returns an error if the
+// effective policy is being closed, or if policy refresh fails with an error.
+func (p *Policy) removeSource(source *source.Source) error {
+	return p.applySourcesChange(nil, source)
+}
+
+// replaceSource replaces the old source with the new source atomically,
+// and triggers a synchronous policy refresh. It returns an error
+// if the source is not a valid source for this effective policy,
+// or if the effective policy is being closed,
+// or if policy refresh fails with an error.
+func (p *Policy) replaceSource(old, new *source.Source) error {
+	return p.applySourcesChange(new, old)
+}
+
+func (p *Policy) applySourcesChange(toAdd, toRemove *source.Source) error {
+	if toAdd == toRemove {
+		return nil
+	}
+	if toAdd != nil && !toAdd.Scope().Contains(p.scope) {
+		return errors.New("scope mismatch")
+	}
+
+	changed, err := func() (changed bool, err error) {
+		p.mu.Lock()
+		defer p.mu.Unlock()
+		if toAdd != nil && !p.sources.Contains(toAdd) {
+			reader, err := toAdd.Reader()
+			if err != nil {
+				return false, fmt.Errorf("failed to get a store reader: %w", err)
+			}
+			session, err := reader.OpenSession()
+			if err != nil {
+				return false, fmt.Errorf("failed to open a reading session: %w", err)
+			}
+
+			addAt := p.sources.InsertionIndexOf(toAdd)
+			toAdd := source.ReadableSource{
+				Source:         toAdd,
+				ReadingSession: session,
+			}
+			p.sources = slices.Insert(p.sources, addAt, toAdd)
+			go p.watchPolicyChanges(toAdd)
+			changed = true
+		}
+		if toRemove != nil {
+			if deleteAt := p.sources.IndexOf(toRemove); deleteAt != -1 {
+				p.sources.DeleteAt(deleteAt)
+				changed = true
+			}
+		}
+		return changed, nil
+	}()
+	if changed {
+		_, err = p.reload(false)
+	}
+	return err // may be nil or non-nil
+}
+
+func (p *Policy) watchPolicyChanges(s source.ReadableSource) {
+	for {
+		select {
+		case _, ok := <-s.ReadingSession.PolicyChanged():
+			if !ok {
+				p.mu.Lock()
+				abruptlyClosed := slices.Contains(p.sources, s)
+				p.mu.Unlock()
+				if abruptlyClosed {
+					// The underlying [source.Source] was closed abruptly without
+					// being properly removed or replaced by another policy source.
+					// We can't keep this [Policy] up to date, so we should close it.
+					p.Close()
+				}
+				return
+			}
+			// The PolicyChanged channel was signaled.
+			// Request an asynchronous policy reload.
+			p.reloadAsync()
+		case <-p.closeCh:
+			// The [Policy] is being closed.
+			return
+		}
+	}
+}
+
+// startWatchReloadIfNeeded starts [Policy.watchReload] in a new goroutine
+// if the list of policy sources is not empty, it hasn't been started yet,
+// and the [Policy] is not being closed.
+// It reports whether [Policy.watchReload] has ever been started.
+//
+// It must not be called with p.mu held.
+func (p *Policy) startWatchReloadIfNeeded() bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if len(p.sources) != 0 && !p.watcherStarted && !p.closing {
+		go p.watchReload()
+		for i := range p.sources {
+			go p.watchPolicyChanges(p.sources[i])
+		}
+		p.watcherStarted = true
+	}
+	return p.watcherStarted
+}
+
+// reloadRequest describes a policy reload request.
+type reloadRequest struct {
+	// force policy reload regardless of whether a policy change was detected.
+	force bool
+	// respCh is an optional channel. If non-nil, it makes the reload request
+	// synchronous and receives the result.
+	respCh chan<- reloadResponse
+}
+
+// reloadResponse is a result of a synchronous policy reload.
+type reloadResponse struct {
+	policy *setting.Snapshot
+	err    error
+}
+
+// watchReload processes incoming synchronous and asynchronous policy reload requests.
+//
+// Synchronous requests (with a non-nil respCh) are served immediately.
+//
+// Asynchronous requests are debounced and throttled: they are executed at least
+// [policyReloadMinDelay] after the last request, but no later than [policyReloadMaxDelay]
+// after the first request in a batch.
+func (p *Policy) watchReload() {
+	defer p.closeInternal()
+
+	force := false // whether a forced refresh was requested
+	var delayCh, timeoutCh <-chan time.Time
+	reload := func(respCh chan<- reloadResponse) {
+		delayCh, timeoutCh = nil, nil
+		policy, err := p.reloadNow(force)
+		if err != nil {
+			loggerx.Errorf("%v policy reload failed: %v\n", p.scope, err)
+		}
+		if respCh != nil {
+			respCh <- reloadResponse{policy: policy, err: err}
+		}
+		force = false
+	}
+
+loop:
+	for {
+		select {
+		case req := <-p.reloadCh:
+			if req.force {
+				force = true
+			}
+			if req.respCh != nil {
+				reload(req.respCh)
+				continue
+			}
+			if delayCh == nil {
+				timeoutCh = time.After(policyReloadMinDelay)
+			}
+			delayCh = time.After(policyReloadMaxDelay)
+		case <-delayCh:
+			reload(nil)
+		case <-timeoutCh:
+			reload(nil)
+		case <-p.closeCh:
+			break loop
+		}
+	}
+}
+
+func (p *Policy) closeInternal() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.sources.Close()
+	p.changeCallbacks.Close()
+	close(p.doneCh)
+	deletePolicy(p)
+}
+
+// Close initiates the closing of the policy.
+// The [Policy.Done] channel is closed to signal that the operation has been completed.
+func (p *Policy) Close() {
+	p.mu.Lock()
+	alreadyClosing := p.closing
+	watcherStarted := p.watcherStarted
+	p.closing = true
+	p.mu.Unlock()
+
+	if alreadyClosing {
+		return
+	}
+
+	close(p.closeCh)
+	if !watcherStarted {
+		// Normally, closing p.closeCh signals [Policy.watchReload] to exit,
+		// and [Policy.closeInternal] performs the actual closing when
+		// [Policy.watchReload] returns. However, if the watcher was never
+		// started, we need to call [Policy.closeInternal] manually.
+		go p.closeInternal()
+	}
+}
+
+func setForTest[T any](tb internal.TB, target *T, newValue T) {
+	oldValue := *target
+	tb.Cleanup(func() { *target = oldValue })
+	*target = newValue
+}

vendor/tailscale.com/util/syspolicy/rsop/rsop.go

@@ -0,0 +1,174 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+// Package rsop facilitates [source.Store] registration via [RegisterStore]
+// and provides access to the effective policy merged from all registered sources
+// via [PolicyFor].
+package rsop
+
+import (
+	"errors"
+	"fmt"
+	"slices"
+	"sync"
+
+	"tailscale.com/syncs"
+	"tailscale.com/util/slicesx"
+	"tailscale.com/util/syspolicy/internal"
+	"tailscale.com/util/syspolicy/setting"
+	"tailscale.com/util/syspolicy/source"
+)
+
+var (
+	policyMu          sync.Mutex       // protects [policySources] and [effectivePolicies]
+	policySources     []*source.Source // all registered policy sources
+	effectivePolicies []*Policy        // all active (non-closed) effective policies returned by [PolicyFor]
+
+	// effectivePolicyLRU is an LRU cache of [Policy] by [setting.Scope].
+	// Although there could be multiple [setting.PolicyScope] instances with the same [setting.Scope],
+	// such as two user scopes for different users, there is only one [setting.DeviceScope], only one
+	// [setting.CurrentProfileScope], and in most cases, only one active user scope.
+	// Therefore, cache misses that require falling back to [effectivePolicies] are extremely rare.
+	// It's a fixed-size array of atomic values and can be accessed without [policyMu] held.
+	effectivePolicyLRU [setting.NumScopes]syncs.AtomicValue[*Policy]
+)
+
+// PolicyFor returns the [Policy] for the specified scope,
+// creating it from the registered [source.Store]s if it doesn't already exist.
+func PolicyFor(scope setting.PolicyScope) (*Policy, error) {
+	if err := internal.Init.Do(); err != nil {
+		return nil, err
+	}
+	policy := effectivePolicyLRU[scope.Kind()].Load()
+	if policy != nil && policy.Scope() == scope && policy.IsValid() {
+		return policy, nil
+	}
+	return policyForSlow(scope)
+}
+
+func policyForSlow(scope setting.PolicyScope) (policy *Policy, err error) {
+	defer func() {
+		// Always update the LRU cache on exit if we found (or created)
+		// a policy for the specified scope.
+		if policy != nil {
+			effectivePolicyLRU[scope.Kind()].Store(policy)
+		}
+	}()
+
+	policyMu.Lock()
+	defer policyMu.Unlock()
+	if policy, ok := findPolicyByScopeLocked(scope); ok {
+		return policy, nil
+	}
+
+	// If there is no existing effective policy for the specified scope,
+	// we need to create one using the policy sources registered for that scope.
+	sources := slicesx.Filter(nil, policySources, func(source *source.Source) bool {
+		return source.Scope().Contains(scope)
+	})
+	policy, err = newPolicy(scope, sources...)
+	if err != nil {
+		return nil, err
+	}
+	effectivePolicies = append(effectivePolicies, policy)
+	return policy, nil
+}
+
+// findPolicyByScopeLocked returns a policy with the specified scope and true if
+// one exists in the [effectivePolicies] list, otherwise it returns nil, false.
+// [policyMu] must be held.
+func findPolicyByScopeLocked(target setting.PolicyScope) (policy *Policy, ok bool) {
+	for _, policy := range effectivePolicies {
+		if policy.Scope() == target && policy.IsValid() {
+			return policy, true
+		}
+	}
+	return nil, false
+}
+
+// deletePolicy deletes the specified effective policy from [effectivePolicies]
+// and [effectivePolicyLRU].
+func deletePolicy(policy *Policy) {
+	policyMu.Lock()
+	defer policyMu.Unlock()
+	if i := slices.Index(effectivePolicies, policy); i != -1 {
+		effectivePolicies = slices.Delete(effectivePolicies, i, i+1)
+	}
+	effectivePolicyLRU[policy.Scope().Kind()].CompareAndSwap(policy, nil)
+}
+
+// registerSource registers the specified [source.Source] to be used by the package.
+// It updates existing [Policy]s returned by [PolicyFor] to use this source if
+// they are within the source's [setting.PolicyScope].
+func registerSource(source *source.Source) error {
+	policyMu.Lock()
+	defer policyMu.Unlock()
+	if slices.Contains(policySources, source) {
+		// already registered
+		return nil
+	}
+	policySources = append(policySources, source)
+	return forEachEffectivePolicyLocked(func(policy *Policy) error {
+		if !source.Scope().Contains(policy.Scope()) {
+			// Policy settings in the specified source do not apply
+			// to the scope of this effective policy.
+			// For example, a user policy source is being registered
+			// while the effective policy is for the device (or another user).
+			return nil
+		}
+		return policy.addSource(source)
+	})
+}
+
+// replaceSource is like [unregisterSource](old) followed by [registerSource](new),
+// but performed atomically: the effective policy will contain settings
+// either from the old source or the new source, never both and never neither.
+func replaceSource(old, new *source.Source) error {
+	policyMu.Lock()
+	defer policyMu.Unlock()
+	oldIndex := slices.Index(policySources, old)
+	if oldIndex == -1 {
+		return fmt.Errorf("the source is not registered: %v", old)
+	}
+	policySources[oldIndex] = new
+	return forEachEffectivePolicyLocked(func(policy *Policy) error {
+		if !old.Scope().Contains(policy.Scope()) || !new.Scope().Contains(policy.Scope()) {
+			return nil
+		}
+		return policy.replaceSource(old, new)
+	})
+}
+
+// unregisterSource unregisters the specified [source.Source],
+// so that it won't be used by any new or existing [Policy].
+func unregisterSource(source *source.Source) error {
+	policyMu.Lock()
+	defer policyMu.Unlock()
+	index := slices.Index(policySources, source)
+	if index == -1 {
+		return nil
+	}
+	policySources = slices.Delete(policySources, index, index+1)
+	return forEachEffectivePolicyLocked(func(policy *Policy) error {
+		if !source.Scope().Contains(policy.Scope()) {
+			return nil
+		}
+		return policy.removeSource(source)
+	})
+}
+
+// forEachEffectivePolicyLocked calls fn for every non-closed [Policy] in [effectivePolicies].
+// It accumulates the returned errors and returns an error that wraps all errors returned by fn.
+// The [policyMu] mutex must be held while this function is executed.
+func forEachEffectivePolicyLocked(fn func(p *Policy) error) error {
+	var errs []error
+	for _, policy := range effectivePolicies {
+		if policy.IsValid() {
+			err := fn(policy)
+			if err != nil && !errors.Is(err, ErrPolicyClosed) {
+				errs = append(errs, err)
+			}
+		}
+	}
+	return errors.Join(errs...)
+}

vendor/tailscale.com/util/syspolicy/rsop/store_registration.go

@@ -0,0 +1,98 @@
+// Copyright (c) Tailscale Inc & AUTHORS
+// SPDX-License-Identifier: BSD-3-Clause
+
+package rsop
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"tailscale.com/util/syspolicy/internal"
+	"tailscale.com/util/syspolicy/setting"
+	"tailscale.com/util/syspolicy/source"
+)
+
+// ErrAlreadyConsumed is the error returned when [StoreRegistration.ReplaceStore]
+// or [StoreRegistration.Unregister] is called more than once.
+var ErrAlreadyConsumed = errors.New("the store registration is no longer valid")
+
+// StoreRegistration is a [source.Store] registered for use in the specified scope.
+// It can be used to unregister the store, or replace it with another one.
+type StoreRegistration struct {
+	source   *source.Source
+	m        sync.Mutex  // protects the [StoreRegistration.consumeSlow] path
+	consumed atomic.Bool // can be read without holding m, but must be written with m held
+}
+
+// RegisterStore registers a new policy [source.Store] with the specified name and [setting.PolicyScope].
+func RegisterStore(name string, scope setting.PolicyScope, store source.Store) (*StoreRegistration, error) {
+	return newStoreRegistration(name, scope, store)
+}
+
+// RegisterStoreForTest is like [RegisterStore], but unregisters the store when
+// tb and all its subtests complete.
+func RegisterStoreForTest(tb internal.TB, name string, scope setting.PolicyScope, store source.Store) (*StoreRegistration, error) {
+	setForTest(tb, &policyReloadMinDelay, 10*time.Millisecond)
+	setForTest(tb, &policyReloadMaxDelay, 500*time.Millisecond)
+
+	reg, err := RegisterStore(name, scope, store)
+	if err == nil {
+		tb.Cleanup(func() {
+			if err := reg.Unregister(); err != nil && !errors.Is(err, ErrAlreadyConsumed) {
+				tb.Fatalf("Unregister failed: %v", err)
+			}
+		})
+	}
+	return reg, err // may be nil or non-nil
+}
+
+func newStoreRegistration(name string, scope setting.PolicyScope, store source.Store) (*StoreRegistration, error) {
+	source := source.NewSource(name, scope, store)
+	if err := registerSource(source); err != nil {
+		return nil, err
+	}
+	return &StoreRegistration{source: source}, nil
+}
+
+// ReplaceStore replaces the registered store with the new one,
+// returning a new [StoreRegistration] or an error.
+func (r *StoreRegistration) ReplaceStore(new source.Store) (*StoreRegistration, error) {
+	var res *StoreRegistration
+	err := r.consume(func() error {
+		newSource := source.NewSource(r.source.Name(), r.source.Scope(), new)
+		if err := replaceSource(r.source, newSource); err != nil {
+			return err
+		}
+		res = &StoreRegistration{source: newSource}
+		return nil
+	})
+	return res, err
+}
+
+// Unregister reverts the registration.
+func (r *StoreRegistration) Unregister() error {
+	return r.consume(func() error { return unregisterSource(r.source) })
+}
+
+// consume invokes fn, consuming r if no error is returned.
+// It returns [ErrAlreadyConsumed] on subsequent calls after the first successful call.
+func (r *StoreRegistration) consume(fn func() error) (err error) {
+	if r.consumed.Load() {
+		return ErrAlreadyConsumed
+	}
+	return r.consumeSlow(fn)
+}
+
+func (r *StoreRegistration) consumeSlow(fn func() error) (err error) {
+	r.m.Lock()
+	defer r.m.Unlock()
+	if r.consumed.Load() {
+		return ErrAlreadyConsumed
+	}
+	if err = fn(); err == nil {
+		r.consumed.Store(true)
+	}
+	return err // may be nil or non-nil
+}