AgentServer/apps/mounter/server.go

package main

// 与平台的通信（docs/design/21-knowledge-sources.md §4.5）：Unix socket，一行一个 JSON 请求和响应，每个请求都要带密钥。

import (
	"bufio"
	"context"
	"crypto/rand"
	"crypto/subtle"
	"encoding/hex"
	"encoding/json"
	"errors"
	"log"
	"net"
	"os"
	"time"
)

type request struct {
	Token  string      `json:"token"`
	Op     string      `json:"op"`
	Path   string      `json:"path,omitempty"`
	Mounts []mountSpec `json:"mounts,omitempty"`
	Remote *remoteSpec `json:"remote,omitempty"`
}

type server struct {
	token   string
	mgr     *manager
	version string
}

func newToken() string {
	b := make([]byte, 32)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}

// writeGroupFile 写入文件并设为目录的组（平台以 agentsvc 组身份读取）
func writeGroupFile(path string, data []byte, mode os.FileMode, gid int) error {
	tmp := path + ".tmp"
	if err := os.WriteFile(tmp, data, mode); err != nil {
		return err
	}
	if err := os.Chown(tmp, -1, gid); err != nil {
		os.Remove(tmp)
		return err
	}
	if err := os.Chmod(tmp, mode); err != nil {
		return err
	}
	return os.Rename(tmp, path)
}

func (s *server) handle(req request) map[string]any {
	fail := func(err error) map[string]any {
		var ce *checkError
		if errors.As(err, &ce) {
			return map[string]any{"ok": false, "error": ce.Message, "state": ce.State}
		}
		return map[string]any{"ok": false, "error": err.Error()}
	}
	switch req.Op {
	case "ping":
		return map[string]any{"ok": true, "version": s.version}
	case "status":
		// 顺便重新检查：路径被删除、改名不会引起挂载表变化
		s.mgr.refreshAll()
		s.mgr.checkRemotes()
		return map[string]any{"ok": true, "version": s.version, "mounts": s.mgr.status()}
	case "list":
		res, cerr := browse(req.Path, s.mgr.roots)
		if cerr != nil {
			return fail(cerr)
		}
		return map[string]any{"ok": true, "result": res}
	case "check":
		real, cerr := checkSource(req.Path, s.mgr.roots)
		if cerr != nil {
			return fail(cerr)
		}
		return map[string]any{"ok": true, "result": identify(s.mgr.sys, real)}
	case "test":
		if req.Remote == nil {
			return map[string]any{"ok": false, "error": "缺少远程存储配置"}
		}
		if s.mgr.runner == nil {
			return map[string]any{"ok": false, "error": "挂载服务没有配置 rclone"}
		}
		env, err := buildRemote(*req.Remote, s.mgr.runner.bin)
		if err != nil {
			return map[string]any{"ok": false, "error": err.Error(), "state": "mount_failed"}
		}
		return map[string]any{"ok": true, "result": s.mgr.runner.probe(context.Background(), env, 20*time.Second)}
	case "sync":
		st, err := s.mgr.sync(req.Mounts, true)
		if err != nil {
			return fail(err)
		}
		return map[string]any{"ok": true, "mounts": st}
	default:
		return map[string]any{"ok": false, "error": "未知请求：" + req.Op}
	}
}

func (s *server) serveConn(c net.Conn) {
	defer c.Close()
	r := bufio.NewReader(c)
	enc := json.NewEncoder(c)
	for {
		_ = c.SetReadDeadline(time.Now().Add(60 * time.Second))
		line, err := r.ReadBytes('\n')
		if len(line) > 4<<20 {
			_ = enc.Encode(map[string]any{"ok": false, "error": "请求过大"})
			return
		}
		if len(line) == 0 && err != nil {
			return
		}
		var req request
		if jerr := json.Unmarshal(line, &req); jerr != nil {
			_ = enc.Encode(map[string]any{"ok": false, "error": "请求格式不正确"})
			return
		}
		if subtle.ConstantTimeCompare([]byte(req.Token), []byte(s.token)) != 1 {
			_ = enc.Encode(map[string]any{"ok": false, "error": "密钥不正确"})
			return
		}
		_ = c.SetWriteDeadline(time.Now().Add(30 * time.Second))
		if err := enc.Encode(s.handle(req)); err != nil {
			return
		}
		if err != nil {
			return
		}
	}
}

func (s *server) serve(l net.Listener) {
	for {
		c, err := l.Accept()
		if err != nil {
			if errors.Is(err, net.ErrClosed) {
				return
			}
			log.Printf("accept：%v", err)
			time.Sleep(100 * time.Millisecond)
			continue
		}
		go s.serveConn(c)
	}
}