Implement a custom invite flow

Implement a list of approved email addresses for third-party sign-ups using the SuperTokens user metadata recipe.

Overview

This guide shows you how to disable public sign-ups to allow only certain people to access your app. For third-party login, maintain a list of approved email addresses and validate users against it.

Before you start

The tutorial assumes that you already have a working application integrated with SuperTokens. If you have not, please check the Quickstart Guide.

Prerequisites

This guide uses the UserMetadata recipe to store the list of approved email addresses. You need to enable it in the SDK initialization step.

Steps

1. Implement the approved email list

You can store this list in your own database or use the metadata feature provided by SuperTokens.

The following code samples show you how to save the approved email list in user metadata.

import UserMetadata from "supertokens-node/recipe/usermetadata";

function allowlistKey(tenantId: string) {
  return `emailAllowList:${tenantId}`;
}

async function addEmailToAllowlist(tenantId: string, email: string) {
  let existingData = await UserMetadata.getUserMetadata(allowlistKey(tenantId));
  let allowList: string[] = existingData.metadata.allowList || [];
  allowList = [...allowList, email];
  await UserMetadata.updateUserMetadata(allowlistKey(tenantId), {
    allowList,
  });
}

async function isEmailAllowed(tenantId: string, email: string) {
  let existingData = await UserMetadata.getUserMetadata(allowlistKey(tenantId));
  let allowList: string[] = existingData.metadata.allowList || [];
  return allowList.includes(email);
}

import (
	"fmt"

	"github.com/supertokens/supertokens-golang/recipe/usermetadata"
)

func allowlistKey(tenantID string) string {
	return fmt.Sprintf("emailAllowList:%s", tenantID)
}

func getAllowList(metadata map[string]interface{}) []string {
	allowList := []string{}
	rawAllowList, ok := metadata["allowList"].([]interface{})
	if !ok {
		return allowList
	}

	for _, value := range rawAllowList {
		email, ok := value.(string)
		if ok {
			allowList = append(allowList, email)
		}
	}
	return allowList
}

func addEmailToAllowlist(tenantID string, email string) error {
	existingData, err := usermetadata.GetUserMetadata(allowlistKey(tenantID))
	if err != nil {
		return err
	}
	allowList := getAllowList(existingData)
	allowList = append(allowList, email)
	_, err = usermetadata.UpdateUserMetadata(allowlistKey(tenantID), map[string]interface{}{
		"allowList": allowList,
	})
	return err
}

func isEmailAllowed(tenantID string, email string) (bool, error) {
	existingData, err := usermetadata.GetUserMetadata(allowlistKey(tenantID))
	if err != nil {
		return false, err
	}
	allowList := getAllowList(existingData)
	for _, allowedEmail := range allowList {
		if allowedEmail == email {
			return true, nil
		}
	}
	return false, nil
}

from typing import List

from supertokens_python.recipe.usermetadata.asyncio import (
    get_user_metadata,
    update_user_metadata,
)


def allowlist_key(tenant_id: str):
    return f"emailAllowList:{tenant_id}"


async def add_email_to_allow_list(tenant_id: str, email: str):
    metadataResult = await get_user_metadata(allowlist_key(tenant_id))
    allow_list: List[str] = metadataResult.metadata["allowList"] if "allowList" in metadataResult.metadata else []
    allow_list.append(email)
    await update_user_metadata(allowlist_key(tenant_id), {
        "allowList": allow_list
    })

async def is_email_allowed(tenant_id: str, email: str):
    metadataResult = await get_user_metadata(allowlist_key(tenant_id))
    allow_list: List[str] = metadataResult.metadata["allowList"] if "allowList" in metadataResult.metadata else []
    return email in allow_list

2. Check if the email is allowed

Update the backend SDK API function to only allow sign-up requests from users whose email addresses are on the approved list. Use the check functions from the previous code snippet.

The overrides reject a provider response without an email before the SDK can generate a synthetic email for a provider configured not to require one.

import ThirdParty from "supertokens-node/recipe/thirdparty";
import supertokens from "supertokens-node";

class SignUpNotAllowedError extends Error {}

ThirdParty.init({
  override: {
    functions: (originalImplementation) => {
      return {
        ...originalImplementation,
        signInUp: async function (input) {
          let existingUsers = await supertokens.listUsersByAccountInfo(input.tenantId, {
            thirdParty: {
              id: input.thirdPartyId,
              userId: input.thirdPartyUserId,
            },
          });
          if (existingUsers.length === 0) {
            if (!input.isVerified) {
              throw new SignUpNotAllowedError();
            }
            if (!(await isEmailAllowed(input.tenantId, input.email))) {
              throw new SignUpNotAllowedError();
            }
          }
          return originalImplementation.signInUp(input);
        },
      };
    },
    apis: (originalImplementation) => {
      return {
        ...originalImplementation,
        signInUpPOST: async function (input) {
          try {
            const provider = input.provider;
            const providerWithEmailCheck =
              provider.type === "oauth2"
                ? {
                    ...provider,
                    getUserInfo: async (getUserInfoInput: Parameters<typeof provider.getUserInfo>[0]) => {
                      const userInfo = await provider.getUserInfo(getUserInfoInput);
                      if (userInfo.email === undefined) {
                        throw new SignUpNotAllowedError();
                      }
                      return userInfo;
                    },
                  }
                : {
                    ...provider,
                    getUserInfo: async (getUserInfoInput: Parameters<typeof provider.getUserInfo>[0]) => {
                      const userInfo = await provider.getUserInfo(getUserInfoInput);
                      if (userInfo.email === undefined) {
                        throw new SignUpNotAllowedError();
                      }
                      return userInfo;
                    },
                  };

            return await originalImplementation.signInUpPOST!({
              ...input,
              provider: providerWithEmailCheck,
            });
          } catch (err: unknown) {
            if (err instanceof SignUpNotAllowedError) {
              return {
                status: "GENERAL_ERROR",
                message: "Sign-ups are disabled. Please contact the admin.",
              };
            }
            throw err;
          }
        },
      };
    },
  },
});

import (
	"errors"

	"github.com/supertokens/supertokens-golang/recipe/thirdparty"
	"github.com/supertokens/supertokens-golang/recipe/thirdparty/tpmodels"
	"github.com/supertokens/supertokens-golang/supertokens"
)

var errSignUpNotAllowed = errors.New("sign up not allowed")

func main() {
	thirdparty.Init(&tpmodels.TypeInput{
		Override: &tpmodels.OverrideStruct{
			APIs: func(originalImplementation tpmodels.APIInterface) tpmodels.APIInterface {
				originalSignInUpPOST := *originalImplementation.SignInUpPOST

				(*originalImplementation.SignInUpPOST) = func(provider *tpmodels.TypeProvider, input tpmodels.TypeSignInUpInput, tenantId string, options tpmodels.APIOptions, userContext supertokens.UserContext) (tpmodels.SignInUpPOSTResponse, error) {
					providerWithInviteCheck := *provider
					originalGetUserInfo := provider.GetUserInfo
					providerWithInviteCheck.GetUserInfo = func(oAuthTokens tpmodels.TypeOAuthTokens, userContext supertokens.UserContext) (tpmodels.TypeUserInfo, error) {
						userInfo, err := originalGetUserInfo(oAuthTokens, userContext)
						if err != nil {
							return tpmodels.TypeUserInfo{}, err
						}

						if userInfo.Email == nil {
							return tpmodels.TypeUserInfo{}, errSignUpNotAllowed
						}

						existingUser, err := thirdparty.GetUserByThirdPartyInfo(tenantId, provider.ID, userInfo.ThirdPartyUserId, userContext)
						if err != nil {
							return tpmodels.TypeUserInfo{}, err
						}

						if existingUser == nil {
							if !userInfo.Email.IsVerified {
								return tpmodels.TypeUserInfo{}, errSignUpNotAllowed
							}

							allowed, err := isEmailAllowed(tenantId, userInfo.Email.ID)
							if err != nil {
								return tpmodels.TypeUserInfo{}, err
							}
							if !allowed {
								return tpmodels.TypeUserInfo{}, errSignUpNotAllowed
							}
						}

						return userInfo, nil
					}

					resp, err := originalSignInUpPOST(&providerWithInviteCheck, input, tenantId, options, userContext)

					if errors.Is(err, errSignUpNotAllowed) {
						return tpmodels.SignInUpPOSTResponse{
							GeneralError: &supertokens.GeneralErrorResponse{
								Message: "Sign-ups are disabled. Please contact the admin.",
							},
						}, nil
					}

					return resp, err
				}

				return originalImplementation
			},
		},
	})
}

from copy import copy
from typing import Any, Dict, Optional, Union

from supertokens_python import InputAppInfo, init
from supertokens_python.asyncio import list_users_by_account_info
from supertokens_python.recipe import thirdparty
from supertokens_python.recipe.session.interfaces import SessionContainer
from supertokens_python.recipe.thirdparty.interfaces import (
    APIInterface,
    APIOptions,
    RecipeInterface,
)
from supertokens_python.recipe.thirdparty.provider import Provider, RedirectUriInfo
from supertokens_python.recipe.thirdparty.types import (
    RawUserInfoFromProvider,
    ThirdPartyInfo,
    UserInfo,
)
from supertokens_python.types import GeneralErrorResponse
from supertokens_python.types.base import AccountInfoInput


async def is_email_allowed(tenant_id: str, email: str):
    # from previous code snippet..
    return False


class SignUpNotAllowedError(Exception):
    pass


def override_thirdparty_functions(original_implementation: RecipeInterface):
    original_thirdparty_sign_in_up = original_implementation.sign_in_up

    async def thirdparty_sign_in_up(
        third_party_id: str,
        third_party_user_id: str,
        email: str,
        is_verified: bool,
        oauth_tokens: Dict[str, Any],
        raw_user_info_from_provider: RawUserInfoFromProvider,
        session: Optional[SessionContainer],
        should_try_linking_with_session_user: Union[bool, None],
        tenant_id: str,
        user_context: Dict[str, Any],
    ):
        existing_users = await list_users_by_account_info(
            tenant_id,
            AccountInfoInput(
                third_party=ThirdPartyInfo(
                    third_party_user_id=third_party_user_id,
                    third_party_id=third_party_id,
                )
            ),
        )
        if len(existing_users) == 0:
            if not is_verified:
                raise SignUpNotAllowedError()
            if not await is_email_allowed(tenant_id, email):
                raise SignUpNotAllowedError()

        return await original_thirdparty_sign_in_up(
            third_party_id,
            third_party_user_id,
            email,
            is_verified,
            oauth_tokens,
            raw_user_info_from_provider,
            session,
            should_try_linking_with_session_user,
            tenant_id,
            user_context,
        )

    original_implementation.sign_in_up = thirdparty_sign_in_up

    return original_implementation


def override_thirdparty_apis(original_implementation: APIInterface):
    original_sign_in_up_post = original_implementation.sign_in_up_post

    async def thirdparty_sign_in_up_post(
        provider: Provider,
        redirect_uri_info: Optional[RedirectUriInfo],
        oauth_tokens: Optional[Dict[str, Any]],
        session: Optional[SessionContainer],
        should_try_linking_with_session_user: Union[bool, None],
        tenant_id: str,
        api_options: APIOptions,
        user_context: Dict[str, Any],
    ):
        provider_with_email_check = copy(provider)
        original_get_user_info = provider.get_user_info

        async def get_user_info_with_email_check(
            oauth_tokens: Dict[str, Any], user_context: Dict[str, Any]
        ) -> UserInfo:
            user_info = await original_get_user_info(oauth_tokens, user_context)
            if user_info.email is None:
                raise SignUpNotAllowedError()
            return user_info

        setattr(
            provider_with_email_check,
            "get_user_info",
            get_user_info_with_email_check,
        )

        try:
            return await original_sign_in_up_post(
                provider_with_email_check,
                redirect_uri_info,
                oauth_tokens,
                session,
                should_try_linking_with_session_user,
                tenant_id,
                api_options,
                user_context,
            )
        except SignUpNotAllowedError:
            return GeneralErrorResponse(
                "Sign-ups are disabled. Please contact the admin."
            )

    original_implementation.sign_in_up_post = thirdparty_sign_in_up_post
    return original_implementation


init(
    app_info=InputAppInfo(api_domain="...", app_name="...", website_domain="..."),
    framework="...",
    recipe_list=[
        thirdparty.init(
            override=thirdparty.InputOverrideConfig(
                apis=override_thirdparty_apis, functions=override_thirdparty_functions
            ),
        )
    ],
)