johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package requester

import (
	"cmp"
	"context"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"

	commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market"
	atprotooauth "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/oauth"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/pdsclient"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/xrpc"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/pds"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/lease"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/market/participant"
)

func (q *Requester) resolveOAuthIdentity(ctx context.Context) error {
	ref := q.cfg.Spec.Identity.MarketIdentityRef
	workspaces := IdentityWorkspaces(q.cfg.Workspace)
	found, ok := q.deps.Inputs.AwaitIdentity(ctx, workspaces, ref.Name, IdentityWait)
	if !ok {
		return fmt.Errorf("requester: reading its MarketIdentity %s: not found", ref.Name)
	}
	did := strings.TrimSpace(found.Status.Did)
	if did == "" {
		return fmt.Errorf("requester: its MarketIdentity %s is not signed in (phase %s)", ref.Name, found.Status.Phase)
	}
	q.noteIdentityRef(ref.Name)
	q.did, q.repoDID, q.marketDID = did, did, did
	q.pdsURL = cmp.Or(q.pdsURL, strings.TrimSpace(found.Status.PdsURL))
	q.handle = cmp.Or(q.handle, found.Status.Handle)
	return nil
}

func (q *Requester) oauthServiceAuth() ServiceAuthSource {
	if q.deps.ServiceAuth == nil {
		return nil
	}
	identity, did, open := q.identityRef, q.did, q.deps.ServiceAuth
	return participant.PDSServiceAuth{DID: did, Open: func(ctx context.Context) (pds.ServiceAuth, error) {
		return open(ctx, identity, did)
	}}
}

type ownerIdentityServiceAuth struct {
	participant.PDSServiceAuth
}

func (q *Requester) ownerIdentityServiceAuth() ServiceAuthSource {
	if q.deps.ServiceAuth == nil || q.did == "" || q.cfg.LocalSigner || q.cfg.IdentityMode != IdentityModeCallback || q.identityRef.Name == "" {
		return nil
	}
	identity, did, open := q.identityRef, q.did, q.deps.ServiceAuth
	return ownerIdentityServiceAuth{participant.PDSServiceAuth{DID: did, Open: func(ctx context.Context) (pds.ServiceAuth, error) {
		return open(ctx, identity, did)
	}}}
}

func (q *Requester) atircServiceAuth(market ServiceAuthSource) (ServiceAuthSource, string, string) {
	if market != nil {
		return market, serviceAuthName(market), ""
	}
	if q.deps.ServiceAuth == nil || q.did == "" {
		return nil, "none", "the requester has no service-auth handle"
	}
	open, did := q.deps.ServiceAuth, q.did
	switch pdsRef := q.cfg.Spec.PdsRef; {
	case q.cfg.LocalSigner && pdsRef != nil && pdsRef.Name != "":
		plcURL := ""
		if network := q.cfg.Spec.Network; network != nil {
			plcURL = network.PLCDirectoryURL
		}
		if ok, reason := LocalPdsSignsAtirc(plcURL, q.pdsURL); !ok {
			return nil, "none", reason
		}
		target := commonv1.ObjectRef{Group: cmp.Or(pdsRef.Group, v1alpha1.GroupName), Kind: cmp.Or(pdsRef.Kind, "Pds"), Name: pdsRef.Name,
			Workspace: cmp.Or(pdsRef.Workspace, q.cfg.Workspace)}
		return participant.PDSServiceAuth{DID: did, Open: func(ctx context.Context) (pds.ServiceAuth, error) {
			return open(ctx, target, did)
		}}, "local-pds", ""
	}
	return nil, "none", "the identity mode signs no service auth"
}

func (q *Requester) oauthGranted(ctx context.Context, scope string) (bool, error) {
	granted, known, err := q.oauthGrants(ctx)
	if err != nil || !known {
		return false, err
	}
	return len(atprotooauth.MissingScopes([]string{scope}, granted)) == 0, nil
}

func (q *Requester) oauthGrants(ctx context.Context) ([]string, bool, error) {
	if q.cfg.IdentityMode != IdentityModeOAuth || q.deps.ServiceAuth == nil {
		return nil, false, nil
	}
	handle, err := q.deps.ServiceAuth(ctx, q.identityRef, q.did)
	if err != nil {
		return nil, false, err
	}
	grants, ok := handle.(pds.ScopeGrants)
	if !ok {
		return nil, false, nil
	}
	granted, ok := grants.GrantedScopes(q.did)
	if !ok || !atprotooauth.KnownGrant(granted) {
		return nil, false, nil
	}
	return granted, true, nil
}

func (q *Requester) authenticateOwnerSession(ctx context.Context) error {
	session, ok := q.cfg.OwnerSessions[q.repoDID]
	if !ok {
		return fmt.Errorf("requester: the owner sessions file names no account %s", q.repoDID)
	}
	did, err := syntax.ParseDID(q.repoDID)
	if err != nil {
		return fmt.Errorf("requester: owner DID: %w", err)
	}
	if _, err := pdsclient.NewRemote(pdsclient.RemoteConfig{
		BaseURL:    session.Service,
		DID:        did,
		HTTPClient: q.deps.HTTP,
		Timeout:    DefaultHTTPTimeout,
	}); err != nil {
		return fmt.Errorf("requester: the owner PDS %s: %w", session.Service, err)
	}
	tokens := newPDSAccountTokens(session.Service, session.Identifier, session.Password, q.deps.HTTP)
	if _, err := tokens.RefreshAccessToken(ctx); err != nil {
		return fmt.Errorf("requester: the owner session of %s: %w", q.repoDID, err)
	}
	q.pdsURL = session.Service
	q.ownerServiceAuth = newOwnerSessionServiceAuth(session.Service, tokens, q.deps.HTTP)
	return nil
}

func (q *Requester) watchBinds() (<-chan struct{}, func()) {
	if q.deps.WatchBinds == nil {
		return nil, func() {}
	}
	return q.deps.WatchBinds()
}

func (q *Requester) bindSignerKeyAfterBinds(ctx context.Context) error {
	if q.deps.WatchBinds == nil {
		return q.bindSignerKey(ctx)
	}
	wake, cancel := q.deps.WatchBinds()
	defer cancel()
	waiting := false
	for {
		err := q.bindSignerKey(ctx)
		if err == nil {
			if waiting {
				q.log.Info("requester_signer_key_vault_ready")
			}
			return nil
		}
		if !waiting {
			waiting = true
			q.log.Info("requester_signer_key_waits_for_vault", "reason", "a Vault, MarketIdentity or AtprotoRepo bind wakes the next try; no bind within "+signerBindVaultWait.String()+" fails the start", "err", err.Error())
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-wake:
		case <-time.After(signerBindVaultWait):
			return err
		}
	}
}

func (q *Requester) bindSignerKey(ctx context.Context) error {
	if q.repo == nil || q.signer == nil {
		return errors.New("requester: the signer binding has no repo or signer")
	}
	key, err := q.signer.PublicDID()
	if err != nil {
		return fmt.Errorf("requester: the signer key: %w", err)
	}
	rkey := market.BadgeBlueKeysRkey(q.repoDID, key)
	if _, err := market.CreateBadgeBlueKeysRecord(ctx, q.repo, market.BadgeBlueKeysRecordOptions{
		DID:       q.repoDID,
		KeyID:     key,
		Service:   ownerSessionSigningService,
		CreatedAt: q.nowISO(),
	}); err != nil {
		return fmt.Errorf("requester: publishing the signer key binding: %w", err)
	}
	q.log.Info("requester_signer_key_bound", "did", q.repoDID, "key", key, "rkey", rkey)
	return nil
}

func serviceAuthName(source ServiceAuthSource) string {
	switch source.(type) {
	case nil:
		return "local"
	case participant.PDSServiceAuth:
		return "oauth"
	case ownerIdentityServiceAuth:
		return "owner-identity"
	case *ownerSessionServiceAuth:
		return "owner-session"
	}
	return "other"
}

type ownerSessionServiceAuth struct {
	minter xrpc.PDSMinter
}

var (
	_ ServiceAuthSource        = (*ownerSessionServiceAuth)(nil)
	_ ServiceAuthSource        = participant.PDSServiceAuth{}
	_ pdsclient.TokenRefresher = (*pdsAccountTokens)(nil)
)

func newOwnerSessionServiceAuth(pdsURL string, tokens pdsclient.TokenSource, client *http.Client) *ownerSessionServiceAuth {
	if client == nil {
		client = &http.Client{Timeout: DefaultHTTPTimeout}
	}
	var refresh xrpc.TokenFunc
	if refresher, ok := tokens.(pdsclient.TokenRefresher); ok {
		refresh = refresher.RefreshAccessToken
	}
	auth := &xrpc.Bearer{Tokens: xrpc.Refreshing(tokens.AccessToken, refresh)}
	return &ownerSessionServiceAuth{minter: xrpc.PDSMinter{Client: &xrpc.Client{Host: pdsURL, HTTP: client, Auth: auth}}}
}

func (a *ownerSessionServiceAuth) ServiceAuth(ctx context.Context, audience, lxm string) (string, error) {
	token, err := a.minter.ServiceAuth(ctx, audience, lxm)
	if err != nil {
		return "", fmt.Errorf("requester: getServiceAuth: %w", err)
	}
	return token, nil
}

type pdsAccountTokens struct {
	pdsURL string

	identifier string

	password string

	client *http.Client

	access *lease.Source[string]
}

func newPDSAccountTokens(pdsURL, identifier, password string, client *http.Client) *pdsAccountTokens {
	t := &pdsAccountTokens{pdsURL: pdsURL, identifier: identifier, password: password, client: client}
	t.access = lease.New(t.login)
	return t
}

func (t *pdsAccountTokens) AccessToken(ctx context.Context) (string, error) {
	return t.access.Get(ctx)
}

func (t *pdsAccountTokens) RefreshAccessToken(ctx context.Context) (string, error) {
	t.access.Invalidate()
	return t.access.Get(ctx)
}

func (t *pdsAccountTokens) login(ctx context.Context) (string, time.Time, error) {
	pds, err := pdsclient.NewRemote(pdsclient.RemoteConfig{BaseURL: t.pdsURL, HTTPClient: t.client, Timeout: DefaultHTTPTimeout})
	if err != nil {
		return "", time.Time{}, err
	}
	session, err := pds.CreateSession(ctx, t.identifier, t.password)
	if err != nil {
		return "", time.Time{}, err
	}
	expires, ok := pdsclient.TokenExpiry(session.AccessJWT)
	if !ok {
		return session.AccessJWT, time.Now(), nil
	}
	return session.AccessJWT, expires.Add(-pdsclient.TokenExpirySkew), nil
}