mirror of
https://github.com/navidrome/navidrome.git
synced 2026-10-18 07:30:29 +00:00
dafd57ca25
isHostAllowed cut the host at the last colon, which mangles an IPv6 literal: "[::1]:8080" became "[::1]" and "[::1]" became "[:". A plugin manifest could therefore never allow an IPv6 host. It now uses net.SplitHostPort, falling back to unwrapping the brackets when there is no port.
397 lines
11 KiB
Go
397 lines
11 KiB
Go
package plugins
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"maps"
|
|
"net"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/gorilla/websocket"
|
|
"github.com/navidrome/navidrome/log"
|
|
"github.com/navidrome/navidrome/model/id"
|
|
"github.com/navidrome/navidrome/plugins/capabilities"
|
|
"github.com/navidrome/navidrome/plugins/host"
|
|
)
|
|
|
|
// CapabilityWebSocket indicates the plugin can receive WebSocket callbacks.
|
|
// Detected when the plugin exports any of the WebSocket callback functions.
|
|
const CapabilityWebSocket Capability = "WebSocket"
|
|
|
|
// webSocketCallbackTimeout is the maximum duration allowed for a WebSocket callback.
|
|
const webSocketCallbackTimeout = 30 * time.Second
|
|
|
|
// WebSocket callback function names
|
|
const (
|
|
FuncWebSocketOnTextMessage = "nd_websocket_on_text_message"
|
|
FuncWebSocketOnBinaryMessage = "nd_websocket_on_binary_message"
|
|
FuncWebSocketOnError = "nd_websocket_on_error"
|
|
FuncWebSocketOnClose = "nd_websocket_on_close"
|
|
)
|
|
|
|
func init() {
|
|
registerCapability(
|
|
CapabilityWebSocket,
|
|
FuncWebSocketOnTextMessage,
|
|
FuncWebSocketOnBinaryMessage,
|
|
FuncWebSocketOnError,
|
|
FuncWebSocketOnClose,
|
|
)
|
|
}
|
|
|
|
// wsConnection represents an active WebSocket connection.
|
|
type wsConnection struct {
|
|
conn *websocket.Conn
|
|
done chan struct{}
|
|
closeMu sync.Mutex
|
|
isClosed bool
|
|
}
|
|
|
|
// webSocketServiceImpl implements host.WebSocketService.
|
|
// It provides plugins with WebSocket communication capabilities.
|
|
type webSocketServiceImpl struct {
|
|
baseCtx context.Context // bounds the read loops, which outlive the Connect() call
|
|
pluginName string
|
|
manager *Manager
|
|
requiredHosts []string
|
|
|
|
mu sync.RWMutex
|
|
connections map[string]*wsConnection
|
|
}
|
|
|
|
// newWebSocketService creates a new WebSocketService for a plugin.
|
|
func newWebSocketService(ctx context.Context, pluginName string, manager *Manager, permission *WebSocketPermission) *webSocketServiceImpl {
|
|
return &webSocketServiceImpl{
|
|
baseCtx: ctx,
|
|
pluginName: pluginName,
|
|
manager: manager,
|
|
requiredHosts: permission.RequiredHosts,
|
|
connections: make(map[string]*wsConnection),
|
|
}
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) Connect(ctx context.Context, urlStr string, headers map[string]string, connectionID string) (string, error) {
|
|
// Parse and validate URL
|
|
parsedURL, err := url.Parse(urlStr)
|
|
if err != nil {
|
|
return "", fmt.Errorf("invalid URL: %w", err)
|
|
}
|
|
|
|
// Validate scheme
|
|
if parsedURL.Scheme != "ws" && parsedURL.Scheme != "wss" {
|
|
return "", fmt.Errorf("invalid URL scheme: must be ws:// or wss://")
|
|
}
|
|
|
|
// Validate host against allowed hosts
|
|
if !s.isHostAllowed(parsedURL.Host) {
|
|
return "", fmt.Errorf("host %q is not allowed", parsedURL.Host)
|
|
}
|
|
|
|
// Generate connection ID if not provided
|
|
if connectionID == "" {
|
|
connectionID = id.NewRandom()
|
|
}
|
|
|
|
s.mu.Lock()
|
|
if _, exists := s.connections[connectionID]; exists {
|
|
s.mu.Unlock()
|
|
return "", fmt.Errorf("connection ID %q already exists", connectionID)
|
|
}
|
|
s.mu.Unlock()
|
|
|
|
// Create HTTP headers for handshake
|
|
httpHeaders := http.Header{}
|
|
for k, v := range headers {
|
|
httpHeaders.Set(k, v)
|
|
}
|
|
|
|
// Establish WebSocket connection
|
|
dialer := websocket.Dialer{
|
|
HandshakeTimeout: 30 * time.Second,
|
|
}
|
|
|
|
conn, resp, err := dialer.DialContext(ctx, urlStr, httpHeaders)
|
|
if resp != nil && resp.Body != nil {
|
|
_ = resp.Body.Close()
|
|
}
|
|
if err != nil {
|
|
return "", fmt.Errorf("failed to connect: %w", err)
|
|
}
|
|
|
|
wsConn := &wsConnection{
|
|
conn: conn,
|
|
done: make(chan struct{}),
|
|
}
|
|
|
|
s.mu.Lock()
|
|
s.connections[connectionID] = wsConn
|
|
s.mu.Unlock()
|
|
|
|
// Start read goroutine with the service's base context instead of the
|
|
// caller's ctx, because the readLoop must outlive the Connect() call.
|
|
// Connections are closed by Close() when the plugin is unloaded, which ends
|
|
// the readLoop; the base context is a backstop that also ends it on server
|
|
// shutdown (it is never cancelled in one-shot CLI runs).
|
|
go s.readLoop(s.baseCtx, connectionID, wsConn)
|
|
|
|
log.Debug(ctx, "WebSocket connected", "plugin", s.pluginName, "connectionID", connectionID, "url", urlStr)
|
|
return connectionID, nil
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) SendText(ctx context.Context, connectionID, message string) error {
|
|
wsConn, err := s.getConnection(connectionID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := wsConn.conn.WriteMessage(websocket.TextMessage, []byte(message)); err != nil {
|
|
return fmt.Errorf("failed to send text message: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) SendBinary(ctx context.Context, connectionID string, data []byte) error {
|
|
wsConn, err := s.getConnection(connectionID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := wsConn.conn.WriteMessage(websocket.BinaryMessage, data); err != nil {
|
|
return fmt.Errorf("failed to send binary message: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) CloseConnection(ctx context.Context, connectionID string, code int32, reason string) error {
|
|
s.mu.Lock()
|
|
wsConn, exists := s.connections[connectionID]
|
|
if !exists {
|
|
s.mu.Unlock()
|
|
return fmt.Errorf("connection ID %q not found", connectionID)
|
|
}
|
|
delete(s.connections, connectionID)
|
|
s.mu.Unlock()
|
|
|
|
// Mark as closed to prevent callback
|
|
wsConn.closeMu.Lock()
|
|
wsConn.isClosed = true
|
|
wsConn.closeMu.Unlock()
|
|
|
|
// Send close message
|
|
closeMsg := websocket.FormatCloseMessage(int(code), reason)
|
|
_ = wsConn.conn.WriteControl(websocket.CloseMessage, closeMsg, time.Now().Add(5*time.Second))
|
|
_ = wsConn.conn.Close()
|
|
|
|
// Signal read goroutine to stop
|
|
close(wsConn.done)
|
|
|
|
// Invoke close callback
|
|
s.invokeOnClose(ctx, connectionID, code, reason)
|
|
|
|
log.Debug(ctx, "WebSocket connection closed", "plugin", s.pluginName, "connectionID", connectionID, "code", code)
|
|
return nil
|
|
}
|
|
|
|
// Close closes all connections for this plugin.
|
|
// This is called when the plugin is unloaded.
|
|
func (s *webSocketServiceImpl) Close() error {
|
|
s.mu.Lock()
|
|
connections := make(map[string]*wsConnection, len(s.connections))
|
|
maps.Copy(connections, s.connections)
|
|
s.connections = make(map[string]*wsConnection)
|
|
s.mu.Unlock()
|
|
|
|
ctx := context.Background()
|
|
for connID, wsConn := range connections {
|
|
wsConn.closeMu.Lock()
|
|
wsConn.isClosed = true
|
|
wsConn.closeMu.Unlock()
|
|
|
|
closeMsg := websocket.FormatCloseMessage(websocket.CloseGoingAway, "plugin unloaded")
|
|
err := wsConn.conn.WriteControl(websocket.CloseMessage, closeMsg, time.Now().Add(2*time.Second))
|
|
if err != nil {
|
|
log.Warn("Failed to send WebSocket close message on plugin unload", "plugin", s.pluginName, "connectionID", connID, "error", err)
|
|
}
|
|
err = wsConn.conn.Close()
|
|
if err != nil {
|
|
log.Warn("Failed to close WebSocket connection on plugin unload", "plugin", s.pluginName, "connectionID", connID, "error", err)
|
|
}
|
|
close(wsConn.done)
|
|
|
|
s.invokeOnClose(ctx, connID, websocket.CloseGoingAway, "plugin unloaded")
|
|
log.Debug("WebSocket connection closed on plugin unload", "plugin", s.pluginName, "connectionID", connID)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) getConnection(connectionID string) (*wsConnection, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
|
|
wsConn, exists := s.connections[connectionID]
|
|
if !exists {
|
|
return nil, fmt.Errorf("connection ID %q not found", connectionID)
|
|
}
|
|
return wsConn, nil
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) isHostAllowed(host string) bool {
|
|
// Strip port from host if present
|
|
hostWithoutPort := host
|
|
if h, _, err := net.SplitHostPort(host); err == nil {
|
|
hostWithoutPort = h
|
|
} else if h, ok := strings.CutPrefix(host, "["); ok {
|
|
// Bracketed IPv6 literal with no port
|
|
hostWithoutPort = strings.TrimSuffix(h, "]")
|
|
}
|
|
|
|
for _, pattern := range s.requiredHosts {
|
|
if matchHostPattern(pattern, hostWithoutPort) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// matchHostPattern matches a host against a pattern.
|
|
// Supports "*" (allow all) and wildcards like "*.example.com".
|
|
func matchHostPattern(pattern, host string) bool {
|
|
if pattern == "*" {
|
|
return true
|
|
}
|
|
if pattern == host {
|
|
return true
|
|
}
|
|
|
|
// Handle wildcard patterns like *.example.com
|
|
if strings.HasPrefix(pattern, "*.") {
|
|
suffix := pattern[1:] // Get .example.com
|
|
return strings.HasSuffix(host, suffix)
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) readLoop(ctx context.Context, connectionID string, wsConn *wsConnection) {
|
|
defer func() {
|
|
// Remove connection if still present
|
|
s.mu.Lock()
|
|
delete(s.connections, connectionID)
|
|
s.mu.Unlock()
|
|
}()
|
|
|
|
for {
|
|
select {
|
|
case <-wsConn.done:
|
|
return
|
|
default:
|
|
}
|
|
|
|
messageType, data, err := wsConn.conn.ReadMessage()
|
|
if err != nil {
|
|
wsConn.closeMu.Lock()
|
|
isClosed := wsConn.isClosed
|
|
wsConn.closeMu.Unlock()
|
|
|
|
if isClosed {
|
|
return
|
|
}
|
|
|
|
// Check if it's a close error
|
|
if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseNoStatusReceived) {
|
|
closeCode := websocket.CloseNoStatusReceived
|
|
closeReason := ""
|
|
if ce, ok := errors.AsType[*websocket.CloseError](err); ok {
|
|
closeCode = ce.Code
|
|
closeReason = ce.Text
|
|
}
|
|
s.invokeOnClose(ctx, connectionID, int32(closeCode), closeReason)
|
|
return
|
|
}
|
|
|
|
// Other read error
|
|
s.invokeOnError(ctx, connectionID, err.Error())
|
|
return
|
|
}
|
|
|
|
switch messageType {
|
|
case websocket.TextMessage:
|
|
s.invokeOnTextMessage(ctx, connectionID, string(data))
|
|
case websocket.BinaryMessage:
|
|
s.invokeOnBinaryMessage(ctx, connectionID, data)
|
|
}
|
|
}
|
|
}
|
|
|
|
// invokeWebSocketCallback is a generic helper that handles the common callback invocation pattern.
|
|
func invokeWebSocketCallback[I any](ctx context.Context, s *webSocketServiceImpl, funcName string, input I, callbackName string, connectionID string) {
|
|
instance := s.getPluginInstance()
|
|
if instance == nil {
|
|
return
|
|
}
|
|
|
|
callbackCtx, cancel := context.WithTimeout(ctx, webSocketCallbackTimeout)
|
|
defer cancel()
|
|
|
|
start := time.Now()
|
|
err := callPluginFunctionNoOutput(callbackCtx, instance, funcName, input)
|
|
if err != nil {
|
|
if !errors.Is(errFunctionNotFound, err) {
|
|
log.Error(ctx, "WebSocket "+callbackName+" callback failed", "plugin", s.pluginName, "connectionID", connectionID, "duration", time.Since(start), err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) invokeOnTextMessage(ctx context.Context, connectionID, message string) {
|
|
invokeWebSocketCallback(ctx, s, FuncWebSocketOnTextMessage, capabilities.OnTextMessageRequest{
|
|
ConnectionID: connectionID,
|
|
Message: message,
|
|
}, "text message", connectionID)
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) invokeOnBinaryMessage(ctx context.Context, connectionID string, data []byte) {
|
|
invokeWebSocketCallback(ctx, s, FuncWebSocketOnBinaryMessage, capabilities.OnBinaryMessageRequest{
|
|
ConnectionID: connectionID,
|
|
Data: data,
|
|
}, "binary message", connectionID)
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) invokeOnError(ctx context.Context, connectionID, errorMsg string) {
|
|
invokeWebSocketCallback(ctx, s, FuncWebSocketOnError, capabilities.OnErrorRequest{
|
|
ConnectionID: connectionID,
|
|
Error: errorMsg,
|
|
}, "error", connectionID)
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) invokeOnClose(ctx context.Context, connectionID string, code int32, reason string) {
|
|
invokeWebSocketCallback(ctx, s, FuncWebSocketOnClose, capabilities.OnCloseRequest{
|
|
ConnectionID: connectionID,
|
|
Code: code,
|
|
Reason: reason,
|
|
}, "close", connectionID)
|
|
}
|
|
|
|
func (s *webSocketServiceImpl) getPluginInstance() *plugin {
|
|
s.manager.mu.RLock()
|
|
instance, ok := s.manager.plugins[s.pluginName]
|
|
s.manager.mu.RUnlock()
|
|
|
|
if !ok {
|
|
log.Warn("Plugin not loaded for WebSocket callback", "plugin", s.pluginName)
|
|
return nil
|
|
}
|
|
|
|
return instance
|
|
}
|
|
|
|
// Verify interface implementation
|
|
var _ host.WebSocketService = (*webSocketServiceImpl)(nil)
|