teleport/lib/join/server.go

// Teleport
// Copyright (C) 2025 Gravitational, Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package join

import (
	"cmp"
	"context"
	"crypto"
	"crypto/x509"
	"encoding/pem"
	"errors"
	"log/slog"
	"slices"
	"strings"
	"sync"
	"time"

	"github.com/gravitational/trace"
	"github.com/jonboulle/clockwork"
	"golang.org/x/crypto/ssh"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"

	"github.com/gravitational/teleport"
	"github.com/gravitational/teleport/api/client/proto"
	"github.com/gravitational/teleport/api/constants"
	joiningv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/scopes/joining/v1"
	workloadidentityv1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/workloadidentity/v1"
	"github.com/gravitational/teleport/api/types"
	apievents "github.com/gravitational/teleport/api/types/events"
	"github.com/gravitational/teleport/api/utils/keys"
	"github.com/gravitational/teleport/lib/auth/keystore"
	"github.com/gravitational/teleport/lib/authz"
	"github.com/gravitational/teleport/lib/events"
	"github.com/gravitational/teleport/lib/join/azuredevops"
	"github.com/gravitational/teleport/lib/join/azurejoin"
	"github.com/gravitational/teleport/lib/join/bitbucket"
	"github.com/gravitational/teleport/lib/join/circleci"
	"github.com/gravitational/teleport/lib/join/ec2join"
	"github.com/gravitational/teleport/lib/join/gcp"
	"github.com/gravitational/teleport/lib/join/githubactions"
	"github.com/gravitational/teleport/lib/join/gitlab"
	"github.com/gravitational/teleport/lib/join/iamjoin"
	joinauthz "github.com/gravitational/teleport/lib/join/internal/authz"
	"github.com/gravitational/teleport/lib/join/internal/diagnostic"
	"github.com/gravitational/teleport/lib/join/internal/messages"
	"github.com/gravitational/teleport/lib/join/joinutils"
	"github.com/gravitational/teleport/lib/join/oraclejoin"
	"github.com/gravitational/teleport/lib/join/provision"
	"github.com/gravitational/teleport/lib/join/spacelift"
	"github.com/gravitational/teleport/lib/join/terraformcloud"
	"github.com/gravitational/teleport/lib/join/tpmjoin"
	kubetoken "github.com/gravitational/teleport/lib/kube/token"
	"github.com/gravitational/teleport/lib/modules"
	"github.com/gravitational/teleport/lib/scopes/joining"
	"github.com/gravitational/teleport/lib/services"
	"github.com/gravitational/teleport/lib/services/readonly"
	"github.com/gravitational/teleport/lib/utils"
	"github.com/gravitational/teleport/lib/utils/hostid"
	logutils "github.com/gravitational/teleport/lib/utils/log"
)

var log = logutils.NewPackageLogger(teleport.ComponentKey, "join")

// AuthService is the subset of the Auth service interface required by the
// JoinServer to implement joining.
type AuthService interface {
	ValidateToken(ctx context.Context, tokenName string) (types.ProvisionToken, error)
	GenerateHostCertsForJoin(ctx context.Context, token provision.Token, req *HostCertsParams) (*proto.Certs, error)
	GenerateBotCertsForJoin(ctx context.Context, token provision.Token, req *BotCertsParams) (*proto.Certs, string, error)
	EmitAuditEvent(ctx context.Context, e apievents.AuditEvent) error
	GetAuthPreference(ctx context.Context) (types.AuthPreference, error)
	GetReadOnlyAuthPreference(context.Context) (readonly.AuthPreference, error)
	GetClusterName(context.Context) (types.ClusterName, error)
	GetCertAuthority(context.Context, types.CertAuthID, bool) (types.CertAuthority, error)
	GetKeyStore() *keystore.Manager
	PatchToken(context.Context, string, func(types.ProvisionToken) (types.ProvisionToken, error)) (types.ProvisionToken, error)
	UpsertLock(context.Context, types.Lock) error
	CheckLockInForce(constants.LockingMode, []types.LockTarget) error
	GetClock() clockwork.Clock
	GetHTTPClientForAWSSTS() utils.HTTPDoClient
	GetAWSOrganizationsClientGetter() iamjoin.OrganizationsAPIGetter
	GenerateAWSOIDCToken(ctx context.Context, integrationName string) (string, error)
	GetIntegration(ctx context.Context, name string) (types.Integration, error)
	GetAzureDevopsIDTokenValidator() azuredevops.Validator
	GetBitbucketIDTokenValidator() bitbucket.Validator
	GetEC2ClientForEC2JoinMethod() ec2join.EC2Client
	GetCircleCITokenValidator() circleci.Validator
	GetEnv0IDTokenValidator() Env0TokenValidator
	GetGCPIDTokenValidator() gcp.Validator
	GetGHAIDTokenValidator() githubactions.GithubIDTokenValidator
	GetGHAIDTokenJWKSValidator() githubactions.GithubIDTokenJWKSValidator
	GetGitlabIDTokenValidator() gitlab.Validator
	GetTPMValidator() tpmjoin.TPMValidator
	GetK8sTokenReviewValidator() kubetoken.InClusterValidator
	GetK8sJWKSValidator() kubetoken.JWKSValidator
	GetK8sOIDCValidator() *kubetoken.KubernetesOIDCTokenValidator
	GetSpaceliftIDTokenValidator() spacelift.Validator
	GetTerraformIDTokenValidator() terraformcloud.Validator
	GetAzureJoinConfig() *azurejoin.AzureJoinConfig
	services.Presence
	GetStaticScopedTokens(context.Context) (*joiningv1.StaticScopedTokens, error)
}

// ServerConfig holds configuration parameters for [Server].
type ServerConfig struct {
	AuthService        AuthService
	ScopedAuthorizer   authz.ScopedAuthorizer
	FIPS               bool
	ScopedTokenService services.ScopedTokenService
	OracleHTTPClient   utils.HTTPDoClient
	Logger             *slog.Logger
	Modules            modules.Modules
}

// Server implements cluster joining for nodes and bots.
type Server struct {
	cfg               *ServerConfig
	oracleRootCACache *oraclejoin.RootCACache
}

// NewServer returns a new [Server] instance.
func NewServer(cfg *ServerConfig) *Server {
	if cfg.Logger == nil {
		cfg.Logger = slog.With(teleport.ComponentKey, "join")
	}
	return &Server{
		cfg:               cfg,
		oracleRootCACache: oraclejoin.NewRootCACache(),
	}
}

// getProvisionToken attempts to resolve a name to a [provision.Token] by first attempting to
// fetch a [joiningv1.ScopedToken] and then falling back to a [types.ProvisionTokenV2] if a
// scoped token can not be found.
func (s *Server) getProvisionToken(ctx context.Context, name string) (provision.Token, error) {
	var scoped provision.Token
	var scopedErr error

	var classic provision.Token
	var classicErr error

	wg := &sync.WaitGroup{}
	wg.Go(func() {
		staticTokens, err := s.cfg.AuthService.GetStaticScopedTokens(ctx)
		if err != nil {
			if !trace.IsNotFound(err) {
				s.cfg.Logger.ErrorContext(ctx, "could not fetch static scoped tokens", "error", err)
			}
		}

		// short circuit if a matching static scoped token is found
		for _, tok := range staticTokens.GetSpec().GetTokens() {
			if tok.GetMetadata().GetName() == name {
				scoped, scopedErr = joining.NewToken(tok)
				return
			}
		}

		res, err := s.cfg.ScopedTokenService.GetScopedToken(ctx, &joiningv1.GetScopedTokenRequest{
			Name:       name,
			WithSecret: true,
		})
		if err != nil {
			scopedErr = err
			return
		}
		if err := joining.ValidateTokenForUse(res.GetToken()); err != nil {
			scopedErr = err
			return
		}

		scoped, scopedErr = joining.NewToken(res.GetToken())
	})
	wg.Go(func() {
		// Fetch the provision token and validate that it is not expired.
		classic, classicErr = s.cfg.AuthService.ValidateToken(ctx, name)
	})
	wg.Wait()

	// we explicitly disallow a join if the provided token name returns both a scoped and classic provision token
	if scoped != nil && classic != nil {
		return nil, trace.AccessDenied("joining with an ambiguous token name is not permitted")
	}

	if scoped != nil {
		return scoped, nil
	}

	if classic != nil {
		return classic, nil
	}

	// if both errors are [trace.NotFoundError], just return a single err
	if trace.IsNotFound(scopedErr) && trace.IsNotFound(classicErr) {
		return nil, trace.NotFound("token expired or not found")
	}

	// prefer reporting errors other than [trace.NotFoundError]
	if trace.IsNotFound(scopedErr) {
		return nil, trace.Wrap(classicErr)
	}

	if trace.IsNotFound(classicErr) {
		return nil, trace.Wrap(scopedErr)
	}

	// return both errors as an aggregate if we couldn't reasonably return one
	return nil, trace.NewAggregate(scopedErr, classicErr)
}

// Join implements cluster joining for nodes and bots.
//
// It returns credentials for a node or bot to join the Teleport cluster using
// a provision token.
//
// The client must request a specific role (and the role must match one of the
// roles the token was generated for).
//
// If a token was generated with a TTL, it gets enforced (can't join after the
// token expires).
//
// Only secret tokens are currently supported.
func (s *Server) Join(stream messages.ServerStream) (err error) {
	ctx := stream.Context()
	diag := stream.Diagnostic()
	defer func() {
		if err != nil {
			diag.Set(func(i *diagnostic.Info) { i.Error = err })
			handleJoinFailure(ctx, s.cfg.AuthService, diag)
		}
	}()

	// Receive the first message from the client, which must always be ClientInit.
	clientInit, err := messages.RecvRequest[*messages.ClientInit](stream)
	if err != nil {
		return trace.Wrap(err)
	}
	// Set any diagnostic info we can get from the ClientInit message.
	diag.Set(func(i *diagnostic.Info) {
		i.Role = joinutils.SanitizeUntrustedString(clientInit.SystemRole)
		if clientInit.JoinMethod != nil {
			i.RequestedJoinMethod = joinutils.SanitizeUntrustedString(*clientInit.JoinMethod)
		}
	})
	if err := clientInit.Check(); err != nil {
		return trace.Wrap(err, "validating ClientInit message")
	}

	// Authenticate the request in case the node/bot is rejoining with previous
	// credentials.
	authCtx, err := s.authenticate(ctx, diag, clientInit)
	if err != nil {
		return trace.Wrap(err)
	}

	token, err := s.getProvisionToken(ctx, clientInit.TokenName)
	if err != nil {
		return trace.Wrap(err)
	}

	// Set any diagnostic info we can get from the token.
	diag.Set(func(i *diagnostic.Info) {
		i.SafeTokenName = token.GetSafeName()
		i.TokenJoinMethod = string(configuredJoinMethod(token))
		i.TokenExpires = token.Expiry()
		i.BotName = token.GetBotName()

		// It's not worth fetching the true bot scope here (via bot user label)
		// so we'll just include the one embedded in the token.
		i.BotScope = token.GetBotScope()
	})

	// Validate that the requested join method matches the join method
	// configured on the token, or that the client did not specify a specific
	// join method and allow the server to choose it from the token.
	joinMethod, err := checkJoinMethod(token, clientInit.JoinMethod)
	if err != nil {
		return trace.Wrap(err)
	}

	// Assert that the provision token allows the requested system role.
	if err := TokenAllowsRole(token, types.SystemRole(clientInit.SystemRole)); err != nil {
		return trace.Wrap(err)
	}

	if authCtx.IsInstance && authCtx.Scope != token.GetAssignedScope() {
		return trace.BadParameter("tried to re-join instance from scope %q into %q", authCtx.Scope, token.GetAssignedScope())
	}

	authPref, err := s.cfg.AuthService.GetAuthPreference(ctx)
	if err != nil {
		return trace.Wrap(err, "getting cluster auth preference")
	}

	// Send the ServerInit message with the selected join method and current
	// cluster signature algorithm suite so that the client can generate keys
	// with the right algorithm.
	if err := stream.Send(&messages.ServerInit{
		JoinMethod:              string(joinMethod),
		SignatureAlgorithmSuite: authPref.GetSignatureAlgorithmSuite(),
	}); err != nil {
		return trace.Wrap(err, "sending ServerInit message")
	}

	// Call out to the handler for the specific join method.
	result, err := s.handleJoinMethod(stream, authCtx, clientInit, token, joinMethod)
	if err != nil {
		return trace.Wrap(err)
	}

	// Finally, send the result back to the client.
	return trace.Wrap(stream.Send(result))
}

func (s *Server) handleJoinMethod(
	stream messages.ServerStream,
	authCtx *joinauthz.Context,
	clientInit *messages.ClientInit,
	token provision.Token,
	joinMethod types.JoinMethod,
) (messages.Response, error) {
	switch joinMethod {
	case types.JoinMethodAzure:
		return s.handleAzureJoin(stream, authCtx, clientInit, token)
	case types.JoinMethodAzureDevops:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateAzureDevopsToken)
	case types.JoinMethodBitbucket:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateBitbucketToken)
	case types.JoinMethodBoundKeypair:
		return s.handleBoundKeypairJoin(stream, authCtx, clientInit, token)
	case types.JoinMethodCircleCI:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateCircleCIToken)
	case types.JoinMethodEC2:
		return s.handleEC2Join(stream, authCtx, clientInit, token)
	case types.JoinMethodEnv0:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateEnv0Token)
	case types.JoinMethodGCP:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateGCPToken)
	case types.JoinMethodGitHub:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateGithubToken)
	case types.JoinMethodGitLab:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateGitlabToken)
	case types.JoinMethodIAM:
		return s.handleIAMJoin(stream, authCtx, clientInit, token)
	case types.JoinMethodKubernetes:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateKubernetesToken)
	case types.JoinMethodOracle:
		return s.handleOracleJoin(stream, authCtx, clientInit, token)
	case types.JoinMethodSpacelift:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateSpaceliftToken)
	case types.JoinMethodTerraformCloud:
		return s.handleOIDCJoin(stream, authCtx, clientInit, token, s.validateTerraformCloudToken)
	case types.JoinMethodToken:
		return s.handleTokenJoin(stream, authCtx, clientInit, token)
	case types.JoinMethodTPM:
		return s.handleTPMJoin(stream, authCtx, clientInit, token)
	default:
		return nil, trace.NotImplemented("join method %s is not implemented", joinMethod)
	}
}

func (s *Server) authenticate(ctx context.Context, diag *diagnostic.Diagnostic, clientInit *messages.ClientInit) (*joinauthz.Context, error) {
	authCtx, err := s.cfg.ScopedAuthorizer.AuthorizeScoped(ctx)
	if err != nil && !trace.IsAccessDenied(err) {
		return nil, trace.Wrap(err, "unexpected error authorizing request")
	}
	if trace.IsAccessDenied(err) || authCtx == nil {
		// No authentication or AccessDenied is okay, this is not normally an
		// authenticated endpoint unless the client is re-joining or the
		// request was forwarded by a proxy, just return an empty Context.

		// A note around use of ScopedAuthorizer: it will return an empty
		// context if the scopes feature is disabled even if an otherwise-valid
		// scoped identity is presented, so they will be treated as
		// unauthenticated and ultimately will fail to join. This edge case will
		// be resolved when the scopes feature flag is removed.
		return &joinauthz.Context{}, nil
	}
	var isProxy bool
	if unscopedCtx, ok := authCtx.UnscopedContext(); ok {
		// Proxy identities are always unscoped, so the unscoped context should
		// always be available.
		isProxy = authz.HasBuiltinRole(*unscopedCtx, types.RoleProxy.String())
	}
	if !isProxy && clientInit.ProxySuppliedParams != nil {
		return nil, trace.AccessDenied("client set ProxySuppliedParameters but did not authenticate as a proxy")
	}
	if clientInit.ForwardedByProxy {
		if !isProxy {
			return nil, trace.BadParameter("client claims to be a proxy forwarding the request but did not authenticate as a proxy (this is a bug)")
		}
		if clientInit.ProxySuppliedParams == nil {
			return nil, trace.BadParameter("proxy forwarded request but did not set ProxySuppliedParams (this is a bug)")
		}
		diag.Set(func(i *diagnostic.Info) {
			i.RemoteAddr = clientInit.ProxySuppliedParams.RemoteAddr
			i.ClientVersion = clientInit.ProxySuppliedParams.ClientVersion
		})
		// Must ignore any authentication if the request was forwarded by a
		// proxy to avoid forgery of a host ID or system role via the proxy
		// credentials.
		return &joinauthz.Context{
			IsForwardedByProxy: true,
		}, nil
	}

	id := authCtx.Identity.GetIdentity()

	isInstance := slices.Equal(id.Groups, []string{types.RoleInstance.String()})
	var systemRoles types.SystemRoles
	if isInstance {
		systemRoles, err = types.NewTeleportRoles(id.SystemRoles)
		if err != nil {
			return nil, trace.Wrap(err)
		}
	}

	hostID := ""
	botInstanceID := ""
	botGeneration := uint64(0)
	if id.IsBot() {
		botInstanceID = id.BotInstanceID
		botGeneration = id.Generation
	} else {
		hostID = strings.SplitN(id.Username, ".", 2)[0]
	}

	// Set any diagnostic info we can get from the authenticated identity.
	diag.Set(func(i *diagnostic.Info) {
		i.HostID = hostID
		i.SystemRoles = systemRoles.StringSlice()
		i.BotInstanceID = botInstanceID
		i.BotGeneration = botGeneration
	})

	return &joinauthz.Context{
		IsInstance:    isInstance,
		IsBot:         id.IsBot(),
		SystemRoles:   systemRoles,
		HostID:        hostID,
		BotInstanceID: botInstanceID,
		BotGeneration: botGeneration,
		Scope:         id.AgentScope,
	}, nil
}

func checkJoinMethod(token provision.Token, requestedJoinMethod *string) (types.JoinMethod, error) {
	tokenJoinMethod := configuredJoinMethod(token)
	if requestedJoinMethod == nil {
		// Auto join method mode, the client didn't specify so use whatever is on the token.
		return tokenJoinMethod, nil
	}
	if types.JoinMethod(*requestedJoinMethod) != tokenJoinMethod {
		return "", trace.BadParameter(
			"client requested join method %s, provision token only supports method %s",
			*requestedJoinMethod, tokenJoinMethod)
	}
	return tokenJoinMethod, nil
}

// TokenAllowsRole asserts that the given provision token allows the
// requested role, or else it returns an error.
func TokenAllowsRole(token provision.Token, role types.SystemRole) error {
	// Instance certs can be requested if the provision token allows at least
	// one local service role (e.g. proxy, node, etc).
	if role == types.RoleInstance {
		hasLocalServiceRole := false
		for _, role := range token.GetRoles() {
			if role.IsLocalService() {
				hasLocalServiceRole = true
				break
			}
		}
		if !hasLocalServiceRole {
			return trace.AccessDenied("cannot requisition instance certs (token contains no local service roles)")
		}
	}

	// Make sure the caller is requesting a role allowed by the token.
	if !token.GetRoles().Include(role) && role != types.RoleInstance {
		return trace.BadParameter("can not join the cluster, the token does not allow role %s", role)
	}

	return nil
}

func (s *Server) makeResult(
	ctx context.Context,
	diag *diagnostic.Diagnostic,
	authCtx *joinauthz.Context,
	clientInit *messages.ClientInit,
	clientParams *messages.ClientParams,
	token provision.Token,
	rawClaims any,
	attrs *workloadidentityv1pb.JoinAttrs,
) (messages.Response, error) {
	switch types.SystemRole(clientInit.SystemRole) {
	case types.RoleInstance:
		return s.makeHostResult(ctx, diag, authCtx, clientParams.HostParams, token, rawClaims)
	case types.RoleBot:
		result, _, err := s.makeBotResult(ctx, diag, authCtx, clientParams.BotParams, token, rawClaims, attrs)
		return result, trace.Wrap(err)
	default:
		return nil, trace.NotImplemented("new join service only supports Instance and Bot system roles, client requested %s", clientInit.SystemRole)
	}
}

func (s *Server) makeHostResult(
	ctx context.Context,
	diag *diagnostic.Diagnostic,
	authCtx *joinauthz.Context,
	hostParams *messages.HostParams,
	token provision.Token,
	rawClaims any,
) (*messages.HostResult, error) {
	certsParams, err := makeHostCertsParams(ctx, diag, authCtx, hostParams, configuredJoinMethod(token), rawClaims)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	certs, err := s.cfg.AuthService.GenerateHostCertsForJoin(ctx, token, certsParams)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	certificates, err := convertCerts(certs)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return &messages.HostResult{
		Certificates:    *certificates,
		HostID:          certsParams.HostID,
		ImmutableLabels: token.GetImmutableLabels(),
	}, nil
}

// MakeHostCertsParams returns [HostCertsParams] populated by the ClientInit
// message and context of the request.
func makeHostCertsParams(
	ctx context.Context,
	diag *diagnostic.Diagnostic,
	authCtx *joinauthz.Context,
	hostParams *messages.HostParams,
	joinMethod types.JoinMethod,
	rawClaims any,
) (*HostCertsParams, error) {
	// GenerateHostCertsForJoin requires the TLS key to be PEM-encoded.
	tlsPub, err := x509.ParsePKIXPublicKey(hostParams.PublicKeys.PublicTLSKey)
	if err != nil {
		return nil, trace.BadParameter("failed to parse TLS public key")
	}
	tlsPubPEM, err := keys.MarshalPublicKey(crypto.PublicKey(tlsPub))
	if err != nil {
		return nil, trace.Wrap(err)
	}

	// GenerateHostCertsForJoin requires the SSH key to be in authorized keys format.
	sshPub, err := ssh.ParsePublicKey(hostParams.PublicKeys.PublicSSHKey)
	if err != nil {
		return nil, trace.BadParameter("failed to parse SSH public key")
	}
	sshAuthorizedKey := ssh.MarshalAuthorizedKey(sshPub)

	params := &HostCertsParams{
		SystemRole:           types.RoleInstance,
		PublicTLSKey:         tlsPubPEM,
		PublicSSHKey:         sshAuthorizedKey,
		HostName:             hostParams.HostName,
		AdditionalPrincipals: hostParams.AdditionalPrincipals,
		DNSNames:             hostParams.DNSNames,
		RemoteAddr:           diag.Get().RemoteAddr,
		RawJoinClaims:        rawClaims,
	}

	if authCtx.IsInstance {
		// Only authenticated Instance certs are allowed to re-join and
		// maintain their existing host ID and authenticate additional system
		// roles.
		params.HostID = authCtx.HostID
		params.AuthenticatedSystemRoles = authCtx.SystemRoles
	} else if joinMethod == types.JoinMethodEC2 {
		// EC2 join method uses a special host ID format that will be set in
		// authCtx by the EC2 method handler.
		params.HostID = authCtx.HostID
	} else {
		// Generate a new host ID to assign to the client.
		hostID, err := hostid.Generate(ctx, joinMethod)
		if err != nil {
			return nil, trace.Wrap(err)
		}
		params.HostID = hostID
	}

	return params, nil
}

func (s *Server) makeBotResult(
	ctx context.Context,
	diag *diagnostic.Diagnostic,
	authCtx *joinauthz.Context,
	botParams *messages.BotParams,
	token provision.Token,
	rawClaims any,
	attrs *workloadidentityv1pb.JoinAttrs,
) (*messages.BotResult, string, error) {
	certsParams, err := makeBotCertsParams(diag, authCtx, botParams, rawClaims, attrs)
	if err != nil {
		return nil, "", trace.Wrap(err)
	}
	certs, botInstanceID, err := s.cfg.AuthService.GenerateBotCertsForJoin(ctx, token, certsParams)
	if err != nil {
		return nil, "", trace.Wrap(err)
	}
	certificates, err := convertCerts(certs)
	if err != nil {
		return nil, "", trace.Wrap(err)
	}
	return &messages.BotResult{
		Certificates: *certificates,
	}, botInstanceID, nil
}

// makeBotCertsParams returns [BotCertsParams] populated by the
// ClientInit message and context of the request.
func makeBotCertsParams(
	diag *diagnostic.Diagnostic,
	authCtx *joinauthz.Context,
	botParams *messages.BotParams,
	rawClaims any,
	attrs *workloadidentityv1pb.JoinAttrs,
) (*BotCertsParams, error) {
	// GenerateBotCertsForJoin requires the TLS key to be PEM-encoded.
	tlsPub, err := x509.ParsePKIXPublicKey(botParams.PublicKeys.PublicTLSKey)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	tlsPubPEM, err := keys.MarshalPublicKey(crypto.PublicKey(tlsPub))
	if err != nil {
		return nil, trace.Wrap(err)
	}

	// GenerateBotCertsForJoin requires the SSH key to be in authorized keys format.
	sshPub, err := ssh.ParsePublicKey(botParams.PublicKeys.PublicSSHKey)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	sshAuthorizedKey := ssh.MarshalAuthorizedKey(sshPub)

	return &BotCertsParams{
		PublicTLSKey:  tlsPubPEM,
		PublicSSHKey:  sshAuthorizedKey,
		BotInstanceID: authCtx.BotInstanceID,
		BotGeneration: int32(authCtx.BotGeneration),
		Expires:       botParams.Expires,
		RemoteAddr:    diag.Get().RemoteAddr,
		RawJoinClaims: rawClaims,
		Attrs:         attrs,
	}, nil
}

// convertCerts converts the PEM-encoded TLS certs and the authorized_keys
// formats SSH cert + keys into the wire formats required for
// [messages.Certificates].
func convertCerts(certs *proto.Certs) (*messages.Certificates, error) {
	sshCert, err := rawSSHCert(certs.SSH)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	// certs.SSHCACerts is a misnomer, SSH CAs are just public keys, not certificates.
	sshCAKeys, err := rawSSHPublicKeys(certs.SSHCACerts)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return &messages.Certificates{
		TLSCert:    rawTLSCert(certs.TLS),
		TLSCACerts: rawTLSCerts(certs.TLSCACerts),
		SSHCert:    sshCert,
		SSHCAKeys:  sshCAKeys,
	}, nil
}

// rawTLSCerts converts a slice of PEM-encoded TLS certificates to the raw ASN.1
// DER form as required by [Result].
func rawTLSCerts(pemBytes [][]byte) [][]byte {
	out := make([][]byte, len(pemBytes))
	for i, bytes := range pemBytes {
		out[i] = rawTLSCert(bytes)
	}
	return out
}

// rawTLSCert converts a PEM-encoded TLS certificate to the raw ASN.1 DER form
// as required by [Result].
func rawTLSCert(pemBytes []byte) []byte {
	pemBlock, _ := pem.Decode(pemBytes)
	return pemBlock.Bytes
}

// rawSSHCert converts an SSH certificate or public key in SSH authorized_keys
// format to the SSH wire format as required by [messages.Result].
func rawSSHCert(authorizedKey []byte) ([]byte, error) {
	pub, _, _, _, err := ssh.ParseAuthorizedKey(authorizedKey)
	if err != nil {
		return nil, trace.Wrap(err)
	}
	return pub.Marshal(), nil
}

// rawSSHPublicKeys converts a slices of SSH public keys in SSH authorized_keys
// format to the SSH wire format as required by [messages.Result].
func rawSSHPublicKeys(authorizedKeys [][]byte) ([][]byte, error) {
	out := make([][]byte, len(authorizedKeys))
	for i, authorizedKey := range authorizedKeys {
		var err error
		out[i], err = rawSSHCert(authorizedKey)
		if err != nil {
			return nil, trace.Wrap(err)
		}
	}
	return out, nil
}

func setDiagnosticClientParams(diag *diagnostic.Diagnostic, clientParams *messages.ClientParams) {
	if clientParams.HostParams != nil {
		diag.Set(func(i *diagnostic.Info) {
			i.NodeName = joinutils.SanitizeUntrustedString(clientParams.HostParams.HostName)
		})
	}
}

func handleJoinFailure(ctx context.Context, emitter apievents.Emitter, diag *diagnostic.Diagnostic) {
	diagInfo := diag.Get()
	slogAttrs := diagInfo.SlogAttrs()

	// Fetch and encode RawJoinAttrs if they are available.
	attributesStruct, err := joinutils.RawJoinAttrsToStruct(diagInfo.RawJoinAttrs)
	if err != nil {
		log.WarnContext(ctx, "Unable to fetch join attributes from join method", "error", err)
	}
	if attributesStruct != nil {
		slogAttrs = append(slogAttrs, slog.Any("attributes", attributesStruct))
	}

	log.LogAttrs(ctx, slog.LevelWarn, "Failure to join cluster occurred", slogAttrs...)
	if err := emitter.EmitAuditEvent(context.WithoutCancel(ctx), makeAuditEvent(diagInfo, attributesStruct)); err != nil {
		log.WarnContext(ctx, "Failed to emit failed join event", "error", err)
	}
}

func makeAuditEvent(info diagnostic.Info, attributesStruct *apievents.Struct) apievents.AuditEvent {
	errorMessage := info.Error.Error()
	if errors.Is(info.Error, context.Canceled) || status.Code(info.Error) == codes.Canceled {
		errorMessage = "join attempt timed out or was aborted"
	}
	status := apievents.Status{
		Success: false,
		Error:   errorMessage,
	}
	if info.Role == types.RoleBot.String() {
		return &apievents.BotJoin{
			Metadata: apievents.Metadata{
				Type: events.BotJoinEvent,
				Code: events.BotJoinFailureCode,
				Time: time.Now(),
			},
			Status: status,
			ConnectionMetadata: apievents.ConnectionMetadata{
				RemoteAddr: info.RemoteAddr,
			},
			Method:        cmp.Or(info.TokenJoinMethod, info.RequestedJoinMethod),
			TokenName:     info.SafeTokenName,
			BotName:       info.BotName,
			BotInstanceID: info.BotInstanceID,
			Scope:         info.BotScope,
			Attributes:    attributesStruct,
		}
	}
	return &apievents.InstanceJoin{
		Metadata: apievents.Metadata{
			Type: events.InstanceJoinEvent,
			Code: events.InstanceJoinFailureCode,
			Time: time.Now(),
		},
		Status: status,
		ConnectionMetadata: apievents.ConnectionMetadata{
			RemoteAddr: info.RemoteAddr,
		},
		Method:       cmp.Or(info.TokenJoinMethod, info.RequestedJoinMethod),
		TokenName:    info.SafeTokenName,
		TokenExpires: info.TokenExpires,
		Role:         info.Role,
		NodeName:     info.NodeName,
		Attributes:   attributesStruct,
	}
}

func configuredJoinMethod(token provision.Token) types.JoinMethod {
	method := token.GetJoinMethod()
	if method == types.JoinMethodUnspecified {
		return types.JoinMethodToken
	}
	return method
}