83 lines
2.1 KiB
Go
83 lines
2.1 KiB
Go
package eventstream
|
|
|
|
import (
|
|
"bytes"
|
|
"io"
|
|
"time"
|
|
)
|
|
|
|
// MessageSigner signs event stream message header and payload byte pairs.
|
|
// Each invocation chains off the previous signature.
|
|
type MessageSigner interface {
|
|
SignMessage(headers, payload []byte, signingTime time.Time) ([]byte, error)
|
|
}
|
|
|
|
// SigningWriter wraps an io.WriteCloser and signs each event stream message
|
|
// frame written to it. Each Write call MUST contain exactly one complete
|
|
// encoded event stream message frame.
|
|
//
|
|
// The signing writer wraps each incoming frame in an outer event stream
|
|
// message with :date and :chunk-signature headers, then encodes the outer
|
|
// message to the underlying writer.
|
|
//
|
|
// Close sends a signed empty message to signal end-of-stream, then closes
|
|
// the underlying writer.
|
|
type SigningWriter struct {
|
|
writer io.WriteCloser
|
|
signer MessageSigner
|
|
encoder *Encoder
|
|
|
|
headersBuf bytes.Buffer
|
|
}
|
|
|
|
// NewSigningWriter returns a SigningWriter that signs frames and writes them
|
|
// to w.
|
|
func NewSigningWriter(w io.WriteCloser, signer MessageSigner) *SigningWriter {
|
|
return &SigningWriter{
|
|
writer: w,
|
|
signer: signer,
|
|
encoder: NewEncoder(),
|
|
}
|
|
}
|
|
|
|
// Write signs a complete event stream message frame and writes the signed
|
|
// outer envelope to the underlying writer.
|
|
func (s *SigningWriter) Write(frame []byte) (int, error) {
|
|
if err := s.signAndWrite(frame); err != nil {
|
|
return 0, err
|
|
}
|
|
return len(frame), nil
|
|
}
|
|
|
|
// Close sends a signed empty message to signal end-of-stream, then closes
|
|
// the underlying writer.
|
|
func (s *SigningWriter) Close() error {
|
|
if err := s.signAndWrite([]byte{}); err != nil {
|
|
_ = s.writer.Close()
|
|
return err
|
|
}
|
|
return s.writer.Close()
|
|
}
|
|
|
|
func (s *SigningWriter) signAndWrite(payload []byte) error {
|
|
now := time.Now().UTC()
|
|
|
|
var msg Message
|
|
msg.Headers.Set(DateHeader, TimestampValue(now))
|
|
msg.Payload = payload
|
|
|
|
s.headersBuf.Reset()
|
|
if err := EncodeHeaders(&s.headersBuf, msg.Headers); err != nil {
|
|
return err
|
|
}
|
|
|
|
sig, err := s.signer.SignMessage(s.headersBuf.Bytes(), payload, now)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
msg.Headers.Set(ChunkSignatureHeader, BytesValue(sig))
|
|
|
|
return s.encoder.Encode(s.writer, msg)
|
|
}
|