chore(release): 发布 v1.9.2
CI / verify (push) Canceled after 0s
Docker Release / build-and-push (amd64, linux/amd64) (push) Canceled after 0s
Docker Release / build-and-push (arm64, linux/arm64) (push) Canceled after 0s
GitHub Release / release (push) Canceled after 0s
Docker Release / create-manifest (push) Canceled after 0s

- 「公网服务端」新增 Nginx 启停控制(Agent 0.15.2,停止带二次确认)
- frps 启动失败不再误报成功,并增强启动诊断
- frps vhost 端口禁止 80/443,避免与 VPS 上 Nginx 冲突
- 旧服务端配置的 tcp_mux / login_fail_exit 默认值补齐
- 新增安全审查、页面布局方案与场景操作文档
This commit is contained in:
2026-10-07 19:26:02 +08:00
parent a7d71752e3
commit 2a0740a60f
55 changed files with 5065 additions and 467 deletions
+78
View File
@@ -3,6 +3,7 @@ package agent
import (
"bytes"
"crypto/x509"
"encoding/json"
"encoding/pem"
"fmt"
"io"
@@ -309,6 +310,83 @@ func (s *Server) nginxReload(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, payload)
}
// nginxAction 启动 / 停止 / 重启 VPS 上的 Nginx。
//
// 不走 systemctl:宝塔、源码编译等安装方式没有 systemd unit,
// 统一用 nginx 自身信号控制,与 reloadNginx 走同一条路径,行为一致可预期。
func (s *Server) nginxAction(w http.ResponseWriter, r *http.Request) {
var req struct {
Action string `json:"action"`
}
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<10)).Decode(&req); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
return
}
if req.Action != "start" && req.Action != "stop" && req.Action != "restart" {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "action 需为 start/stop/restart"})
return
}
nginx := s.nginxPath()
if nginx == "" {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "未找到 nginx 可执行文件:请先安装 Nginx,或用 HAVLINE_AGENT_NGINX_BIN 指定路径"})
return
}
s.mu.Lock()
defer s.mu.Unlock()
if err := runNginxAction(nginx, req.Action); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
return
}
// 启停异步生效,留出短暂窗口再重新探测,避免页面读到旧状态
time.Sleep(600 * time.Millisecond)
s.invalidateNginxProbe()
probe := s.probeNginx()
writeJSON(w, http.StatusOK, map[string]any{
"ok": true,
"action": req.Action,
"nginx_running": probe.Running,
"nginx_ok": probe.OK,
"nginx_error": probe.Error,
})
}
// runNginxAction 执行一次启停;已经是目标状态时按幂等处理,不报错。
func runNginxAction(nginx, action string) error {
running := nginxRunning()
switch action {
case "start":
if running {
return nil
}
return startNginx(nginx)
case "stop":
if !running {
return nil
}
return stopNginx(nginx)
default: // restart
if running {
_ = stopNginx(nginx)
time.Sleep(500 * time.Millisecond)
}
return startNginx(nginx)
}
}
func startNginx(nginx string) error {
if out, err := exec.Command(nginx).CombinedOutput(); err != nil {
return fmt.Errorf("启动 Nginx 失败:%s", strings.TrimSpace(string(out)))
}
return nil
}
func stopNginx(nginx string) error {
if out, err := exec.Command(nginx, "-s", "stop").CombinedOutput(); err != nil {
return fmt.Errorf("停止 Nginx 失败:%s", strings.TrimSpace(string(out)))
}
return nil
}
// frpsConfig 回读 VPS 上的 frps.toml,供应用侧与「即将下发的内容」比对,检出被手工修改的情况
func (s *Server) frpsConfig(w http.ResponseWriter, _ *http.Request) {
path := strings.TrimSpace(s.cfg.FrpsConfigPath)
+2 -1
View File
@@ -31,7 +31,7 @@ import (
"github.com/havline/havline/internal/sysinfo"
)
const AgentVersion = "0.15.1"
const AgentVersion = "0.15.2"
// Config 是 havline-agent 的运行配置。agent 只在 VPS 本地管理自己的 Nginx 片段、证书和 frps 配置。
type Config struct {
@@ -224,6 +224,7 @@ func agentRoutes(s *Server) []agentRoute {
{http.MethodGet, "/api/v1/frps/config", s.frpsConfig},
{http.MethodGet, "/api/v1/certs", s.certs},
{http.MethodPost, "/api/v1/nginx/reload", s.nginxReload},
{http.MethodPost, "/api/v1/nginx/action", s.nginxAction},
}
}
+28 -1
View File
@@ -277,7 +277,12 @@ func (h *FRPMultiHandler) AgentInstall(w http.ResponseWriter, r *http.Request) {
writeError(r, w, http.StatusBadRequest, "FRP 服务端 ID 无效")
return
}
server, err := h.svc.AgentInstall(r.Context(), id)
var server frp.Server
if r.URL.Query().Get("trust_new_host_key") == "true" {
server, err = h.svc.AgentInstallTrustHostKey(r.Context(), id)
} else {
server, err = h.svc.AgentInstall(r.Context(), id)
}
if err != nil {
writeError(r, w, http.StatusBadGateway, err.Error())
return
@@ -796,6 +801,28 @@ func (h *FRPMultiHandler) AgentNginxReload(w http.ResponseWriter, r *http.Reques
writeJSON(w, http.StatusOK, result)
}
// AgentNginxAction 启动 / 停止 / 重启 VPS 上的 Nginx(需 agent ≥ 0.15.2)
func (h *FRPMultiHandler) AgentNginxAction(w http.ResponseWriter, r *http.Request) {
id, err := parseID(r.PathValue("id"))
if err != nil {
writeError(r, w, http.StatusBadRequest, "FRP 服务端 ID 无效")
return
}
var body struct {
Action string `json:"action"`
}
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
writeError(r, w, http.StatusBadRequest, "请求格式无效")
return
}
result, err := h.svc.AgentNginxAction(r.Context(), id, body.Action)
if err != nil {
writeError(r, w, http.StatusBadGateway, err.Error())
return
}
writeJSON(w, http.StatusOK, result)
}
func (h *FRPMultiHandler) AgentFRPSAction(w http.ResponseWriter, r *http.Request) {
id, err := parseID(r.PathValue("id"))
if err != nil {
+1
View File
@@ -168,6 +168,7 @@ func newRouterWithSpecs(deps Deps) (http.Handler, []routeSpec) {
protect("GET /api/frpmulti/servers/{id}/agent/certs", frpHandler.AgentCerts)
protect("POST /api/frpmulti/servers/{id}/agent/nginx/reload", frpHandler.AgentNginxReload)
protect("POST /api/frpmulti/servers/{id}/agent/nginx/install", frpHandler.AgentInstallNginx)
protect("POST /api/frpmulti/servers/{id}/agent/nginx/action", frpHandler.AgentNginxAction)
protect("GET /api/frpmulti/servers/{id}/agent/routes/{proxyID}/conf", frpHandler.AgentRouteConf)
protect("GET /api/frpmulti/servers/{id}/agent/routes/{proxyID}/versions", frpHandler.AgentRouteVersions)
protect("GET /api/frpmulti/servers/{id}/agent/routes/{proxyID}/versions/{name}", frpHandler.AgentRouteVersion)
+47 -1
View File
@@ -6,9 +6,11 @@ import (
"io"
"io/fs"
"log/slog"
"net"
"net/http"
"os"
"path/filepath"
"strings"
"time"
"github.com/havline/havline/internal/acme"
@@ -152,6 +154,9 @@ func New(cfg config.Config, staticFS fs.FS, migrationsDir string) (*App, error)
StaticFS: staticFS,
StartedAt: startedAt,
})
if prefix := strings.TrimSpace(cfg.GatewayPrefix); prefix != "" {
handler = stripPrefixIfPresent(prefix, handler)
}
server := &http.Server{
Addr: cfg.ListenAddr,
Handler: handler,
@@ -243,7 +248,32 @@ func (a *App) bootstrapNginx(ctx context.Context, proxySvc *service.ProxyService
}
func (a *App) Run() error {
return a.server.ListenAndServe()
if strings.TrimSpace(a.cfg.GatewaySocket) == "" {
return a.server.ListenAndServe()
}
// 先准备两端监听,任何一端失败都不留下半启动的服务。
tcp, err := net.Listen("tcp", a.server.Addr)
if err != nil {
return err
}
defer tcp.Close()
socket := strings.TrimSpace(a.cfg.GatewaySocket)
if err := os.Remove(socket); err != nil && !os.IsNotExist(err) {
return err
}
listener, err := net.Listen("unix", socket)
if err != nil {
return err
}
defer listener.Close()
defer os.Remove(socket)
if err := os.Chmod(socket, 0o660); err != nil {
return err
}
errCh := make(chan error, 2)
go func() { errCh <- a.server.Serve(tcp) }()
go func() { errCh <- a.server.Serve(listener) }()
return <-errCh
}
func (a *App) Shutdown(ctx context.Context) error {
@@ -254,6 +284,9 @@ func (a *App) Shutdown(ctx context.Context) error {
if err := a.server.Shutdown(ctx); err != nil {
return err
}
if socket := strings.TrimSpace(a.cfg.GatewaySocket); socket != "" {
_ = os.Remove(socket)
}
if err := a.nginx.Stop(ctx); err != nil {
a.logger.Error("nginx stop failed", "module", "NGINX", "error", err.Error())
}
@@ -268,6 +301,19 @@ func (a *App) Shutdown(ctx context.Context) error {
return nil
}
func stripPrefixIfPresent(prefix string, next http.Handler) http.Handler {
prefix = strings.TrimRight(prefix, "/")
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == prefix || strings.HasPrefix(r.URL.Path, prefix+"/") {
r.URL.Path = strings.TrimPrefix(r.URL.Path, prefix)
if r.URL.Path == "" {
r.URL.Path = "/"
}
}
next.ServeHTTP(w, r)
})
}
func ResolveMigrationsDir() (string, error) {
candidates := []string{
"migrations",
+25
View File
@@ -0,0 +1,25 @@
package app
import (
"net/http"
"net/http/httptest"
"testing"
)
func TestGatewayPrefix(t *testing.T) {
for _, tc := range []struct{ input, want string }{
{"/app/havline/api/version", "/api/version"},
{"/app/havline", "/"},
{"/api/version", "/api/version"},
{"/app/havline-other/api/version", "/app/havline-other/api/version"},
} {
t.Run(tc.input, func(t *testing.T) {
handler := stripPrefixIfPresent("/app/havline/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != tc.want {
t.Errorf("路径为 %q,期望 %q", r.URL.Path, tc.want)
}
}))
handler.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, tc.input, nil))
})
}
}
+10
View File
@@ -12,6 +12,8 @@ import (
type Config struct {
ListenAddr string
GatewaySocket string
GatewayPrefix string
DataDir string
SessionSecret string
NginxBin string
@@ -33,6 +35,8 @@ func Load() Config {
}
return Config{
ListenAddr: envOr("HAVLINE_LISTEN", ":6893"),
GatewaySocket: envOr("HAVLINE_GATEWAY_SOCKET", ""),
GatewayPrefix: envOr("HAVLINE_GATEWAY_PREFIX", ""),
DataDir: dataDir,
SessionSecret: envOr("HAVLINE_SESSION_SECRET", ""),
NginxBin: envOr("HAVLINE_NGINX_BIN", "nginx"),
@@ -108,6 +112,12 @@ func (c Config) SSHKnownHostsPath() string {
return filepath.Join(c.DataDir, "ssh_known_hosts")
}
// SSHKnownHostsPathForServer 为每台 FRP 服务端提供独立的 known_hosts,
// 避免一台 VPS 的主机密钥记录影响另一台服务器的 SSH 连接。
func (c Config) SSHKnownHostsPathForServer(serverID int64) string {
return filepath.Join(c.DataDir, "ssh_known_hosts", fmt.Sprintf("server-%d", serverID))
}
func (c Config) ErrorsDir() string {
return c.NginxDir() + "/errors"
}
+9
View File
@@ -2,6 +2,7 @@ package config
import (
"os"
"path/filepath"
"testing"
)
@@ -42,3 +43,11 @@ func TestLoadUpdaterTokenFile(t *testing.T) {
t.Fatalf("updater token file %q", cfg.UpdaterTokenFile)
}
}
func TestSSHKnownHostsPathForServer(t *testing.T) {
cfg := Config{DataDir: "/data"}
want := filepath.Join("/data", "ssh_known_hosts", "server-3")
if got := cfg.SSHKnownHostsPathForServer(3); got != want {
t.Fatalf("per-server known_hosts path %q", got)
}
}
+16
View File
@@ -0,0 +1,16 @@
package config
import "testing"
func TestNativeGatewayConfig(t *testing.T) {
t.Setenv("HAVLINE_GATEWAY_SOCKET", "/tmp/havline-test.sock")
t.Setenv("HAVLINE_GATEWAY_PREFIX", "/app/havline")
t.Setenv("HAVLINE_UPDATER_SOCKET", "")
cfg := Load()
if cfg.GatewaySocket != "/tmp/havline-test.sock" || cfg.GatewayPrefix != "/app/havline" {
t.Fatal("原生网关配置未生效")
}
if cfg.UpdaterSocket != "" {
t.Fatal("原生应用显式空值应禁用 Docker 升级")
}
}
+24
View File
@@ -219,9 +219,15 @@ const agentMinVersionInsights = "0.11.0"
// agentMinVersionRouteVersions 是域名 vhost 版本列表 / 对比 / 回滚端点要求的最低 agent 版本
const agentMinVersionRouteVersions = "0.12.2"
// agentMinVersionNginxAction 是 Nginx 启停(start / stop / restart)端点要求的最低 agent 版本
const agentMinVersionNginxAction = "0.15.2"
// ErrAgentEndpointMissing 表示 VPS 上的 agent 版本过旧、没有该端点(HTTP 404/405)
var ErrAgentEndpointMissing = errors.New("VPS 上的 agent 版本过低,不支持该接口")
// ErrAgentTokenMismatch 表示 Havline 保存的 Token 与 VPS 上 Agent 配置不一致。
var ErrAgentTokenMismatch = errors.New("Agent Token 不匹配,请点击「升级 Agent」重新同步")
// statusError 保留 HTTP 状态码,便于上层区分「端点不存在」与真实错误
type statusError struct {
code int
@@ -232,6 +238,14 @@ func (e *statusError) Error() string {
return fmt.Sprintf("agent 返回 HTTP %d: %s", e.code, e.body)
}
func wrapAgentAuth(err error) error {
var statusErr *statusError
if errors.As(err, &statusErr) && statusErr.code == http.StatusUnauthorized {
return fmt.Errorf("%w: %s", ErrAgentTokenMismatch, statusErr.body)
}
return err
}
// wrapEndpointMissing 把「端点不存在」统一成带升级提示的错误(按端点给出最低版本),其它错误原样返回
func wrapEndpointMissing(err error, minVersion string) error {
var statusErr *statusError
@@ -280,6 +294,16 @@ func (c *AgentClient) NginxReload(ctx context.Context) (map[string]any, error) {
return result, nil
}
// NginxAction 启动 / 停止 / 重启 VPS 上的 Nginx(需 agent ≥ 0.15.2)
func (c *AgentClient) NginxAction(ctx context.Context, action string) (map[string]any, error) {
var result map[string]any
payload := map[string]string{"action": action}
if err := c.request(ctx, http.MethodPost, "/api/v1/nginx/action", payload, &result); err != nil {
return nil, wrapEndpointMissing(err, agentMinVersionNginxAction)
}
return result, nil
}
// FRPSConfig 回读 VPS 上的 frps.toml(用于与下发内容比对)
func (c *AgentClient) FRPSConfig(ctx context.Context) (map[string]any, error) {
var result map[string]any
+20 -5
View File
@@ -14,7 +14,8 @@ func (s *Service) AgentMetrics(ctx context.Context, serverID int64) (map[string]
if !ok {
return nil, fmt.Errorf("公网 agent 未配置或 Token 未保存")
}
return client.Metrics(ctx)
result, err := client.Metrics(ctx)
return result, wrapAgentAuth(err)
}
// AgentNginxLogs 读取 VPS 上 Nginx 的 access / error 日志尾部
@@ -29,7 +30,8 @@ func (s *Service) AgentNginxLogs(ctx context.Context, serverID int64, kind strin
if lines <= 0 || lines > 2000 {
lines = 200
}
return client.NginxLogs(ctx, kind, lines)
result, err := client.NginxLogs(ctx, kind, lines)
return result, wrapAgentAuth(err)
}
// AgentCerts 列出 VPS 证书目录下的域名与到期时间,供与 Havline 证书库交叉核对
@@ -38,7 +40,8 @@ func (s *Service) AgentCerts(ctx context.Context, serverID int64) (map[string]an
if !ok {
return nil, fmt.Errorf("公网 agent 未配置或 Token 未保存")
}
return client.Certs(ctx)
result, err := client.Certs(ctx)
return result, wrapAgentAuth(err)
}
// AgentNginxReload 显式校验并重载 VPS 上的 Nginx,返回失败原因
@@ -47,7 +50,18 @@ func (s *Service) AgentNginxReload(ctx context.Context, serverID int64) (map[str
if !ok {
return nil, fmt.Errorf("公网 agent 未配置或 Token 未保存")
}
return client.NginxReload(ctx)
result, err := client.NginxReload(ctx)
return result, wrapAgentAuth(err)
}
// AgentNginxAction 启动 / 停止 / 重启 VPS 上的 Nginx(需 agent ≥ 0.15.2)
func (s *Service) AgentNginxAction(ctx context.Context, serverID int64, action string) (map[string]any, error) {
client, ok := s.agentClient(ctx, serverID)
if !ok {
return nil, fmt.Errorf("公网 agent 未配置或 Token 未保存")
}
result, err := client.NginxAction(ctx, action)
return result, wrapAgentAuth(err)
}
// AgentFRPSConfig 回读 VPS 上的 frps.toml,供与「即将下发的内容」比对
@@ -56,5 +70,6 @@ func (s *Service) AgentFRPSConfig(ctx context.Context, serverID int64) (map[stri
if !ok {
return nil, fmt.Errorf("公网 agent 未配置或 Token 未保存")
}
return client.FRPSConfig(ctx)
result, err := client.FRPSConfig(ctx)
return result, wrapAgentAuth(err)
}
+5 -1
View File
@@ -116,7 +116,11 @@ func extractToken(output string) string {
func extractAgentVersion(output string) string {
for _, line := range strings.Split(output, "\n") {
if version, found := strings.CutPrefix(strings.TrimSpace(line), "agent version: "); found {
return strings.TrimSpace(version)
fields := strings.Fields(version)
if len(fields) == 0 {
return ""
}
return strings.TrimPrefix(fields[len(fields)-1], "v")
}
}
return ""
+6
View File
@@ -7,6 +7,12 @@ func TestExtractAgentVersion(t *testing.T) {
if got := extractAgentVersion(output); got != "0.15.1" {
t.Fatalf("版本解析失败,got %q", got)
}
if got := extractAgentVersion("agent version: havline-agent 0.15.1\nhavline-agent installed"); got != "0.15.1" {
t.Fatalf("带程序名前缀的版本解析失败,got %q", got)
}
if got := extractAgentVersion("agent version: v0.15.1"); got != "0.15.1" {
t.Fatalf("带 v 前缀的版本解析失败,got %q", got)
}
if got := extractAgentVersion("havline-agent installed\nToken: abc123"); got != "" {
t.Fatalf("无版本行时应返回空串,got %q", got)
}
+138 -25
View File
@@ -2,6 +2,7 @@ package frp
import (
"context"
"encoding/json"
"fmt"
"io"
"net"
@@ -14,25 +15,31 @@ import (
const agentRemoteAddr = "127.0.0.1:7700"
const (
agentTunnelKeepalive = 30 * time.Second
agentTunnelBackoffMin = time.Second
agentTunnelBackoffMax = 60 * time.Second
agentTunnelKeepalive = 30 * time.Second
agentTunnelKeepaliveTimeout = 5 * time.Second
agentTunnelBackoffMin = time.Second
agentTunnelBackoffMax = 60 * time.Second
)
type agentTunnel struct {
serverID int64
localPort int
ssh SSHInstaller
serverID int64
localPort int
remoteAddr string
ssh SSHInstaller
ctx context.Context
cancel context.CancelFunc
connectMu sync.Mutex
mu sync.Mutex
listener net.Listener
client *ssh.Client
state string
lastError string
wg sync.WaitGroup
connsMu sync.Mutex
conns map[net.Conn]struct{}
}
func newAgentTunnel(parent context.Context, serverID int64, localPort int, installer SSHInstaller) *agentTunnel {
@@ -42,13 +49,15 @@ func newAgentTunnel(parent context.Context, serverID int64, localPort int, insta
func newAgentTunnelWithListener(parent context.Context, serverID int64, localPort int, installer SSHInstaller, listener net.Listener) *agentTunnel {
ctx, cancel := context.WithCancel(parent)
return &agentTunnel{
serverID: serverID,
localPort: localPort,
ssh: installer,
ctx: ctx,
cancel: cancel,
listener: listener,
state: "connecting",
serverID: serverID,
localPort: localPort,
remoteAddr: agentRemoteAddr,
ssh: installer,
ctx: ctx,
cancel: cancel,
listener: listener,
state: "connecting",
conns: map[net.Conn]struct{}{},
}
}
@@ -61,8 +70,13 @@ func (t *agentTunnel) Start() error {
}
t.listener = listener
}
if t.localPort == 0 {
if addr, ok := t.listener.Addr().(*net.TCPAddr); ok {
t.localPort = addr.Port
}
}
t.mu.Lock()
t.state = "connected"
t.state = "connecting"
t.lastError = ""
t.mu.Unlock()
t.wg.Add(2)
@@ -76,7 +90,7 @@ func (t *agentTunnel) monitorLoop() {
defer t.wg.Done()
backoff := agentTunnelBackoffMin
for {
_, err := t.ensureClient()
err := t.check()
wait := agentTunnelKeepalive
if err != nil {
t.setState("reconnecting", err)
@@ -92,6 +106,24 @@ func (t *agentTunnel) monitorLoop() {
}
}
// check 同时验证 SSH 连接和远端 Agent 端口,避免只连上 SSH 就误判隧道可用。
func (t *agentTunnel) check() error {
client, err := t.ensureClient()
if err != nil {
return err
}
if err := t.keepalive(client); err != nil {
t.invalidateClient(client)
return fmt.Errorf("SSH keepalive 失败: %w", err)
}
remote, err := client.Dial("tcp", t.remoteAddr)
if err != nil {
return fmt.Errorf("连接远端 Agent 端口失败: %w", err)
}
_ = remote.Close()
return nil
}
func (t *agentTunnel) sleep(d time.Duration) bool {
timer := time.NewTimer(d)
defer timer.Stop()
@@ -126,8 +158,13 @@ func (t *agentTunnel) acceptLoop() {
return
}
t.wg.Add(1)
if !t.trackConn(conn) {
t.wg.Done()
return
}
go func() {
defer t.wg.Done()
defer t.untrackConn(conn)
t.handle(conn)
}()
}
@@ -138,31 +175,53 @@ func (t *agentTunnel) handle(local net.Conn) {
client, err := t.ensureClient()
if err != nil {
t.setState("reconnecting", err)
writeTunnelError(local, err)
return
}
remote, err := client.Dial("tcp", agentRemoteAddr)
remote, err := client.Dial("tcp", t.remoteAddr)
if err != nil {
t.invalidateClient(client)
t.setState("reconnecting", err)
return
remote, err = t.retryRemoteDial(client, err)
if err != nil {
t.setState("reconnecting", err)
writeTunnelError(local, err)
return
}
}
defer remote.Close()
t.setState("connected", nil)
proxyTunnel(local, remote)
}
// retryRemoteDial 仅在 SSH 连接本身失效时重连,远端 Agent 端口暂时不可用不会关闭共享连接。
func (t *agentTunnel) retryRemoteDial(client *ssh.Client, firstErr error) (net.Conn, error) {
if err := t.keepalive(client); err == nil {
return nil, fmt.Errorf("连接远端 Agent 端口失败: %w", firstErr)
}
t.invalidateClient(client)
reconnected, err := t.ensureClient()
if err != nil {
return nil, fmt.Errorf("SSH 连接失效,重连失败: %w", err)
}
remote, err := reconnected.Dial("tcp", t.remoteAddr)
if err != nil {
return nil, fmt.Errorf("SSH 已重连,但远端 Agent 端口仍不可用: %w", err)
}
return remote, nil
}
func (t *agentTunnel) ensureClient() (*ssh.Client, error) {
// 单飞:监控循环和请求处理可能同时触发建连,串行化后只允许一条 SSH 握手。
// 正常请求只复用连接,不在热路径做 keepalive,避免多个接口互相排队。
t.connectMu.Lock()
defer t.connectMu.Unlock()
t.mu.Lock()
if t.client != nil {
client := t.client
t.mu.Unlock()
if _, _, err := client.SendRequest("keepalive@openssh.com", true, nil); err == nil {
return client, nil
}
t.invalidateClient(client)
} else {
t.mu.Unlock()
return client, nil
}
t.mu.Unlock()
ctx, cancel := context.WithTimeout(t.ctx, 15*time.Second)
defer cancel()
client, err := t.ssh.client(ctx)
@@ -181,6 +240,30 @@ func (t *agentTunnel) ensureClient() (*ssh.Client, error) {
return client, nil
}
func (t *agentTunnel) keepalive(client *ssh.Client) error {
done := make(chan error, 1)
go func() {
_, _, err := client.SendRequest("keepalive@openssh.com", true, nil)
done <- err
}()
select {
case err := <-done:
return err
case <-time.After(agentTunnelKeepaliveTimeout):
return fmt.Errorf("SSH keepalive 超时")
case <-t.ctx.Done():
return t.ctx.Err()
}
}
// writeTunnelError 让本地 HTTP 客户端立即得到明确错误,而不是一直等到 context deadline。
func writeTunnelError(conn net.Conn, err error) {
body, _ := json.Marshal(map[string]string{"error": "SSH 隧道不可用: " + err.Error()})
_, _ = fmt.Fprintf(conn,
"HTTP/1.1 502 Bad Gateway\r\nContent-Type: application/json\r\nContent-Length: %d\r\nConnection: close\r\n\r\n%s",
len(body), body)
}
func (t *agentTunnel) invalidateClient(client *ssh.Client) {
t.mu.Lock()
if t.client == client {
@@ -217,9 +300,39 @@ func (t *agentTunnel) Stop() {
_ = t.client.Close()
}
t.mu.Unlock()
t.closeConns()
t.wg.Wait()
}
func (t *agentTunnel) trackConn(conn net.Conn) bool {
t.connsMu.Lock()
defer t.connsMu.Unlock()
if t.ctx.Err() != nil {
_ = conn.Close()
return false
}
t.conns[conn] = struct{}{}
return true
}
func (t *agentTunnel) untrackConn(conn net.Conn) {
t.connsMu.Lock()
delete(t.conns, conn)
t.connsMu.Unlock()
}
func (t *agentTunnel) closeConns() {
t.connsMu.Lock()
conns := make([]net.Conn, 0, len(t.conns))
for conn := range t.conns {
conns = append(conns, conn)
}
t.connsMu.Unlock()
for _, conn := range conns {
_ = conn.Close()
}
}
func proxyTunnel(a, b net.Conn) {
var wg sync.WaitGroup
wg.Add(2)
@@ -0,0 +1,246 @@
package frp
import (
"context"
"crypto/ed25519"
"crypto/rand"
"fmt"
"io"
"net"
"net/http"
"net/http/httptest"
"path/filepath"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"golang.org/x/crypto/ssh"
)
type directTCPIPPayload struct {
DestAddr string
DestPort uint32
OriginAddr string
OriginPort uint32
}
func TestAgentTunnelReturnsBadGatewayWhenRemoteAgentIsDown(t *testing.T) {
sshAddr := startFakeSSHServer(t)
tunnel := startTestAgentTunnel(t, sshAddr, "127.0.0.1:1")
response, err := (&http.Client{Timeout: 5 * time.Second}).Get(
fmt.Sprintf("http://127.0.0.1:%d/api/v1/status", tunnel.localPort),
)
if err != nil {
t.Fatalf("应返回 HTTP 502,而不是 EOF 或超时: %v", err)
}
defer response.Body.Close()
if response.StatusCode != http.StatusBadGateway {
t.Fatalf("状态码 = %d,期望 502", response.StatusCode)
}
body, _ := io.ReadAll(response.Body)
if !strings.Contains(string(body), "SSH 隧道不可用") {
t.Fatalf("响应应包含隧道失败原因,实际为 %q", string(body))
}
}
func TestAgentTunnelForwardsHTTPToRemoteAgent(t *testing.T) {
remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"ok":true}`))
}))
defer remote.Close()
sshAddr := startFakeSSHServer(t)
tunnel := startTestAgentTunnel(t, sshAddr, strings.TrimPrefix(remote.URL, "http://"))
response, err := (&http.Client{Timeout: 5 * time.Second}).Get(
fmt.Sprintf("http://127.0.0.1:%d/api/v1/status", tunnel.localPort),
)
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
body, _ := io.ReadAll(response.Body)
if response.StatusCode != http.StatusOK || string(body) != `{"ok":true}` {
t.Fatalf("隧道转发失败: status=%d body=%q", response.StatusCode, string(body))
}
}
func TestAgentTunnelSerializesConcurrentSSHConnections(t *testing.T) {
remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, _ = w.Write([]byte(`{"ok":true}`))
}))
defer remote.Close()
sshAddr, connectionCount := startFakeSSHServerWithCounter(t, 300*time.Millisecond)
tunnel := startTestAgentTunnel(t, sshAddr, strings.TrimPrefix(remote.URL, "http://"))
client := &http.Client{Timeout: 5 * time.Second}
var wait sync.WaitGroup
for index := 0; index < 4; index++ {
wait.Add(1)
go func() {
defer wait.Done()
response, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d/api/v1/status", tunnel.localPort))
if err == nil {
_, _ = io.Copy(io.Discard, response.Body)
_ = response.Body.Close()
}
}()
}
wait.Wait()
if got := connectionCount.Load(); got != 1 {
t.Fatalf("并发请求建立了 %d 条 SSH 连接,期望单飞复用 1 条", got)
}
}
func TestAgentTunnelDoesNotBlockRequestsOnSlowKeepalive(t *testing.T) {
remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, _ = w.Write([]byte(`{"ok":true}`))
}))
defer remote.Close()
sshAddr, _ := startFakeSSHServerWithDelays(t, 0, 3*time.Second)
tunnel := startTestAgentTunnel(t, sshAddr, strings.TrimPrefix(remote.URL, "http://"))
started := time.Now()
response, err := (&http.Client{Timeout: 5 * time.Second}).Get(
fmt.Sprintf("http://127.0.0.1:%d/api/v1/status", tunnel.localPort),
)
if err != nil {
t.Fatal(err)
}
defer response.Body.Close()
_, _ = io.Copy(io.Discard, response.Body)
if elapsed := time.Since(started); elapsed > time.Second {
t.Fatalf("正常请求被慢 keepalive 阻塞了 %v", elapsed)
}
}
func startTestAgentTunnel(t *testing.T, sshAddr, remoteAddr string) *agentTunnel {
t.Helper()
host, portText, err := net.SplitHostPort(sshAddr)
if err != nil {
t.Fatal(err)
}
port, err := strconv.Atoi(portText)
if err != nil {
t.Fatal(err)
}
tunnel := newAgentTunnel(context.Background(), 1, 0, SSHInstaller{
Host: host, Port: port, User: "test", Password: "test",
Timeout: 2 * time.Second, KnownHostsPath: filepath.Join(t.TempDir(), "known_hosts"),
})
tunnel.remoteAddr = remoteAddr
if err := tunnel.Start(); err != nil {
t.Fatal(err)
}
t.Cleanup(tunnel.Stop)
return tunnel
}
func startFakeSSHServer(t *testing.T) string {
t.Helper()
addr, _ := startFakeSSHServerWithCounter(t, 0)
return addr
}
func startFakeSSHServerWithCounter(t *testing.T, handshakeDelay time.Duration) (string, *atomic.Int32) {
t.Helper()
return startFakeSSHServerWithDelays(t, handshakeDelay, 0)
}
func startFakeSSHServerWithDelays(t *testing.T, handshakeDelay, keepaliveDelay time.Duration) (string, *atomic.Int32) {
t.Helper()
_, privateKey, err := ed25519.GenerateKey(rand.Reader)
if err != nil {
t.Fatal(err)
}
signer, err := ssh.NewSignerFromKey(privateKey)
if err != nil {
t.Fatal(err)
}
config := &ssh.ServerConfig{
PasswordCallback: func(ssh.ConnMetadata, []byte) (*ssh.Permissions, error) {
return nil, nil
},
}
config.AddHostKey(signer)
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = listener.Close() })
var connectionCount atomic.Int32
go func() {
for {
conn, err := listener.Accept()
if err != nil {
return
}
go serveFakeSSHConn(conn, config, &connectionCount, handshakeDelay, keepaliveDelay)
}
}()
return listener.Addr().String(), &connectionCount
}
func serveFakeSSHConn(
conn net.Conn,
config *ssh.ServerConfig,
connectionCount *atomic.Int32,
handshakeDelay, keepaliveDelay time.Duration,
) {
if handshakeDelay > 0 {
time.Sleep(handshakeDelay)
}
connectionCount.Add(1)
serverConn, channels, requests, err := ssh.NewServerConn(conn, config)
if err != nil {
_ = conn.Close()
return
}
defer serverConn.Close()
go func() {
for request := range requests {
if request.Type == "keepalive@openssh.com" && keepaliveDelay > 0 {
time.Sleep(keepaliveDelay)
}
if request.WantReply {
_ = request.Reply(true, nil)
}
}
}()
for newChannel := range channels {
if newChannel.ChannelType() != "direct-tcpip" {
_ = newChannel.Reject(ssh.UnknownChannelType, "unsupported")
continue
}
var payload directTCPIPPayload
if err := ssh.Unmarshal(newChannel.ExtraData(), &payload); err != nil {
_ = newChannel.Reject(ssh.ConnectionFailed, "invalid direct-tcpip payload")
continue
}
target := net.JoinHostPort(payload.DestAddr, strconv.Itoa(int(payload.DestPort)))
remote, err := net.DialTimeout("tcp", target, 2*time.Second)
if err != nil {
_ = newChannel.Reject(ssh.ConnectionFailed, err.Error())
continue
}
channel, channelRequests, err := newChannel.Accept()
if err != nil {
_ = remote.Close()
continue
}
go ssh.DiscardRequests(channelRequests)
go func() {
_, _ = io.Copy(remote, channel)
_ = remote.Close()
}()
go func() {
_, _ = io.Copy(channel, remote)
_ = channel.Close()
}()
}
}
+16 -2
View File
@@ -24,8 +24,8 @@ func TestAgentTunnelBindsLoopbackAndReleasesPort(t *testing.T) {
t.Fatal("隧道启动后本地端口应已被占用")
}
state, _ := tunnel.State()
if state != "connected" && state != "reconnecting" {
t.Fatalf("state=%q, want connected 或 reconnecting", state)
if state != "connecting" && state != "connected" && state != "reconnecting" {
t.Fatalf("state=%q, want connecting、connected 或 reconnecting", state)
}
tunnel.Stop()
if listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", port)); err != nil {
@@ -51,3 +51,17 @@ func TestNextTunnelBackoff(t *testing.T) {
}
}
}
func TestAgentTunnelStopClosesActiveConnections(t *testing.T) {
tunnel := newAgentTunnel(context.Background(), 1, 0, SSHInstaller{})
server, client := net.Pipe()
defer client.Close()
if !tunnel.trackConn(server) {
t.Fatal("隧道运行时应能登记活动连接")
}
tunnel.Stop()
_ = client.SetReadDeadline(time.Now().Add(time.Second))
if _, err := client.Read(make([]byte, 1)); err == nil {
t.Fatal("隧道停止后活动连接应被关闭")
}
}
Binary file not shown.
Binary file not shown.
+26
View File
@@ -1,6 +1,7 @@
package frp
import (
"encoding/json"
"strings"
"testing"
)
@@ -62,4 +63,29 @@ func TestRenderNATTraversalOnlyWhenDisabledAssistedAddrs(t *testing.T) {
}
}
// 旧数据(config_json = "{}")缺省布尔键时必须补回表单默认值,
// 否则会生成 transport.tcpMux = false,让每条新连接都重建 TCP+TLS。
func TestServerOptionsBooleanDefaultsWhenKeysMissing(t *testing.T) {
var legacy ServerOptions
if err := json.Unmarshal([]byte("{}"), &legacy); err != nil {
t.Fatalf("解析旧版空配置失败:%v", err)
}
legacy = normalizeServerOptions(legacy)
if !legacy.TCPMux || !legacy.LoginFailExit {
t.Fatalf("缺省键应补默认值 true,实际 tcpMux=%v loginFailExit=%v", legacy.TCPMux, legacy.LoginFailExit)
}
content := renderTOML(renderedConfig{Server: Server{ServerAddr: "vps.example.com", ServerPort: 7000, Options: legacy}})
if !strings.Contains(content, "transport.tcpMux = true") {
t.Fatalf("旧版配置应渲染 transport.tcpMux = true:\n%s", content)
}
var explicit ServerOptions
if err := json.Unmarshal([]byte(`{"tcp_mux":false,"login_fail_exit":false}`), &explicit); err != nil {
t.Fatalf("解析显式配置失败:%v", err)
}
if explicit.TCPMux || explicit.LoginFailExit {
t.Fatalf("显式 false 不应被默认值覆盖,实际 tcpMux=%v loginFailExit=%v", explicit.TCPMux, explicit.LoginFailExit)
}
}
func intPtr(value int) *int { return &value }
+8
View File
@@ -102,3 +102,11 @@ func (s *Service) waitForManagers(ctx context.Context, action func(*Manager) err
}
return nil
}
func sshSettingsChanged(existing, updated Server, secretUpdated bool) bool {
return existing.SSHHost != updated.SSHHost ||
existing.SSHPort != updated.SSHPort ||
existing.SSHUser != updated.SSHUser ||
existing.SSHAuth != updated.SSHAuth ||
secretUpdated
}
+198 -14
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"log/slog"
"net"
"net/url"
"os"
"path/filepath"
"sort"
@@ -416,6 +417,17 @@ func (s *Service) StartServer(ctx context.Context, id int64) (Server, error) {
if err != nil {
return Server{}, err
}
hasProxies, err := s.hasEnabledProxies(ctx, id)
if err != nil {
return Server{}, err
}
if !hasProxies {
// 空服务端没有 frpc 需要运行,保持 stopped,避免后续监控把它误判为异常退出。
s.setServerRuntime(ctx, id, "stopped", "")
s.enrichTLSStatus(&server)
s.enrichServerRuntime(ctx, &server)
return server, nil
}
s.setServerRuntime(ctx, id, "running", "")
s.enrichTLSStatus(&server)
s.enrichServerRuntime(ctx, &server)
@@ -638,6 +650,14 @@ func (s *Service) UpdateServer(ctx context.Context, id int64, in ServerInput) (S
if !server.Enabled {
_ = s.managerForServer(id).Stop(ctx)
}
sshChanged := sshSettingsChanged(existing, server, sshSecretUpdate)
if sshChanged {
// SSH 目标或凭据变化后必须废弃旧隧道,避免新服务端继续复用旧 VPS 的连接。
s.stopAgentTunnel(id)
}
if sshChanged || agentTokenUpdate || existing.AgentURL != server.AgentURL {
s.invalidateAgentClient(id)
}
if !server.MgmtEnabled || server.AgentTransport != "tunnel" {
s.stopAgentTunnel(id)
}
@@ -654,6 +674,7 @@ func (s *Service) DeleteServer(ctx context.Context, id int64) error {
}
_ = os.Remove(s.serverConfigPath(id))
_ = os.Remove(s.serverLogPath(id))
_ = os.Remove(s.cfg.SSHKnownHostsPathForServer(id))
s.removeManager(id)
return s.persistConfig(ctx)
}
@@ -715,7 +736,7 @@ func (s *Service) sshInstaller(ctx context.Context, serverID int64) (Server, SSH
return Server{}, SSHInstaller{}, err
}
}
installer := SSHInstaller{Host: server.SSHHost, Port: server.SSHPort, User: server.SSHUser, KnownHostsPath: s.cfg.SSHKnownHostsPath()}
installer := SSHInstaller{Host: server.SSHHost, Port: server.SSHPort, User: server.SSHUser, KnownHostsPath: s.cfg.SSHKnownHostsPathForServer(serverID)}
if server.SSHAuth == "password" {
installer.Password = secret
} else {
@@ -733,6 +754,20 @@ func (s *Service) AgentProbe(ctx context.Context, serverID int64) (map[string]an
}
func (s *Service) AgentInstall(ctx context.Context, serverID int64) (Server, error) {
return s.agentInstall(ctx, serverID, false)
}
// AgentInstallTrustHostKey 在用户确认 VPS 已重装系统后,先更新该服务端的 SSH 主机密钥记录,再安装 Agent。
func (s *Service) AgentInstallTrustHostKey(ctx context.Context, serverID int64) (Server, error) {
return s.agentInstall(ctx, serverID, true)
}
func (s *Service) agentInstall(ctx context.Context, serverID int64, trustHostKey bool) (Server, error) {
if trustHostKey {
if err := s.resetAgentSSHKnownHost(ctx, serverID); err != nil {
return Server{}, err
}
}
server, installer, err := s.sshInstaller(ctx, serverID)
if err != nil {
return Server{}, err
@@ -745,9 +780,17 @@ func (s *Service) AgentInstall(ctx context.Context, serverID int64) (Server, err
if err != nil {
return Server{}, err
}
// 已是 HTTPS 直连时,升级 Agent 不应把用户的传输方式改回隧道。
// 已是 HTTPS 直连时优先保持原传输方式;服务器重置或证书丢失时自动回退隧道,
// 避免 SSH 安装已经成功,却因为旧证书指纹失效而把整个升级判定为失败。
if server.AgentTransport == "https-pin" {
return s.keepAgentHTTPSAfterInstall(ctx, server, token, agentToken)
updated, keepErr := s.keepAgentHTTPSAfterInstall(ctx, server, token, agentToken)
if keepErr == nil {
return updated, nil
}
s.logger.Warn("HTTPS 直连验证失败,自动回退 SSH 隧道",
"server_id", serverID,
"error", keepErr.Error(),
)
}
localPort, listener, err := s.reserveAgentTunnelPort(ctx, serverID, server.AgentLocalPort)
if err != nil {
@@ -796,6 +839,23 @@ func (s *Service) AgentInstall(ctx context.Context, serverID int64) (Server, err
return updated, nil
}
func (s *Service) resetAgentSSHKnownHost(ctx context.Context, serverID int64) error {
server, err := s.store.GetServer(ctx, serverID)
if err != nil {
return err
}
if strings.TrimSpace(server.SSHHost) == "" || strings.TrimSpace(server.SSHUser) == "" {
return fmt.Errorf("请先填写 SSH 地址与用户")
}
s.stopAgentTunnel(serverID)
s.invalidateAgentClient(serverID)
path := s.cfg.SSHKnownHostsPathForServer(serverID)
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("重置 SSH 主机密钥记录失败: %w", err)
}
return nil
}
// keepAgentHTTPSAfterInstall 在 HTTPS 直连模式下完成 Agent 升级:
// 保持 agent_url、pin 与 transport 不变,只更新 Token 密文并重新验证连通。
func (s *Service) keepAgentHTTPSAfterInstall(ctx context.Context, server Server, token, agentToken string) (Server, error) {
@@ -1264,16 +1324,26 @@ func (s *Service) SwitchAgentToTunnel(ctx context.Context, serverID int64) (Serv
return Server{}, err
}
if server.AgentTransport == "https-pin" {
tunnel, err := s.ensureAgentTunnel(ctx, serverID)
if err != nil {
return Server{}, fmt.Errorf("建立 SSH 隧道失败: %w", err)
}
token, err := s.agentToken(ctx, serverID)
if err != nil {
return Server{}, err
}
client := NewAgentClientWithPin(agentHTTPSBaseURL(server), token, server.AgentTLSPin)
if err := client.DisableHTTPS(ctx); err != nil {
return Server{}, fmt.Errorf("关闭 HTTPS 直连失败: %w", err)
// 切换时不再依赖旧 HTTPS 直连;先通过隧道验证 Agent,再清理旧 HTTPS 配置。
client := NewAgentClient(fmt.Sprintf("http://127.0.0.1:%d", tunnel.localPort), token)
if err := client.Ping(ctx); err != nil {
return Server{}, fmt.Errorf("SSH 隧道已建立,但 Agent 验证失败: %w", err)
}
}
if _, err := s.ensureAgentTunnel(ctx, serverID); err != nil {
if err := client.DisableHTTPS(ctx); err != nil {
s.logger.Warn("关闭旧 HTTPS 直连失败,已保留 SSH 隧道",
"server_id", serverID,
"error", err.Error(),
)
}
} else if _, err := s.ensureAgentTunnel(ctx, serverID); err != nil {
return Server{}, err
}
updated, err := s.store.GetServer(ctx, serverID)
@@ -1319,11 +1389,64 @@ func (s *Service) AgentClient(ctx context.Context, serverID int64) (*AgentClient
}
func (s *Service) AgentStatus(ctx context.Context, serverID int64) (map[string]any, error) {
client, ok := s.agentClient(ctx, serverID)
if !ok {
return nil, fmt.Errorf("公网 agent 未配置或 Token 未保存")
statusCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
server, err := s.store.GetServer(statusCtx, serverID)
if err != nil {
return nil, err
}
return client.Status(ctx)
if !server.AgentConfigured {
return map[string]any{
"configured": false,
"mgmt_enabled": false,
"available": false,
"message": "公网 agent 未配置或 Token 未保存",
}, nil
}
if !server.MgmtEnabled {
return map[string]any{
"configured": true,
"mgmt_enabled": false,
"available": false,
"message": "Agent 已配置,但未启用日常管理",
}, nil
}
if server.AgentTransport == "tunnel" {
if state, lastError, ok := s.tunnelState(serverID); ok && (state == "failed" || state == "reconnecting") {
if strings.TrimSpace(lastError) == "" {
lastError = state
}
return nil, fmt.Errorf("SSH 隧道未连接:%s", lastError)
}
}
client, ok := s.agentClient(statusCtx, serverID)
if !ok {
return map[string]any{
"configured": false,
"mgmt_enabled": false,
"available": false,
"message": "公网 agent 未配置或 Token 未保存",
}, nil
}
status, err := client.Status(statusCtx)
if err != nil {
if statusErr, ok := err.(*statusError); ok && statusErr.code == 401 {
return map[string]any{
"configured": true,
"mgmt_enabled": true,
"available": false,
"token_mismatch": true,
"message": "Agent Token 不匹配,请点击「升级 Agent」重新同步",
}, nil
}
return nil, err
}
if status == nil {
status = map[string]any{}
}
status["configured"] = true
status["mgmt_enabled"] = true
return status, nil
}
func (s *Service) InstallFRPSOnAgent(ctx context.Context, serverID int64, version, proxy string) (map[string]any, error) {
@@ -1598,7 +1721,7 @@ func (s *Service) RouteHealth(ctx context.Context, serverID int64) ([]RouteHealt
}
item.DNSOK = matched
if !matched {
item.DNSDetail = "未指向服务端地址"
item.DNSDetail = fmt.Sprintf("当前解析 %s;服务端地址 %s", formatIPAddrs(ips), formatIPAddrs(serverIPs))
}
}
}
@@ -1668,6 +1791,17 @@ func (s *Service) RouteHealth(ctx context.Context, serverID int64) ([]RouteHealt
return out, nil
}
func formatIPAddrs(addrs []net.IPAddr) string {
if len(addrs) == 0 {
return "无"
}
values := make([]string, 0, len(addrs))
for _, addr := range addrs {
values = append(values, addr.IP.String())
}
return strings.Join(values, ", ")
}
// tunnelMismatchDetail 区分「frps 一个代理都没有」和「有代理但没绑定当前域名」,
// 避免空列表被拼成 “frps 已注册代理: ;” 这种误导性提示。
func tunnelMismatchDetail(names []string, domain string) string {
@@ -1917,7 +2051,7 @@ func (s *Service) agentClient(ctx context.Context, serverID int64) (*AgentClient
}
baseURL := server.AgentURL
var pins []string
if server.AgentTransport == "tunnel" {
if server.AgentTransport == "tunnel" || (server.AgentTransport == "http" && isLoopbackAgentURL(server.AgentURL)) {
tunnel, err := s.ensureAgentTunnel(ctx, serverID)
if err != nil {
return nil, false
@@ -1962,6 +2096,22 @@ func (s *Service) agentClient(ctx context.Context, serverID int64) (*AgentClient
return client, true
}
func isLoopbackAgentURL(raw string) bool {
parsed, err := url.Parse(strings.TrimSpace(raw))
if err != nil {
return false
}
host := strings.TrimSpace(parsed.Hostname())
if host == "" {
return false
}
if strings.EqualFold(host, "localhost") {
return true
}
ip := net.ParseIP(host)
return ip != nil && ip.IsLoopback()
}
func (s *Service) syncAgentRoute(ctx context.Context, proxy Proxy) {
if !proxy.Enabled || (proxy.Type != "http" && proxy.Type != "https") || len(proxy.CustomDomains) == 0 {
return
@@ -2013,6 +2163,16 @@ func (s *Service) Start(ctx context.Context) error {
var failures []string
for _, server := range servers {
manager := s.managerForServer(server.ID)
hasProxies, proxyErr := s.hasEnabledProxies(ctx, server.ID)
if proxyErr != nil {
failures = append(failures, fmt.Sprintf("服务端 %s 读取规则失败:%v", server.Name, proxyErr))
continue
}
if !hasProxies {
_ = manager.Stop(ctx)
s.recordProcessState(ctx, server.ID, manager, "")
continue
}
if err := manager.Verify(ctx, s.serverConfigPath(server.ID)); err != nil {
s.recordProcessState(ctx, server.ID, manager, err.Error())
failures = append(failures, fmt.Sprintf("服务端 %s 校验失败:%v", server.Name, err))
@@ -2085,6 +2245,16 @@ func (s *Service) Reload(ctx context.Context) error {
var failures []string
for _, server := range servers {
manager := s.managerForServer(server.ID)
hasProxies, proxyErr := s.hasEnabledProxies(ctx, server.ID)
if proxyErr != nil {
failures = append(failures, fmt.Sprintf("服务端 %s 读取规则失败:%v", server.Name, proxyErr))
continue
}
if !hasProxies {
_ = manager.Stop(ctx)
s.recordProcessState(ctx, server.ID, manager, "")
continue
}
if err := manager.Verify(ctx, s.serverConfigPath(server.ID)); err != nil {
s.recordProcessState(ctx, server.ID, manager, err.Error())
failures = append(failures, fmt.Sprintf("服务端 %s 校验失败:%v", server.Name, err))
@@ -2223,12 +2393,26 @@ func (s *Service) Status(ctx context.Context) RuntimeStatus {
current.Configured = s.serverConfigured(server)
if proxies, proxyErr := s.store.ListEnabledProxies(ctx, server.ID); proxyErr == nil {
current.EnabledProxies = len(proxies)
if len(proxies) == 0 {
// 没有启用规则的服务端不应启动 frpc,也不应把它的历史退出错误聚合到全局状态。
current.Status = "stopped"
current.PID = 0
current.LastError = ""
}
}
statuses = append(statuses, current)
}
return latestRuntimeStatus(statuses)
}
func (s *Service) hasEnabledProxies(ctx context.Context, serverID int64) (bool, error) {
proxies, err := s.store.ListEnabledProxies(ctx, serverID)
if err != nil {
return false, err
}
return len(proxies) > 0, nil
}
// statusWithVersion 取服务端运行状态,并在版本探测失败时按二进制目录名推导版本:
// frpc 进程未运行或 --version 探测失败时,RuntimeStatus.Version 因带 omitempty 会被省略,
// 前端(FRP 穿透页状态卡、仪表盘)就只能显示「未检测 / 未知」。
+42
View File
@@ -62,3 +62,45 @@ func TestSummarizeServerSamplesUsesLatestStatus(t *testing.T) {
t.Fatalf("unexpected last error: %s", detail.LastError)
}
}
func TestSSHSettingsChanged(t *testing.T) {
base := Server{SSHHost: "vps-a.example.com", SSHPort: 22, SSHUser: "root", SSHAuth: "key"}
if sshSettingsChanged(base, base, false) {
t.Fatal("相同 SSH 配置不应判定为变化")
}
cases := []Server{
{SSHHost: "vps-b.example.com", SSHPort: 22, SSHUser: "root", SSHAuth: "key"},
{SSHHost: "vps-a.example.com", SSHPort: 2222, SSHUser: "root", SSHAuth: "key"},
{SSHHost: "vps-a.example.com", SSHPort: 22, SSHUser: "admin", SSHAuth: "key"},
{SSHHost: "vps-a.example.com", SSHPort: 22, SSHUser: "root", SSHAuth: "password"},
}
for _, updated := range cases {
if !sshSettingsChanged(base, updated, false) {
t.Fatalf("SSH 配置变化未识别: %+v", updated)
}
}
if !sshSettingsChanged(base, base, true) {
t.Fatal("SSH 凭据更新应判定为变化")
}
}
func TestIsLoopbackAgentURL(t *testing.T) {
for _, raw := range []string{
"http://127.0.0.1:17700",
"http://localhost:17700",
"https://[::1]:17700",
} {
if !isLoopbackAgentURL(raw) {
t.Fatalf("%q 应识别为回环地址", raw)
}
}
for _, raw := range []string{
"http://38.95.74.87:7700",
"https://agent.example.com",
"",
} {
if isLoopbackAgentURL(raw) {
t.Fatalf("%q 不应识别为回环地址", raw)
}
}
}
+2 -2
View File
@@ -57,8 +57,8 @@ func (i SSHInstaller) hostKeyCallback() (ssh.HostKeyCallback, error) {
if len(keyErr.Want) == 0 {
return appendKnownHost(path, remote, key)
}
return fmt.Errorf("SSH 主机密钥与首次记录不一致(%s 期望 %s,实际 %s);若确认 VPS 重装过系统,请删除 %s 后重连",
remote.String(), ssh.FingerprintSHA256(keyErr.Want[0].Key), ssh.FingerprintSHA256(key), path)
return fmt.Errorf("SSH 主机密钥与首次记录不一致(%s 期望 %s,实际 %s);若确认 VPS 重装过系统,请重新点击「安装 Agent」并选择「信任并重装」",
remote.String(), ssh.FingerprintSHA256(keyErr.Want[0].Key), ssh.FingerprintSHA256(key))
}
return fmt.Errorf("SSH 主机密钥校验失败: %w", err)
}, nil
+40 -6
View File
@@ -7,9 +7,15 @@ import (
"io"
"net/http"
"strings"
"sync"
"time"
)
const (
legacyProxyQueryTimeout = 8 * time.Second
legacyProxyConcurrency = 4
)
// frps 管理接口(webServer)返回的代理状态结构,仅取本模块用到的字段。
type frpsProxyStatus struct {
Name string `json:"name"`
@@ -118,15 +124,43 @@ func (c *frpsTrafficClient) fetchLegacyProxyByName(ctx context.Context, server S
// fetchLegacyRouteProxies 按候选规则逐条查询旧版 frps,并保留域名到真实代理名的映射。
func (c *frpsTrafficClient) fetchLegacyRouteProxies(ctx context.Context, server Server, password string, names []string) ServerTraffic {
empty := map[string]ProxyTraffic{}
if len(names) == 0 {
return ServerTraffic{ServerID: server.ID, OK: true, Proxies: empty}
}
queryCtx, cancel := context.WithTimeout(ctx, legacyProxyQueryTimeout)
defer cancel()
type queryResult struct {
info legacyProxyInfo
err error
}
results := make(chan queryResult, len(names))
semaphore := make(chan struct{}, legacyProxyConcurrency)
var wait sync.WaitGroup
for _, name := range names {
wait.Add(1)
go func(name string) {
defer wait.Done()
semaphore <- struct{}{}
defer func() { <-semaphore }()
proxyName := name
if user := strings.TrimSpace(server.Options.User); user != "" && !strings.HasPrefix(proxyName, user+".") {
proxyName = user + "." + proxyName
}
info, _, err := c.fetchLegacyProxyByName(queryCtx, server, password, proxyName)
results <- queryResult{info: info, err: err}
}(name)
}
go func() {
wait.Wait()
close(results)
}()
proxies := make(map[string]ProxyTraffic)
proxyNamesByDomain := make(map[string]string)
var lastErr error
for _, name := range names {
proxyName := name
if user := strings.TrimSpace(server.Options.User); user != "" && !strings.HasPrefix(proxyName, user+".") {
proxyName = user + "." + proxyName
}
info, _, err := c.fetchLegacyProxyByName(ctx, server, password, proxyName)
for result := range results {
info, err := result.info, result.err
if err != nil {
lastErr = err
continue
+50
View File
@@ -0,0 +1,50 @@
package frp
import (
"context"
"fmt"
"net"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"testing"
"time"
)
func TestFetchLegacyRouteProxiesUsesBoundedConcurrency(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
time.Sleep(200 * time.Millisecond)
name := strings.TrimPrefix(r.URL.Path, "/api/proxies/")
w.Header().Set("Content-Type", "application/json")
_, _ = fmt.Fprintf(w, `{"name":%q,"todayTrafficIn":1,"todayTrafficOut":2,"curConns":3}`, name)
}))
defer server.Close()
host, portText, err := net.SplitHostPort(strings.TrimPrefix(server.URL, "http://"))
if err != nil {
t.Fatal(err)
}
port, err := strconv.Atoi(portText)
if err != nil {
t.Fatal(err)
}
client := newFRPSTrafficClient()
names := []string{"a", "b", "c", "d", "e", "f", "g", "h"}
started := time.Now()
result := client.fetchLegacyRouteProxies(context.Background(), Server{
ID: 1,
Options: ServerOptions{
DashboardAddr: host,
DashboardPort: port,
},
}, "", names)
elapsed := time.Since(started)
if !result.OK || len(result.Proxies) != len(names) {
t.Fatalf("逐条查询结果不完整: ok=%v proxies=%d", result.OK, len(result.Proxies))
}
if elapsed >= time.Second {
t.Fatalf("并发查询耗时 %v,仍接近串行等待", elapsed)
}
}
+26
View File
@@ -1,6 +1,7 @@
package frp
import (
"encoding/json"
"fmt"
"net"
"net/url"
@@ -280,6 +281,31 @@ func normalizeServerOptions(in ServerOptions) ServerOptions {
return in
}
// ServerOptions 的布尔字段在表单默认值里是 true,但 JSON 缺省时零值是 false。
// 016 迁移给已有服务端补的 config_json 默认值为 "{}",非 UI 路径创建的服务端也可能不带这些键;
// 缺省时若不兜底,生成的 frpc 配置会变成 transport.tcpMux = false(每条新连接都要重新建
// TCP+TLS,首连明显变慢)与 loginFailExit = false(登录失败不再退出,排障信息被吞掉)。
// 因此按「键缺省即取默认值」反序列化;显式写入 false 的配置仍然保留 false。
func (in *ServerOptions) UnmarshalJSON(data []byte) error {
type plain ServerOptions
var decoded plain
if err := json.Unmarshal(data, &decoded); err != nil {
return err
}
var keys map[string]json.RawMessage
if err := json.Unmarshal(data, &keys); err != nil {
return err
}
*in = ServerOptions(decoded)
if _, ok := keys["tcp_mux"]; !ok {
in.TCPMux = true
}
if _, ok := keys["login_fail_exit"]; !ok {
in.LoginFailExit = true
}
return nil
}
func validAuthMethod(value string) bool {
return value == "none" || value == "token" || value == "oidc"
}