From 50c8c3f6faf57dca83ef5ff630de12a83b02cae1 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Wed, 2 Sep 2026 13:55:52 +0800 Subject: [PATCH 1/2] =?UTF-8?q?Implement=20terminal=20flag=20(SPEC=20v0.6.?= =?UTF-8?q?0=20=C2=A73=20bit=206)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A terminal message is a leaf of its thread: no message may reference it via pid. Per SPEC v0.6.0: - pkg/fmsg: FlagTerminal (bit 6); only bit 7 remains reserved. - Header validation: an add-to message with terminal set is rejected with code 1 without a lookup, since an add-to copies the original's flags and so signals the original is terminal (§10.3 step 3). - Reply path: a reply whose stored parent is terminal is rejected with code 1 (§10.3 step 7). Add-to path: a batch whose stored parent is terminal is rejected with code 1, catching a sender that stripped the bit. - Store: new msg.is_terminal column persisted from the wire flags and loaded back into every outgoing wire form via baseFlags. - dd.sql: replying to a terminal message (pid trigger) and adding a batch to one (new msg_add_to_batch trigger) raise, so neither fmsgd nor fmsg-webapi can create a unit the Sending Host must not send. - Sender: defensively records code 1 (not retryable) instead of transmitting a reply to a terminal parent or a batch of a terminal message (§10.2), so such rows stop being retried. - Store lookups used by the validation paths are now overridable function vars so the reply and add-to paths are unit-testable. - SPEC.md synced from fmsg-spec v0.6.0. Verified: go test ./... passes; dd.sql applied to a scratch PostgreSQL, both triggers refuse the terminal cases and allow ordinary ones, and the script re-applies idempotently. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MW4FXDGacoWXeKPU5cQTHS --- SPEC.md | 14 ++-- cmd/fmsgd/defs.go | 1 + cmd/fmsgd/host.go | 52 ++++++++++--- cmd/fmsgd/host_test.go | 9 ++- cmd/fmsgd/sender.go | 60 ++++++++++++++- cmd/fmsgd/store.go | 42 ++++++++-- cmd/fmsgd/store_test.go | 14 ++++ cmd/fmsgd/terminal_test.go | 154 +++++++++++++++++++++++++++++++++++++ dd.sql | 33 +++++++- pkg/fmsg/fmsg.go | 7 +- 10 files changed, 355 insertions(+), 31 deletions(-) create mode 100644 cmd/fmsgd/terminal_test.go diff --git a/SPEC.md b/SPEC.md index 695e6b3..d35fcb9 100644 --- a/SPEC.md +++ b/SPEC.md @@ -52,7 +52,8 @@ The hash MUST be computed over the full message bytes: message header fields exa | 3 | important | Sender flags message as important. | | 4 | no reply | Sender will discard any reply. | | 5 | zlib-deflate | Message data compressed with zlib/deflate (RFC 1950/1951); _expanded size_ field present. | -| 6–7 | reserved | Must be 0. | +| 6 | terminal | Message is a leaf: no message may reference it via _pid_. A reply to, or add-to of, a terminal message → code 1. Enforced by hosts, unlike advisory _no reply_. | +| 7 | reserved | MUST be 0; set → code 1. | ## 4. Common Media Types @@ -101,7 +102,7 @@ Each attachment header, in order: | Field | Type | Notes | |-------|------|-------| -| flags | uint8 | Bit 0 = common type (same lookup as §4). Bit 1 = zlib-deflate. Bits 2–7 reserved. | +| flags | uint8 | Bit 0 = common type (same lookup as §4). Bit 1 = zlib-deflate. Bits 2–7 reserved, MUST be 0; set → code 1. | | type | uint8 + [ASCII string] | Same encoding rule as message type, using this attachment's own common type flag. | | filename | uint8 length + UTF-8 | < 256 bytes. Unicode letters/numbers, plus `-` `_` ` ` `.` non-consecutively, not at start/end. Unique per message (case-insensitive). | | size | uint32 | Byte length of this attachment's data on the wire (after compression, if zlib-deflate set). | @@ -175,7 +176,7 @@ One message per connection. Two TCP connections used: Connection 1 (message tran ### 10.2 Sending (Host A perspective) -Host A delivers iff _from_ or _add to from_ belongs to Host A's domain. +Host A delivers iff _from_ or _add to from_ belongs to Host A's domain. Host A MUST NOT send a message whose _pid_ references a message it holds with _terminal_ set. When _has add to_ is NOT set: perform the steps below for each unique recipient domain. @@ -207,6 +208,8 @@ When _has add to_ IS set: perform the steps below for each unique participant do - If _has add to_ not set: ≥ 1 recipient in _to_ belongs to Host B's domain. If _has add to_ set: ≥ 1 participant (_from_, _to_, _add to from_ or _add to_) belongs to Host B's domain. - Common type IDs (message and attachment) are mapped. - _expanded size_ fields are present iff the corresponding zlib-deflate flag is set. + - If _has add to_: _terminal_ is NOT set (an add-to copies the original's flags, so _terminal_ means the original is terminal and cannot be referenced). + - No reserved flag bit is set (message bit 7, attachment bits 2–7). 4. DNS-verify sender IP: resolve `fmsg.`, check Connection 1 source IP is in result set. Fail → TERMINATE. 5. If _size_ + attachment sizes > MAX_SIZE, or total expanded size > MAX_EXPANDED_SIZE → respond code 4, close. Total expanded size uses _expanded size_ for compressed parts and _size_ for uncompressed parts. 6. Compute DELTA = now − _time_: @@ -218,10 +221,11 @@ When _has add to_ IS set: perform the steps below for each unique participant do - Verify parent stored (§11). Not found → respond code 6, close. - Parent time − MAX_TIME_SKEW must be before incoming time. Fail → respond code 9, close. - _from_ must be a participant of the parent. Fail → respond code 1, close. + - Parent must not be terminal. Fail → respond code 1, close. - **add-to set** (adding recipients): - pid MUST also be set. Fail → respond code 1, close. - Check if parent stored (§11): - - **Stored**: check time travel (code 9 if fail). + - **Stored**: parent must not be terminal (code 1 if fail); check time travel (code 9 if fail). - **Not stored**: if ≥ 1 recipient in _to_ or _add to_ belongs to Host B's domain, treat as full message delivery. Otherwise (Host B hosts only non-recipient participants) respond code 6 (parent not found), close. 8. Optionally issue a CHALLENGE on Connection 2 (see §10.5). @@ -289,7 +293,7 @@ An add-to message is a duplicate of the original message with these differences: An add-to message MUST be sent to every participant domain per §10.2, so all participants of the message being added to — including the original sender, when not themselves the _add to from_ — learn of the added recipients, not only the domains hosting the new recipients. This is required because a subsequent reply may reference this add-to message via _pid_, and a host can only accept a reply whose parent it holds. -Add-to batches do not chain: recipients are always added to the original message; an add-to message's _pid_ MUST NOT reference another add-to message. A message therefore has 0 or more add-to batches, each a sibling branch under the original — the thread evolves as a tree. +Add-to batches do not chain: recipients are always added to the original message; an add-to message's _pid_ MUST NOT reference another add-to message. An add-to message MUST NOT reference a terminal message (§3). A message therefore has 0 or more add-to batches, each a sibling branch under the original — the thread evolves as a tree. A recipient added by a batch and not already in _to_ is a participant of that batch message only, not of the original: their replies MUST reference the batch message via _pid_ (referencing the original would fail the participant check, §10.3 step 7) and extend the batch's branch. An address in both _to_ and _add to_ was already a participant of the original and may reply on either branch. diff --git a/cmd/fmsgd/defs.go b/cmd/fmsgd/defs.go index cf18055..1aa40a9 100644 --- a/cmd/fmsgd/defs.go +++ b/cmd/fmsgd/defs.go @@ -15,4 +15,5 @@ const ( FlagImportant = fmsg.FlagImportant FlagNoReply = fmsg.FlagNoReply FlagDeflate = fmsg.FlagDeflate + FlagTerminal = fmsg.FlagTerminal ) diff --git a/cmd/fmsgd/host.go b/cmd/fmsgd/host.go index d855d98..bc9ac1b 100644 --- a/cmd/fmsgd/host.go +++ b/cmd/fmsgd/host.go @@ -57,7 +57,7 @@ const ( RejectCodeAccept uint8 = 200 - messageReservedBitsMask uint8 = 0b11000000 + messageReservedBitsMask uint8 = 0b10000000 // bit 7 (SPEC §3); bit 6 is terminal attachmentReservedBitsMask uint8 = 0b11111100 ) @@ -349,11 +349,11 @@ func isMessageRetrievable(msg *FMsgHeader) bool { if len(msg.Pid) == 0 { return false } - parentID, err := lookupMsgIdByHash(msg.Pid) + parentID, err := lookupMsgIdByHashFn(msg.Pid) if err != nil || parentID == 0 { return false } - parentMsg, err := getMsgByID(parentID) + parentMsg, err := getMsgByIDFn(parentID) if err != nil { return false } @@ -498,6 +498,15 @@ func validateMessageFlags(c net.Conn, flags uint8) error { } return fmt.Errorf("reserved message flag bits set: %#08b", flags) } + // An add-to message duplicates the original's flags, so terminal set on + // an add-to means the original is terminal and cannot be referenced + // (SPEC §10.3 step 3, §12). + if flags&FlagHasAddTo != 0 && flags&FlagTerminal != 0 { + if err := sendCode(c, RejectCodeInvalid); err != nil { + return err + } + return fmt.Errorf("add-to message has terminal flag set: %#08b", flags) + } return nil } @@ -586,7 +595,7 @@ func handleAddToPath(c net.Conn, h *FMsgHeader) (*FMsgHeader, error) { // Deliberately resolves canonical message hashes only: batches do not // chain, so an add-to whose pid is another batch's hash must not resolve // (SPEC §12). - parentID, err := lookupMsgIdByHash(h.Pid) + parentID, err := lookupMsgIdByHashFn(h.Pid) if err != nil { return h, err } @@ -595,7 +604,7 @@ func handleAddToPath(c net.Conn, h *FMsgHeader) (*FMsgHeader, error) { return handleAddToParentNotStored(c, h, hasLocalRecipient) } - parentMsg, err := getMsgByID(parentID) + parentMsg, err := getMsgByIDFn(parentID) if err != nil { return h, err } @@ -603,6 +612,15 @@ func handleAddToPath(c net.Conn, h *FMsgHeader) (*FMsgHeader, error) { return handleAddToParentNotStored(c, h, hasLocalRecipient) } + // Recipients cannot be added to a terminal message (SPEC §10.3 step 7, + // §12). + if parentMsg.Flags&FlagTerminal != 0 { + if err := sendCode(c, RejectCodeInvalid); err != nil { + return h, err + } + return h, fmt.Errorf("add-to: parent msg %d is terminal", parentID) + } + if parentMsg.Timestamp-FutureTimeDelta > h.Timestamp { if err := sendCode(c, RejectCodeTimeTravel); err != nil { return h, err @@ -627,7 +645,7 @@ func handleAddToPath(c net.Conn, h *FMsgHeader) (*FMsgHeader, error) { // A batch this host already recorded is a duplicate (SPEC §10.4 step 1). // The same addresses re-issued at a new time hash differently and are a // distinct batch — a new sibling branch — not a duplicate (SPEC §12). - recorded, err := addToBatchRecorded(parentID, batchHash) + recorded, err := addToBatchRecordedFn(parentID, batchHash) if err != nil { return h, err } @@ -652,19 +670,19 @@ func validatePidReplyPath(c net.Conn, h *FMsgHeader) error { return nil } - parentID, err := lookupMsgIdByHash(h.Pid) + parentID, err := lookupMsgIdByHashFn(h.Pid) if err != nil { return err } var parentMsg *FMsgHeader if parentID != 0 { - parentMsg, err = getMsgByID(parentID) + parentMsg, err = getMsgByIDFn(parentID) } else { // A reply may reference an add-to batch message via pid (SPEC §12); // its wire form is reconstructed from the stored shared message and // batch fields (SPEC §11). - parentMsg, err = getMsgByBatchHash(h.Pid) + parentMsg, err = getMsgByBatchHashFn(h.Pid) } if err != nil { return err @@ -694,10 +712,26 @@ func validatePidReplyPath(c net.Conn, h *FMsgHeader) error { } return fmt.Errorf("pid reply: sender %s was not a participant of parent", h.From.ToString()) } + // A terminal message is a leaf: nothing may reference it via pid (SPEC + // §10.3 step 7). + if parentMsg.Flags&FlagTerminal != 0 { + if err := sendCode(c, RejectCodeInvalid); err != nil { + return err + } + return fmt.Errorf("pid reply: parent %s is terminal", hex.EncodeToString(h.Pid)) + } return nil } +// Store lookups used by the header validation paths. Overridable in tests. +var ( + lookupMsgIdByHashFn = lookupMsgIdByHash + getMsgByIDFn = getMsgByID + getMsgByBatchHashFn = getMsgByBatchHash + addToBatchRecordedFn = addToBatchRecorded +) + func readVersionOrChallenge(c net.Conn, r *bufio.Reader, h *FMsgHeader) (bool, error) { v, err := r.ReadByte() if err != nil { diff --git a/cmd/fmsgd/host_test.go b/cmd/fmsgd/host_test.go index b264d7f..ab39f23 100644 --- a/cmd/fmsgd/host_test.go +++ b/cmd/fmsgd/host_test.go @@ -216,6 +216,9 @@ func TestFlagConstants(t *testing.T) { if FlagDeflate != 32 { t.Errorf("FlagDeflate = %d, want 32 (bit 5)", FlagDeflate) } + if FlagTerminal != 64 { + t.Errorf("FlagTerminal = %d, want 64 (bit 6)", FlagTerminal) + } } func encodeUInt8String(t *testing.T, s string) []byte { @@ -482,7 +485,7 @@ func TestReadAttachmentHeadersRejectsTooBig(t *testing.T) { func TestValidateMessageFlagsRejectsReservedBits(t *testing.T) { c := &testConn{} - err := validateMessageFlags(c, 1<<6) + err := validateMessageFlags(c, 1<<7) if err == nil { t.Fatalf("expected error for reserved message flag bit") } @@ -724,8 +727,8 @@ func TestReadAttachmentHeadersReadsExpandedSizeForCompressedAttachment(t *testin }) h := &FMsgHeader{Size: 0} - b := []byte{1} // 1 attachment - b = append(b, 1<<1) // attachment flags: zlib-deflate (bit 1) + b := []byte{1} // 1 attachment + b = append(b, 1<<1) // attachment flags: zlib-deflate (bit 1) b = append(b, encodeUInt8String(t, "text/plain")...) b = append(b, encodeUInt8String(t, "file.txt")...) diff --git a/cmd/fmsgd/sender.go b/cmd/fmsgd/sender.go index 5526ed7..f2db113 100644 --- a/cmd/fmsgd/sender.go +++ b/cmd/fmsgd/sender.go @@ -481,14 +481,34 @@ func deliverMessage(target pendingTarget) { return } + // SPEC §10.2: a host must not transmit a reply to a terminal message, nor + // an add-to batch of one. dd.sql refuses to create such rows, so this is + // defensive; matching units are recorded as code 1 (invalid, not + // retryable) rather than sent, so they stop being picked up. + parentTerminal, err := isStoredMsgTerminal(db, m.parentPid) + if err != nil { + log.Printf("ERROR: sender: checking parent of msg %d for terminal: %s", target.MsgID, err) + return + } + // Deliver the original message to its pending msg_to recipients. - deliverUnit(db, target, orig, "msg_to", 0) + if parentTerminal { + log.Printf("ERROR: sender: msg %d references a terminal parent; not sending (SPEC §10.2)", target.MsgID) + recordUnitInvalid(db, target, "msg_to", 0) + } else { + deliverUnit(db, target, orig, "msg_to", 0) + } // Deliver each add-to batch as its own add-to message (one sender each). for _, b := range batches { if len(b.Recipients) == 0 { continue // degenerate batch; an add-to message needs ≥ 1 recipient } + if m.isTerminal { + log.Printf("ERROR: sender: msg %d is terminal; not sending add-to batch %d (SPEC §12)", target.MsgID, b.ID) + recordUnitInvalid(db, target, "msg_add_to", b.ID) + continue + } h := m.addToHeader(b, sharedHash) d.applyTo(h) // Persist the batch hash — the batch's identity (SPEC §11) — once, @@ -508,6 +528,44 @@ func deliverMessage(target pendingTarget) { } } +// recordUnitInvalid records code 1 (invalid) against one delivery unit's +// pending recipients and notify row for a domain without transmitting +// anything, for a unit the protocol forbids sending. Code 1 is not retryable, +// so the rows drop out of findPendingTargets. +func recordUnitInvalid(db *sql.DB, target pendingTarget, table string, batchID int64) { + tx, err := db.Begin() + if err != nil { + log.Printf("ERROR: sender: begin tx: %s", err) + return + } + committed := false + defer func() { + if !committed { + tx.Rollback() + } + }() + now := timeutil.TimestampNow().Float64() + locked, err := lockPendingRecipients(tx, table, target.MsgID, target.Domain, batchID, now) + if err != nil { + log.Printf("ERROR: sender: lock %s rows for msg %d: %s", table, target.MsgID, err) + return + } + var notifyID int64 + if table == "msg_add_to" { + notifyID, err = lockPendingNotify(tx, batchID, target.Domain, now) + if err != nil { + log.Printf("ERROR: sender: lock notify row for batch %d: %s", batchID, err) + return + } + } + if len(locked) == 0 && notifyID == 0 { + return + } + updateLocked(tx, table, locked, target.MsgID, now, int(RejectCodeInvalid), false) + updateNotify(tx, notifyID, now, int(RejectCodeInvalid), false) + commitOrLog(tx, &committed, target.MsgID) +} + // markLocalDelivered marks a message's local-domain msg_to recipients delivered // rather than sending over the network. (findPendingTargets skips the local // domain, so this is a defensive path.) diff --git a/cmd/fmsgd/store.go b/cmd/fmsgd/store.go index 1c0311b..a556f6f 100644 --- a/cmd/fmsgd/store.go +++ b/cmd/fmsgd/store.go @@ -506,6 +506,7 @@ func storeMsgDetail(msg *FMsgHeader, localOutcome map[string]uint8) error { , no_reply , is_important , is_deflate + , is_terminal , time_sent , from_addr , topic @@ -515,12 +516,13 @@ func storeMsgDetail(msg *FMsgHeader, localOutcome map[string]uint8) error { , size , filepath , wire_header) -values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) +values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) returning id`, msg.Version, msg.Flags&FlagNoReply != 0, msg.Flags&FlagImportant != 0, msg.Flags&FlagDeflate != 0, + msg.Flags&FlagTerminal != 0, msg.Timestamp, msg.From.ToString(), msg.Topic, @@ -646,6 +648,7 @@ func storeMsgHeaderOnly(msg *FMsgHeader) error { , no_reply , is_important , is_deflate + , is_terminal , time_sent , from_addr , topic @@ -655,12 +658,13 @@ func storeMsgHeaderOnly(msg *FMsgHeader) error { , size , filepath , wire_header) -values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) +values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) returning id`, msg.Version, msg.Flags&FlagNoReply != 0, msg.Flags&FlagImportant != 0, msg.Flags&FlagDeflate != 0, + msg.Flags&FlagTerminal != 0, msg.Timestamp, msg.From.ToString(), msg.Topic, @@ -742,6 +746,7 @@ type msgFields struct { version int size int noReply, isImportant, isDeflate bool + isTerminal bool // SPEC §3 bit 6: no message may reference this one via pid parentPid []byte // relational parent hash (stored pid column) storedHash []byte // stored sha256; empty when not yet persisted from FMsgAddress @@ -757,9 +762,9 @@ func loadMsgFields(tx *sql.Tx, msgID int64) (*msgFields, error) { var m msgFields var fromAddr string if err := tx.QueryRow(` - SELECT version, no_reply, is_important, is_deflate, psha256, sha256, from_addr, topic, type, time_sent, size, filepath + SELECT version, no_reply, is_important, is_deflate, is_terminal, psha256, sha256, from_addr, topic, type, time_sent, size, filepath FROM msg WHERE id = $1 - `, msgID).Scan(&m.version, &m.noReply, &m.isImportant, &m.isDeflate, &m.parentPid, &m.storedHash, + `, msgID).Scan(&m.version, &m.noReply, &m.isImportant, &m.isDeflate, &m.isTerminal, &m.parentPid, &m.storedHash, &fromAddr, &m.topic, &m.typ, &m.timeSent, &m.size, &m.filepath); err != nil { return nil, fmt.Errorf("load msg %d: %w", msgID, err) } @@ -829,8 +834,8 @@ func loadRecipientAddrs(tx *sql.Tx, query string, msgID int64) ([]FMsgAddress, e return addrs, rows.Err() } -// baseFlags returns the persisted flag bits (no_reply/important/deflate) shared -// by every wire form of the message. +// baseFlags returns the persisted flag bits (no_reply/important/deflate/ +// terminal) shared by every wire form of the message. func (m *msgFields) baseFlags() uint8 { var f uint8 if m.noReply { @@ -842,9 +847,34 @@ func (m *msgFields) baseFlags() uint8 { if m.isDeflate { f |= FlagDeflate } + if m.isTerminal { + f |= FlagTerminal + } return f } +// isStoredMsgTerminal reports whether the stored message identified by hash — +// a message's canonical hash or one of its add-to batch hashes (SPEC §11) — +// has the terminal flag set. False when no such message is stored. +func isStoredMsgTerminal(db *sql.DB, hash []byte) (bool, error) { + if len(hash) == 0 { + return false, nil + } + var terminal bool + err := db.QueryRow(` + SELECT m.is_terminal FROM msg m WHERE m.sha256 = $1 + UNION ALL + SELECT m.is_terminal FROM msg m + INNER JOIN msg_add_to_batch b ON b.msg_id = m.id + WHERE b.sha256 = $1 + LIMIT 1 + `, hash).Scan(&terminal) + if err == sql.ErrNoRows { + return false, nil + } + return terminal, err +} + // originalHeader builds the message in its original (non-add-to) wire form, // whose pid (if any) references the relational parent. func (m *msgFields) originalHeader() *FMsgHeader { diff --git a/cmd/fmsgd/store_test.go b/cmd/fmsgd/store_test.go index b5bfd07..bc34311 100644 --- a/cmd/fmsgd/store_test.go +++ b/cmd/fmsgd/store_test.go @@ -225,3 +225,17 @@ func TestInboundRecipientRow(t *testing.T) { t.Fatalf("remote: got (%v, %v), want (nil, %d)", delivered, code, localResponseCodeNotOurDelivery) } } + +func TestBaseFlagsIncludesTerminal(t *testing.T) { + m := &msgFields{noReply: true, isTerminal: true} + got := m.baseFlags() + if got&FlagTerminal == 0 { + t.Fatalf("baseFlags() = %#08b, want terminal bit set", got) + } + if got&FlagNoReply == 0 { + t.Fatalf("baseFlags() = %#08b, want no reply bit set", got) + } + if (&msgFields{}).baseFlags()&FlagTerminal != 0 { + t.Fatalf("baseFlags() set terminal for a non-terminal message") + } +} diff --git a/cmd/fmsgd/terminal_test.go b/cmd/fmsgd/terminal_test.go new file mode 100644 index 0000000..415d2c7 --- /dev/null +++ b/cmd/fmsgd/terminal_test.go @@ -0,0 +1,154 @@ +package main + +import ( + "os" + "path/filepath" + "testing" +) + +// Tests for the terminal flag (SPEC v0.6.0 §3 bit 6): a terminal message is a +// leaf, so a reply to it or an add-to batch of it is rejected with code 1. + +func TestValidateMessageFlagsAcceptsTerminal(t *testing.T) { + c := &testConn{} + if err := validateMessageFlags(c, FlagHasPid|FlagNoReply|FlagTerminal); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if c.Len() != 0 { + t.Fatalf("unexpected bytes written: %v", c.Bytes()) + } +} + +func TestValidateMessageFlagsRejectsAddToWithTerminal(t *testing.T) { + c := &testConn{} + err := validateMessageFlags(c, FlagHasPid|FlagHasAddTo|FlagTerminal) + if err == nil { + t.Fatal("expected error for add-to message with terminal flag set") + } + if got := c.Bytes(); len(got) != 1 || got[0] != RejectCodeInvalid { + t.Fatalf("wrote %v, want single code %d (invalid)", got, RejectCodeInvalid) + } +} + +var ( + testAlice = FMsgAddress{User: "alice", Domain: "example.com"} + testBob = FMsgAddress{User: "bob", Domain: "example.edu"} + testCarol = FMsgAddress{User: "carol", Domain: "example.edu"} +) + +// storedParentForTest returns a retrievable stored message from alice to bob +// with the given flags. +func storedParentForTest(t *testing.T, flags uint8) *FMsgHeader { + t.Helper() + path := filepath.Join(t.TempDir(), "data") + if err := os.WriteFile(path, []byte("hi"), 0o600); err != nil { + t.Fatal(err) + } + return &FMsgHeader{ + Version: 1, + Flags: flags, + From: testAlice, + To: []FMsgAddress{testBob}, + Timestamp: 1000, + Type: "text/plain", + Size: 2, + Filepath: path, + } +} + +// stubStoredParent makes every store lookup resolve to parent (message id 1) +// with no add-to batches recorded, and restores the real lookups afterwards. +func stubStoredParent(t *testing.T, parent *FMsgHeader) { + t.Helper() + origLookup, origGet, origBatch, origRecorded := lookupMsgIdByHashFn, getMsgByIDFn, getMsgByBatchHashFn, addToBatchRecordedFn + origDomain := Domain + t.Cleanup(func() { + lookupMsgIdByHashFn, getMsgByIDFn, getMsgByBatchHashFn, addToBatchRecordedFn = origLookup, origGet, origBatch, origRecorded + Domain = origDomain + }) + lookupMsgIdByHashFn = func([]byte) (int64, error) { return 1, nil } + getMsgByIDFn = func(int64) (*FMsgHeader, error) { return parent, nil } + getMsgByBatchHashFn = func([]byte) (*FMsgHeader, error) { return nil, nil } + addToBatchRecordedFn = func(int64, []byte) (bool, error) { return false, nil } + Domain = testBob.Domain +} + +func replyForTest(from FMsgAddress) *FMsgHeader { + return &FMsgHeader{ + Version: 1, + Flags: FlagHasPid, + Pid: make([]byte, 32), + From: from, + To: []FMsgAddress{testAlice}, + Timestamp: 2000, + Type: "text/plain", + } +} + +func TestValidatePidReplyPathRejectsTerminalParent(t *testing.T) { + stubStoredParent(t, storedParentForTest(t, FlagTerminal)) + c := &testConn{} + err := validatePidReplyPath(c, replyForTest(testBob)) + if err == nil { + t.Fatal("expected error for reply to terminal parent") + } + if got := c.Bytes(); len(got) != 1 || got[0] != RejectCodeInvalid { + t.Fatalf("wrote %v, want single code %d (invalid)", got, RejectCodeInvalid) + } +} + +func TestValidatePidReplyPathAcceptsNonTerminalParent(t *testing.T) { + stubStoredParent(t, storedParentForTest(t, FlagNoReply)) + c := &testConn{} + if err := validatePidReplyPath(c, replyForTest(testBob)); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if c.Len() != 0 { + t.Fatalf("unexpected bytes written: %v", c.Bytes()) + } +} + +func addToForTest() *FMsgHeader { + from := testAlice + return &FMsgHeader{ + Version: 1, + Flags: FlagHasPid | FlagHasAddTo, + Pid: make([]byte, 32), + From: testAlice, + To: []FMsgAddress{testBob}, + AddToFrom: &from, + AddTo: []FMsgAddress{testCarol}, + Timestamp: 2000, + Type: "text/plain", + Size: 2, + } +} + +// A sender that strips the terminal bit from an add-to copy of a terminal +// message is still caught by the stored-parent check. +func TestHandleAddToPathRejectsTerminalParent(t *testing.T) { + stubStoredParent(t, storedParentForTest(t, FlagTerminal)) + c := &testConn{} + _, err := handleAddToPath(c, addToForTest()) + if err == nil { + t.Fatal("expected error for add-to of terminal parent") + } + if got := c.Bytes(); len(got) != 1 || got[0] != RejectCodeInvalid { + t.Fatalf("wrote %v, want single code %d (invalid)", got, RejectCodeInvalid) + } +} + +func TestHandleAddToPathAcceptsNonTerminalParent(t *testing.T) { + stubStoredParent(t, storedParentForTest(t, 0)) + c := &testConn{} + h, err := handleAddToPath(c, addToForTest()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if h.InitialResponseCode != AcceptCodeSkipData { + t.Fatalf("InitialResponseCode = %d, want %d (skip data)", h.InitialResponseCode, AcceptCodeSkipData) + } + if c.Len() != 0 { + t.Fatalf("unexpected bytes written: %v", c.Bytes()) + } +} diff --git a/dd.sql b/dd.sql index 2d4e14b..2116073 100644 --- a/dd.sql +++ b/dd.sql @@ -26,6 +26,7 @@ create table if not exists msg ( no_reply boolean not null default false, is_important boolean not null default false, is_deflate boolean not null default false, + is_terminal boolean not null default false, -- SPEC §3 bit 6: leaf message, nothing may reference it via pid time_sent double precision, -- time sending host recieved message for sending, message timestamp field, NULL means message not ready for sending i.e. draft from_addr varchar(255) not null, topic varchar(255) not null, @@ -38,6 +39,7 @@ create table if not exists msg ( ); create index if not exists msg_lower_idx on msg ((lower(from_addr))); alter table msg add column if not exists wire_header bytea; -- upgrade path for databases created before this column +alter table msg add column if not exists is_terminal boolean not null default false; -- upgrade path (SPEC v0.6.0) create table if not exists msg_to ( id bigserial primary key, @@ -102,19 +104,22 @@ create table if not exists msg_attachment ( ); -- keep protocol parent hash populated for locally-created replies that set --- the relational parent id. A reply cannot reference a draft parent, and any --- explicit psha256 must match the referenced parent's sha256. +-- the relational parent id. A reply cannot reference a draft parent or a +-- terminal parent (SPEC v0.6.0 §3: a Sending Host must not transmit a reply +-- to a terminal message, so refuse to create one), and any explicit psha256 +-- must match the referenced parent's sha256. create or replace function populate_msg_psha256_from_pid() returns trigger as $$ declare parent_time_sent double precision; parent_sha256 bytea; + parent_is_terminal boolean; begin if NEW.pid is null then return NEW; end if; - select parent.time_sent, parent.sha256 - into parent_time_sent, parent_sha256 + select parent.time_sent, parent.sha256, parent.is_terminal + into parent_time_sent, parent_sha256, parent_is_terminal from msg parent where parent.id = NEW.pid; @@ -126,6 +131,10 @@ begin raise exception 'cannot set pid %: parent message is a draft', NEW.pid; end if; + if parent_is_terminal then + raise exception 'cannot set pid %: parent message is terminal', NEW.pid; + end if; + if parent_sha256 is null or octet_length(parent_sha256) = 0 then -- parent was delivered locally only and has no sha256 yet; psha256 cannot be populated return NEW; @@ -153,6 +162,22 @@ create trigger trg_msg_populate_psha256 before insert or update of pid, psha256 on msg for each row execute function populate_msg_psha256_from_pid(); +-- recipients cannot be added to a terminal message (SPEC §12): refuse to +-- create a batch for one, so the sender never has such a unit to transmit. +create or replace function prevent_add_to_terminal_msg() returns trigger as $$ +begin + if exists (select 1 from msg where id = NEW.msg_id and is_terminal) then + raise exception 'cannot add recipients to message %: it is terminal', NEW.msg_id; + end if; + return NEW; +end; +$$ language plpgsql; + +drop trigger if exists trg_msg_add_to_batch_terminal on msg_add_to_batch; +create trigger trg_msg_add_to_batch_terminal + before insert on msg_add_to_batch + for each row execute function prevent_add_to_terminal_msg(); + -- once a message has replies, it must remain referenceable by protocol hash. create or replace function prevent_referenced_msg_from_becoming_unreferenceable() returns trigger as $$ begin diff --git a/pkg/fmsg/fmsg.go b/pkg/fmsg/fmsg.go index fb489e1..3a955a1 100644 --- a/pkg/fmsg/fmsg.go +++ b/pkg/fmsg/fmsg.go @@ -18,15 +18,16 @@ import ( "strings" ) -// Flag bit assignments per SPEC.md. -// Bits 6–7 are reserved and must be zero on the wire. +// Flag bit assignments per SPEC.md §3. +// Bit 7 is reserved and must be zero on the wire. const ( FlagHasPid uint8 = 1 // bit 0: pid field present; message is a reply FlagHasAddTo uint8 = 1 << 1 // bit 1: add-to addresses present FlagCommonType uint8 = 1 << 2 // bit 2: type encoded as common type ID, not string FlagImportant uint8 = 1 << 3 // bit 3: sender marks message as important - FlagNoReply uint8 = 1 << 4 // bit 4: sender will discard replies + FlagNoReply uint8 = 1 << 4 // bit 4: sender will discard replies (advisory) FlagDeflate uint8 = 1 << 5 // bit 5: message body is zlib-deflate compressed + FlagTerminal uint8 = 1 << 6 // bit 6: leaf message; no message may reference it via pid (enforced by hosts) ) // Address is an fmsg address of the form @user@domain. From 40e5f8a290b3f5b4a06f99e4cc964621a94494b3 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Wed, 2 Sep 2026 15:36:15 +0800 Subject: [PATCH 2/2] Encode Common Media Type IDs on outgoing headers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sender always encoded types as strings, so a host could never send the one-byte common type form (SPEC §4) that FMSG-005 requires for reactions (ID 56). Outgoing headers now set the common type flag and ID for the message type and each attachment type that has one. The message hash covers the header exactly as transmitted, so a message whose hash was recorded by an earlier delivery keeps the form that reproduces it: encodeForWire tries the common-type form, compares its hash with the stored one, and falls back to string types when they differ. Add-to batches follow the original's form, and a batch with its own stored hash is checked the same way (batch hashes are now loaded). Also fixes originalHeader sharing the loaded attachment slice between wire forms, so flags set on one header leaked into the next. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01MW4FXDGacoWXeKPU5cQTHS --- cmd/fmsgd/common_type_test.go | 122 ++++++++++++++++++++++++++++++++++ cmd/fmsgd/sender.go | 67 +++++++++++++++++-- cmd/fmsgd/store.go | 10 +-- 3 files changed, 191 insertions(+), 8 deletions(-) create mode 100644 cmd/fmsgd/common_type_test.go diff --git a/cmd/fmsgd/common_type_test.go b/cmd/fmsgd/common_type_test.go new file mode 100644 index 0000000..f153f57 --- /dev/null +++ b/cmd/fmsgd/common_type_test.go @@ -0,0 +1,122 @@ +package main + +import ( + "bytes" + "os" + "path/filepath" + "testing" +) + +// Outgoing headers encode Common Media Type IDs (SPEC §4) where the stored +// type string has one, as FMSG-005 requires for reactions (ID 56). + +func commonTypeTestFields(t *testing.T) *msgFields { + t.Helper() + dir := t.TempDir() + bodyPath := filepath.Join(dir, "data.txt") + if err := os.WriteFile(bodyPath, []byte("👍"), 0o600); err != nil { + t.Fatal(err) + } + attPath := filepath.Join(dir, "pic.png") + if err := os.WriteFile(attPath, []byte("png"), 0o600); err != nil { + t.Fatal(err) + } + return &msgFields{ + version: 1, + size: 4, + from: FMsgAddress{User: "alice", Domain: "example.com"}, + to: []FMsgAddress{{User: "bob", Domain: "example.org"}}, + timeSent: 1754280000, + topic: "types", + typ: "text/plain;charset=UTF-8", + filepath: bodyPath, + attachments: []FMsgAttachmentHeader{ + {Type: "image/png", Filename: "pic.png", Size: 3, Filepath: attPath}, + {Type: "application/x-custom", Filename: "custom.bin", Size: 3, Filepath: attPath}, + }, + } +} + +func TestApplyCommonTypesEncodesIDs(t *testing.T) { + h := commonTypeTestFields(t).originalHeader() + if !applyCommonTypes(h) { + t.Fatal("applyCommonTypes reported no change") + } + if h.Flags&FlagCommonType == 0 || h.TypeID != 56 { + t.Errorf("message type: flags=%#08b id=%d, want common type ID 56", h.Flags, h.TypeID) + } + if h.Attachments[0].Flags&1 == 0 || h.Attachments[0].TypeID != 38 { + t.Errorf("png attachment: flags=%#08b id=%d, want common type ID 38", h.Attachments[0].Flags, h.Attachments[0].TypeID) + } + if h.Attachments[1].Flags&1 != 0 { + t.Errorf("unmapped attachment type must stay a string, flags=%#08b", h.Attachments[1].Flags) + } + wire := h.Encode() + if bytes.Contains(wire, []byte("text/plain")) || bytes.Contains(wire, []byte("image/png")) { + t.Error("common type strings must not appear on the wire") + } + if !bytes.Contains(wire, []byte("application/x-custom")) { + t.Error("unmapped type string must appear on the wire") + } + if applyCommonTypes(h) { + t.Error("second application must be a no-op") + } +} + +func TestEncodeForWireUsesCommonTypesForNewMessage(t *testing.T) { + m := commonTypeTestFields(t) + h, common, err := encodeForWire(m.originalHeader, deflateState{}, true, nil) + if err != nil { + t.Fatal(err) + } + if !common || h.Flags&FlagCommonType == 0 { + t.Errorf("new message should use common type IDs (common=%v flags=%#08b)", common, h.Flags) + } + h, common, err = encodeForWire(m.originalHeader, deflateState{}, false, nil) + if err != nil { + t.Fatal(err) + } + if common || h.Flags&FlagCommonType != 0 { + t.Errorf("commonTypes=false must keep string types (common=%v flags=%#08b)", common, h.Flags) + } +} + +// A message whose hash was recorded before this host encoded common type IDs +// keeps its string types, so every delivery reproduces the stored hash. +func TestEncodeForWireKeepsRecordedForm(t *testing.T) { + m := commonTypeTestFields(t) + + stringForm := m.originalHeader() + stringHash, err := stringForm.GetMessageHash() + if err != nil { + t.Fatal(err) + } + h, common, err := encodeForWire(m.originalHeader, deflateState{}, true, stringHash) + if err != nil { + t.Fatal(err) + } + if common || h.Flags&FlagCommonType != 0 { + t.Errorf("hash recorded in string form must keep string form (common=%v flags=%#08b)", common, h.Flags) + } + got, _ := h.GetMessageHash() + if !bytes.Equal(got, stringHash) { + t.Error("string form does not reproduce the recorded hash") + } + + commonForm := m.originalHeader() + applyCommonTypes(commonForm) + commonHash, err := commonForm.GetMessageHash() + if err != nil { + t.Fatal(err) + } + if bytes.Equal(commonHash, stringHash) { + t.Fatal("forms hash identically; test no longer discriminates") + } + h, common, err = encodeForWire(m.originalHeader, deflateState{}, true, commonHash) + if err != nil { + t.Fatal(err) + } + if !common || h.Flags&FlagCommonType == 0 { + t.Errorf("hash recorded in common form must keep common form (common=%v flags=%#08b)", common, h.Flags) + } +} diff --git a/cmd/fmsgd/sender.go b/cmd/fmsgd/sender.go index f2db113..de302be 100644 --- a/cmd/fmsgd/sender.go +++ b/cmd/fmsgd/sender.go @@ -1,10 +1,12 @@ package main import ( + "bytes" "crypto/tls" "database/sql" "encoding/hex" "fmt" + "github.com/markmnl/fmsgd/pkg/fmsg" "io" "log" "net" @@ -358,6 +360,57 @@ func (d deflateState) applyTo(h *FMsgHeader) { } } +// applyCommonTypes encodes h's type, and each attachment's type, as a Common +// Media Type ID (SPEC §4) where the type string has one, so the wire carries +// one byte instead of the string. Standards such as FMSG-005 require the ID +// form. It reports whether anything changed. +func applyCommonTypes(h *FMsgHeader) bool { + changed := false + if id, ok := fmsg.GetCommonMediaTypeID(h.Type); ok && h.Flags&FlagCommonType == 0 { + h.Flags |= FlagCommonType + h.TypeID = id + changed = true + } + for i := range h.Attachments { + att := &h.Attachments[i] + if id, ok := fmsg.GetCommonMediaTypeID(att.Type); ok && att.Flags&1 == 0 { + att.Flags |= 1 // attachment flag bit 0: common type (SPEC §5) + att.TypeID = id + changed = true + } + } + return changed +} + +// encodeForWire builds a unit header in its transmitted form: build, apply +// deflate, then, when commonTypes is set, common type IDs. The message hash +// covers the header exactly as transmitted, so when storedHash was recorded by +// an earlier delivery the form that reproduces it wins: a message first sent +// before this host encoded common type IDs keeps string types for every later +// delivery. It reports whether common type IDs were used, so a message's +// add-to batches can follow the original's form. +func encodeForWire(build func() *FMsgHeader, d deflateState, commonTypes bool, storedHash []byte) (*FMsgHeader, bool, error) { + h := build() + d.applyTo(h) + if !commonTypes || !applyCommonTypes(h) { + return h, commonTypes, nil + } + if len(storedHash) == 0 { + return h, true, nil + } + hash, err := h.GetMessageHash() + if err != nil { + return nil, false, err + } + if bytes.Equal(hash, storedHash) { + return h, true, nil + } + // Recorded in string form before this host encoded common type IDs. + h = build() + d.applyTo(h) + return h, false, nil +} + // removeTempFiles deletes the compression temp files. func (d deflateState) removeTempFiles() { for _, p := range d.cleanup { @@ -462,8 +515,11 @@ func deliverMessage(target pendingTarget) { d := computeDeflate(m, target.MsgID) defer d.removeTempFiles() - orig := m.originalHeader() - d.applyTo(orig) + orig, useCommonTypes, err := encodeForWire(m.originalHeader, d, true, m.storedHash) + if err != nil { + log.Printf("ERROR: sender: building wire header for msg %d: %s", target.MsgID, err) + return + } sharedHash := m.storedHash if len(sharedHash) == 0 { @@ -509,8 +565,11 @@ func deliverMessage(target pendingTarget) { recordUnitInvalid(db, target, "msg_add_to", b.ID) continue } - h := m.addToHeader(b, sharedHash) - d.applyTo(h) + h, _, err := encodeForWire(func() *FMsgHeader { return m.addToHeader(b, sharedHash) }, d, useCommonTypes, b.Hash) + if err != nil { + log.Printf("ERROR: sender: building add-to wire header for batch %d of msg %d: %s", b.ID, target.MsgID, err) + continue + } // Persist the batch hash — the batch's identity (SPEC §11) — once, // so replies referencing this batch resolve at this host too, which // must verify messages it sent, not only ones it received. Cached on diff --git a/cmd/fmsgd/store.go b/cmd/fmsgd/store.go index 44c256d..dda2aaf 100644 --- a/cmd/fmsgd/store.go +++ b/cmd/fmsgd/store.go @@ -800,7 +800,7 @@ func (m *msgFields) originalHeader() *FMsgHeader { Topic: m.topic, Type: m.typ, Size: uint32(m.size), - Attachments: m.attachments, + Attachments: append([]FMsgAttachmentHeader(nil), m.attachments...), // own copy: wire forms mutate attachment flags Filepath: m.filepath, } } @@ -841,13 +841,14 @@ type addToBatch struct { From FMsgAddress TimeAdded float64 Recipients []FMsgAddress + Hash []byte // batch message hash once persisted (SPEC §11); nil before first delivery } // loadAddToBatches returns every add-to batch for a message, each with its // sender, timestamp and recipients, ordered by when it was added. func loadAddToBatches(tx *sql.Tx, msgID int64) ([]addToBatch, error) { rows, err := tx.Query(` - SELECT b.id, b.add_to_from, b.time_added, a.addr + SELECT b.id, b.add_to_from, b.time_added, b.sha256, a.addr FROM msg_add_to_batch b LEFT JOIN msg_add_to a ON a.batch_id = b.id WHERE b.msg_id = $1 @@ -864,8 +865,9 @@ func loadAddToBatches(tx *sql.Tx, msgID int64) ([]addToBatch, error) { var id int64 var fromStr string var timeAdded float64 + var hash []byte var addr sql.NullString - if err := rows.Scan(&id, &fromStr, &timeAdded, &addr); err != nil { + if err := rows.Scan(&id, &fromStr, &timeAdded, &hash, &addr); err != nil { return nil, fmt.Errorf("scan add-to batch row: %w", err) } idx, ok := byID[id] @@ -874,7 +876,7 @@ func loadAddToBatches(tx *sql.Tx, msgID int64) ([]addToBatch, error) { if err != nil { return nil, fmt.Errorf("invalid add_to_from address %s: %w", fromStr, err) } - batches = append(batches, addToBatch{ID: id, From: *from, TimeAdded: timeAdded}) + batches = append(batches, addToBatch{ID: id, From: *from, TimeAdded: timeAdded, Hash: hash}) idx = len(batches) - 1 byID[id] = idx }