diff --git a/go.mod b/go.mod index 2f81efb8..ab54cbaf 100644 --- a/go.mod +++ b/go.mod @@ -9,6 +9,7 @@ require ( github.com/go-ldap/ldap/v3 v3.4.12 github.com/goccy/go-json v0.10.5 github.com/google/uuid v1.6.0 + github.com/gorilla/websocket v1.5.3 github.com/joho/godotenv v1.5.1 github.com/mymmrac/telego v1.3.1 github.com/nicksnyder/go-i18n/v2 v2.6.0 @@ -51,7 +52,6 @@ require ( github.com/gorilla/context v1.1.2 // indirect github.com/gorilla/securecookie v1.1.2 // indirect github.com/gorilla/sessions v1.4.0 // indirect - github.com/gorilla/websocket v1.5.3 // indirect github.com/grbit/go-json v0.11.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect diff --git a/web.rar b/web.rar new file mode 100644 index 00000000..8e3093b7 Binary files /dev/null and b/web.rar differ diff --git a/web/assets/js/websocket.js b/web/assets/js/websocket.js new file mode 100644 index 00000000..2a551a8a --- /dev/null +++ b/web/assets/js/websocket.js @@ -0,0 +1,138 @@ +/** + * WebSocket client for real-time updates + */ +class WebSocketClient { + constructor(basePath = '') { + this.basePath = basePath; + this.ws = null; + this.reconnectAttempts = 0; + this.maxReconnectAttempts = 10; + this.reconnectDelay = 1000; + this.listeners = new Map(); + this.isConnected = false; + this.shouldReconnect = true; + } + + connect() { + if (this.ws && this.ws.readyState === WebSocket.OPEN) { + return; + } + + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const wsUrl = `${protocol}//${window.location.host}${this.basePath}ws`; + + try { + this.ws = new WebSocket(wsUrl); + + this.ws.onopen = () => { + console.log('WebSocket connected'); + this.isConnected = true; + this.reconnectAttempts = 0; + this.emit('connected'); + }; + + this.ws.onmessage = (event) => { + try { + // Validate message size (prevent memory issues) + const maxMessageSize = 10 * 1024 * 1024; // 10MB + if (event.data && event.data.length > maxMessageSize) { + console.error('WebSocket message too large:', event.data.length, 'bytes'); + this.ws.close(); + return; + } + + const message = JSON.parse(event.data); + if (!message || typeof message !== 'object') { + console.error('Invalid WebSocket message format'); + return; + } + + this.handleMessage(message); + } catch (e) { + console.error('Failed to parse WebSocket message:', e); + } + }; + + this.ws.onerror = (error) => { + console.error('WebSocket error:', error); + this.emit('error', error); + }; + + this.ws.onclose = () => { + console.log('WebSocket disconnected'); + this.isConnected = false; + this.emit('disconnected'); + + if (this.shouldReconnect && this.reconnectAttempts < this.maxReconnectAttempts) { + this.reconnectAttempts++; + const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1); + console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})`); + setTimeout(() => this.connect(), delay); + } + }; + } catch (e) { + console.error('Failed to create WebSocket connection:', e); + this.emit('error', e); + } + } + + handleMessage(message) { + const { type, payload, time } = message; + + // Emit to specific type listeners + this.emit(type, payload, time); + + // Emit to all listeners + this.emit('message', { type, payload, time }); + } + + on(event, callback) { + if (!this.listeners.has(event)) { + this.listeners.set(event, []); + } + this.listeners.get(event).push(callback); + } + + off(event, callback) { + if (!this.listeners.has(event)) { + return; + } + const callbacks = this.listeners.get(event); + const index = callbacks.indexOf(callback); + if (index > -1) { + callbacks.splice(index, 1); + } + } + + emit(event, ...args) { + if (this.listeners.has(event)) { + this.listeners.get(event).forEach(callback => { + try { + callback(...args); + } catch (e) { + console.error('Error in WebSocket event handler:', e); + } + }); + } + } + + disconnect() { + this.shouldReconnect = false; + if (this.ws) { + this.ws.close(); + this.ws = null; + } + } + + send(data) { + if (this.ws && this.ws.readyState === WebSocket.OPEN) { + this.ws.send(JSON.stringify(data)); + } else { + console.warn('WebSocket is not connected'); + } + } +} + +// Create global WebSocket client instance +// Safely get basePath from global scope (defined in page.html) +window.wsClient = new WebSocketClient(typeof basePath !== 'undefined' ? basePath : ''); diff --git a/web/controller/inbound.go b/web/controller/inbound.go index eeb160d6..8317de31 100644 --- a/web/controller/inbound.go +++ b/web/controller/inbound.go @@ -8,6 +8,7 @@ import ( "github.com/mhsanaei/3x-ui/v2/database/model" "github.com/mhsanaei/3x-ui/v2/web/service" "github.com/mhsanaei/3x-ui/v2/web/session" + "github.com/mhsanaei/3x-ui/v2/web/websocket" "github.com/gin-gonic/gin" ) @@ -125,6 +126,9 @@ func (a *InboundController) addInbound(c *gin.Context) { if needRestart { a.xrayService.SetToNeedRestart() } + // Broadcast inbounds update via WebSocket + inbounds, _ := a.inboundService.GetInbounds(user.Id) + websocket.BroadcastInbounds(inbounds) } // delInbound deletes an inbound configuration by its ID. @@ -143,6 +147,10 @@ func (a *InboundController) delInbound(c *gin.Context) { if needRestart { a.xrayService.SetToNeedRestart() } + // Broadcast inbounds update via WebSocket + user := session.GetLoginUser(c) + inbounds, _ := a.inboundService.GetInbounds(user.Id) + websocket.BroadcastInbounds(inbounds) } // updateInbound updates an existing inbound configuration. @@ -169,6 +177,10 @@ func (a *InboundController) updateInbound(c *gin.Context) { if needRestart { a.xrayService.SetToNeedRestart() } + // Broadcast inbounds update via WebSocket + user := session.GetLoginUser(c) + inbounds, _ := a.inboundService.GetInbounds(user.Id) + websocket.BroadcastInbounds(inbounds) } // getClientIps retrieves the IP addresses associated with a client by email. diff --git a/web/controller/server.go b/web/controller/server.go index 292ef338..5b39700e 100644 --- a/web/controller/server.go +++ b/web/controller/server.go @@ -9,6 +9,7 @@ import ( "github.com/mhsanaei/3x-ui/v2/web/global" "github.com/mhsanaei/3x-ui/v2/web/service" + "github.com/mhsanaei/3x-ui/v2/web/websocket" "github.com/gin-gonic/gin" ) @@ -67,6 +68,8 @@ func (a *ServerController) refreshStatus() { // collect cpu history when status is fresh if a.lastStatus != nil { a.serverService.AppendCpuSample(time.Now(), a.lastStatus.Cpu) + // Broadcast status update via WebSocket + websocket.BroadcastStatus(a.lastStatus) } } @@ -155,9 +158,16 @@ func (a *ServerController) stopXrayService(c *gin.Context) { err := a.serverService.StopXrayService() if err != nil { jsonMsg(c, I18nWeb(c, "pages.xray.stopError"), err) + websocket.BroadcastXrayState("error", err.Error()) return } jsonMsg(c, I18nWeb(c, "pages.xray.stopSuccess"), err) + websocket.BroadcastXrayState("stop", "") + websocket.BroadcastNotification( + I18nWeb(c, "pages.xray.stopSuccess"), + "Xray service has been stopped", + "warning", + ) } // restartXrayService restarts the Xray service. @@ -165,9 +175,16 @@ func (a *ServerController) restartXrayService(c *gin.Context) { err := a.serverService.RestartXrayService() if err != nil { jsonMsg(c, I18nWeb(c, "pages.xray.restartError"), err) + websocket.BroadcastXrayState("error", err.Error()) return } jsonMsg(c, I18nWeb(c, "pages.xray.restartSuccess"), err) + websocket.BroadcastXrayState("running", "") + websocket.BroadcastNotification( + I18nWeb(c, "pages.xray.restartSuccess"), + "Xray service has been restarted successfully", + "success", + ) } // getLogs retrieves the application logs based on count, level, and syslog filters. diff --git a/web/controller/websocket.go b/web/controller/websocket.go new file mode 100644 index 00000000..b3f9622c --- /dev/null +++ b/web/controller/websocket.go @@ -0,0 +1,232 @@ +package controller + +import ( + "net/http" + "strings" + "time" + + "github.com/google/uuid" + "github.com/mhsanaei/3x-ui/v2/logger" + "github.com/mhsanaei/3x-ui/v2/util/common" + "github.com/mhsanaei/3x-ui/v2/web/session" + "github.com/mhsanaei/3x-ui/v2/web/websocket" + + "github.com/gin-gonic/gin" + ws "github.com/gorilla/websocket" +) + +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 = 512 +) + +var upgrader = ws.Upgrader{ + ReadBufferSize: 4096, // Увеличено с 1024 для лучшей производительности + WriteBufferSize: 4096, // Увеличено с 1024 для лучшей производительности + CheckOrigin: func(r *http.Request) bool { + // Check origin for security + origin := r.Header.Get("Origin") + if origin == "" { + // Allow connections without Origin header (same-origin requests) + return true + } + // Get the host from the request + host := r.Host + // Extract scheme and host from origin + originURL := origin + // Simple check: origin should match the request host + // This prevents cross-origin WebSocket hijacking + if strings.HasPrefix(originURL, "http://") || strings.HasPrefix(originURL, "https://") { + // Extract host from origin + originHost := strings.TrimPrefix(strings.TrimPrefix(originURL, "http://"), "https://") + if idx := strings.Index(originHost, "/"); idx != -1 { + originHost = originHost[:idx] + } + if idx := strings.Index(originHost, ":"); idx != -1 { + originHost = originHost[:idx] + } + // Compare hosts (without port) + requestHost := host + if idx := strings.Index(requestHost, ":"); idx != -1 { + requestHost = requestHost[:idx] + } + return originHost == requestHost || originHost == "" || requestHost == "" + } + return false + }, +} + +// WebSocketController handles WebSocket connections for real-time updates +type WebSocketController struct { + BaseController + hub *websocket.Hub +} + +// NewWebSocketController creates a new WebSocket controller +func NewWebSocketController(hub *websocket.Hub) *WebSocketController { + return &WebSocketController{ + hub: hub, + } +} + +// HandleWebSocket handles WebSocket connections +func (w *WebSocketController) HandleWebSocket(c *gin.Context) { + // Check authentication + if !session.IsLogin(c) { + logger.Warningf("Unauthorized WebSocket connection attempt from %s", getRemoteIp(c)) + c.AbortWithStatus(http.StatusUnauthorized) + return + } + + // Upgrade connection to WebSocket + conn, err := upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + logger.Error("Failed to upgrade WebSocket connection:", err) + return + } + + // Create client + clientID := uuid.New().String() + client := &websocket.Client{ + ID: clientID, + Hub: w.hub, + Send: make(chan []byte, 512), // Увеличено с 256 до 512 для предотвращения переполнения + Topics: make(map[websocket.MessageType]bool), + } + + // Register client + w.hub.Register(client) + logger.Infof("WebSocket client %s registered from %s", clientID, getRemoteIp(c)) + + // Start goroutines for reading and writing + go w.writePump(client, conn) + go w.readPump(client, conn) +} + +// readPump pumps messages from the WebSocket connection to the hub +func (w *WebSocketController) readPump(client *websocket.Client, conn *ws.Conn) { + defer func() { + if r := common.Recover("WebSocket readPump panic"); r != nil { + logger.Error("WebSocket readPump panic recovered:", r) + } + w.hub.Unregister(client) + conn.Close() + }() + + conn.SetReadDeadline(time.Now().Add(pongWait)) + conn.SetPongHandler(func(string) error { + conn.SetReadDeadline(time.Now().Add(pongWait)) + return nil + }) + conn.SetReadLimit(maxMessageSize) + + for { + _, message, err := conn.ReadMessage() + if err != nil { + if ws.IsUnexpectedCloseError(err, ws.CloseGoingAway, ws.CloseAbnormalClosure) { + logger.Debugf("WebSocket read error for client %s: %v", client.ID, err) + } + break + } + + // Validate message size + if len(message) > maxMessageSize { + logger.Warningf("WebSocket message from client %s exceeds max size: %d bytes", client.ID, len(message)) + continue + } + + // Handle incoming messages (e.g., subscription requests) + // For now, we'll just log them + logger.Debugf("Received WebSocket message from client %s: %s", client.ID, string(message)) + } +} + +// writePump pumps messages from the hub to the WebSocket connection +func (w *WebSocketController) writePump(client *websocket.Client, conn *ws.Conn) { + ticker := time.NewTicker(pingPeriod) + defer func() { + if r := common.Recover("WebSocket writePump panic"); r != nil { + logger.Error("WebSocket writePump panic recovered:", r) + } + ticker.Stop() + conn.Close() + }() + + for { + select { + case message, ok := <-client.Send: + conn.SetWriteDeadline(time.Now().Add(writeWait)) + if !ok { + // Hub closed the channel + conn.WriteMessage(ws.CloseMessage, []byte{}) + return + } + + writer, err := conn.NextWriter(ws.TextMessage) + if err != nil { + logger.Debugf("WebSocket write error for client %s: %v", client.ID, err) + return + } + writer.Write(message) + + // Оптимизация: батчинг сообщений с умным ограничением + // Обрабатываем накопленные сообщения, но ограничиваем для предотвращения задержек + n := len(client.Send) + maxQueued := 20 // Увеличено с 10 до 20 для лучшей пропускной способности + if n > maxQueued { + // Пропускаем старые сообщения, оставляем только последние для актуальности + skipped := n - maxQueued + for i := 0; i < skipped; i++ { + select { + case <-client.Send: + // Пропускаем старое сообщение + default: + // Канал закрыт или пуст, прекращаем пропуск + goto skipDone + } + } + skipDone: + n = len(client.Send) // Обновляем количество после пропуска + } + + // Батчинг: отправляем несколько сообщений в одном фрейме + // Безопасное чтение с проверкой закрытия канала + for i := 0; i < n; i++ { + select { + case msg, ok := <-client.Send: + if !ok { + // Канал закрыт, выходим + return + } + writer.Write([]byte{'\n'}) + writer.Write(msg) + default: + // Больше нет сообщений в очереди, прекращаем батчинг + goto batchDone + } + } + batchDone: + + if err := writer.Close(); err != nil { + logger.Debugf("WebSocket writer close error for client %s: %v", client.ID, err) + return + } + + case <-ticker.C: + conn.SetWriteDeadline(time.Now().Add(writeWait)) + if err := conn.WriteMessage(ws.PingMessage, nil); err != nil { + logger.Debugf("WebSocket ping error for client %s: %v", client.ID, err) + return + } + } + } +} diff --git a/web/global/global.go b/web/global/global.go index 025fa081..f72c7bfe 100644 --- a/web/global/global.go +++ b/web/global/global.go @@ -17,6 +17,7 @@ var ( type WebServer interface { GetCron() *cron.Cron // Get the cron scheduler GetCtx() context.Context // Get the server context + GetWSHub() interface{} // Get the WebSocket hub (using interface{} to avoid circular dependency) } // SubServer interface defines methods for accessing the subscription server instance. diff --git a/web/html/common/page.html b/web/html/common/page.html index c0a7ca63..0af63afb 100644 --- a/web/html/common/page.html +++ b/web/html/common/page.html @@ -49,6 +49,7 @@ const basePath = '{{ .base_path }}'; axios.defaults.baseURL = basePath; + {{ end }} {{ define "page/body_end" }} diff --git a/web/html/form/protocol/vless.html b/web/html/form/protocol/vless.html index 83950063..65e12a68 100644 --- a/web/html/form/protocol/vless.html +++ b/web/html/form/protocol/vless.html @@ -39,6 +39,7 @@ +