mesh-catalog/modules/gitea/cmd/forge-account-provider/service.go

package main

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"sort"
	"strings"
	"sync"
	"time"

	"git.novox.be/novox/mesh-sdk/go/provider"

	"gitea/internal/forgeaccount"
	"gitea/internal/forgehttp"
)

// ProvisionName is what gitea provides here, as its manifest says (novox/hq ADR 0400).
const ProvisionName = "forge-account"

// Module is gitea's name in the mesh.
const Module = "gitea"

// Runner runs a program; a test fakes it.
type Runner interface {
	Run(ctx context.Context, name string, args ...string) ([]byte, error)
}

// KeySource is the machines' public keys as the bus keeps them (ssh-client's state `keys`): the whole current set,
// then each change. handle is told the key, the operation ("put" or "delete"), the value and whether it was there
// when the watch began.
type KeySource interface {
	Watch(handle func(key, op string, value []byte, current bool)) error
}

// Config is what the service needs from its environment.
type Config struct {
	// Node is the forge's machine.
	Node string
	// HostFile holds the forge machine's own mesh name (a fact of the machine), OriginsFile the forge's public
	// origins, one per line.
	HostFile, OriginsFile string
	// Port is where the forge's sshd is published on the mesh.
	Port int
	// HostKeyFile is the forge's own ed25519 host key (the .pub file in its data directory). It is read first; when it
	// is unset or unreadable the forge's sshd is asked with ssh-keyscan instead (issue 1939).
	HostKeyFile string
	Forge       forgehttp.Client
	Runner      Runner
	Log         func(format string, a ...any)
}

// Service answers the consumers of the provision `forge-account` (novox/hq ADR 0399, ADR 0400; issue 955): for each
// machine whose ssh-client reported a public key it makes the machine's agent a forge account, registers the key
// for it, and serves where and as whom to reach the forge over ssh. A key is no secret: the offer has no credential.
type Service struct {
	cfg      Config
	accounts *forgeaccount.Accounts

	// OnKeys is told the node whose key changed, so its consumer is answered without waiting for the next pass.
	OnKeys func(node string)
	// Pause waits between asks of a watch the runtime does not answer yet.
	Pause func(time.Duration)

	mu     sync.Mutex
	keys   map[string]machineKey
	loaded bool // the machines' keys have been read: until then a node with none has not reported none
	// loadedFrom says of each build module's state that its keys have been read, as loaded says of ssh-client's: a build
	// module's state that never answers (the module is not installed) holds back its own entries and no other.
	loadedFrom map[string]bool
	known      string // the forge's host key line, as the machines write it into known_hosts

	bound provider.Bus // the bus the entries are written to (set by Provider)

	letGo map[string]bool // nodes whose per-node user of before has been said to be let go

	making sync.Mutex // one Make at a time: they register keys, and two at once would race each other
}

type machineKey struct {
	Node, Account, Kind, Public string
	// Source is the module whose state the key was read from: ssh-client's for a machine's, a build module's for a
	// builder's. Empty is ssh-client's.
	Source string
}

// SSHClient is the module whose `keys` state holds the machines' keys (kind machine); the build modules' states hold
// the builders' (kind builder, ADR 0466 §2).
const SSHClient = "ssh-client"

// sourceOf is the module a stored key was read from.
func (k machineKey) sourceOf() string {
	if k.Source == "" {
		return SSHClient
	}
	return k.Source
}

// storageKey is the key a source's key is kept under: a machine's as it is, a builder's under its module's name.
func storageKey(source, key string) string {
	if source == "" || source == SSHClient {
		return key
	}
	return source + "/" + key
}

// kindOf is the kind of key a consumer module's entry is made for.
func kindOf(module string) string { return forgeaccount.KindOfModule(module) }

// NewService is the service over its environment.
func NewService(cfg Config) *Service {
	if cfg.Log == nil {
		cfg.Log = func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[gitea forge-account] "+f+"\n", a...) }
	}
	return &Service{cfg: cfg, keys: map[string]machineKey{}, letGo: map[string]bool{}, loadedFrom: map[string]bool{}, Pause: time.Sleep,
		accounts: &forgeaccount.Accounts{Forge: cfg.Forge, Log: cfg.Log}}
}

// Provider is the provider helper answering the consumers over bus. Every consumer is made again every ten minutes,
// which puts back an account, its team or a key changed by hand.
func (s *Service) Provider(bus provider.Bus) *provider.Provider {
	vb := &valuesBus{Bus: bus, s: s, written: map[string][]byte{}}
	s.bound = vb
	return provider.New(provider.Options{
		Node: s.cfg.Node, Module: Module,
		Offers: map[string]provider.Offer{ProvisionName: {Credential: provider.NoCredential}},
		Bus:    vb,
		Make:   s.Make, Gone: s.Gone,
		Every: 10 * time.Minute,
		Logf:  s.cfg.Log,
	})
}

// errNoKey is a consumer whose machine has reported no key yet: nothing is made, and the client says it waits.
type errNoKey struct{ node, module string }

func (e errNoKey) Error() string {
	if kind := kindOf(e.module); kind == "builder" {
		return "no key from " + e.node + " yet: its " + e.module + " has not reported a builder key, so there is no " +
			"entry to make"
	}
	return "no key from " + e.node + " yet: its ssh-client has not reported a machine key, so there is no account " +
		"and no entry to make"
}

// sourceOfModule is the module whose state holds the keys a consumer module's entry is made from: a build module's
// own, ssh-client's for any other consumer.
func sourceOfModule(module string) string {
	if kindOf(module) == "builder" {
		return module
	}
	return SSHClient
}

// scopeFor is the title prefix of the keys one entry registered: every key of a machine for a machine's entry, one
// build module's key for a builder's (nox-builder holds the keys of every build module of the mesh).
func scopeFor(kind, node, module string) string {
	if kind == "builder" {
		return forgeaccount.BuilderScope(node, module)
	}
	return forgeaccount.MachineScope(node)
}

// Make brings one consumer's account, its team membership and its keys into being, as the machine reported them. It
// leaves what is as asked alone, so it is asked again freely.
func (s *Service) Make(ctx context.Context, c provider.Consumer) error {
	s.making.Lock()
	defer s.making.Unlock()
	node := c.Entry.Consumer
	module, kind := c.Entry.Module, kindOf(c.Entry.Module)
	user := forgeaccount.ForgeUser(node, kind)
	if user == "" {
		return fmt.Errorf("%q is not a machine's name an account can be named for", node)
	}
	source := sourceOfModule(module)
	s.mu.Lock()
	loaded := s.loaded
	if source != SSHClient {
		loaded = s.loadedFrom[source]
	}
	s.mu.Unlock()
	if !loaded {
		// After a restart the first pass can come before the keys arrive: nothing is removed or deleted on that.
		// Not ready, not failed: the entry waits and keeps serving what it served (provider.ErrNotReady).
		return fmt.Errorf("the keys of %s are not read yet, so nothing is made, removed or deleted: %w", source,
			provider.ErrNotReady)
	}
	have := s.keysFrom(source, node, kind)
	if len(have) == 0 {
		if kind == "machine" {
			if err := s.releasePerNodeUser(ctx, node); err != nil {
				return err
			}
		}
		// A key the machine no longer reports is no longer wanted (ADR 0400 §2): the mesh's keys go, the operator's
		// stay, and the entry that served the machine goes with them.
		if _, err := s.accounts.RemoveKeysTitled(ctx, user, scopeFor(kind, node, module)); err != nil {
			return err
		}
		if s.bound != nil {
			if err := s.bound.Delete(ctx, provider.BucketOf(Module), c.Entry.Key()); err != nil {
				return err
			}
		}
		return errNoKey{node: node, module: module}
	}
	if _, err := s.hostName(); err != nil {
		return err
	}
	known, err := s.scan(ctx)
	if err != nil {
		return err
	}
	if err := s.accounts.EnsureAccount(ctx, user); err != nil {
		return err
	}
	if err := s.accounts.EnsureInTeamOf(ctx, forgeaccount.TeamOfUser(user), user); err != nil {
		return err
	}
	var wanted []forgeaccount.WantedKey
	for _, k := range have {
		// A builder's key may read and never push (ADR 0466 §3): the key says it beside the team that does.
		wanted = append(wanted, forgeaccount.WantedKey{Title: forgeaccount.KeyTitle(k.Node, k.Account), Key: k.Public, ReadOnly: kind == "builder"})
	}
	// The forge holds one public key on one user only, so the key leaves agent-<node> before it joins nox-agent. That
	// is done here, after everything that can fail before it, so the window in which the machine's key is on no user
	// is only the span between these two calls; if the add fails, the next pass (at most ten minutes) puts it back.
	if kind == "machine" {
		if err := s.releasePerNodeUser(ctx, node); err != nil {
			return err
		}
	}
	if err := s.accounts.EnsureKeys(ctx, user, scopeFor(kind, node, module), wanted); err != nil {
		return err
	}
	s.mu.Lock()
	s.known = known
	s.mu.Unlock()
	return nil
}

// Gone removes the keys the mesh registered for a machine (or, for a build module, the module's key on that machine)
// that no longer binds the forge. The account stays: it may
// hold what the machine's agent wrote, and the keys of the operator's own that were added by hand are not touched
// (ADR 0399 §2).
func (s *Service) Gone(ctx context.Context, e provider.Entry) error {
	s.making.Lock()
	defer s.making.Unlock()
	kind := kindOf(e.Module)
	user := forgeaccount.ForgeUser(e.Consumer, kind)
	if user == "" {
		return nil
	}
	if kind == "machine" {
		if err := s.releasePerNodeUser(ctx, e.Consumer); err != nil {
			return err
		}
	}
	// A builder's entry gone removes that entry's key only: the user and the team serve the other machines.
	_, err := s.accounts.RemoveKeysTitled(ctx, user, scopeFor(kind, e.Consumer, e.Module))
	return err
}

// releasePerNodeUser lets go the per-node user (agent-<node>) a machine had before one user held every machine's keys
// (ADR 0410 §3): the mesh keys registered on it are removed. The user is never
// deleted and stays in the agents team while agent-token mints it a token, and the keys of the operator's own stay. It is said once per machine.
func (s *Service) releasePerNodeUser(ctx context.Context, node string) error {
	old := forgeaccount.NodeAccount(node)
	if !forgeaccount.NodeName.MatchString(node) || old == forgeaccount.AgentUser {
		return nil
	}
	removed, err := s.accounts.RemoveKeysTitled(ctx, old, forgeaccount.MachineScope(node))
	if err != nil && !errors.Is(err, forgehttp.ErrNotFound) {
		return err
	}
	if removed == 0 {
		return nil // never existed, or holds no mesh key of the machine's: nothing was let go, nothing is said
	}
	s.mu.Lock()
	said := s.letGo[node]
	s.letGo[node] = true
	s.mu.Unlock()
	if !said {
		s.cfg.Log("let go the forge user %s of before: its mesh keys are removed; the user "+
			"is not deleted and stays in the agents team while its token is minted (the agents of every machine are now %s)", old, forgeaccount.AgentUser)
	}
	return nil
}

// KeyChanged takes in one machine key as the bus keeps it, and tells OnKeys its node.
func (s *Service) KeyChanged(key string, value []byte) { s.SourceKeyChanged(SSHClient, key, value) }

// KeyGone forgets one machine key, and tells OnKeys its node.
func (s *Service) KeyGone(key string) { s.SourceKeyGone(SSHClient, key) }

// SourceKeyChanged takes in one key of the state `keys` of the module source (ssh-client's, or a build module's), and
// tells OnKeys its node.
func (s *Service) SourceKeyChanged(source, key string, value []byte) {
	if node := s.storeFrom(source, key, value); node != "" && s.OnKeys != nil {
		s.OnKeys(node)
	}
}

// SourceKeyGone forgets one key of the module source's state, and tells OnKeys its node.
func (s *Service) SourceKeyGone(source, key string) {
	node, _, _ := strings.Cut(key, ".")
	s.mu.Lock()
	delete(s.keys, storageKey(source, key))
	s.mu.Unlock()
	if s.OnKeys != nil {
		s.OnKeys(node)
	}
}

// store keeps a machine key that reads as one: its node and account the key's own, and a public half that is a
// public key line and nothing private. It answers the node, or "" for a value it did not keep (and drops what was
// kept under that key, since the key now says something else).
func (s *Service) store(key string, value []byte) string { return s.storeFrom(SSHClient, key, value) }

// storeFrom is store for a key of the module source's state.
func (s *Service) storeFrom(source, key string, value []byte) string {
	node, account, _ := strings.Cut(key, ".")
	stored := storageKey(source, key)
	var k machineKey
	var raw struct {
		Node, Account, Kind, Public string
	}
	err := json.Unmarshal(value, &raw)
	k = machineKey{Node: raw.Node, Account: raw.Account, Kind: raw.Kind, Public: raw.Public}
	if source != SSHClient {
		k.Source = source
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if want := kindOf(source); err == nil && k.Kind != want && k.Node == node && k.Account == account {
		// Each state holds one kind: ssh-client's the machines', a build module's the builders' (ADR 0466 §2). The other
		// kind in either is not this entry's to serve, and is said.
		delete(s.keys, stored)
		s.cfg.Log("the key %s in the state keys of %s is of kind %q, not %q; it is ignored", key, source, k.Kind, want)
		return ""
	}
	if err != nil || k.Node != node || k.Account != account || !publicKeyLine(k.Public) {
		if _, had := s.keys[stored]; had {
			delete(s.keys, stored)
		}
		s.cfg.Log("the key %s of %s is not a public key that can be registered; it is not registered", key, source)
		return ""
	}
	s.keys[stored] = k
	return node
}

func publicKeyLine(line string) bool {
	f := strings.Fields(line)
	return len(f) >= 2 && f[0] == "ssh-ed25519" && !strings.Contains(line, "PRIVATE") && !strings.ContainsAny(line, "\r\n")
}

// keysOf are the machine keys one node reported, by key.
func (s *Service) keysOf(node string) []machineKey { return s.keysFrom(SSHClient, node, "machine") }

// keysFrom are the keys of the kind that one node reported in the module source's state, by key.
func (s *Service) keysFrom(source, node, kind string) []machineKey {
	s.mu.Lock()
	defer s.mu.Unlock()
	var names []string
	for key, k := range s.keys {
		if k.Node == node && k.Kind == kind && k.sourceOf() == source && strings.HasPrefix(key, storageKey(source, node+".")) {
			names = append(names, key)
		}
	}
	sort.Strings(names)
	out := make([]machineKey, 0, len(names))
	for _, n := range names {
		out = append(out, s.keys[n])
	}
	return out
}

// WatchKeys keeps the machines' keys as the bus holds them (ssh-client's state `keys`), asking again until the runtime
// answers (the channel to it opens as the bundle starts, and the state of ssh-client may arrive after this one). A
// change after the first values is told to OnKeys; the first values are not, since one pass answers them all.
func (s *Service) WatchKeys(ctx context.Context, src KeySource) error {
	return s.WatchSource(ctx, SSHClient, src)
}

// WatchSource is WatchKeys for the state `keys` of any module whose keys the provider registers: ssh-client's, or a
// build module's (ADR 0466 §3). Once its first values are in, the entries made from them are no longer held back.
func (s *Service) WatchSource(ctx context.Context, source string, src KeySource) error {
	waits := []time.Duration{2 * time.Second, 5 * time.Second, 10 * time.Second, 30 * time.Second}
	for attempt := 0; ; attempt++ {
		err := src.Watch(func(key, op string, value []byte, current bool) {
			switch {
			case op != "put" && current:
			case op != "put":
				s.SourceKeyGone(source, key)
			case current:
				s.storeFrom(source, key, value)
			default:
				s.SourceKeyChanged(source, key, value)
			}
		})
		if err == nil {
			// The watch answers once the current values are all handled (stdio KeptState.Watch).
			s.mu.Lock()
			if source == SSHClient {
				s.loaded = true
			} else {
				s.loadedFrom[source] = true
			}
			s.mu.Unlock()
			return nil
		}
		pause := time.Minute
		if attempt < len(waits) {
			pause = waits[attempt]
		}
		s.cfg.Log("the keys of %s cannot be watched yet (%v); asking again in %s", source, err, pause)
		if ctx.Err() != nil {
			return ctx.Err()
		}
		s.Pause(pause)
	}
}

// hostName is the forge machine's own mesh name, from the fact the mesh writes.
func (s *Service) hostName() (string, error) {
	raw, err := os.ReadFile(s.cfg.HostFile)
	if err != nil {
		return "", fmt.Errorf("the forge machine's mesh name cannot be read (%s): %w", s.cfg.HostFile, err)
	}
	host := strings.TrimSpace(string(raw))
	if host == "" || strings.ContainsAny(host, " \t\n[]:/") {
		return "", fmt.Errorf("%s holds %q, which is not a machine's mesh name", s.cfg.HostFile, host)
	}
	return host, nil
}

// httpsBase is the forge's public https root with a trailing slash.
func (s *Service) httpsBase() (string, error) {
	raw, err := os.ReadFile(s.cfg.OriginsFile)
	if err != nil {
		return "", fmt.Errorf("the forge's origins cannot be read (%s): %w", s.cfg.OriginsFile, err)
	}
	for _, line := range strings.Split(string(raw), "\n") {
		line = strings.TrimRight(strings.TrimSpace(line), "/")
		if strings.HasPrefix(line, "https://") {
			return line + "/", nil
		}
	}
	return "", fmt.Errorf("%s names no https origin", s.cfg.OriginsFile)
}

// scan gives the forge's ed25519 host key as the machines' known_hosts line. The key the forge itself holds is read
// from its file first; only when that cannot be read is the forge's sshd asked over the network, and when that fails
// too (the forge's container is restarting) the key read before is kept, as a host key does not change (issue 1939).
func (s *Service) scan(ctx context.Context) (string, error) {
	host, err := s.hostName()
	if err != nil {
		return "", err
	}
	if s.cfg.HostKeyFile != "" {
		raw, ferr := os.ReadFile(s.cfg.HostKeyFile)
		if ferr == nil {
			if line, perr := knownHostLine("127.0.0.1 "+strings.TrimSpace(string(raw)), host, s.cfg.Port); perr == nil {
				return line, nil
			}
		}
	}
	out, err := s.cfg.Runner.Run(ctx, "ssh-keyscan", "-t", "ed25519", "-p", fmt.Sprint(s.cfg.Port), "127.0.0.1")
	if err != nil {
		s.mu.Lock()
		kept := s.known
		s.mu.Unlock()
		if kept != "" {
			return kept, nil
		}
		return "", fmt.Errorf("the forge's ssh host key cannot be read: %w", err)
	}
	return knownHostLine(string(out), host, s.cfg.Port)
}

// sshAlias is the forge's own name in ssh's configuration (issue 955, ADR 0409): the machines' ssh-client writes a
// Host of this name, so no other Host (the operator's own entry for the forge machine) is merged with it.
const sshAlias = "mesh-forge"

// knownHostLine is the ed25519 line of ssh-keyscan's output with its host written as the mesh name and port. It is
// the old value of `known_host`, kept for machines whose ssh-client has not moved to ssh_alias and host_key yet.
func knownHostLine(scanned, host string, port int) (string, error) {
	for _, line := range strings.Split(scanned, "\n") {
		f := strings.Fields(line)
		if len(f) >= 3 && !strings.HasPrefix(f[0], "#") && f[1] == "ssh-ed25519" {
			return fmt.Sprintf("[%s]:%d %s %s", host, port, f[1], f[2]), nil
		}
	}
	return "", errors.New("the forge's sshd gave no ed25519 host key")
}

// valuesForUser is the forge user a consumer module's entry is served as.
func (s *Service) valuesForUser(node, module string) (string, error) {
	user := forgeaccount.ForgeUser(node, kindOf(module))
	if user == "" {
		return "", fmt.Errorf("%q is not a machine's name an account can be named for", node)
	}
	return user, nil
}

// valuesFor are the values served to one machine's consumer (ADR 0400 §3, §4).
func (s *Service) valuesFor(node, module string) (map[string]any, error) {
	user := forgeaccount.ForgeUser(node, kindOf(module))
	host, err := s.hostName()
	if err != nil {
		return nil, err
	}
	https, err := s.httpsBase()
	if err != nil {
		return nil, err
	}
	s.mu.Lock()
	known := s.known
	s.mu.Unlock()
	if user == "" || known == "" {
		return nil, fmt.Errorf("nothing is known to serve %s yet", node)
	}
	hostKey := strings.Join(strings.Fields(known)[1:], " ")
	return map[string]any{"user": user, "ssh_host": host, "ssh_port": s.cfg.Port, "known_host": known,
		"ssh_alias": sshAlias, "host_key": hostKey,
		// known_host and ssh_base keep their old values: a machine whose ssh-client has not moved yet renders its old
		// templates from them. They are dropped in a later change once every machine uses ssh_alias and host_key (ADR 0409).
		"ssh_base": fmt.Sprintf("ssh://git@%s:%d/", host, s.cfg.Port), "https_base": https}, nil
}

// valuesBus writes each consumer's entry with the values that differ per consumer or come from the machine's own
// facts. The provider helper's offer states what is the same for all; the helper has no place for the rest.
type valuesBus struct {
	provider.Bus
	s *Service

	mu      sync.Mutex
	written map[string][]byte
}

func (b *valuesBus) Put(ctx context.Context, bucket, key string, value []byte) error {
	if bucket != provider.BucketOf(Module) {
		return b.Bus.Put(ctx, bucket, key, value)
	}
	var e provider.Entry
	if err := json.Unmarshal(value, &e); err != nil {
		return err
	}
	if e.Refused == "" {
		extra, err := b.s.valuesFor(e.Consumer, e.Module)
		if err != nil {
			return err
		}
		if e.Values == nil {
			e.Values = map[string]any{}
		}
		for k, v := range extra {
			e.Values[k] = v
		}
	}
	out, err := json.Marshal(e)
	if err != nil {
		return err
	}
	b.mu.Lock()
	same := bytes.Equal(b.written[key], out)
	b.mu.Unlock()
	if same {
		return nil // the helper writes what it composed on every pass; this one stands
	}
	if err := b.Bus.Put(ctx, bucket, key, out); err != nil {
		return err
	}
	b.mu.Lock()
	b.written[key] = out
	b.mu.Unlock()
	return nil
}

func (b *valuesBus) Delete(ctx context.Context, bucket, key string) error {
	b.mu.Lock()
	delete(b.written, key)
	b.mu.Unlock()
	return b.Bus.Delete(ctx, bucket, key)
}

// Backoff of KeepRunning: the first pause and the longest.
const (
	restartFirst = 5 * time.Second
	restartLast  = 2 * time.Minute
)

// KeepRunning runs the SDK provider and starts it again whenever it ends, however it ended (a watch or a read the
// runtime refuses, because a grant has not reached it yet, ends Run): a provider stopped for good answers no consumer
// until something restarts the process, and nothing does when its bundle is unchanged (issue 955). It says so once
// per spell of the same error, pauses 5 s doubling to 2 min between starts, and returns only when ctx ends.
func (s *Service) KeepRunning(ctx context.Context, run func(context.Context) error) {
	pause := restartFirst
	lastSaid := ""
	for ctx.Err() == nil {
		started := time.Now()
		err := run(ctx)
		if ctx.Err() != nil {
			return
		}
		if time.Since(started) >= restartLast {
			// It served a while: a new spell.
			pause, lastSaid = restartFirst, ""
		}
		why := "it returned without an error"
		if err != nil {
			why = err.Error()
		}
		if why != lastSaid {
			s.cfg.Log("the provider stopped: %s; starting it again every %s up to %s until it stays up", why, pause, restartLast)
			lastSaid = why
		}
		s.Pause(pause)
		if pause *= 2; pause > restartLast {
			pause = restartLast
		}
	}
}