knowledge-governance/internal/libraryagent/agent.go

// Package libraryagent is the least-privilege execution boundary for
// whole-library capture on the target host.
//
// Why it exists: capturing a library has to stop the source writer for a few
// seconds and read source credentials. Handing the governance containers the
// Docker socket (a full write API) or the source credentials would be a much
// larger privilege than that operation needs. Instead this agent runs as the
// ordinary operator account on the host, listens on one unix socket which only
// that account can connect to, and executes one fixed command with a fixed
// writer declaration. A caller can choose the scope and (optionally) a job id —
// nothing else.
package libraryagent

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"strings"
	"sync"
	"time"
)

// Config is fixed at start-up by the operator (systemd unit / CLI flags). It is
// never taken from a request.
type Config struct {
	Socket      string
	Binary      string
	ArchiveRoot string
	JournalRoot string
	// Writer declaration passed to the binary, exactly as reviewed in
	// scripts/library-capture.sh. Clients cannot influence it.
	SourceApp    string
	SourcePG     string
	SourceES     string
	GovernancePG string
	Project      string
	Writers      string
	NonWriters   string
	MaxDuration  time.Duration
	// Run executes the binary; replaced in tests.
	Run func(ctx context.Context, name string, args ...string) ([]byte, error)
}

type request struct {
	Action string `json:"action"`
	Scope  string `json:"scope,omitempty"`
	Job    string `json:"job,omitempty"`
}

type response struct {
	OK      bool            `json:"ok"`
	Action  string          `json:"action"`
	Scope   string          `json:"scope,omitempty"`
	Job     string          `json:"job,omitempty"`
	Error   string          `json:"error,omitempty"`
	Payload json.RawMessage `json:"payload,omitempty"`
}

var jobPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`)

// Agent serves capture requests. Only one operation runs at a time: a library
// capture owns the writer group, so overlapping requests would race the window.
type Agent struct {
	config Config
	mu     sync.Mutex
}

func New(config Config) (*Agent, error) {
	if !filepath.IsAbs(config.Socket) || !filepath.IsAbs(config.ArchiveRoot) || !filepath.IsAbs(config.JournalRoot) {
		return nil, errors.New("library agent paths must be absolute")
	}
	if config.Binary == "" || config.Project == "" || config.Writers == "" || !strings.Contains(config.NonWriters, "=") {
		return nil, errors.New("library agent needs a binary, project, writers and a non-writer declaration")
	}
	if config.MaxDuration <= 0 || config.MaxDuration > 30*time.Minute {
		return nil, errors.New("library agent duration bound is invalid")
	}
	return &Agent{config: config}, nil
}

// Serve listens until ctx is cancelled. The socket is created 0600 inside a
// 0700 directory and removed on shutdown.
func (a *Agent) Serve(ctx context.Context) error {
	if err := os.MkdirAll(filepath.Dir(a.config.Socket), 0700); err != nil {
		return err
	}
	if err := os.Chmod(filepath.Dir(a.config.Socket), 0700); err != nil {
		return err
	}
	_ = os.Remove(a.config.Socket)
	listener, err := net.Listen("unix", a.config.Socket)
	if err != nil {
		return err
	}
	defer func() {
		listener.Close()
		_ = os.Remove(a.config.Socket)
	}()
	if err := os.Chmod(a.config.Socket, 0600); err != nil {
		return err
	}
	go func() {
		<-ctx.Done()
		_ = listener.Close()
	}()
	for {
		conn, err := listener.Accept()
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			return err
		}
		go a.handle(ctx, conn)
	}
}

func (a *Agent) handle(ctx context.Context, conn net.Conn) {
	defer conn.Close()
	if err := peerAllowed(conn); err != nil {
		_ = json.NewEncoder(conn).Encode(response{OK: false, Error: err.Error()})
		return
	}
	_ = conn.SetDeadline(time.Now().Add(a.config.MaxDuration + 30*time.Second))
	decoder := json.NewDecoder(bufio.NewReader(conn))
	var req request
	if err := decoder.Decode(&req); err != nil {
		_ = json.NewEncoder(conn).Encode(response{OK: false, Error: "invalid request"})
		return
	}
	result := a.execute(ctx, req)
	_ = json.NewEncoder(conn).Encode(result)
}

// execute serialises operations and maps a request onto the one fixed command.
func (a *Agent) execute(ctx context.Context, req request) response {
	a.mu.Lock()
	defer a.mu.Unlock()
	out := response{Action: req.Action, Scope: req.Scope, Job: req.Job}
	var args []string
	switch req.Action {
	case "status":
		args = []string{"status"}
	case "recover":
		args = []string{"recover"}
	case "capture":
		if req.Scope != "internal" && req.Scope != "external" {
			out.Error = "scope must be internal or external"
			return out
		}
		if req.Job != "" && !jobPattern.MatchString(req.Job) {
			out.Error = "invalid job id"
			return out
		}
		args = []string{"capture", "-scope", req.Scope}
		if req.Job != "" {
			args = append(args, "-job", req.Job)
		}
	default:
		out.Error = "unsupported action"
		return out
	}
	args = append(args, a.config.declaration()...)
	bounded, cancel := context.WithTimeout(ctx, a.config.MaxDuration)
	defer cancel()
	payload, err := a.config.run(bounded, a.config.Binary, args...)
	if err != nil {
		out.Error = "operation failed"
		if message := strings.TrimSpace(string(payload)); message != "" && len(message) <= 2000 {
			out.Error = message
		}
		return out
	}
	out.OK = true
	if json.Valid(payload) {
		out.Payload = json.RawMessage(payload)
	} else {
		out.Payload = json.RawMessage(`{}`)
	}
	return out
}

// declaration pins every path and the writer group so a request cannot redirect
// the operation to another stack, archive or journal.
func (c Config) declaration() []string {
	return []string{
		"-source-app", c.SourceApp,
		"-source-pg", c.SourcePG,
		"-source-es", c.SourceES,
		"-governance-pg", c.GovernancePG,
		"-project", c.Project,
		"-writers", c.Writers,
		"-non-writers", c.NonWriters,
		"-archive", c.ArchiveRoot,
		"-journal", c.JournalRoot,
	}
}

func (c Config) run(ctx context.Context, name string, args ...string) ([]byte, error) {
	if c.Run != nil {
		return c.Run(ctx, name, args...)
	}
	command := exec.CommandContext(ctx, name, args...)
	command.Env = append(os.Environ(), "LIBRARYCTL_AGENT=1")
	return command.CombinedOutput()
}

// Describe returns a redacted view for logs and health checks.
func (a *Agent) Describe() map[string]any {
	return map[string]any{
		"socket":       a.config.Socket,
		"archive_root": a.config.ArchiveRoot,
		"journal_root": a.config.JournalRoot,
		"project":      a.config.Project,
		"writers":      a.config.Writers,
		"scope_choice": "request may select internal or external only",
	}
}