Tag: AWS

Sending logs to AWS CloudWatch with a sidecar pattern and Python

~ Gonzalo Ayuso ~ Leave a comment

In a Docker Swarm environment, the sidecar pattern is a common architectural approach used to extend the functionality of a primary container without directly modifying it.

First there’s a primary container. This is the main application or service you want to run within a Docker Swarm service. It’s responsible for the core functionality of your application. In our case, this container will be the log generator. It will save logs within a json format.

The sidecar container is a secondary container that runs alongside the primary container. It’s tightly coupled with the primary container and assists it by providing additional services or functionality. In our example, the sidecar responsibility will be push logs to AWS CloudWatch. The idea is sharing a docker volume between both containers. Whit this technique, our primary container will not be affected in network latency generating logs.

The idea is to generate something like this:

version: '3.6'

services:
  api:
    image: api:latest
    logging:
      options:
        max-size: 10m
    deploy:
      restart_policy:
        condition: any
    volumes:
      - logs_volume:/src/logs
    environment:
      - ENVIRONMENT=production
      - PROCESS_ID=api
    ports:
      - 5000:5000
    command: gunicorn -w 1 app:app -b 0.0.0.0:5000 --timeout 180

  filebeat:
    image: cw:production
    command: bash -c "python cw.py && sleep 1m"
    deploy:
      restart_policy:
        condition: any
    environment:
      - LOG_GROUP=python_logs_example
      - LOG_STREAM_PREFIX=default
    volumes:
      - logs_volume:/src/logs
volumes:
  logs_volume:

Let’s go. First, we need to setup our aws credentials. We can use a profile or a IAM user

if AWS_PROFILE_NAME:
    session = boto3.Session(profile_name=AWS_PROFILE_NAME, region_name=AWS_REGION)
else:
    session = boto3.Session(
        aws_access_key_id=AWS_ACCESS_KEY_ID,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
        region_name=AWS_REGION)

logs = session. Client('logs')

then we need to setup the logGroup and logStream in CloudWatch. We can generate them by hand, but I preffer to generate them programatically, if they don’t exist.

def init_cloudwatch_stream():
    log_stream_name = f"{LOG_STREAM_PREFIX}_{datetime.now().strftime('%Y%m%d')}"

    log_groups = logs.describe_log_groups(logGroupNamePrefix=LOG_GROUP)['logGroups']
    if not any(group['logGroupName'] == LOG_GROUP for group in log_groups):
        logs.create_log_group(logGroupName=LOG_GROUP)

    log_streams = logs.describe_log_streams(
        logGroupName=LOG_GROUP,
        logStreamNamePrefix=log_stream_name
    )['logStreams']

    if not any(stream['logStreamName'] == log_stream_name for stream in log_streams):
        logs.create_log_stream(logGroupName=LOG_GROUP, logStreamName=log_stream_name)

    return log_stream_name

Now we need to upload logs to CloudWatch. We need to use put_log_events. To send multiple logs we need to use a sequenceToken (not the first time). To do that I use this trick.

function_parameters = dict(
    logGroupName=LOG_GROUP,
    logStreamName=log_stream_name
)

for f in glob(f'{LOG_PATH}/*.{LOG_EXTENSION}'):
    function_parameters['logEvents'] = get_log_events_from_file(f)
    response = logs.put_log_events(**function_parameters)
    function_parameters['sequenceToken'] = response['nextSequenceToken']

We also need to read log files and maybe change the fields according to your needs

def get_log_events_from_file(file):
    exclude_fields = ('@timestamp', 'logger')
    return [
        dict(
            timestamp=int(datetime.fromisoformat(d['@timestamp']).timestamp() * 1000),
            message=json.dumps({k: v for k, v in d.items() if k not in exclude_fields})
        ) for d in [json.loads(linea) for linea in open(file, 'r')]]

I like to have all the settings of my application in a file called settings.py. It’s a pattern that I’ve copied from Django. In this file also I read environment variables from a dotenv file.

import os
from pathlib import Path

from dotenv import load_dotenv

BASE_DIR = Path(__file__).resolve().parent
ENVIRONMENT = os.getenv('ENVIRONMENT', 'local')

load_dotenv(dotenv_path=Path(BASE_DIR).resolve().joinpath('env', ENVIRONMENT, '.env'))

AWS_ACCESS_KEY_ID = os.getenv('AWS_ACCESS_KEY_ID')
AWS_SECRET_ACCESS_KEY = os.getenv('AWS_SECRET_ACCESS_KEY')

AWS_PROFILE_NAME = os.getenv('AWS_PROFILE_NAME', False)
AWS_REGION = os.getenv('AWS_REGION')

LOG_GROUP = os.getenv('LOG_GROUP', 'python_logs_example')
LOG_STREAM_PREFIX = os.getenv('LOG_STREAM_PREFIX', 'default')

LOG_EXTENSION = 'log'
LOG_PATH = os.getenv('LOG_PATH', Path(BASE_DIR).resolve())

And that’s all. Your logs in CloudWatch uploaded in a background process decoupled from the main process.

Source code available in my github.

Real-time Object Detection and Streaming with Python, OpenCV, AWS Kinesis, and YOLOv8

~ Gonzalo Ayuso ~ 2 Comments

I’ve got the following idea in my mind. I want to detect objects with YoloV8 and OpenCV. Here a simple example:

import cv2 as cv
from ultralytics import YOLO
import logging

logging.getLogger("ultralytics").setLevel(logging.WARNING)

logging.basicConfig(
    format='%(asctime)s [%(levelname)s] %(message)s',
    level='INFO',
    datefmt='%d/%m/%Y %X')

logger = logging.getLogger(__name__)

X_RESIZE = 800
Y_RESIZE = 450

YOLO_MODEL="yolov8m.pt"
YOLO_CLASSES_TO_DETECT = 'bottle',
THRESHOLD = 0.80

cap = cv.VideoCapture(0)
model = YOLO(YOLO_MODEL)
while True:
    _, original_frame = cap.read()
    frame = cv.resize(original_frame, (X_RESIZE, Y_RESIZE), interpolation=cv.INTER_LINEAR)

    results = model.predict(frame)

    color = (0, 255, 0)
    for result in results:
        for box in result.boxes:
            class_id = result.names[box.cls[0].item()]
            status = id == class_id
            cords = box.xyxy[0].tolist()
            probability = round(box.conf[0].item(), 2)

            if probability > THRESHOLD and class_id in YOLO_CLASSES_TO_DETECT:
                logger.info(f"{class_id} {round(probability * 100)}%")
                x1, y1, x2, y2 = [round(x) for x in cords]
                cv.rectangle(frame, (x1, y1), (x2, y2), color, 2)

    cv.imshow(f"Cam", frame)

    if cv.waitKey(1) & 0xFF == ord('q'):
        break

cap.release()
cv.destroyAllWindows()

However, my idea is slightly different. I intend to capture the camera stream, transmit it to AWS Kinesis, and subsequently, within a separate process, ingest this Kinesis stream for object detection or any other purpose using the camera frames. To achieve this, our initial step is to publish our OpenCV frames into Kinesis. producer.py

import logging

from lib.video import kinesis_producer
from settings import KINESIS_STREAM_NAME, CAM_URI, WIDTH

logging.basicConfig(
    format='%(asctime)s [%(levelname)s] %(message)s',
    level='INFO',
    datefmt='%d/%m/%Y %X')

logger = logging.getLogger(__name__)

if __name__ == '__main__':
    kinesis_producer(
        kinesis_stream_name=KINESIS_STREAM_NAME,
        cam_uri=CAM_URI,
        width=WIDTH
    )

def kinesis_producer(kinesis_stream_name, cam_uri, width):
    logger.info(f"start emitting stream from cam={cam_uri} to kinesis")
    kinesis = aws_get_service('kinesis')
    cap = cv.VideoCapture(cam_uri)
    try:
        while True:
            ret, frame = cap.read()
            if ret:
                scale = width / frame.shape[1]
                height = int(frame.shape[0] * scale)

                scaled_frame = cv.resize(frame, (width, height))
                _, img_encoded = cv.imencode('.jpg', scaled_frame)
                kinesis.put_record(
                    StreamName=kinesis_stream_name,
                    Data=img_encoded.tobytes(),
                    PartitionKey='1'
                )

    except KeyboardInterrupt:
        cap.release()

Now our consumer.py

import logging

from lib.video import kinesis_consumer
from settings import KINESIS_STREAM_NAME

logging.getLogger("ultralytics").setLevel(logging.WARNING)

logging.basicConfig(
    format='%(asctime)s [%(levelname)s] %(message)s',
    level='INFO',
    datefmt='%d/%m/%Y %X')

logger = logging.getLogger(__name__)

if __name__ == '__main__':
    kinesis_consumer(
        kinesis_stream_name=KINESIS_STREAM_NAME
    )

To get the frames from kinesis stream we need to consider the shards. According to your needs you should change it a little bit.

def kinesis_consumer(kinesis_stream_name):
    logger.info(f"start reading kinesis stream from={kinesis_stream_name}")
    kinesis = aws_get_service('kinesis')
    model = YOLO(YOLO_MODEL)
    response = kinesis.describe_stream(StreamName=kinesis_stream_name)
    shard_id = response['StreamDescription']['Shards'][0]['ShardId']
    while True:
        shard_iterator_response = kinesis.get_shard_iterator(
            StreamName=kinesis_stream_name,
            ShardId=shard_id,
            ShardIteratorType='LATEST'
        )
        shard_iterator = shard_iterator_response['ShardIterator']
        response = kinesis.get_records(
            ShardIterator=shard_iterator,
            Limit=10
        )

        for record in response['Records']:
            image_data = record['Data']
            frame = cv.imdecode(np.frombuffer(image_data, np.uint8), cv.IMREAD_COLOR)
            results = model.predict(frame)
            if detect_id_in_results(results):
                logger.info('Detected')

def detect_id_in_results(results):
    status = False
    for result in results:
        for box in result.boxes:
            class_id = result.names[box.cls[0].item()]
            probability = round(box.conf[0].item(), 2)
            if probability > THRESHOLD and class_id in YOLO_CLASSES_TO_DETECT:
                status = True
    return status

And that’s all. Real-time Object Detection with Kinesis stream. However, computer vision examples are often simple and quite straightforward. On the other hand, real-world computer vision projects tend to be more intricate, involving considerations such as cameras, lighting conditions, performance optimization, and accuracy. But remember, this is just an example.

Full code in my github.

AWS SQS 2 HTTP server with Python

~ Gonzalo Ayuso ~ Leave a comment

Simple service that listens to a sqs queue and bypass the payload to a http server, sending a POST request with the SQS’s payload and a Bearer Token.

The main script read a .env file with the aws credentials and http endpoint.

# AWS CREDENTIALS
AWS_ACCESS_KEY_ID=xxx
AWS_SECRET_ACCESS_KEY=xxx
AWS_REGION=eu-west-1
#AWS_PROFILE=xxx
# SQS
SQS_QUEUE_URL=https://sqs.eu-west-1.amazonaws.com/xxx/name
# REST ENDPOINT
HTTP_ENDPOINT=http://localhost:5555
AUTH_BEARER_TOKEN=xxx

import logging

from dotenv import load_dotenv, find_dotenv

_ = load_dotenv(find_dotenv())

from lib.sqs2http import loop_step, get_sqs

for library in ['botocore', 'boto3']:
    logging.getLogger(library).setLevel(logging.WARNING)

logging.basicConfig(
    format='%(asctime)s [%(levelname)s] %(message)s',
    level='INFO',
    datefmt='%d/%m/%Y %X'
)

logger = logging.getLogger(__name__)

if __name__ == "__main__":
    logger.info("sqs2http start")
    sqs = get_sqs()
    while True:
        loop_step(sqs)

The main loop is like this:

def loop_step(sqs):
    response = sqs.receive_message(
        QueueUrl=SQS_QUEUE_URL,
        MaxNumberOfMessages=SQS_MAX_NUMBER_OF_MESSAGES)

    if 'Messages' in response:
        for message in response.get('Messages', []):
            try:
                body = message.get('Body')
                logger.info(body)
                do_post(body)
                remove_from_sqs(sqs, message.get('ReceiptHandle'))

            except Exception as e:
                logger. Exception(e)

To post to the http server we can use requests.

import requests
import json

from settings import HTTP_ENDPOINT, AUTH_BEARER_TOKEN

def do_post(body):
    requests.post(
        url=HTTP_ENDPOINT,
        data=json.dumps(body),
        headers={
            'Authorization': AUTH_BEARER_TOKEN,
            'Content-Type': 'application/json'
        })

With this simple script you can set up a web server that process all incoming SQS messages. Source code in my github.

Deploying Django application to AWS EC2 instance with Docker

~ Gonzalo Ayuso ~ Leave a comment

In AWS we have several ways to deploy Django (and not Django applications) with Docker. We can use ECS or EKS clusters. If we don’t have one ECS or Kubernetes cluster up and running, maybe it can be complex. Today I want to show how deploy a Django application in production mode within a EC2 host. Let’s start.

I’m getting older to provision one host by hand I prefer to use docker. The idea is create one EC2 instance (one simple Amazon Linux AMI AWS-supported image). This host don’t have docker installed. We need to install it. When we launch one instance, when we’re configuring the instance, we can specify user data to configure an instance or run a configuration script during launch.

We only need to put this shell script to set up docker

#! /bin/bash
yum update -y
yum install -y docker
usermod -a -G docker ec2-user
curl -L https://github.com/docker/compose/releases/download/1.25.5/docker-compose-`uname -s`-`uname -m` | sudo tee /usr/local/bin/docker-compose > /dev/null
chmod +x /usr/local/bin/docker-compose
service docker start
chkconfig docker on

rm /etc/localtime
ln -s /usr/share/zoneinfo/Europe/Madrid /etc/localtime

ln -s /usr/local/bin/docker-compose /usr/bin/docker-compose

docker swarm init

We also need to attach one IAM role to our instance. This IAM role only need to allow us the following policies:

  • AmazonEC2ContainerRegistryReadOnly (because we’re going to use AWS ECR as container registry)
  • CloudWatchAgentServerPolicy (because we’re going to emit our logs to Cloudwatch)

Also we need to set up a security group to allow incoming SSH connections to port 22 and HTTP connections (in our example to port 8000)

When we launch our instance we need to provide a keypair to connect via ssh. I like to put this keypair in my .ssh/config

Host xxx.eu-central-1.compute.amazonaws.com
    User ec2-user
    Identityfile ~/.ssh/keypair-xxx.pem

To deploy our application we need to follow those steps:

  • Build our docker images
  • Push our images to a container registry (in this case ECR)
  • Deploy the application.

I’ve created a simple shell script called deploy.sh to perform all tasks:

#!/usr/bin/env bash

set -a
[ -f deploy.env ] && . deploy.env
set +a

echo "$(tput setaf 1)Building docker images ...$(tput sgr0)"
docker build -t ec2-web -t ec2-web:latest -t $ECR/ec2-web:latest .
docker build -t ec2-nginx -t $ECR/ec2-nginx:latest .docker/nginx

echo "$(tput setaf 1)Pusing to ECR ...$(tput sgr0)"
aws ecr get-login-password --region $REGION --profile $PROFILE |
  docker login --username AWS --password-stdin $ECR
docker push $ECR/ec2-web:latest
docker push $ECR/ec2-nginx:latest

CMD="docker stack deploy -c $DOCKER_COMPOSE_YML ec2 --with-registry-auth"
echo "$(tput setaf 1)Deploying to EC2 ($CMD)...$(tput sgr0)"
echo "$CMD"

DOCKER_HOST="ssh://$HOST" $CMD
echo "$(tput setaf 1)Building finished $(date +'%Y%m%d.%H%M%S')$(tput sgr0)"

This script assumes that there’s a deploy.env file with our personal configuration (AWS profile, the host of the EC2, instance, The ECR and things like that)

PROFILE=xxxxxxx

DOKER_COMPOSE_YML=docker-compose.yml
HOST=ec2-user@xxxx.eu-central-1.compute.amazonaws.com

ECR=9999999999.dkr.ecr.eu-central-1.amazonaws.com
REGION=eu-central-1

In this example I’m using docker swarm to deploy the application. I want to play also with secrets. This dummy application don’t have any sensitive information but I’ve created one "ec2.supersecret" variable

echo "super secret text" | docker secret create ec2.supersecret -

That’s the docker-compose.yml file:

version: '3.8'
services:
  web:
    image: 999999999.dkr.ecr.eu-central-1.amazonaws.com/ec2-web:latest
    command: /bin/bash ./docker-entrypoint.sh
    environment:
      DEBUG: 'False'
    secrets:
      - ec2.supersecret
    deploy:
      replicas: 1
    logging:
      driver: awslogs
      options:
        awslogs-group: /projects/ec2
        awslogs-region: eu-central-1
        awslogs-stream: app
    volumes:
      - static_volume:/src/staticfiles
  nginx:
    image: 99999999.dkr.ecr.eu-central-1.amazonaws.com/ec2-nginx:latest
    deploy:
      replicas: 1
    logging:
      driver: awslogs
      options:
        awslogs-group: /projects/ec2
        awslogs-region: eu-central-1
        awslogs-stream: nginx
    volumes:
      - static_volume:/src/staticfiles:ro
    ports:
      - 8000:80
    depends_on:
      - web
volumes:
  static_volume:

secrets:
  ec2.supersecret:
    external: true

And that’s all. Maybe ECS or EKS are better solutions to deploy docker applications in AWS, but we also can deploy easily to one docker host in a EC2 instance that it can be ready within a couple of minutes.

Source code in my github

Django reactive users with Celery and Channels

~ Gonzalo Ayuso ~ Leave a comment

Today I want to build a prototype. The idea is to create two Django applications. One application will be the master and the other one will the client. Both applications will have their User model but each change within master User model will be propagated through the client (or clients). Let me show you what I’ve got in my mind:

We’re going to create one signal in User model (at Master) to detect user modifications:

  • If certain fields have been changed (for example we’re going to ignore last_login, password and things like that) we’re going to emit a event
  • I normally work with AWS, so the event will be a SNS event.
  • The idea to have multiple clients, so each client will be listening to one SQS queue. Those SQSs queues will be mapped to the SNS event.
  • To decouple the SNS sending og the message we’re going to send it via Celery worker.
  • The second application (the Client) will have one listener to the SQS queue.
  • Each time the listener have a message it will persists the user information within the client’s User model
  • And also it will emit on message to one Django Channel’s consumer to be sent via websockets to the browser.

The Master

We’re going to emit the event each time the User model changes (and also when we create or delete one user). To detect changes we’re going to register on signal in the pre_save to mark if the model has been changed and later in the post_save we’re going to emit the event via Celery worker.

@receiver(pre_save, sender=User)
def pre_user_modified(sender, instance, **kwargs):
    instance.is_modified = None

    if instance.is_staff is False and instance.id is not None:
        modified_user_data = UserSerializer(instance).data
        user = User.objects.get(username=modified_user_data['username'])
        user_serializer_data = UserSerializer(user).data

        if user_serializer_data != modified_user_data:
            instance.is_modified = True

@receiver(post_save, sender=User)
def post_user_modified(sender, instance, created, **kwargs):
    if instance.is_staff is False:
        if created or instance.is_modified:
            modified_user_data = UserSerializer(instance).data
            user_changed_event.delay(modified_user_data, action=Actions.INSERT if created else Actions.UPDATE)

@receiver(post_delete, sender=User)
def post_user_deleted(sender, instance, **kwargs):
    deleted_user_data = UserSerializer(instance).data
    user_changed_event.delay(deleted_user_data, action=Actions.DELETE)

We need to register our signals in apps.py

from django.apps import AppConfig

class MasterConfig(AppConfig):
    name = 'master'

    def ready(self):
        from master.signals import pre_user_modified
        from master.signals import post_user_modified
        from master.signals import post_user_deleted

Our Celery task will send the message to sns queue

@shared_task()
def user_changed_event(body, action):
    sns = boto3.client('sns')
    message = {
        "user": body,
        "action": action
    }
    response = sns.publish(
        TargetArn=settings.SNS_REACTIVE_TABLE_ARN,
        Message=json.dumps({'default': json.dumps(message)}),
        MessageStructure='json'
    )
    logger.info(response)

AWS

In Aws We need to create one SNS messaging service and one SQS queue linked to this SNS.

The Client

First we need one command to run the listener.

class Actions:
    INSERT = 0
    UPDATE = 1
    DELETE = 2

switch_actions = {
    Actions.INSERT: insert_user,
    Actions.UPDATE: update_user,
    Actions.DELETE: delete_user,
}

class Command(BaseCommand):
    help = 'sqs listener'

    def handle(self, *args, **options):
        self.stdout.write(self.style.WARNING("starting listener"))
        sqs = boto3.client('sqs')

        queue_url = settings.SQS_REACTIVE_TABLES

        def process_message(message):
            decoded_body = json.loads(message['Body'])
            data = json.loads(decoded_body['Message'])

            switch_actions.get(data['action'])(
                data=data['user'],
                timestamp=message['Attributes']['SentTimestamp']
            )

            notify_to_user(data['user'])

            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle'])

        def loop():
            response = sqs.receive_message(
                QueueUrl=queue_url,
                AttributeNames=[
                    'SentTimestamp'
                ],
                MaxNumberOfMessages=10,
                MessageAttributeNames=[
                    'All'
                ],
                WaitTimeSeconds=20
            )

            if 'Messages' in response:
                messages = [message for message in response['Messages'] if 'Body' in message]
                [process_message(message) for message in messages]

        try:
            while True:
                loop()
        except KeyboardInterrupt:
            sys.exit(0)

Here we persists the model in Client’s database

def insert_user(data, timestamp):
    username = data['username']
    serialized_user = UserSerializer(data=data)
    serialized_user.create(validated_data=data)
    logging.info(f"user: {username} created at {timestamp}")

def update_user(data, timestamp):
    username = data['username']
    try:
        user = User.objects.get(username=data['username'])
        serialized_user = UserSerializer(user)
        serialized_user.update(user, data)
        logging.info(f"user: {username} updated at {timestamp}")
    except User.DoesNotExist:
        logging.info(f"user: {username} don't exits. Creating ...")
        insert_user(data, timestamp)

def delete_user(data, timestamp):
    username = data['username']
    try:
        user = User.objects.get(username=username)
        user.delete()
        logging.info(f"user: {username} deleted at {timestamp}")
    except User.DoesNotExist:
        logging.info(f"user: {username} don't exits. Don't deleted")

And also emit one message to channel’s consumer

def notify_to_user(user):
    username = user['username']
    serialized_user = UserSerializer(user)
    emit_message_to_user(
        message=serialized_user.data,
        username=username, )

Here the Consumer:

class WsConsumer(AsyncWebsocketConsumer):
    @personal_consumer
    async def connect(self):
        await self.channel_layer.group_add(
            self._get_personal_room(),
            self.channel_name
        )

    @private_consumer_event
    async def emit_message(self, event):
        message = event['message']
        await self.send(text_data=json.dumps(message))

    def _get_personal_room(self):
        username = self.scope['user'].username
        return self.get_room_name(username)

    @staticmethod
    def get_room_name(room):
        return f"{'ws_room'}_{room}"

def emit_message_to_user(message, username):
    group = WsConsumer.get_room_name(username)
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(group, {
        'type': WsConsumer.emit_message.__name__,
        'message': message
    })

Our consumer will only allow to connect only if the user is authenticated. That’s because I like Django Channels. This kind of thing are really simple to to (I’ve done similar things using PHP applications connected to a socket.io server and it was a nightmare). I’ve created a couple of decorators to ensure authentication in the consumer.

def personal_consumer(func):
    @wraps(func)
    async def wrapper_decorator(*args, **kwargs):
        self = args[0]

        async def accept():
            value = await func(*args, **kwargs)
            await self.accept()
            return value

        if self.scope['user'].is_authenticated:
            username = self.scope['user'].username
            room_name = self.scope['url_route']['kwargs']['username']
            if username == room_name:
                return await accept()

        await self.close()

    return wrapper_decorator

def private_consumer_event(func):
    @wraps(func)
    async def wrapper_decorator(*args, **kwargs):
        self = args[0]
        if self.scope['user'].is_authenticated:
            return await func(*args, **kwargs)

    return wrapper_decorator

That’s the websocket route

from django.urls import re_path

from client import consumers

websocket_urlpatterns = [
    re_path(r'ws/(?P<username>\w+)$', consumers.WsConsumer),
]

Finally we only need to connect our HTML page to the websocket

{% block title %}Example{% endblock %}
{% block header_text %}Hello <span id="name">{{ request.user.first_name }}</span>{% endblock %}

{% block extra_body %}
  <script>
    var ws_scheme = window.location.protocol === "https:" ? "wss" : "ws"
    var ws_path = ws_scheme + '://' + window.location.host + "/ws/{{ request.user.username }}"
    var ws = new ReconnectingWebSocket(ws_path)
    var render = function (key, value) {
      document.querySelector(`#${key}`).innerHTML = value
    }
    ws.onmessage = function (e) {
      const data = JSON.parse(e.data);
      render('name', data.first_name)
    }

    ws.onopen = function () {
      console.log('Connected')
    };
  </script>
{% endblock %}

Here a docker-compose with the project:

version: '3.4'

services:
  redis:
    image: redis
  master:
    image: reactive_master:latest
    command: python manage.py runserver 0.0.0.0:8001
    build:
      context: ./master
      dockerfile: Dockerfile
    depends_on:
      - "redis"
    ports:
      - 8001:8001
    environment:
      REDIS_HOST: redis
  celery:
    image: reactive_master:latest
    command: celery -A master worker --uid=nobody --gid=nogroup
    depends_on:
      - "redis"
      - "master"
    environment:
      REDIS_HOST: redis
      SNS_REACTIVE_TABLE_ARN: ${SNS_REACTIVE_TABLE_ARN}
      AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION}
      AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
      AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
  client:
    image: reactive_client:latest
    command: python manage.py runserver 0.0.0.0:8000
    build:
      context: ./client
      dockerfile: Dockerfile
    depends_on:
      - "redis"
    ports:
      - 8000:8000
    environment:
      REDIS_HOST: redis
  listener:
    image: reactive_client:latest
    command: python manage.py listener
    build:
      context: ./client
      dockerfile: Dockerfile
    depends_on:
      - "redis"
    environment:
      REDIS_HOST: redis
      SQS_REACTIVE_TABLES: ${SQS_REACTIVE_TABLES}
      AWS_DEFAULT_REGION: ${AWS_DEFAULT_REGION}
      AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
      AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}

And that’s all. Here a working example of the prototype in action:

Source code in my github.

Alexa skill and account linking with serverless and Cognito

~ Gonzalo Ayuso ~ 1 Comment

Sometimes when we’re building one Alexa skill, we need to identify the user. To do that Alexa provides account linking. Basically we need an Oauth2 server to link our account within our Alexa skill. AWS provide us a managed Oauth2 service called Cognito, so we can use use Cognito identity pool to handle the authentication for our Alexa Skills.

In this example I’ve followed the following blog post. Cognito is is a bit weird to set up but after following all the steps we can use Account Linking in Alexa skill.

There’s also a good sample skill here. I’ve studied a little bit this example and create a working prototype by my own, basically to understand the process.

That’s my skill:

const Alexa = require('ask-sdk')

const RequestInterceptor = require('./interceptors/RequestInterceptor')
const ResponseInterceptor = require('./interceptors/ResponseInterceptor')
const LocalizationInterceptor = require('./interceptors/LocalizationInterceptor')
const GetLinkedInfoInterceptor = require('./interceptors/GetLinkedInfoInterceptor')

const LaunchRequestHandler = require('./handlers/LaunchRequestHandler')
const CheckAccountLinkedHandler = require('./handlers/CheckAccountLinkedHandler')
const HelloWorldIntentHandler = require('./handlers/HelloWorldIntentHandler')
const HelpIntentHandler = require('./handlers/HelpIntentHandler')
const CancelAndStopIntentHandler = require('./handlers/CancelAndStopIntentHandler')
const SessionEndedRequestHandler = require('./handlers/SessionEndedRequestHandler')
const FallbackHandler = require('./handlers/FallbackHandler')
const ErrorHandler = require('./handlers/ErrorHandler')
const RequestInfoHandler = require('./handlers/RequestInfoHandler')

let skill

module.exports.handler = async (event, context) => {
  if (!skill) {
    skill = Alexa.SkillBuilders.custom().
      addRequestInterceptors(
        RequestInterceptor,
        ResponseInterceptor,
        LocalizationInterceptor,
        GetLinkedInfoInterceptor
      ).
      addRequestHandlers(
        LaunchRequestHandler,
        CheckAccountLinkedHandler,
        HelloWorldIntentHandler,
        RequestInfoHandler,
        HelpIntentHandler,
        CancelAndStopIntentHandler,
        SessionEndedRequestHandler,
        FallbackHandler).
      addErrorHandlers(
        ErrorHandler).
      create()
  }

  return await skill.invoke(event, context)
}

The most important thing here is maybe GetLinkedInfoInterceptor.

const log = require('../lib/log')
const cognito = require('../lib/cognito')
const utils = require('../lib/utils')

const GetLinkedInfoInterceptor = {
  async process (handlerInput) {
    if (utils.isAccountLinked(handlerInput)) {
      const userData = await cognito.getUserData(handlerInput.requestEnvelope.session.user.accessToken)
      log.info('GetLinkedInfoInterceptor: getUserData: ', userData)
      const sessionAttributes = handlerInput.attributesManager.getSessionAttributes()
      if (userData.Username !== undefined) {
        sessionAttributes.auth = true
        sessionAttributes.emailAddress = cognito.getAttribute(userData.UserAttributes, 'email')
        sessionAttributes.userName = userData.Username
        handlerInput.attributesManager.setSessionAttributes(sessionAttributes)
      } else {
        sessionAttributes.auth = false
        log.error('GetLinkedInfoInterceptor: No user data was found.')
      }
    }
  }
}

module.exports = GetLinkedInfoInterceptor

This interceptor retrieves the user info from cognito when we provide the accessToken. We can obtain the accessToken from session (if our skill is account linked). Then we inject the user information (in my example the email and the username of the Cognito identity pool) into the session.

Then we can create one intent in our request handlers chain called CheckAccountLinkedHandler. With this intent we check if our skill is account linked. If not we can provide ‘withLinkAccountCard’ to force user to login with Cognito and link the skill’s account.

const utils = require('../lib/utils')

const CheckAccountLinkedHandler = {
  canHandle (handlerInput) {
    return !utils.isAccountLinked(handlerInput)
  },
  handle (handlerInput) {
    const requestAttributes = handlerInput.attributesManager.getRequestAttributes()
    const speakOutput = requestAttributes.t('NEED_TO_LINK_MESSAGE', 'SKILL_NAME')
    return handlerInput.responseBuilder.
      speak(speakOutput).
      withLinkAccountCard().
      getResponse()
  }
}

module.exports = CheckAccountLinkedHandler

Later we can create one intent to give the information to the user of maybe, in another case, perform an authorization workflow

const RequestInfoHandler = {
  canHandle (handlerInput) {
    const request = handlerInput.requestEnvelope.request
    return (request.type === 'IntentRequest'
      && request.intent.name === 'RequestInfoIntent')
  },
  handle (handlerInput) {
    const request = handlerInput.requestEnvelope.request
    const requestAttributes = handlerInput.attributesManager.getRequestAttributes()
    const sessionAttributes = handlerInput.attributesManager.getSessionAttributes()
    const repromptOutput = requestAttributes.t('FOLLOW_UP_MESSAGE')
    const cardTitle = requestAttributes.t('SKILL_NAME')

    let speakOutput = ''

    let inquiryTypeId = getResolvedSlotIDValue(request, 'infoTypeRequested')
    if (!inquiryTypeId) {
      inquiryTypeId = 'fullProfile'
      speakOutput += requestAttributes.t('NOT_SURE_OF_TYPE_MESSAGE')
    } else {
      if (inquiryTypeId === 'emailAddress' || inquiryTypeId === 'fullProfile') {
        speakOutput += requestAttributes.t('REPORT_EMAIL_ADDRESS', sessionAttributes.emailAddress)
      }

      if (inquiryTypeId === 'userName' || inquiryTypeId === 'fullProfile') {
        speakOutput += requestAttributes.t('REPORT_USERNAME', sessionAttributes.userName)
      }
    }

    speakOutput += repromptOutput

    return handlerInput.responseBuilder.
      speak(speakOutput).
      reprompt(repromptOutput).
      withSimpleCard(cardTitle, speakOutput).
      getResponse()
  }
}

module.exports = RequestInfoHandler

And basically that’s all. In fact isn’t very different than traditional web authentication. Maybe the most complicated part especially if you’re not used to Oauth2 is to configure Cognito properly.

Here you can see the source code in my github.

Playing with lambda, serverless and Python

~ Gonzalo Ayuso ~ Leave a comment

Couple of weeks ago I attended to serverless course. I’ve played with lambdas from time to time (basically when AWS forced me to use them) but without knowing exactly what I was doing. After this course I know how to work with the serverless framework and I understand better lambda world. Today I want to hack a little bit and create a simple Python service to obtain random numbers. Let’s start

We don’t need Flask to create lambdas but as I’m very comfortable with it so we’ll use it here.
Basically I follow the steps that I’ve read here.

from flask import Flask

app = Flask(__name__)


@app.route("/", methods=["GET"])
def hello():
    return "Hello from lambda"


if __name__ == '__main__':
    app.run()

And serverless yaml to configure the service

service: random

plugins:
  - serverless-wsgi
  - serverless-python-requirements
  - serverless-pseudo-parameters

custom:
  defaultRegion: eu-central-1
  defaultStage: dev
  wsgi:
    app: app.app
    packRequirements: false
  pythonRequirements:
    dockerizePip: non-linux

provider:
  name: aws
  runtime: python3.7
  region: ${opt:region, self:custom.defaultRegion}
  stage: ${opt:stage, self:custom.defaultStage}

functions:
  home:
    handler: wsgi_handler.handler
    events:
      - http: GET /

We’re going to use serverless plugins. We need to install them:

npx serverless plugin install -n serverless-wsgi
npx serverless plugin install -n serverless-python-requirements
npx serverless plugin install -n serverless-pseudo-parameters

And that’s all. Our “Hello world” lambda service with Python and Flask is up and running.
Now We’re going to create a “more complex” service. We’re going to return a random number with random.randint function.
randint requires two parameters: start, end. We’re going to pass the end parameter to our service. The start value will be parameterized. I’ll parameterize it only because I want to play with AWS’s Parameter Store (SSM). It’s just an excuse.

Let’s start with the service:

from random import randint
from flask import Flask, jsonify
import boto3
from ssm_parameter_store import SSMParameterStore

import os
from dotenv import load_dotenv

current_dir = os.path.dirname(os.path.abspath(__file__))
load_dotenv(dotenv_path="{}/.env".format(current_dir))

app = Flask(__name__)

app.config.update(
    STORE=SSMParameterStore(
        prefix="{}/{}".format(os.environ.get('ssm_prefix'), os.environ.get('stage')),
        ssm_client=boto3.client('ssm', region_name=os.environ.get('region')),
        ttl=int(os.environ.get('ssm_ttl'))
    )
)


@app.route("/", methods=["GET"])
def hello():
    return "Hello from lambda"


@app.route("/random/<int:to_int>", methods=["GET"])
def get_random_quote(to_int):
    from_int = app.config['STORE']['from_int']
    return jsonify(randint(from_int, to_int))


if __name__ == '__main__':
    app.run()

Now the serverless configuration. I can use only one function, handling all routes and letting Flask do the job.

functions:
  app:
    handler: wsgi_handler.handler
    events:
      - http: ANY /
      - http: 'ANY {proxy+}'

But in this example I want to create two different functions. Only for fun (and to use different role statements and different logs in cloudwatch).

service: random

plugins:
  - serverless-wsgi
  - serverless-python-requirements
  - serverless-pseudo-parameters
  - serverless-iam-roles-per-function

custom:
  defaultRegion: eu-central-1
  defaultStage: dev
  wsgi:
    app: app.app
    packRequirements: false
  pythonRequirements:
    dockerizePip: non-linux

provider:
  name: aws
  runtime: python3.7
  region: ${opt:region, self:custom.defaultRegion}
  stage: ${opt:stage, self:custom.defaultStage}
  memorySize: 128
  environment:
    region: ${self:provider.region}
    stage: ${self:provider.stage}

functions:
  app:
    handler: wsgi_handler.handler
    events:
      - http: ANY /
      - http: 'ANY {proxy+}'
    iamRoleStatements:
      - Effect: Allow
        Action: ssm:DescribeParameters
        Resource: arn:aws:ssm:${self:provider.region}:#{AWS::AccountId}:*
      - Effect: Allow
        Action: ssm:GetParameter
        Resource: arn:aws:ssm:${self:provider.region}:#{AWS::AccountId}:parameter/random/*
  home:
    handler: wsgi_handler.handler
    events:
      - http: GET /

And that’s all. “npx serverless deploy” and my random generator is running.

Handling Amazon SNS messages with PHP, Lumen and CloudWatch

~ Gonzalo Ayuso ~ 4 Comments

This days I’m involve with Amazon’s AWS and since I am migrating my backends to Lumen I’m going to play a little bit with AWS and Lumen. Today I want to create a simple Lumen server to handle SNS notifications. One end-point to listen to SNS and another one to emit notifications. I also want to register logs within CloudWatch. Let’s start.

First the Lumen server. 

use Laravel\Lumen\Application;

require __DIR__ . '/../vendor/autoload.php';

(new Dotenv\Dotenv(__DIR__ . "/../env"))->load();

$app = new Application();

$app->register(App\Providers\LogServiceProvider::class);
$app->register(App\Providers\AwsServiceProvider::class);

$app->group(['namespace' => 'App\Http\Controllers'], function (Application $app) {
    $app->get("/push", "SnsController@push");
    $app->post("/read", "SnsController@read");
});

$app->run();

As we can see there’s a route to push notifications and another one to read messages.

To work with SNS I will create a simple service provider

namespace App\Providers;

use Illuminate\Support\ServiceProvider;
use Aws\Sns\SnsClient;

class AwsServiceProvider extends ServiceProvider
{
    public function register()
    {
        $awsCredentials = [
            'region'      => getenv('AWS_REGION'),
            'version'     => getenv('AWS_VERSION'),
            'credentials' => [
                'key'    => getenv('AWS_CREDENTIALS_KEY'),
                'secret' => getenv('AWS_CREDENTIALS_SECRET'),
            ],
        ];

        $this->app->instance(SnsClient::class, new SnsClient($awsCredentials));
    }
}

Now We can create the routes in SnsController. Sns has a confirmation mechanism to validate endpoints. It’s well explained here

namespace App\Http\Controllers;

use Aws\Sns\SnsClient;
use Illuminate\Http\Request;
use Laravel\Lumen\Routing\Controller;
use Monolog\Logger;

class SnsController extends Controller
{
    private $request;
    private $logger;

    public function __construct(Request $request, Logger $logger)
    {
        $this->request = $request;
        $this->logger  = $logger;
    }

    public function push(SnsClient $snsClient)
    {
        $snsClient->publish([
            'TopicArn' => getenv('AWS_SNS_TOPIC1'),
            'Message'  => 'hi',
            'Subject'  => 'Subject',
        ]);

        return ['push'];
    }

    public function read(SnsClient $snsClient)
    {
        $data = $this->request->json()->all();

        if ($this->request->headers->get('X-Amz-Sns-Message-Type') == 'SubscriptionConfirmation') {
            $this->logger->notice("sns:confirmSubscription");
            $snsClient->confirmSubscription([
                'TopicArn' => getenv('AWS_SNS_TOPIC1'),
                'Token'    => $data['Token'],
            ]);
        } else {
            $this->logger->warn("read", [
                'Subject'   => $data['Subject'],
                'Message'   => $data['Message'],
                'Timestamp' => $data['Timestamp'],
            ]);
        }

        return "OK";
    }
}

Finally I want to use CloudWatch so I will configure Monolog with another service provider. It’s also well explained here:

namespace App\Providers;

use Aws\CloudWatchLogs\CloudWatchLogsClient;
use Illuminate\Support\ServiceProvider;
use Maxbanton\Cwh\Handler\CloudWatch;
use Monolog\Formatter\LineFormatter;
use Monolog\Logger;

class LogServiceProvider extends ServiceProvider
{
    public function register()
    {
        $awsCredentials = [
            'region'      => getenv('AWS_REGION'),
            'version'     => getenv('AWS_VERSION'),
            'credentials' => [
                'key'    => getenv('AWS_CREDENTIALS_KEY'),
                'secret' => getenv('AWS_CREDENTIALS_SECRET'),
            ],
        ];

        $cwClient = new CloudWatchLogsClient($awsCredentials);

        $cwRetentionDays      = getenv('CW_RETENTIONDAYS');
        $cwGroupName          = getenv('CW_GROUPNAME');
        $cwStreamNameInstance = getenv('CW_STREAMNAMEINSTANCE');
        $loggerName           = getenv('CF_LOGGERNAME');

        $logger  = new Logger($loggerName);
        $handler = new CloudWatch($cwClient, $cwGroupName, $cwStreamNameInstance, $cwRetentionDays);
        $handler->setFormatter(new LineFormatter(null, null, false, true));

        $logger->pushHandler($handler);

        $this->app->instance(Logger::class, $logger);
    }
}

Debugging those kind of webhooks with a EC2 instance sometimes is a bit hard. But we can easily expose our local webserver to internet with ngrok.
We only need to start our local server 

php -S 0.0.0.0:8080 -t www

And create a tunnel wiht ngrok

ngrok http 8080

And that’s up. Lumen and SNS up and running.

Code available in my github

Authenticate OpenUI5 applications and Lumen backends with Amazon Cognito and JWT

~ Gonzalo Ayuso ~ 1 Comment

Today I want to create an UI5/OpenUI5 boilerplate that plays with Lumen backends. Simple, isn’t it? We only need to create a Lumen API server and connect our OpenUI5 application with this API server. But today I also want to create a Login also. The typical user/password input form. I don’t want to build it from scratch (a user database, oauth provider or something like that). Since this days I’m involved with Amazon AWS projects I want to try Amazon Cognito.

Cognito has a great javaScript SDK. In fact we can do all the authentication flow (create users, validate passwords, change password, multifactor authentication, …) with Cognito. To create this project first I’ve create the following steps within Amazon AWS Cognito Console: Create a user pool with required attributes (email only in this example), without MFA and only allow administrators to create users. I’ve also created a App client inside this pool, so I’ve got a UserPoolId and a ClientId.

Let’s start with the OpenUI5 application. I’ve created an small application with one route called “home”. To handle the login process I will work in Component.js init function. The idea is check the cognito session. If there’s an active one (that’s means a Json Web Token stored in the local storage) we’ll display to “home” route and if there isn’t we’ll show login one.

sap.ui.define([
        "sap/ui/core/UIComponent",
        "sap/ui/Device",
        "app/model/models",
        "app/model/cognito"
    ], function (UIComponent, Device, models, cognito) {
        "use strict";

        return UIComponent.extend("app.Component", {

            metadata: {
                manifest: "json"
            },

            init: function () {
                UIComponent.prototype.init.apply(this, arguments);
                this.setModel(models.createDeviceModel(), "device");
                this.getRouter().initialize();

                var targets = this.getTargets();
                cognito.hasSession(function (err) {
                    if (err) {
                        targets.display("login");
                        return;
                    }
                    targets.display("home");
                });
            },

            /* *** */
        });
    }
);

To encapsulate the cognito operations I’ve create a model called cognito.js. It’s not perfect, but it allows me to abstract cognito stuff in the OpenUI5 application.

sap.ui.define([
        "app/conf/env"
    ], function (env) {
        "use strict";

        AWSCognito.config.region = env.region;

        var poolData = {
            UserPoolId: env.UserPoolId,
            ClientId: env.ClientId
        };

        var userPool = new AWSCognito.CognitoIdentityServiceProvider.CognitoUserPool(poolData);
        var jwt;

        var cognito = {
            getJwt: function () {
                return jwt;
            },

            hasSession: function (cbk) {
                var cognitoUser = cognito.getCurrentUser();
                if (cognitoUser != null) {
                    cognitoUser.getSession(function (err, session) {
                        if (err) {
                            cbk(err);
                            return;
                        }
                        if (session.isValid()) {
                            jwt = session.idToken.getJwtToken();
                            cbk(false, session)
                        } else {
                            cbk(true);
                        }
                    });
                } else {
                    cbk(true);
                }
            },

            getCurrentUser: function () {
                return userPool.getCurrentUser();
            },

            signOut: function () {
                var currentUser = cognito.getCurrentUser();
                if (currentUser) {
                    currentUser.signOut()
                }
            },

            getUsername: function () {
                var currentUser = cognito.getCurrentUser();
                return (currentUser) ? currentUser.username : undefined;
            },

            getUserData: function (user) {
                return {
                    Username: user,
                    Pool: userPool
                };
            },

            getCognitoUser: function (user) {
                return new AWSCognito.CognitoIdentityServiceProvider.CognitoUser(cognito.getUserData(user));
            },

            authenticateUser: function (user, pass, cbk) {
                var authenticationData = {
                    Username: user,
                    Password: pass
                };

                var authenticationDetails = new AWSCognito.CognitoIdentityServiceProvider.AuthenticationDetails(authenticationData);
                var cognitoUser = new AWSCognito.CognitoIdentityServiceProvider.CognitoUser(cognito.getUserData(user));

                cognitoUser.authenticateUser(authenticationDetails, cbk);

                return cognitoUser;
            }
        };

        return cognito;
    }
);

The login route has the following xml view:

<core:View
        xmlns:core="sap.ui.core"
        xmlns:f="sap.ui.layout.form"
        xmlns="sap.m"
        controllerName="app.controller.Login"
>
    <Image class="bg"></Image>
    <VBox class="sapUiSmallMargin loginForm">
        <f:SimpleForm visible="{= ${/flow} === 'login' }">
            <f:toolbar>
                <Toolbar>
                    <Title text="{i18n>Login_Title}" level="H4" titleStyle="H4"/>
                </Toolbar>
            </f:toolbar>
            <f:content>
                <Label text="{i18n>Login_user}"/>
                <Input placeholder="{i18n>Login_userPlaceholder}" value="{/user}"/>
                <Label text="{i18n>Login_pass}"/>
                <Input type="Password" placeholder="{i18n>Login_passPlaceholder}" value="{/pass}"/>
                <Button type="Accept" text="{i18n>OK}" press="loginPressHandle"/>
            </f:content>
        </f:SimpleForm>
        
        <f:SimpleForm visible="{= ${/flow} === 'PasswordReset' }">
            <f:toolbar>
                <Toolbar>
                    <Title text="{i18n>Login_PasswordReset}" level="H4" titleStyle="H4"/>
                </Toolbar>
            </f:toolbar>
            <f:content>
                <Label text="{i18n>Login_verificationCode}"/>
                <Input type="Number" placeholder="{i18n>Login_verificationCodePlaceholder}" value="{/verificationCode}"/>
                <Label text="{i18n>Login_newpass}"/>
                <Input type="Password" placeholder="{i18n>Login_newpassPlaceholder}" value="{/newPass}"/>
                <Button type="Accept" text="{i18n>OK}" press="newPassVerificationPressHandle"/>
            </f:content>
        </f:SimpleForm>
        
        <f:SimpleForm visible="{= ${/flow} === 'newPasswordRequired' }">
            <f:toolbar>
                <Toolbar>
                    <Title text="{i18n>Login_PasswordReset}" level="H4" titleStyle="H4"/>
                </Toolbar>
            </f:toolbar>
            <f:content>
                <Label text="{i18n>Login_newpass}"/>
                <Input type="Password" placeholder="{i18n>Login_newpassPlaceholder}" value="{/newPass}"/>
                <Button type="Accept" text="{i18n>OK}" press="newPassPressHandle"/>
            </f:content>
        </f:SimpleForm>
    </VBox>
</core:View>

It has three different stages: “login”, “PasswordReset” and “newPasswordRequired”
“login” is the main one. In this stage the user can input his login credentials. If credentials are OK then we’ll display home route.
The first time a user log in in the application with the password provided by the administrator, Cognito will force to change the password. Then We’ll show newPasswordRequired flow. I’m not going to explain each step. We developers prefer code than texts. That’s the code:

sap.ui.define([
        "app/controller/BaseController",
        "sap/ui/model/json/JSONModel",
        "sap/m/MessageToast",
        "app/model/cognito"
    ], function (BaseController, JSONModel, MessageToast, cognito) {
        "use strict";

        var cognitoUser;
        return BaseController.extend("app.controller.Login", {
            model: {
                user: "",
                pass: "",
                flow: "login",
                verificationCode: undefined,
                newPass: undefined
            },

            onInit: function () {
                this.getView().setModel(new JSONModel(this.model));
            },

            newPassPressHandle: function () {
                var that = this;
                var targets = this.getOwnerComponent().getTargets();
                var attributesData = {};
                sap.ui.core.BusyIndicator.show();
                cognitoUser.completeNewPasswordChallenge(this.model.newPass, attributesData, {
                    onFailure: function (err) {
                        sap.ui.core.BusyIndicator.hide();
                        MessageToast.show(err.message);
                    },
                    onSuccess: function (data) {
                        sap.ui.core.BusyIndicator.hide();
                        that.getModel().setProperty("/flow", "login");
                        targets.display("home");
                    }
                })
            },

            newPassVerificationPressHandle: function () {
                var that = this;
                var targets = this.getOwnerComponent().getTargets();
                sap.ui.core.BusyIndicator.show();
                cognito.getCognitoUser(this.model.user).confirmPassword(this.model.verificationCode, this.model.newPass, {
                    onFailure: function (err) {
                        sap.ui.core.BusyIndicator.hide();
                        MessageToast.show(err);
                    },
                    onSuccess: function (result) {
                        sap.ui.core.BusyIndicator.hide();
                        that.getModel().setProperty("/flow", "PasswordReset");
                        targets.display("home");
                    }
                });
            },

            loginPressHandle: function () {
                var that = this;
                var targets = this.getOwnerComponent().getTargets();
                sap.ui.core.BusyIndicator.show();
                cognitoUser = cognito.authenticateUser(this.model.user, this.model.pass, {
                    onSuccess: function (result) {
                        sap.ui.core.BusyIndicator.hide();
                        targets.display("home");
                    },

                    onFailure: function (err) {
                        sap.ui.core.BusyIndicator.hide();
                        switch (err.code) {
                            case "PasswordResetRequiredException":
                                that.getModel().setProperty("/flow", "PasswordReset");
                                break;
                            default:
                                MessageToast.show(err.message);
                        }
                    },

                    newPasswordRequired: function (userAttributes, requiredAttributes) {
                        sap.ui.core.BusyIndicator.hide();
                        that.getModel().setProperty("/flow", "newPasswordRequired");
                    }
                });
            }
        });
    }
);

The home route is the main one. It asumes that there’s an active Cognito session enabled.

<mvc:View
        controllerName="app.controller.Home"
        xmlns="sap.m"
        xmlns:mvc="sap.ui.core.mvc"
        xmlns:semantic="sap.m.semantic">
    <semantic:FullscreenPage
            id="page"
            semanticRuleSet="Optimized"
            showNavButton="false"
            title="{i18n>loggedUser}: {/userName}">
        <semantic:content>
            <Panel width="auto" class="sapUiResponsiveMargin" accessibleRole="Region">
                <headerToolbar>
                    <Toolbar height="3rem">
                        <Title text="Title"/>
                    </Toolbar>
                </headerToolbar>
                <content>
                    <Text text="Lorem ipsum dolor st amet, consetetur sadipscing elitr, sed diam nonumy eirmod tempor invidunt ut labore et dolore magna aliquyam erat, sed diam voluptua. At vero eos et accusam et justo duo dolores et ea rebum. Stet clita kasd gubergren, no sea takimata sanctus est Lorem ipsum dolor sit amet. Lorem ipsum dolor sit amet, consetetur sadipscing elitr, sed diam nonumy eirmod tempor invidunt ut labore et dolore magna aliquyam erat, sed diam voluptua. Lorem ipsum dolor sit amet, consetetur sadipscing elitr, sed diam nonumy eirmod tempor invidunt ut labore et dolore magna aliquyam erat"/>
                    <Button text="{i18n>Hello}" icon="sap-icon://hello-world" press="helloPress"/>
                </content>
            </Panel>
        </semantic:content>
        <semantic:customFooterContent>
            <Button text="{i18n>LogOff}" icon="sap-icon://visits" press="onLogOffPress"/>
        </semantic:customFooterContent>
    </semantic:FullscreenPage>
</mvc:View>

It shows the Cognito login name. It alos has a simple logff button and also one button that calls to the backend.

sap.ui.define([
        "app/controller/BaseController",
        "sap/ui/model/json/JSONModel",
        "sap/m/MessageToast",
        "app/model/cognito",
        "app/model/api"
    ], function (BaseController, JSONModel, MessageToast, cognito, api) {
        "use strict";

        return BaseController.extend("app.controller.Home", {
            model: {
                userName: ""
            },

            onInit: function () {
                this.model.userName = cognito.getUsername();
                this.getView().setModel(new JSONModel(this.model));
            },

            helloPress: function () {
                api.get("/api/hi", {}, function (data) {
                    MessageToast.show("Hello user " + data.userInfo.username + " (" + data.userInfo.email + ")");
                });
            },

            onLogOffPress: function () {
                cognito.signOut();
                this.getOwnerComponent().getTargets().display("login");
            }
        });
    }
);

To handle ajax requests I’ve create an api model. This model injects jwt inside every request.

sap.ui.define([
    "sap/m/MessageToast",
    "app/model/cognito"
], function (MessageToast, cognito) {
    "use strict";

    var backend = "";

    return {
        get: function (uri, params, cb) {
            params = params || {};
            params._jwt = cognito.getJwt();
            sap.ui.core.BusyIndicator.show(1000);

            jQuery.ajax({
                type: "GET",
                contentType: "application/json",
                data: params,
                url: backend + uri,
                cache: false,
                dataType: "json",
                async: true,
                success: function (data, textStatus, jqXHR) {
                    sap.ui.core.BusyIndicator.hide();
                    cb(data);
                },
                error: function (data, textStatus, jqXHR) {
                    sap.ui.core.BusyIndicator.hide();
                    switch (data.status) {
                        case 403: // Forbidden
                            MessageToast.show('Auth error');
                            break;
                        default:
                            console.log('Error', data);
                    }
                }
            });
        }
    };
});

That’s the frontend. Now it’s time to backend. Our Backend will be a simple Lumen server.

use App\Http\Middleware;
use Illuminate\Contracts\Debug\ExceptionHandler;
use Laravel\Lumen\Application;

(new Dotenv\Dotenv(__DIR__ . "/../env/"))->load();

$app = new Application();

$app->singleton(ExceptionHandler::class, App\Exceptions\Handler::class);

$app->routeMiddleware([
    'cognito' => Middleware\AuthCognitoMiddleware::class,
]);

$app->register(App\Providers\RedisServiceProvider::class);

$app->group([
    'middleware' => 'cognito',
    'namespace'  => 'App\Http\Controllers',
], function (Application $app) {
    $app->get("/api/hi", "DemoController@hi");
});

$app->run();

As you can see I’ve created a middelware to handle the authentication. This middleware will check the jwt provided by the frontend. We will use “spomky-labs/jose” library to validate the token. 

namespace App\Http\Middleware;

use Closure;
use Illuminate\Http\Request;
use Jose\Factory\JWKFactory;
use Jose\Loader;
use Monolog\Logger;
use Symfony\Component\Cache\Adapter\RedisAdapter;

class AuthCognitoMiddleware
{
    public function handle(Request $request, Closure $next)
    {
        try {
            $payload = $this->getPayload($request->get('_jwt'), $this->getJwtWebKeys());
            config([
                "userInfo" => [
                    'username' => $payload['cognito:username'],
                    'email'    => $payload['email'],
                ],
            ]);
        } catch (\Exception $e) {
            $log = app(Logger::class);
            $log->alert($e->getMessage());

            return response('Token Error', 403);
        }

        return $next($request);
    }

    private function getJwtWebKeys()
    {
        $url      = sprintf(
            'https://cognito-idp.%s.amazonaws.com/%s/.well-known/jwks.json',
            getenv('AWS_REGION'),
            getenv('AWS_COGNITO_POOL')
        );
        $cacheKey = sprintf('JWKFactory-Content-%s', hash('sha512', $url));

        $cache = app(RedisAdapter::class);

        $item = $cache->getItem($cacheKey);
        if (!$item->isHit()) {
            $item->set($this->getContent($url));
            $item->expiresAfter((int)getenv("TTL_JWK_CACHE"));
            $cache->save($item);
        }

        return JWKFactory::createFromJKU($url, false, $cache);
    }

    private function getPayload($accessToken, $jwtWebKeys)
    {
        $loader  = new Loader();
        $jwt     = $loader->loadAndVerifySignatureUsingKeySet($accessToken, $jwtWebKeys, ['RS256']);
        $payload = $jwt->getPayload();

        return $payload;
    }

    private function getContent($url)
    {
        $ch = curl_init();
        curl_setopt_array($ch, [
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_URL            => $url,
            CURLOPT_SSL_VERIFYPEER => true,
            CURLOPT_SSL_VERIFYHOST => 2,
        ]);
        $content = curl_exec($ch);
        curl_close($ch);

        return $content;
    }
}

To validate jwt Cognito tokens we need to obtain JwtWebKeys from this url

https://cognito-idp.my_aws_region.amazonaws.com/my_aws_cognito_pool_id/.well-known/jwks.json

That means that we need to fetch this url within every backend request, and that’s not cool. spomky-labs/jose allows us to use a cache to avoid fetch the request again and again. This cache is an instance of something that implementes the interface Psr\Cache\CacheItemPoolInterface. I’m not going to create a Cache from scratch. I’m not crazy. I’ll use symfony/cache here with a Redis adapter

And basically that’s all. Full application in my github