AgentServer/apps/mounter/remote.go

package main

// 远程存储（docs/design/21-knowledge-sources.md §6）：每个资料库一个 rclone 子进程，只读挂到 mounter/remote/<id>（内部，只有本用户能访问），
// 再由本进程的文件系统（隐藏文件过滤、只读）提供给 knowledge/mnt/<id>。
// 连接参数和凭据通过子进程的环境变量传给 rclone，不放在命令行里（命令行在进程列表中所有人可见）。

import (
	"bytes"
	"context"
	"fmt"
	"log"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"sort"
	"strings"
	"sync"
	"syscall"
	"time"
)

type remoteSpec struct {
	Type    string            `json:"type"`
	Params  map[string]string `json:"params"`
	Secrets map[string]string `json:"secrets"`
}

// allowLocalRemote 仅测试使用：rclone 的 local 后端，不需要网络
var allowLocalRemote = false

var (
	bucketPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,254}$`)
	s3Providers   = map[string]bool{"Minio": true, "AWS": true, "Alibaba": true, "TencentCOS": true, "HuaweiOBS": true, "Qiniu": true, "Ceph": true, "Cloudflare": true, "Other": true}
	webdavVendors = map[string]bool{"other": true, "nextcloud": true, "owncloud": true, "sharepoint": true, "sharepoint-ntlm": true}
)

type rcloneEnv struct {
	env  []string // RCLONE_CONFIG_KB_*
	path string   // kb:bucket/prefix
}

func clean(v string) (string, error) {
	v = strings.TrimSpace(v)
	if strings.ContainsAny(v, "\x00\r\n") {
		return "", fmt.Errorf("参数中不能有换行")
	}
	return v, nil
}

func httpURL(v, name string) (string, error) {
	u, err := url.Parse(v)
	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
		return "", fmt.Errorf("%s必须是 http 或 https 地址", name)
	}
	if u.User != nil {
		return "", fmt.Errorf("%s中不能包含用户名或密码", name)
	}
	return v, nil
}

// buildRemote 按存储类型生成 rclone 的环境变量与远程路径
func buildRemote(r remoteSpec, rclone string) (*rcloneEnv, error) {
	p := map[string]string{}
	for k, v := range r.Params {
		c, err := clean(v)
		if err != nil {
			return nil, err
		}
		p[k] = c
	}
	sec := func(k string) (string, error) {
		v := r.Secrets[k]
		if strings.ContainsAny(v, "\x00\r\n") {
			return "", fmt.Errorf("凭据中不能有换行")
		}
		return v, nil
	}
	out := &rcloneEnv{}
	add := func(k, v string) { out.env = append(out.env, "RCLONE_CONFIG_KB_"+k+"="+v) }
	switch r.Type {
	case "s3":
		provider := p["provider"]
		if provider == "" {
			provider = "Other"
		}
		if !s3Providers[provider] {
			return nil, fmt.Errorf("不支持的 S3 服务商：%s", provider)
		}
		if p["endpoint"] == "" && provider != "AWS" {
			return nil, fmt.Errorf("请填写服务地址（endpoint）")
		}
		if p["endpoint"] != "" {
			if _, err := httpURL(p["endpoint"], "服务地址"); err != nil {
				return nil, err
			}
		}
		if !bucketPattern.MatchString(p["bucket"]) {
			return nil, fmt.Errorf("存储桶名称不正确")
		}
		ak, err := sec("accessKeyId")
		if err != nil {
			return nil, err
		}
		sk, err := sec("secretAccessKey")
		if err != nil {
			return nil, err
		}
		if ak == "" || sk == "" {
			return nil, fmt.Errorf("请填写 Access Key 和 Secret Key")
		}
		add("TYPE", "s3")
		add("PROVIDER", provider)
		add("ENV_AUTH", "false")
		add("ACCESS_KEY_ID", ak)
		add("SECRET_ACCESS_KEY", sk)
		add("NO_CHECK_BUCKET", "true")
		if p["endpoint"] != "" {
			add("ENDPOINT", p["endpoint"])
		}
		if p["region"] != "" {
			add("REGION", p["region"])
		}
		prefix := strings.Trim(p["prefix"], "/")
		if strings.Contains("/"+prefix+"/", "/../") {
			return nil, fmt.Errorf("路径前缀不能包含 ..")
		}
		out.path = "kb:" + p["bucket"]
		if prefix != "" {
			out.path += "/" + prefix
		}
	case "webdav":
		u, err := httpURL(p["url"], "地址")
		if err != nil {
			return nil, err
		}
		vendor := p["vendor"]
		if vendor == "" {
			vendor = "other"
		}
		if !webdavVendors[vendor] {
			return nil, fmt.Errorf("不支持的 WebDAV 服务类型：%s", vendor)
		}
		add("TYPE", "webdav")
		add("URL", u)
		add("VENDOR", vendor)
		if p["user"] != "" {
			add("USER", p["user"])
		}
		pass, err := sec("password")
		if err != nil {
			return nil, err
		}
		if pass != "" {
			ob, err := obscure(rclone, pass)
			if err != nil {
				return nil, err
			}
			add("PASS", ob)
		}
		out.path = "kb:"
	case "local":
		if !allowLocalRemote {
			return nil, fmt.Errorf("不支持的存储类型：local")
		}
		add("TYPE", "local")
		out.path = "kb:" + p["path"]
	default:
		return nil, fmt.Errorf("不支持的存储类型：%s", r.Type)
	}
	return out, nil
}

// obscure rclone 的 WebDAV 密码需要先混淆；明文从标准输入传入，不出现在命令行
func obscure(rclone, pass string) (string, error) {
	cmd := exec.Command(rclone, "obscure", "-")
	cmd.Env = []string{"PATH=/usr/bin:/bin"}
	cmd.Stdin = strings.NewReader(pass)
	out, err := cmd.Output()
	if err != nil {
		return "", fmt.Errorf("处理密码失败：%v", err)
	}
	return strings.TrimSpace(string(out)), nil
}

// classify 根据 rclone 的错误输出判断状态
func classify(stderr string) (state, msg string) {
	low := strings.ToLower(stderr)
	detail := lastError(stderr)
	switch {
	case containsAny(low, "signaturedoesnotmatch", "invalidaccesskeyid", "accessdenied", "access denied", "statuscode: 401", "statuscode: 403", "401 unauthorized", "403 forbidden", "authentication", "unauthorized"):
		return "auth_failed", "认证失败或没有权限：" + detail
	case containsAny(low, "nosuchbucket", "bucket does not exist", "directory not found", "404 not found", "statuscode: 404"):
		return "path_missing", "存储桶或路径不存在：" + detail
	case containsAny(low, "connection refused", "no such host", "i/o timeout", "timeout", "network is unreachable", "tls:", "x509", "connection reset", "unexpected eof"):
		return "unreachable", "无法连接：" + detail
	}
	if detail == "" {
		detail = "未知错误"
	}
	return "unreachable", detail
}

var (
	apiErr = regexp.MustCompile(`api error ([A-Za-z]+): ([^\n]+)`)
	netErr = regexp.MustCompile(`(dial tcp [^\n]*|lookup [^\n]*|tls: [^\n]*|x509: [^\n]*)`)
)

func lastError(s string) string {
	if m := apiErr.FindAllStringSubmatch(s, -1); len(m) > 0 {
		last := m[len(m)-1]
		return strings.TrimSpace(last[1] + "：" + last[2])
	}
	if m := netErr.FindAllString(s, -1); len(m) > 0 {
		return strings.TrimSpace(m[len(m)-1])
	}
	lines := strings.Split(strings.TrimSpace(s), "\n")
	for i := len(lines) - 1; i >= 0; i-- {
		l := lines[i]
		if strings.Contains(l, "ERROR") || strings.Contains(l, "CRITICAL") || strings.Contains(l, "Failed") {
			if j := strings.Index(l, ": "); j >= 0 && strings.Contains(l[:j], "/") {
				l = l[j+2:]
			}
			if len(l) > 300 {
				l = l[:300] + "…"
			}
			return strings.TrimSpace(l)
		}
	}
	return ""
}

func containsAny(s string, subs ...string) bool {
	for _, x := range subs {
		if strings.Contains(s, x) {
			return true
		}
	}
	return false
}

// rcloneRunner 运行 rclone 的公共参数
type rcloneRunner struct {
	bin    string
	config string // 空的配置文件，避免读取部署用户自己的 rclone.conf
	home   string
}

func (r *rcloneRunner) command(ctx context.Context, env *rcloneEnv, args ...string) *exec.Cmd {
	cmd := exec.CommandContext(ctx, r.bin, append(args, "--config", r.config)...)
	cmd.Env = append([]string{"PATH=/usr/bin:/bin", "HOME=" + r.home}, env.env...)
	return cmd
}

type probeResult struct {
	State   string   `json:"state"`
	Message string   `json:"message"`
	Entries []string `json:"entries"`
}

// probe 列出远程根目录（带超时），用于"测试连接"与状态检查
func (r *rcloneRunner) probe(parent context.Context, env *rcloneEnv, timeout time.Duration) probeResult {
	ctx, cancel := context.WithTimeout(parent, timeout)
	defer cancel()
	cmd := r.command(ctx, env, "lsf", env.path, "--max-depth", "1", "--low-level-retries", "1", "--retries", "1", "--contimeout", "5s", "--timeout", "10s", "--log-level", "ERROR")
	var out, errb bytes.Buffer
	cmd.Stdout, cmd.Stderr = &limitWriter{w: &out, n: 64 << 10}, &limitWriter{w: &errb, n: 64 << 10}
	err := cmd.Run()
	if ctx.Err() == context.DeadlineExceeded {
		return probeResult{State: "unreachable", Message: "连接超时"}
	}
	if ctx.Err() != nil {
		return probeResult{State: "pending", Message: "已取消"}
	}
	// rclone lsf 出错时退出码可能为 0，以错误输出为准
	if err != nil || strings.Contains(errb.String(), "ERROR") {
		st, msg := classify(errb.String())
		return probeResult{State: st, Message: msg}
	}
	var entries []string
	for _, l := range strings.Split(strings.TrimSpace(out.String()), "\n") {
		if l != "" && !strings.HasPrefix(l, ".") && len(entries) < 50 {
			entries = append(entries, l)
		}
	}
	return probeResult{State: "ok", Entries: entries}
}

type limitWriter struct {
	w *bytes.Buffer
	n int
}

func (l *limitWriter) Write(p []byte) (int, error) {
	if rest := l.n - l.w.Len(); rest > 0 {
		if len(p) > rest {
			l.w.Write(p[:rest])
		} else {
			l.w.Write(p)
		}
	}
	return len(p), nil
}

// ringLog 保留 rclone 最近的错误输出
type ringLog struct {
	mu  sync.Mutex
	buf []string
}

func (r *ringLog) Write(p []byte) (int, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	for _, l := range strings.Split(strings.TrimRight(string(p), "\n"), "\n") {
		if l == "" {
			continue
		}
		r.buf = append(r.buf, l)
		if len(r.buf) > 50 {
			r.buf = r.buf[len(r.buf)-50:]
		}
	}
	return len(p), nil
}

func (r *ringLog) String() string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return strings.Join(r.buf, "\n")
}

func (r *ringLog) reset() {
	r.mu.Lock()
	r.buf = nil
	r.mu.Unlock()
}

// remoteProc 一个远程存储的 rclone 挂载及其状态
type remoteProc struct {
	id     string
	point  string // 内部挂载点
	cache  string
	runner *rcloneRunner
	sys    system
	env    *rcloneEnv
	key    string // 配置的指纹，变化时重启

	mu        sync.Mutex
	mounted   bool
	state     string
	message   string
	probedAt  time.Time
	ready     chan struct{} // 首次得出状态后关闭
	readyOnce sync.Once
	stop      chan struct{}
	done      chan struct{}
	cmd       *exec.Cmd
	logs      ringLog
	onChange  func()
}

func (p *remoteProc) status() (mounted bool, state, msg string) {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.mounted, p.state, p.message
}

func (p *remoteProc) set(mounted bool, state, msg string) {
	p.mu.Lock()
	changed := p.mounted != mounted || p.state != state || p.message != msg
	p.mounted, p.state, p.message = mounted, state, msg
	p.mu.Unlock()
	if state != "pending" {
		p.readyOnce.Do(func() { close(p.ready) })
	}
	if changed && p.onChange != nil {
		p.onChange()
	}
}

// isMounted 内部挂载点上确实是 rclone 的挂载（而不是下面的空目录）
func (p *remoteProc) isMounted() bool {
	for _, e := range p.sys.mountinfo() {
		if e.Point == p.point && strings.HasPrefix(e.FSType, "fuse.rclone") {
			return true
		}
	}
	return false
}

// run 启动并看护 rclone：退出时按退避间隔重启
func (p *remoteProc) run() {
	defer close(p.done)
	backoff := 5 * time.Second
	for {
		started := time.Now()
		p.runOnce()
		select {
		case <-p.stop:
			return
		default:
		}
		if time.Since(started) > time.Minute {
			backoff = 5 * time.Second
		}
		select {
		case <-p.stop:
			return
		case <-time.After(backoff):
		}
		if backoff < 5*time.Minute {
			backoff *= 2
		}
	}
}

func (p *remoteProc) runOnce() {
	_ = exec.Command("fusermount3", "-uz", p.point).Run()
	if err := os.MkdirAll(p.point, 0o700); err != nil {
		p.set(false, "mount_failed", "创建内部挂载点失败："+err.Error())
		return
	}
	_ = os.MkdirAll(p.cache, 0o700)
	p.logs.reset()
	cmd := p.runner.command(context.Background(), p.env, "mount", p.env.path, p.point,
		"--read-only", "--vfs-cache-mode", "full", "--cache-dir", p.cache,
		"--vfs-cache-max-size", "1G", "--vfs-cache-max-age", "1h", "--dir-cache-time", "1m",
		"--low-level-retries", "3", "--contimeout", "10s", "--timeout", "60s", "--log-level", "NOTICE")
	cmd.Stdout, cmd.Stderr = &p.logs, &p.logs
	cmd.SysProcAttr = &syscall.SysProcAttr{Pdeathsig: syscall.SIGTERM}
	if err := cmd.Start(); err != nil {
		p.set(false, "mount_failed", "启动 rclone 失败："+err.Error())
		return
	}
	p.mu.Lock()
	p.cmd = cmd
	p.mu.Unlock()
	exited := make(chan error, 1)
	go func() { exited <- cmd.Wait() }()

	// 等挂载出现
	deadline := time.After(30 * time.Second)
	tick := time.NewTicker(200 * time.Millisecond)
	defer tick.Stop()
wait:
	for {
		select {
		case err := <-exited:
			st, msg := classify(p.logs.String())
			if msg == "未知错误" || msg == "" {
				msg = fmt.Sprintf("rclone 退出：%v", err)
			}
			log.Printf("%s：rclone 退出：%s", p.id, lastError(p.logs.String()))
			p.set(false, st, msg)
			return
		case <-p.stop:
			p.terminate(exited)
			return
		case <-deadline:
			log.Printf("%s：rclone 30 秒内没有完成挂载", p.id)
			p.set(false, "unreachable", "挂载超时")
			p.terminate(exited)
			return
		case <-tick.C:
			if p.isMounted() {
				break wait
			}
		}
	}
	log.Printf("%s：rclone 已挂载 %s", p.id, p.env.path)
	p.set(true, "pending", "正在连接")
	p.checkNow()

	// 运行期间：未连通时每分钟检查一次，正常时每 5 分钟
	for {
		_, state, _ := p.status()
		every := 5 * time.Minute
		if state != "ok" {
			every = time.Minute
		}
		select {
		case err := <-exited:
			st, msg := "unreachable", fmt.Sprintf("rclone 意外退出（%v），稍后自动重启", err)
			if detail := lastError(p.logs.String()); detail != "" {
				st, msg = classify(p.logs.String())
			}
			log.Printf("%s：rclone 意外退出（%v）：%s", p.id, err, lastError(p.logs.String()))
			p.set(false, st, msg)
			_ = exec.Command("fusermount3", "-uz", p.point).Run()
			return
		case <-p.stop:
			p.terminate(exited)
			return
		case <-time.After(every):
			p.checkNow()
		}
	}
}

// checkNow 列远程根目录，得出连接状态
func (p *remoteProc) checkNow() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		select {
		case <-p.stop:
			cancel()
		case <-ctx.Done():
		}
	}()
	r := p.runner.probe(ctx, p.env, 20*time.Second)
	if r.State == "pending" {
		return
	}
	if r.State == "ok" {
		// 之前的错误已经过去，rclone 之后退出时不再据此判断
		p.logs.reset()
	}
	p.mu.Lock()
	p.probedAt = time.Now()
	mounted := p.mounted
	p.mu.Unlock()
	p.set(mounted, r.State, r.Message)
}

// checkIfStale 状态查询时：超过 30 秒没检查过的，在后台重新检查
func (p *remoteProc) checkIfStale() {
	p.mu.Lock()
	stale := p.mounted && time.Since(p.probedAt) > 30*time.Second
	if stale {
		p.probedAt = time.Now()
	}
	p.mu.Unlock()
	if stale {
		go p.checkNow()
	}
}

func (p *remoteProc) terminate(exited chan error) {
	p.mu.Lock()
	cmd := p.cmd
	p.mu.Unlock()
	if cmd != nil && cmd.Process != nil {
		_ = cmd.Process.Signal(syscall.SIGTERM)
		select {
		case <-exited:
		case <-time.After(5 * time.Second):
			_ = cmd.Process.Kill()
			<-exited
		}
	}
	_ = exec.Command("fusermount3", "-uz", p.point).Run()
}

func newRemoteProc(id, remoteDir, cacheDir string, runner *rcloneRunner, sys system, env *rcloneEnv, key string, onChange func()) *remoteProc {
	return &remoteProc{
		id: id, point: filepath.Join(remoteDir, id), cache: filepath.Join(cacheDir, id), runner: runner, sys: sys, env: env, key: key,
		state: "pending", message: "正在连接", ready: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), onChange: onChange,
	}
}

func (p *remoteProc) start() { go p.run() }

// shutdown 停止 rclone、卸载、删除缓存
func (p *remoteProc) shutdown() {
	close(p.stop)
	<-p.done
	_ = os.Remove(p.point)
	_ = os.RemoveAll(p.cache)
}

// remoteKey 配置指纹（含凭据），用于判断是否需要重启 rclone
func remoteKey(r *remoteSpec) string {
	var b strings.Builder
	b.WriteString(r.Type)
	for _, m := range []map[string]string{r.Params, r.Secrets} {
		keys := make([]string, 0, len(m))
		for k := range m {
			keys = append(keys, k)
		}
		sort.Strings(keys)
		for _, k := range keys {
			fmt.Fprintf(&b, "\x00%s=%s", k, m[k])
		}
		b.WriteString("\x01")
	}
	return b.String()
}