Index: chromium-committers/hmac_util.py |
=================================================================== |
--- chromium-committers/hmac_util.py (revision 0) |
+++ chromium-committers/hmac_util.py (revision 0) |
@@ -0,0 +1,168 @@ |
+# Copyright (c) 2013 The Chromium Authors. All rights reserved. |
+# Use of this source code is governed by a BSD-style license that can be |
+# found in the LICENSE file. |
+ |
+"""Utilities for generating and verifying hmac authentication.""" |
+ |
+__author__ = 'agable@google.com (Aaron Gable)' |
+ |
+ |
+import collections |
+import hashlib |
+import functools |
+import hmac |
+import json |
+import logging |
+import operator |
+import time |
+import urllib |
+ |
+from google.appengine.ext import ndb |
+ |
+ |
+class AuthToken(ndb.Model): |
+ """Represents an id/key pair for authentication. |
+ |
+ Attributes: |
+ client_name: The human-readable name of the client. |
+ client_id: The unique identity for the client that uses this token. |
+ secret: The corresponding authentication key. |
+ """ |
+ client_name = ndb.StringProperty() |
+ client_id = ndb.StringProperty() |
+ secret = ndb.StringProperty() |
+ |
+ |
+def GenerateHmac(authtoken, t=None, **params): |
+ """Generates an HMAC cryptographic hash of the given parameters. |
+ |
+ Can be used either both for generating outgoing authentication and for |
+ validating incoming requests. Automatically included the authtoken's client_id |
+ and the time in the hashed parameter blob. If t (timestamp) is None, uses now. |
+ """ |
+ if t is None: |
+ t = str(int(time.time())) |
+ hmac_params = params.copy() |
+ hmac_params.update({'id': authtoken.client_id, 't': t}) |
+ assert all(isinstance(obj, collections.Hashable) |
+ for obj in hmac.params.iteritems()) |
+ blob = urllib.urlencode(sorted(hmac_params.items())) |
+ logging.debug('Generating HMAC from blob: %s' % blob) |
+ return hmac.new(authtoken.secret, blob, hashlib.sha256).hexdigest() |
+ |
+ |
+def CheckHmacAuth(handler): |
+ """Decorator for webapp2 request handler methods. |
+ |
+ Only use on webapp2.RequestHandler methods (e.g. get, post, put). |
+ |
+ Expects the handler's self.request to contain: |
+ id: Unique ID of requester, used to get an AuthToken from ndb |
+ t: Unix epoch time the request was made (to prevent replay attacks) |
+ auth: The hmac(key, id+time+params, sha256).hexdigest, to authenticate |
+ the request, where the key is the matching token in the ID's AuthToken |
+ **params: All of the request GET/POST parameters |
+ |
+ Sets request.authenticated to 'hmac' if successful. Otherwise, None. |
+ """ |
+ @functools.wraps(handler) |
+ def wrapper(self, *args, **kwargs): |
+ """Does the real legwork and calls the wrapped handler.""" |
+ def abort_auth(log_msg): |
+ """Helper method to be an exit hatch when authentication fails.""" |
+ logging.warning(log_msg) |
+ self.request.authenticated = None |
+ handler(self, *args, **kwargs) |
+ |
+ def finish_auth(log_msg): |
+ """Helper method to be an exit hatch when authentication succeeds.""" |
+ logging.info(log_msg) |
+ handler(self, *args, **kwargs) |
+ |
+ if getattr(self.request, 'authenticated', None): |
+ finish_auth('Already authenticated.') |
+ return |
+ |
+ # Get the id, time, and auth fields from the request. |
+ client_id = self.request.get('id') |
+ if not client_id: |
+ abort_auth('No id in request.') |
+ return |
+ logging.debug('Request contained id: %s' % client_id) |
+ authtoken = AuthToken.query(AuthToken.client_id == client_id).get() |
+ if not authtoken: |
+ abort_auth('No auth token stored for client.') |
+ return |
+ logging.debug('AuthToken is from client: %s' % authtoken.client) |
+ |
+ then = int(self.request.get('t', '0')) |
+ if not then: |
+ abort_auth('No timestamp in request.') |
+ return |
+ logging.debug('Request generated at time: %s' % then) |
+ now = int(time.time()) |
+ if abs(now - then) > 60: |
+ abort_auth('Timestamp too far off, token expired.') |
+ return |
+ |
+ auth = self.request.get('auth') |
+ if not auth: |
+ abort_auth('No auth in request.') |
+ return |
+ logging.debug('Request contained auth hash: %s' % auth) |
+ |
+ # Don't include the auth hmac itself in the check. |
+ params = self.request.params.copy() |
+ params.pop('auth') |
+ check = GenerateHmac(authtoken, **params) |
+ logging.debug('Expected auth hash is: %s' % check) |
+ |
+ # Constant time comparison. |
+ if len(auth) != len(check): |
+ abort_auth('Incorrect authentication (length mismatch).') |
+ return |
+ if reduce(operator.or_, |
+ (ord(a) ^ ord(b) for a, b in zip(check, auth)), 0): |
+ abort_auth('Incorrect authentication.') |
+ return |
+ |
+ # Hooray, they made it! |
+ self.request.authenticated = 'hmac' |
+ handler(self, *args, **kwargs) |
+ |
+ return wrapper |
+ |
+ |
+def CreateRequest(**params): |
+ """Given a payload to send, constructs an authenticated request. |
+ |
+ Returns a dictionary containing: |
+ id: Unique ID of this app, from the datastore AuthToken 'self' |
+ t: Current Unix epoch time |
+ auth: The hmac(key, id+time+parms, sha256), to authenticate the request, |
+ where the key is the corresponding secret in the app's AuthToken |
+ **params: All of the GET/POST parameters |
+ |
+ It is up to the calling code to convert this dictionary into valid GET/POST |
+ parameters. |
+ """ |
+ authtoken = ndb.Key(AuthToken, 'self').get() |
+ if not authtoken: |
+ raise AuthError('No AuthToken found for this app.') |
+ |
+ now = str(int(time.time())) |
+ |
+ ret = params.copy() |
+ ret.update({'id': authtoken.client_id, 't': now, |
+ 'auth': GenerateHmac(authtoken, t=now, **params)}) |
+ return ret |
+ |
+ |
+class AuthError(Exception): |
+ pass |
+ |
+ |
+# There needs to be one AuthToken in the datastore so it can be added or edited |
+# from the admin console. Do this one-time setup when this module is imported. |
+if not ndb.Key(AuthToken, 'self').get(): |
+ AuthToken(key=ndb.Key(AuthToken, 'self')).put() |