Hire us

GO Serverless! Part 3 - Deploy HTTP API to AWS Lambda and Expose it via API Gateway

GO Serverless! Part 3 - Deploy HTTP API to AWS Lambda and Expose it via API Gateway

If you didn't read the previous parts, we highly advise you to do that!


In the previous parts we have discussed the pros and cons of serverless architectures, different serverless tools and technologies, integration patterns between serverless components and more.

We've built the reusable Terraform modules that helped us decouple Lambda code and dependencies continuous-integration from Terraform and delegate it to external CI services like Codebuild and Codepipeline.

Throughout this part, we will build a simple Flask API, deploy it as a Lambda Function with the CI Terraform module we previously built, and Expose it through AWS API Gateway V2 as an HTTP API.

TLDR

If you are in a rush and you don't have time to read the entire article, we've got your back!

TLDR

Our goal is to build a reusable Terraform module for provisioning all resources needed for a functional, low budget and serverless HTTP API, to achieve this we will leverage the Terraform modules from our last part, and we will build other modules, so this is what we need:

  1. AWS Lambda API Code compatible with API Gateway - an API built with any Restfull API framework like Flask API and Fast API, and able to adapt Lambda API Gateway events into HTTP Requests and HTTP Responses into API Gateway Responses. an AWS Lambda Fast API Starter and AWS Lambda Flask API Starter are provided.

  2. Codeless AWS Lambda Function - from the previous part, a reusable Terraform module for provisioning codeless Lambda resources in order for code/dependencies build and deployment to be delegated to an external CI/CD process.

  3. AWS Lambda Function CI/CD - from the previous part, a reusable Terraform module for provisioning pipeline resources that will take care of the Lambda code and dependencies continuous integration/deployment.

  4. AWS API Gateway HTTP API - a Terraform reusable module for provisioning an AWS API Gateway HTTP API that integrates with an Upstream Lambda Function, authorize and proxy requests.

  5. AWS API Gateway APIs Exposer - a Terraform reusable module for provisioning AWS API Gateway resources for a custom domain, mapping the domain with the API Gateway HTTP API and exposing the API through route53 or cloudflare records.

  6. An Identity as a Service Provider - any JWT compatible provider that API Gateway can integrate with for authorizing requests based on users JWT access tokens.

  7. AWS Lambda API - This is the main reusable Terraform module (The Brain!) which processes/trigger other modules that are listed above. It encapsulates Codeless Lambda Function(2), Lambda Function CI/CD(3) and API Gateway HTTP API(4).

The API Gateway APIs Exposer(5) is placed in a separate module because it's more generic and it can be used to expose multiple API Gateway HTTP APIs(4) or even API Gateway Websocket APIs that we will discuss more in the next article.

The Lambda API Code compatible with API Gateway(1) and Identity as a Service Provider(6) components are your responsibility to build and configure but a AWS Lambda Fast API Starter and AWS Lambda Flask API Starter are provided to you for inspiration and demo.

Prerequisites

  • HTTP API application compatible with Lambda/AWS-APIGW (A starter app is provided)
  • Route53 or Cloudflare zone (That you own of course)
  • ACM certificate for your AWS API Gateway custom domain (For HTTPs)
  • Codestar connection to your Github account.
  • S3 bucket for holding CI/CD artifacts and Lambda Code/Dependencies
  • Firebase project or a project in any other IaaS providers.
  • Slack channel(s) for notifying success adn failure deployments.

Usage

  • Version Control your Lambda Function API source code in Github and Provision the AWS Lambda API(7)
module "aws_flask_lambda_api" {
  source      = "git::https://github.com/obytes/terraform-aws-lambda-api.git//modules/api"
  prefix      = "${local.prefix}-flask"
  common_tags = local.common_tags

  # Lambda API
  description                        = "Flask Lambda API"
  runtime                            = "python3.7"
  handler                            = "app.runtime.lambda.main.handler"
  memory_size                        = 512
  envs                               = {
    FIREBASE_APP_API_KEY   = "AIzaSyAbiq3L6lVT9TyM_Lik6C5rgSLEGCiqJhM"
    AWS_API_GW_MAPPING_KEY = "flask"
  }
  policy_json                        = null
  logs_retention_in_days             = 3
  jwt_authorization_groups_attr_name = "groups"

  # CI/CD
  github                          = {
     owner          = "obytes"
     webhook_secret = "not-secret"
     connection_arn = "arn:aws:codestar-connections:us-east-1:{ACCOUNT_ID}:connection/{CONNECTION_ID}"
  }
  pre_release                     = true
  github_repository               = {
    name   = "lambda-flask-api"
    branch = "main"
  }
  s3_artifacts                    = {
     arn    = aws_s3_bucket.artifacts.arn
     bucket = aws_s3_bucket.artifacts.bucket
  }
  app_src_path                    = "src"
  packages_descriptor_path        = "src/requirements/lambda.txt"
  ci_notifications_slack_channels = {
     info  = "ci-info"
     alert = "ci-alert"
  }

  # API Gateway
  stage_name                    = "mvp"
  jwt_authorizer                = {
    issuer   = "https://securetoken.google.com/flask-lambda"
    audience = [ "flask-lambda" ]
  }
  routes_definitions            = {
    health_check = {
      operation_name = "Service Health Check"
      route_key      = "GET /v1/manage/hc"
    }
    token = {
      operation_name = "Get authorization token"
      route_key      = "POST /v1/auth/token"
    }
    whoami = {
      operation_name = "Get user claims"
      route_key      = "GET /v1/users/whoami"
      # Authorization
      api_key_required     = false
      authorization_type   = "JWT"
      authorization_scopes = []
    }
    site_map = {
      operation_name = "Get endpoints list"
      route_key      = "GET /v1/admin/endpoints"
      # Authorization
      api_key_required     = false
      authorization_type   = "JWT"
      authorization_scopes = []
    }
    swagger_specification = {
      operation_name = "Swagger Specification"
      route_key      = "GET /v1/swagger.json"
    }
    swagger_ui = {
      operation_name = "Swagger UI"
      route_key      = "GET /v1/docs"
    }
  }
  access_logs_retention_in_days = 3
}
  • For FastAPI lovers, we've got your back:
module "aws_fast_lambda_api" {
  source      = "git::https://github.com/obytes/terraform-aws-lambda-api.git//modules/api"
  prefix      = "${local.prefix}-fast"
  common_tags = local.common_tags

  # Lambda API
  description                        = "Fast Lambda API"
  runtime                            = "python3.7"
  handler                            = "app.runtime.lambda.main.handler"
  memory_size                        = 512
  envs                               = {
    FIREBASE_APP_API_KEY   = "AIzaSyAbiq3L6lVT9TyM_Lik6C5rgSLEGCiqJhM"
    AWS_API_GW_MAPPING_KEY = "fast"
  }
  policy_json                        = null
  logs_retention_in_days             = 3
  jwt_authorization_groups_attr_name = "groups"

  # CI/CD
  github                          = {
     owner          = "obytes"
     webhook_secret = "not-secret"
     connection_arn = "arn:aws:codestar-connections:us-east-1:{ACCOUNT_ID}:connection/{CONNECTION_ID}"
  }
  pre_release                     = true
  github_repository               = {
    name   = "lambda-fast-api"
    branch = "main"
  }
  s3_artifacts                    = {
     arn    = aws_s3_bucket.artifacts.arn
     bucket = aws_s3_bucket.artifacts.bucket
  }
  app_src_path                    = "src"
  packages_descriptor_path        = "src/requirements/lambda.txt"
  ci_notifications_slack_channels = {
     info  = "ci-info"
     alert = "ci-alert"
  }

  # API Gateway
  stage_name                    = "mvp"
  jwt_authorizer                = {
    issuer   = "https://securetoken.google.com/flask-lambda"
    audience = [ "flask-lambda" ]
  }
  routes_definitions            = {
    health_check = {
      operation_name = "Service Health Check"
      route_key      = "GET /v1/manage/hc"
    }
    token = {
      operation_name = "Get authorization token"
      route_key      = "POST /v1/auth/token"
    }
    whoami = {
      operation_name = "Get user claims"
      route_key      = "GET /v1/users/whoami"
      # Authorization
      api_key_required     = false
      authorization_type   = "JWT"
      authorization_scopes = []
    }
    site_map = {
      operation_name = "Get site map"
      route_key      = "GET /v1/admin/endpoints"
      # Authorization
      api_key_required     = false
      authorization_type   = "JWT"
      authorization_scopes = []
    }
    openapi = {
      operation_name = "OpenAPI Specification"
      route_key      = "GET /v1/openapi.json"
    }
    swagger = {
      operation_name = "Swagger UI"
      route_key      = "GET /v1/docs"
    }
    redoc = {
      operation_name = "ReDoc UI"
      route_key      = "GET /v1/redoc"
    }
  }
  access_logs_retention_in_days = 3
}
  • Provision The AWS API Gateway APIs Exposer(5)
module "gato" {
  source      = "git::https://github.com/obytes/terraform-aws-gato.git//modules/core-route53"
  prefix      = local.prefix
  common_tags = local.common_tags

  # DNS
  r53_zone_id = aws_route53_zone.prerequisite.zone_id
  cert_arn    = aws_acm_certificate.prerequisite.arn
  domain_name = "kodhive.com"
  sub_domains = {
    stateless = "api"
    statefull = "ws"
  }

  # Rest APIS
  http_apis = [
    {
      id    = module.aws_flask_lambda_api.http_api_id
      key   = "flask"
      stage = module.aws_flask_lambda_api.http_api_stage_name
    },
    {
      id    = module.aws_fast_lambda_api.http_api_id
      key   = "fast"
      stage = module.aws_fast_lambda_api.http_api_stage_name
    },
  ]
  ws_apis = []
}

With this configuration, our Lambda API Gateway final base URLs will be https://api.kodhive.com/flask/ for Flask API and https://api.kodhive.com/fast/ for Fast API and these endpoints will be exposed:

Flask Docs
Fast Docs

Demo time, Bring it on!

APIGW Ready

Demo Credentials

Demo

  • Public Endpoint [ALLOW]
curl -X GET https://api.kodhive.com/flask/v1/manage/hc
{
    "status": "I'm sexy and I know It"
}
  • Auth Endpoint [DENY]
curl -X POST -F '[email protected]' -F 'password=not-secret' https://api.kodhive.com/flask/v1/auth/token
{
    "error": {
        "code": "002401",
        "title": "Unauthorized",
        "message": "Access unauthorized",
        "reason": "EMAIL_NOT_FOUND"
    },
    "message": "EMAIL_NOT_FOUND"
}
  • Auth Endpoint [ALLOW]
curl -X POST -F '[email protected]' -F 'password=not-secret' https://api.kodhive.com/flask/v1/auth/token
{
    "kind": "identitytoolkit#VerifyPasswordResponse",
    "localId": "gf30eciYKjVJrA5XMHK0NKDbKeC2",
    "email": "[email protected]",
    "displayName": "Super Admin",
    "registered": true,
    "profilePicture": "https://img2.freepng.fr/20180402/ogw/kisspng-computer-icons-user-profile-clip-art-user-avatar-5ac208105c03d6.9558906215226654883769.jpg",
    "refreshToken": "TOO_LONG_TOKEN",
    "expiresIn": "3600",
    "token_type": "bearer",
    "access_token": "TOO_LONG_TOKEN"
}
  • Private Endpoint [DENY]
curl -X GET https://api.kodhive.com/flask/v1/users/whoami
{"message":"Unauthorized"}%
  • Private Endpoint [ALLOW]
curl -X GET https://api.kodhive.com/flask/v1/users/whoami -H "Authorization: Bearer NORMAL_USER_FIREBASE_JWT_TOKEN"
{
    "claims": {
        "aud": "flask-lambda",
        "auth_time": "1635015339",
        "email": "[email protected]",
        "email_verified": "true",
        "exp": "1635018939",
        "firebase": "map[identities:map[email:[[email protected]]] sign_in_provider:password]",
        "groups": "[USERS ADMINS]",
        "iat": "1635015339",
        "iss": "https://securetoken.google.com/flask-lambda",
        "name": "Hamza Adami",
        "picture": "https://siasky.net/_AlWdFnwvbHwXoDeVk-4DrMKcmQajKIJ2z-maOkXsDfYNw",
        "sub": "NcpGCnZ9B0cFDqRllYbtTYG8awE2",
        "user_id": "NcpGCnZ9B0cFDqRllYbtTYG8awE2"
    }
}
  • Admin Endpoint [DENY]
curl -X GET https://api.kodhive.com/flask/v1/admin/endpoints -H "Authorization: Bearer NORMAL_USER_FIREBASE_JWT_TOKEN"
{
    "error": {
        "code": "002401",
        "title": "Unauthorized",
        "message": "Access unauthorized",
        "reason": "Only ['ADMINS'] can access this endpoint"
    },
    "message": "Only ['ADMINS'] can access this endpoint"
}
  • Admin Endpoint [ALLOW]
curl -X GET https://api.kodhive.com/flask/v1/admin/endpoints -H "Authorization: Bearer ADMIN_USER_FIREBASE_JWT_TOKEN"
{
    "endpoints": [
        {
            "path": "/v1/manage/hc",
            "name": "api.manage_health_check"
        },
        {
            "path": "/v1/admin/endpoints",
            "name": "api.admin_list_endpoints"
        },
        {
            "path": "/v1/users/whoami",
            "name": "api.users_who_am_i"
        },
        {
            "path": "/v1/swagger.json",
            "name": "api.specs"
        },
        {
            "path": "/v1/docs",
            "name": "api.doc"
        },
        {
            "path": "/v1/",
            "name": "api.root"
        }
    ]
}

Notice that even though the request URL has the flask mapping key prefix on it and also the stage name is added in the background by APIGW to the HTTP path sent to Lambda Function but it still works! this is thanks to API Gateway Mapping that strips the mapping key flask from the original path in the background and the lambda function adapter that strips the stage name mvp before matching the APIGW path with Flask route path that does not have the mapping key nor the stage name.

It's working

That's all. enjoy your low budget API. However, if you are not in a rush, continue the article to see how we've built it. you will not regret that.

“You take the blue pill, the story ends, you wake up in your bed and believe whatever you want to believe. You take the red pill, you stay in wonderland, and I show you how deep the rabbit hole goes.” Morpheus

Lambda Compatible API?

Source: AWS Lambda Flask API Starter Source: AWS Lambda Fast API Starter

Possible

You can make any API compatible with Lambda and API Gateway as long as it's written in a supported Lambda runtime and respects the HTTP specifications and standards. for this article we will take Flask APIs as an example.

Well, Most of you knows how to build a Flask application, so we will skip that and go directly to the important part which is how to make the Flask application compatible with Lambda Function and capable of serving requests originated from API Gateway.

Structure

To achieve that, we will need an event/request adapter, The adapter will be responsible of the following:

  • Adapt the received API Gateway event and translate it to a WSGI environ that contains information about the server configuration and client request.

  • Create a WSGI HTTP request, with headers and body taken from the WSGI environment. Which has properties and methods for using the functionality defined by various HTTP specs.

  • Start a WSGI Application to process the request, dispatch it to the target route and return a WSGI HTTP response with body, status, and headers.

  • Adapt the WSGI HTTP Response to a format that API Gateway understand and return it.

# src/app/runtime/lambda/flask_lambda.py
import base64
import sys

from app.api.conf.settings import AWS_API_GW_STAGE_NAME
from .warmer import warmer

try:
    from urllib import urlencode
except ImportError:
    from urllib.parse import urlencode

from flask import Flask
from io import BytesIO

from werkzeug._internal import _to_bytes


def strip_api_gw_stage_name(path: str) -> str:
    if path.startswith(f"/{AWS_API_GW_STAGE_NAME}"):
        return path[len(f"/{AWS_API_GW_STAGE_NAME}"):]
    return path


def adapt(event):
    environ = {'SCRIPT_NAME': ''}

    context = event['requestContext']
    http = context['http']

    # Construct HEADERS
    for hdr_name, hdr_value in event['headers'].items():
        hdr_name = hdr_name.replace('-', '_').upper()
        if hdr_name in ['CONTENT_TYPE', 'CONTENT_LENGTH']:
            environ[hdr_name] = hdr_value
            continue

        http_hdr_name = 'HTTP_%s' % hdr_name
        environ[http_hdr_name] = hdr_value

    # Construct QUERY Params
    qs = event.get('queryStringParameters')
    environ['QUERY_STRING'] = urlencode(qs) if qs else ''

    # Construct HTTP
    environ['REQUEST_METHOD'] = http['method']
    environ['PATH_INFO'] = strip_api_gw_stage_name(http['path'])
    environ['SERVER_PROTOCOL'] = http['protocol']
    environ['REMOTE_ADDR'] = http['sourceIp']
    environ['HOST'] = '%(HTTP_HOST)s:%(HTTP_X_FORWARDED_PORT)s' % environ
    environ['SERVER_PORT'] = environ['HTTP_X_FORWARDED_PORT']
    environ['wsgi.url_scheme'] = environ['HTTP_X_FORWARDED_PROTO']

    # Authorizer
    environ['AUTHORIZER'] = context.get('authorizer')
    environ['IDENTITY'] = context.get('identity')

    # Body
    body = event.get(u"body", "")
    if event.get("isBase64Encoded", False):
        body = base64.b64decode(body)
    if isinstance(body, (str,)):
        body = _to_bytes(body, charset="utf-8")
    environ['CONTENT_LENGTH'] = str(len(body))

    # WSGI
    environ['wsgi.input'] = BytesIO(body)
    environ['wsgi.version'] = (1, 0)
    environ['wsgi.errors'] = sys.stderr
    environ['wsgi.multithread'] = False
    environ['wsgi.run_once'] = True
    environ['wsgi.multiprocess'] = False

    return environ


class LambdaResponse(object):
    def __init__(self):
        self.status = None
        self.response_headers = None

    def start_response(self, status, response_headers, exc_info=None):
        self.status = int(status[:3])
        self.response_headers = dict(response_headers)


class FlaskLambda(Flask):

    @warmer(send_metric=False)
    def __call__(self, event, context):
        response = LambdaResponse()
        response_body = next(self.wsgi_app(
            adapt(event),
            response.start_response
        ))
        res = {
            'statusCode': response.status,
            'headers': response.response_headers,
            'body': response_body.decode("utf-8")
        }
        return res

AWS API Gateway sends Requests HTTP Paths that already contains a stage name to Lambda Function and the Flask application will not be able to match the request with the available target routes.

To make sure all blueprints routes match the path sent from API Gateway, for each request the adapter will strip the API Gateway stage name AWS_API_GW_STAGE_NAME from the original HTTP Path.

The Terraform AWS Lambda API reusable modules will ensure that the same stage name is used for both AWS API Gateway and Flask Application so this logic can work.

Now that we have the adapter in place, we can now use it to create the Flask application and make it as our Lambda Function entrypoint, AKA the handler. additionally, we have to make sure that the handler is set to the Adapter Object which is in our case app.runtime.lambda.main.handler.

# src/app/runtime/lambda/main.py
from __future__ import print_function

from app.blueprints import register_blueprints
from .flask_lambda import FlaskLambda


def create_app():
    # Init app
    app = FlaskLambda(__name__)
    app.url_map.strict_slashes = False
    app.config["RESTX_JSON"] = {"indent": 4}
    register_blueprints(app)
    return app


handler = create_app()

The app is registering a root blueprint for our v1 root resource and 3 sub blueprints:

# src/app/blueprints.py
from app.api.api_v1 import api_v1_blueprint
from app.api.internal.manage import manage_bp
from app.api.internal.admin import admin_bp
from app.api.routes.users import users_bp
from app.api.routes.auth import auth_bp


def register_blueprints(app):
    # Register blueprints
    app.register_blueprint(api_v1_blueprint)
    app.register_blueprint(manage_bp)
    app.register_blueprint(auth_bp)
    app.register_blueprint(admin_bp)
    app.register_blueprint(users_bp)

We are overriding the base_path because it will be used by flask_restx to generate swagger specification base URL, it should include the AWS_API_GW_MAPPING_KEY which is flask in our case in order for the browser to fetch /flask/v1/swagger.json correctly.

We are also adding an OAuth authorizer to the swagger UI, so we can authenticate and test our endpoints directly from swagger instead of using cURL.

# src/app/api/api_v1.py
import os

from flask import Blueprint
from flask_restx import Api

from app.api.conf.settings import AWS_API_GW_MAPPING_KEY
from app.api.docs import get_swagger_ui
from app.api.errors.errors_handler import handle_errors

api_v1_blueprint = Blueprint('api', __name__, url_prefix=f'/v1')


class FlaskAPI(Api):

    @Api.base_path.getter
    def base_path(self):
        """
        The API path

        :rtype: str
        """
        return os.path.join(f"/{AWS_API_GW_MAPPING_KEY}", "v1")


authorizations = {
    'oauth2': {
        'type': 'oauth2',
        'flow': 'password',
        'tokenUrl': os.path.join(f"/{AWS_API_GW_MAPPING_KEY}", "v1/auth/token"),
        'refreshUrl': os.path.join(f"/{AWS_API_GW_MAPPING_KEY}", "v1/auth/refresh"),
    }
}

api_v1 = FlaskAPI(
    api_v1_blueprint,
    version='0.1.0',
    title="Lambda Flask API Starter",
    description="Fast API Starter, Deployed on AWS Lambda and served with AWS API Gateway",
    doc="/docs",
    authorizations=authorizations
)
api_v1.documentation(get_swagger_ui)
handle_errors(api_v1)

Endpoints

Test

To test all use cases we will cook a public endpoint, an authentication endpoint, a private endpoint and an admin endpoint:

  • Public Endpoint: simple health check endpoint.
# src/app/api/internal/manage.py
import logging

from flask import Blueprint
from flask_restx import Resource

from app.api.api_v1 import api_v1

manage_bp = Blueprint('manage', __name__, url_prefix='/v1/manage')
manage_ns = api_v1.namespace('manage', 'Operations related to management.')

logger = logging.getLogger(__name__)


@manage_ns.route('/hc')
class HealthCheck(Resource):

    def get(self):
        """
        Useful to prevent cold start, should be called periodically by another lambda
        """
        return {"status": "I'm sexy and I know It"}, 200
  • Auth Endpoint: OAuth Password Authentication Flow.
# src/app/api/routes/auth.py
import logging

import requests
from flask import Blueprint, request
from flask_restx import Resource

from app.api.api_v1 import api_v1
from app.api.conf.settings import FIREBASE_APP_API_KEY
from app.api.exceptions import LambdaAuthorizationError

auth_bp = Blueprint('auth', __name__, url_prefix='/v1/auth')
auth_ns = api_v1.namespace('auth', 'Operations related to auth.')

logger = logging.getLogger(__name__)


@auth_ns.route("/token")
class Token(Resource):

    def post(self):
        base_path = "https://identitytoolkit.googleapis.com"
        payload = {
            "email": request.form["username"],
            "password": request.form["password"],
            'returnSecureToken': True
        }
        # Post request
        r = requests.post(
            f"{base_path}/v1/accounts:signInWithPassword?key={FIREBASE_APP_API_KEY}",
            data=payload
        )
        keys = r.json().keys()
        # Check for errors
        if "error" in keys:
            error = r.json()["error"]
            raise LambdaAuthorizationError(
                errors=error["message"]
            )
        # success
        auth = r.json()
        auth["token_type"] = "bearer"
        auth["access_token"] = auth.pop("idToken")
        return auth
  • Private Endpoint: whoami endpoint that returns to the calling user his JWT decoded claims.
# src/app/api/routes/users.py
import logging

from flask import Blueprint, g
from flask_restx import Resource

from app.api.api_v1 import api_v1
from app.api.decorators import auth

users_bp = Blueprint('users', __name__, url_prefix='/v1/users')
users_ns = api_v1.namespace('users', 'Operations related to users.')

logger = logging.getLogger(__name__)


@users_ns.route('/whoami')
class WhoAmI(Resource):

    @users_ns.doc(security=[{'oauth2': []}])
    @auth(allowed_groups=['USERS', 'ADMINS'])  # OR Simply @auth
    def get(self):
        """
        Return user specific JWT decoded claims
        """
        return {"claims": g.claims}, 200
  • Admin Endpoint: returns to site admins the available Flask routes as a list.
# src/app/api/internal/admin.py
import logging

from flask import Blueprint, current_app, url_for
from flask_restx import Resource

from app.api.api_v1 import api_v1
from app.api.decorators import auth

admin_bp = Blueprint('admin', __name__, url_prefix='/v1/admin')
admin_ns = api_v1.namespace('admin', 'Operations related to administration')

logger = logging.getLogger(__name__)


def has_no_empty_params(rule):
    defaults = rule.defaults if rule.defaults is not None else ()
    arguments = rule.arguments if rule.arguments is not None else ()
    return len(defaults) >= len(arguments)


@admin_ns.route('/endpoints')
class ListEndpoints(Resource):

    @admin_ns.doc(security=[{'oauth2': []}])
    @auth(allowed_groups=['ADMINS'])
    def get(self):
        """
        Get flask app available urls
        """
        endpoints = []
        for rule in current_app.url_map.iter_rules():
            # Filter out rules we can't navigate to in a browser
            # and rules that require parameters
            if "GET" in rule.methods and has_no_empty_params(rule):
                url = url_for(rule.endpoint, **(rule.defaults or {}))
                endpoints.append({"path": url, "name": rule.endpoint})
        return {"endpoints": endpoints}, 200

Authentication & Authorization

Authorization

Authentication

The public endpoint will be open for all users without prior authentication but how about the private and admin endpoints? They certainly need an authentication system in place, for that we will not reinvent the wheel, and we will leverage an IaaS (Identity as a Service) provider like Firebase.

We have agreed to use an IaaS to authenticate users but how can we verify the users issued JWT access tokens? fortunately, AWS API Gateway can take that burden as it is capable of:

  • Allowing only access tokens that passed the integrity check.
  • Verify that access tokens are not yet expired.
  • Verify that access tokens are issued for an audience which is in the whitelisted audiences list.
  • Verify that access tokens have sufficient OAuth scopes to consume the endpoints.

NOTE: audiences and scopes checks are Authorization checks and not Authentication checks

Authorization

Authorization is an important aspect when building APIs, so we want certain functionalities/endpoints to be available to only a subset of our users. to tackle this, there are two famous approaches Role Based Access Control (RBAC) and OAuth Scopes Authorization.

Role Based Access Control (RBAC)

Authorization can be achieved by implementing a Role Based Access Control (RBAC) model. where we assign each user a role or multiple roles by adding them to groups and then decorate each route with the list of groups that can consume it.

When using an Identity as a Service Provider like Auth0, Firebase or Cognito we have to make sure to assign users to groups and during user's authentication, the JWT token service will embed the user's groups into the JWT Access/ID tokens claims.

After authenticating to Identity Provider, the user can send its JWT access token to API Gateway that will verify the token integrity/expiration and dispatch the request with decoded JWT token to Lambda Function. Finally, the Lambda Function will compare user's groups claim with the whitelisted groups at route level and decide whether to allow it or forbid it.

# src/app/api/decorators.py
from functools import wraps
from typing import List

from flask import g, request

from app.api.conf import settings
from app.api.exceptions import LambdaAuthorizationError
from app.api.auth import decode_jwt_token


def get_claims():
    g.access_token = request.headers.get('Authorization').split()[1]
    if settings.RUNTIME == 'LAMBDA':
        g.claims = request.environ['AUTHORIZER']['jwt']['claims']
        g.username = g.claims['sub']
        g.groups = g.claims.get(settings.JWT_AUTHORIZATION_GROUPS_ATTR_NAME, "[]").strip("[]").split()
    elif settings.RUNTIME == 'CONTAINERIZED':
        g.claims = decode_jwt_token(g.access_token)
        g.username = g.claims['sub']
        g.groups = g.claims.get(settings.JWT_AUTHORIZATION_GROUPS_ATTR_NAME, [])
    else:
        raise Exception("No runtime specified, Please set RUNTIME environment variable!")


def auth(_route=None, allowed_groups: List = None):
    def decorator(route):
        @wraps(route)
        def wrapper(*args, **kwargs):
            get_claims()
            if allowed_groups:
                if not g.groups:
                    raise LambdaAuthorizationError(
                        'The endpoint has authorization check and the caller does not belong to any groups'
                    )
                else:
                    if not any(group in allowed_groups for group in g.groups):
                        raise LambdaAuthorizationError(f'Only {allowed_groups} can access this endpoint')
            return route(*args, **kwargs)

        return wrapper

    if _route:
        return decorator(_route)
    return decorator

This approach comes with many benefits but also with drawbacks:

  • Requests will not be authorized at the API Gateway level, and they need to travel to Lambda Function to run authorization logic.

  • Authorization rules will be writen in backend code, which will be messy from a DevOps perspective but backend developers will favour that because they will have a better visibility when coding/debugging, and they will easily know who can call any endpoint without going to infrastructure code.

OAuth Scopes Authorization

The second approach is by using OAuth Scopes Authorization model, and for each functionality/route we have to:

  • Create an OAuth scope.
  • Assign to users the list of OAuth scopes that they can claim.
  • At AWS API Gateway level specify the list of OAuth scopes that the user should have at least one of them for the API Gateway to let it reach the Lambda Function API.

The advantages of this approach are:

  • The ability to change permissions scopes at Identity Provider and AWS API Gateway Level without changing/deploying new code.
  • Unauthorized requests will be revoked at API Gateway Level and before reaching the Lambda Function.

The Terraform AWS Lambda API module supports this authorization model and you can customize it using the module's routes_definitions Terraform variable.

Deploy The Lambda Function

Source 1: Codeless AWS Lambda Function Source 2: AWS Lambda Function CI/CD

Deploy

To deploy our Lambda Flask Application we will leverage the reusable modules we built in the previous part, the first module will provision the Lambda Function and related resources and the second module will provision the External CI module:

module "flask_api" {
  source      = "git::https://github.com/obytes/terraform-aws-codeless-lambda.git//modules/lambda"
  prefix      = "${local.prefix}-flask-api"
  common_tags = local.common_tags

  envs = {
    RUNTIME = "LAMBDA"

    # API
    AWS_API_GW_STAGE_NAME = "mvp"

    # Authentication/Authorization
    JWT_AUTHORIZATION_GROUPS_ATTR_NAME = "groups"
  }
}

module "flask_api_ci" {
  source      = "git::https://github.com/obytes/terraform-aws-lambda-ci.git//modules/ci"
  prefix      = "${local.prefix}-flask-api-ci"
  common_tags = local.common_tags

  # Lambda
  lambda                   = module.flask_api.lambda
  app_src_path             = "src"
  packages_descriptor_path = "src/requirements/lambda.txt"

  # Github
  s3_artifacts      = {
    arn    = aws_s3_bucket.artifacts.arn
    bucket = aws_s3_bucket.artifacts.bucket
  }
  github            = {
    connection_arn = "arn:aws:codestar-connections:us-east-1:[YOUR_ACCOUNT_ID]:connection/[YOUR_CONNECTION_ID]"
    owner          = "YOUR_GITHUB_USERNAME"
    webhook_secret = "YOUR_WEBHOOK_SECRET"
  }
  pre_release       = true
  github_repository = {
    name = "lambda-flask-api"
    branch = "main"
  }

  # Notifications
  ci_notifications_slack_channels = var.ci_notifications_slack_channels
}

The RUNTIME environment variable is used by Flask Application to distinguish between LAMBDA and CONTAINERIZED runtimes so the application can work as a lambda or as a docker container.

The AWS_API_GW_STAGE_NAME environment variable as discussed earlier, It should be set to the same stage as the AWS API Gateway.

The JWT_AUTHORIZATION_GROUPS_ATTR_NAME environment is passed to the Flask application as the claim attribute name that will be used by authentication decorator for RBAC authorization.

The AWS API Gateway

Source: AWS API Gateway HTTP API

Home

The first thing we do is to create an API Gateway V2 HTTP API, we specify the selection expressions for method, path and api key, and we configure CORS to allow all headers, methods and origins.

# modules/gw/gateway_http_api.tf
resource "aws_apigatewayv2_api" "_" {
  name          = "${local.prefix}-http-api"
  description   = "Lambda HTTP API"
  protocol_type = "HTTP"

  route_selection_expression   = "$request.method $request.path"
  api_key_selection_expression = "$request.header.x-api-key"

  cors_configuration {
    allow_credentials = false

    allow_headers = [
      "*"
    ]
    allow_methods = [
      "*"
    ]
    allow_origins = [
      "*"
    ]
    expose_headers = [
      "*"
    ]
  }

  tags          = local.common_tags
}

Next, we create the integration with the upstream service which is in this case our Lambda Function, the integration type is set to AWS_PROXY for direct interactions between the client and the integrated Lambda function. and we've chosen INTERNET as connection type because the connection is through the public routable internet and not inside a VPC.

# modules/gw/gateway_integrations.tf
resource "aws_apigatewayv2_integration" "_" {
  api_id                 = aws_apigatewayv2_api._.id
  description            = "Lambda Serverless Upstream Service"

  passthrough_behavior   = "WHEN_NO_MATCH"
  payload_format_version = "2.0"

  # Upstream
  integration_type     = "AWS_PROXY"
  integration_uri      = var.api_lambda.invoke_arn
  connection_type      = "INTERNET"
  integration_method   = "POST"
  timeout_milliseconds = 29000

  lifecycle {
    ignore_changes = [
      passthrough_behavior
    ]
  }
}

For the authorizers, we need a JWT Authorizer because we will leverage a token-based Authentication and Authorization standard to allow an application to access our API. however, the routes can also support NONE for open access mode and IAM for authorization with IAM STS tokens generated by Cognito Identity Pools.

The JWT issuer(iss) and audience(aud) depends on the IaaS provider that you will use. in our case we are using Firebase. so these are the issuer and audience format:

For AWS Cognito:

For Auth0:

  • issuer` - https://[YOUR_AUTH0_DOMAIN]/
  • audience - YOUR_AUTH0_API_ID
# modules/gw/gateway_authorizers.tf
resource "aws_apigatewayv2_authorizer" "_" {
  name             = "${var.prefix}-jwt-authz"
  api_id           = aws_apigatewayv2_api._.id
  authorizer_type  = "JWT"
  identity_sources = ["$request.header.Authorization"]

  jwt_configuration {
    issuer   = var.jwt_authorizer.issuer
    audience = var.jwt_authorizer.audience
  }
}

locals {
  authorizers_ids = {
    JWT  = aws_apigatewayv2_authorizer._.id
    IAM  = null
    NONE = null
  }
}

After that, we will create the most important resources which are the routes, we are using the for_each to create a route for every route definition element provided in routes definitions variable, the upstream target is our API Gateway Lambda integration and the authorization will be dynamically configured:

resource "aws_apigatewayv2_route" "routes" {
  for_each       = var.routes_definitions
  api_id         = aws_apigatewayv2_api._.id

  # UPSTREAM
  target         = "integrations/${aws_apigatewayv2_integration._.id}"
  route_key      = each.value.route_key
  operation_name = each.value.operation_name

  # AUTHORIZATION
  authorizer_id        = lookup(local.authorizers_ids, lookup(each.value, "authorization_type", "NONE"), null)
  api_key_required     = lookup(each.value, "api_key_required", false)
  authorization_type   = lookup(each.value, "authorization_type", "NONE")
  authorization_scopes = lookup(each.value, "authorization_scopes", null)
}

Now that we have prepared all the required resources for the API Gateway to integrate with Lambda Function we should add a permission rule to the Upstream Lambda Function allowing invocations from API Gateway:

# modules/gw/permission.tf
resource "aws_lambda_permission" "_" {
  statement_id  = "AllowExecutionFromAPIGatewayV2"
  action        = "lambda:InvokeFunction"
  function_name = var.api_lambda.name
  principal     = "apigateway.amazonaws.com"
  qualifier     = var.api_lambda.alias
  source_arn    = "${aws_apigatewayv2_api._.execution_arn}/${aws_apigatewayv2_stage._.name}/*/*"
}

Finally, we deploy our HTTP API to an API Gateway Stage, we've opted for automatic deploy to avoid manual deploys each time we change routes definitions:

# modules/gw/gateway_stage.tf
resource "aws_apigatewayv2_stage" "_" {
  name        = var.stage_name
  api_id      = aws_apigatewayv2_api._.id
  description = "Default Stage"
  auto_deploy = true

  access_log_settings {
    format          = jsonencode(local.access_logs_format)
    destination_arn = aws_cloudwatch_log_group.access.arn
  }

  lifecycle {
    ignore_changes = [
      deployment_id,
      default_route_settings
    ]
  }
}

The reusable Terraform Lambda APIGW module can be called like this:

module "flask_api_gw" {
  source      = "git::https://github.com/obytes/terraform-aws-lambda-apigw.git//modules/gw"
  prefix      = local.prefix
  common_tags = local.common_tags

  stage_name     = "mvp"
  api_lambda     = {
    name       = aws_lambda_function.function.function_name
    arn        = aws_lambda_function.function.arn
    runtime    = aws_lambda_function.function.runtime
    alias      = aws_lambda_alias.alias.name
    invoke_arn = aws_lambda_alias.alias.invoke_arn
  }
  jwt_authorizer = {
    issuer   = "https://securetoken.google.com/flask-lambda"
    audience = [ "flask-lambda" ]
  }
  routes_definitions = {
    health_check = {
      operation_name = "Service Health Check"
      route_key      = "GET /v1/manage/hc"
    }
    token = {
      operation_name = "Get authorization token"
      route_key      = "POST /v1/auth/token"
    }
    whoami = {
      operation_name = "Get user claims"
      route_key      = "GET /v1/users/whoami"
      # Authorization
      api_key_required     = false
      authorization_type   = "JWT"
      authorization_scopes = []
    }
    site_map = {
      operation_name = "Get endpoints list"
      route_key      = "GET /v1/admin/endpoints"
      # Authorization
      api_key_required     = false
      authorization_type   = "JWT"
      authorization_scopes = []
    }
    swagger_specification = {
      operation_name = "Swagger Specification"
      route_key      = "GET /v1/swagger.json"
    }
    swagger_ui = {
      operation_name = "Swagger UI"
      route_key      = "GET /v1/docs"
    }
  }
  access_logs_retention_in_days = 3
}

Expose it!

Source: AWS API Gateway APIs Exposer

Expose

After deploying the Flask Application as a Lambda Function and integrate it with API Gateway, now we need to expose it to the outer world with a beautiful domain name instead of the ugly one generated by AWS.

We need these prerequisites before exposing our API:

  • AWS route53 or cloudflare zone.
  • AWS ACM Certificate for the subdomain that we will use with our API.
  • An A record pointing to APEX domain (the custom domain creation will fail otherwise).

If you already have these requirements let's create our custom API Gateway domain which will replace the default invoke URL provided by API gateway:

# modules/core-route53/gateway_domains.tf
resource "aws_apigatewayv2_domain_name" "stateless" {
  domain_name = "${var.sub_domains.stateless}.${var.domain_name}"

  domain_name_configuration {
    certificate_arn = var.cert_arn
    endpoint_type   = "REGIONAL"
    security_policy = "TLS_1_2"
  }
}

After that we will create an API Gateway Mapping to map the deployed API/Stage with the custom domain and the mapping key for our Lambda API will be flask. count is used here to support multiple HTTP APIs in case we have multiple Lambda APIs.

# modules/core-route53/microservices.tf
resource "aws_apigatewayv2_api_mapping" "stateless_microservices" {
  count           = length(var.http_apis)
  stage           = var.http_apis[count.index].stage
  api_id          = var.http_apis[count.index].id
  api_mapping_key = var.http_apis[count.index].key
  domain_name     = aws_apigatewayv2_domain_name.stateless.domain_name
}

Consider the mapping key as API microservice namespace, for example in a shopping site we can have accounts, products and payments microservices, so we will have three APIs with three different mapping keys (prefixes), this will expose these URLs:

Finally, we create the DNS record pointing to our API Gateway target domain name:

# modules/core-route53/records.tf
resource "aws_route53_record" "stateless" {
  zone_id = var.r53_zone_id
  name    = var.sub_domains.stateless
  type    = "A"

  alias {
    name                   = aws_apigatewayv2_domain_name.stateless.domain_name_configuration[0].target_domain_name
    zone_id                = aws_apigatewayv2_domain_name.stateless.domain_name_configuration[0].hosted_zone_id
    evaluate_target_health = true
  }
}

If you prefer Cloudflare, the reusable module can also create Cloudflare records:

# modules/core-cloudflare/records.tf
resource "cloudflare_record" "stateless" {
  zone_id = var.cloudflare_zone_id
  name    = var.sub_domains.stateless
  type    = "CNAME"
  proxied = true
  value   = aws_apigatewayv2_domain_name.stateless.domain_name_configuration[0].target_domain_name
}

The reusable AWS API Gateway APIs Exposer module can be called like this:

module "gato" {
  source      = "git::https://github.com/obytes/terraform-aws-gato//modules/core-route53"
  prefix      = local.prefix
  common_tags = local.common_tags

  # DNS
  r53_zone_id = aws_route53_zone.prerequisite.zone_id
  cert_arn    = aws_acm_certificate.prerequisite.arn
  domain_name = "kodhive.com"
  sub_domains = {
    stateless = "api"
    statefull = "live"
  }

  # Rest APIS
  http_apis = [
    {
      id    = module.flask_api_gw.http_api_id
      key   = "flask"
      stage = module.flask_api_gw.http_api_stage_name
    },
  ]

  ws_apis = []
}

What's next?

Share

HTTP APIs operates on stateless protocol and usually the client sends requests, and then the server responds with requested data. There is no generic way for the server to communicate with the client on its own. This is referred to as a one-way communication.

Next we will see how we will build a Serverless WebSocket API that works over persistent TCP communication, and it's possible for both the server and client to send data independent of each other, This is referred to as bi-directional communication.

Share if you like the article and Stay tuned for the next article. it's just the beginning!

Hamza Adami
Hamza Adami
2021-11-02 | 26 min read
Share article

More articles