[![Open in Cloud Shell][shell_img]][shell_link]

[shell_img]: http://gstatic.com/cloudssh/images/open-btn.png

[shell_link]: https://console.cloud.google.com/cloudshell/open?git_repo=https://github.com/GoogleCloudPlatform/python-docs-samples&page=editor&open_in_editor=appengine/standard/pubsub/README.md

This demonstrates how to send and receive messages using [Google Cloud Pub/Sub](https://cloud.google.com/pubsub) on [Google App Engine Standard Environment](https://cloud.google.com/appengine/docs/standard/).

Before you can run or deploy the sample, you will need to do the following:

1. Enable the Cloud Pub/Sub API in the [Google Developers Console](https://console.developers.google.com/project/_/apiui/apiview/pubsub/overview).

2. Create a topic and subscription. The push auth service account must have Service Account Token Creator Role assigned, which can be done in the Cloud Console [IAM & admin](https://console.cloud.google.com/iam-admin/iam) UI. `--push-auth-token-audience` is optional. If set, remember to modify the audience field check in `main.py` (line 88).

$ gcloud pubsub topics create [your-topic-name]

$ gcloud beta pubsub subscriptions create [your-subscription-name] \

--topic=[your-topic-name] \

--push-endpoint=\

https://[your-app-id].appspot.com/_ah/push-handlers/receive_messages/token=[your-token] \

--ack-deadline=30 \

--push-auth-service-account=[your-service-account-email] \

--push-auth-token-audience=example.com

3. Update the environment variables in ``app.yaml``.

When running locally, you can use the [Google Cloud SDK](https://cloud.google.com/sdk) to provide authentication to use Google Cloud APIs:

$ gcloud init

Install dependencies, preferably with a virtualenv:

$ virtualenv env

$ source env/bin/activate

$ pip install -r requirements.txt

Then set environment variables before starting your application:

$ export GOOGLE_CLOUD_PROJECT=[your-project-name]

$ export PUBSUB_VERIFICATION_TOKEN=[your-verification-token]

$ export PUBSUB_TOPIC=[your-topic]

$ python main.py

The application can send messages locally, but it is not able to receive push messages locally. You can, however, simulate a push message by making an HTTP request to the local push notification endpoint. There is an included ``sample_message.json``. You can use

``curl`` or [httpie](https://github.com/jkbrzt/httpie) to POST this:

$ curl -i --data @sample_message.json "localhost:8080/_ah/push-handlers/receive_messages?token=[your-token]"

Or

$ http POST ":8080/_ah/push-handlers/receive_messages?token=[your-token]" < sample_message.json

Response:

HTTP/1.0 400 BAD REQUEST

Content-Type: text/html; charset=utf-8

Content-Length: 58

Server: Werkzeug/0.15.2 Python/3.7.3

Date: Sat, 06 Apr 2019 04:56:12 GMT

Invalid token: 'NoneType' object has no attribute 'split'

The simulated push request fails because it does not have a Cloud Pub/Sub-generated JWT in the "Authorization" header.

Note: Not all the files in the current directory are needed to run your code on App Engine. Specifically, `main_test.py` and the `data` directory, which contains a mocked private key file and a mocked public certs file, are for testing purposes only. They SHOULD NOT be included in when deploying your app. When your app is up and running, Cloud Pub/Sub creates tokens using a private key, then the Google Auth Python library takes care of verifying and decoding the token using Google's public certs, to confirm that the push requests indeed come from Cloud Pub/Sub.

In the current directory, deploy using `gcloud`:

$ gcloud app deploy app.yaml

You can now access the application at `https://[your-app-id].appspot.com`. You can use the form to submit messages, but it's non-deterministic which instance of your application will receive the notification. You can send multiple messages and refresh the page to see the received message.

runtime: python37

env_variables:

PUBSUB_TOPIC: your-topic

# This token is used to verify that requests originate from your

# application. It can be any sufficiently random string.

PUBSUB_VERIFICATION_TOKEN: 1234abc

-----BEGIN RSA PRIVATE KEY-----

MIIEpAIBAAKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj

7wZgkdmM7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/

xmVU1WeruQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYs

SliS5qQpgyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18

pe+zpyl4+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xk

SBc//fy3ZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABAoIBAQDGGHzQxGKX+ANk

nQi53v/c6632dJKYXVJC+PDAz4+bzU800Y+n/bOYsWf/kCp94XcG4Lgsdd0Gx+Zq

HD9CI1IcqqBRR2AFscsmmX6YzPLTuEKBGMW8twaYy3utlFxElMwoUEsrSWRcCA1y

nHSDzTt871c7nxCXHxuZ6Nm/XCL7Bg8uidRTSC1sQrQyKgTPhtQdYrPQ4WZ1A4J9

IisyDYmZodSNZe5P+LTJ6M1SCgH8KH9ZGIxv3diMwzNNpk3kxJc9yCnja4mjiGE2

YCNusSycU5IhZwVeCTlhQGcNeV/skfg64xkiJE34c2y2ttFbdwBTPixStGaF09nU

Z422D40BAoGBAPvVyRRsC3BF+qZdaSMFwI1yiXY7vQw5+JZh01tD28NuYdRFzjcJ

vzT2n8LFpj5ZfZFvSMLMVEFVMgQvWnN0O6xdXvGov6qlRUSGaH9u+TCPNnIldjMP

B8+xTwFMqI7uQr54wBB+Poq7dVRP+0oHb0NYAwUBXoEuvYo3c/nDoRcZAoGBAOWl

aLHjMv4CJbArzT8sPfic/8waSiLV9Ixs3Re5YREUTtnLq7LoymqB57UXJB3BNz/2

eCueuW71avlWlRtE/wXASj5jx6y5mIrlV4nZbVuyYff0QlcG+fgb6pcJQuO9DxMI

aqFGrWP3zye+LK87a6iR76dS9vRU+bHZpSVvGMKJAoGAFGt3TIKeQtJJyqeUWNSk

klORNdcOMymYMIlqG+JatXQD1rR6ThgqOt8sgRyJqFCVT++YFMOAqXOBBLnaObZZ

CFbh1fJ66BlSjoXff0W+SuOx5HuJJAa5+WtFHrPajwxeuRcNa8jwxUsB7n41wADu

UqWWSRedVBg4Ijbw3nWwYDECgYB0pLew4z4bVuvdt+HgnJA9n0EuYowVdadpTEJg

soBjNHV4msLzdNqbjrAqgz6M/n8Ztg8D2PNHMNDNJPVHjJwcR7duSTA6w2p/4k28

bvvk/45Ta3XmzlxZcZSOct3O31Cw0i2XDVc018IY5be8qendDYM08icNo7vQYkRH

504kQQKBgQDjx60zpz8ozvm1XAj0wVhi7GwXe+5lTxiLi9Fxq721WDxPMiHDW2XL

YXfFVy/9/GIMvEiGYdmarK1NW+VhWl1DC5xhDg0kvMfxplt4tynoq1uTsQTY31Mx

BeF5CT/JuNYk3bEBF0H/Q3VGO1/ggVS+YezdFbLWIRoMnLj6XCFEGg==

-----END RSA PRIVATE KEY-----

-----BEGIN CERTIFICATE-----

MIIDIzCCAgugAwIBAgIJAMfISuBQ5m+5MA0GCSqGSIb3DQEBBQUAMBUxEzARBgNV

BAMTCnVuaXQtdGVzdHMwHhcNMTExMjA2MTYyNjAyWhcNMjExMjAzMTYyNjAyWjAV

MRMwEQYDVQQDEwp1bml0LXRlc3RzMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIB

CgKCAQEA4ej0p7bQ7L/r4rVGUz9RN4VQWoej1Bg1mYWIDYslvKrk1gpj7wZgkdmM

7oVK2OfgrSj/FCTkInKPqaCR0gD7K80q+mLBrN3PUkDrJQZpvRZIff3/xmVU1Wer

uQLFJjnFb2dqu0s/FY/2kWiJtBCakXvXEOb7zfbINuayL+MSsCGSdVYsSliS5qQp

gyDap+8b5fpXZVJkq92hrcNtbkg7hCYUJczt8n9hcCTJCfUpApvaFQ18pe+zpyl4

+WzkP66I28hniMQyUlA1hBiskT7qiouq0m8IOodhv2fagSZKjOTTU2xkSBc//fy3

ZpsL7WqgsZS7Q+0VRK8gKfqkxg5OYQIDAQABo3YwdDAdBgNVHQ4EFgQU2RQ8yO+O

gN8oVW2SW7RLrfYd9jEwRQYDVR0jBD4wPIAU2RQ8yO+OgN8oVW2SW7RLrfYd9jGh

GaQXMBUxEzARBgNVBAMTCnVuaXQtdGVzdHOCCQDHyErgUOZvuTAMBgNVHRMEBTAD

AQH/MA0GCSqGSIb3DQEBBQUAA4IBAQBRv+M/6+FiVu7KXNjFI5pSN17OcW5QUtPr

odJMlWrJBtynn/TA1oJlYu3yV5clc/71Vr/AxuX5xGP+IXL32YDF9lTUJXG/uUGk

+JETpKmQviPbRsvzYhz4pf6ZIOZMc3/GIcNq92ECbseGO+yAgyWUVKMmZM0HqXC9

ovNslqe0M8C1sLm1zAR5z/h/litE7/8O2ietija3Q/qtl2TOXJdCA6sgjJX2WUql

ybrC55ct18NKf3qhpcEkGQvFU40rVYApJpi98DiZPYFdx1oBDp/f4uZ3ojpxRVFT

cDwcJLfNRCPUhormsY7fDS9xSyThiHsW9mjJYdcaKQkwYZ0F11yB

-----END CERTIFICATE-----

import base64

from flask import current_app, Flask, render_template, request

import json

import logging

import os

from google.auth import jwt

from google.auth.transport import requests

from google.cloud import pubsub_v1

from google.oauth2 import id_token

app = Flask(__name__)

app.config['PUBSUB_VERIFICATION_TOKEN'] = \

os.environ['PUBSUB_VERIFICATION_TOKEN']

app.config['PUBSUB_TOPIC'] = os.environ['PUBSUB_TOPIC']

app.config['GCLOUD_PROJECT'] = os.environ['GOOGLE_CLOUD_PROJECT']

MESSAGES = []

TOKENS = []

HEADERS = []

CLAIMS = []

def index():

if request.method == 'GET':

return render_template('index.html', messages=MESSAGES, tokens=TOKENS,

headers=HEADERS, claims=CLAIMS)

data = request.form.get('payload', 'Example payload').encode('utf-8')

publisher = pubsub_v1.PublisherClient()

topic_path = publisher.topic_path(app.config['GCLOUD_PROJECT'],

app.config['PUBSUB_TOPIC'])

future = publisher.publish(topic_path, data)

future.result()

return 'OK', 200

def receive_messages_handler():

# Verify that the request originates from the application.

if (request.args.get('token', '') !=

current_app.config['PUBSUB_VERIFICATION_TOKEN']):

return 'Invalid request', 400

# Verify that the push request originates from Cloud Pub/Sub.

try:

# Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.

bearer_token = request.headers.get('Authorization')

token = bearer_token.split(' ')[1]

TOKENS.append(token)

header = jwt.decode_header(token)

HEADERS.append(header)

# Verify and decode the JWT. Underneath it checks the signature against

# Google's public certs at https://www.googleapis.com/oauth2/v1/certs.

# It also checks the token expiration time.

claim = id_token.verify_oauth2_token(token, requests.Request())

CLAIMS.append(claim)

# Check the audience field in the claim. It was specified in

# `--push-auth-token-audience` when you created the subscription.

assert claim['aud'] == 'example.com'

except Exception as e:

return 'Invalid token: {}\n'.format(e), 400

envelope = json.loads(request.data.decode('utf-8'))

payload = base64.b64decode(envelope['message']['data'])

MESSAGES.append(payload)

# Returning any 2xx status indicates successful receipt of the message.

return 'OK', 200

def server_error(e):

logging.exception('An error occurred during a request.')

return """

""".format(e), 500

if __name__ == '__main__':

# This is used when running locally. Gunicorn is used to run the

# application on Google App Engine. See entrypoint in app.yaml.

app.run(host='127.0.0.1', port=8080, debug=True)

import base64

import calendar

import datetime

import json

import os

import pytest

from google.auth import crypt

from google.auth import jwt

from google.oauth2 import id_token

import main

DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')

with open(os.path.join(DATA_DIR, 'privatekey.pem'), 'rb') as fh:

PRIVATE_KEY_BYTES = fh.read()

with open(os.path.join(DATA_DIR, 'public_cert.pem'), 'rb') as fh:

PUBLIC_CERT_BYTES = fh.read()

def client():

main.app.testing = True

return main.app.test_client()

def signer():

return crypt.RSASigner.from_string(PRIVATE_KEY_BYTES, '1')

def fake_token(signer):

now = calendar.timegm(datetime.datetime.utcnow().utctimetuple())

payload = {

'aud': 'example.com',

'azp': '1234567890',

'email': 'pubsub@example.iam.gserviceaccount.com',

'email_verified': True,

'iat': now,

'exp': now + 3600,

'iss': 'https://accounts.google.com',

'sub': '1234567890'

}

header = {

'alg': 'RS256',

'kid': signer.key_id,

'typ': 'JWT'

}

yield jwt.encode(signer, payload, header=header)

def _verify_mocked_oauth2_token(token, request):

claims = jwt.decode(token, certs=PUBLIC_CERT_BYTES, verify=True)

return claims

def test_index(client):

r = client.get('/')

assert r.status_code == 200

def test_post_index(client):

r = client.post('/', data={'payload': 'Test payload'})

assert r.status_code == 200

def test_push_endpoint(monkeypatch, client, fake_token):

monkeypatch.setattr(id_token, 'verify_oauth2_token',

_verify_mocked_oauth2_token)

url = '/_ah/push-handlers/receive_messages?token=' + \

os.environ['PUBSUB_VERIFICATION_TOKEN']

r = client.post(

url,

data=json.dumps({

"message": {

"data": base64.b64encode(

u'Test message'.encode('utf-8')

).decode('utf-8')

}

}),

headers=dict(

Authorization="Bearer " + fake_token.decode('utf-8')

)

)

assert r.status_code == 200

# Make sure the message is visible on the home page.

r = client.get('/')

assert r.status_code == 200

assert 'Test message' in r.data.decode('utf-8')

def test_push_endpoint_errors(client):

# no token

r = client.post('/_ah/push-handlers/receive_messages')

assert r.status_code == 400

# invalid token

r = client.post('/_ah/push-handlers/receive_messages?token=bad')

assert r.status_code == 400

{

"message": {

"data": "SGVsbG8sIFdvcmxkIQ=="

}

}