johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package guestidentity

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sync"
	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"

	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/opt"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
)

const (
	DefaultResync = 30 * time.Second

	DefaultRetryDelay = 5 * time.Second

	DefaultMaxRetry = 6
)

type GrantStore interface {
	List(ctx context.Context) ([]v1alpha1.GuestIdentityGrant, error)

	Apply(ctx context.Context, grant v1alpha1.GuestIdentityGrant) error

	Delete(ctx context.Context, name string) error
}

type GrantStoreFuncs struct {
	ListFunc func(context.Context) ([]v1alpha1.GuestIdentityGrant, error)

	ApplyFunc func(context.Context, v1alpha1.GuestIdentityGrant) error

	DeleteFunc func(context.Context, string) error
}

func (f GrantStoreFuncs) List(ctx context.Context) ([]v1alpha1.GuestIdentityGrant, error) {
	return f.ListFunc(ctx)
}

func (f GrantStoreFuncs) Apply(ctx context.Context, grant v1alpha1.GuestIdentityGrant) error {
	return f.ApplyFunc(ctx, grant)
}

func (f GrantStoreFuncs) Delete(ctx context.Context, name string) error {
	return f.DeleteFunc(ctx, name)
}

type RBACSource interface {
	Records(ctx context.Context, requesterDID, acceptURI string, acceptedAt time.Time) ([]RBACRecord, error)
}

type RBACSourceFunc func(context.Context, string, string, time.Time) ([]RBACRecord, error)

func (f RBACSourceFunc) Records(ctx context.Context, requesterDID, acceptURI string, acceptedAt time.Time) ([]RBACRecord, error) {
	return f(ctx, requesterDID, acceptURI, acceptedAt)
}

type GrantSubject struct {
	ComputeVM string

	OwnerUID string

	RequesterDID string

	AcceptURI string

	AcceptedAt time.Time

	Subject string
}

func GrantFor(subject GrantSubject, wish GrantWish) v1alpha1.GuestIdentityGrant {
	scopes := append([]string(nil), wish.Scopes...)
	claims := append([]v1alpha1.GuestIdentityClaim(nil), wish.Claims...)
	g := v1alpha1.GuestIdentityGrant{
		TypeMeta: metav1.TypeMeta{APIVersion: v1alpha1.GroupVersion.String(), Kind: "GuestIdentityGrant"},
		ObjectMeta: metav1.ObjectMeta{
			Name: GrantName(subject.ComputeVM, wish),
		},
		Spec: v1alpha1.GuestIdentityGrantSpec{
			ComputeVMRef: commonv1.LocalObjectRef{Name: subject.ComputeVM},
			Entity:       Entity(subject.ComputeVM),
			Api:          wish.API,
			Actx:         wish.Actx,
			Scopes:       scopes,
			Claims:       claims,
		},
	}
	if subject.OwnerUID != "" {
		block := true
		g.OwnerReferences = []metav1.OwnerReference{{
			APIVersion:         v1alpha1.GroupVersion.String(),
			Kind:               "ComputeVM",
			Name:               subject.ComputeVM,
			UID:                types.UID(subject.OwnerUID),
			BlockOwnerDeletion: &block,
		}}
	}
	return g
}

type SyncerOptions struct {
	Store GrantStore

	RBAC RBACSource

	Issuer func() string

	Log *slog.Logger

	Resync time.Duration

	RetryDelay time.Duration

	MaxRetry int
}

type Syncer struct {
	opts   SyncerOptions
	ctx    context.Context
	cancel context.CancelFunc
	wg     sync.WaitGroup

	mu     sync.Mutex
	guests map[string]*guestSync
	closed bool

	rbacWarned map[string]rbacWarning
}

type rbacWarning struct {
	at time.Time

	suppressed int
}

const rbacWarnEvery = 10 * time.Minute

type guestSync struct {
	subject GrantSubject
	ctx     context.Context
	cancel  context.CancelFunc
	kick    chan struct{}
}

func NewSyncer(ctx context.Context, opts SyncerOptions) (*Syncer, error) {
	if opts.Store == nil {
		return nil, errors.New("guestidentity: the grant syncer has no store")
	}
	if opts.RBAC == nil {
		return nil, errors.New("guestidentity: the grant syncer has no RBAC source")
	}
	if opts.Log == nil {
		opts.Log = slog.New(slog.DiscardHandler)
	}
	opts.Resync = opt.Positive(opts.Resync, DefaultResync)
	opts.RetryDelay = opt.Positive(opts.RetryDelay, DefaultRetryDelay)
	opts.MaxRetry = opt.Positive(opts.MaxRetry, DefaultMaxRetry)
	root, cancel := context.WithCancel(ctx)
	return &Syncer{opts: opts, ctx: root, cancel: cancel, guests: map[string]*guestSync{}, rbacWarned: map[string]rbacWarning{}}, nil
}

func (s *Syncer) Register(ctx context.Context, subject GrantSubject) error {
	if err := checkSubject(subject); err != nil {
		return err
	}
	if err := s.syncOnce(ctx, subject); err != nil {
		return err
	}
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return errors.New("guestidentity: the grant syncer is closed")
	}
	if old, ok := s.guests[subject.ComputeVM]; ok {
		old.cancel()
	}
	stateCtx, cancel := context.WithCancel(s.ctx)
	state := &guestSync{subject: subject, ctx: stateCtx, cancel: cancel, kick: make(chan struct{}, 1)}
	s.guests[subject.ComputeVM] = state
	s.wg.Go(func() { s.run(state) })
	s.mu.Unlock()
	return nil
}

func (s *Syncer) Forget(computevm string) {
	s.mu.Lock()
	state, ok := s.guests[computevm]
	if ok {
		delete(s.guests, computevm)
	}
	s.mu.Unlock()
	if ok {
		state.cancel()
	}
}

func (s *Syncer) RequesterRBACChanged(requesterDID string) {
	s.mu.Lock()
	states := make([]*guestSync, 0, len(s.guests))
	for _, state := range s.guests {
		if state.subject.RequesterDID == requesterDID {
			states = append(states, state)
		}
	}
	s.mu.Unlock()
	for _, state := range states {
		select {
		case state.kick <- struct{}{}:
		default:
		}
	}
}

func (s *Syncer) Close() {
	s.mu.Lock()
	if s.closed {
		s.mu.Unlock()
		return
	}
	s.closed = true
	states := make([]*guestSync, 0, len(s.guests))
	for _, state := range s.guests {
		states = append(states, state)
	}
	s.guests = map[string]*guestSync{}
	s.mu.Unlock()
	s.cancel()
	for _, state := range states {
		state.cancel()
	}
	s.wg.Wait()
}

func (s *Syncer) run(state *guestSync) {
	failures := 0
	for {
		delay := s.opts.Resync
		if failures > 0 {
			delay = s.opts.RetryDelay
			for i := 1; i < failures && i < s.opts.MaxRetry; i++ {
				delay *= 2
			}
			if delay > s.opts.Resync {
				delay = s.opts.Resync
			}
		}
		timer := time.NewTimer(delay)
		select {
		case <-state.ctx.Done():
			timer.Stop()
			return
		case <-state.kick:
			timer.Stop()
		case <-timer.C:
		}
		if state.ctx.Err() != nil {
			return
		}
		if err := s.syncOnce(state.ctx, state.subject); err != nil {
			if state.ctx.Err() != nil {
				return
			}
			failures++
			s.opts.Log.Warn("guest identity grants did not sync", "computevm", state.subject.ComputeVM, "err", err.Error())
			continue
		}
		failures = 0
	}
}

func (s *Syncer) syncOnce(ctx context.Context, subject GrantSubject) error {
	issuer := ""
	if s.opts.Issuer != nil {
		issuer = s.opts.Issuer()
	}
	records, readable, err := s.records(ctx, subject.RequesterDID, subject.AcceptURI, subject.AcceptedAt)
	if err != nil {
		if suppressed, due := s.rbacWarnDue(subject.RequesterDID); due {
			s.opts.Log.Warn("the requester's RBAC records are unreadable; the grants it already holds stay",
				"computevm", subject.ComputeVM, "requester", subject.RequesterDID, "err", err.Error(), "suppressed", suppressed)
		}
	} else {
		s.mu.Lock()
		delete(s.rbacWarned, subject.RequesterDID)
		s.mu.Unlock()
	}
	wishes := Wishes(subject.ComputeVM, records, issuer, subject.Subject)
	held, err := s.opts.Store.List(ctx)
	if err != nil {
		return fmt.Errorf("listing the GuestIdentityGrants: %w", err)
	}
	current := map[string]v1alpha1.GuestIdentityGrant{}
	for _, g := range held {
		if g.Spec.ComputeVMRef.Name == subject.ComputeVM && g.DeletionTimestamp == nil {
			current[g.Name] = g
		}
	}
	wanted := map[string]bool{}
	for _, wish := range wishes {
		name := GrantName(subject.ComputeVM, wish)
		wanted[name] = true
		if g, ok := current[name]; ok && HeldKey(g) == WishKey(wish) {
			continue
		}
		if err := s.opts.Store.Apply(ctx, GrantFor(subject, wish)); err != nil {
			return fmt.Errorf("applying the grant for %s: %w", wish.Audience(), err)
		}
		s.opts.Log.Info("a guest identity grant is written", "computevm", subject.ComputeVM,
			"api", wish.API, "actx", wish.Actx, "scopes", wish.Scopes)
	}
	if !readable {
		return nil
	}
	for name := range current {
		if wanted[name] {
			continue
		}
		if err := s.opts.Store.Delete(ctx, name); err != nil && !apierrors.IsNotFound(err) {
			return fmt.Errorf("deleting the grant %s: %w", name, err)
		}
		s.opts.Log.Info("a guest identity grant is revoked", "computevm", subject.ComputeVM, "grant", name)
	}
	return nil
}

func (s *Syncer) rbacWarnDue(requesterDID string) (int, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	now := time.Now()
	last, seen := s.rbacWarned[requesterDID]
	if seen && now.Sub(last.at) < rbacWarnEvery {
		last.suppressed++
		s.rbacWarned[requesterDID] = last
		return 0, false
	}
	s.rbacWarned[requesterDID] = rbacWarning{at: now}
	return last.suppressed, true
}

func (s *Syncer) records(ctx context.Context, requesterDID, acceptURI string, acceptedAt time.Time) ([]RBACRecord, bool, error) {
	records, err := s.opts.RBAC.Records(ctx, requesterDID, acceptURI, acceptedAt)
	if err != nil {
		return nil, false, err
	}
	return records, true, nil
}

func checkSubject(subject GrantSubject) error {
	switch {
	case !ValidComputeVM(subject.ComputeVM):
		return fmt.Errorf("guestidentity: %q is not a ComputeVM name", subject.ComputeVM)
	case subject.RequesterDID == "":
		return errors.New("guestidentity: the grant subject names no requester")
	case subject.Subject == "":
		return errors.New("guestidentity: the grant subject names no OIDC subject")
	}
	return nil
}