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
	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
	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)

	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
}

// 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{}, 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 string }

func (e errNoKey) Error() string {
	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"
}

// 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
	user := forgeaccount.ForgeUser(node, "machine")
	if user == "" {
		return fmt.Errorf("%q is not a machine's name an account can be named for", node)
	}
	s.mu.Lock()
	loaded := s.loaded
	s.mu.Unlock()
	if !loaded {
		// After a restart the first pass can come before the keys arrive: nothing is removed or deleted on that.
		return errors.New("the machines' keys are not read yet, so nothing is made, removed or deleted")
	}
	have := s.keysOf(node)
	if len(have) == 0 {
		// 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, forgeaccount.TitlePrefix+node+"."); 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}
	}
	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.EnsureInTeam(ctx, user); err != nil {
		return err
	}
	var wanted []forgeaccount.WantedKey
	for _, k := range have {
		wanted = append(wanted, forgeaccount.WantedKey{Title: forgeaccount.KeyTitle(k.Node, k.Account), Key: k.Public})
	}
	if err := s.accounts.EnsureKeys(ctx, user, 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 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()
	user := forgeaccount.ForgeUser(e.Consumer, "machine")
	if user == "" {
		return nil
	}
	return s.accounts.RemoveKeysTitled(ctx, user, forgeaccount.TitlePrefix+e.Consumer+".")
}

// KeyChanged takes in one machine key as the bus keeps it, and tells OnKeys its node.
func (s *Service) KeyChanged(key string, value []byte) {
	if node := s.store(key, value); node != "" && s.OnKeys != nil {
		s.OnKeys(node)
	}
}

// KeyGone forgets one machine key, and tells OnKeys its node.
func (s *Service) KeyGone(key string) {
	node, _, _ := strings.Cut(key, ".")
	s.mu.Lock()
	delete(s.keys, 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 {
	node, account, _ := strings.Cut(key, ".")
	var k machineKey
	var raw struct {
		Node, Account, Kind, Public string
	}
	err := json.Unmarshal(value, &raw)
	k = machineKey(raw)
	s.mu.Lock()
	defer s.mu.Unlock()
	if err != nil || k.Node != node || k.Account != account || !publicKeyLine(k.Public) {
		if _, had := s.keys[key]; had {
			delete(s.keys, key)
		}
		s.cfg.Log("the key %s is not a machine's public key; it is not registered", key)
		return ""
	}
	s.keys[key] = 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 {
	s.mu.Lock()
	defer s.mu.Unlock()
	var names []string
	for key, k := range s.keys {
		if k.Node == node && k.Kind == "machine" && strings.HasPrefix(key, 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, 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 {
	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.KeyGone(key)
			case current:
				s.store(key, value)
			default:
				s.KeyChanged(key, value)
			}
		})
		if err == nil {
			// The watch answers once the current values are all handled (stdio KeptState.Watch).
			s.mu.Lock()
			s.loaded = true
			s.mu.Unlock()
			return nil
		}
		pause := time.Minute
		if attempt < len(waits) {
			pause = waits[attempt]
		}
		s.cfg.Log("the machines' keys cannot be watched yet (%v); asking again in %s", 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 reads the forge's ed25519 host key from its own sshd and writes it as the machines' known_hosts line.
func (s *Service) scan(ctx context.Context) (string, error) {
	host, err := s.hostName()
	if err != nil {
		return "", err
	}
	out, err := s.cfg.Runner.Run(ctx, "ssh-keyscan", "-t", "ed25519", "-p", fmt.Sprint(s.cfg.Port), "127.0.0.1")
	if err != 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")
}

// valuesFor are the values served to one machine's consumer (ADR 0400 §3, §4).
func (s *Service) valuesFor(node string) (map[string]any, error) {
	user := forgeaccount.ForgeUser(node, "machine")
	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)
		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
		}
	}
}