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
	known string // the forge's host key line, as the machines write it into known_hosts

	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 {
	return provider.New(provider.Options{
		Node: s.cfg.Node, Module: Module,
		Offers: map[string]provider.Offer{ProvisionName: {Credential: provider.NoCredential}},
		Bus:    &valuesBus{Bus: bus, s: s, written: map[string][]byte{}},
		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)
	}
	have := s.keysOf(node)
	if len(have) == 0 {
		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 {
			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)
}

// knownHostLine is the ed25519 line of ssh-keyscan's output with its host written as the mesh name and port.
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)
	}
	return map[string]any{"user": user, "ssh_host": host, "ssh_port": s.cfg.Port, "known_host": known,
		"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)
}