mesh-host/internal/filling/filling.go

// Package filling is the node-engine's narrow local write verb (novox/hq ADR 0349 §4, design 57 P2): a socket on
// this machine through which the provisioner's client has a file it filled written, and nothing else.
//
// **The client fills; the node-engine writes.** The client renders a module's template with the values each domain
// owns and hands the node-engine the content; the node-engine writes it as the declared file, with the declared
// file's path, owner, mode and ownership at runtime, and makes what reads it take the change. So the one process that
// writes the machine's files is still the node-engine, and the client needs no right to write any of them.
//
// **Narrow by three checks, each from something the client cannot say for itself.** The account asking, from the
// kernel (SO_PEERCRED), in this machine's own namespaces, must be the account the node's current declaration names
// as the provisioner's client's; the path, clean and absolute, must be one the declaration names as filled; and the
// write takes the apply's lock, so a write and an apply never write one file at once. While the declaration names no
// account — every declaration before the provisioner's client has one — every write is refused.
//
// **No listening control surface** (novox/hq ADR 0004): the socket has no network address. And the content is never
// said: the journal has the path, the account and the content's digest.
package filling

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"os/user"
	"path/filepath"
	"sort"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/novox/mesh-host/internal/declaration"
	"github.com/novox/mesh-host/internal/meshcli"
)

// Socket is where the node-engine accepts a write of a filled file, beside mesh-cli's socket.
const Socket = "/run/nox-mesh-host/fill.sock"

// MaxRequest bounds one request, its content encoded in it: a configuration file, never an archive.
const MaxRequest = 8 << 20

// Request is one write, sent as one JSON object on one line: the path of a filled file and what the client filled
// into it (base64 in the JSON). The content only: the declaration says everything else about the file.
type Request struct {
	Path    string `json:"path"`
	Content []byte `json:"content"`
}

// Answer is what came of one request. Refused is a write that was not attempted, with why; Failed one that was
// attempted and did not complete, with why — the file, or something that reads it. Otherwise Action is what the
// write did to the file (created, updated, corrected, unchanged, kept) and Digest what the file holds now, as the
// node-engine records it.
type Answer struct {
	Path    string `json:"path,omitempty"`
	Action  string `json:"action,omitempty"`
	Digest  string `json:"digest,omitempty"`
	Refused string `json:"refused,omitempty"`
	Failed  string `json:"failed,omitempty"`
}

// Refusal is an answer that nothing was written, and why.
func Refusal(path, why string) Answer { return Answer{Path: path, Refused: why} }

// Writer is the node-engine's side of the socket.
type Writer interface {
	// Accepts judges the asking account before its request is read: nil when the current declaration names it as
	// the provisioner's client's.
	Accepts(uid uint32) error
	// Write judges the request and writes it, holding the apply's lock.
	Write(ctx context.Context, uid uint32, req Request) Answer
}

// Engine is the Writer the node-engine serves: the declaration it applies now, read again under the apply's lock.
type Engine struct {
	// Declared reads the declaration this node applies now, verified.
	Declared func() (*declaration.Declaration, error)
	// UIDOf is the user id of an account, by name, from this machine's user database.
	UIDOf func(account string) (uint32, error)
	// Lock takes the apply's lock and returns what releases it: the same the apply takes, so a write and an apply
	// are one at a time.
	Lock func() (func(), error)
	// Place writes the content into the filled file at path, as the declared file, and makes what reads it take the
	// change. Called holding the lock, with the declaration read under it.
	Place func(ctx context.Context, d *declaration.Declaration, path string, content []byte) (action, digest string,
		err error)
}

// Accepts reads the declaration and judges the account.
func (e *Engine) Accepts(uid uint32) error {
	d, err := e.Declared()
	if err != nil {
		return fmt.Errorf("this node's declaration cannot be read, so no write is accepted: %w", err)
	}
	return JudgeAccount(d, uid, e.UIDOf)
}

// Write takes the apply's lock, reads the declaration under it, judges the account and the path again — the
// declaration may have changed since the request was accepted — and writes.
func (e *Engine) Write(ctx context.Context, uid uint32, req Request) Answer {
	if err := JudgeShape(req.Path); err != nil {
		return Refusal(req.Path, err.Error())
	}
	unlock, err := e.Lock()
	if err != nil {
		return Refusal(req.Path, "the apply's lock could not be taken, so nothing was written: "+err.Error())
	}
	defer unlock()
	d, err := e.Declared()
	if err != nil {
		return Refusal(req.Path, "this node's declaration cannot be read, so nothing was written: "+err.Error())
	}
	if err := JudgeAccount(d, uid, e.UIDOf); err != nil {
		return Refusal(req.Path, err.Error())
	}
	if err := JudgePath(d, req.Path); err != nil {
		return Refusal(req.Path, err.Error())
	}
	action, digest, err := e.Place(ctx, d, req.Path, req.Content)
	if err != nil {
		return Answer{Path: req.Path, Action: action, Digest: digest, Failed: err.Error()}
	}
	return Answer{Path: req.Path, Action: action, Digest: digest}
}

// JudgeAccount refuses every account but the one the declaration names as the provisioner's client's, by its user
// id here — and every account while the declaration names none.
func JudgeAccount(d *declaration.Declaration, uid uint32, uidOf func(string) (uint32, error)) error {
	if d.Provisioner == "" {
		return errors.New("this node's declaration names no provisioner client's account, so no write is accepted")
	}
	want, err := uidOf(d.Provisioner)
	if err != nil {
		return fmt.Errorf("the provisioner client's account %s is not on this machine, so no write is accepted: %w",
			d.Provisioner, err)
	}
	if want == 0 {
		return fmt.Errorf("the provisioner client's account %s is root here, so no write is accepted", d.Provisioner)
	}
	if uid != want {
		return fmt.Errorf("user id %d is not the provisioner client's account %s (user id %d), so nothing was written",
			uid, d.Provisioner, want)
	}
	return nil
}

// JudgeShape refuses a path that is not clean and absolute: a relative path, `..`, a doubled or trailing slash.
// Compared as written with the declaration's, a path that is not clean could name a file the declaration does not.
func JudgeShape(path string) error {
	switch {
	case path == "":
		return errors.New("the request names no path, so nothing was written")
	case !filepath.IsAbs(path):
		return fmt.Errorf("%q is not an absolute path, so nothing was written", path)
	case filepath.Clean(path) != path:
		return fmt.Errorf("%q is not a clean path (it is %q cleaned), so nothing was written", path,
			filepath.Clean(path))
	}
	return nil
}

// JudgePath refuses a path that is not clean and absolute, or that the declaration does not name as filled. Where it
// leads — a link on the way, a place nothing is written — is judged by the write, by the apply's own rules.
func JudgePath(d *declaration.Declaration, path string) error {
	if err := JudgeShape(path); err != nil {
		return err
	}
	if !d.Fills(path) {
		return fmt.Errorf("%s is not a file this node's declaration names as filled, so nothing was written", path)
	}
	return nil
}

// UIDOf is the user id of an account by name, from this machine's user database.
func UIDOf(account string) (uint32, error) {
	u, err := user.Lookup(account)
	if err != nil {
		return 0, err
	}
	id, err := strconv.ParseUint(u.Uid, 10, 32)
	if err != nil {
		return 0, fmt.Errorf("user id %q: %w", u.Uid, err)
	}
	return uint32(id), nil
}

// peerOf and sameNamespaces are the kernel's word on the asking process, as mesh-cli's socket reads it. Variables so
// a test can say.
var (
	peerOf         = meshcli.PeerCredentials
	sameNamespaces = meshcli.SameNamespaces
)

// within bounds how long an accepted connection may take to say its request, and refusedWithin how long a busy one
// is read and dropped before its refusal is written (so the client is not answered with a broken pipe). Variables so
// a test need not wait.
var (
	within        = 30 * time.Second
	refusedWithin = 1 * time.Second
)

// atOnce bounds the accepted connections served at the same time: one client, writing one file at a time. One over
// is refused busy. busyWaitingAtOnce bounds how many busy refusals wait for their request at once, per Serve; over it
// a busy refusal is written at once, as mesh-cli's socket does. judgingAtOnce bounds the callers judged at once, and
// judgeWaiting the connections waiting to be judged — each an open file, so far below the engine's open-file limit;
// one over is closed at once, unread. Variables so a test can say.
var (
	atOnce            = 4
	busyWaitingAtOnce = 8
	judgingAtOnce     = 16
	judgeWaiting      = 256
)

// refusalsSaidEvery is how often the refusals counted and not yet said are said: each kind of refusal of each account
// is said at once the first time, and what follows it within this time is counted and said at its end, or when the
// socket stops. A variable so a test need not wait.
var refusalsSaidEvery = time.Minute

// acceptBackoff is how long the socket waits after an accept that failed before it accepts again, at first and at
// most: a socket that stopped at the first such failure (too many open files, say) would stop the client's writes
// until the engine restarted. Variables so a test need not wait.
var (
	acceptBackoff    = 50 * time.Millisecond
	acceptBackoffMax = 5 * time.Second
)

// acceptFrom accepts one connection. A variable so a test can make it fail.
var acceptFrom = func(l *net.UnixListener) (*net.UnixConn, error) { return l.AcceptUnix() }

// Serve listens on socket until ctx ends, and answers each connection: one request, one answer.
//
// The directory is made 0755 and the socket 0666, as mesh-cli's: who may write is judged per connection from the
// kernel's word, against the current declaration, so the socket needs no account that may not exist yet. **The
// caller is judged before it holds anything**: one the kernel does not say is the provisioner client's account is
// answered and closed at once, and only the client's connections are served or wait as busy, so no other account can
// hold the socket from it. A file at the socket's path that is not a socket is never removed: the engine says so and
// serves nothing.
func Serve(ctx context.Context, socket string, w Writer, say func(string)) error {
	if say == nil {
		say = func(string) {}
	}
	dir := filepath.Dir(socket)
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return fmt.Errorf("the filled files' socket directory: %w", err)
	}
	if problem := meshcli.NotOwnDirectory(dir); problem != "" {
		return fmt.Errorf("filled files are not written on request on this machine: %s", problem)
	}
	if err := os.Chmod(dir, 0o755); err != nil {
		return fmt.Errorf("the filled files' socket directory: %w", err)
	}
	if info, err := os.Lstat(socket); err == nil {
		if info.Mode()&os.ModeSocket == 0 {
			return fmt.Errorf("%s is there and is not a socket, and the node-engine did not put it there; filled "+
				"files are not written on request until it is moved", socket)
		}
		if err := os.Remove(socket); err != nil {
			return fmt.Errorf("the socket a previous engine left at %s: %w", socket, err)
		}
	}
	l, err := net.ListenUnix("unix", &net.UnixAddr{Name: socket, Net: "unix"})
	if err != nil {
		return fmt.Errorf("listening for filled files at %s: %w", socket, err)
	}
	if err := os.Chmod(socket, 0o666); err != nil {
		l.Close()
		return fmt.Errorf("opening the filled files' socket: %w", err)
	}
	go func() {
		<-ctx.Done()
		l.Close()
	}()
	slots := make(chan struct{}, atOnce)
	// Judging reads the declaration: at most judging callers are judged at once, and the rest wait their turn — a
	// few milliseconds each — rather than being refused, so a flood delays the client's write and never stops it.
	judging := make(chan struct{}, judgingAtOnce)
	var waiting atomic.Int64
	busyWaiting := make(chan struct{}, busyWaitingAtOnce)
	refusals := &refusalWords{say: say}
	// Said once everything has stopped: what was counted is never lost with the socket.
	defer refusals.flush()
	var wg sync.WaitGroup
	defer wg.Wait()
	wg.Add(1)
	go func() {
		defer wg.Done()
		tick := time.NewTicker(refusalsSaidEvery)
		defer tick.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-tick.C:
				refusals.flush()
			}
		}
	}()
	backoff := time.Duration(0)
	for {
		conn, err := acceptFrom(l)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			if errors.Is(err, net.ErrClosed) {
				return fmt.Errorf("the filled files' socket was closed: %w", err)
			}
			// Said, waited for, and accepted again: the failure is the machine's for a moment, not the socket's.
			backoff = min(max(2*backoff, acceptBackoff), acceptBackoffMax)
			say(fmt.Sprintf("the filled files' socket could not accept a connection (%v); accepting again in %s",
				err, backoff))
			select {
			case <-ctx.Done():
				return nil
			case <-time.After(backoff):
			}
			continue
		}
		backoff = 0
		if waiting.Add(1) > int64(judgeWaiting) {
			// Too many waiting to be judged already: this one is closed unread, and counted.
			waiting.Add(-1)
			refusals.said(^uint32(0), "overflow", "too many connections are waiting to be judged; closed unread")
			conn.Close()
			continue
		}
		wg.Add(1)
		go func() {
			defer wg.Done()
			// Judged before it holds a slot or is waited for.
			judging <- struct{}{}
			uid, ok := judgePeer(conn, w, refusals)
			<-judging
			waiting.Add(-1)
			if !ok {
				return
			}
			readBy := time.Now().Add(within)
			_ = conn.SetDeadline(readBy)
			select {
			case slots <- struct{}{}:
				defer func() { <-slots }()
				answer(ctx, conn, uid, w, say)
			default:
				busy(conn, readBy, busyWaiting, say)
			}
		}()
	}
}

// judgePeer reads the asking account from the kernel and judges it. A caller that is not the provisioner client's
// account is answered with why and closed at once, its request never read: false.
func judgePeer(conn *net.UnixConn, w Writer, refusals *refusalWords) (uint32, bool) {
	closeWith := func(uid uint32, kind, why string) (uint32, bool) {
		refusals.said(uid, kind, why)
		_ = conn.SetWriteDeadline(time.Now().Add(refusedWithin))
		_ = json.NewEncoder(conn).Encode(Refusal("", why))
		conn.Close()
		return uid, false
	}
	uid, pid, err := peerOf(conn)
	if err != nil {
		return closeWith(^uint32(0), "unread", "the node-engine could not read which account is asking, so nothing was "+
			"written: "+err.Error())
	}
	if err := sameNamespaces(pid); err != nil {
		return closeWith(uid, "namespace", "the asking process does not run in this machine's own namespaces, so nothing was "+
			"written: "+err.Error())
	}
	if err := w.Accepts(uid); err != nil {
		return closeWith(uid, "account", err.Error())
	}
	return uid, true
}

// refusalWords says refused callers in the journal without letting a flood flood it: the first refusal of each kind
// for each account at once, and those that follow it counted and said by flush — on a timer and when the socket stops
// — so an account that stops still has its count said, and a refusal of another kind never hides behind one said.
type refusalWords struct {
	mu      sync.Mutex
	say     func(string)
	pending map[refusalKey]*refusalCount
}

type refusalKey struct {
	uid  uint32
	kind string
}

type refusalCount struct {
	n   int
	why string
}

func (r *refusalWords) said(uid uint32, kind, why string) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.pending == nil {
		r.pending = map[refusalKey]*refusalCount{}
	}
	k := refusalKey{uid, kind}
	if c, ok := r.pending[k]; ok {
		c.n++
		c.why = why
		return
	}
	r.pending[k] = &refusalCount{}
	r.say(fmt.Sprintf("a filled file's write from %s was refused (%s): %s", who(k), kind, why))
}

// flush says what was counted since each was last said, and forgets every key, so the next refusal of each is said at
// once again.
func (r *refusalWords) flush() {
	r.mu.Lock()
	defer r.mu.Unlock()
	keys := make([]refusalKey, 0, len(r.pending))
	for k := range r.pending {
		keys = append(keys, k)
	}
	sort.Slice(keys, func(i, j int) bool {
		if keys[i].uid != keys[j].uid {
			return keys[i].uid < keys[j].uid
		}
		return keys[i].kind < keys[j].kind
	})
	for _, k := range keys {
		if c := r.pending[k]; c.n > 0 {
			r.say(fmt.Sprintf("%d more writes of a filled file from %s were refused (%s) since it was last said; the last: %s",
				c.n, who(k), k.kind, c.why))
		}
	}
	r.pending = nil
}

// who names a caller by its user id, or says why it has none.
func who(k refusalKey) string {
	switch {
	case k.kind == "overflow":
		return "a caller not yet judged"
	case k.uid == ^uint32(0):
		return "an account that could not be read"
	}
	return fmt.Sprintf("user id %d", k.uid)
}

// answer serves one connection from the provisioner client's account.
func answer(ctx context.Context, conn *net.UnixConn, uid uint32, w Writer, say func(string)) {
	var req Request
	dec := json.NewDecoder(io.LimitReader(conn, MaxRequest))
	dec.DisallowUnknownFields()
	if err := dec.Decode(&req); err != nil {
		// Answered even when the deadline to say it passed: the deadline covered the write too.
		_ = conn.SetWriteDeadline(time.Now().Add(refusedWithin))
		reply(conn, Refusal("", "the request could not be read, so nothing was written: "+err.Error()), say)
		return
	}
	// The write waits for the apply's lock as long as an apply takes, and the client waits with it.
	_ = conn.SetDeadline(time.Time{})
	a := w.Write(ctx, uid, req)
	switch {
	case a.Refused != "":
		say(fmt.Sprintf("a write of %s from user id %d was refused: %s", req.Path, uid, a.Refused))
	case a.Failed != "":
		say(fmt.Sprintf("a write of %s from user id %d failed: %s", req.Path, uid, a.Failed))
	default:
		say(fmt.Sprintf("wrote %s for the provisioner's client (user id %d): %s, %s", req.Path, uid, a.Action,
			a.Digest))
	}
	_ = conn.SetWriteDeadline(time.Now().Add(within))
	reply(conn, a, say)
}

// busy answers the client's connection over atOnce: its request read and dropped first, briefly, so the client reads
// the answer rather than a broken pipe — while no more than busyWaiting such refusals wait at once; over that, at once.
func busy(conn *net.UnixConn, readBy time.Time, busyWaiting chan struct{}, say func(string)) {
	select {
	case busyWaiting <- struct{}{}:
		by := time.Now().Add(refusedWithin)
		if readBy.Before(by) {
			by = readBy
		}
		_ = conn.SetReadDeadline(by)
		drain(conn)
		<-busyWaiting
	default:
	}
	_ = conn.SetWriteDeadline(time.Now().Add(refusedWithin))
	reply(conn, Refusal("", "busy: other writes are being served; ask again"), say)
}

// drain reads and drops a request up to its end — the newline after it, the end of the stream or the deadline —
// and at most MaxRequest bytes, keeping none of it.
func drain(r io.Reader) {
	var buf [4096]byte
	for read := 0; read < MaxRequest; {
		n, err := r.Read(buf[:])
		read += n
		if bytes.IndexByte(buf[:n], '\n') >= 0 || err != nil {
			return
		}
	}
}

func reply(conn *net.UnixConn, a Answer, say func(string)) {
	defer conn.Close()
	if err := json.NewEncoder(conn).Encode(a); err != nil {
		say("a filled file's answer could not be written back: " + err.Error())
	}
}