修复添加用户后引起的编辑错误

This commit is contained in:
qaq
2025-06-28 22:42:20 +08:00
parent f044152c0c
commit 34f5015369
7 changed files with 368 additions and 54 deletions
+27 -2
View File
@@ -2,7 +2,9 @@ package api
import (
"fmt"
"net/http"
"strings"
"time"
@@ -10,6 +12,7 @@ import (
"github.com/go-gost/core/service"
"github.com/go-gost/x/config"
parser "github.com/go-gost/x/config/parsing/service"
kill "github.com/go-gost/x/internal/util/port"
"github.com/go-gost/x/registry"
)
@@ -534,8 +537,24 @@ func pauseService(ctx *gin.Context) {
return
}
// 关闭服务但保持注册状态
// 获取服务地址用于强制断开连接
var serviceAddr string
cfg := config.Global()
for _, s := range cfg.Services {
if s.Name == name {
serviceAddr = s.Addr
break
}
}
// 使用和 updateService 相同的方法彻底断开所有连接
svc.Close()
registry.ServiceRegistry().Unregister(name)
// 强制断开端口的所有连接
if serviceAddr != "" {
_ = kill.ForceClosePortConnections(serviceAddr)
}
// 更新配置中的暂停状态
config.OnUpdate(func(c *config.Config) error {
@@ -893,8 +912,14 @@ func pauseServices(ctx *gin.Context) {
return
}
// 暂停服务
// 使用和 updateService 相同的方法彻底断开所有连接
stp.service.Close()
registry.ServiceRegistry().Unregister(stp.name)
// 强制断开端口的所有连接
if serviceConfig.Addr != "" {
_ = kill.ForceClosePortConnections(serviceConfig.Addr)
}
// 记录已暂停的服务
pausedServices = append(pausedServices, struct {
+193
View File
@@ -0,0 +1,193 @@
package port
import (
"bufio"
"fmt"
"net"
"os"
"runtime"
"strconv"
"strings"
)
// ForceClosePortConnections 强制断开指定端口的所有连接
func ForceClosePortConnections(addr string) error {
if addr == "" {
return nil
}
// 解析地址和端口
host, portStr, err := net.SplitHostPort(addr)
if err != nil {
// 如果解析失败,可能是因为没有端口,或者格式不对
return fmt.Errorf("failed to parse address %s: %v", addr, err)
}
port, err := strconv.Atoi(portStr)
if err != nil {
return fmt.Errorf("invalid port %s: %v", portStr, err)
}
switch runtime.GOOS {
case "linux":
return forceClosePortConnectionsLinux(host, port)
default:
// 对于非Linux系统,静默忽略
return nil
}
}
// forceClosePortConnectionsLinux 在Linux上通过/proc/net/tcp*读取连接并关闭
func forceClosePortConnectionsLinux(host string, port int) error {
// 关闭IPv4连接
if err := closeConnectionsFromProcNet("/proc/net/tcp", port); err != nil {
fmt.Printf("Warning: failed to close IPv4 connections on port %d: %v\n", port, err)
}
// 关闭IPv6连接
if err := closeConnectionsFromProcNet("/proc/net/tcp6", port); err != nil {
fmt.Printf("Warning: failed to close IPv6 connections on port %d: %v\n", port, err)
}
return nil
}
// closeConnectionsFromProcNet 从/proc/net/tcp*文件读取连接信息并关闭连接
func closeConnectionsFromProcNet(procFile string, targetPort int) error {
file, err := os.Open(procFile)
if err != nil {
return err
}
defer file.Close()
scanner := bufio.NewScanner(file)
// 跳过第一行(header)
scanner.Scan()
for scanner.Scan() {
line := scanner.Text()
fields := strings.Fields(line)
if len(fields) < 10 {
continue
}
// local_address字段格式: IP:PORT (hex)
localAddr := fields[1]
parts := strings.Split(localAddr, ":")
if len(parts) != 2 {
continue
}
// 解析端口(十六进制)
portHex := parts[1]
port, err := strconv.ParseInt(portHex, 16, 32)
if err != nil {
continue
}
// 检查是否是目标端口
if int(port) == targetPort {
// 获取远程地址信息
remoteAddr := fields[2]
state := fields[3]
// 只关闭已建立的连接,跳过监听状态
if state == "0A" { // LISTEN状态
continue
}
// 尝试通过socket操作关闭连接
if err := closeConnectionByAddresses(localAddr, remoteAddr); err != nil {
fmt.Printf("Warning: failed to close connection %s->%s: %v\n", localAddr, remoteAddr, err)
}
}
}
return scanner.Err()
}
// closeConnectionByAddresses 通过地址信息关闭连接
func closeConnectionByAddresses(localAddr, remoteAddr string) error {
// 解析本地地址
localParts := strings.Split(localAddr, ":")
if len(localParts) != 2 {
return fmt.Errorf("invalid local address format")
}
// 解析远程地址
remoteParts := strings.Split(remoteAddr, ":")
if len(remoteParts) != 2 {
return fmt.Errorf("invalid remote address format")
}
// 转换十六进制IP和端口
localIP, err := hexToIP(localParts[0])
if err != nil {
return err
}
localPort, err := strconv.ParseInt(localParts[1], 16, 32)
if err != nil {
return err
}
remoteIP, err := hexToIP(remoteParts[0])
if err != nil {
return err
}
remotePort, err := strconv.ParseInt(remoteParts[1], 16, 32)
if err != nil {
return err
}
// 创建连接并立即关闭来触发RST
localAddr4 := fmt.Sprintf("%s:%d", localIP, localPort)
remoteAddr4 := fmt.Sprintf("%s:%d", remoteIP, remotePort)
return sendRSTToConnection(localAddr4, remoteAddr4)
}
// hexToIP 将十六进制IP地址转换为可读格式
func hexToIP(hexIP string) (string, error) {
if len(hexIP) == 8 { // IPv4
ip := make([]byte, 4)
for i := 0; i < 4; i++ {
val, err := strconv.ParseUint(hexIP[i*2:(i+1)*2], 16, 8)
if err != nil {
return "", err
}
ip[3-i] = byte(val) // 小端序
}
return fmt.Sprintf("%d.%d.%d.%d", ip[0], ip[1], ip[2], ip[3]), nil
}
// 对于IPv6,处理更复杂,这里简化处理
return "", fmt.Errorf("IPv6 not implemented yet")
}
// sendRSTToConnection 向指定连接发送RST包
func sendRSTToConnection(localAddr, remoteAddr string) error {
// 尝试创建一个到远程地址的连接,然后立即使用SO_LINGER=0关闭
// 这会发送RST包而不是正常的FIN包
conn, err := net.Dial("tcp", remoteAddr)
if err != nil {
// 如果无法连接,说明连接可能已经不存在了
return nil
}
// 获取底层的TCP连接
tcpConn, ok := conn.(*net.TCPConn)
if !ok {
conn.Close()
return fmt.Errorf("not a TCP connection")
}
// 设置SO_LINGER为0,这样关闭时会发送RST而不是FIN
if err := tcpConn.SetLinger(0); err != nil {
tcpConn.Close()
return err
}
// 关闭连接,这会发送RST包
return tcpConn.Close()
}
+8 -1
View File
@@ -9,6 +9,7 @@ import (
"github.com/go-gost/core/service"
"github.com/go-gost/x/config"
parser "github.com/go-gost/x/config/parsing/service"
kill "github.com/go-gost/x/internal/util/port"
"github.com/go-gost/x/registry"
)
@@ -273,8 +274,14 @@ func pauseServices(req pauseServicesRequest) error {
return errors.New(fmt.Sprintf("service %s configuration not found", stp.name))
}
// 暂停服务
// 使用和 updateService 相同的方法彻底断开所有连接
stp.service.Close()
registry.ServiceRegistry().Unregister(stp.name)
// 强制断开端口的所有连接
if serviceConfig.Addr != "" {
_ = kill.ForceClosePortConnections(serviceConfig.Addr)
}
// 记录已暂停的服务
pausedServices = append(pausedServices, struct {
+7 -8
View File
@@ -341,14 +341,13 @@ func (w *WebSocketReporter) handleReceivedMessage(messageType int, message []byt
// routeCommand 路由命令到对应的处理函数
func (w *WebSocketReporter) routeCommand(cmd CommandMessage) {
//jsonBytes, errs := json.MarshalIndent(cmd, "", " ")
//if errs != nil {
// fmt.Println("Error marshaling JSON:", errs)
// return
//}
//
//// 打印 JSON 字符串
//fmt.Println("🔔 收到命令: ", string(jsonBytes))
jsonBytes, errs := json.Marshal(cmd)
if errs != nil {
fmt.Println("Error marshaling JSON:", errs)
return
}
fmt.Println("🔔 收到命令: ", string(jsonBytes))
var err error
var response CommandResponse
@@ -83,7 +83,7 @@ public class GostUtil {
JSONArray services = new JSONArray();
services.add(name + "_tls");
data.put("services", services);
return WebSocketServer.send_msg(node_id, data, "PauseRemoteService");
return WebSocketServer.send_msg(node_id, data, "PauseService");
}
public static GostDto ResumeRemoteService(Long node_id, String name) {
@@ -91,7 +91,7 @@ public class GostUtil {
JSONArray services = new JSONArray();
services.add(name + "_tls");
data.put("services", services);
return WebSocketServer.send_msg(node_id, data, "ResumeRemoteService");
return WebSocketServer.send_msg(node_id, data, "ResumeService");
}
public static GostDto AddChains(Long node_id, String name, String remoteAddr, String protocol) {
@@ -300,6 +300,12 @@ public class FlowController extends BaseController {
*/
private void pauseServiceDueToTunnelLimit(Integer tunnelId, String forwardId,
String userId, String userTunnelId) {
// 先检查转发状态,如果已经是暂停状态就不需要调用暂停服务
Forward currentForward = forwardService.getById(forwardId);
if (currentForward == null || currentForward.getStatus() == 0) {
return; // 转发不存在或已经暂停,无需处理
}
Tunnel tunnel = tunnelService.getById(tunnelId);
if (tunnel != null) {
Node node = nodeService.getNodeById(tunnel.getInNodeId());
@@ -317,12 +323,11 @@ public class FlowController extends BaseController {
}
}
// 更新转发状态为暂停
Forward forward = forwardService.getById(forwardId);
if (forward != null) {
forward.setStatus(0);
forwardService.updateById(forward);
}
// 更新转发状态为暂停(只更新状态字段,避免覆盖流量数据)
UpdateWrapper<Forward> updateWrapper = new UpdateWrapper<>();
updateWrapper.eq("id", forwardId);
updateWrapper.set("status", 0);
forwardService.update(null, updateWrapper);
}
/**
@@ -330,6 +335,12 @@ public class FlowController extends BaseController {
*/
private void pauseServiceDueToTunnelExpiration(Integer tunnelId, String forwardId,
String userId, String userTunnelId) {
// 先检查转发状态,如果已经是暂停状态就不需要调用暂停服务
Forward currentForward = forwardService.getById(forwardId);
if (currentForward == null || currentForward.getStatus() == 0) {
return; // 转发不存在或已经暂停,无需处理
}
Tunnel tunnel = tunnelService.getById(tunnelId);
if (tunnel != null) {
Node node = nodeService.getNodeById(tunnel.getInNodeId());
@@ -347,12 +358,11 @@ public class FlowController extends BaseController {
}
}
// 更新转发状态为暂停
Forward forward = forwardService.getById(forwardId);
if (forward != null) {
forward.setStatus(0);
forwardService.updateById(forward);
}
// 更新转发状态为暂停(只更新状态字段,避免覆盖流量数据)
UpdateWrapper<Forward> updateWrapper = new UpdateWrapper<>();
updateWrapper.eq("id", forwardId);
updateWrapper.set("status", 0);
forwardService.update(null, updateWrapper);
}
/**
@@ -416,8 +426,8 @@ public class FlowController extends BaseController {
* 检查转发状态 - 优化版本,使用传入的Forward实体
*/
private void checkForwardStatus(Forward forward, String userId, String userTunnelId) {
// 检查转发状态是否为正常(1)
if (forward.getStatus() != 1) {
// 检查转发状态是否为正常(1),如果不正常且不是暂停状态,才需要暂停
if (forward.getStatus() != 1 && forward.getStatus() != 0) {
pauseSpecificForward(forward, userId, userTunnelId);
}
}
@@ -426,6 +436,11 @@ public class FlowController extends BaseController {
* 暂停指定的转发服务
*/
private void pauseSpecificForward(Forward forward, String userId, String userTunnelId) {
// 先检查转发状态,如果已经是暂停状态就不需要调用暂停服务
if (forward.getStatus() == 0) {
return; // 已经暂停,无需处理
}
Tunnel tunnel = tunnelService.getById(forward.getTunnelId());
if (tunnel != null) {
Node node = nodeService.getNodeById(tunnel.getInNodeId());
@@ -443,9 +458,11 @@ public class FlowController extends BaseController {
}
}
// 更新转发状态为暂停
forward.setStatus(0);
forwardService.updateById(forward);
// 更新转发状态为暂停(只更新状态字段,避免覆盖流量数据)
UpdateWrapper<Forward> updateWrapper = new UpdateWrapper<>();
updateWrapper.eq("id", forward.getId());
updateWrapper.set("status", 0);
forwardService.update(null, updateWrapper);
}
/**
@@ -455,6 +472,11 @@ public class FlowController extends BaseController {
List<Forward> userForwards = forwardService.list(new QueryWrapper<Forward>().eq("user_id", userId));
for (Forward forward : userForwards) {
// 先检查转发状态,如果已经是暂停状态就跳过
if (forward.getStatus() == 0) {
continue; // 已经暂停,跳过此转发
}
Tunnel tunnel = tunnelService.getById(forward.getTunnelId());
if (tunnel != null) {
Node node = nodeService.getNodeById(tunnel.getInNodeId());
@@ -474,8 +496,11 @@ public class FlowController extends BaseController {
}
}
forward.setStatus(0);
forwardService.updateById(forward);
// 更新转发状态为暂停(只更新状态字段,避免覆盖流量数据)
UpdateWrapper<Forward> updateWrapper = new UpdateWrapper<>();
updateWrapper.eq("id", forward.getId());
updateWrapper.set("status", 0);
forwardService.update(null, updateWrapper);
}
}
@@ -148,7 +148,7 @@ public class ForwardServiceImpl extends ServiceImpl<ForwardMapper, Forward> impl
return R.err("隧道已禁用,无法更新转发");
}
// 4. 检查权限和限制(仅当隧道发生变化时)
// 4. 检查权限和限制
UserPermissionResult permissionResult = null;
if (isTunnelChanged(existForward, forwardUpdateDto)) {
permissionResult = checkUserPermissions(currentUser, tunnel, forwardUpdateDto.getId());
@@ -156,35 +156,48 @@ public class ForwardServiceImpl extends ServiceImpl<ForwardMapper, Forward> impl
return R.err(permissionResult.getErrorMessage());
}
}
// 5. 获取UserTunnel(即使隧道未变化也需要获取,用于构建服务名称)
UserTunnel userTunnel = null;
if (currentUser.getRoleId() != ADMIN_ROLE_ID) {
userTunnel = getUserTunnel(currentUser.getUserId(), tunnel.getId().intValue());
if (userTunnel == null) {
return R.err("你没有该隧道权限");
}
} else {
// 管理员用户也需要获取UserTunnel(如果存在的话),用于构建正确的服务名称
// 通过forward记录获取原始的用户ID
userTunnel = getUserTunnel(existForward.getUserId(), tunnel.getId().intValue());
}
// 5. 更新Forward对象
// 6. 更新Forward对象
Forward updatedForward = updateForwardEntity(forwardUpdateDto, existForward, tunnel);
// 6. 获取所需的节点信息
// 7. 获取所需的节点信息
NodeInfo nodeInfo = getRequiredNodes(tunnel);
if (nodeInfo.isHasError()) {
return R.err(nodeInfo.getErrorMessage());
}
// 7. 调用Gost服务更新转发
// 8. 调用Gost服务更新转发
R gostResult;
if (isTunnelChanged(existForward, forwardUpdateDto)) {
// 隧道变化时:先删除原配置,再创建新配置
gostResult = updateGostServicesWithTunnelChange(existForward, updatedForward, tunnel,
permissionResult != null ? permissionResult.getLimiter() : null,
nodeInfo, permissionResult != null ? permissionResult.getUserTunnel() : null);
nodeInfo, userTunnel);
} else {
// 隧道未变化时:直接更新配置
gostResult = updateGostServices(updatedForward, tunnel,
permissionResult != null ? permissionResult.getLimiter() : null,
nodeInfo, permissionResult != null ? permissionResult.getUserTunnel() : null);
nodeInfo, userTunnel);
}
if (gostResult.getCode() != 0) {
return gostResult;
}
updatedForward.setStatus(1);
// 8. 保存更新
// 9. 保存更新
boolean result = this.updateById(updatedForward);
return result ? R.ok("端口转发更新成功") : R.err("端口转发更新失败");
}
@@ -213,6 +226,9 @@ public class ForwardServiceImpl extends ServiceImpl<ForwardMapper, Forward> impl
if (userTunnel == null) {
return R.err("你没有该隧道权限");
}
} else {
// 管理员删除用户记录时,需要获取对应的UserTunnel用于构建正确的服务名称
userTunnel = getUserTunnel(forward.getUserId(), tunnel.getId().intValue());
}
// 5. 获取所需的节点信息
@@ -317,14 +333,20 @@ public class ForwardServiceImpl extends ServiceImpl<ForwardMapper, Forward> impl
return R.err("你没有该隧道权限");
}
}
// 6. 确保获取UserTunnel用于构建服务名称(包括管理员用户)
if (userTunnel == null) {
// 通过forward记录获取原始的用户ID来查找UserTunnel
userTunnel = getUserTunnel(forward.getUserId(), tunnel.getId().intValue());
}
// 6. 获取所需的节点信息
// 7. 获取所需的节点信息
NodeInfo nodeInfo = getRequiredNodes(tunnel);
if (nodeInfo.isHasError()) {
return R.err(nodeInfo.getErrorMessage());
}
// 7. 调用Gost服务
// 8. 调用Gost服务
String serviceName = buildServiceName(forward.getId(), forward.getUserId(), forward.getTunnelId(), userTunnel);
GostDto gostResult;
@@ -354,7 +376,7 @@ public class ForwardServiceImpl extends ServiceImpl<ForwardMapper, Forward> impl
return R.err(operation + "服务失败:" + gostResult.getMsg());
}
// 8. 更新转发状态
// 9. 更新转发状态
forward.setStatus(targetStatus);
forward.setUpdatedTime(System.currentTimeMillis());
boolean result = this.updateById(forward);
@@ -691,20 +713,14 @@ public class ForwardServiceImpl extends ServiceImpl<ForwardMapper, Forward> impl
return R.err("原隧道不存在,无法删除旧配置");
}
// 2. 获取原隧道的节点信息
NodeInfo oldNodeInfo = getRequiredNodes(oldTunnel);
if (oldNodeInfo.isHasError()) {
log.warn("获取原隧道{}的节点信息失败: {}", oldTunnel.getId(), oldNodeInfo.getErrorMessage());
} else {
// 3. 删除原有的Gost服务配置
R deleteResult = deleteGostServices(existForward, oldTunnel, oldNodeInfo, userTunnel);
if (deleteResult.getCode() != 0) {
// 删除失败时记录日志,但不影响后续创建(可能原配置已不存在)
log.warn("删除原隧道{}的Gost配置失败: {}", oldTunnel.getId(), deleteResult.getMsg());
}
// 2. 删除原有的Gost服务配置
R deleteResult = deleteOldGostServices(existForward, oldTunnel, userTunnel);
if (deleteResult.getCode() != 0) {
// 删除失败时记录日志,但不影响后续创建(可能原配置已不存在)
log.warn("删除原隧道{}的Gost配置失败: {}", oldTunnel.getId(), deleteResult.getMsg());
}
// 4. 创建新的Gost服务配置
// 3. 创建新的Gost服务配置
R createResult = createGostServices(updatedForward, newTunnel, limiter, nodeInfo, userTunnel);
if (createResult.getCode() != 0) {
updateForwardStatusToError(updatedForward);
@@ -714,6 +730,55 @@ public class ForwardServiceImpl extends ServiceImpl<ForwardMapper, Forward> impl
return R.ok();
}
/**
* 删除原有的Gost服务(隧道变化时专用)
*/
private R deleteOldGostServices(Forward forward, Tunnel oldTunnel, UserTunnel userTunnel) {
// 获取原隧道的用户隧道关系
UserTunnel oldUserTunnel = getUserTunnel(forward.getUserId(), oldTunnel.getId().intValue());
String serviceName = buildServiceName(forward.getId(), forward.getUserId(), forward.getTunnelId(), oldUserTunnel);
// 获取原隧道的节点信息
NodeInfo oldNodeInfo = getRequiredNodes(oldTunnel);
// 删除主服务(使用原隧道的入口节点)
if (!oldNodeInfo.isHasError() && oldNodeInfo.getInNode() != null) {
GostDto serviceResult = GostUtil.DeleteService(oldNodeInfo.getInNode().getId(), serviceName);
if (!isGostOperationSuccess(serviceResult)) {
log.warn("删除主服务失败: {}", serviceResult.getMsg());
}
}
// 如果原隧道是隧道转发类型,需要删除链和远程服务
if (oldTunnel.getType() == TUNNEL_TYPE_TUNNEL_FORWARD) {
// 删除链服务
if (!oldNodeInfo.isHasError() && oldNodeInfo.getInNode() != null) {
GostDto chainResult = GostUtil.DeleteChains(oldNodeInfo.getInNode().getId(), serviceName);
if (!isGostOperationSuccess(chainResult)) {
log.warn("删除链服务失败: {}", chainResult.getMsg());
}
}
// 删除远程服务(即使节点信息获取失败,也要尝试删除)
Node outNode = null;
if (!oldNodeInfo.isHasError()) {
outNode = oldNodeInfo.getOutNode();
} else {
// 即使获取节点信息失败,也尝试直接获取出口节点来删除远程服务
outNode = nodeService.getNodeById(oldTunnel.getOutNodeId());
}
if (outNode != null) {
GostDto remoteResult = GostUtil.DeleteRemoteService(outNode.getId(), serviceName);
if (!isGostOperationSuccess(remoteResult)) {
log.warn("删除远程服务失败: {}", remoteResult.getMsg());
}
}
}
return R.ok();
}
/**
* 删除Gost服务
*/