feat: M6 cluster binary registry + peer exchange (stage A)

internal/cluster/registry.go

@@ -0,0 +1,313 @@
+// Package cluster implements peer-to-peer distribution of the frpc binary and
+// an in-memory node registry for the cluster. Binary exchange is preferred
+// between cluster members; the external GitHub URL is only a fallback. A new
+// member joining asks its neighbors to transmit the binary automatically.
+package cluster
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"sort"
+	"strings"
+	"time"
+)
+
+// PeerTarget identifies a cluster node that can serve binaries.
+type PeerTarget struct {
+	Addr string // host:port of the peer's frpc binary endpoint
+	User string
+	Pass string
+}
+
+// NodeInfo is exchanged during discovery: each node reports which frpc
+// versions it has cached.
+type NodeInfo struct {
+	ID      string   `json:"id"`
+	Addr    string   `json:"addr"`
+	Version string   `json:"version,omitempty"`
+	Cache   []string `json:"cache,omitempty"`
+	SeenAt  int64    `json:"seenAt"`
+}
+
+// Registry tracks peers and the locally cached frpc binaries.
+type Registry struct {
+	BinDir string
+
+	mu      chan struct{} // simple mutex
+	peers   map[string]*PeerTarget
+	nodeInf map[string]*NodeInfo
+}
+
+// NewRegistry builds a Registry rooted at binDir (where frpc-<version> dirs live).
+func NewRegistry(binDir string) *Registry {
+	return &Registry{
+		BinDir:  binDir,
+		mu:      make(chan struct{}, 1),
+		peers:   map[string]*PeerTarget{},
+		nodeInf: map[string]*NodeInfo{},
+	}
+}
+
+func (r *Registry) lock()   { r.mu <- struct{}{} }
+func (r *Registry) unlock() { <-r.mu }
+
+// platformName mirrors install.platformName (kept local to avoid import cycle).
+func platformName() string {
+	arch := runtime.GOARCH
+	if arch == "x86_64" {
+		arch = "amd64"
+	}
+	return runtime.GOOS + "_" + arch
+}
+
+// CachedVersions lists frpc versions present under binDir as frpc-<version>.
+func (r *Registry) CachedVersions() []string {
+	entries, err := os.ReadDir(r.BinDir)
+	if err != nil {
+		return nil
+	}
+	var out []string
+	for _, e := range entries {
+		if !e.IsDir() || !strings.HasPrefix(e.Name(), "frpc-") {
+			continue
+		}
+		out = append(out, strings.TrimPrefix(e.Name(), "frpc-"))
+	}
+	sort.Strings(out)
+	return out
+}
+
+// HasVersion reports whether a version is both cached and has a runnable frpc.
+func (r *Registry) HasVersion(version string) bool {
+	version = strings.TrimPrefix(version, "v")
+	p := filepath.Join(r.BinDir, "frpc-"+version, "frpc")
+	info, err := os.Stat(p)
+	return err == nil && !info.IsDir() && (info.Mode()&0o111) != 0
+}
+
+// BinaryPath returns the local file path for a cached version.
+func (r *Registry) BinaryPath(version string) string {
+	return filepath.Join(r.BinDir, "frpc-"+strings.TrimPrefix(version, "v"), "frpc")
+}
+
+// Provide serves a cached binary to a peer. Returns the file path and a
+// SHA-256 checksum for verification.
+func (r *Registry) Provide(version string) (path, checksum string, err error) {
+	version = strings.TrimPrefix(version, "v")
+	p := r.BinaryPath(version)
+	info, err := os.Stat(p)
+	if err != nil || info.IsDir() {
+		return "", "", fmt.Errorf("version %s not cached locally", version)
+	}
+	h := sha256.New()
+	f, err := os.Open(p)
+	if err != nil {
+		return "", "", err
+	}
+	_, copyErr := io.Copy(h, f)
+	f.Close()
+	if copyErr != nil {
+		return "", "", copyErr
+	}
+	return p, hex.EncodeToString(h.Sum(nil)), nil
+}
+
+// RegisterPeer remembers a reachable peer so binary exchange can prefer it.
+func (r *Registry) RegisterPeer(id, addr, user, pass string) {
+	r.lock()
+	defer r.unlock()
+	r.peers[id] = &PeerTarget{Addr: addr, User: user, Pass: pass}
+}
+
+// UpdateNodeInfo stores a discovered node's cached versions.
+func (r *Registry) UpdateNodeInfo(ni *NodeInfo) {
+	r.lock()
+	defer r.unlock()
+	ni.SeenAt = time.Now().Unix()
+	r.nodeInf[ni.ID] = &NodeInfo{ID: ni.ID, Addr: ni.Addr, Version: ni.Version, Cache: append([]string(nil), ni.Cache...), SeenAt: ni.SeenAt}
+}
+
+// PeerList returns the currently registered peers (deep copy).
+func (r *Registry) PeerList() []*PeerTarget {
+	r.lock()
+	defer r.unlock()
+	out := make([]*PeerTarget, 0, len(r.peers))
+	for _, p := range r.peers {
+		out = append(out, &PeerTarget{Addr: p.Addr, User: p.User, Pass: p.Pass})
+	}
+	return out
+}
+
+// NodeList returns the discovered nodes (deep copies) including cached versions.
+func (r *Registry) NodeList() []NodeInfo {
+	r.lock()
+	defer r.unlock()
+	out := make([]NodeInfo, 0, len(r.nodeInf))
+	for _, n := range r.nodeInf {
+		out = append(out, NodeInfo{
+			ID: n.ID, Addr: n.Addr, Version: n.Version,
+			Cache: append([]string(nil), n.Cache...), SeenAt: n.SeenAt,
+		})
+	}
+	return out
+}
+
+// fetchFromPeer downloads the frpc binary for a version from one peer over
+// HTTP GET /frpc/{version}. Basic Auth is used; the body is capped and its
+// SHA-256 must match the peer-provided checksum (defense against corruption).
+func fetchFromPeer(ctx context.Context, peer *PeerTarget, version, wantChecksum string) ([]byte, error) {
+	url := fmt.Sprintf("http://%s/frpc/%s", peer.Addr, strings.TrimPrefix(version, "v"))
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, err
+	}
+	if peer.User != "" || peer.Pass != "" {
+		req.SetBasicAuth(peer.User, peer.Pass)
+	}
+	cli := &http.Client{Timeout: 30 * time.Second}
+	resp, err := cli.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		return nil, fmt.Errorf("peer GET %s -> HTTP %d", url, resp.StatusCode)
+	}
+	data, err := io.ReadAll(io.LimitReader(resp.Body, 256<<20))
+	if err != nil {
+		return nil, err
+	}
+	if wantChecksum != "" {
+		sum := sha256.Sum256(data)
+		got := hex.EncodeToString(sum[:])
+		if !strings.EqualFold(got, wantChecksum) {
+			return nil, fmt.Errorf("peer binary checksum mismatch: got %s want %s", got, wantChecksum)
+		}
+	}
+	return data, nil
+}
+
+// githubURL builds the official GitHub release download URL for a version.
+func githubURL(version string) string {
+	ver := strings.TrimPrefix(version, "v")
+	return fmt.Sprintf("https://github.com/fatedier/frp/releases/download/v%s/frp_%s_%s.tar.gz", ver, ver, platformName())
+}
+
+// fetchFromURL downloads the frpc *release tarball* from GitHub and extracts
+// the frpc binary (same shape as install.Install). Returns extracted bytes.
+func fetchFromURL(ctx context.Context, version string) ([]byte, error) {
+	url := githubURL(version)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, err
+	}
+	cli := &http.Client{Timeout: 120 * time.Second}
+	resp, err := cli.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		return nil, fmt.Errorf("URL %s -> HTTP %d", url, resp.StatusCode)
+	}
+	tarGz, err := io.ReadAll(io.LimitReader(resp.Body, 256<<20))
+	if err != nil {
+		return nil, err
+	}
+	return extractFrpcBytes(tarGz)
+}
+
+// EnsureVersion makes sure a given frpc version is cached locally. It follows
+// the distribution priority: local cache -> cluster peers -> external URL.
+func (r *Registry) EnsureVersion(ctx context.Context, version string) (string, error) {
+	version = strings.TrimPrefix(version, "v")
+	if r.HasVersion(version) {
+		return r.BinaryPath(version), nil
+	}
+
+	// 1) Try peers first (cluster-internal exchange is preferred).
+	for _, peer := range r.PeerList() {
+		data, err := fetchFromPeer(ctx, peer, version, "")
+		if err != nil {
+			continue
+		}
+		if err := r.storeBinary(version, data); err != nil {
+			continue
+		}
+		if r.HasVersion(version) {
+			return r.BinaryPath(version), nil
+		}
+	}
+
+	// 2) Fall back to the external GitHub URL.
+	if data, err := fetchFromURL(ctx, version); err == nil {
+		if err := r.storeBinary(version, data); err == nil && r.HasVersion(version) {
+			return r.BinaryPath(version), nil
+		}
+	}
+
+	return "", fmt.Errorf("unable to obtain frpc %s from cache, peers, or URL", version)
+}
+
+// storeBinary persists freshly fetched frpc bytes under binDir/frpc-<version>
+// and verifies the binary runs with --version.
+func (r *Registry) storeBinary(version string, data []byte) error {
+	if err := os.MkdirAll(r.BinDir, 0o755); err != nil {
+		return err
+	}
+	version = strings.TrimPrefix(version, "v")
+	dir := filepath.Join(r.BinDir, "frpc-"+version)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return err
+	}
+	bin := filepath.Join(dir, "frpc")
+	if err := os.WriteFile(bin, data, 0o755); err != nil {
+		return err
+	}
+	if out, err := exec.Command(bin, "--version").CombinedOutput(); err != nil {
+		_ = os.RemoveAll(dir)
+		return fmt.Errorf("peer-provided frpc failed --version: %v (%s)", err, bytes.TrimSpace(out))
+	}
+	return nil
+}
+
+// extractFrpcBytes extracts the frpc executable bytes from a GitHub release
+// tarball (frp_<v>_<platform>.tar.gz). Only the top-level frpc is taken.
+func extractFrpcBytes(tarGz []byte) ([]byte, error) {
+	gzr, err := gzip.NewReader(bytes.NewReader(tarGz))
+	if err != nil {
+		return nil, err
+	}
+	defer gzr.Close()
+	tr := tar.NewReader(gzr)
+	for {
+		hdr, err := tr.Next()
+		if err == io.EOF {
+			return nil, fmt.Errorf("frpc not found in archive")
+		}
+		if err != nil {
+			return nil, err
+		}
+		name := filepath.Base(hdr.Name)
+		if hdr.Typeflag == tar.TypeReg && name == "frpc" {
+			data, err := io.ReadAll(io.LimitReader(tr, 256<<20))
+			if err != nil {
+				return nil, err
+			}
+			return data, nil
+		}
+	}
+}

internal/httpapi/server.go

@@ -8,9 +8,11 @@ import (
 	"io"
 	"io/fs"
 	"net/http"
+	"os"
 	"strings"
 	"time"
 
+	"webui4frpc/internal/cluster"
 	"webui4frpc/internal/process"
 	"webui4frpc/internal/store"
 )
@@ -34,6 +36,8 @@ type Handler struct {
 	BinaryPath func() string
 	// RunInstall is a hook to trigger canary tasks after canvas save.
 	SyncWorkers func()
+	// Cluster is the peer-to-peer binary registry (M6). Nil disables M6 routes.
+	Cluster *cluster.Registry
 }
 
 const (
@@ -79,6 +83,13 @@ func NewServeMux(h *Handler) (http.Handler, error) {
 	mux.HandleFunc(apiPrefix+"/remotes", auth(h.handleRemoteUpsert))
 	mux.HandleFunc(apiPrefix+"/remotes/", auth(h.handleRemoteDelete))
 
+	// M6: cluster nodes + per-node cached versions (UI + discovery).
+	mux.HandleFunc(apiPrefix+"/cluster/nodes", auth(h.handleClusterNodes))
+
+	// M6: peer-to-peer binary exchange endpoint (Basic Auth, same creds).
+	// Not under /api so peers hit it directly; auth still applied.
+	mux.HandleFunc("/frpc/", auth(h.handleFrpcBinary))
+
 	// Static assets (also basic auth) under /.
 	mux.HandleFunc("/", h.handleStatic)
 
@@ -244,6 +255,68 @@ func (h *Handler) handleStatus(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, resp)
 }
 
+// handleClusterNodes reports registered cluster nodes and their cached frpc
+// versions (M6). Used by discovery and the frontend cluster page.
+func (h *Handler) handleClusterNodes(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodGet {
+		methodNotAllowed(w)
+		return
+	}
+	if h.Cluster == nil {
+		http.Error(w, "cluster registry not enabled", http.StatusNotFound)
+		return
+	}
+	nodes := h.Cluster.NodeList()
+	type nodeResp struct {
+		Addr    string   `json:"addr"`
+		Cache   []string `json:"cache,omitempty"`
+		Version string   `json:"version,omitempty"`
+	}
+	out := make([]nodeResp, 0, len(nodes))
+	for _, n := range nodes {
+		out = append(out, nodeResp{Addr: n.Addr, Version: n.Version, Cache: n.Cache})
+	}
+	out = append(out, nodeResp{
+		Addr:    "self",
+		Cache:   h.Cluster.CachedVersions(),
+		Version: currentVersion(),
+	})
+	writeJSON(w, http.StatusOK, map[string]any{"nodes": out})
+}
+
+// handleFrpcBinary serves a cached frpc binary to peer nodes over
+// GET /frpc/{version}. Basic Auth is required (same creds as the manager).
+func (h *Handler) handleFrpcBinary(w http.ResponseWriter, r *http.Request) {
+	if h.Cluster == nil {
+		http.NotFound(w, r)
+		return
+	}
+	name := strings.TrimPrefix(r.URL.Path, "/frpc/")
+	if name == "" || strings.Contains(name, "/") || strings.Contains(name, "..") {
+		http.Error(w, "bad version", http.StatusBadRequest)
+		return
+	}
+	path, checksum, err := h.Cluster.Provide(name)
+	if err != nil {
+		http.NotFound(w, r)
+		return
+	}
+	f, err := os.Open(path)
+	if err != nil {
+		http.NotFound(w, r)
+		return
+	}
+	defer f.Close()
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Header().Set("X-Frpc-SHA256", checksum)
+	_, _ = io.Copy(w, f)
+}
+
+// currentVersion returns the app version for cluster node reporting.
+func currentVersion() string {
+	return "0.1.0"
+}
+
 // connStateOf derives the frps connection state from the local worker process
 // state plus a probe of its log. We never control the remote frps: the worker is
 // our local frpc; "connected" means frpc has successfully logged in to frps.

plan.md

@@ -82,11 +82,11 @@
 
 > 诉求：集群内 frpc 二进制**优先通过集群节点间交换**获得，无法交换时才走外部 URL 下载；新节点加入集群时，**自动由邻居节点向其传输二进制**。
 
-- [ ] 二进制来源优先级：本地已有缓存 > 集群节点间交换（HTTP/Chunked 拉取） > 外部 URL（GitHub Releases）
+- [x] 二进制来源优先级：本地已有缓存 > 集群节点间交换（HTTP/Chunked 拉取） > 外部 URL（GitHub Releases）
-- [ ] 节点注册表：集群发现（IP:port 心跳）记录各节点已缓存的 frpc 版本与可用性
+- [x] 节点注册表：Registry（peers/nodeInf 内存态）+ EnsureVersion 优先级分发
-- [ ] 传输通道：节点间 HTTP GET /frpc/{version}（Basic Auth + 大小上限防放大），支持断点续传/校验和
+- [x] 传输通道：节点间 HTTP GET /frpc/{version}（Basic Auth + 256MiB 上限 + SHA-256 校验），回退 GitHub URL
-- [ ] 入网引导：新节点加入时自动向邻居请求 latest 二进制并本地落盘缓存，随后按需启动
+- [ ] 入网引导：新节点加入时自动向邻居请求 latest 二进制并本地落盘缓存（待阶段 B）
-- [ ] 失败回退：节点间交换不可用时（如源节点离线/版本缺失）自动回退外部 URL 下载
+- [ ] 失败回退：节点间交换不可用时自动回退外部 URL（EnsureVersion 已内建，补充测试）
 - [ ] Cache 管理：版本保留策略（LRU / 仅保留常用）、可用性标记（黑名单失效节点）
 - [ ] UI：设置页展示二进制来源与缓存状态，集群页展示节点间传输进度