AUTH-1943 hooked up uploader to logger, added timestamp to session logs, add tests

awsuploader/directory_upload_manager.go

@@ -77,7 +77,7 @@ func (m *DirectoryUploadManager) sweep() {
 		checkTime := time.Now().Add(-time.Duration(retentionTime))
 
 		//delete the file it is stale
-		if info.ModTime().After(checkTime) {
+		if info.ModTime().Before(checkTime) {
 			os.Remove(path)
 			return nil
 		}

cmd/cloudflared/tunnel/cmd.go

@@ -370,6 +370,7 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
 
 		logger.Infof("ssh-server set")
 
+		logManager := sshlog.NewEmptyManager()
 		if c.IsSet(bucketNameFlag) && c.IsSet(regionNameFlag) && c.IsSet(accessKeyIDFlag) && c.IsSet(secretIDFlag) {
 			uploader, err := awsuploader.NewFileUploader(c.String(bucketNameFlag), c.String(regionNameFlag),
 				c.String(accessKeyIDFlag), c.String(secretIDFlag), c.String(sessionTokenIDFlag), c.String(s3URLFlag))
@@ -379,12 +380,12 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
 			}
 
 			os.Mkdir(sshLogFileDirectory, 0600)
+			logManager = sshlog.New(sshLogFileDirectory)
 
 			uploadManager := awsuploader.NewDirectoryUploadManager(logger, uploader, sshLogFileDirectory, 30*time.Minute, shutdownC)
 			uploadManager.Start()
 		}
 
-		logManager := sshlog.New()
 		sshServerAddress := "127.0.0.1:" + c.String(sshPortFlag)
 		server, err := sshserver.New(logManager, logger, sshServerAddress, shutdownC, c.Duration(sshIdleTimeoutFlag), c.Duration(sshMaxTimeoutFlag))
 		if err != nil {

sshlog/empty_manager.go

@@ -0,0 +1,37 @@
+package sshlog
+
+import (
+	"io"
+
+	"github.com/sirupsen/logrus"
+)
+
+//empty manager implements the Manager but does nothing (for testing and to disable logging unless the logs are set)
+type emptyManager struct {
+}
+
+type emptyWriteCloser struct {
+}
+
+// NewEmptyManager creates a new instance of a log empty log manager that does nothing
+func NewEmptyManager() Manager {
+	return &emptyManager{}
+}
+
+func (m *emptyManager) NewLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
+	return &emptyWriteCloser{}, nil
+}
+
+func (m *emptyManager) NewSessionLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
+	return &emptyWriteCloser{}, nil
+}
+
+// emptyWriteCloser
+
+func (w *emptyWriteCloser) Write(p []byte) (n int, err error) {
+	return 0, nil
+}
+
+func (w *emptyWriteCloser) Close() error {
+	return nil
+}

sshlog/go.capnp

@@ -0,0 +1,15 @@
+# Generate go.capnp.out with:
+# capnp compile -o- go.capnp > go.capnp.out
+# Must run inside this directory to preserve paths.
+
+@0xd12a1c51fedd6c88;
+
+annotation package(file) :Text;
+annotation import(file) :Text;
+annotation doc(struct, field, enum) :Text;
+annotation tag(enumerant) :Text;
+annotation notag(enumerant) :Void;
+annotation customtype(field) :Text;
+annotation name(struct, field, union, enum, enumerant, interface, method, param, annotation, const, group) :Text;
+
+$package("capnp");

sshlog/logger.go

@@ -2,6 +2,7 @@ package sshlog
 
 import (
 	"bufio"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -12,19 +13,22 @@ import (
 )
 
 const (
-	logTimeFormat = "2006-01-02T15-04-05.000"
+	logTimeFormat        = "2006-01-02T15-04-05.000"
-	megabyte      = 1024 * 1024
+	megabyte             = 1024 * 1024
+	defaultFileSizeLimit = 100 * megabyte
 )
 
 // Logger will buffer and write events to disk
 type Logger struct {
 	sync.Mutex
-	filename    string
+	filename      string
-	file        *os.File
+	file          *os.File
-	writeBuffer *bufio.Writer
+	writeBuffer   *bufio.Writer
-	logger      *logrus.Logger
+	logger        *logrus.Logger
-	done        chan struct{}
+	flushInterval time.Duration
-	once        sync.Once
+	maxFileSize   int64
+	done          chan struct{}
+	once          sync.Once
 }
 
 // NewLogger creates a Logger instance. A buffer is created that needs to be
@@ -34,16 +38,23 @@ type Logger struct {
 // logger variable is a logrus that will log all i/o, filesystem error etc, that
 // that shouldn't end execution of the logger, but are useful to report to the
 // caller.
-func NewLogger(filename string, logger *logrus.Logger) (*Logger, error) {
+func NewLogger(filename string, logger *logrus.Logger, flushInterval time.Duration, maxFileSize int64) (*Logger, error) {
+	if logger == nil {
+		return nil, errors.New("logger can't be nil")
+	}
 	f, err := os.OpenFile(filename, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.FileMode(0600))
 	if err != nil {
 		return nil, err
 	}
 	l := &Logger{filename: filename,
-		file:        f,
+		file:          f,
-		writeBuffer: bufio.NewWriter(f),
+		writeBuffer:   bufio.NewWriter(f),
-		logger:      logger,
+		logger:        logger,
-		done:        make(chan struct{})}
+		flushInterval: flushInterval,
+		maxFileSize:   maxFileSize,
+		done:          make(chan struct{}),
+	}
+
 	go l.writer()
 	return l, nil
 }
@@ -70,7 +81,7 @@ func (l *Logger) Close() error {
 // writer is the run loop that handles draining the write buffer and syncing
 // data to disk.
 func (l *Logger) writer() {
-	ticker := time.NewTicker(time.Second)
+	ticker := time.NewTicker(l.flushInterval)
 	defer ticker.Stop()
 	for {
 		select {
@@ -118,7 +129,7 @@ func (l *Logger) shouldRotate() bool {
 		return false
 	}
 
-	return info.Size() >= 100*megabyte
+	return info.Size() >= l.maxFileSize
 }
 
 // rotate creates a new logfile with the existing filename and renames the

sshlog/logger_test.go

@@ -0,0 +1,90 @@
+package sshlog
+
+import (
+	"io/ioutil"
+	"log"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+const logFileName = "test-logger.log"
+
+func createLogger(t *testing.T) *Logger {
+	os.Remove(logFileName)
+	l := logrus.New()
+	logger, err := NewLogger(logFileName, l, time.Millisecond, 1024)
+	if err != nil {
+		t.Fatal("couldn't create the logger!", err)
+	}
+	return logger
+}
+
+func TestWrite(t *testing.T) {
+	testStr := "hi"
+	logger := createLogger(t)
+	defer func() {
+		logger.Close()
+		os.Remove(logFileName)
+	}()
+
+	logger.Write([]byte(testStr))
+	time.Sleep(2 * time.Millisecond)
+	data, err := ioutil.ReadFile(logFileName)
+	if err != nil {
+		t.Fatal("couldn't read the log file!", err)
+	}
+	checkStr := string(data)
+	if checkStr != testStr {
+		t.Fatal("file data doesn't match!")
+	}
+}
+
+func TestFilenameRotation(t *testing.T) {
+	newName := rotationName("dir/bob/acoolloggername.log")
+
+	dir := filepath.Dir(newName)
+	if dir != "dir/bob" {
+		t.Fatal("rotation name doesn't respect the directory filepath:", newName)
+	}
+
+	filename := filepath.Base(newName)
+	if !strings.HasPrefix(filename, "acoolloggername") {
+		t.Fatal("rotation filename is wrong:", filename)
+	}
+
+	ext := filepath.Ext(newName)
+	if ext != ".log" {
+		t.Fatal("rotation file extension is wrong:", ext)
+	}
+}
+
+func TestRotation(t *testing.T) {
+	logger := createLogger(t)
+
+	for i := 0; i < 2000; i++ {
+		logger.Write([]byte("a string for testing rotation\n"))
+	}
+	logger.Close()
+
+	count := 0
+	filepath.Walk(".", func(path string, info os.FileInfo, err error) error {
+		if err != nil || info.IsDir() {
+			return nil
+		}
+		if strings.HasPrefix(info.Name(), "test-logger") {
+			log.Println("deleting: ", path)
+			os.Remove(path)
+			count++
+		}
+		return nil
+	})
+	if count < 2 {
+		t.Fatal("rotation didn't roll files:", count)
+	}
+
+}

sshlog/manager.go

@@ -2,6 +2,8 @@ package sshlog
 
 import (
 	"io"
+	"path/filepath"
+	"time"
 
 	"github.com/sirupsen/logrus"
 )
@@ -9,15 +11,24 @@ import (
 // Manager be managing logs bruh
 type Manager interface {
 	NewLogger(string, *logrus.Logger) (io.WriteCloser, error)
+	NewSessionLogger(string, *logrus.Logger) (io.WriteCloser, error)
 }
 
-type manager struct{}
+type manager struct {
+	baseDirectory string
+}
 
 // New creates a new instance of a log manager
-func New() Manager {
+func New(baseDirectory string) Manager {
-	return &manager{}
+	return &manager{
+		baseDirectory: baseDirectory,
+	}
 }
 
 func (m *manager) NewLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
-	return NewLogger(name, logger)
+	return NewLogger(filepath.Join(m.baseDirectory, name), logger, time.Second, defaultFileSizeLimit)
+}
+
+func (m *manager) NewSessionLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
+	return NewSessionLogger(name, logger, time.Second, defaultFileSizeLimit)
 }

sshlog/session_log.capnp

@@ -0,0 +1,9 @@
+using Go = import "go.capnp";
+@0x8f43375162194466;
+$Go.package("sshlog");
+$Go.import("github.com/cloudflare/cloudflared/sshlog");
+
+struct SessionLog {
+    timestamp @0 :Text;
+    content @1 :Data;
+}

sshlog/session_log.capnp.go

@@ -0,0 +1,110 @@
+// Code generated by capnpc-go. DO NOT EDIT.
+
+package sshlog
+
+import (
+	capnp "zombiezen.com/go/capnproto2"
+	text "zombiezen.com/go/capnproto2/encoding/text"
+	schemas "zombiezen.com/go/capnproto2/schemas"
+)
+
+type SessionLog struct{ capnp.Struct }
+
+// SessionLog_TypeID is the unique identifier for the type SessionLog.
+const SessionLog_TypeID = 0xa13a07c504a5ab64
+
+func NewSessionLog(s *capnp.Segment) (SessionLog, error) {
+	st, err := capnp.NewStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2})
+	return SessionLog{st}, err
+}
+
+func NewRootSessionLog(s *capnp.Segment) (SessionLog, error) {
+	st, err := capnp.NewRootStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2})
+	return SessionLog{st}, err
+}
+
+func ReadRootSessionLog(msg *capnp.Message) (SessionLog, error) {
+	root, err := msg.RootPtr()
+	return SessionLog{root.Struct()}, err
+}
+
+func (s SessionLog) String() string {
+	str, _ := text.Marshal(0xa13a07c504a5ab64, s.Struct)
+	return str
+}
+
+func (s SessionLog) Timestamp() (string, error) {
+	p, err := s.Struct.Ptr(0)
+	return p.Text(), err
+}
+
+func (s SessionLog) HasTimestamp() bool {
+	p, err := s.Struct.Ptr(0)
+	return p.IsValid() || err != nil
+}
+
+func (s SessionLog) TimestampBytes() ([]byte, error) {
+	p, err := s.Struct.Ptr(0)
+	return p.TextBytes(), err
+}
+
+func (s SessionLog) SetTimestamp(v string) error {
+	return s.Struct.SetText(0, v)
+}
+
+func (s SessionLog) Content() ([]byte, error) {
+	p, err := s.Struct.Ptr(1)
+	return []byte(p.Data()), err
+}
+
+func (s SessionLog) HasContent() bool {
+	p, err := s.Struct.Ptr(1)
+	return p.IsValid() || err != nil
+}
+
+func (s SessionLog) SetContent(v []byte) error {
+	return s.Struct.SetData(1, v)
+}
+
+// SessionLog_List is a list of SessionLog.
+type SessionLog_List struct{ capnp.List }
+
+// NewSessionLog creates a new list of SessionLog.
+func NewSessionLog_List(s *capnp.Segment, sz int32) (SessionLog_List, error) {
+	l, err := capnp.NewCompositeList(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2}, sz)
+	return SessionLog_List{l}, err
+}
+
+func (s SessionLog_List) At(i int) SessionLog { return SessionLog{s.List.Struct(i)} }
+
+func (s SessionLog_List) Set(i int, v SessionLog) error { return s.List.SetStruct(i, v.Struct) }
+
+func (s SessionLog_List) String() string {
+	str, _ := text.MarshalList(0xa13a07c504a5ab64, s.List)
+	return str
+}
+
+// SessionLog_Promise is a wrapper for a SessionLog promised by a client call.
+type SessionLog_Promise struct{ *capnp.Pipeline }
+
+func (p SessionLog_Promise) Struct() (SessionLog, error) {
+	s, err := p.Pipeline.Struct()
+	return SessionLog{s}, err
+}
+
+const schema_8f43375162194466 = "x\xda\x120q`\x12d\x8dg`\x08dae\xfb" +
+	"\x9f\xb2z)\xcbQv\xab\x85\x0c\x82B\x8c\xff\xd3\\" +
+	"$\x93\x02\xcd\x9d\xfb\x19X\x99\xd8\x19\x18\x04E_\x09" +
+	"*\x82h\xd9r\x06\xc6\xff\xc5\xa9\xc5\xc5\x99\xf9y\xf1" +
+	"L9\xf9\xe9z\xc9\x89\x05y\x05V\xc1`!\xfe<" +
+	"\x9f\xfc\xf4\x00F\xc6@\x0ef\x16\x06\x06\x16F\x06\x06" +
+	"A\xcd \x06\x86@\x0df\xc6@\x13&FAFF" +
+	"\x11F\x90\xa0\xa1\x13\x03C\xa0\x0e3c\xa0\x05\x13\xe3" +
+	"\xff\x92\xcc\xdc\xd4\xe2\x92\xc4\\\x06\xc6\x02F\x1e\x06&" +
+	"F\x1e\x06\xc6\xfa\xe4\xfc\xbc\x92\xd4\xbc\x12F^\x06&" +
+	"F^\x06F@\x00\x00\x00\xff\xff\xdaK$\x1a"
+
+func init() {
+	schemas.Register(schema_8f43375162194466,
+		0xa13a07c504a5ab64)
+}

sshlog/session_logger.go

@@ -0,0 +1,71 @@
+package sshlog
+
+import (
+	"time"
+
+	"github.com/sirupsen/logrus"
+	capnp "zombiezen.com/go/capnproto2"
+	"zombiezen.com/go/capnproto2/pogs"
+)
+
+// SessionLogger will buffer and write events to disk using capnp proto for session replay
+type SessionLogger struct {
+	logger  *Logger
+	encoder *capnp.Encoder
+}
+
+type sessionLogData struct {
+	Timestamp string // The UTC timestamp of when the log occurred
+	Content   []byte // The shell output
+}
+
+// NewSessionLogger creates a new session logger by encapsulating a Logger object and writing capnp encoded messages to it
+func NewSessionLogger(filename string, logger *logrus.Logger, flushInterval time.Duration, maxFileSize int64) (*SessionLogger, error) {
+	l, err := NewLogger(filename, logger, flushInterval, maxFileSize)
+	if err != nil {
+		return nil, err
+	}
+	sessionLogger := &SessionLogger{
+		logger:  l,
+		encoder: capnp.NewEncoder(l),
+	}
+	return sessionLogger, nil
+}
+
+// Writes to a log buffer. Implements the io.Writer interface.
+func (l *SessionLogger) Write(p []byte) (n int, err error) {
+	return l.writeSessionLog(&sessionLogData{
+		Timestamp: time.Now().UTC().Format(time.RFC3339),
+		Content:   p,
+	})
+}
+
+// Close drains anything left in the buffer and cleans up any resources still
+// in use.
+func (l *SessionLogger) Close() error {
+	return l.logger.Close()
+}
+
+func (l *SessionLogger) writeSessionLog(p *sessionLogData) (int, error) {
+	msg, seg, err := capnp.NewMessage(capnp.SingleSegment(nil))
+	if err != nil {
+		return 0, err
+	}
+	log, err := NewRootSessionLog(seg)
+	if err != nil {
+		return 0, err
+	}
+	log.SetTimestamp(p.Timestamp)
+	log.SetContent(p.Content)
+
+	if err := l.encoder.Encode(msg); err != nil {
+		return 0, err
+	}
+	return len(p.Content), nil
+}
+
+func unmarshalSessionLog(s SessionLog) (*sessionLogData, error) {
+	p := new(sessionLogData)
+	err := pogs.Extract(p, SessionLog_TypeID, s.Struct)
+	return p, err
+}

sshlog/session_logger_test.go

@@ -0,0 +1,69 @@
+package sshlog
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	capnp "zombiezen.com/go/capnproto2"
+)
+
+const sessionLogFileName = "test-session-logger.log"
+
+func createSessionLogger(t *testing.T) *SessionLogger {
+	os.Remove(sessionLogFileName)
+	l := logrus.New()
+	logger, err := NewSessionLogger(sessionLogFileName, l, time.Millisecond, 1024)
+	if err != nil {
+		t.Fatal("couldn't create the logger!", err)
+	}
+	return logger
+}
+
+func TestSessionLogWrite(t *testing.T) {
+	testStr := "hi"
+	logger := createSessionLogger(t)
+	defer func() {
+		logger.Close()
+		os.Remove(sessionLogFileName)
+	}()
+
+	logger.Write([]byte(testStr))
+	time.Sleep(2 * time.Millisecond)
+	f, err := os.Open(sessionLogFileName)
+	if err != nil {
+		t.Fatal("couldn't read the log file!", err)
+	}
+	defer f.Close()
+
+	msg, err := capnp.NewDecoder(f).Decode()
+	if err != nil {
+		t.Fatal("couldn't read the capnp msg file!", err)
+	}
+
+	sessionLog, err := ReadRootSessionLog(msg)
+	if err != nil {
+		t.Fatal("couldn't read the session log from the msg!", err)
+	}
+
+	timeStr, err := sessionLog.Timestamp()
+	if err != nil {
+		t.Fatal("couldn't read the Timestamp field!", err)
+	}
+
+	_, terr := time.Parse(time.RFC3339, timeStr)
+	if terr != nil {
+		t.Fatal("couldn't parse the Timestamp into the expected RFC3339 format", terr)
+	}
+
+	data, err := sessionLog.Content()
+	if err != nil {
+		t.Fatal("couldn't read the Content field!", err)
+	}
+
+	checkStr := string(data)
+	if checkStr != testStr {
+		t.Fatal("file data doesn't match!")
+	}
+}

sshserver/authentication.go

@@ -98,7 +98,7 @@ func getCACert() (ssh.PublicKey, error) {
 	caCertPath := path.Join(systemConfigPath, "ca.pub")
 	caCertBytes, err := ioutil.ReadFile(caCertPath)
 	if err != nil {
-		return nil, errors.Wrap(err, fmt.Sprintf("Failed to load CA certertificate %s", caCertPath))
+		return nil, errors.Wrap(err, fmt.Sprintf("Failed to load CA certificate %s", caCertPath))
 	}
 	caCert, _, _, _, err := ssh.ParseAuthorizedKey(caCertBytes)
 	if err != nil {

sshserver/sshserver_unix.go

@@ -146,16 +146,16 @@ func (s *SSHServer) connectionHandler(session ssh.Session) {
 	defer pr.Close()
 	defer pw.Close()
 
-	logger, err := s.logManager.NewLogger(fmt.Sprintf("%s-session.log", sessionID), s.logger)
+	sessionLogger, err := s.logManager.NewSessionLogger(fmt.Sprintf("%s-session.log", sessionID), s.logger)
 	if err != nil {
 		if _, err := io.WriteString(session, "Failed to create log\n"); err != nil {
 			s.logger.WithError(err).Error("Failed to create log: Failed to write to SSH session")
 		}
 		s.CloseSession(session)
 	}
-	defer logger.Close()
+	defer sessionLogger.Close()
 	go func() {
-		io.Copy(logger, pr)
+		io.Copy(sessionLogger, pr)
 	}()
 
 	// Write outgoing command output to both the command recorder, and remote user