master: bind heartbeat claims to the connecting peer (#9443)
SendHeartbeat used to accept whatever Ip/Port/Volumes the caller put on
the wire. Three changes tighten that:
- Reject heartbeats whose Ip does not match the gRPC peer's source
address. Loopback peers are still trusted; operators behind a proxy
can opt out with -master.allowUntrustedHeartbeat.
- Track which (ip, port) first claimed a volume id or an ec shard slot
and drop foreign re-claims. Non-EC volume claims are bounded by the
replica copy count so legitimate replicas still register. EC
ownership is keyed by (vid, shard_id) so the same vid can legitimately
be split across many peers as long as their EcIndexBits are disjoint;
rejected bits are cleared from the bitmap and the parallel ShardSizes
array is compacted in lock-step.
- Maintain reverse indexes owner -> volumes and owner -> ec shard slots
so disconnect cleanup is O(M) in what that peer held rather than O(N)
over the whole map.
Bindings are also released when a heartbeat reports that the peer no
longer holds an id, either via explicit Deleted{Volumes,EcShards}
entries or by omitting it from a full snapshot. Without this, a planned
rebalance that moved a vid or an ec shard from peer A to peer B would
leave B's heartbeats permanently filtered out until A disconnected,
breaking ec encode/decode flows that delete shards on the source as
soon as the move completes.
The (vid -> owners) binding still does not track which replica slot
each peer occupies, so the first N claims under the copy count win;
strict per-slot mapping is a follow-up.
This commit is contained in:
Chris Lu
GitHub
parent
10cc06333b
commit
f28c7ce6df
@ -74,6 +74,10 @@ type MasterOptions struct {
|
||||
telemetryEnabled *bool
|
||||
debug *bool
|
||||
debugPort *int
|
||||
// allowUntrustedHeartbeat disables the source-ip binding check on
|
||||
// SendHeartbeat. Required for deployments where a NAT or proxy sits
|
||||
// between volume servers and the master.
|
||||
allowUntrustedHeartbeat *bool
|
||||
// shutdownCtx, when non-nil, tells startMaster to shut down once the ctx
|
||||
// is cancelled. Used by integration tests and by weed mini; nil for
|
||||
// standalone weed master.
|
||||
@ -109,6 +113,7 @@ func init() {
|
||||
m.telemetryEnabled = cmdMaster.Flag.Bool("telemetry", false, "enable telemetry reporting")
|
||||
m.debug = cmdMaster.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port")
|
||||
m.debugPort = cmdMaster.Flag.Int("debug.port", 6060, "http port for debugging")
|
||||
m.allowUntrustedHeartbeat = cmdMaster.Flag.Bool("master.allowUntrustedHeartbeat", false, "accept volume server heartbeats whose advertised ip does not match the gRPC peer address (required behind NAT/proxy)")
|
||||
}
|
||||
|
||||
var cmdMaster = &Command{
|
||||
@ -463,5 +468,6 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp
|
||||
MetricsIntervalSec: *m.metricsIntervalSec,
|
||||
TelemetryUrl: *m.telemetryUrl,
|
||||
TelemetryEnabled: *m.telemetryEnabled,
|
||||
AllowUntrustedHeartbeat: m.allowUntrustedHeartbeat != nil && *m.allowUntrustedHeartbeat,
|
||||
}
|
||||
}
|
||||
|
||||
@ -90,6 +90,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
|
||||
ms.Topo.UnRegisterDataNode(dn)
|
||||
glog.V(0).Infof("unregister disconnected volume server %s:%d", dn.Ip, dn.Port)
|
||||
ms.UnRegisterUuids(dn.Ip, dn.Port)
|
||||
if ms.heartbeatOwnership != nil {
|
||||
ms.heartbeatOwnership.release(dn.Ip, uint32(dn.Port))
|
||||
}
|
||||
|
||||
if ms.Topo.IsLeader() && (len(message.DeletedVids) > 0 || len(message.DeletedEcVids) > 0) {
|
||||
ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
|
||||
@ -135,6 +138,11 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
|
||||
if heartbeat.Ip == "" {
|
||||
continue
|
||||
} // ToDo must be removed after update major version
|
||||
if err := ms.authorizeHeartbeatPeer(stream.Context(), heartbeat); err != nil {
|
||||
glog.Errorf("SendHeartbeat refused: %v", err)
|
||||
stats.MasterReceivedHeartbeatCounter.WithLabelValues("rejected").Inc()
|
||||
return err
|
||||
}
|
||||
dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
|
||||
dc := ms.Topo.GetOrCreateDataCenter(dcName)
|
||||
rack := dc.GetOrCreateRack(rackName)
|
||||
@ -165,6 +173,13 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
|
||||
dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)
|
||||
dn.UpdateDiskTags(heartbeat.DiskTags)
|
||||
|
||||
// Reject claims on volume / ec shard ids that another peer already owns.
|
||||
// Replica-slot semantics (one owner per replica) are approximated by the
|
||||
// declared copy count; the strict per-slot mapping is a follow-up.
|
||||
if ms.heartbeatOwnership != nil {
|
||||
ms.filterVolumeOwnership(dn, heartbeat)
|
||||
}
|
||||
|
||||
glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
|
||||
stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc()
|
||||
|
||||
|
||||
447
weed/server/master_grpc_server_heartbeat_auth.go
Normal file
447
weed/server/master_grpc_server_heartbeat_auth.go
Normal file
@ -0,0 +1,447 @@
|
||||
package weed_server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math/bits"
|
||||
"net"
|
||||
"sync"
|
||||
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/peer"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
||||
"github.com/seaweedfs/seaweedfs/weed/topology"
|
||||
)
|
||||
|
||||
// heartbeatOwner identifies the data node that first claimed a volume or ec
|
||||
// shard id. Two heartbeats are considered to come from the same owner when
|
||||
// both ip and port match.
|
||||
type heartbeatOwner struct {
|
||||
ip string
|
||||
port uint32
|
||||
}
|
||||
|
||||
// ecShardKey identifies a single erasure-coded shard slot, scoped by the
|
||||
// volume id it belongs to. EC distributes shards of the same volume id across
|
||||
// multiple data nodes (one node per shard), so ownership must be tracked per
|
||||
// (vid, shard_id) pair rather than per vid.
|
||||
type ecShardKey struct {
|
||||
vid uint32
|
||||
shardId uint32
|
||||
}
|
||||
|
||||
// heartbeatOwnership tracks which (ip, port) currently owns each volume id
|
||||
// and each ec shard slot. For replicated non-EC volumes, multiple owners are
|
||||
// allowed up to the declared copy count; for EC, each (vid, shard_id) pair
|
||||
// has exactly one owner, but the same vid may be split across many peers
|
||||
// when each peer claims a disjoint subset of shard ids.
|
||||
//
|
||||
// All mutating operations also update reverse indexes keyed by owner so that
|
||||
// stream disconnects can release a peer's bindings in O(M) where M is the
|
||||
// number of slots that specific peer held, rather than walking the full map.
|
||||
type heartbeatOwnership struct {
|
||||
mu sync.Mutex
|
||||
volumes map[uint32][]heartbeatOwner // volume id -> owners
|
||||
ecShards map[ecShardKey]heartbeatOwner // (vid, shard) -> owner
|
||||
ownerToVolumes map[heartbeatOwner]map[uint32]struct{}
|
||||
ownerToEcShards map[heartbeatOwner]map[ecShardKey]struct{}
|
||||
}
|
||||
|
||||
func newHeartbeatOwnership() *heartbeatOwnership {
|
||||
return &heartbeatOwnership{
|
||||
volumes: make(map[uint32][]heartbeatOwner),
|
||||
ecShards: make(map[ecShardKey]heartbeatOwner),
|
||||
ownerToVolumes: make(map[heartbeatOwner]map[uint32]struct{}),
|
||||
ownerToEcShards: make(map[heartbeatOwner]map[ecShardKey]struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// claimVolume returns true if (ip, port) is allowed to claim ownership of vid.
|
||||
// maxOwners caps how many distinct owners the same vid may have (one per
|
||||
// replica slot). Pass 1 if the replica count is unknown.
|
||||
func (o *heartbeatOwnership) claimVolume(vid uint32, ip string, port uint32, maxOwners int) bool {
|
||||
if maxOwners < 1 {
|
||||
maxOwners = 1
|
||||
}
|
||||
owner := heartbeatOwner{ip: ip, port: port}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
owners := o.volumes[vid]
|
||||
for _, existing := range owners {
|
||||
if existing == owner {
|
||||
return true
|
||||
}
|
||||
}
|
||||
if len(owners) >= maxOwners {
|
||||
return false
|
||||
}
|
||||
o.volumes[vid] = append(owners, owner)
|
||||
o.indexVolumeLocked(owner, vid)
|
||||
return true
|
||||
}
|
||||
|
||||
// claimEcShards intersects the requested EcIndexBits with the bits this peer
|
||||
// is allowed to own, claiming each (vid, shard_id) pair independently. The
|
||||
// returned bitmap contains only the shards bound to (ip, port); rejected bits
|
||||
// are dropped so the caller can adjust ShardSizes alignment accordingly.
|
||||
func (o *heartbeatOwnership) claimEcShards(vid uint32, requested uint32, ip string, port uint32) uint32 {
|
||||
if requested == 0 {
|
||||
return 0
|
||||
}
|
||||
owner := heartbeatOwner{ip: ip, port: port}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
var granted uint32
|
||||
for bitmap := requested; bitmap != 0; {
|
||||
shardId := uint32(bits.TrailingZeros32(bitmap))
|
||||
bitmap &^= 1 << shardId
|
||||
key := ecShardKey{vid: vid, shardId: shardId}
|
||||
existing, taken := o.ecShards[key]
|
||||
if taken && existing != owner {
|
||||
continue
|
||||
}
|
||||
if !taken {
|
||||
o.ecShards[key] = owner
|
||||
o.indexEcShardLocked(owner, key)
|
||||
}
|
||||
granted |= 1 << shardId
|
||||
}
|
||||
return granted
|
||||
}
|
||||
|
||||
// release drops every entry owned by (ip, port). Called when a heartbeat
|
||||
// stream ends so that a legitimate restart of the same peer (or a planned
|
||||
// migration) can re-register the same ids. Uses the reverse indexes so the
|
||||
// cost is proportional to what this peer held, not to the size of the cluster.
|
||||
func (o *heartbeatOwnership) release(ip string, port uint32) {
|
||||
owner := heartbeatOwner{ip: ip, port: port}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
for vid := range o.ownerToVolumes[owner] {
|
||||
o.removeVolumeOwnerLocked(vid, owner)
|
||||
}
|
||||
delete(o.ownerToVolumes, owner)
|
||||
for key := range o.ownerToEcShards[owner] {
|
||||
if existing, ok := o.ecShards[key]; ok && existing == owner {
|
||||
delete(o.ecShards, key)
|
||||
}
|
||||
}
|
||||
delete(o.ownerToEcShards, owner)
|
||||
}
|
||||
|
||||
// releaseVolume drops the binding for a single volume id held by (ip, port).
|
||||
// Used when a heartbeat reports the volume as deleted from this peer or when
|
||||
// a full snapshot no longer lists it, so a subsequent peer migration can
|
||||
// claim the same id without waiting for the stream to disconnect.
|
||||
func (o *heartbeatOwnership) releaseVolume(vid uint32, ip string, port uint32) {
|
||||
owner := heartbeatOwner{ip: ip, port: port}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
o.releaseVolumeLocked(vid, owner)
|
||||
}
|
||||
|
||||
// releaseEcShards drops the bindings for every shard set in the bitmap that
|
||||
// is currently owned by (ip, port). Bits owned by other peers are ignored so
|
||||
// a malicious or stale heartbeat cannot evict a legitimate owner.
|
||||
func (o *heartbeatOwnership) releaseEcShards(vid uint32, bitmap uint32, ip string, port uint32) {
|
||||
if bitmap == 0 {
|
||||
return
|
||||
}
|
||||
owner := heartbeatOwner{ip: ip, port: port}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
for b := bitmap; b != 0; {
|
||||
shardId := uint32(bits.TrailingZeros32(b))
|
||||
b &^= 1 << shardId
|
||||
o.releaseEcShardLocked(ecShardKey{vid: vid, shardId: shardId}, owner)
|
||||
}
|
||||
}
|
||||
|
||||
// peerVolumes returns the set of volume ids currently owned by (ip, port).
|
||||
// The caller may mutate the returned slice; the snapshot is taken under the
|
||||
// owner-tracker lock so it is safe to walk concurrently with other heartbeats.
|
||||
func (o *heartbeatOwnership) peerVolumes(ip string, port uint32) []uint32 {
|
||||
owner := heartbeatOwner{ip: ip, port: port}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
held := o.ownerToVolumes[owner]
|
||||
if len(held) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make([]uint32, 0, len(held))
|
||||
for vid := range held {
|
||||
out = append(out, vid)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// peerEcShardBitmaps returns, for each EC volume id this peer currently owns,
|
||||
// the bitmap of shard ids it holds. Used to diff a full snapshot against the
|
||||
// existing bindings so dropped shards can be released.
|
||||
func (o *heartbeatOwnership) peerEcShardBitmaps(ip string, port uint32) map[uint32]uint32 {
|
||||
owner := heartbeatOwner{ip: ip, port: port}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
held := o.ownerToEcShards[owner]
|
||||
if len(held) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make(map[uint32]uint32, len(held))
|
||||
for key := range held {
|
||||
out[key.vid] |= 1 << key.shardId
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// removeVolumeOwnerLocked drops owner from o.volumes[vid] and trims the
|
||||
// underlying slice. Caller must hold o.mu.
|
||||
func (o *heartbeatOwnership) removeVolumeOwnerLocked(vid uint32, owner heartbeatOwner) {
|
||||
owners, ok := o.volumes[vid]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
kept := owners[:0]
|
||||
for _, existing := range owners {
|
||||
if existing != owner {
|
||||
kept = append(kept, existing)
|
||||
}
|
||||
}
|
||||
if len(kept) == 0 {
|
||||
delete(o.volumes, vid)
|
||||
} else {
|
||||
o.volumes[vid] = kept
|
||||
}
|
||||
}
|
||||
|
||||
// releaseVolumeLocked drops a single (vid, owner) binding and tidies the
|
||||
// reverse index. Caller must hold o.mu.
|
||||
func (o *heartbeatOwnership) releaseVolumeLocked(vid uint32, owner heartbeatOwner) {
|
||||
o.removeVolumeOwnerLocked(vid, owner)
|
||||
if set, ok := o.ownerToVolumes[owner]; ok {
|
||||
delete(set, vid)
|
||||
if len(set) == 0 {
|
||||
delete(o.ownerToVolumes, owner)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// releaseEcShardLocked drops a single (vid, shard, owner) binding and tidies
|
||||
// the reverse index. Caller must hold o.mu.
|
||||
func (o *heartbeatOwnership) releaseEcShardLocked(key ecShardKey, owner heartbeatOwner) {
|
||||
if existing, ok := o.ecShards[key]; !ok || existing != owner {
|
||||
return
|
||||
}
|
||||
delete(o.ecShards, key)
|
||||
if set, ok := o.ownerToEcShards[owner]; ok {
|
||||
delete(set, key)
|
||||
if len(set) == 0 {
|
||||
delete(o.ownerToEcShards, owner)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// indexVolumeLocked records the reverse-mapping owner -> vid. Caller must
|
||||
// hold o.mu.
|
||||
func (o *heartbeatOwnership) indexVolumeLocked(owner heartbeatOwner, vid uint32) {
|
||||
set, ok := o.ownerToVolumes[owner]
|
||||
if !ok {
|
||||
set = make(map[uint32]struct{})
|
||||
o.ownerToVolumes[owner] = set
|
||||
}
|
||||
set[vid] = struct{}{}
|
||||
}
|
||||
|
||||
// indexEcShardLocked records the reverse-mapping owner -> (vid, shard).
|
||||
// Caller must hold o.mu.
|
||||
func (o *heartbeatOwnership) indexEcShardLocked(owner heartbeatOwner, key ecShardKey) {
|
||||
set, ok := o.ownerToEcShards[owner]
|
||||
if !ok {
|
||||
set = make(map[ecShardKey]struct{})
|
||||
o.ownerToEcShards[owner] = set
|
||||
}
|
||||
set[key] = struct{}{}
|
||||
}
|
||||
|
||||
// authorizeHeartbeatPeer checks that heartbeat.Ip matches the connection's
|
||||
// source address. Loopback peers are trusted because dev clusters routinely
|
||||
// run master and volume server on the same host with mismatched advertised
|
||||
// ips. In-process / bufconn / unix-socket peers (used by `weed server` where
|
||||
// master and volume run in the same OS process) surface with a non-TCP
|
||||
// net.Addr and are also trusted. Operators with a proxy in front of the
|
||||
// master can disable this check via MasterOption.AllowUntrustedHeartbeat.
|
||||
func (ms *MasterServer) authorizeHeartbeatPeer(ctx context.Context, heartbeat *master_pb.Heartbeat) error {
|
||||
if ms.option != nil && ms.option.AllowUntrustedHeartbeat {
|
||||
return nil
|
||||
}
|
||||
pr, ok := peer.FromContext(ctx)
|
||||
if !ok || pr.Addr == nil {
|
||||
return fmt.Errorf("heartbeat peer info missing")
|
||||
}
|
||||
if tlsInfo, ok := pr.AuthInfo.(credentials.TLSInfo); ok && len(tlsInfo.State.PeerCertificates) > 0 {
|
||||
cn := tlsInfo.State.PeerCertificates[0].Subject.CommonName
|
||||
glog.V(2).Infof("heartbeat from CN=%q claiming ip=%s:%d", cn, heartbeat.Ip, heartbeat.Port)
|
||||
}
|
||||
tcpAddr, isTCP := pr.Addr.(*net.TCPAddr)
|
||||
if !isTCP {
|
||||
// In-process / bufconn / unix-socket peer; same OS process as the
|
||||
// master, so the source IP cannot be spoofed by a remote attacker.
|
||||
return nil
|
||||
}
|
||||
host := tcpAddr.IP.String()
|
||||
if host == heartbeat.Ip {
|
||||
return nil
|
||||
}
|
||||
if tcpAddr.IP.IsLoopback() {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("heartbeat ip %q does not match peer %q", heartbeat.Ip, host)
|
||||
}
|
||||
|
||||
// volumeMaxOwners returns how many distinct peers may legitimately own the
|
||||
// same volume id, derived from its replica placement byte. Falls back to one
|
||||
// owner if the placement byte is malformed.
|
||||
func volumeMaxOwners(replicaPlacement uint32) int {
|
||||
rp, err := super_block.NewReplicaPlacementFromByte(byte(replicaPlacement))
|
||||
if err != nil || rp == nil {
|
||||
return 1
|
||||
}
|
||||
return rp.GetCopyCount()
|
||||
}
|
||||
|
||||
// filterVolumeOwnership drops volume / ec shard claims from the heartbeat
|
||||
// whenever another peer already owns that id. The peer's own heartbeat keeps
|
||||
// its existing claims; only foreign claims are stripped. EC shards are
|
||||
// filtered per (vid, shard_id), so multiple peers can legitimately heartbeat
|
||||
// the same EC vid as long as their shard bitmaps are disjoint.
|
||||
//
|
||||
// Bindings are also released when the heartbeat reports that this peer no
|
||||
// longer owns an id, either explicitly via Deleted{Volumes,EcShards} or
|
||||
// implicitly by omitting it from a full snapshot. Without this, a planned
|
||||
// migration that moves a vid (or an EC shard) from peer A to peer B would
|
||||
// leave B's heartbeats permanently filtered out because the first-claim
|
||||
// binding to A never expires until A disconnects.
|
||||
func (ms *MasterServer) filterVolumeOwnership(dn *topology.DataNode, heartbeat *master_pb.Heartbeat) {
|
||||
ip := dn.Ip
|
||||
port := uint32(dn.Port)
|
||||
|
||||
// Explicit drops: release before we claim survivors so a same-heartbeat
|
||||
// rebalance (delete shard X, claim shard Y) does not race itself.
|
||||
for _, v := range heartbeat.DeletedVolumes {
|
||||
ms.heartbeatOwnership.releaseVolume(v.Id, ip, port)
|
||||
}
|
||||
for _, s := range heartbeat.DeletedEcShards {
|
||||
if s.EcIndexBits != 0 {
|
||||
ms.heartbeatOwnership.releaseEcShards(s.Id, s.EcIndexBits, ip, port)
|
||||
}
|
||||
}
|
||||
|
||||
// Full snapshot of regular volumes: diff against this peer's existing
|
||||
// bindings and release every vid the peer no longer reports. Triggered by
|
||||
// a non-empty Volumes list or by HasNoVolumes=true (the latter means the
|
||||
// peer holds zero volumes after the snapshot).
|
||||
if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
|
||||
present := make(map[uint32]struct{}, len(heartbeat.Volumes))
|
||||
for _, v := range heartbeat.Volumes {
|
||||
present[v.Id] = struct{}{}
|
||||
}
|
||||
for _, vid := range ms.heartbeatOwnership.peerVolumes(ip, port) {
|
||||
if _, still := present[vid]; !still {
|
||||
ms.heartbeatOwnership.releaseVolume(vid, ip, port)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Full snapshot of EC shards: same diff, but bit-wise per vid. Triggered
|
||||
// by EcShards being non-empty or HasNoEcShards=true.
|
||||
if len(heartbeat.EcShards) > 0 || heartbeat.HasNoEcShards {
|
||||
present := make(map[uint32]uint32, len(heartbeat.EcShards))
|
||||
for _, s := range heartbeat.EcShards {
|
||||
present[s.Id] |= s.EcIndexBits
|
||||
}
|
||||
for vid, held := range ms.heartbeatOwnership.peerEcShardBitmaps(ip, port) {
|
||||
dropped := held &^ present[vid]
|
||||
if dropped != 0 {
|
||||
ms.heartbeatOwnership.releaseEcShards(vid, dropped, ip, port)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if n := len(heartbeat.Volumes); n > 0 {
|
||||
kept := heartbeat.Volumes[:0]
|
||||
for _, v := range heartbeat.Volumes {
|
||||
if ms.heartbeatOwnership.claimVolume(v.Id, ip, port, volumeMaxOwners(v.ReplicaPlacement)) {
|
||||
kept = append(kept, v)
|
||||
} else {
|
||||
glog.V(0).Infof("rejecting volume %d claim from %s:%d (owned by another peer)", v.Id, ip, port)
|
||||
}
|
||||
}
|
||||
heartbeat.Volumes = kept
|
||||
}
|
||||
|
||||
if n := len(heartbeat.NewVolumes); n > 0 {
|
||||
kept := heartbeat.NewVolumes[:0]
|
||||
for _, v := range heartbeat.NewVolumes {
|
||||
if ms.heartbeatOwnership.claimVolume(v.Id, ip, port, volumeMaxOwners(v.ReplicaPlacement)) {
|
||||
kept = append(kept, v)
|
||||
} else {
|
||||
glog.V(0).Infof("rejecting new volume %d claim from %s:%d (owned by another peer)", v.Id, ip, port)
|
||||
}
|
||||
}
|
||||
heartbeat.NewVolumes = kept
|
||||
}
|
||||
|
||||
if n := len(heartbeat.EcShards); n > 0 {
|
||||
heartbeat.EcShards = ms.applyEcShardOwnership(heartbeat.EcShards, ip, port, "ec shard")
|
||||
}
|
||||
|
||||
if n := len(heartbeat.NewEcShards); n > 0 {
|
||||
heartbeat.NewEcShards = ms.applyEcShardOwnership(heartbeat.NewEcShards, ip, port, "new ec shard")
|
||||
}
|
||||
}
|
||||
|
||||
// applyEcShardOwnership claims (vid, shard_id) ownership for each shard set
|
||||
// in the message bitmap. Bits owned by a different peer are cleared from
|
||||
// EcIndexBits and the parallel ShardSizes array is compacted in lock-step;
|
||||
// messages that end up with no remaining shards are dropped entirely.
|
||||
func (ms *MasterServer) applyEcShardOwnership(messages []*master_pb.VolumeEcShardInformationMessage, ip string, port uint32, label string) []*master_pb.VolumeEcShardInformationMessage {
|
||||
kept := messages[:0]
|
||||
for _, s := range messages {
|
||||
granted := ms.heartbeatOwnership.claimEcShards(s.Id, s.EcIndexBits, ip, port)
|
||||
if granted == s.EcIndexBits {
|
||||
kept = append(kept, s)
|
||||
continue
|
||||
}
|
||||
rejected := s.EcIndexBits &^ granted
|
||||
glog.V(0).Infof("rejecting %s bits 0x%x for volume %d from %s:%d (owned by another peer)", label, rejected, s.Id, ip, port)
|
||||
if granted == 0 {
|
||||
continue
|
||||
}
|
||||
s.ShardSizes = compactShardSizes(s.EcIndexBits, granted, s.ShardSizes)
|
||||
s.EcIndexBits = granted
|
||||
kept = append(kept, s)
|
||||
}
|
||||
return kept
|
||||
}
|
||||
|
||||
// compactShardSizes returns the subset of sizes that corresponds to the bits
|
||||
// in granted. The input sizes slice is parallel to the set bits in original
|
||||
// (in increasing shard-id order); the same ordering applies to the output.
|
||||
func compactShardSizes(original, granted uint32, sizes []int64) []int64 {
|
||||
if len(sizes) == 0 || granted == original {
|
||||
return sizes
|
||||
}
|
||||
out := sizes[:0]
|
||||
var i int
|
||||
for bitmap := original; bitmap != 0; {
|
||||
shardId := uint32(bits.TrailingZeros32(bitmap))
|
||||
bitmap &^= 1 << shardId
|
||||
if i < len(sizes) && granted&(1<<shardId) != 0 {
|
||||
out = append(out, sizes[i])
|
||||
}
|
||||
i++
|
||||
}
|
||||
return out
|
||||
}
|
||||
613
weed/server/master_grpc_server_heartbeat_auth_test.go
Normal file
613
weed/server/master_grpc_server_heartbeat_auth_test.go
Normal file
@ -0,0 +1,613 @@
|
||||
package weed_server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"testing"
|
||||
|
||||
"google.golang.org/grpc/peer"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/sequence"
|
||||
"github.com/seaweedfs/seaweedfs/weed/topology"
|
||||
)
|
||||
|
||||
func TestHeartbeatOwnership_FirstClaimWins(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
|
||||
if !o.claimVolume(1, "10.0.0.1", 8080, 1) {
|
||||
t.Fatalf("first claim must succeed")
|
||||
}
|
||||
if o.claimVolume(1, "10.6.6.6", 8080, 1) {
|
||||
t.Fatalf("attacker claim from a different peer must be rejected")
|
||||
}
|
||||
if !o.claimVolume(1, "10.0.0.1", 8080, 1) {
|
||||
t.Fatalf("re-claim from the original peer must succeed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHeartbeatOwnership_RespectsReplicaCount(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
|
||||
// 3 legitimate replicas
|
||||
if !o.claimVolume(2, "10.0.0.1", 8080, 3) {
|
||||
t.Fatal()
|
||||
}
|
||||
if !o.claimVolume(2, "10.0.0.2", 8080, 3) {
|
||||
t.Fatal()
|
||||
}
|
||||
if !o.claimVolume(2, "10.0.0.3", 8080, 3) {
|
||||
t.Fatal()
|
||||
}
|
||||
// 4th distinct peer is over count
|
||||
if o.claimVolume(2, "10.6.6.6", 8080, 3) {
|
||||
t.Fatalf("over-count claim must be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHeartbeatOwnership_Release(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
|
||||
if !o.claimVolume(3, "10.0.0.1", 8080, 1) {
|
||||
t.Fatal()
|
||||
}
|
||||
o.release("10.0.0.1", 8080)
|
||||
if !o.claimVolume(3, "10.0.0.2", 8080, 1) {
|
||||
t.Fatalf("after release, another peer should be able to take the slot")
|
||||
}
|
||||
}
|
||||
|
||||
// TestHeartbeatOwnership_EcShardsPerSlot verifies the core EC correctness
|
||||
// property: each (vid, shard_id) pair has exactly one owner, but the same
|
||||
// vid may legitimately be split across multiple peers when they claim
|
||||
// disjoint shard bitmaps. The previous vid-only model rejected those.
|
||||
func TestHeartbeatOwnership_EcShardsPerSlot(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
|
||||
// Peer A claims shards 0,1,2 of vid 7.
|
||||
if got := o.claimEcShards(7, 0b0111, "10.0.0.1", 8080); got != 0b0111 {
|
||||
t.Fatalf("peer A should own shards 0-2, got 0x%x", got)
|
||||
}
|
||||
// Re-claim from the owning peer is fine.
|
||||
if got := o.claimEcShards(7, 0b0111, "10.0.0.1", 8080); got != 0b0111 {
|
||||
t.Fatalf("re-claim from owner should succeed, got 0x%x", got)
|
||||
}
|
||||
// Peer B claims shards 3,4 of the same vid: must succeed because the
|
||||
// slots are disjoint.
|
||||
if got := o.claimEcShards(7, 0b11000, "10.0.0.2", 8080); got != 0b11000 {
|
||||
t.Fatalf("peer B should own shards 3-4, got 0x%x", got)
|
||||
}
|
||||
// Peer C tries to hijack shard 0 (owned by A) and claim shard 5: only
|
||||
// the unowned bit comes through.
|
||||
if got := o.claimEcShards(7, 0b100001, "10.6.6.6", 8080); got != 0b100000 {
|
||||
t.Fatalf("hijack of shard 0 must be rejected; shard 5 must be granted; got 0x%x", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHeartbeatOwnership_EcShardReleaseIsolatesPeers asserts that
|
||||
// disconnecting peer A frees only A's bindings, leaving B's intact.
|
||||
func TestHeartbeatOwnership_EcShardReleaseIsolatesPeers(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
|
||||
o.claimEcShards(9, 0b0111, "10.0.0.1", 8080)
|
||||
o.claimEcShards(9, 0b11000, "10.0.0.2", 8080)
|
||||
|
||||
o.release("10.0.0.1", 8080)
|
||||
|
||||
// B's shards still locked.
|
||||
if got := o.claimEcShards(9, 0b11000, "10.6.6.6", 8080); got != 0 {
|
||||
t.Fatalf("B's bindings must survive A's release, got 0x%x", got)
|
||||
}
|
||||
// A's freed shards may be reclaimed by anyone.
|
||||
if got := o.claimEcShards(9, 0b0111, "10.6.6.6", 8080); got != 0b0111 {
|
||||
t.Fatalf("A's freed shards should be reclaimable, got 0x%x", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHeartbeatOwnership_ReleaseScopedToOwner checks that volume bindings
|
||||
// of peer B are not collateral damage when peer A disconnects.
|
||||
func TestHeartbeatOwnership_ReleaseScopedToOwner(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
|
||||
o.claimVolume(10, "10.0.0.1", 8080, 2)
|
||||
o.claimVolume(10, "10.0.0.2", 8080, 2)
|
||||
|
||||
o.release("10.0.0.1", 8080)
|
||||
|
||||
// Slot vacated by A is reclaimable by a third peer.
|
||||
if !o.claimVolume(10, "10.0.0.3", 8080, 2) {
|
||||
t.Fatalf("A's slot should be reclaimable after release")
|
||||
}
|
||||
// B still owns its slot; a fourth peer must not push past the cap.
|
||||
if o.claimVolume(10, "10.6.6.6", 8080, 2) {
|
||||
t.Fatalf("over-count claim must be rejected; B's slot was not freed")
|
||||
}
|
||||
}
|
||||
|
||||
type fakeAddr struct {
|
||||
network string
|
||||
s string
|
||||
}
|
||||
|
||||
func (f fakeAddr) Network() string { return f.network }
|
||||
func (f fakeAddr) String() string { return f.s }
|
||||
|
||||
func tcpPeerContext(ip string, port int) context.Context {
|
||||
return peer.NewContext(context.Background(), &peer.Peer{
|
||||
Addr: &net.TCPAddr{IP: net.ParseIP(ip), Port: port},
|
||||
})
|
||||
}
|
||||
|
||||
func TestAuthorizeHeartbeatPeer_AcceptsMatchingIp(t *testing.T) {
|
||||
ms := &MasterServer{option: &MasterOption{}}
|
||||
ctx := tcpPeerContext("10.0.0.1", 55555)
|
||||
|
||||
if err := ms.authorizeHeartbeatPeer(ctx, &master_pb.Heartbeat{Ip: "10.0.0.1", Port: 8080}); err != nil {
|
||||
t.Fatalf("unexpected rejection: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthorizeHeartbeatPeer_RejectsMismatch(t *testing.T) {
|
||||
ms := &MasterServer{option: &MasterOption{}}
|
||||
ctx := tcpPeerContext("10.0.0.1", 55555)
|
||||
|
||||
if err := ms.authorizeHeartbeatPeer(ctx, &master_pb.Heartbeat{Ip: "10.6.6.6", Port: 8080}); err == nil {
|
||||
t.Fatalf("attacker should be rejected when source ip does not match claim")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthorizeHeartbeatPeer_LoopbackAccepted(t *testing.T) {
|
||||
ms := &MasterServer{option: &MasterOption{}}
|
||||
ctx := tcpPeerContext("127.0.0.1", 55555)
|
||||
|
||||
// A loopback peer is allowed to advertise its external hostname.
|
||||
if err := ms.authorizeHeartbeatPeer(ctx, &master_pb.Heartbeat{Ip: "host.example", Port: 8080}); err != nil {
|
||||
t.Fatalf("loopback peer should be trusted: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAuthorizeHeartbeatPeer_InProcessAccepted covers the `weed server` case
|
||||
// where master + volume run in the same OS process and the volume server
|
||||
// reaches the master through an in-process / bufconn gRPC connection. Those
|
||||
// peers surface with a non-TCP net.Addr (e.g. "@") and cannot be spoofed by
|
||||
// a remote attacker, so they must be trusted regardless of heartbeat.Ip.
|
||||
func TestAuthorizeHeartbeatPeer_InProcessAccepted(t *testing.T) {
|
||||
ms := &MasterServer{option: &MasterOption{}}
|
||||
ctx := peer.NewContext(context.Background(), &peer.Peer{
|
||||
Addr: fakeAddr{network: "bufconn", s: "@"},
|
||||
})
|
||||
|
||||
if err := ms.authorizeHeartbeatPeer(ctx, &master_pb.Heartbeat{Ip: "host.example", Port: 8080}); err != nil {
|
||||
t.Fatalf("in-process peer should be trusted: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFilterVolumeOwnership_RejectsSecondPeerOnSameVolumeId verifies the
|
||||
// end-to-end behaviour for non-EC: with replica count 1 there is exactly one
|
||||
// legal owner of a vid, and a second peer's claim is stripped before the
|
||||
// topology sees it.
|
||||
func TestFilterVolumeOwnership_RejectsSecondPeerOnSameVolumeId(t *testing.T) {
|
||||
topo := topology.NewTopology("t", sequence.NewMemorySequencer(), 32*1024, 5, false)
|
||||
dc := topo.GetOrCreateDataCenter("dc")
|
||||
rack := dc.GetOrCreateRack("rack")
|
||||
|
||||
ms := &MasterServer{
|
||||
option: &MasterOption{},
|
||||
Topo: topo,
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
|
||||
dnA := rack.GetOrCreateDataNode("10.0.0.1", 8080, 8081, "10.0.0.1", "A", map[string]uint32{"": 10})
|
||||
dnB := rack.GetOrCreateDataNode("10.6.6.6", 8080, 8081, "10.6.6.6", "B", map[string]uint32{"": 10})
|
||||
|
||||
hbA := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
Volumes: []*master_pb.VolumeInformationMessage{
|
||||
{Id: 1, ReplicaPlacement: 0},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbA)
|
||||
if len(hbA.Volumes) != 1 {
|
||||
t.Fatalf("peer A should retain its claim, got %d volumes", len(hbA.Volumes))
|
||||
}
|
||||
|
||||
hbB := &master_pb.Heartbeat{
|
||||
Ip: "10.6.6.6",
|
||||
Port: 8080,
|
||||
Volumes: []*master_pb.VolumeInformationMessage{
|
||||
{Id: 1, ReplicaPlacement: 0},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbB)
|
||||
if len(hbB.Volumes) != 0 {
|
||||
t.Fatalf("peer B's hijack claim should be stripped, got %d volumes", len(hbB.Volumes))
|
||||
}
|
||||
}
|
||||
|
||||
// TestFilterVolumeOwnership_ReleaseLetsNextPeerTakeOver covers the
|
||||
// disconnect-and-replace flow: when peer A's stream ends, ownership is freed,
|
||||
// and a different peer may then claim the same vid (e.g. after planned
|
||||
// migration).
|
||||
func TestFilterVolumeOwnership_ReleaseLetsNextPeerTakeOver(t *testing.T) {
|
||||
ms := &MasterServer{
|
||||
option: &MasterOption{},
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
|
||||
topo := topology.NewTopology("t", sequence.NewMemorySequencer(), 32*1024, 5, false)
|
||||
dc := topo.GetOrCreateDataCenter("dc")
|
||||
rack := dc.GetOrCreateRack("rack")
|
||||
dnA := rack.GetOrCreateDataNode("10.0.0.1", 8080, 8081, "10.0.0.1", "A", map[string]uint32{"": 10})
|
||||
dnB := rack.GetOrCreateDataNode("10.0.0.2", 8080, 8081, "10.0.0.2", "B", map[string]uint32{"": 10})
|
||||
|
||||
hb := func(ip string) *master_pb.Heartbeat {
|
||||
return &master_pb.Heartbeat{Ip: ip, Port: 8080, Volumes: []*master_pb.VolumeInformationMessage{{Id: 5}}}
|
||||
}
|
||||
|
||||
a := hb("10.0.0.1")
|
||||
ms.filterVolumeOwnership(dnA, a)
|
||||
if len(a.Volumes) != 1 {
|
||||
t.Fatalf("peer A should own the volume")
|
||||
}
|
||||
|
||||
b := hb("10.0.0.2")
|
||||
ms.filterVolumeOwnership(dnB, b)
|
||||
if len(b.Volumes) != 0 {
|
||||
t.Fatalf("peer B must not hijack while peer A still owns the id")
|
||||
}
|
||||
|
||||
ms.heartbeatOwnership.release(dnA.Ip, uint32(dnA.Port))
|
||||
|
||||
c := hb("10.0.0.2")
|
||||
ms.filterVolumeOwnership(dnB, c)
|
||||
if len(c.Volumes) != 1 {
|
||||
t.Fatalf("after release, peer B should be able to take over")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFilterVolumeOwnership_EcShardsSplitAcrossPeers exercises the realistic
|
||||
// EC topology where shards of the same vid live on different volume servers.
|
||||
// Both peers must keep their disjoint shard bitmaps after filtering.
|
||||
func TestFilterVolumeOwnership_EcShardsSplitAcrossPeers(t *testing.T) {
|
||||
ms := &MasterServer{
|
||||
option: &MasterOption{},
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
topo := topology.NewTopology("t", sequence.NewMemorySequencer(), 32*1024, 5, false)
|
||||
dc := topo.GetOrCreateDataCenter("dc")
|
||||
rack := dc.GetOrCreateRack("rack")
|
||||
dnA := rack.GetOrCreateDataNode("10.0.0.1", 8080, 8081, "10.0.0.1", "A", map[string]uint32{"": 10})
|
||||
dnB := rack.GetOrCreateDataNode("10.0.0.2", 8080, 8081, "10.0.0.2", "B", map[string]uint32{"": 10})
|
||||
|
||||
hbA := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
EcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 100, EcIndexBits: 0x7F, ShardSizes: []int64{1, 2, 3, 4, 5, 6, 7}}, // shards 0..6
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbA)
|
||||
if len(hbA.EcShards) != 1 || hbA.EcShards[0].EcIndexBits != 0x7F {
|
||||
t.Fatalf("peer A should keep shards 0-6, got %+v", hbA.EcShards)
|
||||
}
|
||||
if len(hbA.EcShards[0].ShardSizes) != 7 {
|
||||
t.Fatalf("shard sizes for peer A should be preserved, got %v", hbA.EcShards[0].ShardSizes)
|
||||
}
|
||||
|
||||
hbB := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.2",
|
||||
Port: 8080,
|
||||
EcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 100, EcIndexBits: 0x3F80, ShardSizes: []int64{8, 9, 10, 11, 12, 13, 14}}, // shards 7..13
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbB)
|
||||
if len(hbB.EcShards) != 1 || hbB.EcShards[0].EcIndexBits != 0x3F80 {
|
||||
t.Fatalf("peer B should keep shards 7-13, got %+v", hbB.EcShards)
|
||||
}
|
||||
if len(hbB.EcShards[0].ShardSizes) != 7 {
|
||||
t.Fatalf("shard sizes for peer B should be preserved, got %v", hbB.EcShards[0].ShardSizes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFilterVolumeOwnership_EcShardHijackStripped checks the partial-overlap
|
||||
// case: the granted bits are kept (with their sizes), the rejected bits are
|
||||
// stripped, and ShardSizes is compacted in lock-step with EcIndexBits.
|
||||
func TestFilterVolumeOwnership_EcShardHijackStripped(t *testing.T) {
|
||||
ms := &MasterServer{
|
||||
option: &MasterOption{},
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
topo := topology.NewTopology("t", sequence.NewMemorySequencer(), 32*1024, 5, false)
|
||||
dc := topo.GetOrCreateDataCenter("dc")
|
||||
rack := dc.GetOrCreateRack("rack")
|
||||
dnA := rack.GetOrCreateDataNode("10.0.0.1", 8080, 8081, "10.0.0.1", "A", map[string]uint32{"": 10})
|
||||
dnAttacker := rack.GetOrCreateDataNode("10.6.6.6", 8080, 8081, "10.6.6.6", "X", map[string]uint32{"": 10})
|
||||
|
||||
hbA := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
EcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 200, EcIndexBits: 0b0111, ShardSizes: []int64{10, 20, 30}}, // shards 0,1,2
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbA)
|
||||
|
||||
hbAttacker := &master_pb.Heartbeat{
|
||||
Ip: "10.6.6.6",
|
||||
Port: 8080,
|
||||
EcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
// claim shards 1 (taken), 3 (free), 5 (free), with sizes in
|
||||
// shard-id order.
|
||||
{Id: 200, EcIndexBits: 0b101010, ShardSizes: []int64{77, 88, 99}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnAttacker, hbAttacker)
|
||||
if len(hbAttacker.EcShards) != 1 {
|
||||
t.Fatalf("attacker should keep the message with the granted bits only, got %d", len(hbAttacker.EcShards))
|
||||
}
|
||||
got := hbAttacker.EcShards[0]
|
||||
if got.EcIndexBits != 0b101000 {
|
||||
t.Fatalf("expected only shards 3 and 5 to survive, got 0x%x", got.EcIndexBits)
|
||||
}
|
||||
if len(got.ShardSizes) != 2 || got.ShardSizes[0] != 88 || got.ShardSizes[1] != 99 {
|
||||
t.Fatalf("ShardSizes must compact alongside EcIndexBits, got %v", got.ShardSizes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHeartbeatOwnership_InvariantsAfterRelease covers the reverse-index
|
||||
// bookkeeping: after release, the owner-keyed maps are empty for that peer.
|
||||
func TestHeartbeatOwnership_InvariantsAfterRelease(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
owner := heartbeatOwner{ip: "10.0.0.1", port: 8080}
|
||||
|
||||
o.claimVolume(1, owner.ip, owner.port, 1)
|
||||
o.claimVolume(2, owner.ip, owner.port, 1)
|
||||
o.claimEcShards(3, 0b101, owner.ip, owner.port)
|
||||
|
||||
o.release(owner.ip, owner.port)
|
||||
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
if len(o.volumes) != 0 || len(o.ecShards) != 0 {
|
||||
t.Fatalf("forward maps must be empty after sole owner releases: volumes=%v ecShards=%v", o.volumes, o.ecShards)
|
||||
}
|
||||
if _, ok := o.ownerToVolumes[owner]; ok {
|
||||
t.Fatalf("owner-volume reverse index leaked")
|
||||
}
|
||||
if _, ok := o.ownerToEcShards[owner]; ok {
|
||||
t.Fatalf("owner-ecshard reverse index leaked")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuthorizeHeartbeatPeer_OptOut(t *testing.T) {
|
||||
ms := &MasterServer{option: &MasterOption{AllowUntrustedHeartbeat: true}}
|
||||
ctx := tcpPeerContext("10.0.0.1", 55555)
|
||||
|
||||
if err := ms.authorizeHeartbeatPeer(ctx, &master_pb.Heartbeat{Ip: "10.6.6.6", Port: 8080}); err != nil {
|
||||
t.Fatalf("opt-out must skip the check: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFilterVolumeOwnership_DeletedEcShardsReleaseBinding covers the
|
||||
// rebalance flow: peer A heartbeats shard 0 of vid 5, then a follow-up
|
||||
// heartbeat from A reports the shard as deleted. The binding must release so
|
||||
// peer B can claim shard 0 on its next heartbeat.
|
||||
func TestFilterVolumeOwnership_DeletedEcShardsReleaseBinding(t *testing.T) {
|
||||
ms := &MasterServer{
|
||||
option: &MasterOption{},
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
topo := topology.NewTopology("t", sequence.NewMemorySequencer(), 32*1024, 5, false)
|
||||
dc := topo.GetOrCreateDataCenter("dc")
|
||||
rack := dc.GetOrCreateRack("rack")
|
||||
dnA := rack.GetOrCreateDataNode("10.0.0.1", 8080, 8081, "10.0.0.1", "A", map[string]uint32{"": 10})
|
||||
dnB := rack.GetOrCreateDataNode("10.0.0.2", 8080, 8081, "10.0.0.2", "B", map[string]uint32{"": 10})
|
||||
|
||||
// Peer A claims shard 0 of vid 5.
|
||||
hbAClaim := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b1, ShardSizes: []int64{100}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbAClaim)
|
||||
if len(hbAClaim.NewEcShards) != 1 || hbAClaim.NewEcShards[0].EcIndexBits != 0b1 {
|
||||
t.Fatalf("peer A initial claim must succeed, got %+v", hbAClaim.NewEcShards)
|
||||
}
|
||||
|
||||
// Peer B tries to claim shard 0 first; the binding still belongs to A so
|
||||
// the claim is filtered out.
|
||||
hbBHijack := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.2",
|
||||
Port: 8080,
|
||||
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b1, ShardSizes: []int64{100}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbBHijack)
|
||||
if len(hbBHijack.NewEcShards) != 0 {
|
||||
t.Fatalf("peer B must not hijack shard 0 while A owns it, got %+v", hbBHijack.NewEcShards)
|
||||
}
|
||||
|
||||
// Peer A heartbeat reports shard 0 deleted. This must release the
|
||||
// binding so a different peer can take ownership next.
|
||||
hbADelete := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
DeletedEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b1},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbADelete)
|
||||
|
||||
// Peer B re-tries the claim; this time the slot is free.
|
||||
hbBTakeover := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.2",
|
||||
Port: 8080,
|
||||
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b1, ShardSizes: []int64{100}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbBTakeover)
|
||||
if len(hbBTakeover.NewEcShards) != 1 || hbBTakeover.NewEcShards[0].EcIndexBits != 0b1 {
|
||||
t.Fatalf("peer B should own shard 0 after A's release, got %+v", hbBTakeover.NewEcShards)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFilterVolumeOwnership_FullSnapshotDropsStaleEcBits covers the case
|
||||
// where a peer drops a shard between heartbeats without sending an explicit
|
||||
// DeletedEcShards entry (e.g. after a restart that reloaded fewer shards).
|
||||
// The full EcShards snapshot must release the shard so another peer can
|
||||
// claim it.
|
||||
func TestFilterVolumeOwnership_FullSnapshotDropsStaleEcBits(t *testing.T) {
|
||||
ms := &MasterServer{
|
||||
option: &MasterOption{},
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
topo := topology.NewTopology("t", sequence.NewMemorySequencer(), 32*1024, 5, false)
|
||||
dc := topo.GetOrCreateDataCenter("dc")
|
||||
rack := dc.GetOrCreateRack("rack")
|
||||
dnA := rack.GetOrCreateDataNode("10.0.0.1", 8080, 8081, "10.0.0.1", "A", map[string]uint32{"": 10})
|
||||
dnB := rack.GetOrCreateDataNode("10.0.0.2", 8080, 8081, "10.0.0.2", "B", map[string]uint32{"": 10})
|
||||
|
||||
// Peer A snapshots shards {0, 1, 3} = 0b1011 for vid 5.
|
||||
hbAFull := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
EcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b1011, ShardSizes: []int64{10, 20, 30}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbAFull)
|
||||
if hbAFull.EcShards[0].EcIndexBits != 0b1011 {
|
||||
t.Fatalf("peer A initial snapshot should keep all three bits, got 0x%x", hbAFull.EcShards[0].EcIndexBits)
|
||||
}
|
||||
|
||||
// Peer A re-snapshots with shard 1 dropped: 0b1001.
|
||||
hbAShrink := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
EcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b1001, ShardSizes: []int64{10, 30}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbAShrink)
|
||||
if hbAShrink.EcShards[0].EcIndexBits != 0b1001 {
|
||||
t.Fatalf("peer A shrink snapshot should keep the two surviving bits, got 0x%x", hbAShrink.EcShards[0].EcIndexBits)
|
||||
}
|
||||
|
||||
// Peer B can now claim shard 1 because the diff released that bit.
|
||||
hbBTakeover := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.2",
|
||||
Port: 8080,
|
||||
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b0010, ShardSizes: []int64{20}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbBTakeover)
|
||||
if len(hbBTakeover.NewEcShards) != 1 || hbBTakeover.NewEcShards[0].EcIndexBits != 0b0010 {
|
||||
t.Fatalf("peer B should pick up shard 1 after A's snapshot diff, got %+v", hbBTakeover.NewEcShards)
|
||||
}
|
||||
|
||||
// Bits 0 and 3 must still be locked to A; peer B cannot hijack them.
|
||||
hbBHijack := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.2",
|
||||
Port: 8080,
|
||||
NewEcShards: []*master_pb.VolumeEcShardInformationMessage{
|
||||
{Id: 5, EcIndexBits: 0b1001, ShardSizes: []int64{10, 30}},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbBHijack)
|
||||
if len(hbBHijack.NewEcShards) != 0 {
|
||||
t.Fatalf("peer B must not steal bits A still owns, got %+v", hbBHijack.NewEcShards)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFilterVolumeOwnership_FullSnapshotDropsStaleVolumes covers the same
|
||||
// pattern for non-EC volumes: when peer A no longer lists a vid in a full
|
||||
// snapshot, the binding must release so another peer can take over.
|
||||
func TestFilterVolumeOwnership_FullSnapshotDropsStaleVolumes(t *testing.T) {
|
||||
ms := &MasterServer{
|
||||
option: &MasterOption{},
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
topo := topology.NewTopology("t", sequence.NewMemorySequencer(), 32*1024, 5, false)
|
||||
dc := topo.GetOrCreateDataCenter("dc")
|
||||
rack := dc.GetOrCreateRack("rack")
|
||||
dnA := rack.GetOrCreateDataNode("10.0.0.1", 8080, 8081, "10.0.0.1", "A", map[string]uint32{"": 10})
|
||||
dnB := rack.GetOrCreateDataNode("10.0.0.2", 8080, 8081, "10.0.0.2", "B", map[string]uint32{"": 10})
|
||||
|
||||
hbAFull := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
Volumes: []*master_pb.VolumeInformationMessage{
|
||||
{Id: 7, ReplicaPlacement: 0},
|
||||
{Id: 8, ReplicaPlacement: 0},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbAFull)
|
||||
|
||||
// Snapshot drops vid 7.
|
||||
hbAShrink := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.1",
|
||||
Port: 8080,
|
||||
Volumes: []*master_pb.VolumeInformationMessage{
|
||||
{Id: 8, ReplicaPlacement: 0},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnA, hbAShrink)
|
||||
|
||||
// Peer B claims vid 7; the binding has been released by the diff.
|
||||
hbB := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.2",
|
||||
Port: 8080,
|
||||
Volumes: []*master_pb.VolumeInformationMessage{
|
||||
{Id: 7, ReplicaPlacement: 0},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbB)
|
||||
if len(hbB.Volumes) != 1 {
|
||||
t.Fatalf("peer B should pick up vid 7 after A's snapshot diff, got %+v", hbB.Volumes)
|
||||
}
|
||||
|
||||
// Peer B cannot hijack vid 8 — A still owns it.
|
||||
hbBHijack := &master_pb.Heartbeat{
|
||||
Ip: "10.0.0.2",
|
||||
Port: 8080,
|
||||
Volumes: []*master_pb.VolumeInformationMessage{
|
||||
{Id: 8, ReplicaPlacement: 0},
|
||||
},
|
||||
}
|
||||
ms.filterVolumeOwnership(dnB, hbBHijack)
|
||||
if len(hbBHijack.Volumes) != 0 {
|
||||
t.Fatalf("peer B must not hijack vid 8 while A owns it, got %+v", hbBHijack.Volumes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHeartbeatOwnership_ReleaseEcShardsIgnoresForeign confirms that a
|
||||
// malicious peer cannot use a forged DeletedEcShards entry to evict the
|
||||
// legitimate owner of a shard slot.
|
||||
func TestHeartbeatOwnership_ReleaseEcShardsIgnoresForeign(t *testing.T) {
|
||||
o := newHeartbeatOwnership()
|
||||
|
||||
if got := o.claimEcShards(11, 0b1, "10.0.0.1", 8080); got != 0b1 {
|
||||
t.Fatalf("peer A must own shard 0")
|
||||
}
|
||||
|
||||
// Attacker tries to release A's binding.
|
||||
o.releaseEcShards(11, 0b1, "10.6.6.6", 8080)
|
||||
|
||||
// Peer C still cannot claim it.
|
||||
if got := o.claimEcShards(11, 0b1, "10.0.0.3", 8080); got != 0 {
|
||||
t.Fatalf("foreign release must be a no-op, got 0x%x", got)
|
||||
}
|
||||
|
||||
// Owner can still release its own binding.
|
||||
o.releaseEcShards(11, 0b1, "10.0.0.1", 8080)
|
||||
if got := o.claimEcShards(11, 0b1, "10.0.0.3", 8080); got != 0b1 {
|
||||
t.Fatalf("after owner release, slot must be reclaimable, got 0x%x", got)
|
||||
}
|
||||
}
|
||||
@ -61,6 +61,10 @@ type MasterOption struct {
|
||||
TelemetryUrl string
|
||||
TelemetryEnabled bool
|
||||
VolumeGrowthDisabled bool
|
||||
// AllowUntrustedHeartbeat skips the source-ip check that binds a
|
||||
// heartbeat's claimed Ip to the gRPC peer address. Operators with a
|
||||
// NAT/proxy between volume servers and the master must enable this.
|
||||
AllowUntrustedHeartbeat bool
|
||||
}
|
||||
|
||||
type MasterServer struct {
|
||||
@ -90,6 +94,11 @@ type MasterServer struct {
|
||||
|
||||
LockRingManager *cluster.LockRingManager
|
||||
|
||||
// heartbeatOwnership binds volume / ec shard ids to the peer that first
|
||||
// claimed them in a heartbeat, so a later attacker cannot hijack lookups
|
||||
// by re-announcing the same ids from a different host.
|
||||
heartbeatOwnership *heartbeatOwnership
|
||||
|
||||
// telemetry
|
||||
telemetryCollector *telemetry.Collector
|
||||
}
|
||||
@ -138,6 +147,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
|
||||
MasterClient: wdclient.NewMasterClient(grpcDialOption, "", cluster.MasterType, option.Master, "", "", *pb.NewServiceDiscoveryFromMap(peers)),
|
||||
adminLocks: NewAdminLocks(),
|
||||
Cluster: cluster.NewCluster(),
|
||||
heartbeatOwnership: newHeartbeatOwnership(),
|
||||
}
|
||||
|
||||
ms.LockRingManager = cluster.NewLockRingManager(ms.broadcastToClients)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
