feat: lazysecrets list prints every source, vault and secret, for fzf and editor pickers
This commit is contained in:
13 files changed
+886
-91
No files matched your search
@@ -0,0 +1,285 @@
|
||||
// Package listing lists what lazysecrets can find, for lazysecrets list:
|
||||
// what Go to searches, without the TUI. What's cached comes first, at
|
||||
// once; then lists that are missing or older than MaxAge load in the
|
||||
// background, a few at a time, and what's new in them follows as they
|
||||
// arrive. Repos are always read, from disk. Values are never listed.
|
||||
package listing
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.kebcloud.no/keb/lazysecrets/internal/cachefile"
|
||||
"git.kebcloud.no/keb/lazysecrets/internal/provider"
|
||||
"git.kebcloud.no/keb/lazysecrets/internal/ref"
|
||||
)
|
||||
|
||||
// Workers is how many lists load at once, as in Go to.
|
||||
const Workers = 4
|
||||
|
||||
// Entry is a source, vault or item, by its lazysecrets:// link.
|
||||
type Entry struct {
|
||||
Link string `json:"link"`
|
||||
// Kind is source (a subscription, workspace or repo), vault (a Key
|
||||
// Vault, secret scope or repo file), secret, key or certificate.
|
||||
Kind string `json:"kind"`
|
||||
// Provider is azure, databricks or repo, as in links.
|
||||
Provider string `json:"provider"`
|
||||
// Path names it: its source's, vault's and own names.
|
||||
Path []string `json:"path"`
|
||||
// Account is the az account it's seen through, when there are several.
|
||||
Account string `json:"account,omitempty"`
|
||||
}
|
||||
|
||||
type Options struct {
|
||||
Providers []provider.Provider
|
||||
// Cache is what was loaded before; Run adds what it loads.
|
||||
Cache cachefile.File
|
||||
// MaxAge: cached lists loaded longer ago are loaded again.
|
||||
MaxAge time.Duration
|
||||
// Cached lists only what's cached (and repos): nothing is loaded.
|
||||
Cached bool
|
||||
Timeout func(provider.Kind) time.Duration
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// Run lists everything, calling emit for each entry once, from this
|
||||
// goroutine. An error from emit (the reader went away) stops it. It
|
||||
// returns the cache with what was loaded, and the lists that couldn't be.
|
||||
func Run(ctx context.Context, o Options, emit func(Entry) error) (cachefile.File, []error) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
l := &lister{o: o, emit: emit, seen: map[string]bool{}, pending: map[string]bool{}, results: make(chan result)}
|
||||
if l.o.Now == nil {
|
||||
l.o.Now = time.Now
|
||||
}
|
||||
for _, p := range o.Providers {
|
||||
if l.stopped != nil {
|
||||
break
|
||||
}
|
||||
l.provider(p)
|
||||
}
|
||||
l.start(ctx)
|
||||
for l.active > 0 {
|
||||
r := <-l.results
|
||||
l.active--
|
||||
if l.stopped == nil {
|
||||
l.handle(r)
|
||||
l.start(ctx)
|
||||
}
|
||||
if l.stopped != nil {
|
||||
cancel()
|
||||
}
|
||||
}
|
||||
return l.o.Cache, l.errs
|
||||
}
|
||||
|
||||
type job struct {
|
||||
p provider.Provider
|
||||
src *provider.Source // list its vaults
|
||||
c *provider.Container // list its secrets
|
||||
key string // where the list is cached
|
||||
}
|
||||
|
||||
type result struct {
|
||||
job
|
||||
sources []provider.Source
|
||||
containers []provider.Container
|
||||
objects []provider.Object
|
||||
err error
|
||||
}
|
||||
|
||||
type lister struct {
|
||||
o Options
|
||||
emit func(Entry) error
|
||||
seen map[string]bool // provider ID + link of entries emitted
|
||||
pending map[string]bool // cache keys queued or loading
|
||||
queue []job
|
||||
active int
|
||||
results chan result
|
||||
errs []error
|
||||
stopped error // emit's error
|
||||
}
|
||||
|
||||
func (l *lister) cacheable(p provider.Provider) bool { return p.Kind() != provider.KindRepo }
|
||||
|
||||
// load reports whether to load a list: a repo's always (from disk), else
|
||||
// unless Cached, when it isn't cached or is older than MaxAge.
|
||||
func (l *lister) load(p provider.Provider, cached bool, at time.Time) bool {
|
||||
if !l.cacheable(p) {
|
||||
return true
|
||||
}
|
||||
return !l.o.Cached && (!cached || l.o.Now().Sub(at) >= l.o.MaxAge)
|
||||
}
|
||||
|
||||
func (l *lister) provider(p provider.Provider) {
|
||||
e, ok := l.o.Cache.Sources[p.ID()]
|
||||
if l.cacheable(p) && ok {
|
||||
for _, src := range sorted(e.Value, func(s provider.Source) string { return s.Name }) {
|
||||
l.source(p, src)
|
||||
}
|
||||
}
|
||||
if l.load(p, ok, e.At) {
|
||||
l.enqueue(job{p: p, key: p.ID()})
|
||||
}
|
||||
}
|
||||
|
||||
// source emits src, and what's cached under it, and queues what's not.
|
||||
func (l *lister) source(p provider.Provider, src provider.Source) {
|
||||
src.Provider, src.ProviderLabel = p.ID(), p.Label()
|
||||
if r, err := ref.ForSource(src); err == nil {
|
||||
l.put(p, Entry{Link: r.String(), Kind: "source", Provider: kindOf(src.Kind), Path: []string{src.Name}, Account: accountOf(src)})
|
||||
}
|
||||
row := cachefile.Row(p.ID(), src.ID)
|
||||
e, ok := l.o.Cache.Containers[row]
|
||||
if l.cacheable(p) && ok {
|
||||
l.containers(p, src, e.Value)
|
||||
}
|
||||
if l.load(p, ok, e.At) {
|
||||
l.enqueue(job{p: p, src: &src, key: row})
|
||||
}
|
||||
}
|
||||
|
||||
func (l *lister) containers(p provider.Provider, src provider.Source, cs []provider.Container) {
|
||||
for _, c := range sorted(cs, func(c provider.Container) string { return c.Name }) {
|
||||
c.Source = src
|
||||
if r, err := ref.ForContainer(c, provider.TypeSecret); err == nil {
|
||||
l.put(p, Entry{Link: r.String(), Kind: "vault", Provider: kindOf(src.Kind), Path: []string{src.Name, c.Name}, Account: accountOf(src)})
|
||||
}
|
||||
key := cachefile.ObjectsKey(c, provider.TypeSecret)
|
||||
e, ok := l.o.Cache.Objects[key]
|
||||
if l.cacheable(p) {
|
||||
// Keys and certificates only as far as they were opened in
|
||||
// the TUI: Go to doesn't load them either.
|
||||
for _, t := range []provider.ObjectType{provider.TypeSecret, provider.TypeKey, provider.TypeCertificate} {
|
||||
l.objects(p, c, l.o.Cache.Objects[cachefile.ObjectsKey(c, t)].Value)
|
||||
}
|
||||
}
|
||||
if l.load(p, ok, e.At) {
|
||||
l.enqueue(job{p: p, c: &c, key: key})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (l *lister) objects(p provider.Provider, c provider.Container, objs []provider.Object) {
|
||||
for _, o := range sorted(objs, func(o provider.Object) string { return o.Name }) {
|
||||
o.Container = c
|
||||
r, err := ref.ForObject(o, "")
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
l.put(p, Entry{Link: r.String(), Kind: strings.ToLower(string(o.Type)), Provider: kindOf(c.Source.Kind),
|
||||
Path: []string{c.Source.Name, c.Name, o.Name}, Account: accountOf(c.Source)})
|
||||
}
|
||||
}
|
||||
|
||||
// put emits e, unless it was already (or emitting has failed).
|
||||
func (l *lister) put(p provider.Provider, e Entry) {
|
||||
id := p.ID() + "\x00" + e.Link
|
||||
if l.stopped != nil || l.seen[id] {
|
||||
return
|
||||
}
|
||||
l.seen[id] = true
|
||||
l.stopped = l.emit(e)
|
||||
}
|
||||
|
||||
func (l *lister) enqueue(j job) {
|
||||
if l.pending[j.key] {
|
||||
return
|
||||
}
|
||||
l.pending[j.key] = true
|
||||
l.queue = append(l.queue, j)
|
||||
}
|
||||
|
||||
func (l *lister) start(ctx context.Context) {
|
||||
for l.stopped == nil && l.active < Workers && len(l.queue) > 0 {
|
||||
j := l.queue[0]
|
||||
l.queue = l.queue[1:]
|
||||
l.active++
|
||||
go func() { l.results <- l.fetch(ctx, j) }()
|
||||
}
|
||||
}
|
||||
|
||||
func (l *lister) fetch(ctx context.Context, j job) result {
|
||||
kind := j.p.Kind()
|
||||
ctx, cancel := context.WithTimeout(ctx, l.o.Timeout(kind))
|
||||
defer cancel()
|
||||
r := result{job: j}
|
||||
switch {
|
||||
case j.c != nil:
|
||||
r.objects, r.err = j.p.Objects(ctx, *j.c, provider.TypeSecret)
|
||||
case j.src != nil:
|
||||
r.containers, r.err = j.p.Containers(ctx, *j.src)
|
||||
default:
|
||||
r.sources, r.err = j.p.Sources(ctx)
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func (l *lister) handle(r result) {
|
||||
delete(l.pending, r.key)
|
||||
now := l.o.Now()
|
||||
if r.err != nil {
|
||||
l.errs = append(l.errs, fmt.Errorf("%s: %w", r.label(), r.err))
|
||||
return
|
||||
}
|
||||
cache := l.cacheable(r.p)
|
||||
switch {
|
||||
case r.c != nil:
|
||||
if cache {
|
||||
l.o.Cache.Objects[r.key] = cachefile.Stamped[[]provider.Object]{At: now, Value: r.objects}
|
||||
}
|
||||
l.objects(r.p, *r.c, r.objects)
|
||||
case r.src != nil:
|
||||
r.containers = slices.Clone(r.containers) // the provider's may be shared
|
||||
for i := range r.containers {
|
||||
r.containers[i].Source = *r.src
|
||||
}
|
||||
if cache {
|
||||
l.o.Cache.Containers[r.key] = cachefile.Stamped[[]provider.Container]{At: now, Value: r.containers}
|
||||
}
|
||||
l.containers(r.p, *r.src, r.containers)
|
||||
default:
|
||||
if cache {
|
||||
l.o.Cache.Sources[r.key] = cachefile.Stamped[[]provider.Source]{At: now, Value: r.sources}
|
||||
}
|
||||
for _, src := range r.sources {
|
||||
l.source(r.p, src)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// label names a list that couldn't be loaded.
|
||||
func (r result) label() string {
|
||||
switch {
|
||||
case r.c != nil:
|
||||
return r.c.Source.Name + " / " + r.c.Name
|
||||
case r.src != nil:
|
||||
return r.src.Name
|
||||
}
|
||||
return r.p.Label()
|
||||
}
|
||||
|
||||
func kindOf(k provider.Kind) string { return strings.ToLower(string(k)) }
|
||||
|
||||
// accountOf is the account part of a source's provider label ("alice" in
|
||||
// "Azure · alice"), or "" when there's a single account.
|
||||
func accountOf(src provider.Source) string {
|
||||
account, ok := strings.CutPrefix(src.ProviderLabel, string(src.Kind)+" · ")
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
return account
|
||||
}
|
||||
|
||||
func sorted[T any](s []T, name func(T) string) []T {
|
||||
s = slices.Clone(s)
|
||||
slices.SortStableFunc(s, func(a, b T) int {
|
||||
return cmp.Compare(strings.ToLower(name(a)), strings.ToLower(name(b)))
|
||||
})
|
||||
return s
|
||||
}
|
||||
@@ -0,0 +1,176 @@
|
||||
package listing
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.kebcloud.no/keb/lazysecrets/internal/cachefile"
|
||||
"git.kebcloud.no/keb/lazysecrets/internal/provider"
|
||||
"git.kebcloud.no/keb/lazysecrets/internal/provider/fake"
|
||||
)
|
||||
|
||||
// counting counts a provider's requests.
|
||||
type counting struct {
|
||||
provider.Provider
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (c *counting) Sources(ctx context.Context) ([]provider.Source, error) {
|
||||
c.calls.Add(1)
|
||||
return c.Provider.Sources(ctx)
|
||||
}
|
||||
|
||||
func (c *counting) Containers(ctx context.Context, src provider.Source) ([]provider.Container, error) {
|
||||
c.calls.Add(1)
|
||||
return c.Provider.Containers(ctx, src)
|
||||
}
|
||||
|
||||
func (c *counting) Objects(ctx context.Context, ct provider.Container, t provider.ObjectType) ([]provider.Object, error) {
|
||||
c.calls.Add(1)
|
||||
return c.Provider.Objects(ctx, ct, t)
|
||||
}
|
||||
|
||||
func run(t *testing.T, o Options) ([]Entry, cachefile.File, []error) {
|
||||
t.Helper()
|
||||
if o.Cache.Version == 0 {
|
||||
o.Cache = cachefile.New()
|
||||
}
|
||||
o.Timeout = func(provider.Kind) time.Duration { return 5 * time.Second }
|
||||
var got []Entry
|
||||
f, errs := Run(context.Background(), o, func(e Entry) error {
|
||||
got = append(got, e)
|
||||
return nil
|
||||
})
|
||||
return got, f, errs
|
||||
}
|
||||
|
||||
func has(es []Entry, kind, path string) bool {
|
||||
for _, e := range es {
|
||||
if e.Kind == kind && strings.Join(e.Path, "/") == path {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func TestListLoadsAndCaches(t *testing.T) {
|
||||
t.Parallel()
|
||||
az := &counting{Provider: fake.Azure(0)}
|
||||
got, f, errs := run(t, Options{Providers: []provider.Provider{az, fake.Repos()}, MaxAge: time.Minute})
|
||||
for _, want := range [][2]string{
|
||||
{"source", "platform-prod"}, {"vault", "platform-prod/kv-app-prod"}, {"secret", "platform-prod/kv-app-prod/db-password"},
|
||||
{"source", "lazysecrets-demo"}, {"secret", "lazysecrets-demo/.env/DATABASE_URL"},
|
||||
} {
|
||||
if !has(got, want[0], want[1]) {
|
||||
t.Errorf("no %s %s", want[0], want[1])
|
||||
}
|
||||
}
|
||||
if len(errs) != 1 || !strings.Contains(errs[0].Error(), "kv-restricted: 403") {
|
||||
t.Errorf("errs = %v", errs)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, e := range got {
|
||||
if seen[e.Link] {
|
||||
t.Errorf("%s listed twice", e.Link)
|
||||
}
|
||||
seen[e.Link] = true
|
||||
if e.Provider != "azure" && e.Provider != "repo" {
|
||||
t.Errorf("provider %q", e.Provider)
|
||||
}
|
||||
}
|
||||
if len(f.Sources) != 1 || len(f.Containers) == 0 || len(f.Objects) == 0 {
|
||||
t.Errorf("not cached: %d sources, %d containers, %d objects", len(f.Sources), len(f.Containers), len(f.Objects))
|
||||
}
|
||||
for k := range f.Containers {
|
||||
if cachefile.ProviderOf(k) != az.ID() {
|
||||
t.Errorf("cached a repo: %s", k)
|
||||
}
|
||||
}
|
||||
|
||||
// Again with that cache: the same, without asking Azure.
|
||||
before := az.calls.Load()
|
||||
again, _, _ := run(t, Options{Providers: []provider.Provider{az, fake.Repos()}, Cache: f, MaxAge: time.Minute})
|
||||
if len(again) != len(got) {
|
||||
t.Errorf("%d entries from the cache, %d loaded", len(again), len(got))
|
||||
}
|
||||
// The vault that couldn't be read wasn't cached: it's asked again.
|
||||
if n := az.calls.Load() - before; n != 1 {
|
||||
t.Errorf("%d requests with a fresh cache, want 1 (kv-restricted)", n)
|
||||
}
|
||||
|
||||
// Stale: loaded again, but each entry is still listed once.
|
||||
before = az.calls.Load()
|
||||
stale, _, _ := run(t, Options{Providers: []provider.Provider{az}, Cache: f, MaxAge: time.Minute,
|
||||
Now: func() time.Time { return time.Now().Add(time.Hour) }})
|
||||
if az.calls.Load()-before < 3 {
|
||||
t.Error("stale lists weren't loaded again")
|
||||
}
|
||||
links := map[string]bool{}
|
||||
for _, e := range stale {
|
||||
if links[e.Link] {
|
||||
t.Errorf("%s listed twice", e.Link)
|
||||
}
|
||||
links[e.Link] = true
|
||||
}
|
||||
}
|
||||
|
||||
func TestListCachedOnly(t *testing.T) {
|
||||
t.Parallel()
|
||||
az := &counting{Provider: fake.Azure(0)}
|
||||
_, f, _ := run(t, Options{Providers: []provider.Provider{az}, MaxAge: time.Minute})
|
||||
before := az.calls.Load()
|
||||
got, _, errs := run(t, Options{Providers: []provider.Provider{az, fake.Repos()}, Cache: f, Cached: true,
|
||||
Now: func() time.Time { return time.Now().Add(24 * time.Hour) }})
|
||||
if az.calls.Load() != before || len(errs) != 0 {
|
||||
t.Errorf("--cached asked Azure: %d requests, %v", az.calls.Load()-before, errs)
|
||||
}
|
||||
if !has(got, "secret", "platform-prod/kv-app-prod/db-password") || !has(got, "secret", "lazysecrets-demo/.env/DATABASE_URL") {
|
||||
t.Error("cached secrets or repos missing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestListCachedFirst(t *testing.T) {
|
||||
t.Parallel()
|
||||
az := fake.Azure(0)
|
||||
_, f, _ := run(t, Options{Providers: []provider.Provider{az}, MaxAge: time.Minute})
|
||||
// Forget one vault's secrets: they load, after everything cached.
|
||||
var dropped string
|
||||
for k := range f.Objects {
|
||||
if strings.Contains(k, "kv-app-prod") && strings.HasSuffix(k, "|Secret") {
|
||||
dropped = k
|
||||
delete(f.Objects, k)
|
||||
break
|
||||
}
|
||||
}
|
||||
if dropped == "" {
|
||||
t.Fatalf("no kv-app-prod in %v", f.Objects)
|
||||
}
|
||||
got, _, _ := run(t, Options{Providers: []provider.Provider{az}, Cache: f, MaxAge: time.Minute})
|
||||
last := got[len(got)-1]
|
||||
if last.Kind != "secret" || last.Path[1] != "kv-app-prod" {
|
||||
t.Errorf("last = %+v, want a kv-app-prod secret, loaded after the cache", last)
|
||||
}
|
||||
}
|
||||
|
||||
func TestListStopsWhenTheReaderGoes(t *testing.T) {
|
||||
t.Parallel()
|
||||
gone := errors.New("broken pipe")
|
||||
n := 0
|
||||
_, errs := Run(context.Background(), Options{
|
||||
Providers: []provider.Provider{fake.Azure(0), fake.Repos()}, Cache: cachefile.New(), MaxAge: time.Minute,
|
||||
Timeout: func(provider.Kind) time.Duration { return time.Second },
|
||||
}, func(Entry) error {
|
||||
n++
|
||||
if n == 2 {
|
||||
return gone
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if n != 2 || len(errs) != 0 {
|
||||
t.Errorf("emitted %d after the reader went, errs %v", n, errs)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user