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 }