[Pkg-freeipa-devel] [Git][freeipa-team/python-jwcrypto][master] 21 commits: Post release bump to 0.7.dev1
Timo Aaltonen
gitlab at salsa.debian.org
Tue Apr 21 15:38:40 BST 2020
Timo Aaltonen pushed to branch master at FreeIPA packaging / python-jwcrypto
Commits:
90dcb8fb by Christian Heimes at 2018-11-05T16:20:16+01:00
Post release bump to 0.7.dev1
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
c49927a1 by Simo Sorce at 2018-12-12T11:11:31-05:00
Allow to use JWKSet on a JWT with no KID
When we are given a JWT and a KeySet, try all keys in the set if no
'kid' attribute is present in the JWT header instead of failing the
operation immediately.
Fixes #143
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
4b823285 by Brandon Lum at 2019-03-19T07:55:47-04:00
b64U -> b64u for pep8
Signed-off-by: Brandon Lum <lumjjb at gmail.com>
- - - - -
92151c2c by Anton Nguyen at 2019-03-20T12:53:07-04:00
Fixed typo
- - - - -
b94b24a5 by Brandon Lum at 2019-03-21T12:34:16-04:00
Fixed JWE jose_header
Signed-off-by: Brandon Lum <lumjjb at gmail.com>
- - - - -
061c654c by Brandon Lum at 2019-03-21T18:06:40-04:00
Added JWE/JWS custom registry header implementation
Signed-off-by: Brandon Lum <lumjjb at gmail.com>
- - - - -
961df898 by unknown at 2019-04-05T08:47:32-04:00
RFC 8037 - Support for Ed25519, Ed448
- - - - -
fc37b72b by Simo Sorce at 2019-06-06T15:21:09-04:00
Stricter OKP key generation parms check
- - - - -
b92643c3 by Simo Sorce at 2019-06-06T15:21:09-04:00
Add X25519/X448 support
Implements the missing part of rfc8037
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
559747af by Simo Sorce at 2019-06-06T15:21:09-04:00
Make X25519/X448 conditional when using older pyca
- - - - -
2cc6dadf by Simo Sorce at 2019-06-06T15:21:09-04:00
Simplify internal code curve selection
Adds a namedtuple table to make curve selection a simple table lookup
instead of long fragile if/elif/else constructs.
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
437ea86c by Simo Sorce at 2019-06-07T13:50:33-04:00
Fix encoding length of EC keys Coordinates
According to RFC 7518 6.2.1.2 the Coordinates need to be encoded such
that the size of the encoded string is fixed (based on private curve
key_size). Fix _encode_int to take in account the bit_size given
python-cryptography gives us python integers and that may lead to
truncation of leading zeros when doing the conversion via hex().
Fixes: #158
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
1f065c0c by Paul M at 2019-08-07T12:48:13-04:00
typo correction
- - - - -
e7f3845e by Simo Sorce at 2019-09-03T10:49:10-04:00
Remove unwanted debug print
This crept in unseen during the work to ad ed448
- - - - -
d7bb9344 by Simo Sorce at 2019-10-01T08:23:20-04:00
Try to fix travis and its old cruft
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
58c1099e by Simo Sorce at 2019-10-01T08:23:20-04:00
Add the ability to verify 'none' signatures.
Apparently this is required by OpenID Connect as an option:
https://openid.net/specs/openid-connect-core-1_0.html#RequestObject
In order to avoid accidental use of the None algoritm, which is disabled
by default but easy to casually enable, add the requirement to pass a
"None Key".
The none key clearly indicates that the user really wants to allow the
'none' algorithm to be used.
The "None Key" is an 'oct' key with key size of 0.
Example:
>>> from jwcrypto import jwk,jwt
>>> e="eyJhbGciOiJub25lIn0.eyJpc3MiOiJqb2UiLCJodHRwOi8vZXhhbXBsZS5jb20vaXNfcm9vdCI6dHJ1ZX0."
>>> k=jwk.JWK(generate='oct', size=0)
>>> E=jwt.JWT(algs=['none'])
>>> E.deserialize(jwt=e, key=k)
>>> E.claims
u'{"iss":"joe","http://example.com/is_root":true}'
Fixes #163
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
c51e10b8 by Karthikeyan Singaravelan at 2020-02-03T20:39:42+00:00
Import ABC from collections.abc instead of collections for Python 3.9 compatibility.
- - - - -
13c434ad by JC at 2020-02-17T18:57:27+00:00
docs: fix spelling
- - - - -
067ce869 by Simo Sorce at 2020-02-19T12:00:41-05:00
Release 0.7
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
51a7e2a8 by Timo Aaltonen at 2020-04-21T17:37:39+03:00
Merge branch 'upstream'
- - - - -
904c7902 by Timo Aaltonen at 2020-04-21T17:38:04+03:00
bump the version
- - - - -
12 changed files:
- debian/changelog
- docs/source/conf.py
- docs/source/jwe.rst
- jwcrypto/common.py
- jwcrypto/jwa.py
- jwcrypto/jwe.py
- jwcrypto/jwk.py
- jwcrypto/jws.py
- jwcrypto/jwt.py
- jwcrypto/tests.py
- setup.py
- tox.ini
Changes:
=====================================
debian/changelog
=====================================
@@ -1,5 +1,6 @@
-python-jwcrypto (0.6.0-3) UNRELEASED; urgency=medium
+python-jwcrypto (0.7.0-1) UNRELEASED; urgency=medium
+ * New upstream release.
* Set debhelper-compat version in Build-Depends.
* Set upstream metadata fields: Bug-Database, Repository, Repository-
Browse.
=====================================
docs/source/conf.py
=====================================
@@ -53,9 +53,9 @@ copyright = u'2016-2018, JWCrypto Contributors'
# built documents.
#
# The short X.Y version.
-version = '0.6'
+version = '0.7'
# The full version, including alpha/beta/rc tags.
-release = '0.6'
+release = '0.7'
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
=====================================
docs/source/jwe.rst
=====================================
@@ -3,7 +3,7 @@ JSON Web Encryption (JWE)
The jwe Module implements the `JSON Web Encryption`_ standard.
A JSON Web Encryption is represented by a JWE object, related utility
-classes and functions are availbale in this module too.
+classes and functions are available in this module too.
.. _JSON Web Encryption: https://tools.ietf.org/html/rfc7516
=====================================
jwcrypto/common.py
=====================================
@@ -1,8 +1,13 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
+import copy
import json
from base64 import urlsafe_b64decode, urlsafe_b64encode
-
+from collections import namedtuple
+try:
+ from collections.abc import MutableMapping
+except ImportError:
+ from collections import MutableMapping
# Padding stripping versions as described in
# RFC 7515 Appendix C
@@ -56,7 +61,7 @@ class InvalidCEKeyLength(JWException):
"""Invalid CEK Key Length.
This exception is raised when a Content Encryption Key does not match
- the required lenght.
+ the required length.
"""
def __init__(self, expected, obtained):
@@ -98,9 +103,91 @@ class InvalidJWEKeyLength(JWException):
"""Invalid JWE Key Length.
This exception is raised when the provided JWK Key does not match
- the lenght required by the sepcified algorithm.
+ the length required by the sepcified algorithm.
"""
def __init__(self, expected, obtained):
- msg = 'Expected key of lenght %d, got %d' % (expected, obtained)
+ msg = 'Expected key of length %d, got %d' % (expected, obtained)
super(InvalidJWEKeyLength, self).__init__(msg)
+
+
+class InvalidJWSERegOperation(JWException):
+ """Invalid JWSE Header Registry Operation.
+
+ This exception is raised when there is an error in trying ot add a JW
+ Signature or Encryption header to the Registry.
+ """
+
+ def __init__(self, message=None, exception=None):
+ msg = None
+ if message:
+ msg = message
+ else:
+ msg = 'Unknown Operation Failure'
+ if exception:
+ msg += ' {%s}' % repr(exception)
+ super(InvalidJWSERegOperation, self).__init__(msg)
+
+
+# JWSE Header Registry definitions
+
+# RFC 7515 - 9.1: JSON Web Signature and Encryption Header Parameters Registry
+# HeaderParameters are for both JWS and JWE
+JWSEHeaderParameter = namedtuple('Parameter',
+ 'description mustprotect supported check_fn')
+
+
+class JWSEHeaderRegistry(MutableMapping):
+ def __init__(self, init_registry=None):
+ if init_registry:
+ if isinstance(init_registry, dict):
+ self._registry = copy.deepcopy(init_registry)
+ else:
+ raise InvalidJWSERegOperation('Unknown input type')
+ else:
+ self._registry = {}
+
+ MutableMapping.__init__(self)
+
+ def check_header(self, h, value):
+ if h not in self._registry:
+ raise InvalidJWSERegOperation('No header "%s" found in registry'
+ % h)
+
+ param = self._registry[h]
+ if param.check_fn is None:
+ return True
+ else:
+ return param.check_fn(value)
+
+ def __getitem__(self, key):
+ return self._registry.__getitem__(key)
+
+ def __iter__(self):
+ return self._registry.__iter__()
+
+ def __delitem__(self, key):
+ if self._registry[key].mustprotect or \
+ self._registry[key].supported:
+ raise InvalidJWSERegOperation('Unable to delete protected or '
+ 'supported field')
+ else:
+ self._registry.__delitem__(key)
+
+ def __setitem__(self, h, jwse_header_param):
+ # Check if a header is not supported
+ if h in self._registry:
+ p = self._registry[h]
+ if p.supported:
+ raise InvalidJWSERegOperation('Supported header already exists'
+ ' in registry')
+ elif p.mustprotect and not jwse_header_param.mustprotect:
+ raise InvalidJWSERegOperation('Header specified should be'
+ 'a protected header')
+ else:
+ del self._registry[h]
+
+ self._registry[h] = jwse_header_param
+
+ def __len__(self):
+ return self._registry.__len__()
=====================================
jwcrypto/jwa.py
=====================================
@@ -70,7 +70,7 @@ def _inbytes(x):
def _randombits(x):
if x % 8 != 0:
- raise ValueError("lenght must be a multiple of 8")
+ raise ValueError("length must be a multiple of 8")
return os.urandom(_inbytes(x))
@@ -161,7 +161,8 @@ class _RawNone(_RawJWS):
return ''
def verify(self, key, payload, signature):
- raise InvalidSignature('The "none" signature cannot be verified')
+ if key.key_type != 'oct' or key.get_op_key() != '':
+ raise InvalidSignature('The "none" signature cannot be verified')
class _HS256(_RawHMAC, JWAAlgorithm):
@@ -693,8 +694,12 @@ class _EcdhEs(_RawKeyMgmt, JWAAlgorithm):
def _check_key(self, key):
if not isinstance(key, JWK):
raise ValueError('key is not a JWK object')
- if key.key_type != 'EC':
- raise InvalidJWEKeyType('EC', key.key_type)
+ if key.key_type not in ['EC', 'OKP']:
+ raise InvalidJWEKeyType('EC or OKP', key.key_type)
+ if key.key_type == 'OKP':
+ if key.key_curve not in ['X25519', 'X448']:
+ raise InvalidJWEKeyType('X25519 or X448',
+ key.key_curve)
def _derive(self, privkey, pubkey, alg, bitsize, headers):
# OtherInfo is defined in NIST SP 56A 5.8.1.2.1
@@ -718,7 +723,13 @@ class _EcdhEs(_RawKeyMgmt, JWAAlgorithm):
# no SuppPrivInfo
- shared_key = privkey.exchange(ec.ECDH(), pubkey)
+ # Shared Key generation
+ if isinstance(privkey, ec.EllipticCurvePrivateKey):
+ shared_key = privkey.exchange(ec.ECDH(), pubkey)
+ else:
+ # X25519/X448
+ shared_key = privkey.exchange(pubkey)
+
ckdf = ConcatKDFHash(algorithm=hashes.SHA256(),
length=_inbytes(bitsize),
otherinfo=otherinfo,
@@ -802,6 +813,28 @@ class _EcdhEsAes256Kw(_EcdhEs):
algorithm_use = 'kex'
+class _EdDsa(_RawJWS, JWAAlgorithm):
+
+ name = 'EdDSA'
+ description = 'EdDSA using Ed25519 or Ed448 algorithms'
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+ keysize = None
+
+ def sign(self, key, payload):
+
+ if key.key_curve in ['Ed25519', 'Ed448']:
+ skey = key.get_op_key('sign')
+ return skey.sign(payload)
+ raise NotImplementedError
+
+ def verify(self, key, payload, signature):
+ if key.key_curve in ['Ed25519', 'Ed448']:
+ pkey = key.get_op_key('verify')
+ return pkey.verify(signature, payload)
+ raise NotImplementedError
+
+
class _RawJWE(object):
def encrypt(self, k, a, m):
@@ -1026,6 +1059,7 @@ class JWA(object):
'ECDH-ES+A128KW': _EcdhEsAes128Kw,
'ECDH-ES+A192KW': _EcdhEsAes192Kw,
'ECDH-ES+A256KW': _EcdhEsAes256Kw,
+ 'EdDSA': _EdDsa,
'A128GCMKW': _A128GcmKw,
'A192GCMKW': _A192GcmKw,
'A256GCMKW': _A256GcmKw,
=====================================
jwcrypto/jwe.py
=====================================
@@ -4,6 +4,7 @@ import zlib
from jwcrypto import common
from jwcrypto.common import JWException
+from jwcrypto.common import JWSEHeaderParameter, JWSEHeaderRegistry
from jwcrypto.common import base64url_decode, base64url_encode
from jwcrypto.common import json_decode, json_encode
from jwcrypto.jwa import JWA
@@ -11,20 +12,23 @@ from jwcrypto.jwa import JWA
# RFC 7516 - 4.1
# name: (description, supported?)
-JWEHeaderRegistry = {'alg': ('Algorithm', True),
- 'enc': ('Encryption Algorithm', True),
- 'zip': ('Compression Algorithm', True),
- 'jku': ('JWK Set URL', False),
- 'jwk': ('JSON Web Key', False),
- 'kid': ('Key ID', True),
- 'x5u': ('X.509 URL', False),
- 'x5c': ('X.509 Certificate Chain', False),
- 'x5t': ('X.509 Certificate SHA-1 Thumbprint', False),
- 'x5t#S256': ('X.509 Certificate SHA-256 Thumbprint',
- False),
- 'typ': ('Type', True),
- 'cty': ('Content Type', True),
- 'crit': ('Critical', True)}
+JWEHeaderRegistry = {
+ 'alg': JWSEHeaderParameter('Algorithm', False, True, None),
+ 'enc': JWSEHeaderParameter('Encryption Algorithm', False, True, None),
+ 'zip': JWSEHeaderParameter('Compression Algorithm', False, True, None),
+ 'jku': JWSEHeaderParameter('JWK Set URL', False, False, None),
+ 'jwk': JWSEHeaderParameter('JSON Web Key', False, False, None),
+ 'kid': JWSEHeaderParameter('Key ID', False, True, None),
+ 'x5u': JWSEHeaderParameter('X.509 URL', False, False, None),
+ 'x5c': JWSEHeaderParameter('X.509 Certificate Chain', False, False, None),
+ 'x5t': JWSEHeaderParameter('X.509 Certificate SHA-1 Thumbprint', False,
+ False, None),
+ 'x5t#S256': JWSEHeaderParameter('X.509 Certificate SHA-256 Thumbprint',
+ False, False, None),
+ 'typ': JWSEHeaderParameter('Type', False, True, None),
+ 'cty': JWSEHeaderParameter('Content Type', False, True, None),
+ 'crit': JWSEHeaderParameter('Critical', True, True, None),
+}
"""Registry of valid header parameters"""
default_allowed_algs = [
@@ -73,7 +77,8 @@ class JWE(object):
"""
def __init__(self, plaintext=None, protected=None, unprotected=None,
- aad=None, algs=None, recipient=None, header=None):
+ aad=None, algs=None, recipient=None, header=None,
+ header_registry=None):
"""Creates a JWE token.
:param plaintext(bytes): An arbitrary plaintext to be encrypted.
@@ -83,10 +88,14 @@ class JWE(object):
:param algs: An optional list of allowed algorithms
:param recipient: An optional, default recipient key
:param header: An optional header for the default recipient
+ :param header_registry: Optional additions to the header registry
"""
self._allowed_algs = None
self.objects = dict()
self.plaintext = None
+ self.header_registry = JWSEHeaderRegistry(JWEHeaderRegistry)
+ if header_registry:
+ self.header_registry.update(header_registry)
if plaintext is not None:
if isinstance(plaintext, bytes):
self.plaintext = plaintext
@@ -339,10 +348,10 @@ class JWE(object):
def _check_crit(self, crit):
for k in crit:
- if k not in JWEHeaderRegistry:
+ if k not in self.header_registry:
raise InvalidJWEData('Unknown critical header: "%s"' % k)
else:
- if not JWEHeaderRegistry[k][1]:
+ if not self.header_registry[k].supported:
raise InvalidJWEData('Unsupported critical header: '
'"%s"' % k)
@@ -354,6 +363,11 @@ class JWE(object):
# TODO: allow caller to specify list of headers it understands
self._check_crit(jh.get('crit', dict()))
+ for hdr in jh:
+ if hdr in self.header_registry:
+ if not self.header_registry.check_header(hdr, self):
+ raise InvalidJWEData('Failed header check')
+
alg = self._jwa_keymgmt(jh.get('alg', None))
enc = self._jwa_enc(jh.get('enc', None))
@@ -492,7 +506,7 @@ class JWE(object):
@property
def jose_header(self):
- jh = self._get_jose_header()
+ jh = self._get_jose_header(self.objects.get('header'))
if len(jh) == 0:
raise InvalidJWEOperation("JOSE Header not available")
return jh
=====================================
jwcrypto/jwk.py
=====================================
@@ -18,10 +18,76 @@ from jwcrypto.common import base64url_decode, base64url_encode
from jwcrypto.common import json_decode, json_encode
-# RFC 7518 - 7.4
+class UnimplementedOKPCurveKey(object):
+ @classmethod
+ def generate(cls):
+ raise NotImplementedError
+
+ @classmethod
+ def from_public_bytes(cls, *args):
+ raise NotImplementedError
+
+ @classmethod
+ def from_private_bytes(cls, *args):
+ raise NotImplementedError
+
+
+ImplementedOkpCurves = []
+
+
+# Handle the best we can older versions of python cryptography that
+# do not yet implement these interfaces properly
+try:
+ from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+ Ed25519PublicKey, Ed25519PrivateKey
+ )
+ ImplementedOkpCurves.append('Ed25519')
+except ImportError:
+ Ed25519PublicKey = UnimplementedOKPCurveKey
+ Ed25519PrivateKey = UnimplementedOKPCurveKey
+try:
+ from cryptography.hazmat.primitives.asymmetric.ed448 import (
+ Ed448PublicKey, Ed448PrivateKey
+ )
+ ImplementedOkpCurves.append('Ed448')
+except ImportError:
+ Ed448PublicKey = UnimplementedOKPCurveKey
+ Ed448PrivateKey = UnimplementedOKPCurveKey
+try:
+ from cryptography.hazmat.primitives.asymmetric.x25519 import (
+ X25519PublicKey, X25519PrivateKey
+ )
+ priv_bytes = getattr(X25519PrivateKey, 'from_private_bytes', None)
+ if priv_bytes is None:
+ raise ImportError
+ ImplementedOkpCurves.append('X25519')
+except ImportError:
+ X25519PublicKey = UnimplementedOKPCurveKey
+ X25519PrivateKey = UnimplementedOKPCurveKey
+try:
+ from cryptography.hazmat.primitives.asymmetric.x448 import (
+ X448PublicKey, X448PrivateKey
+ )
+ ImplementedOkpCurves.append('X448')
+except ImportError:
+ X448PublicKey = UnimplementedOKPCurveKey
+ X448PrivateKey = UnimplementedOKPCurveKey
+
+
+_OKP_CURVE = namedtuple('Name', 'pubkey privkey')
+_OKP_CURVES_TABLE = {
+ 'Ed25519': _OKP_CURVE(Ed25519PublicKey, Ed25519PrivateKey),
+ 'Ed448': _OKP_CURVE(Ed448PublicKey, Ed448PrivateKey),
+ 'X25519': _OKP_CURVE(X25519PublicKey, X25519PrivateKey),
+ 'X448': _OKP_CURVE(X448PublicKey, X448PrivateKey)
+}
+
+
+# RFC 7518 - 7.4 , RFC 8037 - 5
JWKTypesRegistry = {'EC': 'Elliptic Curve',
'RSA': 'RSA',
- 'oct': 'Octet sequence'}
+ 'oct': 'Octet sequence',
+ 'OKP': 'Octet Key Pair'}
"""Registry of valid Key Types"""
@@ -31,7 +97,7 @@ JWKTypesRegistry = {'EC': 'Elliptic Curve',
class ParmType(Enum):
name = 'A string with a name'
b64 = 'Base64url Encoded'
- b64U = 'Base64urlUint Encoded'
+ b64u = 'Base64urlUint Encoded'
unsupported = 'Unsupported Parameter'
@@ -45,21 +111,26 @@ JWKValuesRegistry = {
},
'RSA': {
'n': JWKParameter('Modulus', True, True, ParmType.b64),
- 'e': JWKParameter('Exponent', True, True, ParmType.b64U),
- 'd': JWKParameter('Private Exponent', False, False, ParmType.b64U),
- 'p': JWKParameter('First Prime Factor', False, False, ParmType.b64U),
- 'q': JWKParameter('Second Prime Factor', False, False, ParmType.b64U),
+ 'e': JWKParameter('Exponent', True, True, ParmType.b64u),
+ 'd': JWKParameter('Private Exponent', False, False, ParmType.b64u),
+ 'p': JWKParameter('First Prime Factor', False, False, ParmType.b64u),
+ 'q': JWKParameter('Second Prime Factor', False, False, ParmType.b64u),
'dp': JWKParameter('First Factor CRT Exponent',
- False, False, ParmType.b64U),
+ False, False, ParmType.b64u),
'dq': JWKParameter('Second Factor CRT Exponent',
- False, False, ParmType.b64U),
+ False, False, ParmType.b64u),
'qi': JWKParameter('First CRT Coefficient',
- False, False, ParmType.b64U),
+ False, False, ParmType.b64u),
'oth': JWKParameter('Other Primes Info',
False, False, ParmType.unsupported),
},
'oct': {
'k': JWKParameter('Key Value', False, True, ParmType.b64),
+ },
+ 'OKP': {
+ 'crv': JWKParameter('Curve', True, True, ParmType.name),
+ 'x': JWKParameter('Public Key', True, True, ParmType.b64),
+ 'd': JWKParameter('Private Key', False, False, ParmType.b64),
}
}
"""Registry of valid key values"""
@@ -79,10 +150,14 @@ JWKParamsRegistry = {
}
"""Regstry of valid key parameters"""
-# RFC 7518 - 7.6
+# RFC 7518 - 7.6 , RFC 8037 - 5
JWKEllipticCurveRegistry = {'P-256': 'P-256 curve',
'P-384': 'P-384 curve',
- 'P-521': 'P-521 curve'}
+ 'P-521': 'P-521 curve',
+ 'Ed25519': 'Ed25519 signature algorithm key pairs',
+ 'Ed448': 'Ed448 signature algorithm key pairs',
+ 'X25519': 'X25519 function key pairs',
+ 'X448': 'X448 function key pairs'}
"""Registry of allowed Elliptic Curves"""
# RFC 7517 - 8.2
@@ -212,7 +287,8 @@ class JWK(object):
Valid options per type, when generating new keys:
* oct: size(int)
* RSA: public_exponent(int), size(int)
- * EC: curve(str) (one of P-256, P-384, P-521)
+ * EC: crv(str) (one of P-256, P-384, P-521)
+ * OKP: crv(str) (one of Ed25519, Ed448, X25519, X448)
Deprecated:
Alternatively if the 'generate' parameter is provided, with a
@@ -274,9 +350,17 @@ class JWK(object):
params['k'] = base64url_encode(key)
self.import_key(**params)
- def _encode_int(self, i):
- intg = hex(i).rstrip("L").lstrip("0x")
- return base64url_encode(unhexlify((len(intg) % 2) * '0' + intg))
+ def _encode_int(self, i, bit_size=None):
+ extend = 0
+ if bit_size is not None:
+ extend = ((bit_size + 7) // 8) * 2
+ hexi = hex(i).rstrip("L").lstrip("0x")
+ hexl = len(hexi)
+ if extend > hexl:
+ extend -= hexl
+ else:
+ extend = hexl % 2
+ return base64url_encode(unhexlify(extend * '0' + hexi))
def _generate_RSA(self, params):
pubexp = 65537
@@ -317,6 +401,8 @@ class JWK(object):
return ec.SECP384R1()
elif name == 'P-521':
return ec.SECP521R1()
+ elif name in _OKP_CURVES_TABLE:
+ return name
else:
raise InvalidJWKValue('Unknown Elliptic Curve Type')
@@ -334,12 +420,13 @@ class JWK(object):
def _import_pyca_pri_ec(self, key, **params):
pn = key.private_numbers()
+ key_size = pn.public_numbers.curve.key_size
params.update(
kty='EC',
crv=JWKpycaCurveMap[key.curve.name],
- x=self._encode_int(pn.public_numbers.x),
- y=self._encode_int(pn.public_numbers.y),
- d=self._encode_int(pn.private_value)
+ x=self._encode_int(pn.public_numbers.x, key_size),
+ y=self._encode_int(pn.public_numbers.y, key_size),
+ d=self._encode_int(pn.private_value, key_size)
)
self.import_key(**params)
@@ -353,6 +440,40 @@ class JWK(object):
)
self.import_key(**params)
+ def _generate_OKP(self, params):
+ if 'crv' not in params:
+ raise InvalidJWKValue('Must specify "crv" for OKP key generation')
+ try:
+ key = _OKP_CURVES_TABLE[params['crv']].privkey.generate()
+ except KeyError:
+ raise InvalidJWKValue('"%s" is not a supported curve for the '
+ 'OKP key type' % params['crv'])
+ self._import_pyca_pri_okp(key, **params)
+
+ def _import_pyca_pri_okp(self, key, **params):
+ params.update(
+ kty='OKP',
+ crv=params['crv'],
+ d=base64url_encode(key.private_bytes(
+ serialization.Encoding.Raw,
+ serialization.PrivateFormat.Raw,
+ serialization.NoEncryption())),
+ x=base64url_encode(key.public_key().public_bytes(
+ serialization.Encoding.Raw,
+ serialization.PublicFormat.Raw))
+ )
+ self.import_key(**params)
+
+ def _import_pyca_pub_okp(self, key, **params):
+ params.update(
+ kty='OKP',
+ crv=params['crv'],
+ x=base64url_encode(key.public_bytes(
+ serialization.Encoding.Raw,
+ serialization.PrivateFormat.Raw))
+ )
+ self.import_key(**params)
+
def import_key(self, **kwargs):
names = list(kwargs.keys())
@@ -385,7 +506,7 @@ class JWK(object):
raise InvalidJWKValue(
'"%s" is not base64url encoded' % name
)
- if val[3] == ParmType.b64U and name in self._key:
+ if val[3] == ParmType.b64u and name in self._key:
# Check that the value is Base64urlUInt encoded
try:
self._decode_int(self._key[name])
@@ -547,8 +668,8 @@ class JWK(object):
@property
def key_curve(self):
"""The Curve Name."""
- if self._params['kty'] != 'EC':
- raise InvalidJWKType('Not an EC key')
+ if self._params['kty'] not in ['EC', 'OKP']:
+ raise InvalidJWKType('Not an EC or OKP key')
return self._key['crv']
def get_curve(self, arg):
@@ -556,12 +677,12 @@ class JWK(object):
:param arg: an optional curve name
- :raises InvalidJWKType: the key is not an EC key.
+ :raises InvalidJWKType: the key is not an EC or OKP key.
:raises InvalidJWKValue: if the curve names is invalid.
"""
k = self._key
- if self._params['kty'] != 'EC':
- raise InvalidJWKType('Not an EC key')
+ if self._params['kty'] not in ['EC', 'OKP']:
+ raise InvalidJWKType('Not an EC or OKP key')
if arg and k['crv'] != arg:
raise InvalidJWKValue('Curve requested is "%s", but '
'key curve is "%s"' % (arg, k['crv']))
@@ -605,6 +726,22 @@ class JWK(object):
return ec.EllipticCurvePrivateNumbers(self._decode_int(k['d']),
self._ec_pub(k, curve))
+ def _okp_pub(self, k):
+ try:
+ pubkey = _OKP_CURVES_TABLE[k['crv']].pubkey
+ except KeyError:
+ raise InvalidJWKValue('Unknown curve "%s"' % k['crv'])
+
+ return pubkey.from_public_bytes(base64url_decode(k['x']))
+
+ def _okp_pri(self, k):
+ try:
+ privkey = _OKP_CURVES_TABLE[k['crv']].privkey
+ except KeyError:
+ raise InvalidJWKValue('Unknown curve "%s"' % k['crv'])
+
+ return privkey.from_private_bytes(base64url_decode(k['d']))
+
def _get_public_key(self, arg=None):
if self._params['kty'] == 'oct':
return self._key['k']
@@ -612,6 +749,8 @@ class JWK(object):
return self._rsa_pub(self._key).public_key(default_backend())
elif self._params['kty'] == 'EC':
return self._ec_pub(self._key, arg).public_key(default_backend())
+ elif self._params['kty'] == 'OKP':
+ return self._okp_pub(self._key)
else:
raise NotImplementedError
@@ -622,6 +761,8 @@ class JWK(object):
return self._rsa_pri(self._key).private_key(default_backend())
elif self._params['kty'] == 'EC':
return self._ec_pri(self._key, arg).private_key(default_backend())
+ elif self._params['kty'] == 'OKP':
+ return self._okp_pri(self._key)
else:
raise NotImplementedError
@@ -673,6 +814,10 @@ class JWK(object):
self._import_pyca_pri_ec(key)
elif isinstance(key, ec.EllipticCurvePublicKey):
self._import_pyca_pub_ec(key)
+ elif isinstance(key, (Ed25519PrivateKey, Ed448PrivateKey)):
+ self._import_pyca_pri_okp(key)
+ elif isinstance(key, (Ed25519PublicKey, Ed448PublicKey)):
+ self._import_pyca_pub_okp(key)
else:
raise InvalidJWKValue('Unknown key object %r' % key)
=====================================
jwcrypto/jws.py
=====================================
@@ -1,33 +1,27 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
-from collections import namedtuple
-
from jwcrypto.common import JWException
+from jwcrypto.common import JWSEHeaderParameter, JWSEHeaderRegistry
from jwcrypto.common import base64url_decode, base64url_encode
from jwcrypto.common import json_decode, json_encode
from jwcrypto.jwa import JWA
from jwcrypto.jwk import JWK
-
-# RFC 7515 - 9.1
-# name: (description, supported?)
-JWSHeaderParameter = namedtuple('Parameter',
- 'description mustprotect supported')
JWSHeaderRegistry = {
- 'alg': JWSHeaderParameter('Algorithm', False, True),
- 'jku': JWSHeaderParameter('JWK Set URL', False, False),
- 'jwk': JWSHeaderParameter('JSON Web Key', False, False),
- 'kid': JWSHeaderParameter('Key ID', False, True),
- 'x5u': JWSHeaderParameter('X.509 URL', False, False),
- 'x5c': JWSHeaderParameter('X.509 Certificate Chain', False, False),
- 'x5t': JWSHeaderParameter(
- 'X.509 Certificate SHA-1 Thumbprint', False, False),
- 'x5t#S256': JWSHeaderParameter(
- 'X.509 Certificate SHA-256 Thumbprint', False, False),
- 'typ': JWSHeaderParameter('Type', False, True),
- 'cty': JWSHeaderParameter('Content Type', False, True),
- 'crit': JWSHeaderParameter('Critical', True, True),
- 'b64': JWSHeaderParameter('Base64url-Encode Payload', True, True)
+ 'alg': JWSEHeaderParameter('Algorithm', False, True, None),
+ 'jku': JWSEHeaderParameter('JWK Set URL', False, False, None),
+ 'jwk': JWSEHeaderParameter('JSON Web Key', False, False, None),
+ 'kid': JWSEHeaderParameter('Key ID', False, True, None),
+ 'x5u': JWSEHeaderParameter('X.509 URL', False, False, None),
+ 'x5c': JWSEHeaderParameter('X.509 Certificate Chain', False, False, None),
+ 'x5t': JWSEHeaderParameter(
+ 'X.509 Certificate SHA-1 Thumbprint', False, False, None),
+ 'x5t#S256': JWSEHeaderParameter(
+ 'X.509 Certificate SHA-256 Thumbprint', False, False, None),
+ 'typ': JWSEHeaderParameter('Type', False, True, None),
+ 'cty': JWSEHeaderParameter('Content Type', False, True, None),
+ 'crit': JWSEHeaderParameter('Critical', True, True, None),
+ 'b64': JWSEHeaderParameter('Base64url-Encode Payload', True, True, None)
}
"""Registry of valid header parameters"""
@@ -35,7 +29,8 @@ default_allowed_algs = [
'HS256', 'HS384', 'HS512',
'RS256', 'RS384', 'RS512',
'ES256', 'ES384', 'ES512',
- 'PS256', 'PS384', 'PS512']
+ 'PS256', 'PS384', 'PS512',
+ 'EdDSA']
"""Default allowed algorithms"""
@@ -178,16 +173,20 @@ class JWS(object):
This object represent a JWS token.
"""
- def __init__(self, payload=None):
+ def __init__(self, payload=None, header_registry=None):
"""Creates a JWS object.
:param payload(bytes): An arbitrary value (optional).
+ :param header_registry: Optional additions to the header registry
"""
self.objects = dict()
if payload:
self.objects['payload'] = payload
self.verifylog = None
self._allowed_algs = None
+ self.header_registry = JWSEHeaderRegistry(JWSHeaderRegistry)
+ if header_registry:
+ self.header_registry.update(header_registry)
@property
def allowed_algs(self):
@@ -213,6 +212,7 @@ class JWS(object):
return self.objects.get('valid', False)
# TODO: allow caller to specify list of headers it understands
+ # FIXME: Merge and check to be changed to two separate functions
def _merge_check_headers(self, protected, *headers):
header = None
crit = []
@@ -221,11 +221,11 @@ class JWS(object):
crit = protected['crit']
# Check immediately if we support these critical headers
for k in crit:
- if k not in JWSHeaderRegistry:
+ if k not in self.header_registry:
raise InvalidJWSObject(
'Unknown critical header: "%s"' % k)
else:
- if not JWSHeaderRegistry[k][1]:
+ if not self.header_registry[k].supported:
raise InvalidJWSObject(
'Unsupported critical header: "%s"' % k)
header = protected
@@ -239,8 +239,8 @@ class JWS(object):
if header is None:
header = dict()
for h in list(hn.keys()):
- if h in JWSHeaderRegistry:
- if JWSHeaderRegistry[h].mustprotect:
+ if h in self.header_registry:
+ if self.header_registry[h].mustprotect:
raise InvalidJWSObject('"%s" must be protected' % h)
if h in header:
raise InvalidJWSObject('Duplicate header: "%s"' % h)
@@ -266,7 +266,12 @@ class JWS(object):
raise InvalidJWSSignature('Invalid Unprotected header')
# Merge and check (critical) headers
- self._merge_check_headers(p, header)
+ chk_hdrs = self._merge_check_headers(p, header)
+ for hdr in chk_hdrs:
+ if hdr in self.header_registry:
+ if not self.header_registry.check_header(hdr, self):
+ raise InvalidJWSSignature('Failed header check')
+
# check 'alg' is present
if alg is None and 'alg' not in p:
raise InvalidJWSSignature('No "alg" in headers')
@@ -419,7 +424,7 @@ class JWS(object):
:param alg: An optional algorithm name. If already provided as an
element of the protected or unprotected header it can be safely
omitted.
- :param potected: The Protected Header (optional)
+ :param protected: The Protected Header (optional)
:param header: The Unprotected Header (optional)
:raises InvalidJWSObject: if no payload has been set on the object,
=====================================
jwcrypto/jwt.py
=====================================
@@ -108,6 +108,7 @@ class JWTInvalidClaimFormat(JWException):
super(JWTInvalidClaimFormat, self).__init__(msg)
+# deprecated and not used anymore
class JWTMissingKeyID(JWException):
"""Json Web Token is missing key id.
@@ -187,6 +188,7 @@ class JWT(object):
self._check_claims = None
self._leeway = 60 # 1 minute clock skew allowed
self._validity = 600 # 10 minutes validity (up to 11 with leeway)
+ self.deserializelog = None
if header:
self.header = header
@@ -462,28 +464,37 @@ class JWT(object):
if self._algs:
self.token.allowed_algs = self._algs
+ self.deserializelog = list()
# now deserialize and also decrypt/verify (or raise) if we
# have a key
if key is None:
self.token.deserialize(jwt, None)
elif isinstance(key, JWK):
self.token.deserialize(jwt, key)
+ self.deserializelog.append("Success")
elif isinstance(key, JWKSet):
self.token.deserialize(jwt, None)
- if 'kid' not in self.token.jose_header:
- raise JWTMissingKeyID('No key ID in JWT header')
-
- token_key = key.get_key(self.token.jose_header['kid'])
- if not token_key:
- raise JWTMissingKey('Key ID %s not in key set'
- % self.token.jose_header['kid'])
-
- if isinstance(self.token, JWE):
- self.token.decrypt(token_key)
- elif isinstance(self.token, JWS):
- self.token.verify(token_key)
+ if 'kid' in self.token.jose_header:
+ kid_key = key.get_key(self.token.jose_header['kid'])
+ if not kid_key:
+ raise JWTMissingKey('Key ID %s not in key set'
+ % self.token.jose_header['kid'])
+ self.token.deserialize(jwt, kid_key)
else:
- raise RuntimeError("Unknown Token Type")
+ for k in key:
+ try:
+ self.token.deserialize(jwt, k)
+ self.deserializelog.append("Success")
+ break
+ except Exception as e: # pylint: disable=broad-except
+ keyid = k.key_id
+ if keyid is None:
+ keyid = k.thumbprint()
+ self.deserializelog.append('Key [%s] failed: [%s]' % (
+ keyid, repr(e)))
+ continue
+ if "Success" not in self.deserializelog:
+ raise JWTMissingKey('No working key found in key set')
else:
raise ValueError("Unrecognized Key Type")
=====================================
jwcrypto/tests.py
=====================================
@@ -14,6 +14,8 @@ from jwcrypto import jwe
from jwcrypto import jwk
from jwcrypto import jws
from jwcrypto import jwt
+from jwcrypto.common import InvalidJWSERegOperation
+from jwcrypto.common import JWSEHeaderParameter
from jwcrypto.common import base64url_decode, base64url_encode
from jwcrypto.common import json_decode, json_encode
@@ -234,6 +236,29 @@ zl9HYIMxATFyqSiD9jsx
PublicCertThumbprint = u'7KITkGJF74IZ9NKVvHfuJILbuIZny6j-roaNjB1vgiA'
+# RFC 8037 - A.2
+PublicKeys_EdDsa = {
+ "keys": [
+ {
+ "kty": "OKP",
+ "crv": "Ed25519",
+ "x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo"
+ },
+ ],
+ "thumbprints": ["kPrK_qmxVWaYVA9wwBF6Iuo3vVzz7TxHCTwXBygrS4k"]
+}
+
+# RFC 8037 - A.1
+PrivateKeys_EdDsa = {
+ "keys": [
+ {
+ "kty": "OKP",
+ "crv": "Ed25519",
+ "d": "nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A",
+ "x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo"},
+ ]
+}
+
class TestJWK(unittest.TestCase):
def test_create_pubKeys(self):
@@ -295,6 +320,11 @@ class TestJWK(unittest.TestCase):
key = jwk.JWK.generate(kty='EC', curve='P-256', crv='P-521')
key.get_curve('P-521')
+ def test_generate_OKP_keys(self):
+ for crv in jwk.ImplementedOkpCurves:
+ key = jwk.JWK.generate(kty='OKP', crv=crv)
+ self.assertEqual(key.get_curve(crv), crv)
+
def test_import_pyca_keys(self):
rsa1 = rsa.generate_private_key(65537, 1024, default_backend())
krsa1 = jwk.JWK.from_pyca(rsa1)
@@ -411,6 +441,23 @@ class TestJWK(unittest.TestCase):
with self.assertRaises(jwk.InvalidJWKValue):
jwk.JWK(kty='oct', k=b'\x01')
+ def test_create_pubKeys_eddsa(self):
+ keylist = PublicKeys_EdDsa['keys']
+ for key in keylist:
+ jwk.JWK(**key)
+
+ def test_create_priKeys_eddsa(self):
+ keylist = PrivateKeys_EdDsa['keys']
+ for key in keylist:
+ jwk.JWK(**key)
+
+ def test_thumbprint_eddsa(self):
+ for i in range(0, len(PublicKeys_EdDsa['keys'])):
+ k = jwk.JWK(**PublicKeys_EdDsa['keys'][i])
+ self.assertEqual(
+ k.thumbprint(),
+ PublicKeys_EdDsa['thumbprints'][i])
+
# RFC 7515 - A.1
A1_protected = \
@@ -613,6 +660,20 @@ E_negative = \
'ZJTkVEIl0sDQogImh0dHA6Ly9leGFtcGxlLmNvbS9VTkRFRklORUQiOnRydWUNCn0.' + \
'RkFJTA.'
+customhdr_jws_example = \
+ '{' + \
+ '"payload":' + \
+ '"eyJpc3MiOiJqb2UiLA0KICJleHAiOjEzMDA4MTkzODAsDQogImh0dHA6Ly9leGF' + \
+ 'tcGxlLmNvbS9pc19yb290Ijp0cnVlfQ",' + \
+ '"protected":"eyJhbGciOiJFUzI1NiJ9",' + \
+ '"header":' + \
+ '{"kid":"e9bc097a-ce51-4036-9562-d2ade882db0d", ' + \
+ '"custom1":"custom_val"},' + \
+ '"signature":' + \
+ '"DtEhU3ljbEg8L38VWAfUAqOyKAM6-Xx-F4GawxaepmXFCgfTjDxw5djxLa8IS' + \
+ 'lSApmWQxfKTUJqPP3-Kg6NU1Q"' + \
+ '}'
+
class TestJWS(unittest.TestCase):
def check_sign(self, test):
@@ -651,8 +712,7 @@ class TestJWS(unittest.TestCase):
self.check_sign, A5_example)
a5_bis = {'allowed_algs': ['none']}
a5_bis.update(A5_example)
- with self.assertRaises(jws.InvalidJWSSignature):
- self.check_sign(a5_bis)
+ self.check_sign(a5_bis)
def test_A6(self):
s = jws.JWS(A6_example['payload'])
@@ -679,6 +739,53 @@ class TestJWS(unittest.TestCase):
jws.InvalidJWSSignature(s.deserialize, E_negative)
s.verify(None)
+ def test_customhdr_jws(self):
+ # Test pass header check
+ def jws_chk1(jwobj):
+ return jwobj.jose_header['custom1'] == 'custom_val'
+
+ newhdr = JWSEHeaderParameter('Custom header 1', False, True, jws_chk1)
+ newreg = {'custom1': newhdr}
+ s = jws.JWS(A6_example['payload'], header_registry=newreg)
+ s.deserialize(customhdr_jws_example, A6_example['key2'])
+
+ # Test fail header check
+ def jws_chk2(jwobj):
+ return jwobj.jose_header['custom1'] == 'custom_not'
+
+ newhdr = JWSEHeaderParameter('Custom header 1', False, True, jws_chk2)
+ newreg = {'custom1': newhdr}
+ s = jws.JWS(A6_example['payload'], header_registry=newreg)
+ with self.assertRaises(jws.InvalidJWSSignature):
+ s.deserialize(customhdr_jws_example, A6_example['key2'])
+
+ def test_customhdr_jws_exists(self):
+ newhdr = JWSEHeaderParameter('Custom header 1', False, True, None)
+ newreg = {'alg': newhdr}
+ with self.assertRaises(InvalidJWSERegOperation):
+ jws.JWS(A6_example['payload'], header_registry=newreg)
+
+ def test_EdDsa_signing_and_verification(self):
+ examples = []
+ if 'Ed25519' in jwk.ImplementedOkpCurves:
+ examples = [E_Ed25519]
+ for curve_example in examples:
+ key = jwk.JWK.from_json(curve_example['key_json'])
+ payload = curve_example['payload']
+ protected_header = curve_example['protected_header']
+ jws_test = jws.JWS(payload)
+ jws_test.add_signature(key, None,
+ json_encode(protected_header), None)
+ jws_test_serialization_compact = \
+ jws_test.serialize(compact=True)
+ self.assertEqual(jws_test_serialization_compact,
+ curve_example['jws_serialization_compact'])
+ jws_verify = jws.JWS()
+ jws_verify.deserialize(jws_test_serialization_compact)
+ jws_verify.verify(key.public())
+ self.assertEqual(jws_verify.payload.decode('utf-8'),
+ curve_example['payload'])
+
E_A1_plaintext = \
[84, 104, 101, 32, 116, 114, 117, 101, 32, 115, 105, 103, 110, 32,
@@ -839,6 +946,16 @@ E_A5_ex = \
'"ciphertext":"KDlTtXchhZTGufMYmOYGS4HffxPSUrfmqCHXaI9wOGY",' \
'"tag":"Mz-VPPyU4RlcuYv1IwIvzw"}'
+customhdr_jwe_ex = \
+ '{"protected":"eyJlbmMiOiJBMTI4Q0JDLUhTMjU2In0",' \
+ '"unprotected":{"jku":"https://server.example.com/keys.jwks"},' \
+ '"header":{"alg":"A128KW","kid":"7", "custom1":"custom_val"},' \
+ '"encrypted_key":' \
+ '"6KB707dM9YTIgHtLvtgWQ8mKwboJW3of9locizkDTHzBC2IlrT1oOQ",' \
+ '"iv":"AxY8DCtDaGlsbGljb3RoZQ",' \
+ '"ciphertext":"KDlTtXchhZTGufMYmOYGS4HffxPSUrfmqCHXaI9wOGY",' \
+ '"tag":"Mz-VPPyU4RlcuYv1IwIvzw"}'
+
Issue_136_Protected_Header_no_epk = {
"alg": "ECDH-ES+A256KW",
"enc": "A256CBC-HS512"}
@@ -862,6 +979,23 @@ Issue_136_Contributed_Key = {
"x": "FPrb_xwxe8SBP3kO-e-WsofFp7n5-yc_tGgfAvqAP8g",
"y": "lM3HuyKMYUVsYdGqiWlkwTZbGO3Fh-hyadq8lfkTgBc"}
+# RFC 8037 A.1
+E_Ed25519 = {
+ 'key_json': '{"kty": "OKP",'
+ '"crv": "Ed25519", '
+ '"d": "nWGxne_9WmC6hEr0kuwsxERJxWl7MmkZcDusAxyuf2A", '
+ '"x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo"}',
+ 'payload': 'Example of Ed25519 signing',
+ 'protected_header': {"alg": "EdDSA"},
+ 'jws_serialization_compact': 'eyJhbGciOiJFZERTQSJ9.RXhhbXBsZSBvZiBF'
+ 'ZDI1NTE5IHNpZ25pbmc.hgyY0il_MGCjP0Jzl'
+ 'nLWG1PPOt7-09PGcvMg3AIbQR6dWbhijcNR4ki'
+ '4iylGjg5BhVsPt9g7sVvpAr_MuM0KAg'}
+
+X25519_Protected_Header_no_epk = {
+ "alg": "ECDH-ES+A128KW",
+ "enc": "A128GCM"}
+
class TestJWE(unittest.TestCase):
def check_enc(self, plaintext, protected, key, vector):
@@ -938,6 +1072,42 @@ class TestJWE(unittest.TestCase):
e.deserialize(Issue_136_Contributed_JWE,
jwk.JWK(**Issue_136_Contributed_Key))
+ def test_customhdr_jwe(self):
+ def jwe_chk1(jwobj):
+ return jwobj.jose_header['custom1'] == 'custom_val'
+
+ newhdr = JWSEHeaderParameter('Custom header 1', False, True, jwe_chk1)
+ newreg = {'custom1': newhdr}
+ e = jwe.JWE(header_registry=newreg)
+ e.deserialize(customhdr_jwe_ex, E_A4_ex['key2'])
+
+ def jwe_chk2(jwobj):
+ return jwobj.jose_header['custom1'] == 'custom_not'
+
+ newhdr = JWSEHeaderParameter('Custom header 1', False, True, jwe_chk2)
+ newreg = {'custom1': newhdr}
+ e = jwe.JWE(header_registry=newreg)
+ with self.assertRaises(jwe.InvalidJWEData):
+ e.deserialize(customhdr_jwe_ex, E_A4_ex['key2'])
+
+ def test_customhdr_jwe_exists(self):
+ newhdr = JWSEHeaderParameter('Custom header 1', False, True, None)
+ newreg = {'alg': newhdr}
+ with self.assertRaises(InvalidJWSERegOperation):
+ jwe.JWE(header_registry=newreg)
+
+ def test_X25519_ECDH(self):
+ plaintext = b"plain"
+ protected = json_encode(X25519_Protected_Header_no_epk)
+ if 'X25519' in jwk.ImplementedOkpCurves:
+ x25519key = jwk.JWK.generate(kty='OKP', crv='X25519')
+ e1 = jwe.JWE(plaintext, protected)
+ e1.add_recipient(x25519key)
+ enc = e1.serialize()
+ e2 = jwe.JWE()
+ e2.deserialize(enc, x25519key)
+ self.assertEqual(e2.payload, plaintext)
+
MMA_vector_key = jwk.JWK(**E_A2_key)
MMA_vector_ok_cek = \
@@ -1094,20 +1264,35 @@ class TestJWT(unittest.TestCase):
def test_decrypt_keyset(self):
key = jwk.JWK(kid='testkey', **E_A2_key)
- keyset = jwk.JWKSet()
- # decrypt without keyid
- t = jwt.JWT(A1_header, A1_claims)
+ keyset = jwk.JWKSet.from_json(json_encode(PrivateKeys))
+
+ # encrypt a new JWT with kid
+ header = copy.copy(A1_header)
+ header['kid'] = 'testkey'
+ t = jwt.JWT(header, A1_claims)
t.make_encrypted_token(key)
token = t.serialize()
- self.assertRaises(jwt.JWTMissingKeyID, jwt.JWT, jwt=token,
- key=keyset)
- # encrypt a new JWT
+ # try to decrypt without a matching key
+ self.assertRaises(jwt.JWTMissingKey, jwt.JWT, jwt=token, key=keyset)
+ # now decrypt with key
+ keyset.add(key)
+ jwt.JWT(jwt=token, key=keyset, check_claims={'exp': 1300819380})
+
+ # encrypt a new JWT with wrong kid
header = copy.copy(A1_header)
- header['kid'] = 'testkey'
+ header['kid'] = '1'
t = jwt.JWT(header, A1_claims)
t.make_encrypted_token(key)
token = t.serialize()
- # try to decrypt without key
+ self.assertRaises(jwe.InvalidJWEData, jwt.JWT, jwt=token, key=keyset)
+
+ keyset = jwk.JWKSet.from_json(json_encode(PrivateKeys))
+ # encrypt a new JWT with no kid
+ header = copy.copy(A1_header)
+ t = jwt.JWT(header, A1_claims)
+ t.make_encrypted_token(key)
+ token = t.serialize()
+ # try to decrypt without a matching key
self.assertRaises(jwt.JWTMissingKey, jwt.JWT, jwt=token, key=keyset)
# now decrypt with key
keyset.add(key)
@@ -1238,6 +1423,19 @@ class ConformanceTests(unittest.TestCase):
check.deserialize(enc, key)
self.assertEqual(b'plain', check.payload)
+ def test_none_key(self):
+ e = "eyJhbGciOiJub25lIn0." + \
+ "eyJpc3MiOiJqb2UiLCJodHRwOi8vZXhhbXBsZS5jb20vaXNfcm9vdCI6dHJ1ZX0."
+ token = jwt.JWT(algs=['none'])
+ k = jwk.JWK(generate='oct', size=0)
+ token.deserialize(jwt=e, key=k)
+ self.assertEqual(json_decode(token.claims),
+ {"iss": "joe", "http://example.com/is_root": True})
+ with self.assertRaises(KeyError):
+ token = jwt.JWT()
+ token.deserialize(jwt=e)
+ json_decode(token.claims)
+
class JWATests(unittest.TestCase):
def test_jwa_create(self):
@@ -1246,6 +1444,8 @@ class JWATests(unittest.TestCase):
self.assertIn(cls.algorithm_usage_location, {'alg', 'enc'})
if name == 'ECDH-ES':
self.assertIs(cls.keysize, None)
+ elif name == 'EdDSA':
+ self.assertIs(cls.keysize, None)
else:
self.assertIsInstance(cls.keysize, int)
self.assertGreaterEqual(cls.keysize, 0)
=====================================
setup.py
=====================================
@@ -6,7 +6,7 @@ from setuptools import setup
setup(
name = 'jwcrypto',
- version = '0.6.0',
+ version = '0.7',
license = 'LGPLv3+',
maintainer = 'JWCrypto Project Contributors',
maintainer_email = 'simo at redhat.com',
=====================================
tox.ini
=====================================
@@ -8,7 +8,7 @@ setenv =
deps =
pytest
coverage
-sitepackages = True
+#sitepackages = True
commands =
{envpython} -bb -m coverage run -m pytest --capture=no --strict {posargs}
{envpython} -m coverage report -m
@@ -17,7 +17,7 @@ commands =
basepython = python2.7
deps =
pylint
-sitepackages = True
+#sitepackages = True
commands =
{envpython} -m pylint -d c,r,i,W0613 -r n -f colorized --notes= --disable=star-args ./jwcrypto
View it on GitLab: https://salsa.debian.org/freeipa-team/python-jwcrypto/-/compare/51b5f66a6140449fa91a9b85630332f751bfbdeb...904c7902fff8514f1f42161c8e1fa23d11e75c76
--
View it on GitLab: https://salsa.debian.org/freeipa-team/python-jwcrypto/-/compare/51b5f66a6140449fa91a9b85630332f751bfbdeb...904c7902fff8514f1f42161c8e1fa23d11e75c76
You're receiving this email because of your account on salsa.debian.org.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://alioth-lists.debian.net/pipermail/pkg-freeipa-devel/attachments/20200421/b8b23eea/attachment-0001.html>
More information about the Pkg-freeipa-devel
mailing list