Feat 2k add auth to websockets (#690)

service/internal/acl/acl.go

@@ -71,6 +71,13 @@ func (u *AuthenticatedUser) matchesUsergroupAcl(matchUsergroups []string, sep st
 	return false
 }
 
+type UnauthenticatedUser struct {
+	Username  string
+	Usergroup string
+	Provider  string
+	Sid       string
+}
+
 func logAclNotMatched(cfg *config.Config, aclFunction string, user *AuthenticatedUser, action *config.Action, acl *config.AccessControlList) {
 	if cfg.LogDebugOptions.AclNotMatched {
 		log.WithFields(log.Fields{
@@ -194,6 +201,29 @@ func getMetadataKeyOrEmpty(md metadata.MD, key string) string {
 	return ""
 }
 
+func UserFromUnauthenticatedUser(cfg *config.Config, unauthenticatedUser UnauthenticatedUser) *AuthenticatedUser {
+	ret := &AuthenticatedUser{}
+	ret.Username = unauthenticatedUser.Username
+	ret.UsergroupLine = unauthenticatedUser.Usergroup
+	ret.Provider = unauthenticatedUser.Provider
+	ret.SID = unauthenticatedUser.Sid
+
+	if ret.Username == "" {
+		ret = UserGuest(cfg)
+	}
+
+	buildUserAcls(cfg, ret)
+
+	log.WithFields(log.Fields{
+		"username":      ret.Username,
+		"usergroupLine": ret.UsergroupLine,
+		"provider":      ret.Provider,
+		"acls":          ret.Acls,
+	}).Debugf("UserFromUnauthenticatedUser")
+
+	return ret
+}
+
 // UserFromContext tries to find a user from a grpc context
 func UserFromContext(ctx context.Context, cfg *config.Config) *AuthenticatedUser {
 	var ret *AuthenticatedUser

service/internal/executor/executor.go

@@ -132,9 +132,9 @@ func DefaultExecutor(cfg *config.Config) *Executor {
 }
 
 type listener interface {
-	OnExecutionStarted(logEntry *InternalLogEntry)
-	OnExecutionFinished(logEntry *InternalLogEntry)
-	OnOutputChunk(o []byte, executionTrackingId string)
+	OnExecutionStarted(logEntry *InternalLogEntry, action *config.Action)
+	OnExecutionFinished(logEntry *InternalLogEntry, action *config.Action)
+	OnOutputChunk(o []byte, executionTrackingId string, logEntry *InternalLogEntry, action *config.Action)
 	OnActionMapRebuilt()
 }
 
@@ -509,13 +509,13 @@ func stepLogFinish(req *ExecutionRequest) bool {
 
 func notifyListenersFinished(req *ExecutionRequest) {
 	for _, listener := range req.executor.listeners {
-		listener.OnExecutionFinished(req.logEntry)
+		listener.OnExecutionFinished(req.logEntry, req.Action)
 	}
 }
 
 func notifyListenersStarted(req *ExecutionRequest) {
 	for _, listener := range req.executor.listeners {
-		listener.OnExecutionStarted(req.logEntry)
+		listener.OnExecutionStarted(req.logEntry, req.Action)
 	}
 }
 
@@ -532,7 +532,7 @@ type OutputStreamer struct {
 
 func (ost *OutputStreamer) Write(o []byte) (n int, err error) {
 	for _, listener := range ost.Req.executor.listeners {
-		listener.OnOutputChunk(o, ost.Req.TrackingID)
+		listener.OnOutputChunk(o, ost.Req.TrackingID, ost.Req.logEntry, ost.Req.Action)
 	}
 
 	return ost.output.Write(o)

service/internal/httpservers/restapi.go

@@ -2,6 +2,9 @@ package httpservers
 
 import (
 	"context"
+	"net/http"
+	"strings"
+
 	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
@@ -9,11 +12,10 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/reflect/protoreflect"
-	"net/http"
-	"strings"
 
 	apiv1 "github.com/OliveTin/OliveTin/gen/grpc/olivetin/api/v1"
 
+	"github.com/OliveTin/OliveTin/internal/acl"
 	config "github.com/OliveTin/OliveTin/internal/config"
 	cors "github.com/OliveTin/OliveTin/internal/cors"
 )
@@ -49,42 +51,50 @@ func parseHttpHeaderForAuth(req *http.Request) (string, string) {
 }
 
 //gocyclo:ignore
-func parseRequestMetadata(ctx context.Context, req *http.Request) metadata.MD {
-	username := ""
-	usergroup := ""
-	provider := "unknown"
-	sid := ""
+func authHttpRequest(req *http.Request) acl.UnauthenticatedUser {
+	ret := acl.UnauthenticatedUser{
+		Username:  "",
+		Usergroup: "",
+		Provider:  "unknown",
+		Sid:       "",
+	}
 
 	if cfg.AuthJwtHeader != "" {
-		username, usergroup = parseJwtHeader(req)
-		provider = "jwt-header"
+		ret.Username, ret.Usergroup = parseJwtHeader(req)
+		ret.Provider = "jwt-header"
 	}
 
-	if cfg.AuthJwtCookieName != "" {
-		username, usergroup = parseJwtCookie(req)
-		provider = "jwt-cookie"
+	if cfg.AuthJwtCookieName != "" && ret.Username == "" {
+		ret.Username, ret.Usergroup = parseJwtCookie(req)
+		ret.Provider = "jwt-cookie"
 	}
 
-	if cfg.AuthHttpHeaderUsername != "" && username == "" {
-		username, usergroup = parseHttpHeaderForAuth(req)
-		provider = "http-header"
+	if cfg.AuthHttpHeaderUsername != "" && ret.Username == "" {
+		ret.Username, ret.Usergroup = parseHttpHeaderForAuth(req)
+		ret.Provider = "http-header"
 	}
 
-	if len(cfg.AuthOAuth2Providers) > 0 && username == "" {
-		username, usergroup, sid = parseOAuth2Cookie(req)
-		provider = "oauth2"
+	if len(cfg.AuthOAuth2Providers) > 0 && ret.Username == "" {
+		ret.Username, ret.Usergroup, ret.Sid = parseOAuth2Cookie(req)
+		ret.Provider = "oauth2"
 	}
 
-	if cfg.AuthLocalUsers.Enabled && username == "" {
-		username, usergroup, sid = parseLocalUserCookie(req)
-		provider = "local"
+	if cfg.AuthLocalUsers.Enabled && ret.Username == "" {
+		ret.Username, ret.Usergroup, ret.Sid = parseLocalUserCookie(req)
+		ret.Provider = "local"
 	}
 
+	return ret
+}
+
+func authHttpRequestToMetadata(ctx context.Context, req *http.Request) metadata.MD {
+	authMetadata := authHttpRequest(req)
+
 	md := metadata.New(map[string]string{
-		"username":  username,
-		"usergroup": usergroup,
-		"provider":  provider,
-		"sid":       sid,
+		"username":  authMetadata.Username,
+		"usergroup": authMetadata.Usergroup,
+		"provider":  authMetadata.Provider,
+		"sid":       authMetadata.Sid,
 	})
 
 	log.Tracef("api request metadata: %+v", md)
@@ -177,7 +187,7 @@ func startRestAPIServer(globalConfig *config.Config) error {
 func newMux() *runtime.ServeMux {
 	// The MarshalOptions set some important compatibility settings for the webui. See below.
 	mux := runtime.NewServeMux(
-		runtime.WithMetadata(parseRequestMetadata),
+		runtime.WithMetadata(authHttpRequestToMetadata),
 		runtime.WithForwardResponseOption(forwardResponseHandler),
 		runtime.WithMarshalerOption(runtime.MIMEWildcard, &runtime.HTTPBodyMarshaler{
 			Marshaler: &runtime.JSONPb{

service/internal/httpservers/singleFrontend.go

@@ -9,12 +9,12 @@ away, and several other issues.
 */
 
 import (
-	config "github.com/OliveTin/OliveTin/internal/config"
-	"github.com/OliveTin/OliveTin/internal/websocket"
-	log "github.com/sirupsen/logrus"
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+
+	config "github.com/OliveTin/OliveTin/internal/config"
+	log "github.com/sirupsen/logrus"
 )
 
 func logDebugRequest(cfg *config.Config, source string, r *http.Request) {
@@ -53,7 +53,7 @@ func StartSingleHTTPFrontend(cfg *config.Config) {
 	mux.HandleFunc("/websocket", func(w http.ResponseWriter, r *http.Request) {
 		logDebugRequest(cfg, "ws  ", r)
 
-		websocket.HandleWebsocket(w, r)
+		handleWebsocket(w, r)
 	})
 
 	mux.HandleFunc("/oauth/login", handleOAuthLogin)

service/internal/websocket/websocket.go → service/internal/httpservers/websocket.go

@@ -1,10 +1,12 @@
-package websocket
+package httpservers
 
 import (
 	"net/http"
 	"sync"
 
 	apiv1 "github.com/OliveTin/OliveTin/gen/grpc/olivetin/api/v1"
+	"github.com/OliveTin/OliveTin/internal/acl"
+	"github.com/OliveTin/OliveTin/internal/config"
 	"github.com/OliveTin/OliveTin/internal/executor"
 	ws "github.com/gorilla/websocket"
 	log "github.com/sirupsen/logrus"
@@ -21,7 +23,8 @@ var (
 )
 
 type WebsocketClient struct {
-	conn *ws.Conn
+	conn              *ws.Conn
+	authenticatedUser *acl.AuthenticatedUser
 }
 
 var clients map[*WebsocketClient]struct{}
@@ -35,10 +38,16 @@ var ExecutionListener WebsocketExecutionListener
 
 type WebsocketExecutionListener struct{}
 
-func (WebsocketExecutionListener) OnExecutionStarted(ile *executor.InternalLogEntry) {
-	broadcast(&apiv1.EventExecutionStarted{
+func (WebsocketExecutionListener) OnExecutionStarted(ile *executor.InternalLogEntry, action *config.Action) {
+	evt := &apiv1.EventExecutionStarted{
 		LogEntry: internalLogEntryToPb(ile),
-	})
+	}
+
+	for client := range copyOfClients() {
+		if acl.IsAllowedLogs(cfg, client.authenticatedUser, action) {
+			writeMessageToClient(client, prepareMessage(evt))
+		}
+	}
 }
 
 func OnEntityChanged() {
@@ -66,7 +75,7 @@ func checkOriginPermissive(r *http.Request) bool {
 	return true
 }
 
-func (WebsocketExecutionListener) OnOutputChunk(chunk []byte, executionTrackingId string) {
+func (WebsocketExecutionListener) OnOutputChunk(chunk []byte, executionTrackingId string, logEntry *executor.InternalLogEntry, action *config.Action) {
 	log.Tracef("outputchunk: %s", string(chunk))
 
 	oc := &apiv1.EventOutputChunk{
@@ -74,25 +83,49 @@ func (WebsocketExecutionListener) OnOutputChunk(chunk []byte, executionTrackingI
 		ExecutionTrackingId: executionTrackingId,
 	}
 
-	broadcast(oc)
+	for client := range copyOfClients() {
+		if acl.IsAllowedLogs(cfg, client.authenticatedUser, action) {
+			writeMessageToClient(client, prepareMessage(oc))
+		}
+	}
 }
 
-func (WebsocketExecutionListener) OnExecutionFinished(logEntry *executor.InternalLogEntry) {
+func (WebsocketExecutionListener) OnExecutionFinished(logEntry *executor.InternalLogEntry, action *config.Action) {
 	evt := &apiv1.EventExecutionFinished{
 		LogEntry: internalLogEntryToPb(logEntry),
 	}
 
+	for client := range copyOfClients() {
+		if acl.IsAllowedLogs(cfg, client.authenticatedUser, action) {
+			writeMessageToClient(client, prepareMessage(evt))
+		}
+	}
+
 	log.Infof("WS Execution finished: %v", evt.LogEntry)
+}
+
+func copyOfClients() map[*WebsocketClient]struct{} {
+	sendmutex.Lock()
+	defer sendmutex.Unlock()
+
+	if clients == nil {
+		clients = make(map[*WebsocketClient]struct{})
+	}
+
+	copy := make(map[*WebsocketClient]struct{})
+	for client := range clients {
+		copy[client] = struct{}{}
+	}
 
-	broadcast(evt)
+	return copy
 }
 
-func broadcast(pbmsg protoreflect.ProtoMessage) {
+func prepareMessage(pbmsg protoreflect.ProtoMessage) []byte {
 	payload, err := marshalOptions.Marshal(pbmsg)
 
 	if err != nil {
 		log.Errorf("websocket marshal error: %v", err)
-		return
+		return nil
 	}
 
 	messageType := pbmsg.ProtoReflect().Descriptor().FullName()
@@ -116,16 +149,31 @@ func broadcast(pbmsg protoreflect.ProtoMessage) {
 	hackyMessage = append(hackyMessage, []byte("}")...)
 	// </EVIL>
 
-	sendmutex.Lock()
-	if clients == nil {
-		clients = make(map[*WebsocketClient]struct{})
+	return hackyMessage
+}
+
+func broadcast(pbmsg protoreflect.ProtoMessage) {
+	message := prepareMessage(pbmsg)
+
+	for client := range copyOfClients() {
+		writeMessageToClient(client, message)
 	}
-	for client := range clients {
-		if err := client.conn.WriteMessage(ws.TextMessage, hackyMessage); err != nil {
-			log.Debugf("websocket send error: %v - closing connection", err)
-			_ = client.conn.Close()
-			delete(clients, client)
-		}
+}
+
+func writeMessageToClient(client *WebsocketClient, message []byte) {
+	if message == nil {
+		log.Warnf("writeMessageToClient: message is nil")
+		return
+	}
+
+	sendmutex.Lock()
+	if err := client.conn.WriteMessage(ws.TextMessage, message); err != nil {
+		log.WithFields(log.Fields{
+			"error":  err,
+			"client": client,
+		}).Debugf("websocket send error")
+		_ = client.conn.Close()
+		delete(clients, client)
 	}
 	sendmutex.Unlock()
 }
@@ -143,7 +191,7 @@ func (c *WebsocketClient) messageLoop() {
 	}
 }
 
-func HandleWebsocket(w http.ResponseWriter, r *http.Request) bool {
+func handleWebsocket(w http.ResponseWriter, r *http.Request) bool {
 	c, err := upgrader.Upgrade(w, r, nil)
 
 	if err != nil {
@@ -151,10 +199,13 @@ func HandleWebsocket(w http.ResponseWriter, r *http.Request) bool {
 		return false
 	}
 
-	//	defer c.Close()
+	unauthenticatedUser := authHttpRequest(r)
+
+	authenticatedUser := acl.UserFromUnauthenticatedUser(cfg, unauthenticatedUser)
 
 	wsclient := &WebsocketClient{
-		conn: c,
+		conn:              c,
+		authenticatedUser: authenticatedUser,
 	}
 
 	sendmutex.Lock()

service/main.go

@@ -16,13 +16,13 @@ import (
 	"github.com/OliveTin/OliveTin/internal/onstartup"
 	"github.com/OliveTin/OliveTin/internal/servicehost"
 	updatecheck "github.com/OliveTin/OliveTin/internal/updatecheck"
-	"github.com/OliveTin/OliveTin/internal/websocket"
+
+	"os"
+	"strconv"
 
 	config "github.com/OliveTin/OliveTin/internal/config"
 	"github.com/fsnotify/fsnotify"
 	"github.com/spf13/viper"
-	"os"
-	"strconv"
 )
 
 var (
@@ -171,7 +171,7 @@ func main() {
 
 	executor := executor.DefaultExecutor(cfg)
 	executor.RebuildActionMap()
-	executor.AddListener(websocket.ExecutionListener)
+	executor.AddListener(httpservers.ExecutionListener)
 	config.AddListener(executor.RebuildActionMap)
 
 	go onstartup.Execute(cfg, executor)
@@ -179,7 +179,7 @@ func main() {
 	go onfileindir.WatchFilesInDirectory(cfg, executor)
 	go oncalendarfile.Schedule(cfg, executor)
 
-	entityfiles.AddListener(websocket.OnEntityChanged)
+	entityfiles.AddListener(httpservers.OnEntityChanged)
 	entityfiles.AddListener(executor.RebuildActionMap)
 	go entityfiles.SetupEntityFileWatchers(cfg)