<commit_before>package realtime

import (
	"context"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/cozy/cozy-stack/model/instance"
	"github.com/cozy/cozy-stack/model/permission"
	"github.com/cozy/cozy-stack/pkg/consts"
	"github.com/cozy/cozy-stack/pkg/couchdb"
	"github.com/cozy/cozy-stack/pkg/jsonapi"
	"github.com/cozy/cozy-stack/pkg/logger"
	"github.com/cozy/cozy-stack/pkg/prefixer"
	"github.com/cozy/cozy-stack/pkg/realtime"
	"github.com/cozy/cozy-stack/web/middlewares"
	"github.com/gorilla/websocket"
	"github.com/labstack/echo/v4"
)

const (
	// Time allowed to write a message to the peer
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the peer
	pongWait = 60 * time.Second

	// Send pings to peer with this period (must be less than pongWait)
	pingPeriod = (pongWait * 9) / 10

	// Maximum message size allowed from peer
	maxMessageSize = 1024
)

var upgrader = websocket.Upgrader{
	// Don't check the origin of the connexion, we check authorization later
	CheckOrigin:     func(r *http.Request) bool { return true },
	Subprotocols:    []string{"io.cozy.websocket"},
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
}

type command struct {
	Method  string `json:"method"`
	Payload struct {
		Type string `json:"type"`
		ID   string `json:"id"`
	} `json:"payload"`
}

type wsResponsePayload struct {
	Type string      `json:"type"`
	ID   string      `json:"id"`
	Doc  interface{} `json:"doc,omitempty"`
}

type wsResponse struct {
	Event   string            `json:"event"`
	Payload wsResponsePayload `json:"payload"`
}

type wsErrorPayload struct {
	Status string      `json:"status"`
	Code   string      `json:"code"`
	Title  string      `json:"title"`
	Source interface{} `json:"source"`
}

type wsError struct {
	Event   string         `json:"event"`
	Payload wsErrorPayload `json:"payload"`
}

func unauthorized(cmd interface{}) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "401 Unauthorized",
			Code:   "unauthorized",
			Title:  "The authentication has failed",
			Source: cmd,
		},
	}
}

func forbidden(cmd *command) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "403 Forbidden",
			Code:   "forbidden",
			Title:  fmt.Sprintf("The application can't subscribe to %s", cmd.Payload.Type),
			Source: cmd,
		},
	}
}

func unknownMethod(method string, cmd interface{}) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "405 Method Not Allowed",
			Code:   "method not allowed",
			Title:  fmt.Sprintf("The %s method is not supported", method),
			Source: cmd,
		},
	}
}

func missingType(cmd *command) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "404 Page Not Found",
			Code:   "page not found",
			Title:  "The type parameter is mandatory for SUBSCRIBE",
			Source: cmd,
		},
	}
}

func sendErr(ctx context.Context, errc chan *wsError, e *wsError) {
	select {
	case errc <- e:
	case <-ctx.Done():
	}
}

func readPump(ctx context.Context, c echo.Context, i *instance.Instance, ws *websocket.Conn,
	ds *realtime.DynamicSubscriber, errc chan *wsError, withAuthentication bool) {
	defer close(errc)

	var err error
	var pdoc *permission.Permission

	if withAuthentication {
		var auth map[string]string
		if err = ws.ReadJSON(&auth); err != nil {
			sendErr(ctx, errc, unknownMethod(auth["method"], auth))
			return
		}
		if strings.ToUpper(auth["method"]) != "AUTH" {
			sendErr(ctx, errc, unknownMethod(auth["method"], auth))
			return
		}
		if auth["payload"] == "" {
			sendErr(ctx, errc, unauthorized(auth))
			return
		}
		pdoc, err = middlewares.ParseJWT(c, i, auth["payload"])
		if err != nil {
			sendErr(ctx, errc, unauthorized(auth))
			return
		}
	}

	for {
		cmd := &command{}
		if err = ws.ReadJSON(cmd); err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
				logger.
					WithDomain(ds.DomainName()).
					WithField("nspace", "realtime").
					Debugf("Error: %s", err)
			}
			break
		}

		method := strings.ToUpper(cmd.Method)
		if method != "SUBSCRIBE" && method != "UNSUBSCRIBE" {
			sendErr(ctx, errc, unknownMethod(cmd.Method, cmd))
			continue
		}
		if cmd.Payload.Type == "" {
			sendErr(ctx, errc, missingType(cmd))
			continue
		}
		permType := cmd.Payload.Type
		// XXX: thumbnails is a synthetic doctype, listening to its events
		// requires a permissions on io.cozy.files. Same for note events.
		if permType == consts.Thumbnails || permType == consts.NotesEvents {
			permType = consts.Files
		}
		// XXX: no permissions are required for io.cozy.sharings.initial_sync
		if withAuthentication && cmd.Payload.Type != consts.SharingsInitialSync {
			var authorized bool
			if cmd.Payload.ID == "" {
				authorized = pdoc.Permissions.AllowWholeType(permission.GET, permType)
			} else {
				authorized = pdoc.Permissions.AllowID(permission.GET, permType, cmd.Payload.ID)
			}
			if !authorized {
				sendErr(ctx, errc, forbidden(cmd))
				continue
			}
		}

		if method == "SUBSCRIBE" {
			if cmd.Payload.ID == "" {
				err = ds.Subscribe(cmd.Payload.Type)
			} else {
				err = ds.Watch(cmd.Payload.Type, cmd.Payload.ID)<commit_msg>Fix realtime for a note in a shared directory (#2458)<commit_after>package realtime

import (
	"context"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/cozy/cozy-stack/model/instance"
	"github.com/cozy/cozy-stack/model/permission"
	"github.com/cozy/cozy-stack/model/vfs"
	"github.com/cozy/cozy-stack/pkg/consts"
	"github.com/cozy/cozy-stack/pkg/couchdb"
	"github.com/cozy/cozy-stack/pkg/jsonapi"
	"github.com/cozy/cozy-stack/pkg/logger"
	"github.com/cozy/cozy-stack/pkg/prefixer"
	"github.com/cozy/cozy-stack/pkg/realtime"
	"github.com/cozy/cozy-stack/web/middlewares"
	"github.com/gorilla/websocket"
	"github.com/labstack/echo/v4"
)

const (
	// Time allowed to write a message to the peer
	writeWait = 10 * time.Second

	// Time allowed to read the next pong message from the peer
	pongWait = 60 * time.Second

	// Send pings to peer with this period (must be less than pongWait)
	pingPeriod = (pongWait * 9) / 10

	// Maximum message size allowed from peer
	maxMessageSize = 1024
)

var upgrader = websocket.Upgrader{
	// Don't check the origin of the connexion, we check authorization later
	CheckOrigin:     func(r *http.Request) bool { return true },
	Subprotocols:    []string{"io.cozy.websocket"},
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
}

type command struct {
	Method  string `json:"method"`
	Payload struct {
		Type string `json:"type"`
		ID   string `json:"id"`
	} `json:"payload"`
}

type wsResponsePayload struct {
	Type string      `json:"type"`
	ID   string      `json:"id"`
	Doc  interface{} `json:"doc,omitempty"`
}

type wsResponse struct {
	Event   string            `json:"event"`
	Payload wsResponsePayload `json:"payload"`
}

type wsErrorPayload struct {
	Status string      `json:"status"`
	Code   string      `json:"code"`
	Title  string      `json:"title"`
	Source interface{} `json:"source"`
}

type wsError struct {
	Event   string         `json:"event"`
	Payload wsErrorPayload `json:"payload"`
}

func unauthorized(cmd interface{}) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "401 Unauthorized",
			Code:   "unauthorized",
			Title:  "The authentication has failed",
			Source: cmd,
		},
	}
}

func forbidden(cmd *command) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "403 Forbidden",
			Code:   "forbidden",
			Title:  fmt.Sprintf("The application can't subscribe to %s", cmd.Payload.Type),
			Source: cmd,
		},
	}
}

func unknownMethod(method string, cmd interface{}) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "405 Method Not Allowed",
			Code:   "method not allowed",
			Title:  fmt.Sprintf("The %s method is not supported", method),
			Source: cmd,
		},
	}
}

func missingType(cmd *command) *wsError {
	return &wsError{
		Event: "error",
		Payload: wsErrorPayload{
			Status: "404 Page Not Found",
			Code:   "page not found",
			Title:  "The type parameter is mandatory for SUBSCRIBE",
			Source: cmd,
		},
	}
}

func sendErr(ctx context.Context, errc chan *wsError, e *wsError) {
	select {
	case errc <- e:
	case <-ctx.Done():
	}
}

func authorized(i *instance.Instance, perms permission.Set, permType, id string) bool {
	if perms.AllowWholeType(permission.GET, permType) {
		return true
	} else if id == "" {
		return false
	} else if permType == consts.Files {
		fs := i.VFS()
		dir, file, err := fs.DirOrFileByID(id)
		if dir != nil {
			err = vfs.Allows(fs, perms, permission.GET, dir)
		} else if file != nil {
			err = vfs.Allows(fs, perms, permission.GET, file)
		}
		return err == nil
	} else {
		return perms.AllowID(permission.GET, permType, id)
	}
}

func readPump(ctx context.Context, c echo.Context, i *instance.Instance, ws *websocket.Conn,
	ds *realtime.DynamicSubscriber, errc chan *wsError, withAuthentication bool) {
	defer close(errc)

	var err error
	var pdoc *permission.Permission

	if withAuthentication {
		var auth map[string]string
		if err = ws.ReadJSON(&auth); err != nil {
			sendErr(ctx, errc, unknownMethod(auth["method"], auth))
			return
		}
		if strings.ToUpper(auth["method"]) != "AUTH" {
			sendErr(ctx, errc, unknownMethod(auth["method"], auth))
			return
		}
		if auth["payload"] == "" {
			sendErr(ctx, errc, unauthorized(auth))
			return
		}
		pdoc, err = middlewares.ParseJWT(c, i, auth["payload"])
		if err != nil {
			sendErr(ctx, errc, unauthorized(auth))
			return
		}
	}

	for {
		cmd := &command{}
		if err = ws.ReadJSON(cmd); err != nil {
			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
				logger.
					WithDomain(ds.DomainName()).
					WithField("nspace", "realtime").
					Debugf("Error: %s", err)
			}
			break
		}

		method := strings.ToUpper(cmd.Method)
		if method != "SUBSCRIBE" && method != "UNSUBSCRIBE" {
			sendErr(ctx, errc, unknownMethod(cmd.Method, cmd))
			continue
		}
		if cmd.Payload.Type == "" {
			sendErr(ctx, errc, missingType(cmd))
			continue
		}
		permType := cmd.Payload.Type
		// XXX: thumbnails is a synthetic doctype, listening to its events
		// requires a permissions on io.cozy.files. Same for note events.
		if permType == consts.Thumbnails || permType == consts.NotesEvents {
			permType = consts.Files
		}
		// XXX: no permissions are required for io.cozy.sharings.initial_sync
		if withAuthentication && cmd.Payload.Type != consts.SharingsInitialSync {
			if !authorized(i, pdoc.Permissions, permType, cmd.Payload.ID) {
				sendErr(ctx, errc, forbidden(cmd))
				continue
			}
		}

		if method == "SUBSCRIBE" {
			if cmd.Payload.ID == "" {
				err = ds.Subscribe(cmd.Payload.Type)
			} else {
				err = ds.Watch(cmd.Payload.Type, cmd.Payload.ID)
{'fraction_non_alphanumeric': 0.1354874242702405, 'fraction_numerical': 0.006241968055810538, 'mean_word_length': 3.6361702127659576, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7215599', 'n_tokens_mistral': 4203, 'n_tokens_neox': 3755, 'n_words': 951}