package openbao import ( "cmp" "context" "encoding/json" "errors" "fmt" "log/slog" "net/url" "strings" "sync" "time" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/atprotorepo" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/guestcp" "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/host" "github.com/johnandersen777/socialweb-computer-kcp/internal/guestidentity" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt" ) const ( DefaultVaultClient = "guest-kcp" DefaultTimeout = 10 * time.Second DefaultIssuerRetry = 30 * time.Second hostKeyProbe = time.Minute ) type Config struct { Handle string DID string Mount string VaultClient string Timeout time.Duration GrantSync time.Duration IssuerRetry time.Duration HostKeyAttempts int } type Deps struct { Inputs guestcp.Source Guests guestcp.Guests Grants guestcp.Grants RBAC atprotorepo.RecordReader FallbackIssuer func() string Publish guestcp.Publish Clock host.Clock Log *slog.Logger } type Plane struct { cfg Config deps Deps log *slog.Logger clock host.Clock mu sync.RWMutex inputs guestcp.Inputs probing map[string]bool retrying bool kick chan struct{} issuerKick chan struct{} identity *guestidentity.Client verifier *guestidentity.Verifier guest *guestidentity.Guest syncer *guestidentity.Syncer known map[string]string cancel context.CancelFunc unwatch func() wg sync.WaitGroup once sync.Once } var _ guestcp.Plane = (*Plane)(nil) func New(cfg Config, deps Deps) (*Plane, error) { switch { case strings.TrimSpace(cfg.Handle) == "": return nil, errors.New("guestcp: the guest control plane names no bidder handle") case deps.Inputs == nil: return nil, errors.New("guestcp: the guest control plane has no input source") case deps.Guests == nil: return nil, errors.New("guestcp: the guest control plane has no guest report writer") case deps.Grants == nil: return nil, errors.New("guestcp: the guest control plane has no grant writer") case deps.RBAC == nil: return nil, errors.New("guestcp: the guest control plane has no RBAC record reader") } cfg.VaultClient = cmp.Or(strings.TrimSpace(cfg.VaultClient), DefaultVaultClient) cfg.Timeout = opt.Positive(cfg.Timeout, DefaultTimeout) cfg.IssuerRetry = opt.Positive(cfg.IssuerRetry, DefaultIssuerRetry) log := deps.Log if log == nil { log = slog.New(slog.DiscardHandler) } clock := deps.Clock if clock == nil { clock = systemClock{} } return &Plane{ cfg: cfg, deps: deps, log: log, clock: clock, probing: map[string]bool{}, known: map[string]string{}, kick: make(chan struct{}, 1), issuerKick: make(chan struct{}, 1), }, nil } func (p *Plane) Start(ctx context.Context) error { initial, err := p.deps.Inputs.Inputs(ctx) if err != nil { return fmt.Errorf("guestcp: reading the guest control plane inputs: %w", err) } p.update(initial) runCtx, cancel := context.WithCancel(ctx) p.cancel = cancel identity, err := guestidentity.NewClient(guestidentity.Options{Mount: p.cfg.Mount, Endpoint: p.endpoint, Timeout: p.cfg.Timeout}) if err != nil { p.Close() return err } p.identity = identity issuer := func() string { return guestidentity.OIDCIssuer(p.IssuerURL()) } p.verifier = guestidentity.NewVerifier(guestidentity.IdentityKeys(identity), issuer) syncer, err := guestidentity.NewSyncer(runCtx, guestidentity.SyncerOptions{ Store: grantStore{p: p}, RBAC: &guestidentity.RepoRBAC{Reader: p.deps.RBAC, Log: p.log}, Issuer: issuer, Log: p.log, Resync: p.cfg.GrantSync, }) if err != nil { p.Close() return err } p.syncer = syncer guest, err := guestidentity.NewGuest(runCtx, guestidentity.GuestOptions{ Handle: p.cfg.Handle, Identity: identity, Verifier: p.verifier, VMs: vmStore{p: p}, Log: p.log, Now: p.clock.Now, HostKeyAttempts: p.cfg.HostKeyAttempts, Publish: p.publish, }) if err != nil { p.Close() return err } p.guest = guest p.unwatch = p.deps.Inputs.Watch(p.update) p.wg.Go(func() { p.reconcileLoop(runCtx) }) p.wg.Go(func() { p.issuerLoop(runCtx) }) return nil } func (p *Plane) Close() { p.once.Do(func() { if p.unwatch != nil { p.unwatch() } if p.cancel != nil { p.cancel() } if p.guest != nil { p.guest.Close() } if p.syncer != nil { p.syncer.Close() } p.wg.Wait() }) } func (p *Plane) Ready() error { if p == nil || p.guest == nil { return errors.New("guestcp: the guest identity proxy is not configured") } return nil } func (p *Plane) DID() string { p.mu.RLock() defer p.mu.RUnlock() return cmp.Or(strings.TrimSpace(p.cfg.DID), strings.TrimSpace(p.inputs.DID)) } func (p *Plane) IssuerURL() string { p.mu.RLock() raw := exposureIssuer(p.inputs.Issuer) p.mu.RUnlock() if raw == "" && p.deps.FallbackIssuer != nil { raw = strings.TrimSpace(p.deps.FallbackIssuer()) } return strings.TrimRight(raw, "/") } func (p *Plane) OIDCIssuer() string { raw := p.IssuerURL() if raw == "" { return "" } return guestidentity.OIDCIssuer(raw) } func (p *Plane) Recheck() { nudge(p.issuerKick) } func (p *Plane) LoginStep(ctx context.Context, step string, input map[string]any) guestcp.Reply { return guestcp.Reply(p.guest.LoginStep(ctx, step, input)) } func (p *Plane) OIDCToken(ctx context.Context, role, vaultToken string) guestcp.Reply { return guestcp.Reply(p.guest.OIDCToken(ctx, role, vaultToken)) } func (p *Plane) Discovery(ctx context.Context) guestcp.Reply { return guestcp.Reply(p.guest.Discovery(ctx)) } func (p *Plane) Keys(ctx context.Context) guestcp.Reply { return guestcp.Reply(p.guest.Keys(ctx)) } func (p *Plane) Events(ctx context.Context, authorization string, request guestcp.EventsRequest) guestcp.Reply { reply := p.guest.Events(ctx, authorization, guestidentity.EventsRequest(request)) return guestcp.Reply(p.withGuestLogin(reply)) } func (p *Plane) publish(ctx context.Context, event guestidentity.GuestEvent) error { if p.deps.Publish == nil { return guestcp.ErrNoPublisher } record, err := json.Marshal(event.Record) if err != nil { return err } return p.deps.Publish(ctx, guestcp.GuestEvent{ComputeVM: event.ComputeVM, Collection: event.Collection, Tag: event.Tag, Kind: event.Kind, Record: record}) } func (p *Plane) Authenticate(ctx context.Context, authorization string) (string, error) { name, err := p.guest.Authenticate(ctx, authorization) switch { case errors.Is(err, guestidentity.ErrUnauthenticated): return "", fmt.Errorf("%w: %v", guestcp.ErrUnauthenticated, err) case errors.Is(err, guestidentity.ErrUnknownGuest): return "", fmt.Errorf("%w: %v", guestcp.ErrUnknownGuest, err) } return name, err } func (p *Plane) update(in guestcp.Inputs) { p.mu.Lock() p.inputs = in p.mu.Unlock() nudge(p.kick) nudge(p.issuerKick) } func (p *Plane) current() guestcp.Inputs { p.mu.RLock() defer p.mu.RUnlock() return p.inputs } func (p *Plane) vm(name string) *v1alpha1.ComputeVM { p.mu.RLock() defer p.mu.RUnlock() for i := range p.inputs.ComputeVMs { if p.inputs.ComputeVMs[i].Name == name { return p.inputs.ComputeVMs[i].DeepCopy() } } return nil } func (p *Plane) grants() []v1alpha1.GuestIdentityGrant { p.mu.RLock() defer p.mu.RUnlock() out := make([]v1alpha1.GuestIdentityGrant, 0, len(p.inputs.Grants)) for i := range p.inputs.Grants { out = append(out, *p.inputs.Grants[i].DeepCopy()) } return out } func (p *Plane) endpoint(context.Context) (guestidentity.Endpoint, error) { p.mu.RLock() status := p.inputs.OpenBao p.mu.RUnlock() if status == nil { return guestidentity.Endpoint{}, errors.New("guestcp: the OpenBaoServer reports no status yet") } return guestidentity.EndpointFromServer(*status, p.cfg.VaultClient) } func exposureIssuer(raw string) string { raw = strings.TrimRight(strings.TrimSpace(raw), "/") parsed, err := url.Parse(raw) if err != nil || parsed.Host == "" || (parsed.Scheme != "https" && parsed.Scheme != "http") { return "" } return raw } func nudge(ch chan struct{}) { select { case ch <- struct{}{}: default: } } type systemClock struct{} func (systemClock) Now() time.Time { return time.Now() } func (systemClock) After(d time.Duration) <-chan time.Time { return time.After(d) }