Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import urllib
- import httplib2
- from google.appengine.api import users
- import webapp2
- import json
- import logging
- import gdata.contacts.service
- #~ import gdata.calendar.service
- import os
- import urllib2
- import cookielib
- import logging, md5, urllib, urllib2
- from gdata import gauth
- #~ from google.appengine.tools import appengine_rpc
- email = "myemailaddress"
- password ="password"
- appname = "myapp"
- target_authenticated_google_app_engine_uri = 'http://myapp.appspot.com/secured'
- my_app_name = "myapp"
- def do_auth(appname, user, password, dev=False, admin=False):
- "This is taken from bits of appcfg, specifically: "
- " google/appengine/tools/appengine_rpc.py "
- "It returns the cookie send by the App Engine Login "
- "front-end after authenticating with Google Accounts. "
- if dev:
- return do_auth_dev_appserver(user, admin)
- # get the token
- try:
- auth_token = get_google_authtoken(appname, user, password)
- except AuthError, e:
- if e.reason == "BadAuthentication":
- logging.error( "Invalid username or password." )
- if e.reason == "CaptchaRequired":
- logging.error(
- "Please go to\n"
- "https://www.google.com/accounts/DisplayUnlockCaptcha\n"
- "and verify you are a human. Then try again.")
- if e.reason == "NotVerified":
- logging.error( "Account not verified.")
- if e.reason == "TermsNotAgreed":
- logging.error( "User has not agreed to TOS.")
- if e.reason == "AccountDeleted":
- logging.error( "The user account has been deleted.")
- if e.reason == "AccountDisabled":
- logging.error( "The user account has been disabled.")
- if e.reason == "ServiceDisabled":
- logging.error( "The user's access to the service has been "
- "disabled.")
- if e.reason == "ServiceUnavailable":
- logging.error( "The service is not available; try again later.")
- raise
- # now get the cookie
- cookie = get_gae_cookie(appname, auth_token)
- assert str(cookie)
- return cookie
- def do_auth_dev_appserver(email, admin):
- """Creates cookie payload data.
- Args:
- email, admin: Parameters to incorporate into the cookie.
- Returns:
- String containing the cookie payload.
- """
- admin_string = 'False'
- if admin:
- admin_string = 'True'
- if email:
- user_id_digest = md5.new(email.lower()).digest()
- user_id = '1' + ''.join(['%02d' % ord(x) for x in user_id_digest])[:20]
- else:
- user_id = ''
- return 'dev_appserver_login="%s:%s:%s"; Path=/;' % (email, admin_string, user_id)
- def get_gae_cookie(appname, auth_token):
- """
- Send a token to the App Engine login, again stating the name of the
- application to gain authentication for. Returned is a cookie that may be used
- to authenticate HTTP traffic to the application at App Engine.
- """
- continue_location = "http://myapp.appspot.com/secured"
- args = {"continue": continue_location, "auth": auth_token}
- host = "%s.appspot.com" % appname
- url = "http://%s/_ah/login?%s" % (host,
- urllib.urlencode(args))
- opener = get_opener() # no redirect handler!
- req = urllib2.Request(url)
- try:
- response = opener.open(req)
- except urllib2.HTTPError, e:
- response = e
- logging.info(response.headers)
- if (response.code != 302 or
- response.info()["location"] != continue_location):
- raise urllib2.HTTPError(req.get_full_url(), response.code,
- response.msg, response.headers, response.fp)
- cookie = response.headers.get('set-cookie')
- logging.info(response.headers)
- logging.info("cookie string = %s",cookie)
- assert cookie and cookie.startswith('ACSID')
- #~ return cookie.replace('; HttpOnly', '')
- return cookie, auth_token
- def get_google_authtoken(appname, email_address, password):
- """
- Make secure connection to Google Accounts and retrieve an authorisation
- token for the stated appname.
- The token can be send to the login front-end at appengine using
- get_gae_cookie(), which will return a cookie to use for the user session.
- """
- opener = get_opener()
- # get an AuthToken from Google accounts
- auth_uri = 'https://www.google.com/accounts/ClientLogin'
- authreq_data = urllib.urlencode({ "Email": email_address,
- "Passwd": password,
- "service": "ah",
- "source": appname,
- "accountType": "HOSTED_OR_GOOGLE" })
- req = urllib2.Request(auth_uri, data=authreq_data)
- try:
- response = opener.open(req)
- response_body = response.read()
- response_dict = dict(x.split("=")
- for x in response_body.split("\n") if x)
- return response_dict["Auth"]
- except urllib2.HTTPError, e:
- if e.code == 403:
- body = e.read()
- response_dict = dict(x.split("=", 1) for x in body.split("\n") if x)
- raise AuthError(req.get_full_url(), e.code, e.msg,
- e.headers, response_dict)
- else:
- raise
- """
- auth_req = urllib2.Request(auth_uri, data=authreq_data)
- auth_resp = urllib2.urlopen(auth_req)
- auth_resp_body = auth_resp.read()
- logging.info("authreq %s",auth_req)
- logging.info("auth_resp %s",auth_resp)
- logging.info("auth_resp_body %s",auth_resp_body)
- auth_resp_dict = dict(x.split("=")for x in auth_resp_body.split("\n") if x)
- authtoken = auth_resp_dict["Auth"]
- return authtoken
- """
- class AuthError(urllib2.HTTPError):
- """Raised to indicate there was an error authenticating."""
- def __init__(self, url, code, msg, headers, args):
- urllib2.HTTPError.__init__(self, url, code, msg, headers, None)
- self.args = args
- self.reason = args["Error"]
- def get_opener(cookiejar=None):
- opener = urllib2.OpenerDirector()
- opener.add_handler(urllib2.ProxyHandler())
- opener.add_handler(urllib2.UnknownHandler())
- opener.add_handler(urllib2.HTTPHandler())
- opener.add_handler(urllib2.HTTPDefaultErrorHandler())
- opener.add_handler(urllib2.HTTPErrorProcessor())
- opener.add_handler(urllib2.HTTPSHandler())
- if cookiejar:
- opener.add_handler(urllib2.HTTPCookieProcessor(cookiejar))
- return opener
- def passwdFunc():
- return (email, password)
- class TestSsoHandler(webapp2.RequestHandler):
- def get(self):
- cookieString = do_auth(appname, email, password)
- self.response.set_status(302, 'Use Cookie')
- self.response.headers['Cookie'] = cookieString[0]
- self.response.headers['Location'] = "/secured"
- if users.get_current_user():
- userEmail = users.get_current_user().email()
- self.response.out.write("current user userEmail = "+userEmail)
- else:
- self.response.out.write("current user not found")
- class OpenIDHandler(webapp2.RequestHandler):
- def get(self):
- if users.get_current_user():
- userEmail = users.get_current_user().email()
- logging.info("current user userEmail = %s", userEmail)
- else:
- logging.info("current user : not found")
- self.redirect(users.create_login_url(self.request.uri))
- app.yaml
- - url: /_ah/login_required
- script: urlmapper.py
- - url: /api\/.*\/?([a-z]*)
- script: urlmapper.py # TestSsoHandler
- - url: /secured
- static_files: index.html
- upload: index.html
- login: required
Advertisement
Add Comment
Please, Sign In to add comment