feat(agent): 增加 HTTPS pin 模式并修复隧道端口竞态

This commit is contained in:
2026-09-28 14:13:57 +08:00
parent 1bec91cc5e
commit fac9234e4d
23 changed files with 648 additions and 77 deletions
+31 -26
View File
@@ -220,6 +220,8 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("POST /api/v1/routes/{domain}/versions/{name}/rollback", s.auth(s.rollbackRoute))
mux.HandleFunc("PUT /api/v1/china-cidr", s.auth(s.putChinaCIDR))
mux.HandleFunc("POST /api/v1/nginx/install", s.auth(s.installNginx))
mux.HandleFunc("PUT /api/v1/transport/https", s.auth(s.putHTTPSProxy))
mux.HandleFunc("DELETE /api/v1/transport/https", s.auth(s.deleteHTTPSProxy))
mux.HandleFunc("DELETE /api/v1/routes/{domain}", s.auth(s.deleteRoute))
mux.HandleFunc("PUT /api/v1/frps/config", s.auth(s.putFRPSConfig))
mux.HandleFunc("POST /api/v1/frps/install", s.auth(s.installFRPS))
@@ -278,33 +280,36 @@ func (s *Server) status(w http.ResponseWriter, _ *http.Request) {
ports := map[string]bool{"80": portListening(80), "443": portListening(443), "7000": portListening(7000), "8080": portListening(8080)}
nginxProbe := s.probeNginx()
frpsActive := exec.Command("systemctl", "is-active", "--quiet", "frps").Run() == nil
httpsState, httpsConfigured := readHTTPSProxyState(s.cfg.DataDir)
writeJSON(w, http.StatusOK, map[string]any{
"service": "fonu-agent",
"agent_version": AgentVersion,
"listen_addr": s.cfg.ListenAddr,
"listen_scope": listenScope(s.listenLoopback),
"remote_access_enabled": s.cfg.AllowRemote,
"transport_security": transportSecurity(s.listenLoopback),
"nginx_binary": s.cfg.NginxBinary,
"nginx_conf_dir": s.cfg.NginxConfDir,
"nginx_ok": nginxProbe.OK,
"nginx_missing": nginxProbe.Missing,
"nginx_running": nginxProbe.Running,
"nginx_path": nginxProbe.Path,
"nginx_version": nginxProbe.Version,
"nginx_error": nginxProbe.Error,
"frps_binary": s.cfg.FrpsBinary,
"frps_config_path": s.cfg.FrpsConfigPath,
"ports": ports,
"frps_active": frpsActive,
"frps_vhost_hint": "curl -H 'Host: 你的域名' http://127.0.0.1:8080 可测试 frps vhost 是否通",
"frps_unit_exec": strings.TrimSpace(execCommandOutput("systemctl", "show", "frps", "-p", "ExecStart", "--value")),
"frps_enabled": exec.Command("systemctl", "is-enabled", "--quiet", "frps").Run() == nil,
"frps_version": strings.TrimSpace(frpsVersion(s.cfg.FrpsBinary)),
"frps_log_tail": strings.TrimSpace(journalTailDesc("frps", frpsLogTailLines)),
"dashboard_addr": strings.TrimSpace(dashboardAddr(s.cfg.FrpsConfigPath)),
"routes": s.routeNames(),
"checked_at": time.Now().UTC().Format(time.RFC3339),
"service": "fonu-agent",
"agent_version": AgentVersion,
"listen_addr": s.cfg.ListenAddr,
"listen_scope": listenScope(s.listenLoopback),
"remote_access_enabled": s.cfg.AllowRemote,
"transport_security": transportSecurity(s.listenLoopback),
"https_proxy_configured": httpsConfigured,
"https_proxy_port": httpsState.Port,
"nginx_binary": s.cfg.NginxBinary,
"nginx_conf_dir": s.cfg.NginxConfDir,
"nginx_ok": nginxProbe.OK,
"nginx_missing": nginxProbe.Missing,
"nginx_running": nginxProbe.Running,
"nginx_path": nginxProbe.Path,
"nginx_version": nginxProbe.Version,
"nginx_error": nginxProbe.Error,
"frps_binary": s.cfg.FrpsBinary,
"frps_config_path": s.cfg.FrpsConfigPath,
"ports": ports,
"frps_active": frpsActive,
"frps_vhost_hint": "curl -H 'Host: 你的域名' http://127.0.0.1:8080 可测试 frps vhost 是否通",
"frps_unit_exec": strings.TrimSpace(execCommandOutput("systemctl", "show", "frps", "-p", "ExecStart", "--value")),
"frps_enabled": exec.Command("systemctl", "is-enabled", "--quiet", "frps").Run() == nil,
"frps_version": strings.TrimSpace(frpsVersion(s.cfg.FrpsBinary)),
"frps_log_tail": strings.TrimSpace(journalTailDesc("frps", frpsLogTailLines)),
"dashboard_addr": strings.TrimSpace(dashboardAddr(s.cfg.FrpsConfigPath)),
"routes": s.routeNames(),
"checked_at": time.Now().UTC().Format(time.RFC3339),
})
}
+196
View File
@@ -0,0 +1,196 @@
package agent
import (
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"net/http"
"os"
"path/filepath"
"strings"
"github.com/fonu/fonu/internal/fsutil"
)
const httpsProxyConfigName = "fonuagent-https.conf"
type httpsProxyRequest struct {
CertPEM string `json:"cert_pem"`
KeyPEM string `json:"key_pem"`
Port int `json:"port"`
}
type httpsProxyState struct {
Port int `json:"port"`
}
func (s *Server) putHTTPSProxy(w http.ResponseWriter, r *http.Request) {
var req httpsProxyRequest
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 2<<20)).Decode(&req); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON"})
return
}
if req.Port <= 0 || req.Port > 65535 {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "端口需在 1-65535 之间"})
return
}
if _, err := tls.X509KeyPair([]byte(req.CertPEM), []byte(req.KeyPEM)); err != nil {
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "证书或私钥格式无效"})
return
}
if probe := s.probeNginx(); probe.Missing {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": "VPS 未安装 Nginx,无法启用 HTTPS 直连"})
return
}
s.mu.Lock()
defer s.mu.Unlock()
if err := s.configureHTTPSProxy(req); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "port": req.Port})
}
func (s *Server) deleteHTTPSProxy(w http.ResponseWriter, _ *http.Request) {
s.mu.Lock()
defer s.mu.Unlock()
if err := s.removeHTTPSProxy(); err != nil {
writeJSON(w, http.StatusBadGateway, map[string]string{"error": err.Error()})
return
}
writeJSON(w, http.StatusOK, map[string]any{"ok": true})
}
func (s *Server) configureHTTPSProxy(req httpsProxyRequest) error {
dir, err := s.ensureNginxDir()
if err != nil {
return err
}
tlsDir := "/etc/fonu-agent/tls"
if err := os.MkdirAll(tlsDir, 0o750); err != nil {
return err
}
certPath := filepath.Join(tlsDir, "agent.crt")
keyPath := filepath.Join(tlsDir, "agent.key")
if err := atomicWrite(certPath, []byte(req.CertPEM), 0o640); err != nil {
return err
}
if err := atomicWrite(keyPath, []byte(req.KeyPEM), 0o600); err != nil {
return err
}
path := filepath.Join(dir, httpsProxyConfigName)
previous, readErr := os.ReadFile(path)
if _, err := fsutil.BackupVersioned(path, 10); err != nil {
return err
}
content := renderHTTPSProxy(req.Port, certPath, keyPath)
if err := atomicWrite(path, []byte(content), 0o640); err != nil {
return err
}
if err := s.validateNginx(); err != nil {
s.restoreHTTPSConfig(path, previous, readErr)
return fmt.Errorf("nginx config rejected: %w", err)
}
if err := s.reloadNginx(); err != nil {
s.restoreHTTPSConfig(path, previous, readErr)
return err
}
return writeHTTPSProxyState(s.cfg.DataDir, httpsProxyState{Port: req.Port})
}
func (s *Server) removeHTTPSProxy() error {
dir, err := s.ensureNginxDir()
if err != nil {
return err
}
path := filepath.Join(dir, httpsProxyConfigName)
previous, readErr := os.ReadFile(path)
if errors.Is(readErr, os.ErrNotExist) {
_ = os.Remove(httpsProxyStatePath(s.cfg.DataDir))
return nil
}
if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) {
return err
}
if err := s.validateNginx(); err != nil {
s.restoreHTTPSConfig(path, previous, readErr)
return err
}
if err := s.reloadNginx(); err != nil {
s.restoreHTTPSConfig(path, previous, readErr)
return err
}
_ = os.Remove(httpsProxyStatePath(s.cfg.DataDir))
return nil
}
func (s *Server) restoreHTTPSConfig(path string, previous []byte, readErr error) {
if readErr == nil {
_ = atomicWrite(path, previous, 0o640)
} else {
_ = os.Remove(path)
}
_ = s.reloadNginx()
}
func (s *Server) ensureNginxDir() (string, error) {
dir := strings.TrimSpace(s.cfg.NginxConfDir)
if dir != "" {
return dir, nil
}
dir, err := detectNginxConfDir(s.nginxPath())
if err != nil {
return "", err
}
s.cfg.NginxConfDir = dir
return dir, nil
}
func renderHTTPSProxy(port int, certPath, keyPath string) string {
return fmt.Sprintf(`# Managed by fonu-agent. Do not edit.
server {
listen %d ssl;
server_name _;
ssl_certificate %s;
ssl_certificate_key %s;
ssl_protocols TLSv1.2 TLSv1.3;
location /api/v1/ {
proxy_pass http://127.0.0.1:7700;
proxy_http_version 1.1;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header Authorization $http_authorization;
}
}
`, port, certPath, keyPath)
}
func httpsProxyStatePath(dataDir string) string {
return filepath.Join(dataDir, "https_transport.json")
}
func writeHTTPSProxyState(dataDir string, state httpsProxyState) error {
if strings.TrimSpace(dataDir) == "" {
return nil
}
data, err := json.Marshal(state)
if err != nil {
return err
}
return atomicWrite(httpsProxyStatePath(dataDir), data, 0o640)
}
func readHTTPSProxyState(dataDir string) (httpsProxyState, bool) {
data, err := os.ReadFile(httpsProxyStatePath(dataDir))
if err != nil {
return httpsProxyState{}, false
}
var state httpsProxyState
if err := json.Unmarshal(data, &state); err != nil || state.Port <= 0 {
return httpsProxyState{}, false
}
return state, true
}
+31
View File
@@ -338,6 +338,37 @@ func (h *FRPMultiHandler) AgentTunnelRestart(w http.ResponseWriter, r *http.Requ
writeJSON(w, http.StatusOK, server)
}
func (h *FRPMultiHandler) SetAgentTransport(w http.ResponseWriter, r *http.Request) {
id, err := parseID(r.PathValue("id"))
if err != nil {
writeError(r, w, http.StatusBadRequest, "FRP 服务端 ID 无效")
return
}
var request struct {
Transport string `json:"transport"`
Port int `json:"port"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
writeError(r, w, http.StatusBadRequest, "请求格式无效")
return
}
var server frp.Server
switch strings.TrimSpace(request.Transport) {
case "tunnel":
server, err = h.svc.SwitchAgentToTunnel(r.Context(), id)
case "https-pin":
server, err = h.svc.EnableAgentHTTPS(r.Context(), id, request.Port)
default:
writeError(r, w, http.StatusBadRequest, "不支持的传输方式")
return
}
if err != nil {
writeError(r, w, http.StatusBadGateway, err.Error())
return
}
writeJSON(w, http.StatusOK, server)
}
func (h *FRPMultiHandler) AgentStatus(w http.ResponseWriter, r *http.Request) {
id, err := parseID(r.PathValue("id"))
if err != nil {
+1
View File
@@ -132,6 +132,7 @@ func NewRouter(deps Deps) http.Handler {
protect("POST /api/frpmulti/servers/{id}/agent/ssh-diagnose", frpHandler.AgentSSHDiagnose)
protect("POST /api/frpmulti/servers/{id}/agent/test", frpHandler.AgentTest)
protect("POST /api/frpmulti/servers/{id}/agent/tunnel/restart", frpHandler.AgentTunnelRestart)
protect("POST /api/frpmulti/servers/{id}/agent/transport", frpHandler.SetAgentTransport)
protect("GET /api/frpmulti/servers/{id}/agent/status", frpHandler.AgentStatus)
protect("GET /api/frpmulti/servers/{id}/agent/routes", frpHandler.AgentRoutes)
protect("GET /api/frpmulti/servers/{id}/agent/routes/health", frpHandler.AgentRouteHealth)
+64 -2
View File
@@ -3,6 +3,11 @@ package frp
import (
"bytes"
"context"
"crypto/sha256"
"crypto/subtle"
"crypto/tls"
"crypto/x509"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
@@ -18,16 +23,56 @@ import (
type AgentClient struct {
baseURL string
token string
pin string
http *http.Client
}
func NewAgentClient(baseURL, token string) *AgentClient {
return &AgentClient{
return newAgentClient(baseURL, token, "")
}
func NewAgentClientWithPin(baseURL, token, pin string) *AgentClient {
return newAgentClient(baseURL, token, pin)
}
func newAgentClient(baseURL, token, pin string) *AgentClient {
client := &AgentClient{
baseURL: strings.TrimRight(strings.TrimSpace(baseURL), "/"),
token: token,
pin: strings.ToLower(strings.TrimSpace(pin)),
// 路由写入链路含 nginx -T 自省 + -t 校验 + reload(宝塔带 lua/waf 模块时各步可达数秒),15s 会超时
http: &http.Client{Timeout: 60 * time.Second},
}
if client.pin != "" {
client.http.Transport = &http.Transport{TLSClientConfig: pinTLSConfig(client.pin)}
}
return client
}
func pinTLSConfig(pin string) *tls.Config {
return &tls.Config{
MinVersion: tls.VersionTLS12,
InsecureSkipVerify: true,
VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
if len(rawCerts) == 0 {
return fmt.Errorf("服务器未返回证书")
}
cert, err := x509.ParseCertificate(rawCerts[0])
if err != nil {
return err
}
now := time.Now()
if now.Before(cert.NotBefore) || now.After(cert.NotAfter) {
return fmt.Errorf("Agent HTTPS 证书不在有效期内")
}
sum := sha256.Sum256(rawCerts[0])
got := hex.EncodeToString(sum[:])
if subtle.ConstantTimeCompare([]byte(got), []byte(pin)) != 1 {
return fmt.Errorf("Agent HTTPS 证书指纹不匹配")
}
return nil
},
}
}
func (c *AgentClient) request(ctx context.Context, method, path string, payload any, result any) error {
@@ -55,7 +100,10 @@ func (c *AgentClient) request(ctx context.Context, method, path string, payload
return err
}
defer resp.Body.Close()
data, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
data, err := io.ReadAll(io.LimitReader(resp.Body, 2<<20))
if err != nil {
return err
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
return &statusError{code: resp.StatusCode, body: strings.TrimSpace(string(data))}
}
@@ -71,6 +119,20 @@ func (c *AgentClient) Ping(ctx context.Context) error {
return c.request(ctx, http.MethodGet, "/api/v1/ping", nil, nil)
}
func (c *AgentClient) ConfigureHTTPS(ctx context.Context, certPEM, keyPEM string, port int) error {
var result map[string]any
return c.request(ctx, http.MethodPut, "/api/v1/transport/https", map[string]any{
"cert_pem": certPEM,
"key_pem": keyPEM,
"port": port,
}, &result)
}
func (c *AgentClient) DisableHTTPS(ctx context.Context) error {
var result map[string]any
return c.request(ctx, http.MethodDelete, "/api/v1/transport/https", nil, &result)
}
func (c *AgentClient) Status(ctx context.Context) (map[string]any, error) {
var result map[string]any
err := c.request(ctx, http.MethodGet, "/api/v1/status", nil, &result)
+70
View File
@@ -0,0 +1,70 @@
package frp
import (
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/sha256"
"crypto/x509"
"crypto/x509/pkix"
"encoding/hex"
"encoding/pem"
"fmt"
"math/big"
"net"
"time"
)
type agentCertificate struct {
CertPEM string
KeyPEM string
Pin string
}
func generateAgentCertificate(host string) (agentCertificate, error) {
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
if err != nil {
return agentCertificate{}, err
}
serialLimit := new(big.Int).Lsh(big.NewInt(1), 128)
serial, err := rand.Int(rand.Reader, serialLimit)
if err != nil {
return agentCertificate{}, err
}
now := time.Now()
template := &x509.Certificate{
SerialNumber: serial,
Subject: pkix.Name{CommonName: host},
NotBefore: now.Add(-5 * time.Minute),
NotAfter: now.Add(365 * 24 * time.Hour),
KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
}
if ip := net.ParseIP(host); ip != nil {
template.IPAddresses = []net.IP{ip}
} else {
template.DNSNames = []string{host}
}
der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
if err != nil {
return agentCertificate{}, err
}
keyDER, err := x509.MarshalPKCS8PrivateKey(key)
if err != nil {
return agentCertificate{}, err
}
sum := sha256.Sum256(der)
return agentCertificate{
CertPEM: string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})),
KeyPEM: string(pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER})),
Pin: hex.EncodeToString(sum[:]),
}, nil
}
func validatePin(pin string) error {
decoded, err := hex.DecodeString(pin)
if err != nil || len(decoded) != sha256.Size {
return fmt.Errorf("证书指纹格式无效")
}
return nil
}
+48
View File
@@ -0,0 +1,48 @@
package frp
import (
"crypto/x509"
"encoding/pem"
"net"
"testing"
)
func TestGenerateAgentCertificate(t *testing.T) {
cert, err := generateAgentCertificate("127.0.0.1")
if err != nil {
t.Fatal(err)
}
if err := validatePin(cert.Pin); err != nil {
t.Fatalf("pin 无效: %v", err)
}
block, _ := pem.Decode([]byte(cert.CertPEM))
if block == nil {
t.Fatal("证书 PEM 解码失败")
}
parsed, err := x509.ParseCertificate(block.Bytes)
if err != nil {
t.Fatal(err)
}
if len(parsed.IPAddresses) != 1 || !parsed.IPAddresses[0].Equal(net.ParseIP("127.0.0.1")) {
t.Fatalf("证书 IP SAN 不正确: %#v", parsed.IPAddresses)
}
}
func TestPinTLSConfigVerification(t *testing.T) {
cert, err := generateAgentCertificate("agent.example.com")
if err != nil {
t.Fatal(err)
}
block, _ := pem.Decode([]byte(cert.CertPEM))
if block == nil {
t.Fatal("证书 PEM 解码失败")
}
verify := pinTLSConfig(cert.Pin).VerifyPeerCertificate
if err := verify([][]byte{block.Bytes}, nil); err != nil {
t.Fatalf("正确 pin 校验失败: %v", err)
}
wrong := pinTLSConfig("0000000000000000000000000000000000000000000000000000000000000000").VerifyPeerCertificate
if err := wrong([][]byte{block.Bytes}, nil); err == nil {
t.Fatal("错误 pin 应校验失败")
}
}
+12 -5
View File
@@ -30,6 +30,10 @@ type agentTunnel struct {
}
func newAgentTunnel(parent context.Context, serverID int64, localPort int, installer SSHInstaller) *agentTunnel {
return newAgentTunnelWithListener(parent, serverID, localPort, installer, nil)
}
func newAgentTunnelWithListener(parent context.Context, serverID int64, localPort int, installer SSHInstaller, listener net.Listener) *agentTunnel {
ctx, cancel := context.WithCancel(parent)
return &agentTunnel{
serverID: serverID,
@@ -37,18 +41,21 @@ func newAgentTunnel(parent context.Context, serverID int64, localPort int, insta
ssh: installer,
ctx: ctx,
cancel: cancel,
listener: listener,
state: "connecting",
}
}
func (t *agentTunnel) Start() error {
listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", t.localPort))
if err != nil {
t.setState("failed", err)
return fmt.Errorf("监听隧道本地端口失败: %w", err)
if t.listener == nil {
listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", t.localPort))
if err != nil {
t.setState("failed", err)
return fmt.Errorf("监听隧道本地端口失败: %w", err)
}
t.listener = listener
}
t.mu.Lock()
t.listener = listener
t.state = "connected"
t.lastError = ""
t.mu.Unlock()
Binary file not shown.
Binary file not shown.
+117 -22
View File
@@ -744,15 +744,13 @@ func (s *Service) AgentInstall(ctx context.Context, serverID int64) (Server, err
if err != nil {
return Server{}, err
}
localPort := server.AgentLocalPort
if localPort == 0 {
localPort, err = s.allocateAgentTunnelPort(ctx)
if err != nil {
return Server{}, err
}
localPort, listener, err := s.reserveAgentTunnelPort(ctx, serverID, server.AgentLocalPort)
if err != nil {
return Server{}, err
}
tunnel := newAgentTunnel(s.tunnelContext(ctx), serverID, localPort, installer)
tunnel := newAgentTunnelWithListener(s.tunnelContext(ctx), serverID, localPort, installer, listener)
if err := tunnel.Start(); err != nil {
_ = listener.Close()
return Server{}, err
}
baseURL := fmt.Sprintf("http://127.0.0.1:%d", localPort)
@@ -803,19 +801,24 @@ func (s *Service) tunnelContext(fallback context.Context) context.Context {
return context.Background()
}
func (s *Service) allocateAgentTunnelPort(ctx context.Context) (int, error) {
func (s *Service) reserveAgentTunnelPort(ctx context.Context, ownerID int64, preferred int) (int, net.Listener, error) {
s.tunnelMu.Lock()
defer s.tunnelMu.Unlock()
servers, err := s.store.ListServers(ctx)
if err != nil {
return 0, err
return 0, nil, err
}
used := map[int]bool{}
for _, server := range servers {
if server.AgentLocalPort > 0 {
if server.ID != ownerID && server.AgentLocalPort > 0 {
used[server.AgentLocalPort] = true
}
}
if preferred > 0 && !used[preferred] {
if listener, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", preferred)); err == nil {
return preferred, listener, nil
}
}
for port := 17700; port <= 18700; port++ {
if used[port] {
continue
@@ -824,10 +827,9 @@ func (s *Service) allocateAgentTunnelPort(ctx context.Context) (int, error) {
if err != nil {
continue
}
_ = listener.Close()
return port, nil
return port, listener, nil
}
return 0, fmt.Errorf("SSH 隧道本地端口 17700-18700 已耗尽")
return 0, nil, fmt.Errorf("SSH 隧道本地端口 17700-18700 已耗尽")
}
func (s *Service) ensureAgentTunnel(ctx context.Context, serverID int64) (*agentTunnel, error) {
@@ -842,15 +844,13 @@ func (s *Service) ensureAgentTunnel(ctx context.Context, serverID int64) (*agent
if err != nil {
return nil, err
}
localPort := server.AgentLocalPort
if localPort == 0 {
localPort, err = s.allocateAgentTunnelPort(ctx)
if err != nil {
return nil, err
}
localPort, listener, err := s.reserveAgentTunnelPort(ctx, serverID, server.AgentLocalPort)
if err != nil {
return nil, err
}
tunnel := newAgentTunnel(s.tunnelContext(ctx), serverID, localPort, installer)
tunnel := newAgentTunnelWithListener(s.tunnelContext(ctx), serverID, localPort, installer, listener)
if err := tunnel.Start(); err != nil {
_ = listener.Close()
return nil, err
}
s.tunnelMu.Lock()
@@ -937,6 +937,90 @@ func (s *Service) RestartAgentTunnel(ctx context.Context, serverID int64) (Serve
return server, nil
}
func (s *Service) agentToken(ctx context.Context, serverID int64) (string, error) {
encoded, err := s.store.AgentToken(ctx, serverID)
if err != nil || encoded == "" {
return "", fmt.Errorf("Agent Token 未保存")
}
token, err := s.secretBox.Decrypt(encoded)
if err != nil || token == "" {
return "", fmt.Errorf("Agent Token 无法解密")
}
return token, nil
}
func (s *Service) EnableAgentHTTPS(ctx context.Context, serverID int64, port int) (Server, error) {
if port == 0 {
port = 7443
}
if port <= 0 || port > 65535 {
return Server{}, fmt.Errorf("HTTPS 端口无效")
}
server, err := s.store.GetServer(ctx, serverID)
if err != nil {
return Server{}, err
}
if server.AgentTransport == "http" {
return Server{}, fmt.Errorf("请先通过「升级 Agent」迁移到 SSH 隧道,再切换 HTTPS 直连")
}
currentClient, ok := s.agentClient(ctx, serverID)
if !ok {
return Server{}, fmt.Errorf("当前 Agent 连接不可用")
}
token, err := s.agentToken(ctx, serverID)
if err != nil {
return Server{}, err
}
cert, err := generateAgentCertificate(server.SSHHost)
if err != nil {
return Server{}, err
}
if err := currentClient.ConfigureHTTPS(ctx, cert.CertPEM, cert.KeyPEM, port); err != nil {
return Server{}, err
}
baseURL := fmt.Sprintf("https://%s:%d", server.SSHHost, port)
pinned := NewAgentClientWithPin(baseURL, token, cert.Pin)
if err := pinned.Ping(ctx); err != nil {
_ = currentClient.DisableHTTPS(ctx)
return Server{}, fmt.Errorf("HTTPS 直连验证失败: %w", err)
}
if err := s.store.SetAgentTransport(ctx, serverID, "https-pin", server.AgentLocalPort, cert.Pin, fmt.Sprintf("0.0.0.0:%d", port), "connected", ""); err != nil {
return Server{}, err
}
s.stopAgentTunnel(serverID)
updated, err := s.store.GetServer(ctx, serverID)
if err != nil {
return Server{}, err
}
return updated, nil
}
func (s *Service) SwitchAgentToTunnel(ctx context.Context, serverID int64) (Server, error) {
server, err := s.store.GetServer(ctx, serverID)
if err != nil {
return Server{}, err
}
if server.AgentTransport == "https-pin" {
token, err := s.agentToken(ctx, serverID)
if err != nil {
return Server{}, err
}
client := NewAgentClientWithPin(server.AgentURL, token, server.AgentTLSPin)
if err := client.DisableHTTPS(ctx); err != nil {
return Server{}, fmt.Errorf("关闭 HTTPS 直连失败: %w", err)
}
}
if _, err := s.ensureAgentTunnel(ctx, serverID); err != nil {
return Server{}, err
}
updated, err := s.store.GetServer(ctx, serverID)
if err != nil {
return Server{}, err
}
s.enrichServerRuntime(ctx, &updated)
return updated, nil
}
func (s *Service) AgentUninstall(ctx context.Context, serverID int64, keepData bool) (string, error) {
_, installer, err := s.sshInstaller(ctx, serverID)
if err != nil {
@@ -1515,17 +1599,23 @@ func (s *Service) agentClient(ctx context.Context, serverID int64) (*AgentClient
return nil, false
}
baseURL := server.AgentURL
pin := ""
if server.AgentTransport == "tunnel" {
tunnel, err := s.ensureAgentTunnel(ctx, serverID)
if err != nil {
return nil, false
}
baseURL = fmt.Sprintf("http://127.0.0.1:%d", tunnel.localPort)
} else if server.AgentTransport == "https-pin" {
pin = strings.TrimSpace(server.AgentTLSPin)
if baseURL == "" || pin == "" {
return nil, false
}
} else if baseURL == "" {
return nil, false
}
// 复用同一个 http.Client:批量同步 N 条路由时不再每次新建 Transport 与 TLS 握手
fingerprint := baseURL + "|" + encoded
fingerprint := baseURL + "|" + encoded + "|" + pin
s.agentMu.Lock()
defer s.agentMu.Unlock()
if s.agentClients == nil {
@@ -1534,7 +1624,12 @@ func (s *Service) agentClient(ctx context.Context, serverID int64) (*AgentClient
if cached, ok := s.agentClients[serverID]; ok && cached.fingerprint == fingerprint {
return cached.client, true
}
client := NewAgentClient(baseURL, token)
var client *AgentClient
if pin != "" {
client = NewAgentClientWithPin(baseURL, token, pin)
} else {
client = NewAgentClient(baseURL, token)
}
s.agentClients[serverID] = cachedAgentClient{fingerprint: fingerprint, client: client}
return client, true
}