diff --git a/cmd/fmsg-backfill/migrate.go b/cmd/fmsg-backfill/migrate.go index 4c490f6..a3de0e8 100644 --- a/cmd/fmsg-backfill/migrate.go +++ b/cmd/fmsg-backfill/migrate.go @@ -39,7 +39,7 @@ type oldBatch struct { } func address(s string) (fmsg.Address, error) { - p := strings.Split(s, "@") + p := strings.SplitN(s, "@", 3) if len(p) != 3 || p[0] != "" || p[1] == "" || p[2] == "" || len(s) > 255 { return fmsg.Address{}, fmt.Errorf("invalid stored address %q", s) } @@ -176,14 +176,19 @@ func (m *migration) message(id int64) error { return fmt.Errorf("parent %d: %w", *s.pid, err) } if len(s.h.Pid) == 0 { - if len(s.hash) > 0 { - return fmt.Errorf("hashed reply has no parent hash; cannot change its identity") - } if err = m.tx.QueryRow(`SELECT sha256 FROM msg WHERE id=$1`, *s.pid).Scan(&s.h.Pid); err != nil { return err } } } + // Old receivers sometimes kept wire sizes beside already expanded files. + // For published identities the full hash remains the authority: use actual + // expanded lengths when reconstructing, and only persist them after verification. + if len(s.hash) == 32 && len(s.prepared) == 0 { + if err = expandedSizes(s.h); err != nil { + return err + } + } batches, err := m.batches(id) if err != nil { return err @@ -231,7 +236,7 @@ func (m *migration) message(id int64) error { if len(s.hash) == 0 && !strings.EqualFold(s.h.From.Domain, m.domain) { return fmt.Errorf("remote message has no published hash or wire header") } - base, err = m.reconstruct(s.h, s.hash) + base, err = m.reconstruct(s.h, s.hash, batches) } } if err != nil { @@ -259,6 +264,32 @@ func (m *migration) message(id int64) error { if base == nil || len(s.hash) != 32 { return fmt.Errorf("missing canonical identity or payload representation") } + // Early receivers stamped the arrival time on the batch row. An exact + // retained wire header supplies the sending timestamp. Only repair an + // unambiguous, unhashed batch; existing batch identities remain authoritative. + if received != nil && received.Flags&fmsg.FlagHasAddTo != 0 { + candidate := -1 + matches := 0 + exact := false + for i, b := range batches { + h := batchHeader(base, s.hash, b) + if bytes.Equal(h.Encode(), received.Encode()) { + exact = true + break + } + h.Timestamp = received.Timestamp + if len(b.hash) == 0 && bytes.Equal(h.Encode(), received.Encode()) { + candidate = i + matches++ + } + } + if !exact && matches == 1 { + batches[candidate].time = received.Timestamp + if _, err = m.tx.Exec(`UPDATE msg_add_to_batch SET time_added=$2 WHERE id=$1`, batches[candidate].id, received.Timestamp); err != nil { + return err + } + } + } matchedReceived := received == nil || received.Flags&fmsg.FlagHasAddTo == 0 for _, b := range batches { h := batchHeader(base, s.hash, b) @@ -268,10 +299,6 @@ func (m *migration) message(id int64) error { if len(b.prepared) > 0 { h, err = fmsg.UnmarshalPrepared(b.prepared, b.hash) } else { - if len(b.hash) == 0 && !strings.EqualFold(b.from.Domain, m.domain) && - (received == nil || !bytes.Equal(h.Encode(), received.Encode())) { - return fmt.Errorf("remote batch %d has no published hash or exact header", b.id) - } h, err = selectTypes(h, b.hash) } if err != nil { @@ -295,6 +322,14 @@ func (m *migration) message(id int64) error { if !matchedReceived { return fmt.Errorf("received wire header has no matching add-to batch") } + if _, err = m.tx.Exec(`UPDATE msg SET size=$2 WHERE id=$1`, id, s.h.Size); err != nil { + return err + } + for _, a := range s.h.Attachments { + if _, err = m.tx.Exec(`UPDATE msg_attachment SET filesize=$3 WHERE msg_id=$1 AND filename=$2`, id, a.Filename, a.Size); err != nil { + return err + } + } m.done[id] = true return nil } @@ -310,12 +345,20 @@ func batchHeader(base *fmsg.Header, hash []byte, b oldBatch) *fmsg.Header { // Local sends previously selected compression and common types during the // first network delivery. Try those historical forms only in this tool, and // accept a candidate only if it reproduces the entire existing message hash. -func (m *migration) reconstruct(raw *fmsg.Header, expected []byte) (*fmsg.Header, error) { - h, dir, err := fmsg.Prepare(raw) +func (m *migration) reconstruct(raw *fmsg.Header, expected []byte, batches []oldBatch) (*fmsg.Header, error) { + input := raw.Clone() + // Early local notes could have an empty recipient list. Preserve their + // recorded bytes; normal send validation still requires recipients. + if len(input.To) == 0 { + input.To = []fmsg.Address{input.From} + } + h, dir, err := fmsg.Prepare(input) if err != nil { return nil, err } - if chosen, err := selectTypes(h, expected); err == nil { + h = h.Clone() + h.To = raw.To + if chosen, err := selectHistorical(h, expected, batches); err == nil { m.files = append(m.files, dir) return chosen, nil } @@ -324,7 +367,7 @@ func (m *migration) reconstruct(raw *fmsg.Header, expected []byte) (*fmsg.Header if err != nil { return nil, err } - chosen, err := selectTypes(h, expected) + chosen, err := selectHistorical(h, expected, batches) if err != nil { _ = os.RemoveAll(dir) return nil, err @@ -333,6 +376,41 @@ func (m *migration) reconstruct(raw *fmsg.Header, expected []byte) (*fmsg.Header return chosen, nil } +// An early sender could cache a canonical hash after adding batch fields but +// before setting the pid flag. Only retain that form if recorded batch fields +// reproduce the already published hash; never create this encoding for new IDs. +func selectHistorical(h *fmsg.Header, expected []byte, batches []oldBatch) (*fmsg.Header, error) { + chosen, err := selectOriginal(h, expected) + if err == nil || len(expected) != 32 { + return chosen, err + } + for _, b := range batches { + candidate := h.Clone() + candidate.Flags = (candidate.Flags | fmsg.FlagHasAddTo) &^ fmsg.FlagHasPid + candidate.AddToFrom, candidate.AddTo = &b.from, b.to + if chosen, e := selectTypes(candidate, expected); e == nil { + return chosen, nil + } + } + return nil, err +} + +// Some early local writers hashed a root-form header despite retaining a +// relational parent link. Preserve that established identity and local link; +// never select this form for a message without an existing hash to verify. +func selectOriginal(h *fmsg.Header, expected []byte) (*fmsg.Header, error) { + chosen, err := selectTypes(h, expected) + if err == nil { + return chosen, nil + } + if len(expected) == 32 && h.Flags&fmsg.FlagHasPid != 0 && h.Flags&fmsg.FlagHasAddTo == 0 { + root := h.Clone() + root.Flags &^= fmsg.FlagHasPid + return selectTypes(root, expected) + } + return nil, err +} + func selectTypes(h *fmsg.Header, expected []byte) (*fmsg.Header, error) { for _, mode := range []int{0, 1, 2} { c := h.Clone() @@ -388,3 +466,28 @@ func bytesOrNull(b []byte) any { } return b } + +func expandedSizes(h *fmsg.Header) error { + size := func(path string) (uint32, error) { + info, err := os.Stat(path) + if err != nil { + return 0, err + } + if !info.Mode().IsRegular() || info.Size() < 0 || info.Size() > int64(^uint32(0)) { + return 0, fmt.Errorf("invalid payload size: %s", path) + } + return uint32(info.Size()), nil + } + var err error + h.Size, err = size(h.Filepath) + if err != nil { + return err + } + for i := range h.Attachments { + h.Attachments[i].Size, err = size(h.Attachments[i].Filepath) + if err != nil { + return err + } + } + return nil +} diff --git a/cmd/fmsg-backfill/migrate_test.go b/cmd/fmsg-backfill/migrate_test.go index 31b9028..9b2c374 100644 --- a/cmd/fmsg-backfill/migrate_test.go +++ b/cmd/fmsg-backfill/migrate_test.go @@ -326,3 +326,167 @@ func TestDecodeHeaderRejectsTruncation(t *testing.T) { t.Fatal("accepted trailing data") } } + +func TestMigrationExpandedFilesWithOldWireSizes(t *testing.T) { + for _, retainHeader := range []bool{false, true} { + t.Run(fmt.Sprintf("header=%v", retainHeader), func(t *testing.T) { + db := previousStore(t) + raw := rawMessage(t, "example.org", strings.Repeat("body compression ", 200)) + att := rawMessage(t, "example.org", strings.Repeat("attachment compression ", 200)) + raw.Attachments = []fmsg.AttachmentHeader{{Type: "text/plain;charset=UTF-8", Filename: "note.txt", Size: att.Size, Filepath: att.Filepath}} + wire := prepared(t, raw) + published := hashOf(t, wire) + bodySize, attSize := raw.Size, raw.Attachments[0].Size + raw.Size = wire.Size + raw.Attachments[0].Size = wire.Attachments[0].Size + var header []byte + if retainHeader { + header = wire.Encode() + } + id := putOld(t, db, raw, nil, published, header) + if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil { + t.Fatal(err) + } + var gotBody, gotAtt uint32 + var hash, snapshot []byte + if err := db.QueryRow(`SELECT size,sha256,wire_message FROM msg WHERE id=$1`, id).Scan(&gotBody, &hash, &snapshot); err != nil { + t.Fatal(err) + } + if err := db.QueryRow(`SELECT filesize FROM msg_attachment WHERE msg_id=$1`, id).Scan(&gotAtt); err != nil { + t.Fatal(err) + } + if gotBody != bodySize || gotAtt != attSize || !bytes.Equal(hash, published) { + t.Fatal("expanded metadata or published hash changed") + } + if _, err := fmsg.UnmarshalPrepared(snapshot, published); err != nil { + t.Fatal(err) + } + }) + } +} + +func TestMigrationHistoricalLocalLinkAndUnhashedReceivedBatch(t *testing.T) { + db := previousStore(t) + parentRaw := rawMessage(t, "example.com", "parent") + parentHash := hashOf(t, parentRaw) + parent := putOld(t, db, parentRaw, nil, parentHash, nil) + childRaw := rawMessage(t, "example.com", "child") + historicalHash := hashOf(t, childRaw) + child := putOld(t, db, childRaw, parent, historicalHash, nil) + batch := putBatch(t, db, parent, oldBatch{from: fmsg.Address{User: "bob", Domain: "example.org"}, time: 1300, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}}) + if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil { + t.Fatal(err) + } + var pid int64 + var hash, parentSHA, data []byte + if err := db.QueryRow(`SELECT pid,sha256,psha256,wire_message FROM msg WHERE id=$1`, child).Scan(&pid, &hash, &parentSHA, &data); err != nil { + t.Fatal(err) + } + if pid != parent || !bytes.Equal(hash, historicalHash) || !bytes.Equal(parentSHA, parentHash) { + t.Fatal("historical identity or local thread link changed") + } + h, err := fmsg.UnmarshalPrepared(data, historicalHash) + if err != nil { + t.Fatal(err) + } + if h.Flags&fmsg.FlagHasPid != 0 { + t.Fatal("changed historical root-form identity") + } + if err := db.QueryRow(`SELECT sha256,wire_message FROM msg_add_to_batch WHERE id=$1`, batch).Scan(&hash, &data); err != nil { + t.Fatal(err) + } + if _, err := fmsg.UnmarshalPrepared(data, hash); err != nil { + t.Fatal(err) + } +} + +func TestMigrationPreservesEarlyLocalNotesAndLiteralRecipients(t *testing.T) { + db := previousStore(t) + raw := rawMessage(t, "example.com", "local note") + raw.To = nil + hash := hashOf(t, raw) + note := putOld(t, db, raw, nil, hash, nil) + bad := rawMessage(t, "example.com", "literal recipient") + bad.To = []fmsg.Address{{User: "alice", Domain: "example.org,@bob@example.com"}} + literal := putOld(t, db, bad, nil, nil, nil) + child := rawMessage(t, "example.com", "local child") + childHash := hashOf(t, child) + parentRaw := rawMessage(t, "example.com", "unhashed local parent") + parent := putOld(t, db, parentRaw, nil, nil, nil) + reply := putOld(t, db, child, parent, childHash, nil) + if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil { + t.Fatal(err) + } + for _, id := range []int64{note, literal, reply} { + var got, data []byte + if err := db.QueryRow(`SELECT sha256,wire_message FROM msg WHERE id=$1`, id).Scan(&got, &data); err != nil { + t.Fatal(err) + } + h, err := fmsg.UnmarshalPrepared(data, got) + if err != nil { + t.Fatal(err) + } + if id == note && (!bytes.Equal(got, hash) || len(h.To) != 0) { + t.Fatal("local note changed") + } + if id == literal && h.To[0].ToString() != bad.To[0].ToString() { + t.Fatal("literal recipient changed") + } + if id == reply && !bytes.Equal(got, childHash) { + t.Fatal("historical child identity changed") + } + } +} + +func TestMigrationRecoversRecordedBatchWireTime(t *testing.T) { + db := previousStore(t) + raw := rawMessage(t, "example.org", "forwarded") + base := prepared(t, raw) + canonical := hashOf(t, base) + b := oldBatch{from: raw.From, time: 1300, to: []fmsg.Address{{User: "carol", Domain: "example.com"}}} + wire := batchHeader(base, canonical, b) + hash := hashOf(t, wire) + raw.Timestamp = 1300 + id := putOld(t, db, raw, nil, canonical, wire.Encode()) + b.time = 1302 + bid := putBatch(t, db, id, b) + if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil { + t.Fatal(err) + } + var stamp float64 + var got []byte + if err := db.QueryRow(`SELECT time_added,sha256 FROM msg_add_to_batch WHERE id=$1`, bid).Scan(&stamp, &got); err != nil { + t.Fatal(err) + } + if stamp != 1300 || !bytes.Equal(got, hash) { + t.Fatal("recorded wire identity not restored") + } +} + +func TestMigrationPreservesEarlyHashWithBatchFields(t *testing.T) { + db := previousStore(t) + parent := rawMessage(t, "example.com", "parent") + parentID := putOld(t, db, parent, nil, nil, nil) + raw := rawMessage(t, "example.com", "reply") + raw.Topic = "" + b := oldBatch{from: raw.From, to: []fmsg.Address{{User: "carol", Domain: "remote.example"}}, time: raw.Timestamp} + old := raw.Clone() + old.Flags = fmsg.FlagHasAddTo + old.AddToFrom, old.AddTo = &b.from, b.to + hash := hashOf(t, old) + id := putOld(t, db, raw, parentID, hash, nil) + putBatch(t, db, id, b) + if err := migrate(context.Background(), db, "example.com", true, io.Discard); err != nil { + t.Fatal(err) + } + var got, snapshot []byte + if err := db.QueryRow(`SELECT sha256,wire_message FROM msg WHERE id=$1`, id).Scan(&got, &snapshot); err != nil { + t.Fatal(err) + } + if !bytes.Equal(got, hash) { + t.Fatal("changed historical identity") + } + if _, err := fmsg.UnmarshalPrepared(snapshot, hash); err != nil { + t.Fatal(err) + } +}