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 // 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. return fmt.Errorf("the keys of %s are not read yet, so nothing is made, removed or deleted", source) } 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- 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-) 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 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") } // 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 } } }