Initial work on adding audit capabilities to lottip

chat/client.go

@@ -6,8 +6,8 @@ import (
 )
 
 const (
-	pingPeriod          = time.Millisecond * 5000
-	writeDeadlinePeriod = time.Second * 2
+	pingPeriod          = time.Millisecond * 60000
+	writeDeadlinePeriod = time.Second * 60
 )
 
 // Client represents client(browser) connected via websocket

client_to_server_handler.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"io"
+	"lottip/chat"
+	"lottip/protocol"
+)
+
+type ClientToServerHandler struct {
+	connInfo      *ConnectionInfo
+	queryChan     chan chat.Cmd
+	connStateChan chan chat.ConnState
+	server        io.Writer
+}
+
+// CLIENT to SERVER
+func (pp *ClientToServerHandler) Write(buffer []byte) (n int, err error) {
+	extractPacketsFromBuffer(pp.connInfo, pp.connInfo.clientPacketFragment, buffer, func(packet []byte) {
+		// Switch based on the state
+		fsm := pp.connInfo.fsm
+
+		if ok, _ := fsm.IsInState(StateIdle); ok {
+			pp.connStateChan <- chat.ConnState{pp.connInfo.ConnId, protocol.ConnStateFinished}
+		} else if ok, _ := fsm.IsInState(StateAuthRequested); ok {
+			fsm.Fire(MsgLogin, packet)
+		} else {
+			if len(packet) < 4 {
+				LogInvalid(pp.connInfo, "?Request?", packet)
+			} else {
+				// We are processing queries
+				fsm.Fire(GetPacketType(packet), packet)
+			}
+		}
+	})
+
+	return len(buffer), nil
+}

docker/.gitattributes

@@ -0,0 +1,2 @@
+Dockerfile text eol=LF
+*.sh       text eol=LF

docker/Dockerfile

@@ -1,15 +1,17 @@
 FROM frolvlad/alpine-go
 
-
 RUN apk update && apk upgrade && \
     apk add --no-cache bash git
 
-RUN go get github.com/orderbynull/lottip && \
-    go get github.com/mjibson/esc && \
-    go install github.com/mjibson/esc && \
-    cd /root/go/src/github.com/orderbynull/lottip && \
-    /root/go/bin/esc -o fs.go -prefix web -include=".*\.css|.*\.js|.*\.html|.*\.png" web && \
-    go build
+RUN git clone http://github.com/rkoshy/lottip
+WORKDIR lottip
+RUN go get github.com/mjibson/esc
+RUN go install github.com/mjibson/esc
+RUN /root/go/bin/esc -o fs.go -prefix web -include=".*\.css|.*\.js|.*\.html|.*\.png" web
+RUN go get
+RUN go install
+WORKDIR /
+RUN rm -rf /root/go/bin/esc && rm -rf lottip
 
 ADD run.sh /run.sh
 RUN chmod +x /run.sh
@@ -18,6 +20,3 @@ EXPOSE 4041
 EXPOSE 9999
 
 CMD /run.sh
-
-
-

docker/run.sh

@@ -14,11 +14,30 @@ LOTTIP_GUI="${LOTTIP_GUI:-0.0.0.0:9999}"
 # MySQL DSN (credentials)
 LOTTIP_DSN="${LOTTIP_DSN:-root:root@/}"
 
+# Log directory
+LOTTIP_LOG_DIRECTORY="${LOTTIP_LOG_DIRECTORY:-./logs}"
 
-# Run lottip
+# Log filename
+LOTTIP_LOGFILE="${LOTTIP_LOGFILE:-logfile.log}"
+
+# Query Log filename
+LOTTIP_QUERY_LOGFILE="${LOTTIP_QUERY_LOGFILE:-queries.log}"
 
-/root/go/src/github.com/orderbynull/lottip/lottip \
+# Enable
+if [ "${LOTTIP_CONSOLE_LOGGING}" == "true" ];
+then
+  LOTTIP_CONSOLE_LOGGING="--enable-console-logging"
+else
+  LOTTIP_CONSOLE_LOGGING=""
+fi
+
+# Run lottip
+/root/go/bin/lottip \
   --proxy "$LOTTIP_PROXY" \
   --mysql "$LOTTIP_MYSQL" \
-  --gui "$LOTTIP_GUI" \
-  --mysql-dsn "$LOTTIP_DSN"
+  --gui-addr "$LOTTIP_GUI" \
+  --mysql-dsn "$LOTTIP_DSN" \
+  "$LOTTIP_CONSOLE_LOGGING" \
+  --query-log-file "$LOTTIP_QUERY_LOGFILE" \
+  --log-file "$LOTTIP_LOGFILE" \
+  --log-directory "$LOTTIP_LOG_DIRECTORY"

fsm.go

@@ -0,0 +1,207 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/qmuntal/stateless"
+	"lottip/chat"
+	"lottip/protocol"
+	"net"
+	"reflect"
+	"time"
+)
+
+type MySQLProtocolFSM struct {
+	*stateless.StateMachine
+	connectionInfo *ConnectionInfo
+	clientConn     net.Conn
+	serverConn     net.Conn
+}
+
+const (
+	StateIdle          = "Idle"
+	StateAuthRequested = "AuthRequested"
+	StateAuthSent      = "AuthSent"
+	StateAuthorized    = "stateAuthorized"
+	StateUnauthorized  = "stateUnauthorized"
+)
+
+const (
+	PacketReceived    = "PacketReceived"
+	MsgServerHello    = "ServerHello"
+	MsgLogin          = "Login"
+	MsgOK             = "OK"
+	MsgERROR          = "ERROR"
+	MsgServerToClient = "MsgServerToClient"
+)
+
+func CreateStateMachine(ci *ConnectionInfo, clientConn net.Conn, serverConn net.Conn, cmdChan chan chat.Cmd, resultChan chan chat.CmdResult, stateChan chan chat.ConnState) *MySQLProtocolFSM {
+	fsm := stateless.NewStateMachine(StateIdle)
+
+	fsm.SetTriggerParameters(PacketReceived, reflect.TypeOf([]byte{}))
+	fsm.OnUnhandledTrigger(func(ctx context.Context, state stateless.State, trigger stateless.Trigger, unmetGuards []string) error {
+		LogOther(ci, "fsm - Unhandled event", "%d in state %s", trigger, state)
+		return nil
+	})
+
+	fsm.Configure(StateIdle).
+		Permit(MsgServerHello, StateAuthRequested, func(ctx context.Context, args ...interface{}) bool {
+			// Server's initial response asking for loging info -- passthrough after changing compression flag
+			packet := args[0].([]byte)
+			index := 4
+			if packet[index] == 0x0A {
+				// Valid protocol, skip server version
+				for ; packet[index] != 0 && index < len(packet); index++ {
+				}
+				index++
+				// Skip thread-id
+				index += 4
+				// Skip salt
+				for ; packet[index] != 0 && index < len(packet); index++ {
+				}
+				index++
+
+				packet[index] = packet[index] & 0xDF
+				LogResponse(ci, packet, "Handshake", "Server Handshake/Challenge response (forcing uncompressed protocol)")
+
+				//LogOther(ci, "fsm - Writing to client", "% x", packet)
+				clientConn.Write(packet)
+				return true
+			} else {
+				LogOther(ci, "INVALID PROTOCOL", "%x is not a supported protocol version", packet[4])
+
+				clientConn.Close()
+				serverConn.Close()
+				return false
+			}
+		})
+
+	fsm.Configure(StateAuthRequested).
+		Permit(MsgLogin, StateAuthSent, func(ctx context.Context, args ...interface{}) bool {
+			// Client's Auth Info
+			packet := args[0].([]byte)
+
+			start := 3 + 1 + 2 + 2 + 4 + 1
+			for ; packet[start] == 0; start++ {
+			}
+			stop := start + 1
+			for ; packet[stop] != 0; stop++ {
+			}
+			ci.User = string(packet[start:stop])
+			LogRequest(ci, packet, "Login", "Authorizing as user: '%s' (forcing uncompressed protocol)", ci.User)
+
+			// Disable compression
+			packet[4] = packet[4] & 0xDF
+
+			//LogOther(ci, "fsm - Writing to server", "% x", packet)
+			serverConn.Write(packet)
+			return true
+		})
+
+	fsm.Configure(StateAuthSent).
+		Permit(MsgOK, StateAuthorized, func(ctx context.Context, args ...interface{}) bool {
+			packet := args[0].([]byte)
+			LogResponse(ci, packet, "Authorized", "Client auth successful")
+			//LogOther(ci, "fsm - Writing to client", "% x", packet)
+			clientConn.Write(packet)
+			return true
+		}).
+		Permit(MsgERROR, StateUnauthorized, func(ctx context.Context, args ...interface{}) bool {
+			// Server's initial response asking for loging info -- passthrough after changing compression flag
+			packet := args[0].([]byte)
+			LogResponse(ci, packet, "Unauthorized", "Client auth failed")
+			//LogOther(ci, "fsm - Writing to client", "% x", packet)
+			clientConn.Write(packet)
+			return true
+		})
+
+	fsm.Configure(StateAuthorized).
+		InternalTransition(protocol.ComChangeUser, func(ctx context.Context, args ...interface{}) error {
+			// Do not allow this!
+			packet := args[0].([]byte)
+			start := 3 + 1 + 2 + 2 + 4 + 1
+			for ; packet[start] == 0; start++ {
+			}
+			stop := start + 1
+			for ; packet[stop] != 0; stop++ {
+			}
+			LogRequest(ci, packet, "ChangeUser", "Will reject request to change user to %s", string(packet[start:stop]))
+			return nil
+		}).
+		InternalTransition(protocol.ComPing, func(ctx context.Context, args ...interface{}) error {
+			packet := args[0].([]byte)
+			LogRequest(ci, packet, "Ping")
+			serverConn.Write(packet)
+			return nil
+		}).
+		InternalTransition(protocol.ComCreateDB, logAndSendQueryToServer(ci, serverConn, "CreateDB", cmdChan)).
+		InternalTransition(protocol.ComDropDB, logAndSendQueryToServer(ci, serverConn, "DropDB", cmdChan)).
+		InternalTransition(protocol.ComShutdown, logAndDrop(ci, "Shutdown", cmdChan)).
+		InternalTransition(protocol.ComProcessKill, logAndSendQueryToServer(ci, serverConn, "ProcessKill", cmdChan)).
+		InternalTransition(protocol.ComQuery, logAndSendQueryToServer(ci, serverConn, "Query", cmdChan)).
+		InternalTransition(protocol.ComStmtPrepare, logAndSendQueryToServer(ci, serverConn, "StmtPrepare", cmdChan)).
+		InternalTransition(protocol.ComStmtExecute, logAndSendQueryToServer(ci, serverConn, "StmtExecute", cmdChan)).
+		InternalTransition(protocol.ComStmtClose, logAndSendQueryToServer(ci, serverConn, "StmtClose", cmdChan)).
+		InternalTransition(protocol.ComStmtSendLongData, logAndSendQueryToServer(ci, serverConn, "StmtSendLongData", cmdChan)).
+		InternalTransition(protocol.ComStmtReset, logAndSendQueryToServer(ci, serverConn, "StmtReset", cmdChan)).
+		InternalTransition(protocol.ComQuit, logAndSendQueryToServer(ci, serverConn, "Quit", cmdChan)).
+		InternalTransition(protocol.ComBinlogDump, logAndDrop(ci, "BinlogDump", cmdChan)).
+		InternalTransition(protocol.ComBinlogDump, logAndDrop(ci, "BinlogDump", cmdChan)).
+		InternalTransition(protocol.ComTableDump, logAndDrop(ci, "TableDump", cmdChan)).
+		InternalTransition(protocol.ComConnectOut, logAndDrop(ci, "ConnectOut", cmdChan)).
+		InternalTransition(MsgOK, func(ctx context.Context, args ...interface{}) error {
+			packet := args[0].([]byte)
+			LogResponse(ci, packet, "OK")
+			duration := fmt.Sprintf("%.3f", time.Since(*ci.timer).Seconds())
+			resultChan <- chat.CmdResult{ci.ConnId, ci.QueryId, protocol.ResponseOk, "", duration}
+			clientConn.Write(packet)
+			return nil
+		}).
+		InternalTransition(MsgERROR, func(ctx context.Context, args ...interface{}) error {
+			packet := args[0].([]byte)
+			LogResponse(ci, packet, "ERROR")
+			duration := fmt.Sprintf("%.3f", time.Since(*ci.timer).Seconds())
+			resultChan <- chat.CmdResult{ci.ConnId, ci.QueryId, protocol.ResponseErr, "", duration}
+			clientConn.Write(packet)
+			return nil
+		}).
+		InternalTransition(MsgServerToClient, func(ctx context.Context, args ...interface{}) error {
+			packet := args[0].([]byte)
+			LogResponsePacket(ci, packet)
+			clientConn.Write(packet)
+			return nil
+		})
+
+	return &MySQLProtocolFSM{fsm, ci, clientConn, serverConn}
+}
+
+func logAndSendQueryToServer(ci *ConnectionInfo, serverConn net.Conn, command string, cmdChan chan chat.Cmd) func(ctx context.Context, args ...interface{}) error {
+	return func(ctx context.Context, args ...interface{}) error {
+		packet := args[0].([]byte)
+		query := string(packet[5:])
+		LogRequest(ci, packet, command, query)
+
+		ci.QueryId++
+		*ci.timer = time.Now()
+
+		cmdChan <- chat.Cmd{ci.ConnId, ci.QueryId, "", query, nil, false}
+		//LogOther(ci, "fsm - Writing to server", "% x", packet)
+		serverConn.Write(packet)
+		return nil
+	}
+}
+
+func logAndDrop(ci *ConnectionInfo, command string, cmdChan chan chat.Cmd) func(ctx context.Context, args ...interface{}) error {
+	return func(ctx context.Context, args ...interface{}) error {
+		packet := args[0].([]byte)
+		query := string(packet[5:])
+		LogRequest(ci, packet, "BLOCKED:"+command, query)
+		cmdChan <- chat.Cmd{ci.ConnId, ci.QueryId, "", query, nil, false}
+		LogOther(ci, "fsm - Dropping packet", "% x", packet)
+		return nil
+	}
+}
+
+func (fsm *MySQLProtocolFSM) ProcessPacket(packet []byte) {
+	fsm.Fire(MsgERROR)
+}

go.mod

@@ -0,0 +1,24 @@
+module lottip
+
+go 1.18
+
+require (
+	github.com/go-sql-driver/mysql v1.6.0
+	github.com/gorilla/websocket v1.5.0
+	github.com/olekukonko/tablewriter v0.0.5
+)
+
+require (
+	github.com/mattn/go-colorable v0.1.12 // indirect
+	github.com/mattn/go-isatty v0.0.14 // indirect
+	github.com/mattn/go-runewidth v0.0.9 // indirect
+	golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6 // indirect
+)
+
+require (
+	github.com/kjk/dailyrotate v0.0.0-20220410204435-837f1ef47fc4
+	github.com/pubnative/mysqlproto-go v0.0.0-20210816144457-71d8293daef4
+	github.com/qmuntal/stateless v1.6.0
+	github.com/rs/zerolog v1.27.0
+	github.com/stretchr/testify v1.7.1 // indirect
+)

http.go

@@ -2,13 +2,13 @@ package main
 
 import (
 	"fmt"
-	"log"
+	"github.com/rs/zerolog/log"
+	"lottip/chat"
 	"net/http"
 
 	"encoding/json"
 	"github.com/gorilla/websocket"
 	"github.com/olekukonko/tablewriter"
-	"github.com/orderbynull/lottip/chat"
 )
 
 const (
@@ -23,7 +23,7 @@ func runHttpServer(hub *chat.Hub) {
 
 		conn, err := upgr.Upgrade(w, r, nil)
 		if err != nil {
-			log.Println(err)
+			log.Error().Err(err).Msg("Could not upgrade connection from " + r.RemoteAddr + " to websocket")
 			return
 		}
 
@@ -46,7 +46,7 @@ func runHttpServer(hub *chat.Hub) {
 	http.HandleFunc("/execute", func(w http.ResponseWriter, r *http.Request) {
 		err := r.ParseForm()
 		if err != nil {
-			log.Println(err)
+			log.Error().Err(err).Msg("Could not parse query execution request from " + r.RemoteAddr)
 			return
 		}
 
@@ -80,5 +80,8 @@ func runHttpServer(hub *chat.Hub) {
 
 	http.Handle(webRoute, http.FileServer(FS(*useLocalUI)))
 
-	log.Fatal(http.ListenAndServe(*guiAddr, nil))
+	err := http.ListenAndServe(*guiAddr, nil)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Could not start HTTP server at " + *guiAddr)
+	}
 }