feat(pubsub): authenticated push requests

appengine/pubsub/README.md

@@ -11,13 +11,23 @@ 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).
 1. Create a topic and subscription.
 
-        gcloud beta pubsub topics create <your-topic-name>
-        gcloud beta pubsub subscriptions create <your-subscription-name> \
+        gcloud pubsub topics create <your-topic-name>
+        gcloud pubsub subscriptions create <your-subscription-name> \
           --topic <your-topic-name> \
           --push-endpoint \
             https://<your-project-id>.appspot.com/pubsub/push?token=<your-verification-token> \
           --ack-deadline 30
 
+1. Create a subscription for authenticated pushes. 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 `app.js` (line 112).
+
+        gcloud beta pubsub subscriptions create <your-subscription-name> \
+          --topic <your-topic-name> \
+          --push-endpoint \
+            https://<your-project-id>.appspot.com/pubsub/authenticated-push?token=<your-verification-token> \
+          --ack-deadline 30 \
+          --push-auth-service-account=[your-service-account-email] \
+          --push-auth-token-audience=example.com
+
 1. Update the environment variables in `app.standard.yaml` or `app.flexible.yaml`
 (depending on your App Engine environment).
 
@@ -61,3 +71,33 @@ Response:
 
 After the request completes, you can refresh `localhost:8080` and see the
 message in the list of received messages.
+
+### Authenticated push notifications
+
+Simulating authenticated push requests will fail because requests need to contain a Cloud Pub/Sub-generated JWT in the "Authorization" header.
+
+    http POST ":8080/pubsub/authenticated-push?token=<your-verification-token>" < sample_message.json
+
+Response:
+
+    HTTP/1.1 400 Bad Request
+    Connection: keep-alive
+    Date: Thu, 25 Apr 2019 17:47:36 GMT
+    Transfer-Encoding: chunked
+    X-Powered-By: Express
+
+    Invalid token
+
+## Running on App Engine
+
+Note: Not all the files in the current directory are needed to run your code on App Engine. Specifically, the `test` directory, which is for testing purposes only. It 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 Node.js 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.standard.yaml
+
+To deploy to App Engine Node.js Flexible Environment, run
+
+        gcloud app deploy app.flexible.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.

appengine/pubsub/app.js

@@ -17,6 +17,7 @@
 
 const express = require('express');
 const bodyParser = require('body-parser');
+const {OAuth2Client} = require('google-auth-library');
 const path = require('path');
 const Buffer = require('safe-buffer').Buffer;
 const process = require('process'); // Required for mocking environment variables
@@ -29,6 +30,7 @@ const process = require('process'); // Required for mocking environment variable
 const {PubSub} = require('@google-cloud/pubsub');
 
 // Instantiate a pubsub client
+const authClient = new OAuth2Client();
 const pubsub = new PubSub();
 
 const app = express();
@@ -40,6 +42,8 @@ const jsonBodyParser = bodyParser.json();
 
 // List of all messages received by this instance
 const messages = [];
+const claims = [];
+const tokens = [];
 
 // The following environment variables are set by app.yaml when running on GAE,
 // but will need to be manually set when running locally.
@@ -50,7 +54,7 @@ const topic = pubsub.topic(TOPIC);
 
 // [START gae_flex_pubsub_index]
 app.get('/', (req, res) => {
-  res.render('index', {messages: messages});
+  res.render('index', {messages, tokens, claims});
 });
 
 app.post('/', formBodyParser, async (req, res, next) => {
@@ -87,6 +91,45 @@ app.post('/pubsub/push', jsonBodyParser, (req, res) => {
 });
 // [END gae_flex_pubsub_push]
 
+// [START gae_flex_pubsub_auth_push]
+app.post('/pubsub/authenticated-push', jsonBodyParser, async (req, res) => {
+  // Verify that the request originates from the application.
+  if (req.query.token !== PUBSUB_VERIFICATION_TOKEN) {
+    res.status(400).send('Invalid request');
+    return;
+  }
+
+  // Verify that the push request originates from Cloud Pub/Sub.
+  try {
+    // Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
+    const bearer = req.header('Authorization');
+    const token = bearer.match(/Bearer (.*)/)[1];
+    tokens.push(token);
+
+    // Verify and decode the JWT.
+    const ticket = await authClient.verifyIdToken({
+      idToken: token,
+      audience: 'example.com',
+    });
+
+    const claim = ticket.getPayload();
+    claims.push(claim);
+  } catch (e) {
+    res.status(400).send('Invalid token');
+    return;
+  }
+
+  // The message is a unicode string encoded in base64.
+  const message = Buffer.from(req.body.message.data, 'base64').toString(
+    'utf-8'
+  );
+
+  messages.push(message);
+
+  res.status(200).send();
+});
+// [END gae_flex_pubsub_auth_push]
+
 // Start the server
 const PORT = process.env.PORT || 8080;
 app.listen(PORT, () => {

appengine/pubsub/package.json

@@ -16,12 +16,15 @@
     "@google-cloud/pubsub": "^0.28.0",
     "body-parser": "^1.18.3",
     "express": "^4.16.3",
+    "google-auth-library": "^3.1.2",
     "pug": "^2.0.1",
     "safe-buffer": "^5.1.2"
   },
   "devDependencies": {
     "@google-cloud/nodejs-repo-tools": "^3.0.0",
+    "jsonwebtoken": "^8.5.1",
     "mocha": "^6.0.0",
+    "sinon": "^7.3.1",
     "uuid": "^3.3.2"
   },
   "cloud-repo-tools": {

appengine/pubsub/test/app.test.js

@@ -19,7 +19,11 @@
 'use strict';
 
 const assert = require('assert');
+const fs = require('fs');
+const jwt = require('jsonwebtoken');
+const {OAuth2Client} = require('google-auth-library');
 const path = require('path');
+const sinon = require('sinon');
 const utils = require('@google-cloud/nodejs-repo-tools');
 
 const message = 'This is a test message sent at: ';
@@ -28,6 +32,38 @@ const payload = message + Date.now();
 const cwd = path.join(__dirname, '../');
 const requestObj = utils.getRequest({cwd: cwd});
 
+const fixtures = path.join(__dirname, 'fixtures');
+const privateKey = fs.readFileSync(path.join(fixtures, 'privatekey.pem'));
+const publicCert = fs.readFileSync(path.join(fixtures, 'public_cert.pem'));
+
+const sandbox = sinon.createSandbox();
+
+function createFakeToken() {
+  const now = Date.now() / 1000;
+
+  const 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',
+  };
+
+  const options = {
+    algorithm: 'RS256',
+    keyid: 'fake_id',
+  };
+
+  return jwt.sign(payload, privateKey, options);
+}
+
+afterEach(() => {
+  sandbox.restore();
+});
+
 it('should send a message to Pub/Sub', async () => {
   await requestObj
     .post('/')
@@ -51,6 +87,35 @@ it('should receive incoming Pub/Sub messages', async () => {
     .expect(200);
 });
 
+it('should verify incoming Pub/Sub push requests', async () => {
+  sandbox
+    .stub(OAuth2Client.prototype, 'getFederatedSignonCertsAsync')
+    .resolves({
+      certs: {
+        fake_id: publicCert,
+      },
+    });
+
+  await requestObj
+    .post('/pubsub/authenticated-push')
+    .set('Authorization', `Bearer ${createFakeToken()}`)
+    .query({token: process.env.PUBSUB_VERIFICATION_TOKEN})
+    .send({
+      message: {
+        data: Buffer.from(payload).toString('base64'),
+      },
+    })
+    .expect(200);
+
+  // Make sure the message is visible on the home page
+  await requestObj
+    .get('/')
+    .expect(200)
+    .expect(response => {
+      assert(response.text.includes(payload));
+    });
+});
+
 it('should check for verification token on incoming Pub/Sub messages', async () => {
   await requestObj
     .post('/pubsub/push')

appengine/pubsub/test/fixtures/privatekey.pem

@@ -0,0 +1,27 @@
+-----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-----

appengine/pubsub/test/fixtures/public_cert.pem

@@ -0,0 +1,19 @@
+-----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-----

appengine/pubsub/views/index.pug

@@ -4,6 +4,15 @@ html(lang='en')
     title PubSub
     meta(charset='utf-8')
   body
+    p Bearer tokens received by this instance:
+    ul
+      each val in tokens
+        li= val
+    p Claims received by this instance:
+    ul
+      each val in claims
+        li
+          code!= JSON.stringify(val)
     p Messages received by this instance:
     ul
       each val in messages