When you create an Alexa Skill, you can host the code in the AWS Lambda service and let Amazon handle the security requirements for you. Or you can host the skill anywhere you like and use the original Java SDK, which includes support for some of the security requirements. However, if you prefer to write in Python and want to host the skill yourself, you’ll need to take responsibility for all the security stuff.

Security Requirements

To certify and publish a skill, Amazon’s requires the server to perform several validation checks to keep everything secure. These checks are meant to ensure:

1. Our server is who it claims to be;
2. The requests are coming from Amazon;
3. The requests haven’t been altered or replayed;
4. Only our skills can utilize our service; and
5. All traffic is secured with TLS.

Server SSL/TLS

If we want to self-host our skill, we’ll need a server that can supply a certificate with a valid chain of trust to an approved Certificate Authority. In other words, we can’t make our own certificate, but most of the normal certificate products from popular domain name registrars are fine.

Make sure to use a https:// URL for the service endpoint in the configuration section of the skill.

Handling Requests with Python

The code below is from a web service that uses mod_wsgi with Apache. The overall structure of the wsgi code is described in Home Automation with Amazon Echo Apps, Part 2, and the source code for that older version of the echo_handler.wsgi file is available in the Maker Musings Github Repository.

import base64
import cryptography
import datetime
import dateutil
import requests
import subprocess
import tempfile
import urlparse

These are the libraries used by our validation code. Some of these are not included in the default Python installation and need to be installed separately.

def do_alexa(environ, start_response):
    # Alexa requests come as POST messages with a request body
    try:
        length = int(environ.get('CONTENT_LENGTH', '0'))
    except ValueError:
        length = 0

    if length <= 0:
        # This should never happen with a real Echo request but could happen
        # if your URL is accessed by a browser or otherwise.
        start_response('502 No content', [])
        return ['']

    # Get the request body and parse out the Alexa JSON request
    body = environ['wsgi.input'].read(length)
    if not VerifyClient(environ, body):
        start_response('403 Forbidden', [])
        return ['']

This is the code that gets called when a request comes in to the /alexa endpoint of our service. Lines 69 through 78 are defensive, since no valid request should have a content length of zero. Lines 81 and 82 kick off the first set of validation checks by grabbing the JSON body and calling VerifyClient().

def VerifyClient(environ, body):
    cert_url = environ.get('HTTP_SIGNATURECERTCHAINURL', None)
    signature = environ.get('HTTP_SIGNATURE', None)
    if cert_url is None or signature is None:
        return False

Amazon signs all of their requests, so we grab the signature and the certificate chain URL from the requests headers.

    try:
        cert_url_parts = urlparse.urlparse(cert_url)
        if cert_url_parts.scheme.lower() != 'https':
            return False
        netloc_elements = cert_url_parts.netloc.lower().split(':')
        if netloc_elements[0] != 's3.amazonaws.com':
            return False
        if len(netloc_elements) > 2 or (len(netloc_elements) == 2 and netloc_elements[1] != '443'):
            return False
        if cert_url_parts.path.find('/echo.api') != 0:
            return False

        r = requests.get(cert_url)
        if r.status_code != 200:
            return False

Before downloading the certificate chain, the URL must meet certain criteria. Our code here checks that

1. the scheme is https
2. the FQDN of the server is ‘s3.amazonaws.com’
3. the port number, if specified, is 443. If not specified, 443 will be used because of the https scheme
4. the path starts with ‘/echo.api’

If all four of these criteria are satisfied, then it is safe to download the certificate chain from the URL provided. If the download succeeds, we continue.

        pemfile = tempfile.NamedTemporaryFile(suffix='.pem', prefix='alexa_', delete=False)
        pemfile.write(r.text)
        pem_file_name = pemfile.name
        pemfile.close()

        chain_ok = subprocess.call(["/usr/bin/openssl", "verify", "-untrusted", pem_file_name, pem_file_name])
        os.remove(pem_file_name)
        if chain_ok != 0:
            return False

The downloaded certificate is written to a temporary file with a ‘.pem’ extension and then openssl is called to validate the certificate. Inside the PEM file, the signing certificate is first, followed by any number of intermediate certificates that form a chain of trust. Openssl verifies that there is an unbroken chain from the signing certificate to a trusted root certificate authority. The parameters shown will cause openssl to use the server’s default database of trusted root CAs.

        cert = cryptography.x509.load_pem_x509_certificate(str(r.text), cryptography.hazmat.backends.default_backend())
        cert_san_extension = cert.extensions.get_extension_for_oid(cryptography.x509.OID_SUBJECT_ALTERNATIVE_NAME)
        names = cert_san_extension.value.get_values_for_type(cryptography.x509.DNSName)
        if 'echo-api.amazon.com' not in names:
            return False

Now that the certificate is trusted, we extract the Subject Alternative Names section to verify that the certificate was issued for ‘echo-api.amazon.com’. This finishes validating that the signer of the message is, in fact, the Amazon Echo service. Next, we need to use the signature to ensure that the message is the one that was signed and hasn’t been altered.

        public_key = cert.public_key()

        encrypted_signature = base64.b64decode(signature)

        public_key.verify(encrypted_signature, body, cryptography.hazmat.primitives.asymmetric.padding.PKCS1v15(), cryptography.hazmat.primitives.hashes.SHA1())
    except Exception as e:
        return False

    return True

This code gets the public key from the certificate and the (base64-decoded) signature from the message header and uses them to validate the message body. If everything is good, the function returns True and processing continues.

    alexa_msg = json.loads(body)
    alexa_request = alexa_msg['request']
    if not VerifyTimestamp(alexa_request):
        start_response('400 Bad Request', [])
        return ['']

Back in the do_alexa() function, we can safely assume the message body is a JSON-encoded object that includes a ‘request’ element, which is extracted and passed to VerifyTimestamp().

def VerifyTimestamp(request):
    now = datetime.datetime.now(dateutil.tz.tzlocal())
    request_time = dateutil.parser.parse(request['timestamp'])
    if abs((now - request_time).seconds) > 120:
        return False
    return True

Amazon requires that the time sent in the request should match the server’s time plus or minus 150 seconds. We shorten that to 120 seconds to be on the safe side.

We need to take care here not to try to compare a timezone-aware time with a timezone-unaware time. The timestamp passed in Amazon’s request does contain timezone information, so the current server time accessed with datetime.datetime.now() must also include timezone information.

    alexa_session = alexa_msg['session']

    application = alexa_session.get('application', None)
    if application is None:
        start_response('400 Bad Request', [])
        return ['']
    application_id = application.get('applicationId', '')
    if application_id not in VALID_APPLICATION_IDS:
        start_response('400 Bad Request', [])
        return ['']

The final verification step checks that the incoming request is meant for our specific application. If this step isn’t done, and someone else discovers our service’s URL, they could create a skill that uses our server as the backend. Under most circumstances, the list VALID_APPLICATION_IDS contains the appIDs for both the live skill and the development version.

Now that all the required validations are done, our typical code flow from here would look something like this:

    if alexa_request['type'] == 'LaunchRequest':
        # This is the type when you just say "Open <app>"
        response = handle_launch(alexa_session['sessionId'])

    elif alexa_request['type'] == 'IntentRequest':
        # Get the intent being invoked and any slot values sent with it
        intent_name = alexa_request['intent']['name']
        intent_slots = alexa_request['intent'].get('slots', {})
        ...