summaryrefslogtreecommitdiffstats
path: root/katzenpost/dispatcher
diff options
context:
space:
mode:
Diffstat (limited to 'katzenpost/dispatcher')
-rw-r--r--katzenpost/dispatcher/codec.go29
-rw-r--r--katzenpost/dispatcher/envelope.go45
-rw-r--r--katzenpost/dispatcher/envelope_test.go21
-rw-r--r--katzenpost/dispatcher/frame.go84
-rw-r--r--katzenpost/dispatcher/frame_test.go46
-rw-r--r--katzenpost/dispatcher/reassembler.go91
6 files changed, 0 insertions, 316 deletions
diff --git a/katzenpost/dispatcher/codec.go b/katzenpost/dispatcher/codec.go
deleted file mode 100644
index 3053dde..0000000
--- a/katzenpost/dispatcher/codec.go
+++ /dev/null
@@ -1,29 +0,0 @@
-package dispatcher
-
-import (
- "fmt"
-
- "github.com/fxamacker/cbor/v2"
-)
-
-func EncodeEnvelope(envelope Envelope) ([]byte, error) {
- encoded, err := cbor.Marshal(envelope)
- if err != nil {
- return nil, fmt.Errorf("encode envelope: %w", err)
- }
- if len(encoded) > MaxEnvelopeBytes {
- return nil, fmt.Errorf("encoded envelope is too large")
- }
- return encoded, nil
-}
-
-func DecodeEnvelope(encoded []byte) (Envelope, error) {
- if len(encoded) == 0 || len(encoded) > MaxEnvelopeBytes {
- return Envelope{}, fmt.Errorf("invalid encoded envelope size")
- }
- var envelope Envelope
- if err := cbor.Unmarshal(encoded, &envelope); err != nil {
- return Envelope{}, fmt.Errorf("decode envelope: %w", err)
- }
- return envelope, nil
-}
diff --git a/katzenpost/dispatcher/envelope.go b/katzenpost/dispatcher/envelope.go
deleted file mode 100644
index b9ff7b8..0000000
--- a/katzenpost/dispatcher/envelope.go
+++ /dev/null
@@ -1,45 +0,0 @@
-package dispatcher
-
-import (
- "bytes"
- "errors"
- "fmt"
- "net/mail"
- "strings"
-)
-
-const MaxEnvelopeBytes = 64 * 1024
-
-// Envelope is the only application payload accepted by the dispatcher. Message
-// must already be a complete YAMN transport envelope, not user supplied mail.
-type Envelope struct {
- EntryAddress string `json:"entry_address"`
- Message []byte `json:"message"`
-}
-
-func (e Envelope) Validate(allowedDomains map[string]struct{}) error {
- if len(e.Message) == 0 || len(e.Message) > MaxEnvelopeBytes {
- return fmt.Errorf("invalid YAMN envelope size")
- }
- if bytes.Contains(e.Message, []byte("\r")) {
- return errors.New("YAMN envelope must use LF line endings")
- }
- address, err := mail.ParseAddress(e.EntryAddress)
- if err != nil || address.Address != e.EntryAddress {
- return errors.New("invalid entry remailer address")
- }
- parts := strings.Split(address.Address, "@")
- if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
- return errors.New("invalid entry remailer address")
- }
- if _, ok := allowedDomains[strings.ToLower(parts[1])]; !ok {
- return errors.New("entry remailer domain is not allowed")
- }
- if !bytes.HasPrefix(e.Message, []byte("To: "+e.EntryAddress+"\n")) {
- return errors.New("YAMN envelope recipient does not match entry remailer")
- }
- if !bytes.Contains(e.Message, []byte("-----BEGIN REMAILER MESSAGE-----\n")) {
- return errors.New("invalid YAMN envelope")
- }
- return nil
-}
diff --git a/katzenpost/dispatcher/envelope_test.go b/katzenpost/dispatcher/envelope_test.go
deleted file mode 100644
index 712071c..0000000
--- a/katzenpost/dispatcher/envelope_test.go
+++ /dev/null
@@ -1,21 +0,0 @@
-package dispatcher
-
-import "testing"
-
-func TestEnvelopeValidate(t *testing.T) {
- allowed := map[string]struct{}{"remailer.example": {}}
- valid := Envelope{
- EntryAddress: "entry@remailer.example",
- Message: []byte("To: entry@remailer.example\nFrom: mix@nowhere.invalid\n\n-----BEGIN REMAILER MESSAGE-----\nabc\n"),
- }
- if err := valid.Validate(allowed); err != nil {
- t.Fatalf("expected valid envelope: %v", err)
- }
-
- invalidDomain := valid
- invalidDomain.EntryAddress = "entry@other.example"
- invalidDomain.Message = []byte("To: entry@other.example\n-----BEGIN REMAILER MESSAGE-----\n")
- if err := invalidDomain.Validate(allowed); err == nil {
- t.Fatal("expected unknown domain rejection")
- }
-}
diff --git a/katzenpost/dispatcher/frame.go b/katzenpost/dispatcher/frame.go
deleted file mode 100644
index 43f547c..0000000
--- a/katzenpost/dispatcher/frame.go
+++ /dev/null
@@ -1,84 +0,0 @@
-package dispatcher
-
-import (
- "crypto/rand"
- "crypto/sha256"
- "errors"
- "fmt"
-
- "github.com/fxamacker/cbor/v2"
-)
-
-const (
- ProtocolVersion = 1
- MaxSphinxPayload = 2000
- FrameDataBytes = 1500
- MaxFrames = (MaxEnvelopeBytes + FrameDataBytes - 1) / FrameDataBytes
-)
-
-type Frame struct {
- Version uint8 `cbor:"1,keyasint"`
- ID [16]byte `cbor:"2,keyasint"`
- Digest [32]byte `cbor:"3,keyasint"`
- Index uint16 `cbor:"4,keyasint"`
- Total uint16 `cbor:"5,keyasint"`
- Data []byte `cbor:"6,keyasint"`
-}
-
-func SplitEnvelope(encoded []byte) ([][]byte, error) {
- if len(encoded) == 0 || len(encoded) > MaxEnvelopeBytes {
- return nil, errors.New("invalid encoded envelope size")
- }
-
- var id [16]byte
- if _, err := rand.Read(id[:]); err != nil {
- return nil, fmt.Errorf("generate message id: %w", err)
- }
- digest := sha256.Sum256(encoded)
- total := (len(encoded) + FrameDataBytes - 1) / FrameDataBytes
- if total > MaxFrames {
- return nil, errors.New("encoded envelope requires too many frames")
- }
-
- frames := make([][]byte, 0, total)
- for index, start := 0, 0; start < len(encoded); index, start = index+1, start+FrameDataBytes {
- end := min(start+FrameDataBytes, len(encoded))
- frame := Frame{
- Version: ProtocolVersion,
- ID: id,
- Digest: digest,
- Index: uint16(index),
- Total: uint16(total),
- Data: encoded[start:end],
- }
- payload, err := cbor.Marshal(frame)
- if err != nil {
- return nil, fmt.Errorf("encode frame %d: %w", index, err)
- }
- if len(payload) > MaxSphinxPayload {
- return nil, fmt.Errorf("frame %d exceeds Sphinx payload limit", index)
- }
- frames = append(frames, payload)
- }
- return frames, nil
-}
-
-func DecodeFrame(payload []byte) (Frame, error) {
- if len(payload) == 0 || len(payload) > MaxSphinxPayload {
- return Frame{}, errors.New("invalid frame size")
- }
- var frame Frame
- if err := cbor.Unmarshal(payload, &frame); err != nil {
- return Frame{}, fmt.Errorf("decode frame: %w", err)
- }
- if frame.Version != ProtocolVersion {
- return Frame{}, errors.New("unsupported frame version")
- }
- if frame.Total == 0 || int(frame.Total) > MaxFrames || frame.Index >= frame.Total {
- return Frame{}, errors.New("invalid frame sequence")
- }
- if len(frame.Data) == 0 || len(frame.Data) > FrameDataBytes {
- return Frame{}, errors.New("invalid frame data size")
- }
- return frame, nil
-}
diff --git a/katzenpost/dispatcher/frame_test.go b/katzenpost/dispatcher/frame_test.go
deleted file mode 100644
index 21c84ba..0000000
--- a/katzenpost/dispatcher/frame_test.go
+++ /dev/null
@@ -1,46 +0,0 @@
-package dispatcher
-
-import (
- "bytes"
- "testing"
- "time"
-)
-
-func TestSplitAndReassembleEnvelope(t *testing.T) {
- encoded := bytes.Repeat([]byte("yamn-envelope\n"), 900)
- payloads, err := SplitEnvelope(encoded)
- if err != nil {
- t.Fatal(err)
- }
- if len(payloads) < 2 {
- t.Fatal("expected multiple frames")
- }
-
- reassembler := NewReassembler(time.Minute)
- var result []byte
- for _, payload := range payloads {
- if len(payload) > MaxSphinxPayload {
- t.Fatalf("payload exceeds limit: %d", len(payload))
- }
- frame, err := DecodeFrame(payload)
- if err != nil {
- t.Fatal(err)
- }
- assembled, complete, err := reassembler.Add(frame)
- if err != nil {
- t.Fatal(err)
- }
- if complete {
- result = assembled
- }
- }
- if !bytes.Equal(result, encoded) {
- t.Fatal("reassembled envelope differs")
- }
-}
-
-func TestDecodeFrameRejectsOversizedPayload(t *testing.T) {
- if _, err := DecodeFrame(make([]byte, MaxSphinxPayload+1)); err == nil {
- t.Fatal("expected oversized payload rejection")
- }
-}
diff --git a/katzenpost/dispatcher/reassembler.go b/katzenpost/dispatcher/reassembler.go
deleted file mode 100644
index 87af98c..0000000
--- a/katzenpost/dispatcher/reassembler.go
+++ /dev/null
@@ -1,91 +0,0 @@
-package dispatcher
-
-import (
- "bytes"
- "crypto/sha256"
- "errors"
- "sync"
- "time"
-)
-
-const MaxPendingMessages = 64
-
-type pendingMessage struct {
- digest [32]byte
- total uint16
- parts map[uint16][]byte
- created time.Time
-}
-
-type Reassembler struct {
- mu sync.Mutex
- pending map[[16]byte]*pendingMessage
- ttl time.Duration
- now func() time.Time
-}
-
-func NewReassembler(ttl time.Duration) *Reassembler {
- return &Reassembler{
- pending: make(map[[16]byte]*pendingMessage),
- ttl: ttl,
- now: time.Now,
- }
-}
-
-func (r *Reassembler) Add(frame Frame) ([]byte, bool, error) {
- r.mu.Lock()
- defer r.mu.Unlock()
-
- r.expireLocked()
- pending, ok := r.pending[frame.ID]
- if !ok {
- if len(r.pending) >= MaxPendingMessages {
- return nil, false, errors.New("too many pending messages")
- }
- pending = &pendingMessage{
- digest: frame.Digest,
- total: frame.Total,
- parts: make(map[uint16][]byte, frame.Total),
- created: r.now(),
- }
- r.pending[frame.ID] = pending
- }
- if pending.total != frame.Total || pending.digest != frame.Digest {
- delete(r.pending, frame.ID)
- return nil, false, errors.New("inconsistent frame metadata")
- }
- if _, duplicate := pending.parts[frame.Index]; !duplicate {
- pending.parts[frame.Index] = bytes.Clone(frame.Data)
- }
- if len(pending.parts) != int(pending.total) {
- return nil, false, nil
- }
-
- var assembled bytes.Buffer
- for index := uint16(0); index < pending.total; index++ {
- part, exists := pending.parts[index]
- if !exists {
- return nil, false, nil
- }
- if assembled.Len()+len(part) > MaxEnvelopeBytes {
- delete(r.pending, frame.ID)
- return nil, false, errors.New("reassembled envelope is too large")
- }
- assembled.Write(part)
- }
- delete(r.pending, frame.ID)
- result := assembled.Bytes()
- if sha256.Sum256(result) != pending.digest {
- return nil, false, errors.New("reassembled envelope digest mismatch")
- }
- return result, true, nil
-}
-
-func (r *Reassembler) expireLocked() {
- cutoff := r.now().Add(-r.ttl)
- for id, pending := range r.pending {
- if pending.created.Before(cutoff) {
- delete(r.pending, id)
- }
- }
-}