Merge pull request mpdavis#30 from bjmc/at_hash

Adds support for at_hash verification

jose/constants.py

@@ -1,4 +1,4 @@
-
+import hashlib
 
 class ALGORITHMS(object):
  NONE = 'none'
@@ -19,3 +19,15 @@ class ALGORITHMS(object):
  SUPPORTED = HMAC + RSA + EC
 
  ALL = SUPPORTED + (NONE, )
+
+ HASHES = {
+ HS256: hashlib.sha256,
+ HS384: hashlib.sha384,
+ HS512: hashlib.sha512,
+ RS256: hashlib.sha256,
+ RS384: hashlib.sha384,
+ RS512: hashlib.sha512,
+ ES256: hashlib.sha256,
+ ES384: hashlib.sha384,
+ ES512: hashlib.sha512,
+ }

jose/jwt.py

@@ -14,10 +14,11 @@
 from .exceptions import JWTClaimsError
 from .exceptions import JWTError
 from .exceptions import ExpiredSignatureError
-from .utils import timedelta_total_seconds
+from .constants import ALGORITHMS
+from .utils import timedelta_total_seconds, calculate_at_hash
 
 
-def encode(claims, key, algorithm=None, headers=None):
+def encode(claims, key, algorithm=ALGORITHMS.HS256, headers=None, access_token=None):
  """Encodes a claims set and returns a JWT string.
 
  JWTs are JWS signed objects with a few reserved claims.
@@ -30,6 +31,9 @@ def encode(claims, key, algorithm=None, headers=None):
  headers (dict, optional): A set of headers that will be added to
  the default headers. Any headers that are added as additional
  headers will override the default headers.
+ access_token (str, optional): If present, the 'at_hash' claim will
+ be calculated and added to the claims present in the 'claims'
+ parameter.
 
  Returns:
  str: The string representation of the header, claims, and signature.
@@ -50,13 +54,15 @@ def encode(claims, key, algorithm=None, headers=None):
  if isinstance(claims.get(time_claim), datetime):
  claims[time_claim] = timegm(claims[time_claim].utctimetuple())
 
- if algorithm:
- return jws.sign(claims, key, headers=headers, algorithm=algorithm)
+ if access_token:
+ claims['at_hash'] = calculate_at_hash(access_token,
+ ALGORITHMS.HASHES[algorithm])
 
- return jws.sign(claims, key, headers=headers)
+ return jws.sign(claims, key, headers=headers, algorithm=algorithm)
 
 
-def decode(token, key, algorithms=None, options=None, audience=None, issuer=None, subject=None):
+def decode(token, key, algorithms=None, options=None, audience=None,
+ issuer=None, subject=None, access_token=None):
  """Verifies a JWT string's signature and validates reserved claims.
 
  Args:
@@ -72,6 +78,10 @@ def decode(token, key, algorithms=None, options=None, audience=None, issuer=None
  subject (str): The subject of the token. If the "sub" claim is
  included in the claim set, then the subject must be included and must equal
  the provided claim.
+ access_token (str): An access token returned alongside the id_token during
+ the authorization grant flow. If the "at_hash" claim is included in the
+ claim set, then the access_token must be included, and it must match
+ the "at_hash" claim.
  options (dict): A dictionary of options for skipping validation steps.
 
  defaults = {
@@ -109,6 +119,7 @@ def decode(token, key, algorithms=None, options=None, audience=None, issuer=None
  'verify_iss': True,
  'verify_sub': True,
  'verify_jti': True,
+ 'verify_at_hash': True,
  'leeway': 0,
  }
 
@@ -122,6 +133,9 @@ def decode(token, key, algorithms=None, options=None, audience=None, issuer=None
  except JWSError as e:
  raise JWTError(e)
 
+ # Needed for at_hash verification
+ algorithm = jws.get_unverified_header(token)['alg']
+
  try:
  claims = json.loads(payload.decode('utf-8'))
  except ValueError as e:
@@ -130,7 +144,10 @@ def decode(token, key, algorithms=None, options=None, audience=None, issuer=None
  if not isinstance(claims, Mapping):
  raise JWTError('Invalid payload string: must be a json object')
 
- _validate_claims(claims, audience=audience, issuer=issuer, subject=subject, options=defaults)
+ _validate_claims(claims, audience=audience, issuer=issuer,
+ subject=subject, algorithm=algorithm,
+ access_token=access_token,
+ options=defaults)
 
  return claims
 
@@ -384,7 +401,40 @@ def _validate_jti(claims):
  raise JWTClaimsError('JWT ID must be a string.')
 
 
-def _validate_claims(claims, audience=None, issuer=None, subject=None, options=None):
+def _validate_at_hash(claims, access_token, algorithm):
+ """
+ Validates that the 'at_hash' parameter included in the claims matches
+ with the access_token returned alongside the id token as part of
+ the authorization_code flow.
+
+ Args:
+ claims (dict): The claims dictionary to validate.
+ access_token (str): The access token returned by the OpenID Provider.
+ algorithm (str): The algorithm used to sign the JWT, as specified by
+ the token headers.
+ """
+ if 'at_hash' not in claims and not access_token:
+ return
+ elif 'at_hash' in claims and not access_token:
+ msg = 'No access_token provided to compare against at_hash claim.'
+ raise JWTClaimsError(msg)
+ elif access_token and 'at_hash' not in claims:
+ msg = 'at_hash claim missing from token.'
+ raise JWTClaimsError(msg)
+
+ try:
+ expected_hash = calculate_at_hash(access_token,
+ ALGORITHMS.HASHES[algorithm])
+ except (TypeError, ValueError):
+ msg = 'Unable to calculate at_hash to verify against token claims.'
+ raise JWTClaimsError(msg)
+
+ if claims['at_hash'] != expected_hash:
+ raise JWTClaimsError('at_hash claim does not match access_token.')
+
+
+def _validate_claims(claims, audience=None, issuer=None, subject=None,
+ algorithm=None, access_token=None, options=None):
 
  leeway = options.get('leeway', 0)
 
@@ -414,3 +464,6 @@ def _validate_claims(claims, audience=None, issuer=None, subject=None, options=N
 
  if options.get('verify_jti'):
  _validate_jti(claims)
+
+ if options.get('verify_at_hash'):
+ _validate_at_hash(claims, access_token, algorithm)

jose/utils.py

@@ -2,6 +2,29 @@
 import base64
 
 
+def calculate_at_hash(access_token, hash_alg):
+ """Helper method for calculating an access token
+ hash, as described in http:https://openid.net/specs/openid-connect-core-1_0.html#CodeIDToken
+
+ Its value is the base64url encoding of the left-most half of the hash of the octets
+ of the ASCII representation of the access_token value, where the hash algorithm
+ used is the hash algorithm used in the alg Header Parameter of the ID Token's JOSE
+ Header. For instance, if the alg is RS256, hash the access_token value with SHA-256,
+ then take the left-most 128 bits and base64url encode them. The at_hash value is a
+ case sensitive string.
+
+ Args:
+ access_token (str): An access token string.
+ hash_alg (callable): A callable returning a hash object, e.g. hashlib.sha256
+
+ """
+ hash_digest = hash_alg(access_token.encode('utf-8')).digest()
+ cut_at = int(len(hash_digest) / 2)
+ truncated = hash_digest[:cut_at]
+ at_hash = base64url_encode(truncated)
+ return at_hash.decode('utf-8')
+
+
 def base64url_decode(input):
  """Helper method to base64url_decode a string.
 

tests/test_jwt.py

@@ -428,6 +428,32 @@ def test_jti_invalid(self, key):
  with pytest.raises(JWTError):
  jwt.decode(token, key)
 
+ def test_at_hash(self, claims, key):
+ access_token = '<ACCESS_TOKEN>'
+ token = jwt.encode(claims, key, access_token=access_token)
+ payload = jwt.decode(token, key, access_token=access_token)
+ assert 'at_hash' in payload
+
+ def test_at_hash_invalid(self, claims, key):
+ token = jwt.encode(claims, key, access_token='<ACCESS_TOKEN>')
+ with pytest.raises(JWTError):
+ jwt.decode(token, key, access_token='<OTHER_TOKEN>')
+
+ def test_at_hash_missing_access_token(self, claims, key):
+ token = jwt.encode(claims, key, access_token='<ACCESS_TOKEN>')
+ with pytest.raises(JWTError):
+ jwt.decode(token, key)
+
+ def test_at_hash_missing_claim(self, claims, key):
+ token = jwt.encode(claims, key)
+ with pytest.raises(JWTError):
+ jwt.decode(token, key, access_token='<ACCESS_TOKEN>')
+
+ def test_at_hash_unable_to_calculate(self, claims, key):
+ token = jwt.encode(claims, key, access_token='<ACCESS_TOKEN>')
+ with pytest.raises(JWTError):
+ jwt.decode(token, key, access_token='\xe2')
+
  def test_unverified_claims_string(self):
  token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.aW52YWxpZCBjbGFpbQ.iOJ5SiNfaNO_pa2J4Umtb3b3zmk5C18-mhTCVNsjnck'
  with pytest.raises(JWTError):