begin work on chunk saver

api_test.go

@@ -805,9 +805,9 @@ Content-Transfer-Encoding: quoted-printable
 
 --XXXXboundary text
 Content-Type: text/plain;
- name="log_attachment.txt"
+       name="log_attachment.txt"
 Content-Disposition: attachment;
- filename="log_attachment.txt"
+       filename="log_attachment.txt"
 Content-Transfer-Encoding: base64
 
 TUlNRS1WZXJzaW9uOiAxLjANClgtTWFpbGVyOiBNYWlsQmVlLk5FVCA4LjAuNC40MjgNClN1Ympl
@@ -820,7 +820,6 @@ b2R5DQotLS0tLS09X05leHRQYXJ0XzAwMF9BRTZCXzcyNUUwOUFGLjg4QjdGOTM0DQpDb250ZW50
 LVR5cGU6IHRleHQvaHRtbDsNCgljaGFyc2V0PSJ1dGYtOCINCkNvbnRlbnQtVHJhbnNmZXItRW5j
 b2Rpbmc6IHF1b3RlZC1wcmludGFibGUNCg0KPHByZT50ZXN0IGJvZHk8L3ByZT4NCi0tLS0tLT1f
 TmV4dFBhcnRfMDAwX0FFNkJfNzI1RTA5QUYuODhCN0Y5MzQtLQ0K
-
 --XXXXboundary text--
 `
 
@@ -897,6 +896,13 @@ Content-Disposition: attachment;
 ------_=_NextPart_001_01CBE273.65A0E7AA--
 `
 
+/*
+1  0  166  1514
+1.1  186  260  259
+1.2  280  374  416
+1.3  437  530  584
+1.4  605  769  1514
+*/
 func TestStreamMimeProcessor(t *testing.T) {
 	if err := os.Truncate("tests/testlog", 0); err != nil {
 		t.Error(err)
@@ -906,7 +912,7 @@ func TestStreamMimeProcessor(t *testing.T) {
 		AllowedHosts: []string{"grr.la"},
 		BackendConfig: backends.BackendConfig{
 			"save_process":        "HeadersParser|Debugger",
-			"stream_save_process": "Header|mimeanalyzer|headersparser|compress|Decompress|debug",
+			"stream_save_process": "mimeanalyzer|headersparser|compress|Decompress|debug",
 		},
 	}
 	d := Daemon{Config: cfg}

backends/s_mime.go

@@ -1,13 +1,9 @@
 package backends
 
 import (
-	"bytes"
-	"errors"
 	"fmt"
 	"github.com/flashmob/go-guerrilla/mail"
 	"github.com/flashmob/go-guerrilla/mail/mime"
-	"io"
-	"net/textproto"
 	"strconv"
 )
 
@@ -54,13 +50,13 @@ func StreamMimeAnalyzer() *StreamDecorator {
 
 			sd.Open = func(e *mail.Envelope) error {
 				envelope = e
-				return parser.Open()
+				return nil
 			}
 
 			sd.Close = func() error {
 				if parts, ok := envelope.Values["MimeParts"].(*[]*mime.MimeHeader); ok {
 					for _, v := range *parts {
-						fmt.Println(v.part + " " + strconv.Itoa(int(v.startingPos)) + " " + strconv.Itoa(int(v.startingPosBody)) + " " + strconv.Itoa(int(v.endingPosBody)))
+						fmt.Println(v.Part + " " + strconv.Itoa(int(v.StartingPos)) + " " + strconv.Itoa(int(v.StartingPosBody)) + " " + strconv.Itoa(int(v.EndingPosBody)))
 					}
 				}
 
@@ -81,7 +77,7 @@ func StreamMimeAnalyzer() *StreamDecorator {
 					envelope.Values["MimeParts"] = &parser.Parts
 				}
 				if parseErr == nil {
-					_, parseErr = parser.Read(p)
+					parseErr = parser.Parse(p)
 					if parseErr != nil {
 						Log().WithError(parseErr).Error("mime parse error")
 					}

backends/s_mysql_chunksaver.go

@@ -0,0 +1,92 @@
+package backends
+
+import (
+	"bytes"
+	"fmt"
+	"github.com/flashmob/go-guerrilla/mail"
+	"github.com/flashmob/go-guerrilla/mail/mime"
+	"strconv"
+)
+
+func init() {
+	streamers["MysqlChunksaver"] = func() *StreamDecorator {
+		return MysqlChunksaver()
+	}
+}
+
+/**
+ * messages: mid, part_tree, part_count, has_attach, created_at
+ * parts: mid, part_id, chunk_md5, header_data, seq
+ * chunk: md5, references, data, created_at
+ * A chunk ends ether: after 64KB or after end of a part
+ *
+ * - buffer first chunk
+ * - if didn't receive first chunk for more than x bytes, save normally
+ *
+ */
+func MysqlChunksaver() *StreamDecorator {
+
+	sd := &StreamDecorator{}
+	sd.p =
+
+		func(sp StreamProcessor) StreamProcessor {
+
+			var (
+				envelope    *mail.Envelope
+				currentPart int
+				chunkBuffer bytes.Buffer
+			)
+			Svc.AddInitializer(InitializeWith(func(backendConfig BackendConfig) error {
+				return nil
+			}))
+
+			Svc.AddShutdowner(ShutdownWith(func() error {
+
+				return nil
+			}))
+
+			sd.Open = func(e *mail.Envelope) error {
+				// create a new entry & grab the id
+				envelope = e
+				currentPart = 0
+				return nil
+			}
+
+			sd.Close = func() error {
+				if parts, ok := envelope.Values["MimeParts"].(*[]*mime.MimeHeader); ok {
+					for _, v := range *parts {
+						fmt.Println(v.Part + " " + strconv.Itoa(int(v.StartingPos)) + " " + strconv.Itoa(int(v.StartingPosBody)) + " " + strconv.Itoa(int(v.EndingPosBody)))
+					}
+				}
+				chunkBuffer.Reset()
+
+				// finalize the message
+
+				return nil
+			}
+
+			return StreamProcessWith(func(p []byte) (int, error) {
+				_ = envelope
+				if len(envelope.Header) > 0 {
+
+				}
+				var parts []*mime.MimeHeader
+				if val, ok := envelope.Values["MimeParts"]; !ok {
+					//envelope.Values["MimeParts"] = &parser.Parts
+					parts = val.([]*mime.MimeHeader)
+					size := len(parts)
+					if currentPart != size {
+						currentPart = size
+						// a new part! todo: code to start a new part
+						if currentPart == 0 {
+
+						}
+					}
+				}
+
+				return sp.Write(p)
+			})
+		}
+
+	return sd
+}

mail/mime/reader.go → mail/mime/mime.go

@@ -9,6 +9,10 @@ import (
 	"strconv"
 )
 
+// todo
+// - content-disposition
+// - make the error available
+
 type mimepart struct {
 	/*
 			[starting-pos] => 0
@@ -45,6 +49,8 @@ const (
 	startPos       = -1
 )
 
+var NotMime = errors.New("not Mime")
+
 type Parser struct {
 
 	// related to the state of the parser
@@ -136,13 +142,16 @@ func (p *Parser) addPart(mh *MimeHeader, id string) {
 	p.Parts = append(p.Parts, mh)
 }
 
-//
+// more waits for more input, returns false if there is no more
 func (p *Parser) more() bool {
 	p.consumed <- true // signal that we've reached the end of available input
 	gotMore := <-p.gotNewSlice
 	return gotMore
 }
 
+// next reads the next byte and advances the pointer
+// returns 0 if no more input can be read
+// blocks if at the end of the buffer
 func (p *Parser) next() byte {
 	// wait for a new new slice if reached the end
 	if p.pos+1 >= len(p.buf) {
@@ -162,6 +171,8 @@ func (p *Parser) next() byte {
 	return p.ch
 }
 
+// peek does not advance the pointer, but will block if there's no more
+// input in the buffer
 func (p *Parser) peek() byte {
 
 	// reached the end?
@@ -202,7 +213,6 @@ func (p *Parser) set(input []byte) {
 		p.pos = startPos
 	}
 	p.buf = input
-
 }
 
 func (p *Parser) skip(nBytes int) {
@@ -213,7 +223,6 @@ func (p *Parser) skip(nBytes int) {
 			return
 		}
 	}
-
 }
 
 // boundary scans until next boundary string, returns error if not found
@@ -339,7 +348,6 @@ func (p *Parser) header(mh *MimeHeader) (err error) {
 	var name string
 
 	defer func() {
-		fmt.Println(mh.Headers)
 		p.accept.Reset()
 		if val := mh.Headers.Get("Content-Transfer-Encoding"); val != "" {
 			mh.TransferEncoding = val
@@ -667,89 +675,65 @@ func (p *Parser) parameter() (attribute, value string, err error) {
 	}
 }
 
-func (p *Parser) body(mh *MimeHeader) (err error) {
-	var body bytes.Buffer
-
-	if mh.ContentBoundary != "" {
-		if end, err := p.boundary(mh.ContentBoundary); err != nil {
-			return err
-		} else {
-			fmt.Println("boundary end:", end)
-		}
-		return
-	} else {
-		for {
-
-			p.next()
-			if p.ch == 0 {
-				return io.EOF
-			}
-			if p.ch == '\n' && p.peek() == '\n' {
-				p.next()
-				return
-			}
-
-			body.WriteByte(p.ch)
-
-		}
-	}
-
-}
-
 // isBranch determines if we should branch this part, when building
 // the mime tree
 func (p *Parser) isBranch(part *MimeHeader, parent *MimeHeader) bool {
 	ct := part.ContentType
+	if ct == nil {
+		return false
+	}
+	if part.ContentBoundary == "" {
+		return false
+	}
+
+	// tolerate some incorrect messages that re-use the identical content-boundary
 	if parent != nil && ct.superType != "message" {
 		if parent.ContentBoundary == part.ContentBoundary {
 			return false
 		}
 	}
-	if ct != nil {
-		if ct.superType == "multipart" ||
-			ct.superType == "message" {
-			return true
-		}
+
+	// branch on these superTypes
+	if ct.superType == "multipart" ||
+		ct.superType == "message" {
+		return true
 	}
 	return false
 }
 
+// multi finds the boundary and call back to mime() itself
 func (p *Parser) multi(part *MimeHeader, depth string) (err error) {
 	if part.ContentType != nil {
+		// scan until the start of the boundary
 		if part.ContentType.superType == "multipart" {
 			if end, bErr := p.boundary(part.ContentBoundary); bErr != nil {
 				return bErr
 			} else if end {
-
 				part.EndingPosBody = p.lastBoundaryPos
 				return
 			}
 		}
-		if part.ContentType.superType == "message" ||
-			part.ContentType.superType == "multipart" {
-			err = p.mime(part, depth)
-			if err != nil {
-
-				return err
-			}
-
+		// call back to mime() to start working on a new branch
+		err = p.mime(part, depth)
+		if err != nil {
+			return err
 		}
-
 	}
 	return
 }
 
+// mime scans the mime content and builds the mime-part tree in
+// p.Parts on-the-fly, as more bytes get fed in.
 func (p *Parser) mime(parent *MimeHeader, depth string) (err error) {
 
 	count := 1
 	for {
-
-		var part MimeHeader
-
-		part = *newMimeHeader()
+		part := newMimeHeader()
 		part.StartingPos = p.msgPos
+
+		// parse the headers
 		if p.ch >= 33 && p.ch <= 126 {
-			err = p.header(&part)
+			err = p.header(part)
 			if err != nil {
 				return err
 			}
@@ -760,30 +744,50 @@ func (p *Parser) mime(parent *MimeHeader, depth string) (err error) {
 			p.next()
 			p.next()
 		}
-		if part.ContentBoundary == "" {
+
+		// inherit the content boundary from parent if not present
+		if part.ContentBoundary == "" && parent != nil {
 			part.ContentBoundary = parent.ContentBoundary
 		}
 
+		// record the part
 		part.StartingPosBody = p.msgPos
 		partID := strconv.Itoa(count)
 		if depth != "" {
 			partID = depth + "." + strconv.Itoa(count)
 		}
-		p.addPart(&part, partID)
+		p.addPart(part, partID)
 
-		if p.isBranch(&part, parent) {
-
-			err = p.multi(&part, partID)
+		// build the mime tree recursively
+		if p.isBranch(part, parent) {
+			err = p.multi(part, partID)
 			part.EndingPosBody = p.lastBoundaryPos
 			if err != nil {
 				break
 			}
+		}
 
+		// if we didn't branch & we're still at the root (not a mime email)
+		if parent == nil {
+			for {
+				// keep scanning until the end
+				p.next()
+				if p.ch == 0 {
+					break
+				}
+			}
+			part.EndingPosBody = p.msgPos
+			err = NotMime
+			return
 		}
+
+		// after we return from the lower branches (if there were any)
+		// we walk each of the siblings of the parent
 		if end, bErr := p.boundary(parent.ContentBoundary); bErr != nil {
 			part.EndingPosBody = p.lastBoundaryPos
 			return bErr
 		} else if end {
+			// the last sibling
 			part.EndingPosBody = p.lastBoundaryPos
 			return
 		}
@@ -796,8 +800,14 @@ func (p *Parser) mime(parent *MimeHeader, depth string) (err error) {
 // Close tells the MIME Parser there's no more data & waits for it to return a result
 // it will return an io.EOF error if no error with parsing MIME was detected
 // Close is not concurrency safe, must be called synchronously after all calls to
-// Read have completed
+// Parse have completed
 func (p *Parser) Close() error {
+	defer func() {
+		p.lastBoundaryPos = 0
+		p.pos = startPos
+		p.msgPos = 0
+		p.msgLine = 0
+	}()
 	if p.count > 0 {
 		for {
 			// dont't exit unless we get the result.
@@ -812,24 +822,18 @@ func (p *Parser) Close() error {
 		}
 	}
 	return nil
-
 }
 
-func (p *Parser) Open() error {
-	p.lastBoundaryPos = 0
-	p.pos = startPos
-	p.msgPos = 0
-	p.msgLine = 0
-	return nil
-
-}
-
-// Read takes a byte stream, and feeds it to the MIME Parser
+// Parse takes a byte stream, and feeds it to the MIME Parser, then
 // waits for the Parser to consume all input before returning.
-// The Parser will build
+// The Parser will build a parse tree in p.Parts
+// The reader doesn't decode any input. All it does
+// is collect information about where the different MIME parts
+// start and end, and other meta-data. This data can be used
+// by others later to determine how to store/display
+// the messages
 // returns error if there's a parse error
-
-func (p *Parser) Read(buf []byte) (int, error) {
+func (p *Parser) Parse(buf []byte) error {
 	defer func() {
 		p.count++
 	}()
@@ -841,17 +845,16 @@ func (p *Parser) Read(buf []byte) (int, error) {
 			fmt.Println("mine() ret", err)
 			p.result <- parserMsg{err}
 		}()
-		return p.pos + 1, nil
-
+		return nil
 	}
 	select {
 	case <-p.consumed: // wait for prev buf to be consumed
 		p.set(buf)
 		p.gotNewSlice <- true
-		return p.pos + 1, nil
+		return nil
 	case r := <-p.result:
 		// mime() has returned with a result
-		return p.pos + 1, r.err
+		return r.err
 	}
 
 }

mail/mime/reader_test.go → mail/mime/mime_test.go

@@ -363,6 +363,9 @@ Content-Disposition: attachment;
 ------_=_NextPart_003_01CBE272.13692C80--
 ------_=_NextPart_001_01CBE273.65A0E7AA--`
 
+// note: this mime has an error: the boundary for multipart/alternative is re-used.
+// it should use a new unique boundary marker, which then needs to be terminated after
+// the text/html part.
 var email3 = `MIME-Version: 1.0
 X-Mailer: MailBee.NET 8.0.4.428
 Subject: test subject
@@ -408,7 +411,7 @@ TmV4dFBhcnRfMDAwX0FFNkJfNzI1RTA5QUYuODhCN0Y5MzQtLQ0K
 `
 
 func TestNestedEmail(t *testing.T) {
-	email = email
+	email = email3
 	p.inject([]byte(email))
 
 	go func() {
@@ -416,6 +419,7 @@ func TestNestedEmail(t *testing.T) {
 		//panic("here")
 		//		*moo = *moo + 6
 
+		// for debugging deadlocks
 		//pprof.Lookup("goroutine").WriteTo(os.Stdout, 1)
 		//os.Exit(1)
 	}()
@@ -427,7 +431,7 @@ func TestNestedEmail(t *testing.T) {
 		email = replaceAtIndex(email, '#', p.Parts[part].StartingPos)
 		email = replaceAtIndex(email, '&', p.Parts[part].StartingPosBody)
 		email = replaceAtIndex(email, '*', p.Parts[part].EndingPosBody)
-		fmt.Println(p.Parts[part].Part + " " + strconv.Itoa(int(p.Parts[part].StartingPos)) + " " + strconv.Itoa(int(p.Parts[part].StartingPosBody)) + " " + strconv.Itoa(int(p.Parts[part].EndingPosBody)))
+		fmt.Println(p.Parts[part].Part + "  " + strconv.Itoa(int(p.Parts[part].StartingPos)) + "  " + strconv.Itoa(int(p.Parts[part].StartingPosBody)) + "  " + strconv.Itoa(int(p.Parts[part].EndingPosBody)))
 	}
 	fmt.Print(email)
 	//fmt.Println(strings.Index(email, "--D7F------------D7FD5A0B8AB9C65CCDBFA872--"))
@@ -440,3 +444,21 @@ func TestNestedEmail(t *testing.T) {
 func replaceAtIndex(str string, replacement rune, index uint) string {
 	return str[:index] + string(replacement) + str[index+1:]
 }
+
+var email4 = `Subject: test subject
+To: [email protected]
+
+This is not a an MIME email
+`
+
+func TestNonMineEmail(t *testing.T) {
+	p.inject([]byte(email4))
+	if err := p.mime(nil, ""); err != nil && err != NotMime && err != io.EOF {
+		t.Error(err)
+	} else {
+		for part := range p.Parts {
+			fmt.Println(p.Parts[part].Part + "  " + strconv.Itoa(int(p.Parts[part].StartingPos)) + "  " + strconv.Itoa(int(p.Parts[part].StartingPosBody)) + "  " + strconv.Itoa(int(p.Parts[part].EndingPosBody)))
+		}
+	}
+
+}