fix(web,agent): P1 正确性与鉴权加固,P2 抽 fsutil 并补测试(v1.2.8)
This commit is contained in:
@@ -0,0 +1,65 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestValidateRoute(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
req routeRequest
|
||||
ok bool
|
||||
}{
|
||||
{"正常", routeRequest{Domain: "a.example.com", Upstream: "127.0.0.1:8080"}, true},
|
||||
{"非法域名", routeRequest{Domain: "bad domain", Upstream: "127.0.0.1:8080"}, false},
|
||||
{"空上游", routeRequest{Domain: "a.example.com"}, false},
|
||||
{"上游带分号", routeRequest{Domain: "a.example.com", Upstream: "127.0.0.1:80; rm -rf /"}, false},
|
||||
{"相对证书目录", routeRequest{Domain: "a.example.com", Upstream: "127.0.0.1:80", TLS: true, CertDir: "certs"}, false},
|
||||
}
|
||||
for _, c := range cases {
|
||||
err := validateRoute(c.req)
|
||||
if c.ok && err != nil {
|
||||
t.Errorf("%s:期望通过,实际 %v", c.name, err)
|
||||
}
|
||||
if !c.ok && err == nil {
|
||||
t.Errorf("%s:期望报错", c.name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestNginxRouteRendersUpstream:vhost 模板的关键行(标记、server_name、proxy_pass)。
|
||||
// 不构造 Basic Auth 场景——那条路径会往 /var/lib/fonu-agent 写 htpasswd,不适合在单测里触发。
|
||||
func TestNginxRouteRendersUpstream(t *testing.T) {
|
||||
out := nginxRoute(routeRequest{Domain: "a.example.com", Upstream: "127.0.0.1:8080"}, "")
|
||||
if !strings.HasPrefix(out, "# Managed by fonu-agent.") {
|
||||
t.Errorf("缺少托管标记,其它工具无法识别该文件归属:\n%s", out)
|
||||
}
|
||||
if !strings.Contains(out, "server_name a.example.com;") {
|
||||
t.Errorf("缺少 server_name:\n%s", out)
|
||||
}
|
||||
if !strings.Contains(out, "proxy_pass http://127.0.0.1:8080;") {
|
||||
t.Errorf("缺少 proxy_pass 回源:\n%s", out)
|
||||
}
|
||||
if strings.Contains(out, "listen 443") {
|
||||
t.Errorf("未启用 TLS 时不应监听 443:\n%s", out)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFileSHA256 覆盖 frps 安装包的校验路径。
|
||||
func TestFileSHA256(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "payload.txt")
|
||||
if err := os.WriteFile(path, []byte("hello"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, err := fileSHA256(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
const want = "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
|
||||
if got != want {
|
||||
t.Fatalf("sha256 不符:got %s", got)
|
||||
}
|
||||
}
|
||||
+25
-31
@@ -25,6 +25,7 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/fonu/fonu/internal/fsutil"
|
||||
"github.com/fonu/fonu/internal/sysinfo"
|
||||
)
|
||||
|
||||
@@ -38,7 +39,8 @@ import (
|
||||
// 0.11.0:新增 /api/v1/certs 证书清单与 /api/v1/nginx/reload 显式重载(回传校验/重载失败原因)。
|
||||
// 0.12.0:frps 安装包的 proxy 参数改白名单 + 下载后校验 SHA-256;Basic Auth 自动生成的密码不再写进
|
||||
// nginx 配置注释(改为只在 PUT 响应里回一次)。
|
||||
const AgentVersion = "0.12.0"
|
||||
// 0.12.1:frps 启停失败如实回报(不再无条件写「已启动」);HTTP 服务补 ReadTimeout/IdleTimeout。
|
||||
const AgentVersion = "0.12.1"
|
||||
|
||||
// Config 是 fonu-agent 的运行配置。agent 只在 VPS 本地管理自己的 Nginx 片段、证书和 frps 配置。
|
||||
type Config struct {
|
||||
@@ -181,7 +183,7 @@ func (s *Server) Handler() http.Handler {
|
||||
}
|
||||
|
||||
func (s *Server) ListenAndServe(ctx context.Context) error {
|
||||
server := &http.Server{Addr: s.cfg.ListenAddr, Handler: s.Handler(), ReadHeaderTimeout: 10 * time.Second}
|
||||
server := &http.Server{Addr: s.cfg.ListenAddr, Handler: s.Handler(), ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 60 * time.Second, IdleTimeout: 2 * time.Minute}
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
@@ -341,18 +343,27 @@ func (s *Server) installFRPS(w http.ResponseWriter, r *http.Request) {
|
||||
_ = exec.Command("systemctl", "daemon-reload").Run()
|
||||
}
|
||||
// nohup 手工跑的 frps 让位给 systemd;已在 systemd 下运行则重启以加载新二进制
|
||||
if exec.Command("systemctl", "is-active", "--quiet", "frps").Run() != nil {
|
||||
if exec.Command("pgrep", "-x", "frps").Run() == nil {
|
||||
_ = exec.Command("pkill", "-x", "frps").Run()
|
||||
time.Sleep(time.Second)
|
||||
notes = append(notes, "已停止 nohup 方式运行的旧 frps 进程")
|
||||
}
|
||||
_ = exec.Command("systemctl", "enable", "--now", "frps").Run()
|
||||
notes = append(notes, "frps.service 已启动")
|
||||
} else {
|
||||
_ = exec.Command("systemctl", "restart", "frps").Run()
|
||||
notes = append(notes, "frps.service 已重启以加载新版本")
|
||||
active := exec.Command("systemctl", "is-active", "--quiet", "frps").Run() == nil
|
||||
if !active && exec.Command("pgrep", "-x", "frps").Run() == nil {
|
||||
_ = exec.Command("pkill", "-x", "frps").Run()
|
||||
time.Sleep(time.Second)
|
||||
notes = append(notes, "已停止 nohup 方式运行的旧 frps 进程")
|
||||
}
|
||||
args := []string{"restart", "frps"}
|
||||
label := "已重启以加载新版本"
|
||||
if !active {
|
||||
args = []string{"enable", "--now", "frps"}
|
||||
label = "已启动"
|
||||
}
|
||||
// 启动失败必须如实回报:之前忽略 err 后无条件写「已启动」,运维会以为服务在跑
|
||||
if out, err := exec.Command("systemctl", args...).CombinedOutput(); err != nil {
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"ok": false, "version": version, "binary": installPath, "notes": notes,
|
||||
"error": fmt.Sprintf("frps 二进制已安装,但 systemctl %s 失败:%v:%s", strings.Join(args, " "), err, tailText(string(out), 400)),
|
||||
})
|
||||
return
|
||||
}
|
||||
notes = append(notes, "frps.service "+label)
|
||||
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "version": version, "binary": installPath, "notes": notes})
|
||||
}
|
||||
|
||||
@@ -1511,24 +1522,7 @@ func atomicWrite(path string, data []byte, mode os.FileMode) error {
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
|
||||
return err
|
||||
}
|
||||
tmp, err := os.CreateTemp(filepath.Dir(path), ".fonu-agent-*")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmpPath := tmp.Name()
|
||||
defer os.Remove(tmpPath)
|
||||
if err := tmp.Chmod(mode); err != nil {
|
||||
tmp.Close()
|
||||
return err
|
||||
}
|
||||
if _, err := tmp.Write(data); err != nil {
|
||||
tmp.Close()
|
||||
return err
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmpPath, path)
|
||||
return fsutil.WriteFileAtomic(path, data, mode)
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, value any) {
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/fonu/fonu/internal/auth"
|
||||
@@ -15,10 +16,67 @@ import (
|
||||
type AuthHandler struct {
|
||||
auth *auth.Service
|
||||
notify *notify.Service
|
||||
guard *loginGuard
|
||||
}
|
||||
|
||||
func NewAuthHandler(authSvc *auth.Service, notifySvc *notify.Service) *AuthHandler {
|
||||
return &AuthHandler{auth: authSvc, notify: notifySvc}
|
||||
return &AuthHandler{auth: authSvc, notify: notifySvc, guard: newLoginGuard()}
|
||||
}
|
||||
|
||||
const (
|
||||
loginFailureLimit = 5
|
||||
loginLockoutWindow = 5 * time.Minute
|
||||
)
|
||||
|
||||
// loginGuard 是进程内的登录失败退避:同一 IP 在窗口内失败到阈值后直接拒绝,
|
||||
// 避免明文 HTTP 部署下被无限爆破;与用于通知的失败统计相互独立。
|
||||
type loginGuard struct {
|
||||
mu sync.Mutex
|
||||
failures map[string][]time.Time
|
||||
}
|
||||
|
||||
func newLoginGuard() *loginGuard {
|
||||
return &loginGuard{failures: make(map[string][]time.Time)}
|
||||
}
|
||||
|
||||
// blocked 返回该 IP 是否仍在锁定窗口内(顺带清理过期记录)。
|
||||
func (g *loginGuard) blocked(ip string) bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
kept := g.recentLocked(ip)
|
||||
g.failures[ip] = kept
|
||||
return len(kept) >= loginFailureLimit
|
||||
}
|
||||
|
||||
func (g *loginGuard) fail(ip string) {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if len(g.failures) > 1024 {
|
||||
for key := range g.failures {
|
||||
if len(g.recentLocked(key)) == 0 {
|
||||
delete(g.failures, key)
|
||||
}
|
||||
}
|
||||
}
|
||||
g.failures[ip] = append(g.recentLocked(ip), time.Now())
|
||||
}
|
||||
|
||||
func (g *loginGuard) reset(ip string) {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
delete(g.failures, ip)
|
||||
}
|
||||
|
||||
// recentLocked 返回窗口内的失败时间(调用方需持有锁)。
|
||||
func (g *loginGuard) recentLocked(ip string) []time.Time {
|
||||
now := time.Now()
|
||||
kept := make([]time.Time, 0, len(g.failures[ip]))
|
||||
for _, at := range g.failures[ip] {
|
||||
if now.Sub(at) < loginLockoutWindow {
|
||||
kept = append(kept, at)
|
||||
}
|
||||
}
|
||||
return kept
|
||||
}
|
||||
|
||||
type loginRequest struct {
|
||||
@@ -27,7 +85,7 @@ type loginRequest struct {
|
||||
}
|
||||
|
||||
type authStatusResponse struct {
|
||||
Initialized bool `json:"initialized"`
|
||||
Initialized bool `json:"initialized"`
|
||||
Authenticated bool `json:"authenticated"`
|
||||
}
|
||||
|
||||
@@ -55,11 +113,17 @@ func (h *AuthHandler) Login(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(r, w, http.StatusBadRequest, "请求格式无效")
|
||||
return
|
||||
}
|
||||
ip := clientIP(r)
|
||||
if h.guard.blocked(ip) {
|
||||
writeError(r, w, http.StatusTooManyRequests, "登录失败次数过多,请稍后再试")
|
||||
return
|
||||
}
|
||||
sessionID, err := h.auth.Login(r.Context(), req.Username, req.Password)
|
||||
if err != nil {
|
||||
if errors.Is(err, auth.ErrUnauthorized) {
|
||||
h.guard.fail(ip)
|
||||
if h.notify != nil {
|
||||
h.notify.RecordLoginFailure(r.Context(), clientIP(r))
|
||||
h.notify.RecordLoginFailure(r.Context(), ip)
|
||||
}
|
||||
writeError(r, w, http.StatusUnauthorized, "用户名或密码错误")
|
||||
return
|
||||
@@ -67,7 +131,8 @@ func (h *AuthHandler) Login(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(r, w, http.StatusInternalServerError, "登录失败")
|
||||
return
|
||||
}
|
||||
setSessionCookie(w, sessionID)
|
||||
h.guard.reset(ip)
|
||||
setSessionCookie(w, r, sessionID)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"message": "登录成功"})
|
||||
}
|
||||
|
||||
@@ -86,7 +151,7 @@ func (h *AuthHandler) ChangePassword(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(r, w, http.StatusUnauthorized, "未登录或会话已过期")
|
||||
return
|
||||
}
|
||||
if err := h.auth.ChangePassword(r.Context(), adminID, req.OldPassword, req.NewPassword); err != nil {
|
||||
if err := h.auth.ChangePassword(r.Context(), adminID, req.OldPassword, req.NewPassword, sessionID); err != nil {
|
||||
writeError(r, w, http.StatusBadRequest, err.Error())
|
||||
return
|
||||
}
|
||||
@@ -97,21 +162,31 @@ func (h *AuthHandler) Logout(w http.ResponseWriter, r *http.Request) {
|
||||
if sessionID := sessionIDFromContext(r.Context()); sessionID != "" {
|
||||
_ = h.auth.Logout(r.Context(), sessionID)
|
||||
}
|
||||
clearSessionCookie(w)
|
||||
clearSessionCookie(w, r)
|
||||
writeJSON(w, http.StatusOK, map[string]string{"message": "已退出登录"})
|
||||
}
|
||||
|
||||
func setSessionCookie(w http.ResponseWriter, sessionID string) {
|
||||
func setSessionCookie(w http.ResponseWriter, r *http.Request, sessionID string) {
|
||||
http.SetCookie(w, &http.Cookie{
|
||||
Name: "fonu_session",
|
||||
Value: sessionID,
|
||||
Path: "/",
|
||||
HttpOnly: true,
|
||||
Secure: requestIsHTTPS(r),
|
||||
SameSite: http.SameSiteLaxMode,
|
||||
MaxAge: int((7 * 24 * time.Hour).Seconds()),
|
||||
})
|
||||
}
|
||||
|
||||
// requestIsHTTPS:直连 TLS 或经反代(X-Forwarded-Proto)时给会话 cookie 加 Secure;
|
||||
// 默认明文 HTTP 部署下不加,否则浏览器会直接丢弃 cookie 导致登录不了。
|
||||
func requestIsHTTPS(r *http.Request) bool {
|
||||
if r.TLS != nil {
|
||||
return true
|
||||
}
|
||||
return strings.EqualFold(strings.TrimSpace(r.Header.Get("X-Forwarded-Proto")), "https")
|
||||
}
|
||||
|
||||
func clientIP(r *http.Request) string {
|
||||
host, _, err := net.SplitHostPort(r.RemoteAddr)
|
||||
if err != nil {
|
||||
@@ -120,12 +195,14 @@ func clientIP(r *http.Request) string {
|
||||
return host
|
||||
}
|
||||
|
||||
func clearSessionCookie(w http.ResponseWriter) {
|
||||
func clearSessionCookie(w http.ResponseWriter, r *http.Request) {
|
||||
http.SetCookie(w, &http.Cookie{
|
||||
Name: "fonu_session",
|
||||
Value: "",
|
||||
Path: "/",
|
||||
HttpOnly: true,
|
||||
Secure: requestIsHTTPS(r),
|
||||
SameSite: http.SameSiteLaxMode,
|
||||
MaxAge: -1,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -44,7 +44,10 @@ func (h *CertHandler) Apply(w http.ResponseWriter, r *http.Request) {
|
||||
CA string `json:"ca"`
|
||||
Email string `json:"email"`
|
||||
}
|
||||
_ = json.NewDecoder(r.Body).Decode(&req)
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(r, w, http.StatusBadRequest, "请求格式无效")
|
||||
return
|
||||
}
|
||||
domains := req.Domains
|
||||
if len(domains) == 0 && strings.TrimSpace(req.Domain) != "" {
|
||||
domains = []string{req.Domain}
|
||||
@@ -141,7 +144,10 @@ func (h *CertHandler) Renew(w http.ResponseWriter, r *http.Request) {
|
||||
Domain string `json:"domain"`
|
||||
CA string `json:"ca"`
|
||||
}
|
||||
_ = json.NewDecoder(r.Body).Decode(&req)
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(r, w, http.StatusBadRequest, "请求格式无效")
|
||||
return
|
||||
}
|
||||
record, err := h.svc.Renew(r.Context(), req.Domain, req.CA)
|
||||
if err != nil {
|
||||
writeError(r, w, http.StatusBadRequest, err.Error())
|
||||
|
||||
@@ -23,13 +23,13 @@ type ddnsRequest struct {
|
||||
RecordNames []string `json:"record_names"`
|
||||
Domains []string `json:"domains"`
|
||||
IPv4Enabled *bool `json:"ipv4_enabled"`
|
||||
IPv6Enabled *bool `json:"ipv6_enabled"`
|
||||
Enabled *bool `json:"enabled"`
|
||||
CustomIPv4 string `json:"custom_ipv4"`
|
||||
CustomIPv6 string `json:"custom_ipv6"`
|
||||
APIToken string `json:"api_token"`
|
||||
APITokenID string `json:"api_token_id"`
|
||||
APISecret string `json:"api_secret"`
|
||||
IPv6Enabled *bool `json:"ipv6_enabled"`
|
||||
Enabled *bool `json:"enabled"`
|
||||
CustomIPv4 string `json:"custom_ipv4"`
|
||||
CustomIPv6 string `json:"custom_ipv6"`
|
||||
APIToken string `json:"api_token"`
|
||||
APITokenID string `json:"api_token_id"`
|
||||
APISecret string `json:"api_secret"`
|
||||
}
|
||||
|
||||
func (h *DDNSHandler) toInput(req ddnsRequest) ddns.SaveInput {
|
||||
@@ -119,7 +119,10 @@ func (h *DDNSHandler) Test(w http.ResponseWriter, r *http.Request) {
|
||||
APITokenID string `json:"api_token_id"`
|
||||
APISecret string `json:"api_secret"`
|
||||
}
|
||||
_ = json.NewDecoder(r.Body).Decode(&req)
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeError(r, w, http.StatusBadRequest, "请求格式无效")
|
||||
return
|
||||
}
|
||||
if err := h.svc.Test(r.Context(), ddns.TestInput{
|
||||
ConfigID: req.ConfigID,
|
||||
Provider: req.Provider,
|
||||
|
||||
@@ -2,6 +2,8 @@ package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -606,7 +608,11 @@ func (h *FRPMultiHandler) InstallFRPSOnAgent(w http.ResponseWriter, r *http.Requ
|
||||
Version string `json:"version"`
|
||||
Proxy string `json:"proxy"`
|
||||
}
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
// 允许空请求体(走默认版本),但 JSON 语法错误要报 400
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil && !errors.Is(err, io.EOF) {
|
||||
writeError(r, w, http.StatusBadRequest, "请求格式无效")
|
||||
return
|
||||
}
|
||||
if body.Version == "" {
|
||||
body.Version = "0.71.0"
|
||||
}
|
||||
|
||||
@@ -123,9 +123,13 @@ func New(cfg config.Config, staticFS fs.FS, migrationsDir string) (*App, error)
|
||||
Addr: cfg.ListenAddr,
|
||||
Handler: handler,
|
||||
ReadHeaderTimeout: 10 * time.Second,
|
||||
// 读超时放到 5 分钟:备份恢复要上传最大 64MB;不设 WriteTimeout,否则日志 SSE 会被中途切断
|
||||
ReadTimeout: 5 * time.Minute,
|
||||
IdleTimeout: 2 * time.Minute,
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
frpMultiSvc.SetBaseContext(ctx)
|
||||
intervalMinutes, _ := settingsStore.GetInt(ctx, settings.KeyDDNSCheckInterval)
|
||||
if intervalMinutes <= 0 {
|
||||
intervalMinutes = 5
|
||||
|
||||
@@ -118,7 +118,7 @@ func (s *Service) AdminIDForSession(ctx context.Context, sessionID string) (int,
|
||||
return adminID, nil
|
||||
}
|
||||
|
||||
func (s *Service) ChangePassword(ctx context.Context, adminID int, oldPassword, newPassword string) error {
|
||||
func (s *Service) ChangePassword(ctx context.Context, adminID int, oldPassword, newPassword, keepSessionID string) error {
|
||||
if err := validate.Password(newPassword); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -134,7 +134,11 @@ func (s *Service) ChangePassword(ctx context.Context, adminID int, oldPassword,
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = s.db.ExecContext(ctx, `UPDATE admins SET password_hash = ? WHERE id = ?`, string(newHash), adminID)
|
||||
if _, err := s.db.ExecContext(ctx, `UPDATE admins SET password_hash = ? WHERE id = ?`, string(newHash), adminID); err != nil {
|
||||
return err
|
||||
}
|
||||
// 改密后失效该账号的其他会话(保留当前这次,否则用户会被自己踢下线)
|
||||
_, err = s.db.ExecContext(ctx, `DELETE FROM sessions WHERE admin_id = ? AND id <> ?`, adminID, keepSessionID)
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
package backup
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// writeArchive 造一个含指定条目的 tar.gz,用于验证还原侧的路径校验与权限处理。
|
||||
func writeArchive(t *testing.T, entries map[string]string) string {
|
||||
t.Helper()
|
||||
path := filepath.Join(t.TempDir(), "backup.tar.gz")
|
||||
file, err := os.Create(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer file.Close()
|
||||
gz := gzip.NewWriter(file)
|
||||
tw := tar.NewWriter(gz)
|
||||
for name, content := range entries {
|
||||
header := &tar.Header{Name: name, Mode: 0o777, Size: int64(len(content)), Typeflag: tar.TypeReg}
|
||||
if err := tw.WriteHeader(header); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := tw.Write([]byte(content)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := tw.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := gz.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return path
|
||||
}
|
||||
|
||||
// TestRestoreRejectsEscapingEntries 覆盖「同前缀绕过」:旧实现用 strings.HasPrefix 判断,
|
||||
// DataDir=/data 时 ../data-other/x 会通过前缀匹配写到 DataDir 之外。
|
||||
func TestRestoreRejectsEscapingEntries(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
dataDir := filepath.Join(root, "data")
|
||||
if err := os.MkdirAll(dataDir, 0o750); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
archive := writeArchive(t, map[string]string{"../data-other/evil.txt": "x"})
|
||||
if err := New(dataDir).Restore(context.Background(), archive); err == nil {
|
||||
t.Fatal("应当拒绝逃出 DataDir 的条目")
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(root, "data-other", "evil.txt")); err == nil {
|
||||
t.Fatal("逃逸条目被写到了 DataDir 之外")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRestoreRejectsParentEscape(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
dataDir := filepath.Join(root, "data")
|
||||
if err := os.MkdirAll(dataDir, 0o750); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
archive := writeArchive(t, map[string]string{"../../evil.txt": "x"})
|
||||
if err := New(dataDir).Restore(context.Background(), archive); err == nil {
|
||||
t.Fatal("应当拒绝带 .. 的条目")
|
||||
}
|
||||
}
|
||||
|
||||
// TestRestoreAppliesFixedMode:归档里的权限位不照搬(构造的归档可带 0777 / 执行位)。
|
||||
func TestRestoreAppliesFixedMode(t *testing.T) {
|
||||
dataDir := t.TempDir()
|
||||
archive := writeArchive(t, map[string]string{"nginx/nginx.conf": "worker_processes 1;"})
|
||||
if err := New(dataDir).Restore(context.Background(), archive); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
info, err := os.Stat(filepath.Join(dataDir, "nginx", "nginx.conf"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := info.Mode().Perm(); runtime.GOOS == "windows" {
|
||||
// Windows 不保留 Unix 权限位(可写文件统一报 0666):只断言没带上执行位
|
||||
if got&0o111 != 0 {
|
||||
t.Fatalf("还原后的文件不应带执行位,实际 %o", got)
|
||||
}
|
||||
} else if got != restoreFileMode {
|
||||
t.Fatalf("还原后的权限应为 %o,实际 %o", restoreFileMode, got)
|
||||
}
|
||||
}
|
||||
@@ -14,14 +14,15 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/fonu/fonu/internal/config"
|
||||
"github.com/fonu/fonu/internal/fsutil"
|
||||
"github.com/fonu/fonu/internal/nginx"
|
||||
"github.com/fonu/fonu/internal/proxy"
|
||||
"github.com/fonu/fonu/internal/settings"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultV4URL = "https://raw.githubusercontent.com/17mon/china_ip_list/master/china_ip_list.txt"
|
||||
ghfastProxyPrefix = "https://ghfast.top/"
|
||||
defaultV4URL = "https://raw.githubusercontent.com/17mon/china_ip_list/master/china_ip_list.txt"
|
||||
ghfastProxyPrefix = "https://ghfast.top/"
|
||||
)
|
||||
|
||||
// metowolf/iplist 已下线 CN/ipv6.txt;按优先级尝试可用源。
|
||||
@@ -96,7 +97,7 @@ func (s *Service) Update(ctx context.Context) error {
|
||||
}
|
||||
|
||||
tmpPath := s.cfg.ChinaCIDRTempPath()
|
||||
if err := os.WriteFile(tmpPath, []byte(body), 0o644); err != nil {
|
||||
if err := fsutil.WriteFileAtomic(tmpPath, []byte(body), 0o644); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
|
||||
Binary file not shown.
Binary file not shown.
@@ -42,7 +42,7 @@ func (s *Service) buildFRPSServerConfig(ctx context.Context, server Server) (str
|
||||
|
||||
var b strings.Builder
|
||||
b.WriteString("# 由 Fonu 生成:与本服务端的 frpc 配置同源(bindPort/auth 与客户端一致)\n")
|
||||
b.WriteString("# 格式:frp v0.52.0+ 的 TOML 配置(Docker 镜像内置 frp 0.71.0;INI 自 v0.52.0 起已被官方弃用)\n")
|
||||
b.WriteString("# 格式:frp v0.52.0+ 的 TOML 配置(INI 自 v0.52.0 起已被官方弃用)\n")
|
||||
b.WriteString("# 用法:保存为 VPS 上的 frps.toml,执行 ./frps -c frps.toml\n\n")
|
||||
// TOML 规范:顶层标量键必须写在任何 [table] 段之前,否则会被归入上一个表
|
||||
b.WriteString("bindAddr = \"0.0.0.0\"\n")
|
||||
|
||||
+62
-14
@@ -14,6 +14,7 @@ import (
|
||||
|
||||
"github.com/fonu/fonu/internal/certificate"
|
||||
"github.com/fonu/fonu/internal/config"
|
||||
"github.com/fonu/fonu/internal/fsutil"
|
||||
"github.com/fonu/fonu/internal/secret"
|
||||
)
|
||||
|
||||
@@ -39,8 +40,14 @@ type Service struct {
|
||||
|
||||
downloadMu sync.RWMutex
|
||||
downloadTask DownloadTask
|
||||
locationMu sync.Mutex
|
||||
locations map[string]locationCache
|
||||
// baseCtx 是应用级 ctx:后台下载随之取消(之前用 context.Background(),App 关闭后仍在跑)
|
||||
baseCtx context.Context
|
||||
locationMu sync.Mutex
|
||||
locations map[string]locationCache
|
||||
|
||||
// agentClients 缓存每台服务端的 agent HTTP 客户端(含连接池),按 URL+Token 指纹失效
|
||||
agentMu sync.Mutex
|
||||
agentClients map[int64]cachedAgentClient
|
||||
|
||||
// 诊断短 TTL 缓存与并发上限,避免列表页轮询叠加外部探测
|
||||
diagMu sync.Mutex
|
||||
@@ -167,6 +174,13 @@ func (s *Service) StartDownload(version string) (DownloadTask, error) {
|
||||
return s.downloadTask, nil
|
||||
}
|
||||
|
||||
// SetBaseContext 注入应用级 ctx,供后台下载继承;由 app 启动时调用。
|
||||
func (s *Service) SetBaseContext(ctx context.Context) {
|
||||
s.downloadMu.Lock()
|
||||
defer s.downloadMu.Unlock()
|
||||
s.baseCtx = ctx
|
||||
}
|
||||
|
||||
func (s *Service) DownloadStatus() DownloadTask {
|
||||
s.downloadMu.RLock()
|
||||
defer s.downloadMu.RUnlock()
|
||||
@@ -174,7 +188,13 @@ func (s *Service) DownloadStatus() DownloadTask {
|
||||
}
|
||||
|
||||
func (s *Service) runDownload(version string) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
|
||||
s.downloadMu.RLock()
|
||||
base := s.baseCtx
|
||||
s.downloadMu.RUnlock()
|
||||
if base == nil {
|
||||
base = context.Background()
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(base, 10*time.Minute)
|
||||
defer cancel()
|
||||
binary, err := s.downloadRelease(ctx, version, s.updateDownload)
|
||||
if err != nil {
|
||||
@@ -772,10 +792,20 @@ func (s *Service) InstallFRPSOnAgent(ctx context.Context, serverID int64, versio
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 安装后把同源 frps.toml 一起下发,保证 systemd 启动即指向正确配置
|
||||
if server, err := s.store.GetServer(ctx, serverID); err == nil {
|
||||
if content, err := s.buildFRPSServerConfig(ctx, server); err == nil {
|
||||
_ = client.PushFRPSConfig(ctx, content)
|
||||
// 安装后把同源 frps.toml 一起下发,保证 systemd 启动即指向正确配置。
|
||||
// 失败不能吞掉:否则前端只看到「安装成功」,VPS 上跑的仍是旧配置。
|
||||
warning := ""
|
||||
if server, err := s.store.GetServer(ctx, serverID); err != nil {
|
||||
warning = "frps 已安装,但读取服务端配置失败:" + err.Error()
|
||||
} else if content, err := s.buildFRPSServerConfig(ctx, server); err != nil {
|
||||
warning = "frps 已安装,但生成 frps.toml 失败:" + err.Error()
|
||||
} else if err := client.PushFRPSConfig(ctx, content); err != nil {
|
||||
warning = "frps 已安装,但 frps.toml 下发失败:" + err.Error()
|
||||
}
|
||||
if warning != "" {
|
||||
result["config_warning"] = warning
|
||||
if list, ok := result["notes"].([]any); ok {
|
||||
result["notes"] = append(list, warning)
|
||||
}
|
||||
}
|
||||
return result, nil
|
||||
@@ -1259,6 +1289,12 @@ func (s *Service) DeleteProxy(ctx context.Context, id int64) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// cachedAgentClient 是按 URL+Token 指纹缓存的 agent 客户端;指纹用密文而非明文,避免内存里多存一份 Token。
|
||||
type cachedAgentClient struct {
|
||||
fingerprint string
|
||||
client *AgentClient
|
||||
}
|
||||
|
||||
func (s *Service) agentClient(ctx context.Context, serverID int64) (*AgentClient, bool) {
|
||||
server, err := s.store.GetServer(ctx, serverID)
|
||||
if err != nil || !server.MgmtEnabled || server.AgentURL == "" || !server.AgentConfigured {
|
||||
@@ -1272,7 +1308,19 @@ func (s *Service) agentClient(ctx context.Context, serverID int64) (*AgentClient
|
||||
if err != nil || token == "" {
|
||||
return nil, false
|
||||
}
|
||||
return NewAgentClient(server.AgentURL, token), true
|
||||
// 复用同一个 http.Client:批量同步 N 条路由时不再每次新建 Transport 与 TLS 握手
|
||||
fingerprint := server.AgentURL + "|" + encoded
|
||||
s.agentMu.Lock()
|
||||
defer s.agentMu.Unlock()
|
||||
if s.agentClients == nil {
|
||||
s.agentClients = make(map[int64]cachedAgentClient)
|
||||
}
|
||||
if cached, ok := s.agentClients[serverID]; ok && cached.fingerprint == fingerprint {
|
||||
return cached.client, true
|
||||
}
|
||||
client := NewAgentClient(server.AgentURL, token)
|
||||
s.agentClients[serverID] = cachedAgentClient{fingerprint: fingerprint, client: client}
|
||||
return client, true
|
||||
}
|
||||
|
||||
func (s *Service) syncAgentRoute(ctx context.Context, proxy Proxy) {
|
||||
@@ -1494,8 +1542,10 @@ func (s *Service) persistServerConfigForVerify(ctx context.Context, server Serve
|
||||
_ = os.Remove(officialPath + ".verify")
|
||||
return err
|
||||
}
|
||||
if current, readErr := os.ReadFile(officialPath); readErr == nil {
|
||||
_ = os.WriteFile(officialPath+".bak", current, 0o600)
|
||||
// 备份失败就不能覆盖唯一副本:否则旧配置没有回退、只剩 .verify 里的新内容
|
||||
if err := fsutil.Backup(officialPath); err != nil {
|
||||
_ = os.Remove(officialPath + ".verify")
|
||||
return fmt.Errorf("备份现有 frps 配置失败,已保留原配置: %w", err)
|
||||
}
|
||||
return os.Rename(officialPath+".verify", officialPath)
|
||||
}
|
||||
@@ -1820,10 +1870,8 @@ func writeConfigAtomically(path, content string) error {
|
||||
}
|
||||
// 23.5:写入前把当前版本留存为 .bak。临时文件写入本身已原子(CreateTemp+Rename),
|
||||
// 但 frpc verify 失败时上层需要回退目标,这里提供版本化备份。
|
||||
if current, err := os.ReadFile(path); err == nil {
|
||||
if err := os.WriteFile(path+".bak", current, 0o600); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := fsutil.Backup(path); err != nil {
|
||||
return err
|
||||
}
|
||||
tmp, err := os.CreateTemp(dir, "frpc-*.toml")
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package frp
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestShellPathEscapesSingleQuotes:uploadBytes 把路径拼进远程 shell 命令,
|
||||
// 单引号必须转义,否则路径里的引号能截断命令。
|
||||
func TestShellPathEscapesSingleQuotes(t *testing.T) {
|
||||
cases := []struct{ in, want string }{
|
||||
{"/tmp/fonu-agent", "'/tmp/fonu-agent'"},
|
||||
{"/tmp/it's here", `'/tmp/it'\''s here'`},
|
||||
{"", "''"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := shellPath(c.in); got != c.want {
|
||||
t.Errorf("shellPath(%q) = %s,期望 %s", c.in, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestIsELFBinary:内嵌二进制在开发构建里是占位文件,必须识别出来并提前报错,
|
||||
// 否则 VPS 上 systemd 会反复 Exec format error。
|
||||
func TestIsELFBinary(t *testing.T) {
|
||||
if isELFBinary([]byte("not an elf")) {
|
||||
t.Error("普通文本不应被判为 ELF")
|
||||
}
|
||||
if isELFBinary(nil) {
|
||||
t.Error("空内容不应被判为 ELF")
|
||||
}
|
||||
if !isELFBinary([]byte{0x7f, 'E', 'L', 'F', 2, 1, 1, 0}) {
|
||||
t.Error("ELF 魔数应被识别")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
// Package fsutil 收敛「原子写配置」与「覆盖前备份」两个到处重复的动作。
|
||||
//
|
||||
// 之前 agent、frp、nginx、chinacidr 各自实现了一遍:有的用固定 .tmp 名(并发写同一路径会互相踩),
|
||||
// 有的直接 os.WriteFile(读到写了一半的配置),有的写完不 chmod(权限跟着 umask 走)。
|
||||
package fsutil
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// WriteFileAtomic 先写同目录的临时文件再 rename:读者要么看到旧内容、要么看到新内容。
|
||||
// 临时文件名带随机后缀,避免并发写同一路径时互相覆盖。
|
||||
func WriteFileAtomic(path string, data []byte, mode os.FileMode) error {
|
||||
dir := filepath.Dir(path)
|
||||
tmp, err := os.CreateTemp(dir, filepath.Base(path)+".tmp-*")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmpName := tmp.Name()
|
||||
defer func() {
|
||||
if tmpName != "" {
|
||||
_ = os.Remove(tmpName)
|
||||
}
|
||||
}()
|
||||
if _, err := tmp.Write(data); err != nil {
|
||||
_ = tmp.Close()
|
||||
return err
|
||||
}
|
||||
if err := tmp.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
// CreateTemp 建出来的是 0600,这里显式对齐调用方要求的权限
|
||||
if err := os.Chmod(tmpName, mode); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.Rename(tmpName, path); err != nil {
|
||||
return err
|
||||
}
|
||||
tmpName = "" // rename 成功,不用再清理
|
||||
return nil
|
||||
}
|
||||
|
||||
// Backup 把现有文件复制成 <path>.bak;文件不存在视为无需备份。
|
||||
// 返回的错误必须由调用方处理——备份是「覆盖前留退路」的唯一保障。
|
||||
func Backup(path string) error {
|
||||
data, err := os.ReadFile(path)
|
||||
if os.IsNotExist(err) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return WriteFileAtomic(path+".bak", data, 0o600)
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
package fsutil
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestWriteFileAtomicReplacesContent(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "frps.toml")
|
||||
if err := os.WriteFile(path, []byte("old"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := WriteFileAtomic(path, []byte("new"), 0o640); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(data) != "new" {
|
||||
t.Fatalf("内容未替换:%q", data)
|
||||
}
|
||||
}
|
||||
|
||||
// TestWriteFileAtomicLeavesNoTempFiles:临时文件必须被 rename 或清理,不能在配置目录里留垃圾。
|
||||
func TestWriteFileAtomicLeavesNoTempFiles(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "nginx.conf")
|
||||
if err := WriteFileAtomic(path, []byte("worker_processes 1;"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, entry := range entries {
|
||||
if strings.Contains(entry.Name(), ".tmp-") {
|
||||
t.Fatalf("残留临时文件:%s", entry.Name())
|
||||
}
|
||||
}
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("目录里应只剩目标文件,实际 %d 个", len(entries))
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackupSkipsMissingFile(t *testing.T) {
|
||||
if err := Backup(filepath.Join(t.TempDir(), "nope.conf")); err != nil {
|
||||
t.Fatalf("文件不存在时应视为无需备份,实际 %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBackupCopiesContent(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "frpc.toml")
|
||||
if err := os.WriteFile(path, []byte("token = \"x\""), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := Backup(path); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := os.ReadFile(path + ".bak")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(data) != "token = \"x\"" {
|
||||
t.Fatalf("备份内容不符:%q", data)
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/fonu/fonu/internal/config"
|
||||
"github.com/fonu/fonu/internal/fsutil"
|
||||
)
|
||||
|
||||
const maxCustomBackups = 10
|
||||
@@ -141,21 +142,21 @@ func RestoreBackup(backupDir, backupName, targetPath string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(targetPath, data, 0o644)
|
||||
return fsutil.WriteFileAtomic(targetPath, data, 0o644)
|
||||
}
|
||||
|
||||
func WriteGlobalCustom(cfg config.Config, content string) error {
|
||||
if err := EnsureCustomDirs(cfg); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(cfg.NginxGlobalCustomPath(), []byte(content), 0o644)
|
||||
return fsutil.WriteFileAtomic(cfg.NginxGlobalCustomPath(), []byte(content), 0o644)
|
||||
}
|
||||
|
||||
func WriteRuleCustom(cfg config.Config, ruleID int64, content string) error {
|
||||
if err := os.MkdirAll(cfg.NginxRulesDir(), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(cfg.NginxRuleCustomPath(ruleID), []byte(content), 0o644)
|
||||
return fsutil.WriteFileAtomic(cfg.NginxRuleCustomPath(ruleID), []byte(content), 0o644)
|
||||
}
|
||||
|
||||
func RemoveRuleCustom(cfg config.Config, ruleID int64) error {
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/fonu/fonu/internal/config"
|
||||
"github.com/fonu/fonu/internal/fsutil"
|
||||
"github.com/fonu/fonu/internal/proxy"
|
||||
)
|
||||
|
||||
@@ -112,7 +113,7 @@ func (m *Manager) Apply(ctx context.Context, rules []proxy.Rule, certs []CertSou
|
||||
}
|
||||
|
||||
tmpPath := m.cfg.NginxConfigPath() + ".tmp"
|
||||
if err := os.WriteFile(tmpPath, []byte(content), 0o644); err != nil {
|
||||
if err := fsutil.WriteFileAtomic(tmpPath, []byte(content), 0o644); err != nil {
|
||||
return ApplyResult{}, err
|
||||
}
|
||||
|
||||
@@ -130,12 +131,10 @@ func (m *Manager) Apply(ctx context.Context, rules []proxy.Rule, certs []CertSou
|
||||
}
|
||||
|
||||
currentPath := m.cfg.NginxConfigPath()
|
||||
backupPath := currentPath + ".bak"
|
||||
if _, err := os.Stat(currentPath); err == nil {
|
||||
if err := copyFile(currentPath, backupPath); err != nil {
|
||||
_ = os.Remove(tmpPath)
|
||||
return ApplyResult{}, err
|
||||
}
|
||||
// 覆盖前留 .bak(fsutil.Backup 对文件不存在视为无需备份)
|
||||
if err := fsutil.Backup(currentPath); err != nil {
|
||||
_ = os.Remove(tmpPath)
|
||||
return ApplyResult{}, err
|
||||
}
|
||||
|
||||
if err := os.Rename(tmpPath, currentPath); err != nil {
|
||||
|
||||
@@ -2,6 +2,7 @@ package scheduler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -15,8 +16,17 @@ type Scheduler struct {
|
||||
jobs []Job
|
||||
}
|
||||
|
||||
// New 会丢掉间隔非正的任务:time.NewTicker 对非正间隔直接 panic,这类任务应由调用方单独启动。
|
||||
func New(jobs ...Job) *Scheduler {
|
||||
return &Scheduler{jobs: jobs}
|
||||
valid := make([]Job, 0, len(jobs))
|
||||
for _, job := range jobs {
|
||||
if job.Interval <= 0 {
|
||||
log.Printf("scheduler: 任务 %s 间隔为 %v,已跳过", job.Name, job.Interval)
|
||||
continue
|
||||
}
|
||||
valid = append(valid, job)
|
||||
}
|
||||
return &Scheduler{jobs: valid}
|
||||
}
|
||||
|
||||
func (s *Scheduler) Start(ctx context.Context) {
|
||||
@@ -28,13 +38,22 @@ func (s *Scheduler) Start(ctx context.Context) {
|
||||
func (s *Scheduler) runJob(ctx context.Context, job Job) {
|
||||
ticker := time.NewTicker(job.Interval)
|
||||
defer ticker.Stop()
|
||||
job.Run(ctx)
|
||||
// 单个任务 panic 不能带走整个进程:包一层 recover,记录后继续下一轮
|
||||
run := func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("scheduler: 任务 %s panic: %v", job.Name, r)
|
||||
}
|
||||
}()
|
||||
job.Run(ctx)
|
||||
}
|
||||
run()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
job.Run(ctx)
|
||||
run()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestNewSkipsNonPositiveIntervals:time.NewTicker 对非正间隔会 panic,这类任务必须在建表时丢掉。
|
||||
func TestNewSkipsNonPositiveIntervals(t *testing.T) {
|
||||
s := New(
|
||||
Job{Name: "zero", Interval: 0, Run: func(context.Context) {}},
|
||||
Job{Name: "negative", Interval: -time.Second, Run: func(context.Context) {}},
|
||||
Job{Name: "ok", Interval: time.Hour, Run: func(context.Context) {}},
|
||||
)
|
||||
if len(s.jobs) != 1 || s.jobs[0].Name != "ok" {
|
||||
t.Fatalf("非正间隔的任务应被跳过,实际保留 %d 个", len(s.jobs))
|
||||
}
|
||||
}
|
||||
|
||||
// TestRunJobRecoversFromPanic:单个任务 panic 不能带走整个进程,且要能继续下一轮。
|
||||
func TestRunJobRecoversFromPanic(t *testing.T) {
|
||||
var calls int32
|
||||
s := New(Job{Name: "panic", Interval: 5 * time.Millisecond, Run: func(context.Context) {
|
||||
atomic.AddInt32(&calls, 1)
|
||||
panic("boom")
|
||||
}})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
s.Start(ctx)
|
||||
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) && atomic.LoadInt32(&calls) < 3 {
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
if got := atomic.LoadInt32(&calls); got < 3 {
|
||||
t.Fatalf("panic 后应继续后续轮次,实际只执行了 %d 次", got)
|
||||
}
|
||||
}
|
||||
@@ -17,19 +17,19 @@ import (
|
||||
)
|
||||
|
||||
const (
|
||||
rateWindow = 5 * time.Second
|
||||
activeWindow = 65 * time.Second
|
||||
persistInterval = 30 * time.Second
|
||||
rateWindow = 5 * time.Second
|
||||
activeWindow = 65 * time.Second
|
||||
persistInterval = 30 * time.Second
|
||||
maxSamplesPerHost = 256
|
||||
)
|
||||
|
||||
type RuleTraffic struct {
|
||||
RuleID int64 `json:"rule_id"`
|
||||
UploadTotal int64 `json:"upload_total"`
|
||||
DownloadTotal int64 `json:"download_total"`
|
||||
UploadRate float64 `json:"upload_rate"`
|
||||
DownloadRate float64 `json:"download_rate"`
|
||||
Connections int `json:"connections"`
|
||||
RuleID int64 `json:"rule_id"`
|
||||
UploadTotal int64 `json:"upload_total"`
|
||||
DownloadTotal int64 `json:"download_total"`
|
||||
UploadRate float64 `json:"upload_rate"`
|
||||
DownloadRate float64 `json:"download_rate"`
|
||||
Connections int `json:"connections"`
|
||||
}
|
||||
|
||||
type ClientConn struct {
|
||||
@@ -176,6 +176,7 @@ func (c *Collector) ingestLine(line string, trackLive bool) {
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
hitIP := ""
|
||||
c.mu.Lock()
|
||||
st := c.ensureHost(host)
|
||||
st.uploadTotal += up
|
||||
@@ -189,13 +190,15 @@ func (c *Collector) ingestLine(line string, trackLive bool) {
|
||||
}
|
||||
if entry.ClientIP != "" && entry.ClientIP != "-" {
|
||||
st.clients[entry.ClientIP] = now
|
||||
if c.notify != nil {
|
||||
c.notify.RecordAccessHit(context.Background(), entry.ClientIP)
|
||||
}
|
||||
// 锁内只记 IP:RecordAccessHit 会写库并判断通知,放在锁外,避免阻塞读侧
|
||||
hitIP = entry.ClientIP
|
||||
}
|
||||
c.pruneHost(st, now)
|
||||
}
|
||||
c.mu.Unlock()
|
||||
if hitIP != "" && c.notify != nil {
|
||||
c.notify.RecordAccessHit(context.Background(), hitIP)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Collector) ensureHost(host string) *hostState {
|
||||
|
||||
Reference in New Issue
Block a user