mirror of
https://github.com/juanfont/headscale.git
synced 2026-11-04 12:39:25 +09:00
mapper: send no frame when stripping leaves nothing new
Stripping an unchanged Node and SSHPolicy per stream could leave a frame with only ControlTime, which SaaS never sends.
This commit is contained in:
1 parent
2d094f2243
commit
c2b4c97ce1
4 files changed
+67
-32
No files matched your search
@@ -1758,6 +1758,7 @@ func TestMultiChannelSend_SSHPolicyPerConnection(t *testing.T) {
|
||||
mc.addConnection(b)
|
||||
|
||||
data := testMapResponse()
|
||||
data.Domain = "example.com"
|
||||
data.SSHPolicy = sshPolicyForUser("alice")
|
||||
|
||||
require.NoError(t, mc.send(data))
|
||||
@@ -1810,6 +1811,26 @@ func TestPartialPeerAdditionThenRevocation(t *testing.T) {
|
||||
require.Equal(t, []tailcfg.NodeID{peer}, expectReceive(t, fast, "revocation must remove partially delivered peer").PeersRemoved)
|
||||
}
|
||||
|
||||
// TestSendSkipsResponseLeftEmptyByStripping: a stream that holds this Node and
|
||||
// SSHPolicy already has nothing to read, so it gets no frame.
|
||||
func TestSendSkipsResponseLeftEmptyByStripping(t *testing.T) {
|
||||
mc := newMultiChannelNodeConn(1, nil)
|
||||
ch := make(chan *tailcfg.MapResponse, 2)
|
||||
mc.addConnection(makeConnectionEntry("stream", ch))
|
||||
|
||||
now := time.Now()
|
||||
self := &tailcfg.Node{ID: 1}
|
||||
ssh := &tailcfg.SSHPolicy{}
|
||||
frame := &tailcfg.MapResponse{ControlTime: &now, Node: self, SSHPolicy: ssh}
|
||||
|
||||
require.NoError(t, mc.send(frame))
|
||||
require.Len(t, ch, 1)
|
||||
<-ch
|
||||
|
||||
require.NoError(t, mc.send(frame))
|
||||
require.Empty(t, ch, "nothing new for this stream")
|
||||
}
|
||||
|
||||
func TestEmptyMapResponse(t *testing.T) {
|
||||
now := time.Now()
|
||||
|
||||
|
||||
@@ -515,7 +515,15 @@ func (mc *multiChannelNodeConn) send(data *tailcfg.MapResponse) error {
|
||||
defer func() { mc.sendBlocked.Add(int64(time.Since(start))) }()
|
||||
|
||||
for _, conn := range snapshot {
|
||||
err := conn.send(conn.withSSHPolicyDelta(conn.withSelfDelta(data)))
|
||||
stripped := conn.withSSHPolicyDelta(conn.withSelfDelta(data))
|
||||
|
||||
// Stripping can leave this stream nothing new, like the visibility
|
||||
// filtering at build time.
|
||||
if stripped != data && emptyMapResponse(stripped) {
|
||||
stripped = nil
|
||||
}
|
||||
|
||||
err := conn.send(stripped)
|
||||
if errors.Is(err, errConnectionBackpressure) {
|
||||
blocked = append(blocked, conn)
|
||||
continue
|
||||
|
||||
@@ -1237,7 +1237,7 @@ func TestLongPollShutdownWaitsForCleanup(t *testing.T) {
|
||||
})
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
return writer.WriteCount() >= 2 && app.mapBatcher.IsConnected(node.ID())
|
||||
return writer.WriteCount() >= 1 && app.mapBatcher.IsConnected(node.ID())
|
||||
}, 2*time.Second, 10*time.Millisecond, "map session should connect")
|
||||
|
||||
if inGrace {
|
||||
@@ -1322,17 +1322,19 @@ func TestDeletedNodeInterruptsBlockedWrite(t *testing.T) {
|
||||
}()
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
return writer.WriteCount() >= 2 && app.mapBatcher.IsConnected(nodeView.ID())
|
||||
return writer.WriteCount() >= 1 && app.mapBatcher.IsConnected(nodeView.ID())
|
||||
}, 2*time.Second, 10*time.Millisecond,
|
||||
"expected the initial map and connect update to be delivered")
|
||||
"expected the initial map to be delivered")
|
||||
|
||||
writer.BlockWrites()
|
||||
app.Change(change.SelfUpdate(nodeView.ID()))
|
||||
|
||||
// A self update with nothing new would send no frame.
|
||||
app.Change(change.FullUpdate())
|
||||
|
||||
select {
|
||||
case <-writer.WriteStarted():
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("expected the initial map write to block")
|
||||
t.Fatal("expected the next write to block")
|
||||
}
|
||||
|
||||
cs, err := app.state.DeleteNode(nodeView)
|
||||
|
||||
@@ -207,9 +207,6 @@ func TestGrantPolicies(t *testing.T) { //nolint:gocyclo
|
||||
c2.WaitForPeers(t, 1, 10*time.Second)
|
||||
|
||||
// Phase 1: Set a grant policy allowing only TCP port 22.
|
||||
countC1 := c1.UpdateCount()
|
||||
countC2 := c2.UpdateCount()
|
||||
|
||||
changed, err := srv.State().SetPolicy([]byte(`{
|
||||
"grants": [{
|
||||
"src": ["propagation-user@"],
|
||||
@@ -225,25 +222,22 @@ func TestGrantPolicies(t *testing.T) { //nolint:gocyclo
|
||||
srv.App.Change(changes...)
|
||||
}
|
||||
|
||||
c1.WaitForCondition(t, "first grant policy update",
|
||||
10*time.Second,
|
||||
func(nm *netmap.NetworkMap) bool {
|
||||
return c1.UpdateCount() > countC1
|
||||
})
|
||||
c2.WaitForCondition(t, "first grant policy update",
|
||||
10*time.Second,
|
||||
func(nm *netmap.NetworkMap) bool {
|
||||
return c2.UpdateCount() > countC2
|
||||
})
|
||||
// Wait for the filter itself: any other frame, such as a peer's
|
||||
// endpoint update, would also count as an update.
|
||||
for _, c := range []*servertest.TestClient{c1, c2} {
|
||||
c.WaitForCondition(t, "first grant policy update",
|
||||
10*time.Second,
|
||||
func(nm *netmap.NetworkMap) bool {
|
||||
return hasDstPort(nm.PacketFilter, 22)
|
||||
})
|
||||
}
|
||||
|
||||
// Capture the update count after first policy.
|
||||
// Keep the first policy's netmap to compare against.
|
||||
nm1Before := c1.Netmap()
|
||||
require.NotNil(t, nm1Before)
|
||||
require.NotEmpty(t, nm1Before.PacketFilter,
|
||||
"PacketFilter should have rules from first grant policy")
|
||||
|
||||
updateCountPhase1 := c1.UpdateCount()
|
||||
|
||||
// Phase 2: Change to a grant policy with app capability
|
||||
// (structurally different from IP-only grant).
|
||||
changed, err = srv.State().SetPolicy([]byte(`{
|
||||
@@ -271,16 +265,13 @@ func TestGrantPolicies(t *testing.T) { //nolint:gocyclo
|
||||
}
|
||||
|
||||
// Wait for the second policy update to arrive.
|
||||
c1.WaitForCondition(t, "second grant policy update",
|
||||
10*time.Second,
|
||||
func(nm *netmap.NetworkMap) bool {
|
||||
return c1.UpdateCount() > updateCountPhase1
|
||||
})
|
||||
c2.WaitForCondition(t, "second grant policy update",
|
||||
10*time.Second,
|
||||
func(nm *netmap.NetworkMap) bool {
|
||||
return c2.UpdateCount() > countC2
|
||||
})
|
||||
for _, c := range []*servertest.TestClient{c1, c2} {
|
||||
c.WaitForCondition(t, "second grant policy update",
|
||||
10*time.Second,
|
||||
func(nm *netmap.NetworkMap) bool {
|
||||
return hasCapMatches(nm.PacketFilter)
|
||||
})
|
||||
}
|
||||
|
||||
// Verify the second policy added cap grants that weren't
|
||||
// in the first (IP-only) policy.
|
||||
@@ -1105,6 +1096,19 @@ func hasCapMatches(matches []filtertype.Match) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// hasDstPort reports whether any destination in matches is exactly port.
|
||||
func hasDstPort(matches []filtertype.Match, port uint16) bool {
|
||||
for _, m := range matches {
|
||||
for _, d := range m.Dsts {
|
||||
if d.Ports.First == port && d.Ports.Last == port {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
// hasDstRules returns true if any [filtertype.Match] in the slice contains a
|
||||
// non-empty Dsts list.
|
||||
func hasDstRules(matches []filtertype.Match) bool {
|
||||
|
||||
Reference in new issue
Block a user