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:
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.