Understanding Resource Servers and Scopes in AWS Cognito

What is a Resource Server?

A resource server in AWS Cognito represents a server that hosts protected resources, such as APIs. When you register a resource server in Cognito, you define it in the context of a User Pool, and it acts as an abstraction for grouping APIs or microservices under a single logical entity.

Key attributes of a resource server include:

• Identifier: A unique string that identifies the resource server, typically in URI format (e.g., https://api.example.com or com.example.myapp).
• Scopes: A collection of permissions or access levels associated with the APIs hosted by the resource server.

Why Use a Resource Server?

Resource servers are essential when you want to:

• Secure APIs with token-based authentication.
• Control access to resources based on scopes granted to users.
• Simplify managing permissions for multiple APIs under a single user pool.

What Are Scopes?

Scopes define specific permissions or access levels for APIs hosted by the resource server. They determine what actions a user or client application can perform with an access token issued by Cognito.

Each scope consists of:

• Name: A unique identifier for the scope, typically prefixed by the resource server identifier (e.g., resource_server_identifier/scope_name).
• Description: A human-readable description of the scope’s purpose.

For example, if your resource server is an API for a project management application, you might define the following scopes:

• read:projects — Grants read access to project data.
• write:projects — Grants write access to project data.
• delete:projects — Grants delete access to project data.

Scope Management

When an access token is issued, the scopes associated with it determine the specific permissions granted to the user or application. These scopes are included in the token and can be validated by the API to enforce the appropriate access control.

How Resource Servers and Scopes Work Together

1. Define a Resource Server: Register the resource server in your Cognito User Pool. This creates a logical entity to manage APIs under the same identifier.
2. Create Scopes: Define the necessary scopes within the resource server to reflect the permissions required by your application.
3. Configure Clients: Associate the desired scopes with Cognito App Clients. This step ensures that tokens issued to the clients include only the scopes they’re authorized to request.
4. Integrate with APIs: APIs validate the scopes included in the access token to determine if the user or application has the required permissions for the requested action.
5. PIs validate the scopes included in the access token to determine if the user or application has the required permissions for the requested action.

Example

Suppose you’re building an e-commerce platform with an API to manage products and orders. Here’s how you might configure resource servers and scopes:

Resource Server:

• Identifier: https://api.myecommerce.com

Scopes:

• https://api.myecommerce.com/read:products
• https://api.myecommerce.com/write:products
• https://api.myecommerce.com/read:orders
• https://api.myecommerce.com/write:orders

App Client Configuration:

• Admin Client: Full access to all scopes.
• Customer Client: Limited access, e.g., read:products and read:orders only.

API Integration:

• When a user requests product data, the API validates the token to ensure it includes read:products.
• For order submission, the API checks for write:orders.

Lambda Function with Scope Validation

For this example I will use AWS Lambda Powertools, but there are other options as well.

Get and Validate the Access Token (Requesting Side)

    
import jwt
from jwt import PyJWKClient


# FOR THE REQUESTING SIDE
COGNITO_DOMAIN='mydomain.auth.us-east-1.amazoncognito.com',
CLIENT_ID='your_client_id',
CLIENT_SECRET='your_client_secret',


# FOR THE VALIDATION SIDE
COGNITO_REGION = "us-east-1"
USER_POOL_ID = "your_user_pool_id"
JWKS_URL = f"https://cognito-idp.{COGNITO_REGION}.amazonaws.com/{USER_POOL_ID}/.well-known/jwks.json"
RESOURCE_SERVER_IDENTIFIER = "https://api.myecommerce.com"


# Define required scopes for each endpoint
REQUIRED_SCOPES = {
    "/products": [f"{RESOURCE_SERVER_IDENTIFIER}/read:products"],
    "/orders": [f"{RESOURCE_SERVER_IDENTIFIER}/write:orders"]
}


def validate_token_with_jwks(token: str) -> dict:
    """
    Decode and validate the JWT token using Cognito's JWKS.
    """
    try:
        jwks_client = PyJWKClient(JWKS_URL)
        signing_key = jwks_client.get_signing_key_from_jwt(token)

        decoded_token = jwt.decode(
            token,
            signing_key.key,
            algorithms=["RS256"],
            issuer=f"https://cognito-idp.{COGNITO_REGION}.amazonaws.com/{USER_POOL_ID}"
        )
        logger.info("Token is valid.")
        return decoded_token
    except jwt.ExpiredSignatureError:
        logger.error("Token expired")
        raise Exception("Unauthorized: Token expired")
    except jwt.InvalidTokenError as e:
        logger.error(f"Invalid token: {str(e)}")
        raise Exception("Unauthorized: Invalid token")


def has_required_scopes(required_scopes: list, token_scopes: list) -> bool:
    """
    Check if the token contains at least one of the required scopes.
    """
    return any(scope in token_scopes for scope in required_scopes)



def get_access_token():
      # Prepare the Token endpoint
      token_url = f"https://{COGNITO_DOMAIN}/oauth2/token"

      # Encode Client ID and Secret
      client_credentials = f"{CLIENT_ID}:{CLIENT_SECRET}"
      encoded_credentials = base64.b64encode(client_credentials.encode()).decode()

      headers = {
          "Authorization": f"Basic {encoded_credentials}",
          "Content-Type": "application/x-www-form-urlencoded",
      }
      body = {
          'grant_type': 'client_credentials',
      }

      response = requests.post(token_url, headers=headers, data=body)

      if response.status_code == 200:
          token_data = response.json()
          print("Access Token:", token_data["access_token"])
          return token_data["access_token"]
      else:
          print("Failed to get token:", response.text)
          return None



headers = {
        "Authorization": f"Bearer {token}",
}

token = get_access_token()

decoded_token = validate_token_with_jwks(token)
token_scopes = decoded_token.get("scope", "").split()

if not client.has_required_scopes(REQUIRED_SCOPES["/products"], token_scopes):
    print("Authorization error: Required scopes not found")

if not client.has_required_scopes(REQUIRED_SCOPES["/orders"], token_scopes):
    print("Authorization error: Required scopes not found")


api_url = "https://api.myecommerce.com"

response = requests.get(f"{api_url}/products", headers=headers)
print(response)

response = requests.post(f"{api_url}/orders", headers=headers, json={})
print(response)

API Implementation

pip install pyjwt aws-lambda-powertools

import jwt
from jwt import PyJWKClient
from aws_lambda_powertools import Logger, Tracer
from aws_lambda_powertools.event_handler.api_gateway import ApiGatewayResolver, Response
from aws_lambda_powertools.utilities.typing import LambdaContext

# Initialize Powertools components
logger = Logger()
tracer = Tracer()
app = ApiGatewayResolver()

# Replace these with your AWS Cognito details
COGNITO_REGION = "us-east-1"
USER_POOL_ID = "your_user_pool_id"
JWKS_URL = f"https://cognito-idp.{COGNITO_REGION}.amazonaws.com/{USER_POOL_ID}/.well-known/jwks.json"
RESOURCE_SERVER_IDENTIFIER = "https://api.myecommerce.com"

# Define required scopes for each endpoint
REQUIRED_SCOPES = {
    "/products": [f"{RESOURCE_SERVER_IDENTIFIER}/read:products"],
    "/orders": [f"{RESOURCE_SERVER_IDENTIFIER}/write:orders"]
}

def validate_token_with_jwks(token: str) -> dict:
    """
    Decode and validate the JWT token using Cognito's JWKS.
    """
    try:
        jwks_client = PyJWKClient(JWKS_URL)
        signing_key = jwks_client.get_signing_key_from_jwt(token)

        decoded_token = jwt.decode(
            token,
            signing_key.key,
            algorithms=["RS256"],
            issuer=f"https://cognito-idp.{COGNITO_REGION}.amazonaws.com/{USER_POOL_ID}"
        )
        logger.info("Token is valid.")
        return decoded_token
    except jwt.ExpiredSignatureError:
        logger.error("Token expired")
        raise Exception("Unauthorized: Token expired")
    except jwt.InvalidTokenError as e:
        logger.error(f"Invalid token: {str(e)}")
        raise Exception("Unauthorized: Invalid token")

def has_required_scopes(required_scopes: list, token_scopes: list) -> bool:
    """
    Check if the token contains at least one of the required scopes.
    """
    return any(scope in token_scopes for scope in required_scopes)

@app.get("/products")
@tracer.capture_method
def get_products():
    token = app.current_event.get_header_value("Authorization", "").replace("Bearer ", "")
    if not token:
        logger.warning("Missing token")
        return Response(status_code=401, content="Missing token")

    try:
        decoded_token = validate_token_with_jwks(token)
        token_scopes = decoded_token.get("scope", "").split()
        if not has_required_scopes(REQUIRED_SCOPES["/products"], token_scopes):
            logger.warning("Insufficient permissions for /products")
            return Response(status_code=403, content="Insufficient permissions")
    except Exception as e:
        logger.error(f"Authorization error: {str(e)}")
        return Response(status_code=401, content=str(e))

    return {"message": "Here are your products"}

@app.post("/orders")
@tracer.capture_method
def create_order():
    token = app.current_event.get_header_value("Authorization", "").replace("Bearer ", "")
    if not token:
        logger.warning("Missing token")
        return Response(status_code=401, content="Missing token")

    try:
        decoded_token = validate_token_with_jwks(token)
        token_scopes = decoded_token.get("scope", "").split()
        if not has_required_scopes(REQUIRED_SCOPES["/orders"], token_scopes):
            logger.warning("Insufficient permissions for /orders")
            return Response(status_code=403, content="Insufficient permissions")
    except Exception as e:
        logger.error(f"Authorization error: {str(e)}")
        return Response(status_code=401, content=str(e))

    return {"message": "Order created successfully!"}

@logger.inject_lambda_context
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    """
    Lambda handler that processes API Gateway requests.
    """
    return app.resolve(event, context)

Decoded Token

Token is valid. Claims: 
{'sub': 'some_very_big_id', 
'token_use': 'access', 
'scope': 'https://api.myecomerce.com/read:products https://api.myecomerce.com/write:orders'
'auth_time': 1733870778, 
'iss': 'https://cognito-idp.us-east-1.amazonaws.com/your_user_pool_id', 
'exp': 1733874378, 
'iat': 1733870778,
 'version': 2, 
'jti': 'bdc64f16-0000-0000-0000-0000000000', 
'client_id': 'your_client_id'}