mirror of
https://github.com/juanfont/headscale.git
synced 2026-10-10 08:40:07 +09:00
mapper: reconcile self on policy recomputes
Self renders from the same state as peers, so leaving it out of policy responses hid an exit node's approved routes until it reconnected. Fixes #3502
This commit is contained in:
committed by
Kristoffer Dalby
parent
ba413fca3d
commit
ea6b0bf0b7
@@ -129,9 +129,7 @@ func generateMapResponse(
|
||||
}
|
||||
|
||||
removed = nc.computePeerDiff(currentPeerIDs)
|
||||
// Include self node when this is a self-update (e.g., node's own tags changed)
|
||||
// so the node sees its updated self info along with new packet filters.
|
||||
resp, err = mapper.policyChangeResponse(nodeID, version, currentPeers, isSelfUpdate)
|
||||
resp, err = mapper.policyChangeResponse(nodeID, version, currentPeers)
|
||||
} else if isSelfUpdate {
|
||||
// Non-policy self-update: just send the self node info
|
||||
resp, err = mapper.selfMapResponse(nodeID, version)
|
||||
@@ -386,6 +384,9 @@ func (b *Batcher) AddNode(
|
||||
|
||||
nodeConn.workMu.Unlock()
|
||||
|
||||
// Still pendingInitial, so no broadcast can race this.
|
||||
newEntry.lastSelf.Store(initialMap.Node)
|
||||
|
||||
// Open the connection for broadcast sends now that the initial
|
||||
// map is the stream's first frame; send() requeued any changes
|
||||
// that arrived in the meantime.
|
||||
|
||||
@@ -2715,3 +2715,61 @@ func TestDNSConfigOnlyWithSelfRefresh(t *testing.T) {
|
||||
assert.Nil(t, sent[0].DNSConfig, "policy response must not carry DNSConfig")
|
||||
assert.NotNil(t, sent[1].DNSConfig, "self refresh must carry DNSConfig")
|
||||
}
|
||||
|
||||
// TestSelfSentOnlyWhenChanged pins that a policy recompute carries the
|
||||
// node's own self to each connection that does not hold it yet: self renders
|
||||
// from the same state as peers (issue #3502), and a Node forces a full client
|
||||
// netmap rebuild.
|
||||
func TestSelfSentOnlyWhenChanged(t *testing.T) {
|
||||
testData, cleanup := setupBatcherWithTestData(t, NewBatcherAndMapper, 1, 2, normalBufferSize)
|
||||
defer cleanup()
|
||||
|
||||
b := testData.Batcher.Batcher
|
||||
self := &testData.Nodes[0]
|
||||
|
||||
rename := func(name string) {
|
||||
_, _, err := testData.State.RenameNode(self.n.ID, name)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
require.NoError(t, b.AddNode(self.n.ID, self.ch, tailcfg.CapabilityVersion(100), nil))
|
||||
require.NotNil(t, expectReceive(t, self.ch, "initial map").Node)
|
||||
|
||||
nc, ok := b.nodes.Load(self.n.ID)
|
||||
require.True(t, ok)
|
||||
|
||||
policyFrame := func(chs ...chan *tailcfg.MapResponse) []*tailcfg.MapResponse {
|
||||
nc.workMu.Lock()
|
||||
defer nc.workMu.Unlock()
|
||||
|
||||
require.NoError(t, handleNodeChange(nc, b.mapper, change.PolicyChange()))
|
||||
|
||||
frames := make([]*tailcfg.MapResponse, 0, len(chs))
|
||||
for _, ch := range chs {
|
||||
frames = append(frames, expectReceive(t, ch, "policy frame"))
|
||||
}
|
||||
|
||||
return frames
|
||||
}
|
||||
|
||||
assert.Nil(t, policyFrame(self.ch)[0].Node, "unchanged self must be dropped")
|
||||
|
||||
rename("renamed")
|
||||
|
||||
frame := policyFrame(self.ch)[0]
|
||||
require.NotNil(t, frame.Node, "moved self must be sent")
|
||||
assert.Contains(t, frame.Node.Name, "renamed")
|
||||
|
||||
// A connection joining after self moved gets the new self in its
|
||||
// initial map; the next recompute must still reach the older one.
|
||||
rename("renamed-again")
|
||||
|
||||
ch2 := make(chan *tailcfg.MapResponse, normalBufferSize)
|
||||
require.NoError(t, b.AddNode(self.n.ID, ch2, tailcfg.CapabilityVersion(100), nil))
|
||||
require.NotNil(t, expectReceive(t, ch2, "second connection's initial map").Node)
|
||||
|
||||
frames := policyFrame(self.ch, ch2)
|
||||
require.NotNil(t, frames[0].Node, "first connection still holds the old self")
|
||||
assert.Contains(t, frames[0].Node.Name, "renamed-again")
|
||||
assert.Nil(t, frames[1].Node, "second connection already holds the new self")
|
||||
}
|
||||
|
||||
@@ -309,7 +309,8 @@ func (m *mapper) selfMapResponse(
|
||||
// - PeersChanged for remaining peers (their AllowedIPs may have changed due to policy)
|
||||
// - Updated PacketFilters
|
||||
// - Updated SSHPolicy (SSH rules may reference users/groups that changed)
|
||||
// - Optionally, the node's own self info (when includeSelf is true)
|
||||
// - The node's own self info, which renders from the same state as peers;
|
||||
// dropped per connection when unchanged, see [connectionEntry.withSelfDelta]
|
||||
//
|
||||
// DNSConfig is left out: it forces clients into a full netmap rebuild, and
|
||||
// the node's DNS config inputs arrive with its own [change.SelfUpdate], see
|
||||
@@ -317,24 +318,17 @@ func (m *mapper) selfMapResponse(
|
||||
//
|
||||
// This avoids the issue where an empty Peers slice is interpreted by Tailscale
|
||||
// clients as "no change" rather than "no peers".
|
||||
// When includeSelf is true, the node's self info is included so that a node
|
||||
// whose own attributes changed (e.g., tags via admin API) sees its updated
|
||||
// self info along with the new packet filters.
|
||||
func (m *mapper) policyChangeResponse(
|
||||
nodeID types.NodeID,
|
||||
capVer tailcfg.CapabilityVersion,
|
||||
currentPeers views.Slice[types.NodeView],
|
||||
includeSelf bool,
|
||||
) (*tailcfg.MapResponse, error) {
|
||||
builder := m.NewMapResponseBuilder(nodeID).
|
||||
WithDebugType(policyResponseDebug).
|
||||
WithCapabilityVersion(capVer).
|
||||
WithPacketFilters().
|
||||
WithSSHPolicy()
|
||||
|
||||
if includeSelf {
|
||||
builder = builder.WithSelfNode()
|
||||
}
|
||||
WithSSHPolicy().
|
||||
WithSelfNode()
|
||||
|
||||
// Send remaining peers in PeersChanged - their AllowedIPs may have
|
||||
// changed due to the policy update (e.g., different routes allowed).
|
||||
|
||||
@@ -50,6 +50,25 @@ type connectionEntry struct {
|
||||
// can never become the stream's first frame ahead of the initial
|
||||
// map. The zero value means the connection is ready.
|
||||
pendingInitial atomic.Bool
|
||||
|
||||
// lastSelf is the self node last delivered to this connection's
|
||||
// client, which keeps it until sent another. Every send to an
|
||||
// established connection must go through [multiChannelNodeConn.send]
|
||||
// to keep it current.
|
||||
lastSelf atomic.Pointer[tailcfg.Node]
|
||||
}
|
||||
|
||||
// withSelfDelta returns data without its Node when this client already
|
||||
// holds an equal one: a Node forces a full client netmap rebuild.
|
||||
func (entry *connectionEntry) withSelfDelta(data *tailcfg.MapResponse) *tailcfg.MapResponse {
|
||||
if data.Node == nil || !data.Node.Equal(entry.lastSelf.Load()) {
|
||||
return data
|
||||
}
|
||||
|
||||
stripped := *data
|
||||
stripped.Node = nil
|
||||
|
||||
return &stripped
|
||||
}
|
||||
|
||||
// multiChannelNodeConn manages multiple concurrent connections for a single node.
|
||||
@@ -341,7 +360,7 @@ func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error {
|
||||
)
|
||||
|
||||
for _, conn := range snapshot {
|
||||
err := conn.send(data)
|
||||
err := conn.send(conn.withSelfDelta(data))
|
||||
if err != nil {
|
||||
lastErr = err
|
||||
|
||||
@@ -352,6 +371,10 @@ func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error {
|
||||
Msg("send: connection failed")
|
||||
} else {
|
||||
successCount++
|
||||
|
||||
if data.Node != nil {
|
||||
conn.lastSelf.Store(data.Node)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user