GCS SignedURL with Google AppEngine Standard (1st gen)

2019-12-22

Over the years, I’ve repeatedly had to setup samples that use Google AppEngine’s (v1) to generate a SignedURL.

The following is nothing new but my variation of it (since i just had to do this again from scratch)

Sample code demonstrating GCS SignedURL using original-flavor Appengine (v1).

Note, if your’e reading this and your’e using Cloud Run, Cloud Functions or AppEngine v2, its much, much eaiser: just use the google-cloud-storage python library

Anyway, if you’re really interested:

Setup

export PROJECT_ID=`gcloud config get-value core/project`
export DEST_BUCKET=$PROJECT_ID-gcs-test
gsutil mb gs://$DEST_BUCKET

gsutil iam ch serviceAccount:$PROJECT_ID@appspot.gserviceaccount.com:objectCreator,objectViewer gs://$DEST_BUCKET

$ echo bar > foo.txt
$ gsutil cp foo.txt gs://DEST_BUCKET/
virtualenv env
source env/bin/activate
pip install -r requirements.txt -t lib
deactivate
rm -rf env
$ gcloud app deploy app.yaml --version test --no-promote -q
target version:  [test]
target url:      [https://test-dot-yourproject.appspot.com]
$ curl -s https://test-dot-yourproject.appspot.com/
{"url": "yoursignedurlD"}
$ export SIGNED_URL=`curl -s https://test-dot-yourproject.appspot.com/ | jq -r '.url'`

$ curl -s $SIGNED_URL
bar

that last ‘bar’ is the content from the signedURL get request



from google.appengine.ext import webapp
import json 
from datetime import datetime
import time
from datetime import timedelta
import urllib
import base64

from google.appengine.api import app_identity



class MainPage(webapp.RequestHandler):
  def get(self):

    BUCKET_NAME="mineral-minutia-820"
    object_name="foo.txt"
    expiration = int(time.mktime((datetime.now() +
                                  timedelta(days=1)).timetuple()))

    signature_string = 'GET\n\n\n{}\n{}'.format(str(expiration),
                                                '/' + BUCKET_NAME + '/' +
                                                urllib.quote(object_name))
    signature_signed = self.sign(signature_string)

    urlt = 'https://storage.googleapis.com/{}/{}?{}'
    service_account = app_identity.get_service_account_name()
    url = urlt.format(urllib.quote(BUCKET_NAME),
                      urllib.quote(object_name),
                      urllib.urlencode({'GoogleAccessId': service_account,
                                        'Expires': str(expiration),
                                        'Signature': signature_signed}))
    self.response.write(json.dumps({'url': url}))

  def sign(self, plaintext):
      signature_bytes = app_identity.sign_blob(plaintext)[1]
      return base64.b64encode(signature_bytes)

class HealthCheck(webapp.RequestHandler):
    def get(self):
        self.response.set_status(200)
        self.response.out.write('ok')

app = webapp.WSGIApplication([('/', MainPage), ('/_ah/health',HealthCheck)], debug=True)
  • app.yaml
runtime: python27
api_version: 1
threadsafe: true

handlers:
- url: .*
  script: main.app
  secure: always

libraries:
- name: webapp2
  version: latest
  • appengine_config.py
import os
import google
from google.appengine.ext import vendor

lib_directory = os.path.dirname(__file__) + '/lib'


google.__path__ = [os.path.join(lib_directory, 'google')] + google.__path__

vendor.add(lib_directory)

This site supports webmentions. Send me a mention via this form.