Client Credentials Flow

import supertokens from "supertokens-node";
import OAuth2Provider from "supertokens-node/recipe/oauth2provider";

supertokens.init({
  supertokens: {
    connectionURI: "...",
    apiKey: "...",
  },
  appInfo: {
    appName: "...",
    apiDomain: "...",
    websiteDomain: "...",
  },
  recipeList: [OAuth2Provider.init()],
});

from supertokens_python import init, InputAppInfo, SupertokensConfig
from supertokens_python.recipe import oauth2provider

init(
    app_info=InputAppInfo(
        app_name="...",
        api_domain="...",
        website_domain="...",
    ),
    framework="fastapi",
    supertokens_config=SupertokensConfig(
        connection_uri="...",
        api_key="..."
    ),
    recipe_list=[
        oauth2provider.init()
    ],
)

import OAuth2Provider from "supertokens-node/recipe/oauth2provider";

async function validateClientCredentialsToken(token: string): Promise<boolean> {
  try {
    const result = await OAuth2Provider.validateOAuth2AccessToken(token, {
      audience: "<AUDIENCE>",
      clientId: "<CLIENT_ID>",
      scopes: ["<YOUR_REQUIRED_SCOPE>"],
    });

    return result.payload.iss === "<YOUR_API_DOMAIN>/auth";
  } catch {
    return false;
  }
}

from supertokens_python.recipe.oauth2provider.interfaces import OAuth2TokenValidationRequirements
from supertokens_python.recipe.oauth2provider.syncio import validate_oauth2_access_token


def validate_client_credentials_token(token: str) -> bool:
    try:
        result = validate_oauth2_access_token(
            token=token,
            requirements=OAuth2TokenValidationRequirements(
                audience="<AUDIENCE>",
                client_id="<CLIENT_ID>",
                scopes=["<YOUR_REQUIRED_SCOPE>"],
            ),
        )
        return result.payload.get("iss") == "<YOUR_API_DOMAIN>/auth"
    except Exception:
        return False

import express, { type NextFunction, type Request, type Response } from "express";
import OAuth2Provider from "supertokens-node/recipe/oauth2provider";
import Session from "supertokens-node/recipe/session";

async function verifySessionOrOAuthToken(req: Request, res: Response, next: NextFunction) {
  const authorization = req.headers.authorization;

  if (authorization !== undefined) {
    const separator = authorization.indexOf(" ");
    const scheme = authorization.slice(0, separator);
    const token = authorization.slice(separator + 1);
    if (separator < 1 || scheme.toLowerCase() !== "bearer" || !token) {
      return res.status(401).json({ message: "Unauthorized" });
    }

    try {
      const result = await OAuth2Provider.validateOAuth2AccessToken(token, {
        audience: "<AUDIENCE>",
        clientId: "<CLIENT_ID>",
        scopes: ["<REQUIRED_SCOPE>"],
      });
      if (result.payload.iss === "<YOUR_API_DOMAIN>/auth") {
        return next();
      }
    } catch {
      // The bearer token may be a SuperTokens session access token.
    }
  }

  try {
    await Session.getSession(req, res);
    return next();
  } catch {
    return res.status(401).json({ message: "Unauthorized" });
  }
}

const app = express();
app.get("/protected", verifySessionOrOAuthToken, async (_req, res) => {
  res.json({ message: "Authorized" });
});

from fastapi import HTTPException
from fastapi.requests import Request
from supertokens_python.recipe.oauth2provider.interfaces import OAuth2TokenValidationRequirements
from supertokens_python.recipe.oauth2provider.syncio import validate_oauth2_access_token
from supertokens_python.recipe.session.syncio import get_session


def verify_session_or_oauth_token(request: Request) -> bool:
    authorization = request.headers.get("authorization")

    if authorization is not None:
        scheme, separator, token = authorization.partition(" ")
        if scheme.lower() != "bearer" or not separator or not token:
            raise HTTPException(status_code=401, detail="Unauthorized")

        try:
            result = validate_oauth2_access_token(
                token=token,
                requirements=OAuth2TokenValidationRequirements(
                    audience="<AUDIENCE>",
                    client_id="<CLIENT_ID>",
                    scopes=["<REQUIRED_SCOPE>"],
                ),
            )
            if result.payload.get("iss") == "<YOUR_API_DOMAIN>/auth":
                return True
        except Exception:
            # The bearer token may be a SuperTokens session access token.
            pass

    try:
        get_session(request)
        return True
    except Exception as error:
        raise HTTPException(status_code=401, detail="Unauthorized") from error