filer: let a nested path rule turn worm off (#10503)

* filer: let a nested path rule turn worm off

mergePathConf ORs the booleans, so worm set on a bucket could never be
lifted on a directory under it, while every string field is overridden by
the more specific rule. Make worm tri-state instead: unset inherits, set
wins. readOnly, fsync and disableChunkDeletion keep the OR, so a nested
rule still cannot escape a lock the bucket set.

Configurations written before this carry an explicit "worm": false on
every rule, because they are marshalled with EmitUnpopulated. Reading
those back as an override would quietly drop worm from nested paths, so
filer.conf is now stamped with a version and the flag is dropped to unset
when the version predates it.

* filer: copy the worm value out of the matched rule

mergePathConf aliased the pointer into the merged result, so a caller that
wrote through it would reach into the stored rule.
This commit is contained in:
Chris Lu
2026-07-31 00:34:22 -07:00
committed by GitHub
parent 4dc1b70b2f
commit 3514925581
9 changed files with 146 additions and 43 deletions
+2 -1
View File
@@ -761,7 +761,8 @@ message FilerConf {
string data_node = 11;
uint32 max_file_name_length = 12;
bool disable_chunk_deletion = 13;
bool worm = 14;
// unset inherits from the enclosing path rule, set overrides it
optional bool worm = 14;
uint64 worm_grace_period_seconds = 15;
uint64 worm_retention_time_seconds = 16;
}
+23 -3
View File
@@ -15,6 +15,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/viant/ptrie"
jsonpb "google.golang.org/protobuf/encoding/protojson"
"google.golang.org/protobuf/proto"
)
const (
@@ -27,6 +28,12 @@ const (
IamPoliciesFile = "policies.json"
)
// FilerConfVersion is stamped into every configuration this build writes.
// Version 0 predates worm presence: it was written with EmitUnpopulated, so every
// rule carries an explicit "worm": false that meant nothing. Reading one back as an
// override would silently lift worm off nested paths, so it is dropped to unset.
const FilerConfVersion = 1
type FilerConf struct {
rules ptrie.Trie[*filer_pb.FilerConf_PathConf]
}
@@ -115,6 +122,9 @@ func (fc *FilerConf) LoadFromBytes(data []byte) (err error) {
func (fc *FilerConf) doLoadConf(conf *filer_pb.FilerConf) (err error) {
for _, location := range conf.Locations {
if conf.Version < FilerConfVersion && location.Worm != nil && !*location.Worm {
location.Worm = nil
}
err = fc.SetLocationConf(location)
if err != nil {
// this is not recoverable
@@ -213,6 +223,10 @@ func ClonePathConf(src *filer_pb.FilerConf_PathConf) *filer_pb.FilerConf_PathCon
if src == nil {
return &filer_pb.FilerConf_PathConf{}
}
var worm *bool
if src.Worm != nil {
worm = proto.Bool(*src.Worm)
}
return &filer_pb.FilerConf_PathConf{
LocationPrefix: src.LocationPrefix,
Collection: src.Collection,
@@ -227,7 +241,7 @@ func ClonePathConf(src *filer_pb.FilerConf_PathConf) *filer_pb.FilerConf_PathCon
Rack: src.Rack,
DataNode: src.DataNode,
DisableChunkDeletion: src.DisableChunkDeletion,
Worm: src.Worm,
Worm: worm,
WormGracePeriodSeconds: src.WormGracePeriodSeconds,
WormRetentionTimeSeconds: src.WormRetentionTimeSeconds,
}
@@ -334,7 +348,13 @@ func mergePathConf(a, b *filer_pb.FilerConf_PathConf) {
a.Rack = util.Nvl(b.Rack, a.Rack)
a.DataNode = util.Nvl(b.DataNode, a.DataNode)
a.DisableChunkDeletion = b.DisableChunkDeletion || a.DisableChunkDeletion
a.Worm = b.Worm || a.Worm
// worm merges on presence, so a nested rule can turn it off. readOnly, fsync and
// disableChunkDeletion stay OR'ed on purpose: a nested rule must not be able to
// lift a lock the bucket set.
if b.Worm != nil {
// copy the value: a is often a scratch conf while b is a live trie entry
a.Worm = proto.Bool(*b.Worm)
}
if b.WormRetentionTimeSeconds > 0 {
a.WormRetentionTimeSeconds = b.WormRetentionTimeSeconds
}
@@ -344,7 +364,7 @@ func mergePathConf(a, b *filer_pb.FilerConf_PathConf) {
}
func (fc *FilerConf) ToProto() *filer_pb.FilerConf {
m := &filer_pb.FilerConf{}
m := &filer_pb.FilerConf{Version: FilerConfVersion}
fc.rules.Walk(func(key []byte, value *filer_pb.FilerConf_PathConf) bool {
m.Locations = append(m.Locations, value)
return true
+67 -1
View File
@@ -1,11 +1,13 @@
package filer
import (
"bytes"
"reflect"
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/stretchr/testify/assert"
"google.golang.org/protobuf/proto"
)
func TestFilerConf(t *testing.T) {
@@ -49,6 +51,68 @@ func TestFilerConf(t *testing.T) {
}
func TestWormInheritance(t *testing.T) {
fc := NewFilerConf()
fc.doLoadConf(&filer_pb.FilerConf{
Version: FilerConfVersion,
Locations: []*filer_pb.FilerConf_PathConf{
{LocationPrefix: "/buckets/b/", Worm: proto.Bool(true), Ttl: "7d"},
{LocationPrefix: "/buckets/b/quiet/", Collection: "quiet"},
{LocationPrefix: "/buckets/b/scratch/", Worm: proto.Bool(false)},
{LocationPrefix: "/buckets/b/scratch/keep/", Worm: proto.Bool(true)},
},
})
// a rule that says nothing about worm keeps inheriting it, along with the ttl
rule := fc.MatchStorageRule("/buckets/b/quiet/x")
assert.True(t, rule.GetWorm())
assert.Equal(t, "7d", rule.Ttl)
// an explicit false turns it off, and a deeper rule turns it back on
assert.False(t, fc.MatchStorageRule("/buckets/b/scratch/x").GetWorm())
assert.True(t, fc.MatchStorageRule("/buckets/b/scratch/keep/x").GetWorm())
// paths with no rule at all are unaffected
assert.False(t, fc.MatchStorageRule("/buckets/other/x").GetWorm())
}
// TestWormLegacyFalseIsNotAnOverride pins that the explicit "worm": false every
// version 0 configuration carries does not read as a per-path opt-out.
func TestWormLegacyFalseIsNotAnOverride(t *testing.T) {
const conf = `{
"locations": [
{"locationPrefix": "/buckets/b/", "worm": true},
{"locationPrefix": "/buckets/b/sub/", "collection": "sub", "worm": false}
]
}`
fc := NewFilerConf()
assert.NoError(t, fc.LoadFromBytes([]byte(conf)))
assert.True(t, fc.MatchStorageRule("/buckets/b/sub/x").GetWorm())
// the same file at the current version means what it says
fc = NewFilerConf()
assert.NoError(t, fc.LoadFromBytes([]byte(`{"version": 1,`+conf[1:])))
assert.False(t, fc.MatchStorageRule("/buckets/b/sub/x").GetWorm())
}
// TestWormSurvivesRoundTrip guards the write side: an explicit false has to be
// stamped along with a version that says to honor it.
func TestWormSurvivesRoundTrip(t *testing.T) {
fc := NewFilerConf()
fc.SetLocationConf(&filer_pb.FilerConf_PathConf{LocationPrefix: "/buckets/b/", Worm: proto.Bool(true)})
fc.SetLocationConf(&filer_pb.FilerConf_PathConf{LocationPrefix: "/buckets/b/sub/", Worm: proto.Bool(false)})
fc.SetLocationConf(&filer_pb.FilerConf_PathConf{LocationPrefix: "/buckets/b/other/", Ttl: "7d"})
var buf bytes.Buffer
assert.NoError(t, fc.ToText(&buf))
reloaded := NewFilerConf()
assert.NoError(t, reloaded.LoadFromBytes(buf.Bytes()))
assert.False(t, reloaded.MatchStorageRule("/buckets/b/sub/x").GetWorm())
assert.True(t, reloaded.MatchStorageRule("/buckets/b/other/x").GetWorm())
}
// TestClonePathConf verifies that ClonePathConf copies all exported fields.
// Uses reflection to automatically detect new fields added to the protobuf,
// ensuring the test fails if ClonePathConf is not updated for new fields.
@@ -68,7 +132,7 @@ func TestClonePathConf(t *testing.T) {
Rack: "rack1",
DataNode: "node1",
DisableChunkDeletion: true,
Worm: true,
Worm: proto.Bool(true),
WormGracePeriodSeconds: 3600,
WormRetentionTimeSeconds: 86400,
}
@@ -119,8 +183,10 @@ func TestClonePathConf(t *testing.T) {
// Verify mutation of clone doesn't affect source
clone.Collection = "modified"
clone.ReadOnly = false
*clone.Worm = false
assert.Equal(t, "test_collection", src.Collection, "Modifying clone should not affect source Collection")
assert.Equal(t, true, src.ReadOnly, "Modifying clone should not affect source ReadOnly")
assert.Equal(t, true, src.GetWorm(), "Modifying clone should not affect source Worm")
}
func TestClonePathConfNil(t *testing.T) {
+1 -1
View File
@@ -82,7 +82,7 @@ func (wfs *WFS) wormEnforcedForEntry(path util.FullPath, entry *filer_pb.Entry)
}
rule := wfs.FilerConf.MatchStorageRule(string(path))
if !rule.Worm {
if !rule.GetWorm() {
return false, false
}
+2 -1
View File
@@ -761,7 +761,8 @@ message FilerConf {
string data_node = 11;
uint32 max_file_name_length = 12;
bool disable_chunk_deletion = 13;
bool worm = 14;
// unset inherits from the enclosing path rule, set overrides it
optional bool worm = 14;
uint64 worm_grace_period_seconds = 15;
uint64 worm_retention_time_seconds = 16;
}
+27 -24
View File
@@ -6764,23 +6764,24 @@ func (x *LocateBrokerResponse_Resource) GetResourceCount() int32 {
}
type FilerConf_PathConf struct {
state protoimpl.MessageState `protogen:"open.v1"`
LocationPrefix string `protobuf:"bytes,1,opt,name=location_prefix,json=locationPrefix,proto3" json:"location_prefix,omitempty"`
Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
Replication string `protobuf:"bytes,3,opt,name=replication,proto3" json:"replication,omitempty"`
Ttl string `protobuf:"bytes,4,opt,name=ttl,proto3" json:"ttl,omitempty"`
DiskType string `protobuf:"bytes,5,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
Fsync bool `protobuf:"varint,6,opt,name=fsync,proto3" json:"fsync,omitempty"`
VolumeGrowthCount uint32 `protobuf:"varint,7,opt,name=volume_growth_count,json=volumeGrowthCount,proto3" json:"volume_growth_count,omitempty"`
ReadOnly bool `protobuf:"varint,8,opt,name=read_only,json=readOnly,proto3" json:"read_only,omitempty"`
DataCenter string `protobuf:"bytes,9,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
Rack string `protobuf:"bytes,10,opt,name=rack,proto3" json:"rack,omitempty"`
DataNode string `protobuf:"bytes,11,opt,name=data_node,json=dataNode,proto3" json:"data_node,omitempty"`
MaxFileNameLength uint32 `protobuf:"varint,12,opt,name=max_file_name_length,json=maxFileNameLength,proto3" json:"max_file_name_length,omitempty"`
DisableChunkDeletion bool `protobuf:"varint,13,opt,name=disable_chunk_deletion,json=disableChunkDeletion,proto3" json:"disable_chunk_deletion,omitempty"`
Worm bool `protobuf:"varint,14,opt,name=worm,proto3" json:"worm,omitempty"`
WormGracePeriodSeconds uint64 `protobuf:"varint,15,opt,name=worm_grace_period_seconds,json=wormGracePeriodSeconds,proto3" json:"worm_grace_period_seconds,omitempty"`
WormRetentionTimeSeconds uint64 `protobuf:"varint,16,opt,name=worm_retention_time_seconds,json=wormRetentionTimeSeconds,proto3" json:"worm_retention_time_seconds,omitempty"`
state protoimpl.MessageState `protogen:"open.v1"`
LocationPrefix string `protobuf:"bytes,1,opt,name=location_prefix,json=locationPrefix,proto3" json:"location_prefix,omitempty"`
Collection string `protobuf:"bytes,2,opt,name=collection,proto3" json:"collection,omitempty"`
Replication string `protobuf:"bytes,3,opt,name=replication,proto3" json:"replication,omitempty"`
Ttl string `protobuf:"bytes,4,opt,name=ttl,proto3" json:"ttl,omitempty"`
DiskType string `protobuf:"bytes,5,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
Fsync bool `protobuf:"varint,6,opt,name=fsync,proto3" json:"fsync,omitempty"`
VolumeGrowthCount uint32 `protobuf:"varint,7,opt,name=volume_growth_count,json=volumeGrowthCount,proto3" json:"volume_growth_count,omitempty"`
ReadOnly bool `protobuf:"varint,8,opt,name=read_only,json=readOnly,proto3" json:"read_only,omitempty"`
DataCenter string `protobuf:"bytes,9,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
Rack string `protobuf:"bytes,10,opt,name=rack,proto3" json:"rack,omitempty"`
DataNode string `protobuf:"bytes,11,opt,name=data_node,json=dataNode,proto3" json:"data_node,omitempty"`
MaxFileNameLength uint32 `protobuf:"varint,12,opt,name=max_file_name_length,json=maxFileNameLength,proto3" json:"max_file_name_length,omitempty"`
DisableChunkDeletion bool `protobuf:"varint,13,opt,name=disable_chunk_deletion,json=disableChunkDeletion,proto3" json:"disable_chunk_deletion,omitempty"`
// unset inherits from the enclosing path rule, set overrides it
Worm *bool `protobuf:"varint,14,opt,name=worm,proto3,oneof" json:"worm,omitempty"`
WormGracePeriodSeconds uint64 `protobuf:"varint,15,opt,name=worm_grace_period_seconds,json=wormGracePeriodSeconds,proto3" json:"worm_grace_period_seconds,omitempty"`
WormRetentionTimeSeconds uint64 `protobuf:"varint,16,opt,name=worm_retention_time_seconds,json=wormRetentionTimeSeconds,proto3" json:"worm_retention_time_seconds,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -6907,8 +6908,8 @@ func (x *FilerConf_PathConf) GetDisableChunkDeletion() bool {
}
func (x *FilerConf_PathConf) GetWorm() bool {
if x != nil {
return x.Worm
if x != nil && x.Worm != nil {
return *x.Worm
}
return false
}
@@ -7403,10 +7404,10 @@ const file_filer_proto_rawDesc = "" +
"\x03key\x18\x01 \x01(\fR\x03key\x12\x14\n" +
"\x05value\x18\x02 \x01(\fR\x05value\"%\n" +
"\rKvPutResponse\x12\x14\n" +
"\x05error\x18\x01 \x01(\tR\x05error\"\xb2\x05\n" +
"\x05error\x18\x01 \x01(\tR\x05error\"\xc0\x05\n" +
"\tFilerConf\x12\x18\n" +
"\aversion\x18\x01 \x01(\x05R\aversion\x12:\n" +
"\tlocations\x18\x02 \x03(\v2\x1c.filer_pb.FilerConf.PathConfR\tlocations\x1a\xce\x04\n" +
"\tlocations\x18\x02 \x03(\v2\x1c.filer_pb.FilerConf.PathConfR\tlocations\x1a\xdc\x04\n" +
"\bPathConf\x12'\n" +
"\x0flocation_prefix\x18\x01 \x01(\tR\x0elocationPrefix\x12\x1e\n" +
"\n" +
@@ -7424,10 +7425,11 @@ const file_filer_proto_rawDesc = "" +
" \x01(\tR\x04rack\x12\x1b\n" +
"\tdata_node\x18\v \x01(\tR\bdataNode\x12/\n" +
"\x14max_file_name_length\x18\f \x01(\rR\x11maxFileNameLength\x124\n" +
"\x16disable_chunk_deletion\x18\r \x01(\bR\x14disableChunkDeletion\x12\x12\n" +
"\x04worm\x18\x0e \x01(\bR\x04worm\x129\n" +
"\x16disable_chunk_deletion\x18\r \x01(\bR\x14disableChunkDeletion\x12\x17\n" +
"\x04worm\x18\x0e \x01(\bH\x00R\x04worm\x88\x01\x01\x129\n" +
"\x19worm_grace_period_seconds\x18\x0f \x01(\x04R\x16wormGracePeriodSeconds\x12=\n" +
"\x1bworm_retention_time_seconds\x18\x10 \x01(\x04R\x18wormRetentionTimeSeconds\"\xba\x01\n" +
"\x1bworm_retention_time_seconds\x18\x10 \x01(\x04R\x18wormRetentionTimeSecondsB\a\n" +
"\x05_worm\"\xba\x01\n" +
"&CacheRemoteObjectToLocalClusterRequest\x12\x1c\n" +
"\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x12\n" +
"\x04name\x18\x02 \x01(\tR\x04name\x12+\n" +
@@ -7876,6 +7878,7 @@ func file_filer_proto_init() {
(*StreamMutateEntryResponse_DeleteResponse)(nil),
(*StreamMutateEntryResponse_RenameResponse)(nil),
}
file_filer_proto_msgTypes[98].OneofWrappers = []any{}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
+5 -4
View File
@@ -4814,9 +4814,9 @@ func (m *FilerConf_PathConf) MarshalToSizedBufferVT(dAtA []byte) (int, error) {
i--
dAtA[i] = 0x78
}
if m.Worm {
if m.Worm != nil {
i--
if m.Worm {
if *m.Worm {
dAtA[i] = 1
} else {
dAtA[i] = 0
@@ -8166,7 +8166,7 @@ func (m *FilerConf_PathConf) SizeVT() (n int) {
if m.DisableChunkDeletion {
n += 2
}
if m.Worm {
if m.Worm != nil {
n += 2
}
if m.WormGracePeriodSeconds != 0 {
@@ -21715,7 +21715,8 @@ func (m *FilerConf_PathConf) UnmarshalVT(dAtA []byte) error {
break
}
}
m.Worm = bool(v != 0)
b := bool(v != 0)
m.Worm = &b
case 15:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field WormGracePeriodSeconds", wireType)
@@ -177,7 +177,7 @@ func (fs *FilerServer) checkPermissions(ctx context.Context, r *http.Request, fi
func (fs *FilerServer) wormEnforcedForEntry(ctx context.Context, fullPath string) (bool, error) {
rule := fs.filer.FilerConf.MatchStorageRule(fullPath)
if !rule.Worm {
if !rule.GetWorm() {
return false, nil
}
+18 -7
View File
@@ -12,6 +12,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
"google.golang.org/protobuf/proto"
)
func init() {
@@ -44,6 +45,10 @@ func (c *commandFsConfigure) Help() string {
# example: unlock a bucket that quota enforcement made read-only
fs.configure -locationPrefix=/buckets/my_bucket/ -readOnly=false -apply
# example: keep one directory writable under a worm-protected tree
fs.configure -locationPrefix=/buckets/my_bucket/ -worm -apply
fs.configure -locationPrefix=/buckets/my_bucket/scratch/ -worm=false -apply
# delete the changes
fs.configure -locationPrefix=/my/folder -delete -apply
@@ -64,7 +69,7 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
diskType := fsConfigureCommand.String("disk", "", "[hdd|ssd|<tag>] hard drive or solid state drive or any tag")
fsync := fsConfigureCommand.Bool("fsync", false, "fsync for the writes")
isReadOnly := fsConfigureCommand.Bool("readOnly", false, "disable writes")
worm := fsConfigureCommand.Bool("worm", false, "write-once-read-many, written files are readonly")
worm := fsConfigureCommand.Bool("worm", false, "write-once-read-many, written files are readonly; unset inherits from the parent path")
wormGracePeriod := fsConfigureCommand.Uint64("wormGracePeriod", 0, "grace period before worm is enforced, in seconds")
wormRetentionTime := fsConfigureCommand.Uint64("wormRetentionTime", 0, "retention time for a worm enforced file, in seconds")
maxFileNameLength := fsConfigureCommand.Uint("maxFileNameLength", 0, "file name length limits in bytes for compatibility with Unix-based systems")
@@ -98,11 +103,18 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
DataCenter: *dataCenter,
Rack: *rack,
DataNode: *dataNode,
Worm: *worm,
WormGracePeriodSeconds: *wormGracePeriod,
WormRetentionTimeSeconds: *wormRetentionTime,
}
// worm is only carried when the flag is passed, so a rule that says nothing
// about it keeps inheriting from the enclosing path
fsConfigureCommand.Visit(func(f *flag.Flag) {
if f.Name == "worm" {
locConf.Worm = proto.Bool(*worm)
}
})
// check collection
if *collection != "" && strings.HasPrefix(*locationPrefix, "/buckets/") {
return fmt.Errorf("one s3 bucket goes to one collection and not customizable")
@@ -134,9 +146,10 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
fc.DeleteLocationConf(*locationPrefix)
} else {
fc.AddLocationConf(locConf)
// AddLocationConf merges boolean fields with OR, which can never turn
// a flag off; let an explicitly passed false win, e.g. -readOnly=false
// to reopen a bucket that quota enforcement locked
// AddLocationConf merges these boolean fields with OR, which can never
// turn a flag off; let an explicitly passed false win, e.g. -readOnly=false
// to reopen a bucket that quota enforcement locked. worm does not belong
// here: it merges on presence, so the value set above already wins.
if mergedConf, found := fc.GetLocationConf(*locationPrefix); found {
fsConfigureCommand.Visit(func(f *flag.Flag) {
switch f.Name {
@@ -144,8 +157,6 @@ func (c *commandFsConfigure) Do(args []string, commandEnv *CommandEnv, writer io
mergedConf.ReadOnly = *isReadOnly
case "fsync":
mergedConf.Fsync = *fsync
case "worm":
mergedConf.Worm = *worm
}
})
}