2 changed files with 610 additions and 0 deletions
@@ -0,0 +1,347 @@
+// 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"
+	"strconv"
+	"sync"
+	"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 a connection may take to say its request, and refusedWithin how long a refused 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 connections served at the same time: one client, writing one file at a time. One over is
+// refused busy at once.
+var atOnce = 4
+
+// 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. 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 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)
+	var wg sync.WaitGroup
+	defer wg.Wait()
+	for {
+		conn, err := l.AcceptUnix()
+		if err != nil {
+			if ctx.Err() != nil {
+				return nil
+			}
+			var ne net.Error
+			if errors.As(err, &ne) && ne.Timeout() {
+				continue
+			}
+			return fmt.Errorf("the filled files' socket: %w", err)
+		}
+		readBy := time.Now().Add(within)
+		_ = conn.SetDeadline(readBy)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			select {
+			case slots <- struct{}{}:
+				defer func() { <-slots }()
+				answer(ctx, conn, readBy, w, say)
+			default:
+				refuse(conn, readBy, Refusal("", "busy: other writes are being served; ask again"), say)
+			}
+		}()
+	}
+}
+
+// answer serves one connection.
+func answer(ctx context.Context, conn *net.UnixConn, readBy time.Time, w Writer, say func(string)) {
+	uid, pid, err := peerOf(conn)
+	if err != nil {
+		say("a filled file's write was refused: the asking account could not be read: " + err.Error())
+		refuse(conn, readBy, Refusal("", "the node-engine could not read which account is asking, so nothing was "+
+			"written: "+err.Error()), say)
+		return
+	}
+	if err := sameNamespaces(pid); err != nil {
+		say(fmt.Sprintf("a filled file's write from user id %d was refused: not in this machine's namespaces: %v",
+			uid, err))
+		refuse(conn, readBy, Refusal("", "the asking process does not run in this machine's own namespaces, so "+
+			"nothing was written: "+err.Error()), say)
+		return
+	}
+	if err := w.Accepts(uid); err != nil {
+		say(fmt.Sprintf("a filled file's write from user id %d was refused: %v", uid, err))
+		refuse(conn, readBy, Refusal("", err.Error()), say)
+		return
+	}
+	var req Request
+	if err := json.NewDecoder(io.LimitReader(conn, MaxRequest)).Decode(&req); err != nil {
+		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)
+}
+
+// refuse reads and drops what a refused client wrote, briefly, and answers it: nothing it sent is kept.
+func refuse(conn *net.UnixConn, readBy time.Time, a Answer, say func(string)) {
+	by := time.Now().Add(refusedWithin)
+	if readBy.Before(by) {
+		by = readBy
+	}
+	_ = conn.SetReadDeadline(by)
+	drain(conn)
+	_ = conn.SetWriteDeadline(time.Now().Add(within))
+	reply(conn, a, 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())
+	}
+}
@@ -0,0 +1,263 @@
+package filling
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/novox/mesh-host/internal/declaration"
+)
+
+// Defends novox/hq ADR 0349 §4 (issue 668): the node-engine writes a filled file on request only for the provisioner
+// client's account, only for a clean path the current declaration names as filled, and one at a time with the apply.
+
+const provisionerUID = 4242
+
+func declared(t *testing.T, raw string) *declaration.Declaration {
+	t.Helper()
+	d, err := declaration.Parse([]byte(raw))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return d
+}
+
+const withAccount = `{"declaration":1,"provisioner_account":"mesh-node-provisioner","filled_files":["/etc/m/m.conf"],
+  "resources":[{"id":"m.conf","type":"file","path":"/etc/m/m.conf","content":"x","owner-at-runtime":"owned"},
+  {"id":"m.other","type":"file","path":"/etc/m/other.conf","content":"y"}]}`
+
+const withoutAccount = `{"declaration":1,"filled_files":["/etc/m/m.conf"],
+  "resources":[{"id":"m.conf","type":"file","path":"/etc/m/m.conf","content":"x","owner-at-runtime":"owned"}]}`
+
+// engine is an Engine over a fixed declaration, recording what it was asked to place.
+type engine struct {
+	Engine
+	mu     sync.Mutex
+	placed []string
+}
+
+func newEngine(t *testing.T, raw string, lock sync.Locker) *engine {
+	d := declared(t, raw)
+	e := &engine{}
+	e.Engine = Engine{
+		Declared: func() (*declaration.Declaration, error) { return d, nil },
+		UIDOf: func(name string) (uint32, error) {
+			if name == "mesh-node-provisioner" {
+				return provisionerUID, nil
+			}
+			return 0, errors.New("no such account")
+		},
+		Lock: func() (func(), error) { lock.Lock(); return lock.Unlock, nil },
+		Place: func(_ context.Context, _ *declaration.Declaration, path string, content []byte) (string, string, error) {
+			e.mu.Lock()
+			defer e.mu.Unlock()
+			e.placed = append(e.placed, path)
+			return "updated", "digest", nil
+		},
+	}
+	return e
+}
+
+func TestAWriteFromTheProvisionersAccountOfAFilledPathIsPlaced(t *testing.T) {
+	e := newEngine(t, withAccount, &sync.Mutex{})
+	if err := e.Accepts(provisionerUID); err != nil {
+		t.Fatal(err)
+	}
+	a := e.Write(context.Background(), provisionerUID, Request{Path: "/etc/m/m.conf", Content: []byte("x")})
+	if a.Refused != "" || a.Failed != "" || a.Action != "updated" || len(e.placed) != 1 {
+		t.Fatalf("answer %+v, placed %v", a, e.placed)
+	}
+}
+
+func TestAnUndeclaredPathIsRefused(t *testing.T) {
+	e := newEngine(t, withAccount, &sync.Mutex{})
+	for _, path := range []string{"/etc/m/other.conf", "/etc/m/none.conf", "/etc/passwd"} {
+		a := e.Write(context.Background(), provisionerUID, Request{Path: path, Content: []byte("evil")})
+		if a.Refused == "" || !strings.Contains(a.Refused, "not a file this node's declaration names as filled") {
+			t.Errorf("%s: %+v", path, a)
+		}
+	}
+	if len(e.placed) != 0 {
+		t.Errorf("placed %v", e.placed)
+	}
+}
+
+func TestAnotherAccountIsRefused(t *testing.T) {
+	e := newEngine(t, withAccount, &sync.Mutex{})
+	for _, uid := range []uint32{0, 1000, provisionerUID + 1} {
+		if err := e.Accepts(uid); err == nil {
+			t.Errorf("user id %d was accepted", uid)
+		}
+		a := e.Write(context.Background(), uid, Request{Path: "/etc/m/m.conf", Content: []byte("evil")})
+		if a.Refused == "" || !strings.Contains(a.Refused, "is not the provisioner client's account") {
+			t.Errorf("user id %d: %+v", uid, a)
+		}
+	}
+	if len(e.placed) != 0 {
+		t.Errorf("placed %v", e.placed)
+	}
+}
+
+func TestEveryWriteIsRefusedWhileTheDeclarationNamesNoAccount(t *testing.T) {
+	e := newEngine(t, withoutAccount, &sync.Mutex{})
+	if err := e.Accepts(provisionerUID); err == nil || !strings.Contains(err.Error(), "names no provisioner") {
+		t.Errorf("accepted with no account named: %v", err)
+	}
+	a := e.Write(context.Background(), provisionerUID, Request{Path: "/etc/m/m.conf", Content: []byte("x")})
+	if a.Refused == "" || len(e.placed) != 0 {
+		t.Errorf("written with no account named: %+v", a)
+	}
+}
+
+func TestAnAccountThatIsRootHereIsRefused(t *testing.T) {
+	e := newEngine(t, withAccount, &sync.Mutex{})
+	e.UIDOf = func(string) (uint32, error) { return 0, nil }
+	if err := e.Accepts(0); err == nil || !strings.Contains(err.Error(), "root") {
+		t.Errorf("an account that is root here was accepted: %v", err)
+	}
+}
+
+func TestAnUncleanPathIsRefused(t *testing.T) {
+	e := newEngine(t, withAccount, &sync.Mutex{})
+	for _, path := range []string{"", "etc/m/m.conf", "m.conf", "/etc//m/m.conf", "/etc/m/../m/m.conf",
+		"/etc/m/./m.conf", "/etc/m/m.conf/"} {
+		a := e.Write(context.Background(), provisionerUID, Request{Path: path, Content: []byte("x")})
+		if a.Refused == "" {
+			t.Errorf("%q was not refused: %+v", path, a)
+		}
+	}
+	if len(e.placed) != 0 {
+		t.Errorf("placed %v", e.placed)
+	}
+}
+
+// A write waits for an apply holding the apply's lock, and writes only once it is let go.
+func TestAWriteWaitsForTheApply(t *testing.T) {
+	var applying sync.Mutex
+	e := newEngine(t, withAccount, &applying)
+	applying.Lock()
+	done := make(chan Answer)
+	go func() {
+		done <- e.Write(context.Background(), provisionerUID, Request{Path: "/etc/m/m.conf", Content: []byte("x")})
+	}()
+	select {
+	case a := <-done:
+		t.Fatalf("written while the apply held the lock: %+v", a)
+	case <-time.After(100 * time.Millisecond):
+	}
+	e.mu.Lock()
+	n := len(e.placed)
+	e.mu.Unlock()
+	if n != 0 {
+		t.Fatalf("placed while the apply held the lock")
+	}
+	applying.Unlock()
+	if a := <-done; a.Action != "updated" {
+		t.Fatalf("after the apply: %+v", a)
+	}
+}
+
+// fixed is a Writer that accepts one user id and answers every write.
+type fixed struct {
+	uid    uint32
+	mu     sync.Mutex
+	writes []Request
+}
+
+func (f *fixed) Accepts(uid uint32) error {
+	if uid != f.uid {
+		return errors.New("not the provisioner client's account")
+	}
+	return nil
+}
+
+func (f *fixed) Write(_ context.Context, _ uint32, req Request) Answer {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.writes = append(f.writes, req)
+	return Answer{Path: req.Path, Action: "updated", Digest: "d"}
+}
+
+// serve serves w on a socket in a fresh directory and returns its path, and what was said.
+func serve(t *testing.T, w Writer) (string, func() []string) {
+	t.Helper()
+	dir, err := os.MkdirTemp("", "fill")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	socket := filepath.Join(dir, "fill.sock")
+	ctx, cancel := context.WithCancel(context.Background())
+	var mu sync.Mutex
+	var said []string
+	stopped := make(chan error)
+	go func() {
+		stopped <- Serve(ctx, socket, w, func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() })
+	}()
+	t.Cleanup(func() {
+		cancel()
+		if err := <-stopped; err != nil {
+			t.Error(err)
+		}
+	})
+	for i := 0; i < 100; i++ {
+		if _, err := os.Stat(socket); err == nil {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return socket, func() []string { mu.Lock(); defer mu.Unlock(); return append([]string(nil), said...) }
+}
+
+func ask(t *testing.T, socket string, req Request) Answer {
+	t.Helper()
+	conn, err := net.Dial("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	if err := json.NewEncoder(conn).Encode(req); err != nil {
+		t.Fatal(err)
+	}
+	var a Answer
+	if err := json.NewDecoder(conn).Decode(&a); err != nil {
+		t.Fatal(err)
+	}
+	return a
+}
+
+// Over a real socket, the kernel says who is asking: this test's own account is the provisioner's here.
+func TestTheSocketWritesForTheAccountTheKernelSays(t *testing.T) {
+	w := &fixed{uid: uint32(os.Getuid())}
+	socket, said := serve(t, w)
+	a := ask(t, socket, Request{Path: "/etc/m/m.conf", Content: []byte("secret value")})
+	if a.Refused != "" || a.Action != "updated" || len(w.writes) != 1 || string(w.writes[0].Content) != "secret value" {
+		t.Fatalf("answer %+v, writes %+v", a, w.writes)
+	}
+	for _, line := range said() {
+		if strings.Contains(line, "secret value") {
+			t.Errorf("the content was said: %q", line)
+		}
+	}
+}
+
+func TestTheSocketRefusesAnotherAccountBeforeReadingItsRequest(t *testing.T) {
+	w := &fixed{uid: uint32(os.Getuid()) + 1}
+	socket, said := serve(t, w)
+	a := ask(t, socket, Request{Path: "/etc/m/m.conf", Content: []byte("secret value")})
+	if a.Refused == "" || len(w.writes) != 0 {
+		t.Fatalf("another account was written for: %+v", a)
+	}
+	for _, line := range said() {
+		if strings.Contains(line, "secret value") {
+			t.Errorf("the content was said: %q", line)
+		}
+	}
+}