state, db: validate and store the current policy before installing it

Reject empty input. Validate without changing live state, then store and
install under one lock; prune old rows only with a successful write.
This commit is contained in:
Kristoffer Dalby committed 2026-11-02 17:46:50 +00:00
1 parent 89d56419e2
commit d326143556
12 files changed
+367 -93

No files matched your search

+2 -1
View File
@@ -46,7 +46,8 @@ func mapError(msg string, err error) error {
errors.Is(err, db.ErrUserStillHasNodes),
errors.Is(err, db.ErrCannotChangeOIDCUser),
errors.Is(err, db.ErrPreAuthKeyNotTaggedOrOwned),
errors.Is(err, db.ErrSingleUseAuthKeyHasBeenUsed):
errors.Is(err, db.ErrSingleUseAuthKeyHasBeenUsed),
errors.Is(err, state.ErrInvalidPolicy):
return huma.Error400BadRequest(msg, err)
case errors.Is(err, state.ErrNodeKeyInUse),
+4 -45
View File
@@ -9,7 +9,6 @@ import (
"time"
"github.com/danielgtaylor/huma/v2"
policyv2 "github.com/juanfont/headscale/hscontrol/policy/v2"
"github.com/juanfont/headscale/hscontrol/types"
"github.com/juanfont/headscale/hscontrol/util"
)
@@ -116,37 +115,11 @@ func registerPolicy(api huma.API, b Backend) {
)
}
p := in.Body.Policy
// Reject policy that would fail when building a map response. SSH rule
// validation needs a node, so a server with no nodes can't catch every
// case here.
nodes := b.State.ListNodes()
_, err := b.State.SetPolicy([]byte(p))
if err != nil {
return nil, huma.Error400BadRequest("setting policy", err)
}
if nodes.Len() > 0 {
_, err = b.State.SSHPolicy(nodes.At(0))
if err != nil {
return nil, huma.Error400BadRequest("verifying SSH rules", err)
}
}
updated, err := b.State.SetPolicyInDB(p)
if err != nil {
return nil, huma.Error500InternalServerError("setting policy", err)
}
// Reload even when content is unchanged: routes manually disabled before
// may now qualify for auto-approval, so they must be re-evaluated.
cs, err := b.State.ReloadPolicy()
updated, cs, err := b.State.UpdatePolicy([]byte(in.Body.Policy))
b.Change(cs...)
if err != nil {
return nil, huma.Error500InternalServerError("reloading policy", err)
return nil, mapError("setting policy", err)
}
out := &setPolicyOutput{}
@@ -165,23 +138,9 @@ func registerPolicy(api huma.API, b Backend) {
Tags: []string{"Policy"},
Security: bearerAuth,
}, func(ctx context.Context, in *checkPolicyInput) (*checkPolicyOutput, error) {
polB := []byte(in.Body.Policy)
users, err := b.State.ListAllUsers()
err := b.State.ValidatePolicy([]byte(in.Body.Policy))
if err != nil {
return nil, huma.Error500InternalServerError("loading users", err)
}
nodes := b.State.ListNodes()
pm, err := policyv2.NewPolicyManager(polB, users, nodes)
if err != nil {
return nil, huma.Error400BadRequest(err.Error(), err)
}
_, err = pm.SetPolicy(polB)
if err != nil {
return nil, huma.Error400BadRequest(err.Error(), err)
return nil, mapError(err.Error(), err)
}
return &checkPolicyOutput{}, nil
+2 -22
View File
@@ -130,31 +130,11 @@ func registerACL(api huma.API, b Backend) {
}
}
// Mirror the v1 setPolicy flow: validate, SSH-check, persist, reload.
nodes := b.State.ListNodes()
_, err = b.State.SetPolicy(in.RawBody)
if err != nil {
return nil, huma.Error400BadRequest("setting policy", err)
}
if nodes.Len() > 0 {
_, err = b.State.SSHPolicy(nodes.At(0))
if err != nil {
return nil, huma.Error400BadRequest("verifying SSH rules", err)
}
}
updated, err := b.State.SetPolicyInDB(string(in.RawBody))
if err != nil {
return nil, huma.Error500InternalServerError("setting policy", err)
}
cs, err := b.State.ReloadPolicy()
updated, cs, err := b.State.UpdatePolicy(in.RawBody)
b.Change(cs...)
if err != nil {
return nil, huma.Error500InternalServerError("reloading policy", err)
return nil, mapError("setting policy", err)
}
return streamPolicy([]byte(updated.Data), aclContentType(in.Accept)), nil
+4
View File
@@ -86,6 +86,10 @@ func mapError(msg string, err error) error {
errors.Is(err, state.ErrRequestedTagsInvalidOrNotPermitted):
return huma.Error400BadRequest(msg, err)
// A refused policy is answered with its own reason as the message.
case errors.Is(err, state.ErrInvalidPolicy):
return huma.Error400BadRequest(err.Error(), err)
default:
return huma.Error500InternalServerError(msg, err)
}
+50
View File
@@ -97,6 +97,47 @@ func TestAPIV1PolicySet(t *testing.T) {
assert.NotEmpty(t, got.UpdatedAt)
})
t.Run("persist failure keeps live policy", func(t *testing.T) {
h := newAPIV1Harness(t)
body, err := json.Marshal(map[string]string{"policy": `{"acls":[]}`})
require.NoError(t, err)
require.Equal(t, http.StatusOK,
h.callHuma(http.MethodPut, "/api/v1/policy", body).status)
before, _ := h.app.state.Filter()
require.NoError(t, h.app.state.DB().DB.
Exec("ALTER TABLE policies RENAME TO policies_away").Error)
body, err = json.Marshal(map[string]string{"policy": validPolicy})
require.NoError(t, err)
res := h.callHuma(http.MethodPut, "/api/v1/policy", body)
assertStatus(t, res, http.StatusInternalServerError)
after, _ := h.app.state.Filter()
assert.Equal(t, before, after, "an unstored policy must not be enforced")
})
t.Run("empty rejected", func(t *testing.T) {
h := newAPIV1Harness(t)
body, err := json.Marshal(map[string]string{"policy": validPolicy})
require.NoError(t, err)
require.Equal(t, http.StatusOK,
h.callHuma(http.MethodPut, "/api/v1/policy", body).status)
for _, body := range []string{`{"policy":""}`, `{}`} {
res := h.callHuma(http.MethodPut, "/api/v1/policy", []byte(body))
assertStatus(t, res, http.StatusBadRequest)
}
stored, err := h.app.state.GetPolicy()
require.NoError(t, err)
assert.JSONEq(t, validPolicy, stored.Data)
})
t.Run("not-db mode rejected", func(t *testing.T) {
// Policy updates are only valid in DB mode; file mode rejects with 400.
// createTestApp is DB mode, so build a file-mode app here.
@@ -132,6 +173,15 @@ func TestAPIV1PolicyCheck(t *testing.T) {
assertStatus(t, res, http.StatusBadRequest)
})
t.Run("empty rejected", func(t *testing.T) {
h := newAPIV1Harness(t)
res := h.callHuma(
http.MethodPost, "/api/v1/policy/check", []byte(`{"policy":""}`),
)
assertStatus(t, res, http.StatusBadRequest)
})
t.Run("valid returns empty object", func(t *testing.T) {
h := newAPIV1Harness(t)
+33 -2
View File
@@ -191,9 +191,21 @@ func TestAPIv2ACLInvalidPolicyAtomicity(t *testing.T) {
goodETag := good.Header().Get("Etag")
// Malformed body -> 400, and the stored policy is unchanged (atomicity).
bad := setACL(t, api, `{ this is not valid`, "application/json")
const malformed = `{ this is not valid`
bad := setACL(t, api, malformed, "application/json")
assert.Equal(t, http.StatusBadRequest, bad.Code)
assert.Contains(t, bad.Body.String(), `"message"`)
var answer struct {
Message string `json:"message"`
}
require.NoError(t, json.Unmarshal(bad.Body.Bytes(), &answer))
reason := app.state.ValidatePolicy([]byte(malformed))
require.Error(t, reason)
assert.Equal(t, reason.Error(), answer.Message,
"the answer gives the parser's reason as it is")
assert.NotContains(t, answer.Message, "invalid policy")
assert.JSONEq(t, allowAllPolicy, storedPolicy(t, app))
// Wire-level: GET still serves the good policy + its ETag (no drift).
@@ -202,6 +214,25 @@ func TestAPIv2ACLInvalidPolicyAtomicity(t *testing.T) {
assert.JSONEq(t, allowAllPolicy, storedPolicy(t, app))
}
func TestAPIv2ACLPersistFailureKeepsLivePolicy(t *testing.T) {
app := createTestApp(t)
api := registerAPIV2(t, app)
require.Equal(t, http.StatusOK,
setACL(t, api, `{"acls":[]}`, "application/json").Code)
before, _ := app.state.Filter()
require.NoError(t,
app.state.DB().DB.Exec("ALTER TABLE policies RENAME TO policies_away").Error)
assert.Equal(t, http.StatusInternalServerError,
setACL(t, api, allowAllPolicy, "application/json").Code)
after, _ := app.state.Filter()
assert.Equal(t, before, after, "an unstored policy must not be enforced")
}
// TestAPIv2ACLValidate: validate stores nothing and answers a refused policy
// with 200, its reason as message and one data entry per failing test; only
// a body with no policy at all is a 400.
-3
View File
@@ -181,8 +181,6 @@ var apiv2CompatSkipReasons = map[string]string{
"policy-api/P7/0/body": hujsonFormatted,
"policy-api/P7/1/body": hujsonFormatted,
"policy-api/P2/0/v1-verdict": v1CheckAcceptsEmpty,
}
// apiv2CompatNotCompared lists, by key prefix, checks left out: SaaS's
@@ -220,7 +218,6 @@ const (
taggedAllClientAnyTag = "TAGGED_ALL_CLIENT_ANY_TAG: SaaS lets a tagged all-scope client's token ask only for the client's tags; headscale lets it ask for any tag"
retagRemovesUnownedTag = "RETAG_REMOVES_UNOWNED_TAG: SaaS refuses a token's retag of a device holding a tag outside the token's grant with 400 cannot remove current tags; headscale replaces the tags"
sshTestHostnames = "SSH_TEST_HOSTNAMES: headscale names the nodes behind a failing sshTests destination; SaaS names the destination as written"
v1CheckAcceptsEmpty = "V1_CHECK_ACCEPTS_EMPTY: v1 policy check accepts a zero-length policy; SaaS's validate refuses it"
)
// Differences no planned change removes yet.
+19 -10
View File
@@ -10,19 +10,28 @@ import (
"gorm.io/gorm/clause"
)
// SetPolicy sets the policy in the database.
// SetPolicy replaces the policy in the database. Only the newest row is
// ever read, so older ones are removed rather than kept as history.
func (hsdb *HSDatabase) SetPolicy(policy string) (*types.Policy, error) {
// Create a new policy.
p := types.Policy{
Data: policy,
}
return Write(hsdb.DB, func(tx *gorm.DB) (*types.Policy, error) {
p := types.Policy{
Data: policy,
}
err := hsdb.DB.Clauses(clause.Returning{}).Create(&p).Error
if err != nil {
return nil, err
}
err := tx.Clauses(clause.Returning{}).Create(&p).Error
if err != nil {
return nil, err
}
return &p, nil
// Unscoped: a soft delete would keep the data. Deleting by id
// never removes a newer row from a concurrent writer.
err = tx.Unscoped().Where("id < ?", p.ID).Delete(&types.Policy{}).Error
if err != nil {
return nil, err
}
return &p, nil
})
}
// GetPolicy returns the latest policy in the database.
+31
View File
@@ -0,0 +1,31 @@
package db
import (
"strconv"
"testing"
"github.com/juanfont/headscale/hscontrol/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestSetPolicyKeepsOnlyCurrent checks repeated writes replace the stored
// policy instead of piling up rows.
func TestSetPolicyKeepsOnlyCurrent(t *testing.T) {
db := dbForTest(t)
for i := range 5 {
_, err := db.SetPolicy(
`{"hosts": {"h": "100.64.0.` + strconv.Itoa(i) + `/32"}}`,
)
require.NoError(t, err)
}
var rows int64
require.NoError(t, db.DB.Unscoped().Model(&types.Policy{}).Count(&rows).Error)
assert.Equal(t, int64(1), rows)
p, err := db.GetPolicy()
require.NoError(t, err)
assert.JSONEq(t, `{"hosts": {"h": "100.64.0.4/32"}}`, p.Data)
}
+4 -1
View File
@@ -29,6 +29,9 @@ import (
// ErrInvalidTagOwner is returned when a tag owner is not an [Alias] type.
var ErrInvalidTagOwner = errors.New("tag owner is not an Alias")
// ErrPolicyEmpty is returned by [PolicyManager.SetPolicy] for a zero-length policy.
var ErrPolicyEmpty = errors.New(`policy is empty, use "{}" to allow all`)
type PolicyManager struct {
// RWMutex, not Mutex, so concurrent map generation does not serialise on
// reads. The per-node caches are xsync.Maps so a read can fill them without
@@ -658,7 +661,7 @@ func (pm *PolicyManager) sshRuleAllowsUser(rule SSH, srcNode types.NodeView, loc
func (pm *PolicyManager) SetPolicy(polB []byte) (bool, error) {
if len(polB) == 0 {
return false, nil
return false, ErrPolicyEmpty
}
pol, err := unmarshalPolicy(polB)
+172
View File
@@ -6,6 +6,8 @@ import (
"maps"
"net/http"
"net/netip"
"net/url"
"os"
"path/filepath"
"runtime"
"slices"
@@ -18,6 +20,7 @@ import (
"github.com/juanfont/headscale/hscontrol/db"
"github.com/juanfont/headscale/hscontrol/policy"
policyv2 "github.com/juanfont/headscale/hscontrol/policy/v2"
"github.com/juanfont/headscale/hscontrol/types"
"github.com/juanfont/headscale/hscontrol/types/change"
"github.com/juanfont/headscale/hscontrol/types/testcapture"
@@ -28,6 +31,7 @@ import (
"tailscale.com/tailcfg"
"tailscale.com/types/key"
"tailscale.com/types/views"
"zombiezen.com/go/postgrestest"
)
// persistTestSetup pre-creates a sqlite database on disk with a single
@@ -1310,6 +1314,174 @@ func TestReloadPolicyOverlappingWrites(t *testing.T) {
assert.Equal(t, want, got, "the policy in force must be the one stored last")
}
// TestUpdatePolicyStoreFailure refuses the write at the database and checks
// the live policy was not changed by it.
func TestUpdatePolicyStoreFailure(t *testing.T) {
_, s, _ := persistTestSetup(t)
t.Cleanup(func() { _ = s.Close() })
const (
open = `{"acls": [{"action": "accept", "src": ["*"], "dst": ["*:*"]}]}`
ssh = `{"acls": [{"action": "accept", "src": ["persist-user@"], "dst": ["persist-user@:22"]}]}`
)
_, _, err := s.UpdatePolicy([]byte(ssh))
require.NoError(t, err)
before, _ := s.Filter()
require.NoError(t,
s.db.DB.Exec("ALTER TABLE policies RENAME TO policies_away").Error)
_, _, err = s.UpdatePolicy([]byte(open))
require.Error(t, err)
require.NotErrorIs(t, err, ErrInvalidPolicy)
after, _ := s.Filter()
assert.Equal(t, before, after, "an unstored policy must not be installed")
}
// TestPostgresUpdatePolicyConcurrentWrites runs a second write while the
// first one's insert is not yet committed. Pruning cannot see that row, so
// unless writers are serialised up to the install, it outlives the newer one.
func TestPostgresUpdatePolicyConcurrentWrites(t *testing.T) {
srv, err := postgrestest.Start(t.Context())
if err != nil {
t.Skipf("start postgres: %s", err)
}
t.Cleanup(srv.Cleanup)
dsn, err := srv.CreateDatabase(t.Context())
require.NoError(t, err)
u, err := url.Parse(dsn)
require.NoError(t, err)
pass, _ := u.User.Password()
port, err := strconv.Atoi(u.Port())
require.NoError(t, err)
cfg := persistTestConfig("")
cfg.Database = types.DatabaseConfig{
Type: types.DatabasePostgres,
Postgres: types.PostgresConfig{
Host: u.Hostname(),
Port: port,
Name: strings.TrimLeft(u.Path, "/"),
User: u.User.Username(),
Pass: pass,
Ssl: "disable",
},
}
s, err := NewState(cfg)
require.NoError(t, err)
t.Cleanup(func() { _ = s.Close() })
const (
first = `{"acls": [{"action": "accept", "src": ["*"], "dst": ["*:*"]}]}`
second = `{"acls": [{"action": "accept", "src": ["*"], "dst": ["*:22"]}]}`
)
_, _, err = s.UpdatePolicy([]byte(second))
require.NoError(t, err)
want, _ := s.Filter()
var creates atomic.Int32
written := make(chan struct{})
require.NoError(t, s.db.DB.Callback().Create().After("gorm:create").
Register("write_while_storing", func(tx *gorm.DB) {
if tx.Statement.Table != "policies" || creates.Add(1) != 1 {
return
}
go func() {
defer close(written)
_, _, err := s.UpdatePolicy([]byte(second))
assert.NoError(t, err)
}()
// Give the second write time to finish, unless the first
// write blocks it.
select {
case <-written:
case <-time.After(200 * time.Millisecond):
}
}))
t.Cleanup(func() {
_ = s.db.DB.Callback().Create().Remove("write_while_storing")
})
_, _, err = s.UpdatePolicy([]byte(first))
require.NoError(t, err)
<-written
var rows int64
require.NoError(t,
s.db.DB.Unscoped().Model(&types.Policy{}).Count(&rows).Error)
assert.Equal(t, int64(1), rows, "only the current policy may be stored")
stored, err := s.GetPolicy()
require.NoError(t, err)
assert.JSONEq(t, second, stored.Data, "the write that ran last is stored")
got, _ := s.Filter()
assert.Equal(t, want, got, "the policy in force must be the one stored")
}
// TestReloadPolicyEmpty: with no policy configured a reload has nothing to
// install; an emptied policy file is refused. Both keep the live policy.
func TestReloadPolicyEmpty(t *testing.T) {
const pol = `{"acls": [{"action": "accept", "src": ["persist-user@"], "dst": ["persist-user@:22"]}]}`
t.Run("no policy configured", func(t *testing.T) {
_, s, _ := persistTestSetup(t)
t.Cleanup(func() { _ = s.Close() })
_, err := s.SetPolicy([]byte(pol))
require.NoError(t, err)
before, _ := s.Filter()
_, err = s.ReloadPolicy()
require.NoError(t, err)
after, _ := s.Filter()
assert.Equal(t, before, after)
})
t.Run("empty policy file", func(t *testing.T) {
_, s, _ := persistTestSetup(t)
t.Cleanup(func() { _ = s.Close() })
s.cfg.Policy = types.PolicyConfig{
Mode: types.PolicyModeFile,
Path: filepath.Join(t.TempDir(), "policy.hujson"),
}
require.NoError(t, os.WriteFile(s.cfg.Policy.Path, []byte(pol), 0o600))
_, err := s.ReloadPolicy()
require.NoError(t, err)
before, _ := s.Filter()
// os.ReadFile gives an empty, non-nil slice here.
require.NoError(t, os.WriteFile(s.cfg.Policy.Path, nil, 0o600))
_, err = s.ReloadPolicy()
require.ErrorIs(t, err, policyv2.ErrPolicyEmpty)
after, _ := s.Filter()
assert.Equal(t, before, after)
})
}
// TestNodeWriteChangeWhenPolicyRefreshFails fails both the NodeStore
// writer's SetNodes and the caller's. The write still reached the
// NodeStore, so the change returned with the error must resend the node to
+46 -9
View File
@@ -70,8 +70,8 @@ const (
// ErrUnsupportedPolicyMode is returned for invalid policy modes. Valid modes are "file" and "db".
var ErrUnsupportedPolicyMode = errors.New("unsupported policy mode")
// ErrInvalidPolicy is returned by [State.ValidatePolicy] for a policy it
// refuses.
// ErrInvalidPolicy is returned by [State.UpdatePolicy] and
// [State.ValidatePolicy] for a policy they refuse.
var ErrInvalidPolicy = errors.New("invalid policy")
// invalidPolicyError marks a refusal as [ErrInvalidPolicy] and keeps the
@@ -182,8 +182,8 @@ type State struct {
// polMan handles policy evaluation and management
polMan policy.PolicyManager
// policyMu makes loading the stored policy and installing it one step,
// so of two overlapping reloads the older document cannot land last.
// policyMu makes storing, loading and installing a policy one step, so
// policies land in the database in the order they are installed.
policyMu sync.Mutex
// authCache holds pending Web, OIDC and SSH authentication requests.
@@ -472,14 +472,23 @@ func (s *State) ReloadPolicy() ([]change.Change, error) {
s.policyMu.Lock()
defer s.policyMu.Unlock()
return s.reloadPolicyLocked()
}
// reloadPolicyLocked is [State.ReloadPolicy] for callers holding policyMu.
func (s *State) reloadPolicyLocked() ([]change.Change, error) {
pol, err := hsdb.PolicyBytes(s.db.DB, s.cfg)
if err != nil {
return nil, fmt.Errorf("loading policy: %w", err)
}
policyChanged, err := s.polMan.SetPolicy(pol)
if err != nil {
return nil, fmt.Errorf("setting policy: %w", err)
// nil means no policy is configured: nothing to install.
var policyChanged bool
if pol != nil {
policyChanged, err = s.polMan.SetPolicy(pol)
if err != nil {
return nil, fmt.Errorf("setting policy: %w", err)
}
}
// Clear SSH check auth times when policy changes to ensure stale
@@ -1466,7 +1475,8 @@ func (s *State) NodeCanHaveTag(node types.NodeView, tag string) bool {
return s.polMan.NodeCanHaveTag(node, tag)
}
// SetPolicy updates the policy configuration.
// SetPolicy installs pol without storing it. Policy writes use
// [State.UpdatePolicy].
func (s *State) SetPolicy(pol []byte) (bool, error) {
changed, err := s.polMan.SetPolicy(pol)
if err != nil {
@@ -1528,7 +1538,7 @@ func (s *State) SetPolicyInDB(data string) (*types.Policy, error) {
return s.db.SetPolicy(data)
}
// ValidatePolicy reports whether a policy write would accept pol. It
// ValidatePolicy reports whether [State.UpdatePolicy] would accept pol. It
// compiles pol against the current users and nodes on a throwaway policy
// manager; the live policy is not touched.
func (s *State) ValidatePolicy(pol []byte) error {
@@ -1561,6 +1571,33 @@ func (s *State) ValidatePolicy(pol []byte) error {
return nil
}
// UpdatePolicy validates, stores and installs pol. The live policy is only
// ever installed from what was stored, so a refused or unstored write leaves
// it untouched. The changes are returned alongside a reload error, as from
// [State.ReloadPolicy].
func (s *State) UpdatePolicy(pol []byte) (*types.Policy, []change.Change, error) {
err := s.ValidatePolicy(pol)
if err != nil {
return nil, nil, err
}
// Held from store to install: on PostgreSQL a concurrent writer could
// otherwise commit an older row after a newer one pruned the rest.
s.policyMu.Lock()
defer s.policyMu.Unlock()
stored, err := s.db.SetPolicy(string(pol))
if err != nil {
return nil, nil, fmt.Errorf("storing policy: %w", err)
}
// Reload even when content is unchanged: routes manually disabled before
// may now qualify for auto-approval, so they must be re-evaluated.
cs, err := s.reloadPolicyLocked()
return stored, cs, err
}
// GetNodePrimaryRoutes returns the primary routes for a node.
func (s *State) GetNodePrimaryRoutes(nodeID types.NodeID) []netip.Prefix {
return s.nodeStore.PrimaryRoutesForNode(nodeID)