mimeparse: refactored errors, added custom error types

envelopoe: added mime parse error to enveloper
fixed tests

api_test.go

@@ -1089,7 +1089,7 @@ func TestStreamProcessorBackground(t *testing.T) {
 					"post_process_size":     100,
 					"stream_buffer_size":    1024,
 					"save_workers_size":     8,
-					"save_timeout":          "1s",
+					"save_timeout":          "20s",
 					"val_rcpt_timeout":      "2s",
 				},
 			},

backends/gateway_test.go

@@ -96,7 +96,7 @@ func TestStartProcessStop(t *testing.T) {
 
 	e := &mail.Envelope{
 		RemoteIP: "127.0.0.1",
-		QueuedId: "abc12345",
+		QueuedId: mail.QueuedID(1, 2),
 		Helo:     "helo.example.com",
 		MailFrom: mail.Address{User: "test", Host: "example.com"},
 		TLS:      true,

backends/s_mimeanalyzer.go

@@ -30,7 +30,7 @@ func StreamMimeAnalyzer() *StreamDecorator {
 	sd := &StreamDecorator{}
 	var (
 		envelope *mail.Envelope
-		parseErr error
+		mimeErr  error
 		parser   *mimeparse.Parser
 	)
 	sd.Configure = func(cfg ConfigGroup) error {
@@ -38,15 +38,9 @@ func StreamMimeAnalyzer() *StreamDecorator {
 		return nil
 	}
 	sd.Shutdown = func() error {
-		var err error
-		defer func() {
-			parser = nil
-
-		}()
-		if err = parser.Close(); err != nil {
-			Log().WithError(err).Error("error when closing parser in mimeanalyzer")
-			return err
-		}
+		// assumed that parser has been closed, but we can call close again just to make sure
+		_ = parser.Close()
+		parser = nil
 		return nil
 	}
 
@@ -56,26 +50,39 @@ func StreamMimeAnalyzer() *StreamDecorator {
 			sd.Open = func(e *mail.Envelope) error {
 				parser.Open()
 				envelope = e
+				mimeErr = nil
+				envelope.MimeError = nil
 				return nil
 			}
 
 			sd.Close = func() error {
-				if parseErr == nil {
-					if parseErr = parser.Close(); parseErr != nil {
-						Log().WithError(parseErr).Error("mime parse error in mimeanalyzer on close")
+				closeErr := parser.Close()
+				if mimeErr == nil {
+					mimeErr = closeErr
+				}
+
+				envelope.MimeError = mimeErr
+
+				if mimeErr != nil {
+					Log().WithError(closeErr).Warn("mime parse error in mimeanalyzer on close")
+					envelope.MimeError = nil
+
+					if err, ok := mimeErr.(*mimeparse.Error); ok && err.ParseError() {
+						// dont propagate parse errors && NotMime error
+						return nil
 					}
 				}
-				return parseErr
+				return mimeErr
 			}
 
 			return StreamProcessWith(func(p []byte) (int, error) {
 				if envelope.MimeParts == nil {
 					envelope.MimeParts = &parser.Parts
 				}
-				if parseErr == nil {
-					parseErr = parser.Parse(p)
-					if parseErr != nil {
-						Log().WithError(parseErr).Error("mime parse error in mimeanalyzer")
+				if mimeErr == nil {
+					mimeErr = parser.Parse(p)
+					if mimeErr != nil {
+						Log().WithError(mimeErr).Warn("mime parse error in mimeanalyzer")
 					}
 				}
 				return sp.Write(p)

mail/envelope.go

@@ -158,7 +158,9 @@ type Envelope struct {
 	Size int64
 	// MimeParts contain the information about the mime-parts after they have been parsed
 	MimeParts *mimeparse.Parts
-	// MessageID contains the id of the message after it has been written
+	// MimeError contains any error encountered when parsing mime using the mimeanalyzer
+	MimeError error
+	// MessageID contains theR id of the message after it has been written
 	MessageID uint64
 	// Remote IP address
 	RemoteIP string
@@ -196,11 +198,11 @@ func NewEnvelope(remoteAddr string, clientID uint64, serverID int) *Envelope {
 		RemoteIP: remoteAddr,
 		Values:   make(map[string]interface{}),
 		ServerID: serverID,
-		QueuedId: queuedID(clientID, serverID),
+		QueuedId: QueuedID(clientID, serverID),
 	}
 }
 
-func queuedID(clientID uint64, serverID int) Hash128 {
+func QueuedID(clientID uint64, serverID int) Hash128 {
 	hasher.Lock()
 	defer func() {
 		hasher.h.Reset()
@@ -272,6 +274,7 @@ func (e *Envelope) ResetTransaction() {
 	e.Size = 0
 	e.MessageID = 0
 	e.MimeParts = nil
+	e.MimeError = nil
 	e.Values = make(map[string]interface{})
 }
 
@@ -279,7 +282,7 @@ func (e *Envelope) ResetTransaction() {
 func (e *Envelope) Reseed(remoteIP string, clientID uint64, serverID int) {
 	e.RemoteIP = remoteIP
 	e.ServerID = serverID
-	e.QueuedId = queuedID(clientID, serverID)
+	e.QueuedId = QueuedID(clientID, serverID)
 	e.Helo = ""
 	e.TLS = false
 	e.ESMTP = false

mail/envelope_test.go

@@ -99,7 +99,7 @@ func TestAddressWithIP(t *testing.T) {
 func TestEnvelope(t *testing.T) {
 	e := NewEnvelope("127.0.0.1", 22, 0)
 
-	e.QueuedId = queuedID(2, 33)
+	e.QueuedId = QueuedID(2, 33)
 	e.Helo = "helo.example.com"
 	e.MailFrom = Address{User: "test", Host: "example.com"}
 	e.TLS = true
@@ -162,7 +162,7 @@ func TestEncodedWordAhead(t *testing.T) {
 }
 
 func TestQueuedID(t *testing.T) {
-	h := queuedID(5550000000, 1)
+	h := QueuedID(5550000000, 1)
 
 	if len(h) != 16 { // silly comparison, but there in case of refactoring
 		t.Error("queuedID needs to be 16 bytes in length")
@@ -173,7 +173,7 @@ func TestQueuedID(t *testing.T) {
 		t.Error("queuedID string should be 32 bytes in length")
 	}
 
-	h2 := queuedID(5550000000, 1)
+	h2 := QueuedID(5550000000, 1)
 	if bytes.Equal(h[:], h2[:]) {
 		t.Error("hashes should not be equal")
 	}

mail/mimeparse/mime.go

@@ -11,7 +11,6 @@ doesn't back-track or multi-scan.
 */
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"io"
 	"net/textproto"
@@ -20,6 +19,20 @@ import (
 	"sync"
 )
 
+var (
+	MaxNodesErr *Error
+	NotMineErr  *Error
+)
+
+func init() {
+	NotMineErr = &Error{
+		err: ErrorNotMime,
+	}
+	MaxNodesErr = &Error{
+		err: ErrorMaxNodes,
+	}
+}
+
 const (
 	// maxBoundaryLen limits the length of the content-boundary.
 	// Technically the limit is 79, but here we are more liberal
@@ -46,8 +59,110 @@ const (
 	MaxNodes = 512
 )
 
-var NotMime = errors.New("not Mime")
-var MaxNodesErr = errors.New("too many mime part nodes")
+type MimeError int
+
+const (
+	ErrorNotMime MimeError = iota
+	ErrorMaxNodes
+	ErrorBoundaryTooShort
+	ErrorBoundaryLineExpected
+	ErrorUnexpectedChar
+	ErrorHeaderFieldTooShort
+	ErrorBoundaryExceededLength
+	ErrorHeaderParseError
+	ErrorMissingSubtype
+	ErrorUnexpectedTok
+	ErrorUnexpectedCommentToken
+	ErrorInvalidToken
+	ErrorUnexpectedQuotedStrToken
+	ErrorParameterExpectingEquals
+	ErrorNoHeader
+)
+
+func (e MimeError) Error() string {
+	switch e {
+	case ErrorNotMime:
+		return "not Mime"
+	case ErrorMaxNodes:
+		return "too many mime part nodes"
+	case ErrorBoundaryTooShort:
+		return "content boundary too short"
+	case ErrorBoundaryLineExpected:
+		return "boundary new line expected"
+	case ErrorUnexpectedChar:
+		return "unexpected char"
+	case ErrorHeaderFieldTooShort:
+		return "header field too short"
+	case ErrorBoundaryExceededLength:
+		return "boundary exceeded max length"
+	case ErrorHeaderParseError:
+		return "header parse error"
+	case ErrorMissingSubtype:
+		return "missing subtype"
+	case ErrorUnexpectedTok:
+		return "unexpected tok"
+	case ErrorUnexpectedCommentToken:
+		return "unexpected comment token"
+	case ErrorInvalidToken:
+		return "invalid token"
+	case ErrorUnexpectedQuotedStrToken:
+		return "unexpected token"
+	case ErrorParameterExpectingEquals:
+		return "expecting ="
+	case ErrorNoHeader:
+		return "parse error, no header"
+	}
+	return "unknown mime error"
+}
+
+// Error implements the error interface
+type Error struct {
+	err  error
+	char byte
+	peek byte
+	pos  uint // msgPos
+}
+
+func (e Error) Error() string {
+	if e.char == 0 {
+		return e.err.Error()
+	}
+	return e.err.Error() + " char:[" + string(e.char) + "], peek:" +
+		string(e.peek) + ", pos:" + strconv.Itoa(int(e.pos))
+}
+
+func (e *Error) ParseError() bool {
+	if e.err != io.EOF && e.err != NotMineErr && e.err != MaxNodesErr {
+		return true
+	}
+	return false
+}
+
+func (p *Parser) newParseError(e error) *Error {
+	var peek byte
+	offset := 1
+	for {
+		// reached the end? (don't wait for more bytes to consume)
+		if p.pos+offset >= len(p.buf) {
+			peek = 0
+			break
+		}
+		// peek the next byte
+		peek := p.buf[p.pos+offset]
+		if peek == '\r' {
+			// ignore \r
+			offset++
+			continue
+		}
+		break
+	}
+	return &Error{
+		err:  e,
+		char: p.ch,
+		peek: peek,
+		pos:  p.msgPos,
+	}
+}
 
 type captureBuffer struct {
 	bytes.Buffer
@@ -329,17 +444,15 @@ func (p *Parser) boundary(contentBoundary string) (end bool, err error) {
 	}()
 
 	if len(contentBoundary) < 1 {
-		err = errors.New("content boundary too short")
+		err = ErrorBoundaryTooShort
 	}
 	boundary := doubleDash + contentBoundary
 	p.boundaryMatched = 0
 	for {
 		if i := bytes.Index(p.buf[p.pos:], []byte(boundary)); i > -1 {
-
 			p.skip(i)
 			p.lastBoundaryPos = p.msgPos
 			p.skip(len(boundary))
-
 			if end, err = p.boundaryEnd(); err != nil {
 				return
 			}
@@ -347,10 +460,9 @@ func (p *Parser) boundary(contentBoundary string) (end bool, err error) {
 				return
 			}
 			if p.ch != '\n' {
-				err = errors.New("boundary new line expected")
+				err = ErrorBoundaryLineExpected
 			}
 			return
-
 		} else {
 			// search the tail for partial match
 			// if one is found, load more data and continue the match
@@ -392,7 +504,7 @@ func (p *Parser) boundary(contentBoundary string) (end bool, err error) {
 						return
 					}
 					if p.ch != '\n' {
-						err = errors.New("boundary new line expected")
+						err = ErrorBoundaryLineExpected
 					}
 					return
 				}
@@ -479,14 +591,12 @@ func (p *Parser) header(mh *Part) (err error) {
 					state = 2 // tolerate this error
 					continue
 				}
-				pc := p.peek()
-				err = errors.New("unexpected char:[" + string(p.ch) + "], peek:" +
-					string(pc) + ", pos:" + strconv.Itoa(int(p.msgPos)))
+				err = p.newParseError(ErrorUnexpectedChar)
 				return
 			}
 			if state == 1 {
 				if p.accept.Len() < 2 {
-					err = errors.New("header field too short")
+					err = p.newParseError(ErrorHeaderFieldTooShort)
 					return
 				}
 				p.accept.upper = true
@@ -512,7 +622,7 @@ func (p *Parser) header(mh *Part) (err error) {
 					case contentType.parameters[i].name == "boundary":
 						mh.ContentBoundary = contentType.parameters[i].value
 						if len(mh.ContentBoundary) >= maxBoundaryLen {
-							return errors.New("boundary exceeded max length")
+							return p.newParseError(ErrorBoundaryExceededLength)
 						}
 					case contentType.parameters[i].name == "charset":
 						mh.Charset = strings.ToUpper(contentType.parameters[i].value)
@@ -537,7 +647,7 @@ func (p *Parser) header(mh *Part) (err error) {
 						state = 0
 					}
 				} else {
-					err = errors.New("header parse error, pos:" + strconv.Itoa(p.pos))
+					err = p.newParseError(ErrorHeaderParseError)
 					return
 				}
 			}
@@ -583,7 +693,7 @@ func (p *Parser) contentType() (result contentType, err error) {
 		return
 	}
 	if p.ch != '/' {
-		return result, errors.New("missing subtype")
+		return result, p.newParseError(ErrorMissingSubtype)
 	}
 	p.next()
 
@@ -655,7 +765,7 @@ func (p *Parser) mimeType() (str string, err error) {
 
 		}
 	} else {
-		err = errors.New("unexpected tok")
+		err = p.newParseError(ErrorUnexpectedTok)
 		return
 	}
 }
@@ -675,9 +785,8 @@ func (p *Parser) comment() (err error) {
 	// all header fields except for Content-Disposition
 	// can include RFC 822 comments
 	if p.ch != '(' {
-		err = errors.New("unexpected token")
+		err = p.newParseError(ErrorUnexpectedCommentToken)
 	}
-
 	for {
 		p.next()
 		if p.ch == ')' {
@@ -685,7 +794,6 @@ func (p *Parser) comment() (err error) {
 			return
 		}
 	}
-
 }
 
 func (p *Parser) token(lower bool) (str string, err error) {
@@ -706,7 +814,7 @@ func (p *Parser) token(lower bool) (str string, err error) {
 			_ = p.accept.WriteByte(p.ch)
 			once = true
 		} else if !once {
-			err = errors.New("invalid token")
+			err = p.newParseError(ErrorInvalidToken)
 			return
 		} else {
 			return
@@ -731,7 +839,7 @@ func (p *Parser) quotedString() (str string, err error) {
 	}()
 
 	if p.ch != '"' {
-		err = errors.New("unexpected token")
+		err = p.newParseError(ErrorUnexpectedQuotedStrToken)
 		return
 	}
 	p.next()
@@ -750,7 +858,7 @@ func (p *Parser) quotedString() (str string, err error) {
 			if (p.ch < 127 && p.ch > 32) || p.isWSP(p.ch) {
 				_ = p.accept.WriteByte(p.ch)
 			} else {
-				err = errors.New("unexpected token")
+				err = p.newParseError(ErrorUnexpectedQuotedStrToken)
 				return
 			}
 		case 1:
@@ -759,7 +867,7 @@ func (p *Parser) quotedString() (str string, err error) {
 				_ = p.accept.WriteByte(p.ch)
 				state = 0
 			} else {
-				err = errors.New("unexpected token")
+				err = p.newParseError(ErrorUnexpectedQuotedStrToken)
 				return
 			}
 		}
@@ -785,7 +893,7 @@ func (p *Parser) parameter() (attribute, value string, err error) {
 		if len(attribute) > 0 {
 			return
 		}
-		return "", "", errors.New("expecting =")
+		return "", "", p.newParseError(ErrorParameterExpectingEquals)
 	}
 	p.next()
 	if p.ch == '"' {
@@ -845,7 +953,7 @@ func (p *Parser) mime(part *Part, cb string) (err error) {
 				len(part.Headers) > 0 &&
 				part.Headers.Get("MIME-Version") == "" &&
 				err == nil {
-				err = NotMime
+				err = NotMineErr
 			}
 		}()
 	}
@@ -857,7 +965,7 @@ func (p *Parser) mime(part *Part, cb string) (err error) {
 			return err
 		}
 	} else if root {
-		return errors.New("parse error, no header")
+		return p.newParseError(ErrorNoHeader)
 	}
 	if p.ch == '\n' && p.peek() == '\n' {
 		p.next()
@@ -1052,10 +1160,10 @@ func (p *Parser) Parse(buf []byte) error {
 	}
 }
 
-// ParseError returns true if the type of error was a parse error
+// Error returns true if the type of error was a parse error
 // Returns false if it was an io.EOF or the email was not mime, or exceeded maximum nodes
 func (p *Parser) ParseError(err error) bool {
-	if err != nil && err != io.EOF && err != NotMime && err != MaxNodesErr {
+	if err != nil && err != io.EOF && err != NotMineErr && err != MaxNodesErr {
 		return true
 	}
 	return false

mail/mimeparse/mime_test.go

@@ -603,7 +603,7 @@ This is not a an MIME email
 func TestNonMineEmail(t *testing.T) {
 	p = NewMimeParser()
 	p.inject([]byte(email4))
-	if err := p.mime(nil, ""); err != nil && err != NotMime {
+	if err := p.mime(nil, ""); err != nil && err != NotMineErr {
 		t.Error(err)
 	} else {
 		// err should be NotMime
@@ -618,7 +618,7 @@ func TestNonMineEmail(t *testing.T) {
 
 	// what if we pass an empty string?
 	p.inject([]byte{' '})
-	if err := p.mime(nil, ""); err == nil || err == NotMime {
+	if err := p.mime(nil, ""); err == nil || err == NotMineErr {
 		t.Error("unexpected error", err)
 	}
 
@@ -676,7 +676,7 @@ func TestNonMineEmailBigBody(t *testing.T) {
 		in = append(in, b[i:to+i])
 	}
 	p.inject(in...)
-	if err := p.mime(nil, ""); err != nil && err != NotMime {
+	if err := p.mime(nil, ""); err != nil && err != NotMineErr {
 		t.Error(err)
 	} else {
 		for part := range p.Parts {
@@ -690,7 +690,7 @@ func TestNonMineEmailBigBody(t *testing.T) {
 
 	// what if we pass an empty string?
 	p.inject([]byte{' '})
-	if err := p.mime(nil, ""); err == nil || err == NotMime {
+	if err := p.mime(nil, ""); err == nil || err == NotMineErr {
 		t.Error("unexpected error", err)
 	}
 

server_test.go

@@ -1,19 +1,16 @@
 package guerrilla
 
 import (
-	"os"
-	"testing"
-	"time"
-
 	"bufio"
-	"net/textproto"
-	"strings"
-	"sync"
-
 	"crypto/tls"
 	"fmt"
 	"io/ioutil"
 	"net"
+	"net/textproto"
+	"os"
+	"strings"
+	"sync"
+	"testing"
 
 	"github.com/flashmob/go-guerrilla/backends"
 	"github.com/flashmob/go-guerrilla/log"
@@ -534,7 +531,7 @@ func sendMessage(greet string, TLS bool, w *textproto.Writer, t *testing.T, line
 	line, _ = r.ReadLine()
 	client.Hashes = append(client.Hashes, "abcdef1526777763"+greet)
 	client.TLS = TLS
-	client.QueuedId = time.Now().String()
+	client.QueuedId = mail.QueuedID(1, 1)
 	if err := w.PrintfLine("DATA"); err != nil {
 		t.Error(err)
 	}