eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Redact invalid chunk plan lookup diagnostics

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-18 01:59:10 +00:00
parent b3328b93e5
commit 8c59b6af14
6 changed files with 123 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
internal/framework/chunkplan/store.go
View File

@@ -40,7 +40,7 @@ func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, p
data, err := os.ReadFile(target)
if err != nil {
if os.IsNotExist(err) {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: "chunk plan not found"}, nil
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanMissing, Reason: lookupReason(pipeline.ChunkPlanMissing)}, nil
}
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{}, fmt.Errorf("read chunk plan: %w", err)
}
@@ -50,19 +50,16 @@ func (s *filesystemStore) Load(sourceDigest string) (pipeline.ChunkPlanRecord, p
decoder.DisallowUnknownFields()
decoder.UseNumber()
if err := decoder.Decode(&record); err != nil {
return invalidDecision(fmt.Sprintf("decode stored chunk plan: %v", err))
return invalidDecision()
}
var trailing any
if err := decoder.Decode(&trailing); err != io.EOF {
if err == nil {
return invalidDecision("stored chunk plan contains trailing JSON")
}
return invalidDecision(fmt.Sprintf("decode stored chunk plan trailer: %v", err))
return invalidDecision()
}
if err := validateRecord(record, sourceDigest); err != nil {
return invalidDecision(err.Error())
return invalidDecision()
}
return record, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanHit, Reason: "stored chunk plan is valid"}, nil
return record, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanHit, Reason: lookupReason(pipeline.ChunkPlanHit)}, nil
}
func (s *filesystemStore) Save(record pipeline.ChunkPlanRecord) error {
@@ -165,8 +162,21 @@ func validateRecord(record pipeline.ChunkPlanRecord, requestedDigest string) err
return nil
}
func invalidDecision(reason string) (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: reason}, nil
func invalidDecision() (pipeline.ChunkPlanRecord, pipeline.ChunkPlanDecision, error) {
return pipeline.ChunkPlanRecord{}, pipeline.ChunkPlanDecision{Status: pipeline.ChunkPlanInvalid, Reason: lookupReason(pipeline.ChunkPlanInvalid)}, nil
}
func lookupReason(status pipeline.ChunkPlanStatus) string {
switch status {
case pipeline.ChunkPlanHit:
return "stored chunk plan is valid"
case pipeline.ChunkPlanMissing:
return "chunk plan not found"
case pipeline.ChunkPlanInvalid:
return "stored chunk plan is invalid"
default:
return "chunk plan lookup skipped"
}
}
type atomicWriteHooks struct {

29
internal/framework/chunkplan/store_test.go
View File

@@ -122,14 +122,13 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) {
tests := []struct {
name string
mutate func([]byte) []byte
want string
}{
{name: "unknown field", mutate: func(data []byte) []byte {
return bytes.Replace(data, []byte(`{"schema_version"`), []byte(`{"unknown":true,"schema_version"`), 1)
}, want: "unknown"},
{name: "truncated JSON", mutate: func(data []byte) []byte { return data[:len(data)/2] }, want: "decode"},
{name: "schema mismatch", mutate: replaceJSON(`notarius.chunk-plan.v1`, `notarius.chunk-plan.v2`), want: "schema_version"},
{name: "source mismatch", mutate: replaceJSON(testSourceDigest, "sha256:"+strings.Repeat("b", 64)), want: "source_digest"},
return bytes.Replace(data, []byte(`{"schema_version"`), []byte(`{"SENTINEL_UNKNOWN_FIELD":true,"schema_version"`), 1)
}},
{name: "truncated JSON", mutate: func(data []byte) []byte { return data[:len(data)/2] }},
{name: "schema mismatch", mutate: replaceJSON(`notarius.chunk-plan.v1`, `SENTINEL_SCHEMA_VALUE`)},
{name: "source mismatch", mutate: replaceJSON(testSourceDigest, "sha256:"+strings.Repeat("b", 64))},
{name: "plan digest mismatch", mutate: func(data []byte) []byte {
prefix := []byte(`"plan_digest":"sha256:`)
index := bytes.Index(data, prefix)
@@ -137,14 +136,15 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) {
data[index+len(prefix)] = '0'
}
return data
}, want: "plan_digest"},
}},
{name: "noncanonical annotation", mutate: func(data []byte) []byte {
return bytes.Replace(data, []byte(`{"value":1}`), []byte(`{ "value": 1 }`), 1)
}, want: "canonical JSON"},
return bytes.Replace(data, []byte(`"test/value"`), []byte(`"SENTINEL_ANNOTATION_NAMESPACE"`), 1)
}},
{name: "bad boundary", mutate: func(data []byte) []byte {
return bytes.Replace(data, []byte(`"start_unit_id":1`), []byte(`"start_unit_id":0`), 1)
}, want: "boundaries must be positive"},
{name: "trailing JSON", mutate: func(data []byte) []byte { return append(data, []byte(` {}`)...) }, want: "trailing"},
}},
{name: "timestamp", mutate: replaceJSON(`2026-07-18T12:00:00Z`, `SENTINEL_TIMESTAMP`)},
{name: "trailing JSON", mutate: func(data []byte) []byte { return append(data, []byte(` {"SENTINEL_TRAILING":true}`)...) }},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
@@ -163,9 +163,14 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) {
t.Fatal(err)
}
got, decision, err := store.Load(testSourceDigest)
if err != nil || decision.Status != pipeline.ChunkPlanInvalid || !reflect.DeepEqual(got, pipeline.ChunkPlanRecord{}) || !strings.Contains(decision.Reason, tc.want) {
if err != nil || decision.Status != pipeline.ChunkPlanInvalid || !reflect.DeepEqual(got, pipeline.ChunkPlanRecord{}) || decision.Reason != "stored chunk plan is invalid" {
t.Fatalf("record=%#v decision=%#v error=%v", got, decision, err)
}
for _, sentinel := range []string{"SENTINEL_UNKNOWN_FIELD", "SENTINEL_SCHEMA_VALUE", "SENTINEL_ANNOTATION_NAMESPACE", "SENTINEL_TIMESTAMP", "SENTINEL_TRAILING"} {
if strings.Contains(decision.Reason, sentinel) {
t.Fatalf("decision leaked %q: %#v", sentinel, decision)
}
}
})
}
}