[Pkg-freeipa-devel] [Git][freeipa-team/python-jwcrypto][upstream] 64 commits: Add setup.cfg with release aliases
Timo Aaltonen
gitlab at salsa.debian.org
Tue Apr 2 07:07:37 BST 2019
Timo Aaltonen pushed to branch upstream at FreeIPA packaging / python-jwcrypto
Commits:
a3499b7e by Christian Heimes at 2016-08-19T16:16:41Z
Add setup.cfg with release aliases
setup.cfg adds a flag for universal builds and two aliases to build and
upload packages. A PyPY release becomes as simple as
python setup.py release
Signed-off-by: Christian Heimes <cheimes at redhat.com>
Reviewed-by: Simo Sorce <simo at redhat.com>
Closes #52
- - - - -
af69ffda by Simo Sorce at 2016-08-23T14:03:18Z
Add Examples for JWE and JWS usage
Resolves #53
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Hanno Schlichting <hanno at hannosch.eu>
Closes #54
- - - - -
7c452b96 by Simo Sorce at 2016-08-23T14:03:47Z
Allow to pass in dictionaries for headers
Where it makes sense, allow to pass in a dictionary for the various headers
and auto-encode them as needed.
Resolves #53
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Hanno Schlichting <hanno at hannosch.eu>
Closes #55
- - - - -
e0b3d9d2 by Simo Sorce at 2016-08-23T14:03:57Z
Allow to pass a default reciepient to a JWE
Resolves #53
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Hanno Schlichting <hanno at hannosch.eu>
Closes #56
- - - - -
bf54a6c3 by Simo Sorce at 2016-08-25T17:20:25Z
Move JWS Algorithms implementation
Create a new jwa module that hosts the actual JWA crypto defined in
RFC 7518: JSON Web Algorithms (JWA)
Also slightly changes the "none" algorithm to always raise an InvalidSignature
error on verify(), as the None signature can never be verified after all.
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Christian Heimes <cheimes at redhat.com>
- - - - -
4df4a7b3 by Simo Sorce at 2016-08-25T17:20:30Z
Move JWE algorithms to JWA as well
This keeps all JWA defined algorithms in the same module and make all
algorithms use a common way of instantiating and handling crypto
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Christian Heimes <cheimes at redhat.com>
Closes #57
Closes #58
- - - - -
06da1cd5 by Simo Sorce at 2016-08-26T16:55:48Z
Allow to pass an 'alg' argument to key generation
If an 'alg' argument is apssed in at key generation, then it can be used
to determine the key size appropriate for the desired algorithm.
An explicit 'size' argument always takes precendence.
An 'alg' parameter is accpeted for 'oct' and 'RSA' keys only.
Resolves #50
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Christian Heimes <cheimes at redhat.com>
Closes #59
- - - - -
f7edc391 by Simo Sorce at 2016-08-26T17:17:22Z
Consistently use bit for key sizes
Always use bits to mesure key material sizes through the code.
Resolves #49
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Christian Heimes <cheimes at redhat.com>
Closes #60
- - - - -
ccee4027 by Simo Sorce at 2016-08-26T17:48:08Z
Minor style fixes
Streamline some of the code by using better python
conventions/style/syntax/constricts
Resolves #61
Resolves #62
Resolves #63
Signed-off-by: Simo Sorce <simo at redhat.com>
Reviewed-by: Christian Heimes <cheimes at redhat.com>
Closes #64
- - - - -
9baa1bfa by Simo Sorce at 2016-08-26T18:34:29Z
JWK: Add interface to load from and export to PEM
Keys can be load from PEM files, and Public Keys even from PEM files that have
X509 certificates in it. Keys can be exported (either Private or Public) to
PEM files.
Signed-off-by: Simo Sorce <simo at redhat.com>
Closes: #25
- - - - -
9282e1e9 by Simo Sorce at 2016-08-26T18:34:37Z
Add helpers to test and export specific key forms
New test properties are:
- is_symmetric
True if key is symmetric (kty=oct) otherwise False
- has_private
True if the key is not symmetric and the key has values
marked 'Private' according to the JWKValuesRegistry
- has_public
True if the key is not symmetric and the key has values
marked 'Public' according to the JWKValuesRegistry
New export helpers are:
- export_private()
Succeeds only if the key 'has_private'
- export_symmetric()
Succeeds only if the key 'is_symmetric'
Un-deprecates export_public() but adds checks to make it fail if
'has_public' returns False
Adds tests.
Signed-off-by: Simo Sorce <simo at redhat.com>
Closes #45
- - - - -
eb5be5bd by Simo Sorce at 2016-08-31T19:38:09Z
CVE-2016-6298: Million Messages Attack mitigation
RFC 3218 describes an oracle attack called Million Messages Attack
against RSA with PKCS1 v1.5 padding.
Depending on how JWEs are used a server may become an Oracle, and the
mitigation presecribed in RFC 3218 2.3.2 need to be implemented.
Many thanks to Dennis Detering for his responsible disclosure and help
verifying the mitigation approach.
Resolves #65
Signed-off-by: Simo Sorce <simo at redhat.com>
Closes #66
- - - - -
19490b84 by Simo Sorce at 2016-08-31T19:38:28Z
Add Timing tests for MMA
This test is not very reliable and takes a long time so it is provided but
diasabled by default.
It is only useful to verify if any regression regarding MMA occurs, so it can
be just run occasionally.
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
2869305d by Christian Heimes at 2016-09-08T18:14:22Z
Fix 'ECDH-ES' base classes and add tests
ECDH-ES is an actual JWA algorithm and not a base class. Therefore it
must be a subclass of JWAAlgorithm. New tests will catch these errors in
the future.
Signed-off-by: Christian Heimes <cheimes at redhat.com>
Reviewed-by: Simo Sorce <simo at redhat.com>
Closes #67
- - - - -
1ac6f479 by Simo Sorce at 2016-11-28T11:25:32Z
Fix trvis's flake8 complaints
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
09d05131 by Ash Berlin at 2016-11-28T11:29:00Z
Add 'cryptography' to requirements in setup.py
This is so that you can add just jwcrypto to the requirements of a
downstream project and have it install everything needed -- previously
you would also have to add cryptography yourself.
Since this was the only thing in requirements.txt I have removed the
file
Reviewed-by: Simo Sorce <simo at redhat.com>
Fixes #69
Closes #70
- - - - -
53cc2720 by Carlos Jenkins at 2016-11-28T21:36:46Z
Fixed nbf incorrect validation.
Reviewed-by: Simo Sorce <simo at redhat.com>
Fixes #71
Closes #72
- - - - -
c85a52be by Christian Heimes at 2016-11-29T19:26:37Z
Preparing release 0.4.0
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
b9949a79 by Christian Heimes at 2016-11-29T19:50:28Z
Post release bump
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
bb42eca5 by Simo Sorce at 2017-07-13T08:49:53Z
Remove useless tox command that breaks travis
Signed-off-by: Simo Sorce <simo at redhat.com>
Close #81
- - - - -
19c4d97a by Yann Cézard at 2017-07-13T08:53:44Z
Set claims after reg_claims (or _add_default_claims will never be called).
- - - - -
f38ff2c3 by James Gardiner at 2017-07-13T08:54:46Z
Fix typo in jwk chapter of docs
- - - - -
ffa4ddae by Simo Sorce at 2017-07-24T09:09:45Z
Cyrptography deprecated signer and verifier
Use sign() and verify() directly instead.
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
9203242c by Christian Heimes at 2017-07-24T09:13:03Z
Preparing release 0.4.1
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
dfd400d8 by Christian Heimes at 2017-07-24T09:17:10Z
Post release bump
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
a55d16c7 by Christian Heimes at 2017-08-01T13:08:11Z
Include tox.ini and setup.cfg in sdist
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
323327b6 by Christian Heimes at 2017-08-01T13:21:45Z
Fix bytes/str comparison in JWE
In Python 3, '' != b''. With bytes warning enabled, comparison of bytes
and str raise an exception, too.
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
68bac873 by Christian Heimes at 2017-08-01T13:21:45Z
Remove unnecessary calls to str()
json_decode() returns str instances for str values anyway. In case the
JSON payload contains invalid types, base64 codec will fail with an
appropriate error message.
'some.string'.split('.') returns a list of str.
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
8b41d34e by Christian Heimes at 2017-08-01T15:50:39Z
Support and test with Python 3.6
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
c4637c14 by Christian Heimes at 2017-08-01T15:56:23Z
Preparing release 0.4.2
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
5ecfa17d by Christian Heimes at 2017-08-01T15:57:44Z
Post release bump to 0.5.dev3
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
adeaa746 by Michael Boulton at 2017-08-22T12:48:45Z
Fix minor typo
- - - - -
0dab54b0 by Michael Boulton at 2017-10-09T17:33:07Z
Fix error interpolation on nonstandard jwt claims
If the claim passed was not one of sub, aud, iat, etc. then passing a string as a value to check_claims would raise an error if it was not an integer
Add a working test as well
- - - - -
3db0271a by Simo Sorce at 2017-11-14T15:41:42Z
Make all exception derive from JWException
Fixes #93
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
e21a20ab by Simo Sorce at 2017-11-14T15:41:42Z
Make newer linters happy
- - - - -
fd425f38 by Michael Boulton at 2017-11-30T18:33:38Z
Add example of encrypting/decrypting using asymmetric keys
- - - - -
a03a0360 by Simo Sorce at 2017-11-30T18:34:14Z
Add build pass/fail indicator to main page
- - - - -
8e438cbe by Simo Sorce at 2017-12-05T18:03:34Z
Always augment claims if defaults are provided
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
31079736 by Simo Sorce at 2017-12-05T18:04:15Z
Add function to get a public only JWK
This makes it simpler to get a JWK that contains excluively a pulic
key from an existing JWK. It fails if no public key is available in
the source JWK.
Only known public elements are returned. "Unknown" attributes are
not copied over.
Signed-off-by: Simo Sorce <simo at redhat.com>
Resolves #99
- - - - -
9721a314 by Simo Sorce at 2017-12-13T20:31:01Z
Better validation of JWE compact serialization
Fixes #92
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
3bc514d4 by Simo Sorce at 2017-12-14T16:59:54Z
Test header using string not dict
Change one test to also test using strings instead of dicts as headers.
This is supported in the code but was not tested.
Signed-off-by: Simo Sorce <simo at redhat.com>
Closes #96
- - - - -
4d92702e by Simo Sorce at 2017-12-14T17:49:56Z
Get the protected header from the right place
Modify test to check the header is emitted correctly
Signed-off-by: Simo Sorce <simo at redhat.com>
Fixes #105
- - - - -
75ff511b by Simo Sorce at 2017-12-14T17:59:28Z
Fix potential undefined usage spotted by lgtm.com
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
69c845c7 by Simo Sorce at 2018-01-03T14:37:42Z
Fix lint issues in travis
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
bc3708a4 by Michael Boulton at 2018-01-30T13:40:14Z
Fix misleading error message
- - - - -
39e4ff51 by Eamonn Nugent at 2018-03-19T13:31:42Z
Fixing "A128KW" under "_EcdhEsAes256Kw"
There was a typo in the "_EcdhEsAes256Kw" class. I have changed "A128KW" in the description for the class to become "A256KW" in accordance with the algorithm.
- - - - -
34e7919e by Simo Sorce at 2018-04-11T17:33:24Z
Add Makefile testing vs python 3.6
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
c527234e by Simo Sorce at 2018-04-11T17:33:24Z
Use new pyca function names to avoid deprecation warnings
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
4f3ae07c by Simo Sorce at 2018-05-16T14:20:18Z
Add method to import a JWK directly from json
JWK.from_json(<json_string>) will import a previously exported key.
Signed-off-by: Simo Sorce <simo at redhat.com>
Fixes: #116
- - - - -
0ff425e4 by Simo Sorce at 2018-05-16T14:20:40Z
Make the default iterator return keys directly
It has no value to return the 'keys' attribute itself, instead we now
we return the inner 'keys' object iterator when the JWKSet is iterated
over.
This way iterating over the JWKSet object returns its JWK objects.
Signed-off-by: Simo Sorce <simo at redhat.com>
Fixes #117
- - - - -
b367cfb0 by Simo Sorce at 2018-06-19T14:47:11Z
Validate key parameters on import
Fixes #120
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
e4b08846 by Simo Sorce at 2018-06-27T08:20:37Z
Rework internal JWK structures for clarity
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
0115c859 by Christian Heimes at 2018-06-27T08:21:01Z
Test with Python 3.7
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
76e358d0 by Christian Heimes at 2018-06-27T10:25:58Z
Release 0.5.0
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
02161189 by Christian Heimes at 2018-06-27T10:26:52Z
Post release bump to 0.6.dev1
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
9f441be9 by mbaldwin at 2018-09-14T00:59:20Z
jti should be a string
- - - - -
d0c3ee27 by Li lin at 2018-10-01T12:09:46Z
add missing module name
- - - - -
e319b0c0 by Simo Sorce at 2018-10-31T19:43:20Z
Use python-cryptography's AES key wrapping
Stop using our own, pyca has has AES Key wrap support for long enough
now.
Resolves #137
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
5f9d86ae by Simo Sorce at 2018-10-31T19:59:36Z
Add tests for key wrapping where CEK < KEK
When using ECDH-ES+A[128|192|256]WK key agreement and A128CBC-HS256
encryption the Content Encrypition Key (CEK) has a bigger size than
the Key Encryption Key (KEK).
Thess tests makes sure we properly handle this case.
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
13421b48 by Simo Sorce at 2018-10-31T19:59:36Z
Fix ECDH-ES key exchange for CEK greater than KEK
The code was incorrectly assigning key size for derivation
and wrapping when they differ in size.
Fixes #136
Signed-off-by: Simo Sorce <simo at redhat.com>
- - - - -
2220d100 by Simo Sorce at 2018-10-31T20:10:25Z
Add support for RFC7797
Together with non encoded payloads this commit also adds sull support
for dealing with detached payloads on deserlization and serialization.
The payload must still be provided for any computation, but can be
removed before serialization and add after deserialization of a JWS with
detached payload.
Signed-off-by: Simo Sorce <simo at redhat.com>
Fixes #4
- - - - -
3c9000fb by Jonathan Huot at 2018-10-31T20:11:10Z
Fixed typo
- - - - -
ea21d349 by Jonathan Huot at 2018-10-31T20:14:04Z
Fix JWK.from_json #130
import_keyset no longer returns self.
- - - - -
5d991eb8 by Christian Heimes at 2018-11-05T15:14:47Z
Preparing release 0.6.0
Signed-off-by: Christian Heimes <cheimes at redhat.com>
- - - - -
19 changed files:
- .travis.yml
- MANIFEST.in
- Makefile
- README.md
- docs/source/conf.py
- docs/source/jwe.rst
- docs/source/jwk.rst
- docs/source/jws.rst
- jwcrypto/common.py
- + jwcrypto/jwa.py
- jwcrypto/jwe.py
- jwcrypto/jwk.py
- jwcrypto/jws.py
- jwcrypto/jwt.py
- jwcrypto/tests.py
- − requirements.txt
- + setup.cfg
- setup.py
- tox.ini
Changes:
=====================================
.travis.yml
=====================================
@@ -12,15 +12,17 @@ matrix:
env: TOXENV=py34
- python: 3.5
env: TOXENV=py35
- - python: 3.5
+ - python: 3.6
+ env: TOXENV=py36
+ - python: 3.6
env: TOXENV=doc
- - python: 3.5
+ - python: 3.6
env: TOXENV=sphinx
- - python: 3.5
+ - python: 3.6
env: TOXENV=lint
- python: 2.7
env: TOXENV=pep8py2
- - python: 3.5
+ - python: 3.6
env: TOXENV=pep8py3
install:
=====================================
MANIFEST.in
=====================================
@@ -1 +1,2 @@
include LICENSE README.md
+include tox.ini setup.cfg
=====================================
Makefile
=====================================
@@ -17,11 +17,19 @@ clean:
cscope:
git ls-files | xargs pycscope
+testlong: export JWCRYPTO_TESTS_ENABLE_MMA=True
+testlong: export TOX_TESTENV_PASSENV=JWCRYPTO_TESTS_ENABLE_MMA
+testlong:
+ rm -f .coverage
+ tox -e py36
+
test:
rm -f .coverage
tox -e py27
tox -e py34 --skip-missing-interpreter
tox -e py35 --skip-missing-interpreter
+ tox -e py36 --skip-missing-interpreter
+ tox -e py37 --skip-missing-interpreter
DOCS_DIR = docs
.PHONY: docs
=====================================
README.md
=====================================
@@ -1,3 +1,5 @@
+[](https://travis-ci.org/latchset/jwcrypto)
+
JWCrypto
========
=====================================
docs/source/conf.py
=====================================
@@ -46,16 +46,16 @@ master_doc = 'index'
# General information about the project.
project = u'JWCrypto'
-copyright = u'2016, JWCrypto Contributors'
+copyright = u'2016-2018, JWCrypto Contributors'
# The version info for the project you're documenting, acts as replacement for
# |version| and |release|, also used in various other places throughout the
# built documents.
#
# The short X.Y version.
-version = '0.3'
+version = '0.6'
# The full version, including alpha/beta/rc tags.
-release = '0.3.1'
+release = '0.6'
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
=====================================
docs/source/jwe.rst
=====================================
@@ -47,3 +47,52 @@ Registries
.. autodata:: jwcrypto.jwe.JWEHeaderRegistry
:annotation:
+
+Examples
+--------
+
+Symmetric keys
+~~~~~~~~~~~~~~
+
+Encrypt a JWE token::
+ >>> from jwcrypto import jwk, jwe
+ >>> from jwcrypto.common import json_encode
+ >>> key = jwk.JWK.generate(kty='oct', size=256)
+ >>> payload = "My Encrypted message"
+ >>> jwetoken = jwe.JWE(payload.encode('utf-8'),
+ json_encode({"alg": "A256KW",
+ "enc": "A256CBC-HS512"}))
+ >>> jwetoken.add_recipient(key)
+ >>> enc = jwetoken.serialize()
+
+Decrypt a JWE token::
+ >>> jwetoken = jwe.JWE()
+ >>> jwetoken.deserialize(enc)
+ >>> jwetoken.decrypt(key)
+ >>> payload = jwetoken.payload
+
+Asymmetric keys
+~~~~~~~~~~~~~~~
+
+Encrypt a JWE token::
+ >>> from jwcrypto import jwk, jwe
+ >>> from jwcrypto.common import json_encode, json_decode
+ >>> public_key = jwk.JWK()
+ >>> private_key = jwk.JWK.generate(kty='RSA', size=2048)
+ >>> public_key.import_key(**json_decode(private_key.export_public()))
+ >>> payload = "My Encrypted message"
+ >>> protected_header = {
+ "alg": "RSA-OAEP-256",
+ "enc": "A256CBC-HS512",
+ "typ": "JWE",
+ "kid": public_key.thumbprint(),
+ }
+ >>> jwetoken = jwe.JWE(payload.encode('utf-8'),
+ recipient=public_key,
+ protected=protected_header)
+ >>> enc = jwetoken.serialize()
+
+Decrypt a JWE token::
+ >>> jwetoken = jwe.JWE()
+ >>> jwetoken.deserialize(enc, key=private_key)
+ >>> payload = jwetoken.payload
=====================================
docs/source/jwk.rst
=====================================
@@ -85,3 +85,6 @@ Import a P-256 Public Key::
"crv":"P-256","kty":"EC"}
>>> key = jwk.JWK(**expkey)
+Import a Key from a PEM file::
+ >>> with open("public.pem", "rb") as pemfile:
+ >>> key = jwk.JWK.from_pem(pemfile.read())
=====================================
docs/source/jws.rst
=====================================
@@ -43,3 +43,23 @@ Registries
.. autodata:: jwcrypto.jws.JWSHeaderRegistry
:annotation:
+
+Examples
+--------
+
+Sign a JWS token::
+ >>> from jwcrypto import jwk, jws
+ >>> from jwcrypto.common import json_encode
+ >>> key = jwk.JWK.generate(kty='oct', size=256)
+ >>> payload = "My Integrity protected message"
+ >>> jwstoken = jws.JWS(payload.encode('utf-8'))
+ >>> jwstoken.add_signature(key, None,
+ json_encode({"alg": "HS256"}),
+ json_encode({"kid": key.thumbprint()}))
+ >>> sig = jwstoken.serialize()
+
+Verify a JWS token::
+ >>> jwstoken = jws.JWS()
+ >>> jwstoken.deserialize(sig)
+ >>> jwstoken.verify(key)
+ >>> payload = jwstoken.payload
=====================================
jwcrypto/common.py
=====================================
@@ -16,12 +16,12 @@ def base64url_encode(payload):
def base64url_decode(payload):
- l = len(payload) % 4
- if l == 2:
+ size = len(payload) % 4
+ if size == 2:
payload += '=='
- elif l == 3:
+ elif size == 3:
payload += '='
- elif l != 0:
+ elif size != 0:
raise ValueError('Invalid base64 string')
return urlsafe_b64decode(payload.encode('utf-8'))
@@ -40,9 +40,67 @@ def json_decode(string):
return json.loads(string)
-class InvalidJWAAlgorithm(Exception):
+class JWException(Exception):
+ pass
+
+
+class InvalidJWAAlgorithm(JWException):
def __init__(self, message=None):
- msg = 'Invalid JWS Algorithm name'
+ msg = 'Invalid JWA Algorithm name'
if message:
msg += ' (%s)' % message
super(InvalidJWAAlgorithm, self).__init__(msg)
+
+
+class InvalidCEKeyLength(JWException):
+ """Invalid CEK Key Length.
+
+ This exception is raised when a Content Encryption Key does not match
+ the required lenght.
+ """
+
+ def __init__(self, expected, obtained):
+ msg = 'Expected key of length %d bits, got %d' % (expected, obtained)
+ super(InvalidCEKeyLength, self).__init__(msg)
+
+
+class InvalidJWEOperation(JWException):
+ """Invalid JWS Object.
+
+ This exception is raised when a requested operation cannot
+ be execute due to unsatisfied conditions.
+ """
+
+ 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(InvalidJWEOperation, self).__init__(msg)
+
+
+class InvalidJWEKeyType(JWException):
+ """Invalid JWE Key Type.
+
+ This exception is raised when the provided JWK Key does not match
+ the type required by the sepcified algorithm.
+ """
+
+ def __init__(self, expected, obtained):
+ msg = 'Expected key type %s, got %s' % (expected, obtained)
+ super(InvalidJWEKeyType, self).__init__(msg)
+
+
+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.
+ """
+
+ def __init__(self, expected, obtained):
+ msg = 'Expected key of lenght %d, got %d' % (expected, obtained)
+ super(InvalidJWEKeyLength, self).__init__(msg)
=====================================
jwcrypto/jwa.py
=====================================
@@ -0,0 +1,1072 @@
+# Copyright (C) 2016 JWCrypto Project Contributors - see LICENSE file
+
+import abc
+import os
+import struct
+from binascii import hexlify, unhexlify
+
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import constant_time, hashes, hmac
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.asymmetric import utils as ec_utils
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives.kdf.concatkdf import ConcatKDFHash
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+from cryptography.hazmat.primitives.keywrap import aes_key_unwrap, aes_key_wrap
+from cryptography.hazmat.primitives.padding import PKCS7
+
+import six
+
+from jwcrypto.common import InvalidCEKeyLength
+from jwcrypto.common import InvalidJWAAlgorithm
+from jwcrypto.common import InvalidJWEKeyLength
+from jwcrypto.common import InvalidJWEKeyType
+from jwcrypto.common import InvalidJWEOperation
+from jwcrypto.common import base64url_decode, base64url_encode
+from jwcrypto.common import json_decode
+from jwcrypto.jwk import JWK
+
+# Implements RFC 7518 - JSON Web Algorithms (JWA)
+
+
+ at six.add_metaclass(abc.ABCMeta)
+class JWAAlgorithm(object):
+
+ @abc.abstractproperty
+ def name(self):
+ """The algorithm Name"""
+ pass
+
+ @abc.abstractproperty
+ def description(self):
+ """A short description"""
+ pass
+
+ @abc.abstractproperty
+ def keysize(self):
+ """The actual/recommended/minimum key size"""
+ pass
+
+ @abc.abstractproperty
+ def algorithm_usage_location(self):
+ """One of 'alg', 'enc' or 'JWK'"""
+ pass
+
+ @abc.abstractproperty
+ def algorithm_use(self):
+ """One of 'sig', 'kex', 'enc'"""
+ pass
+
+
+def _bitsize(x):
+ return len(x) * 8
+
+
+def _inbytes(x):
+ return x // 8
+
+
+def _randombits(x):
+ if x % 8 != 0:
+ raise ValueError("lenght must be a multiple of 8")
+ return os.urandom(_inbytes(x))
+
+
+# Note: the number of bits should be a multiple of 16
+def _encode_int(n, bits):
+ e = '{:x}'.format(n)
+ ilen = ((bits + 7) // 8) * 2 # number of bytes rounded up times 2 bytes
+ return unhexlify(e.rjust(ilen, '0')[:ilen])
+
+
+def _decode_int(n):
+ return int(hexlify(n), 16)
+
+
+class _RawJWS(object):
+
+ def sign(self, key, payload):
+ raise NotImplementedError
+
+ def verify(self, key, payload, signature):
+ raise NotImplementedError
+
+
+class _RawHMAC(_RawJWS):
+
+ def __init__(self, hashfn):
+ self.backend = default_backend()
+ self.hashfn = hashfn
+
+ def _hmac_setup(self, key, payload):
+ h = hmac.HMAC(key, self.hashfn, backend=self.backend)
+ h.update(payload)
+ return h
+
+ def sign(self, key, payload):
+ skey = base64url_decode(key.get_op_key('sign'))
+ h = self._hmac_setup(skey, payload)
+ return h.finalize()
+
+ def verify(self, key, payload, signature):
+ vkey = base64url_decode(key.get_op_key('verify'))
+ h = self._hmac_setup(vkey, payload)
+ h.verify(signature)
+
+
+class _RawRSA(_RawJWS):
+ def __init__(self, padfn, hashfn):
+ self.padfn = padfn
+ self.hashfn = hashfn
+
+ def sign(self, key, payload):
+ skey = key.get_op_key('sign')
+ return skey.sign(payload, self.padfn, self.hashfn)
+
+ def verify(self, key, payload, signature):
+ pkey = key.get_op_key('verify')
+ pkey.verify(signature, payload, self.padfn, self.hashfn)
+
+
+class _RawEC(_RawJWS):
+ def __init__(self, curve, hashfn):
+ self._curve = curve
+ self.hashfn = hashfn
+
+ @property
+ def curve(self):
+ return self._curve
+
+ def sign(self, key, payload):
+ skey = key.get_op_key('sign', self._curve)
+ signature = skey.sign(payload, ec.ECDSA(self.hashfn))
+ r, s = ec_utils.decode_dss_signature(signature)
+ size = key.get_curve(self._curve).key_size
+ return _encode_int(r, size) + _encode_int(s, size)
+
+ def verify(self, key, payload, signature):
+ pkey = key.get_op_key('verify', self._curve)
+ r = signature[:len(signature) // 2]
+ s = signature[len(signature) // 2:]
+ enc_signature = ec_utils.encode_dss_signature(
+ int(hexlify(r), 16), int(hexlify(s), 16))
+ pkey.verify(enc_signature, payload, ec.ECDSA(self.hashfn))
+
+
+class _RawNone(_RawJWS):
+
+ def sign(self, key, payload):
+ return ''
+
+ def verify(self, key, payload, signature):
+ raise InvalidSignature('The "none" signature cannot be verified')
+
+
+class _HS256(_RawHMAC, JWAAlgorithm):
+
+ name = "HS256"
+ description = "HMAC using SHA-256"
+ keysize = 256
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_HS256, self).__init__(hashes.SHA256())
+
+
+class _HS384(_RawHMAC, JWAAlgorithm):
+
+ name = "HS384"
+ description = "HMAC using SHA-384"
+ keysize = 384
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_HS384, self).__init__(hashes.SHA384())
+
+
+class _HS512(_RawHMAC, JWAAlgorithm):
+
+ name = "HS512"
+ description = "HMAC using SHA-512"
+ keysize = 512
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_HS512, self).__init__(hashes.SHA512())
+
+
+class _RS256(_RawRSA, JWAAlgorithm):
+
+ name = "RS256"
+ description = "RSASSA-PKCS1-v1_5 using SHA-256"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_RS256, self).__init__(padding.PKCS1v15(), hashes.SHA256())
+
+
+class _RS384(_RawRSA, JWAAlgorithm):
+
+ name = "RS384"
+ description = "RSASSA-PKCS1-v1_5 using SHA-384"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_RS384, self).__init__(padding.PKCS1v15(), hashes.SHA384())
+
+
+class _RS512(_RawRSA, JWAAlgorithm):
+
+ name = "RS512"
+ description = "RSASSA-PKCS1-v1_5 using SHA-512"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_RS512, self).__init__(padding.PKCS1v15(), hashes.SHA512())
+
+
+class _ES256(_RawEC, JWAAlgorithm):
+
+ name = "ES256"
+ description = "ECDSA using P-256 and SHA-256"
+ keysize = 256
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_ES256, self).__init__('P-256', hashes.SHA256())
+
+
+class _ES384(_RawEC, JWAAlgorithm):
+
+ name = "ES384"
+ description = "ECDSA using P-384 and SHA-384"
+ keysize = 384
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_ES384, self).__init__('P-384', hashes.SHA384())
+
+
+class _ES512(_RawEC, JWAAlgorithm):
+
+ name = "ES512"
+ description = "ECDSA using P-521 and SHA-512"
+ keysize = 512
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ super(_ES512, self).__init__('P-521', hashes.SHA512())
+
+
+class _PS256(_RawRSA, JWAAlgorithm):
+
+ name = "PS256"
+ description = "RSASSA-PSS using SHA-256 and MGF1 with SHA-256"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ padfn = padding.PSS(padding.MGF1(hashes.SHA256()),
+ hashes.SHA256.digest_size)
+ super(_PS256, self).__init__(padfn, hashes.SHA256())
+
+
+class _PS384(_RawRSA, JWAAlgorithm):
+
+ name = "PS384"
+ description = "RSASSA-PSS using SHA-384 and MGF1 with SHA-384"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ padfn = padding.PSS(padding.MGF1(hashes.SHA384()),
+ hashes.SHA384.digest_size)
+ super(_PS384, self).__init__(padfn, hashes.SHA384())
+
+
+class _PS512(_RawRSA, JWAAlgorithm):
+
+ name = "PS512"
+ description = "RSASSA-PSS using SHA-512 and MGF1 with SHA-512"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+ def __init__(self):
+ padfn = padding.PSS(padding.MGF1(hashes.SHA512()),
+ hashes.SHA512.digest_size)
+ super(_PS512, self).__init__(padfn, hashes.SHA512())
+
+
+class _None(_RawNone, JWAAlgorithm):
+
+ name = "none"
+ description = "No digital signature or MAC performed"
+ keysize = 0
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'sig'
+
+
+class _RawKeyMgmt(object):
+
+ def wrap(self, key, bitsize, cek, headers):
+ raise NotImplementedError
+
+ def unwrap(self, key, bitsize, ek, headers):
+ raise NotImplementedError
+
+
+class _RSA(_RawKeyMgmt):
+
+ def __init__(self, padfn):
+ self.padfn = padfn
+
+ def _check_key(self, key):
+ if not isinstance(key, JWK):
+ raise ValueError('key is not a JWK object')
+ if key.key_type != 'RSA':
+ raise InvalidJWEKeyType('RSA', key.key_type)
+
+ # FIXME: get key size and insure > 2048 bits
+ def wrap(self, key, bitsize, cek, headers):
+ self._check_key(key)
+ if not cek:
+ cek = _randombits(bitsize)
+ rk = key.get_op_key('wrapKey')
+ ek = rk.encrypt(cek, self.padfn)
+ return {'cek': cek, 'ek': ek}
+
+ def unwrap(self, key, bitsize, ek, headers):
+ self._check_key(key)
+ rk = key.get_op_key('decrypt')
+ cek = rk.decrypt(ek, self.padfn)
+ if _bitsize(cek) != bitsize:
+ raise InvalidJWEKeyLength(bitsize, _bitsize(cek))
+ return cek
+
+
+class _Rsa15(_RSA, JWAAlgorithm):
+
+ name = 'RSA1_5'
+ description = "RSAES-PKCS1-v1_5"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+ def __init__(self):
+ super(_Rsa15, self).__init__(padding.PKCS1v15())
+
+ def unwrap(self, key, bitsize, ek, headers):
+ self._check_key(key)
+ # Address MMA attack by implementing RFC 3218 - 2.3.2. Random Filling
+ # provides a random cek that will cause the decryption engine to
+ # run to the end, but will fail decryption later.
+
+ # always generate a random cek so we spend roughly the
+ # same time as in the exception side of the branch
+ cek = _randombits(bitsize)
+ try:
+ cek = super(_Rsa15, self).unwrap(key, bitsize, ek, headers)
+ # always raise so we always run through the exception handling
+ # code in all cases
+ raise Exception('Dummy')
+ except Exception: # pylint: disable=broad-except
+ return cek
+
+
+class _RsaOaep(_RSA, JWAAlgorithm):
+
+ name = 'RSA-OAEP'
+ description = "RSAES OAEP using default parameters"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+ def __init__(self):
+ super(_RsaOaep, self).__init__(
+ padding.OAEP(padding.MGF1(hashes.SHA1()),
+ hashes.SHA1(), None))
+
+
+class _RsaOaep256(_RSA, JWAAlgorithm): # noqa: ignore=N801
+
+ name = 'RSA-OAEP-256'
+ description = "RSAES OAEP using SHA-256 and MGF1 with SHA-256"
+ keysize = 2048
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+ def __init__(self):
+ super(_RsaOaep256, self).__init__(
+ padding.OAEP(padding.MGF1(hashes.SHA256()),
+ hashes.SHA256(), None))
+
+
+class _AesKw(_RawKeyMgmt):
+
+ keysize = None
+
+ def __init__(self):
+ self.backend = default_backend()
+
+ def _get_key(self, key, op):
+ if not isinstance(key, JWK):
+ raise ValueError('key is not a JWK object')
+ if key.key_type != 'oct':
+ raise InvalidJWEKeyType('oct', key.key_type)
+ rk = base64url_decode(key.get_op_key(op))
+ if _bitsize(rk) != self.keysize:
+ raise InvalidJWEKeyLength(self.keysize, _bitsize(rk))
+ return rk
+
+ def wrap(self, key, bitsize, cek, headers):
+ rk = self._get_key(key, 'encrypt')
+ if not cek:
+ cek = _randombits(bitsize)
+
+ ek = aes_key_wrap(rk, cek, default_backend())
+
+ return {'cek': cek, 'ek': ek}
+
+ def unwrap(self, key, bitsize, ek, headers):
+ rk = self._get_key(key, 'decrypt')
+
+ cek = aes_key_unwrap(rk, ek, default_backend())
+ if _bitsize(cek) != bitsize:
+ raise InvalidJWEKeyLength(bitsize, _bitsize(cek))
+ return cek
+
+
+class _A128KW(_AesKw, JWAAlgorithm):
+
+ name = 'A128KW'
+ description = "AES Key Wrap using 128-bit key"
+ keysize = 128
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _A192KW(_AesKw, JWAAlgorithm):
+
+ name = 'A192KW'
+ description = "AES Key Wrap using 192-bit key"
+ keysize = 192
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _A256KW(_AesKw, JWAAlgorithm):
+
+ name = 'A256KW'
+ description = "AES Key Wrap using 256-bit key"
+ keysize = 256
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _AesGcmKw(_RawKeyMgmt):
+
+ keysize = None
+
+ def __init__(self):
+ self.backend = default_backend()
+
+ def _get_key(self, key, op):
+ if not isinstance(key, JWK):
+ raise ValueError('key is not a JWK object')
+ if key.key_type != 'oct':
+ raise InvalidJWEKeyType('oct', key.key_type)
+ rk = base64url_decode(key.get_op_key(op))
+ if _bitsize(rk) != self.keysize:
+ raise InvalidJWEKeyLength(self.keysize, _bitsize(rk))
+ return rk
+
+ def wrap(self, key, bitsize, cek, headers):
+ rk = self._get_key(key, 'encrypt')
+ if not cek:
+ cek = _randombits(bitsize)
+
+ iv = _randombits(96)
+ cipher = Cipher(algorithms.AES(rk), modes.GCM(iv),
+ backend=self.backend)
+ encryptor = cipher.encryptor()
+ ek = encryptor.update(cek) + encryptor.finalize()
+
+ tag = encryptor.tag
+ return {'cek': cek, 'ek': ek,
+ 'header': {'iv': base64url_encode(iv),
+ 'tag': base64url_encode(tag)}}
+
+ def unwrap(self, key, bitsize, ek, headers):
+ rk = self._get_key(key, 'decrypt')
+
+ if 'iv' not in headers:
+ raise ValueError('Invalid Header, missing "iv" parameter')
+ iv = base64url_decode(headers['iv'])
+ if 'tag' not in headers:
+ raise ValueError('Invalid Header, missing "tag" parameter')
+ tag = base64url_decode(headers['tag'])
+
+ cipher = Cipher(algorithms.AES(rk), modes.GCM(iv, tag),
+ backend=self.backend)
+ decryptor = cipher.decryptor()
+ cek = decryptor.update(ek) + decryptor.finalize()
+ if _bitsize(cek) != bitsize:
+ raise InvalidJWEKeyLength(bitsize, _bitsize(cek))
+ return cek
+
+
+class _A128GcmKw(_AesGcmKw, JWAAlgorithm):
+
+ name = 'A128GCMKW'
+ description = "Key wrapping with AES GCM using 128-bit key"
+ keysize = 128
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _A192GcmKw(_AesGcmKw, JWAAlgorithm):
+
+ name = 'A192GCMKW'
+ description = "Key wrapping with AES GCM using 192-bit key"
+ keysize = 192
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _A256GcmKw(_AesGcmKw, JWAAlgorithm):
+
+ name = 'A256GCMKW'
+ description = "Key wrapping with AES GCM using 256-bit key"
+ keysize = 256
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _Pbes2HsAesKw(_RawKeyMgmt):
+
+ name = None
+ keysize = None
+ hashsize = None
+
+ def __init__(self):
+ self.backend = default_backend()
+ self.aeskwmap = {128: _A128KW, 192: _A192KW, 256: _A256KW}
+
+ def _get_key(self, alg, key, p2s, p2c):
+ if isinstance(key, bytes):
+ plain = key
+ else:
+ plain = key.encode('utf8')
+ salt = bytes(self.name.encode('utf8')) + b'\x00' + p2s
+
+ if self.hashsize == 256:
+ hashalg = hashes.SHA256()
+ elif self.hashsize == 384:
+ hashalg = hashes.SHA384()
+ elif self.hashsize == 512:
+ hashalg = hashes.SHA512()
+ else:
+ raise ValueError('Unknown Hash Size')
+
+ kdf = PBKDF2HMAC(algorithm=hashalg, length=_inbytes(self.keysize),
+ salt=salt, iterations=p2c, backend=self.backend)
+ rk = kdf.derive(plain)
+ if _bitsize(rk) != self.keysize:
+ raise InvalidJWEKeyLength(self.keysize, len(rk))
+ return JWK(kty="oct", use="enc", k=base64url_encode(rk))
+
+ def wrap(self, key, bitsize, cek, headers):
+ p2s = _randombits(128)
+ p2c = 8192
+ kek = self._get_key(headers['alg'], key, p2s, p2c)
+
+ aeskw = self.aeskwmap[self.keysize]()
+ ret = aeskw.wrap(kek, bitsize, cek, headers)
+ ret['header'] = {'p2s': base64url_encode(p2s), 'p2c': p2c}
+ return ret
+
+ def unwrap(self, key, bitsize, ek, headers):
+ if 'p2s' not in headers:
+ raise ValueError('Invalid Header, missing "p2s" parameter')
+ if 'p2c' not in headers:
+ raise ValueError('Invalid Header, missing "p2c" parameter')
+ p2s = base64url_decode(headers['p2s'])
+ p2c = headers['p2c']
+ kek = self._get_key(headers['alg'], key, p2s, p2c)
+
+ aeskw = self.aeskwmap[self.keysize]()
+ return aeskw.unwrap(kek, bitsize, ek, headers)
+
+
+class _Pbes2Hs256A128Kw(_Pbes2HsAesKw, JWAAlgorithm):
+
+ name = 'PBES2-HS256+A128KW'
+ description = 'PBES2 with HMAC SHA-256 and "A128KW" wrapping'
+ keysize = 128
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+ hashsize = 256
+
+
+class _Pbes2Hs384A192Kw(_Pbes2HsAesKw, JWAAlgorithm):
+
+ name = 'PBES2-HS384+A192KW'
+ description = 'PBES2 with HMAC SHA-384 and "A192KW" wrapping'
+ keysize = 192
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+ hashsize = 384
+
+
+class _Pbes2Hs512A256Kw(_Pbes2HsAesKw, JWAAlgorithm):
+
+ name = 'PBES2-HS512+A256KW'
+ description = 'PBES2 with HMAC SHA-512 and "A256KW" wrapping'
+ keysize = 256
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+ hashsize = 512
+
+
+class _Direct(_RawKeyMgmt, JWAAlgorithm):
+
+ name = 'dir'
+ description = "Direct use of a shared symmetric key"
+ keysize = 128
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+ def _check_key(self, key):
+ if not isinstance(key, JWK):
+ raise ValueError('key is not a JWK object')
+ if key.key_type != 'oct':
+ raise InvalidJWEKeyType('oct', key.key_type)
+
+ def wrap(self, key, bitsize, cek, headers):
+ self._check_key(key)
+ if cek:
+ return (cek, None)
+ k = base64url_decode(key.get_op_key('encrypt'))
+ if _bitsize(k) != bitsize:
+ raise InvalidCEKeyLength(bitsize, _bitsize(k))
+ return {'cek': k}
+
+ def unwrap(self, key, bitsize, ek, headers):
+ self._check_key(key)
+ if ek != b'':
+ raise ValueError('Invalid Encryption Key.')
+ cek = base64url_decode(key.get_op_key('decrypt'))
+ if _bitsize(cek) != bitsize:
+ raise InvalidJWEKeyLength(bitsize, _bitsize(cek))
+ return cek
+
+
+class _EcdhEs(_RawKeyMgmt, JWAAlgorithm):
+
+ name = 'ECDH-ES'
+ description = "ECDH-ES using Concat KDF"
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+ keysize = None
+
+ def __init__(self):
+ self.backend = default_backend()
+ self.aeskwmap = {128: _A128KW, 192: _A192KW, 256: _A256KW}
+
+ 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)
+
+ def _derive(self, privkey, pubkey, alg, bitsize, headers):
+ # OtherInfo is defined in NIST SP 56A 5.8.1.2.1
+
+ # AlgorithmID
+ otherinfo = struct.pack('>I', len(alg))
+ otherinfo += bytes(alg.encode('utf8'))
+
+ # PartyUInfo
+ apu = base64url_decode(headers['apu']) if 'apu' in headers else b''
+ otherinfo += struct.pack('>I', len(apu))
+ otherinfo += apu
+
+ # PartyVInfo
+ apv = base64url_decode(headers['apv']) if 'apv' in headers else b''
+ otherinfo += struct.pack('>I', len(apv))
+ otherinfo += apv
+
+ # SuppPubInfo
+ otherinfo += struct.pack('>I', bitsize)
+
+ # no SuppPrivInfo
+
+ shared_key = privkey.exchange(ec.ECDH(), pubkey)
+ ckdf = ConcatKDFHash(algorithm=hashes.SHA256(),
+ length=_inbytes(bitsize),
+ otherinfo=otherinfo,
+ backend=self.backend)
+ return ckdf.derive(shared_key)
+
+ def wrap(self, key, bitsize, cek, headers):
+ self._check_key(key)
+ dk_size = self.keysize
+ if self.keysize is None:
+ if cek is not None:
+ raise InvalidJWEOperation('ECDH-ES cannot use an existing CEK')
+ alg = headers['enc']
+ dk_size = bitsize
+ else:
+ alg = headers['alg']
+
+ epk = JWK.generate(kty=key.key_type, crv=key.key_curve)
+ dk = self._derive(epk.get_op_key('unwrapKey'),
+ key.get_op_key('wrapKey'),
+ alg, dk_size, headers)
+
+ if self.keysize is None:
+ ret = {'cek': dk}
+ else:
+ aeskw = self.aeskwmap[self.keysize]()
+ kek = JWK(kty="oct", use="enc", k=base64url_encode(dk))
+ ret = aeskw.wrap(kek, bitsize, cek, headers)
+
+ ret['header'] = {'epk': json_decode(epk.export_public())}
+ return ret
+
+ def unwrap(self, key, bitsize, ek, headers):
+ if 'epk' not in headers:
+ raise ValueError('Invalid Header, missing "epk" parameter')
+ self._check_key(key)
+ dk_size = self.keysize
+ if self.keysize is None:
+ alg = headers['enc']
+ dk_size = bitsize
+ else:
+ alg = headers['alg']
+
+ epk = JWK(**headers['epk'])
+ dk = self._derive(key.get_op_key('unwrapKey'),
+ epk.get_op_key('wrapKey'),
+ alg, dk_size, headers)
+ if self.keysize is None:
+ return dk
+ else:
+ aeskw = self.aeskwmap[self.keysize]()
+ kek = JWK(kty="oct", use="enc", k=base64url_encode(dk))
+ cek = aeskw.unwrap(kek, bitsize, ek, headers)
+ return cek
+
+
+class _EcdhEsAes128Kw(_EcdhEs):
+
+ name = 'ECDH-ES+A128KW'
+ description = 'ECDH-ES using Concat KDF and "A128KW" wrapping'
+ keysize = 128
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _EcdhEsAes192Kw(_EcdhEs):
+
+ name = 'ECDH-ES+A192KW'
+ description = 'ECDH-ES using Concat KDF and "A192KW" wrapping'
+ keysize = 192
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _EcdhEsAes256Kw(_EcdhEs):
+
+ name = 'ECDH-ES+A256KW'
+ description = 'ECDH-ES using Concat KDF and "A256KW" wrapping'
+ keysize = 256
+ algorithm_usage_location = 'alg'
+ algorithm_use = 'kex'
+
+
+class _RawJWE(object):
+
+ def encrypt(self, k, a, m):
+ raise NotImplementedError
+
+ def decrypt(self, k, a, iv, e, t):
+ raise NotImplementedError
+
+
+class _AesCbcHmacSha2(_RawJWE):
+
+ keysize = None
+
+ def __init__(self, hashfn):
+ self.backend = default_backend()
+ self.hashfn = hashfn
+ self.blocksize = algorithms.AES.block_size
+ self.wrap_key_size = self.keysize * 2
+
+ def _mac(self, k, a, iv, e):
+ al = _encode_int(_bitsize(a), 64)
+ h = hmac.HMAC(k, self.hashfn, backend=self.backend)
+ h.update(a)
+ h.update(iv)
+ h.update(e)
+ h.update(al)
+ m = h.finalize()
+ return m[:_inbytes(self.keysize)]
+
+ # RFC 7518 - 5.2.2
+ def encrypt(self, k, a, m):
+ """ Encrypt according to the selected encryption and hashing
+ functions.
+
+ :param k: Encryption key (optional)
+ :param a: Additional Authentication Data
+ :param m: Plaintext
+
+ Returns a dictionary with the computed data.
+ """
+ hkey = k[:_inbytes(self.keysize)]
+ ekey = k[_inbytes(self.keysize):]
+
+ # encrypt
+ iv = _randombits(self.blocksize)
+ cipher = Cipher(algorithms.AES(ekey), modes.CBC(iv),
+ backend=self.backend)
+ encryptor = cipher.encryptor()
+ padder = PKCS7(self.blocksize).padder()
+ padded_data = padder.update(m) + padder.finalize()
+ e = encryptor.update(padded_data) + encryptor.finalize()
+
+ # mac
+ t = self._mac(hkey, a, iv, e)
+
+ return (iv, e, t)
+
+ def decrypt(self, k, a, iv, e, t):
+ """ Decrypt according to the selected encryption and hashing
+ functions.
+ :param k: Encryption key (optional)
+ :param a: Additional Authenticated Data
+ :param iv: Initialization Vector
+ :param e: Ciphertext
+ :param t: Authentication Tag
+
+ Returns plaintext or raises an error
+ """
+ hkey = k[:_inbytes(self.keysize)]
+ dkey = k[_inbytes(self.keysize):]
+
+ # verify mac
+ if not constant_time.bytes_eq(t, self._mac(hkey, a, iv, e)):
+ raise InvalidSignature('Failed to verify MAC')
+
+ # decrypt
+ cipher = Cipher(algorithms.AES(dkey), modes.CBC(iv),
+ backend=self.backend)
+ decryptor = cipher.decryptor()
+ d = decryptor.update(e) + decryptor.finalize()
+ unpadder = PKCS7(self.blocksize).unpadder()
+ return unpadder.update(d) + unpadder.finalize()
+
+
+class _A128CbcHs256(_AesCbcHmacSha2, JWAAlgorithm):
+
+ name = 'A128CBC-HS256'
+ description = "AES_128_CBC_HMAC_SHA_256 authenticated"
+ keysize = 128
+ algorithm_usage_location = 'enc'
+ algorithm_use = 'enc'
+
+ def __init__(self):
+ super(_A128CbcHs256, self).__init__(hashes.SHA256())
+
+
+class _A192CbcHs384(_AesCbcHmacSha2, JWAAlgorithm):
+
+ name = 'A192CBC-HS384'
+ description = "AES_192_CBC_HMAC_SHA_384 authenticated"
+ keysize = 192
+ algorithm_usage_location = 'enc'
+ algorithm_use = 'enc'
+
+ def __init__(self):
+ super(_A192CbcHs384, self).__init__(hashes.SHA384())
+
+
+class _A256CbcHs512(_AesCbcHmacSha2, JWAAlgorithm):
+
+ name = 'A256CBC-HS512'
+ description = "AES_256_CBC_HMAC_SHA_512 authenticated"
+ keysize = 256
+ algorithm_usage_location = 'enc'
+ algorithm_use = 'enc'
+
+ def __init__(self):
+ super(_A256CbcHs512, self).__init__(hashes.SHA512())
+
+
+class _AesGcm(_RawJWE):
+
+ keysize = None
+
+ def __init__(self):
+ self.backend = default_backend()
+ self.wrap_key_size = self.keysize
+
+ # RFC 7518 - 5.3
+ def encrypt(self, k, a, m):
+ """ Encrypt accoriding to the selected encryption and hashing
+ functions.
+
+ :param k: Encryption key (optional)
+ :param a: Additional Authentication Data
+ :param m: Plaintext
+
+ Returns a dictionary with the computed data.
+ """
+ iv = _randombits(96)
+ cipher = Cipher(algorithms.AES(k), modes.GCM(iv),
+ backend=self.backend)
+ encryptor = cipher.encryptor()
+ encryptor.authenticate_additional_data(a)
+ e = encryptor.update(m) + encryptor.finalize()
+
+ return (iv, e, encryptor.tag)
+
+ def decrypt(self, k, a, iv, e, t):
+ """ Decrypt accoriding to the selected encryption and hashing
+ functions.
+ :param k: Encryption key (optional)
+ :param a: Additional Authenticated Data
+ :param iv: Initialization Vector
+ :param e: Ciphertext
+ :param t: Authentication Tag
+
+ Returns plaintext or raises an error
+ """
+ cipher = Cipher(algorithms.AES(k), modes.GCM(iv, t),
+ backend=self.backend)
+ decryptor = cipher.decryptor()
+ decryptor.authenticate_additional_data(a)
+ return decryptor.update(e) + decryptor.finalize()
+
+
+class _A128Gcm(_AesGcm, JWAAlgorithm):
+
+ name = 'A128GCM'
+ description = "AES GCM using 128-bit key"
+ keysize = 128
+ algorithm_usage_location = 'enc'
+ algorithm_use = 'enc'
+
+
+class _A192Gcm(_AesGcm, JWAAlgorithm):
+
+ name = 'A192GCM'
+ description = "AES GCM using 192-bit key"
+ keysize = 192
+ algorithm_usage_location = 'enc'
+ algorithm_use = 'enc'
+
+
+class _A256Gcm(_AesGcm, JWAAlgorithm):
+
+ name = 'A256GCM'
+ description = "AES GCM using 256-bit key"
+ keysize = 256
+ algorithm_usage_location = 'enc'
+ algorithm_use = 'enc'
+
+
+class JWA(object):
+ """JWA Signing Algorithms.
+
+ This class provides access to all JWA algorithms.
+ """
+
+ algorithms_registry = {
+ 'HS256': _HS256,
+ 'HS384': _HS384,
+ 'HS512': _HS512,
+ 'RS256': _RS256,
+ 'RS384': _RS384,
+ 'RS512': _RS512,
+ 'ES256': _ES256,
+ 'ES384': _ES384,
+ 'ES512': _ES512,
+ 'PS256': _PS256,
+ 'PS384': _PS384,
+ 'PS512': _PS512,
+ 'none': _None,
+ 'RSA1_5': _Rsa15,
+ 'RSA-OAEP': _RsaOaep,
+ 'RSA-OAEP-256': _RsaOaep256,
+ 'A128KW': _A128KW,
+ 'A192KW': _A192KW,
+ 'A256KW': _A256KW,
+ 'dir': _Direct,
+ 'ECDH-ES': _EcdhEs,
+ 'ECDH-ES+A128KW': _EcdhEsAes128Kw,
+ 'ECDH-ES+A192KW': _EcdhEsAes192Kw,
+ 'ECDH-ES+A256KW': _EcdhEsAes256Kw,
+ 'A128GCMKW': _A128GcmKw,
+ 'A192GCMKW': _A192GcmKw,
+ 'A256GCMKW': _A256GcmKw,
+ 'PBES2-HS256+A128KW': _Pbes2Hs256A128Kw,
+ 'PBES2-HS384+A192KW': _Pbes2Hs384A192Kw,
+ 'PBES2-HS512+A256KW': _Pbes2Hs512A256Kw,
+ 'A128CBC-HS256': _A128CbcHs256,
+ 'A192CBC-HS384': _A192CbcHs384,
+ 'A256CBC-HS512': _A256CbcHs512,
+ 'A128GCM': _A128Gcm,
+ 'A192GCM': _A192Gcm,
+ 'A256GCM': _A256Gcm
+ }
+
+ @classmethod
+ def instantiate_alg(cls, name, use=None):
+ alg = cls.algorithms_registry[name]
+ if use is not None and alg.algorithm_use != use:
+ raise KeyError
+ return alg()
+
+ @classmethod
+ def signing_alg(cls, name):
+ try:
+ return cls.instantiate_alg(name, use='sig')
+ except KeyError:
+ raise InvalidJWAAlgorithm(
+ '%s is not a valid Signign algorithm name' % name)
+
+ @classmethod
+ def keymgmt_alg(cls, name):
+ try:
+ return cls.instantiate_alg(name, use='kex')
+ except KeyError:
+ raise InvalidJWAAlgorithm(
+ '%s is not a valid Key Management algorithm name' % name)
+
+ @classmethod
+ def encryption_alg(cls, name):
+ try:
+ return cls.instantiate_alg(name, use='enc')
+ except KeyError:
+ raise InvalidJWAAlgorithm(
+ '%s is not a valid Encryption algorithm name' % name)
=====================================
jwcrypto/jwe.py
=====================================
@@ -1,24 +1,12 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
-import os
-import struct
import zlib
-from binascii import hexlify, unhexlify
-
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import constant_time, hashes, hmac
-from cryptography.hazmat.primitives.asymmetric import ec
-from cryptography.hazmat.primitives.asymmetric import padding
-from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
-from cryptography.hazmat.primitives.kdf.concatkdf import ConcatKDFHash
-from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
-from cryptography.hazmat.primitives.padding import PKCS7
-
-from jwcrypto.common import InvalidJWAAlgorithm
+from jwcrypto import common
+from jwcrypto.common import JWException
from jwcrypto.common import base64url_decode, base64url_encode
from jwcrypto.common import json_decode, json_encode
-from jwcrypto.jwk import JWK
+from jwcrypto.jwa import JWA
# RFC 7516 - 4.1
@@ -53,23 +41,7 @@ default_allowed_algs = [
"""Default allowed algorithms"""
-# Note: l is the number of bits, which should be a multiple of 16
-def _encode_int(n, l):
- e = hex(n).rstrip("L").lstrip("0x")
- elen = len(e)
- ilen = ((l + 7) // 8) * 2 # number of bytes rounded up times 2 chars/bytes
- if elen > ilen:
- e = e[:ilen]
- else:
- e = '0' * (ilen - elen) + e # pad as necessary
- return unhexlify(e)
-
-
-def _decode_int(n):
- return int(hexlify(n), 16)
-
-
-class InvalidJWEData(Exception):
+class InvalidJWEData(JWException):
"""Invalid JWE Object.
This exception is raised when the JWE Object is invalid and/or
@@ -87,726 +59,11 @@ class InvalidJWEData(Exception):
super(InvalidJWEData, self).__init__(msg)
-class InvalidCEKeyLength(Exception):
- """Invalid CEK Key Length.
-
- This exception is raised when a Content Encryption Key does not match
- the required lenght.
- """
-
- def __init__(self, expected, obtained):
- msg = 'Expected key of length %d, got %d' % (expected, obtained)
- super(InvalidCEKeyLength, self).__init__(msg)
-
-
-class InvalidJWEOperation(Exception):
- """Invalid JWS Object.
-
- This exception is raised when a requested operation cannot
- be execute due to unsatisfied conditions.
- """
-
- 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(InvalidJWEOperation, self).__init__(msg)
-
-
-class InvalidJWEKeyType(Exception):
- """Invalid JWE Key Type.
-
- This exception is raised when the provided JWK Key does not match
- the type required by the sepcified algorithm.
- """
-
- def __init__(self, expected, obtained):
- msg = 'Expected key type %s, got %s' % (expected, obtained)
- super(InvalidJWEKeyType, self).__init__(msg)
-
-
-class InvalidJWEKeyLength(Exception):
- """Invalid JWE Key Length.
-
- This exception is raised when the provided JWK Key does not match
- the lenght required by the sepcified algorithm.
- """
-
- def __init__(self, expected, obtained):
- msg = 'Expected key of lenght %d, got %d' % (expected, obtained)
- super(InvalidJWEKeyLength, self).__init__(msg)
-
-
-class _RawKeyMgmt(object):
-
- def wrap(self, key, keylen, cek, headers):
- raise NotImplementedError
-
- def unwrap(self, key, keylen, ek, headers):
- raise NotImplementedError
-
-
-class _RSA(_RawKeyMgmt):
-
- def __init__(self, padfn):
- self.padfn = padfn
-
- def _check_key(self, key):
- if not isinstance(key, JWK):
- raise ValueError('key is not a JWK object')
- if key.key_type != 'RSA':
- raise InvalidJWEKeyType('RSA', key.key_type)
-
- # FIXME: get key size and insure > 2048 bits
- def wrap(self, key, keylen, cek, headers):
- self._check_key(key)
- if not cek:
- cek = os.urandom(keylen)
- rk = key.get_op_key('wrapKey')
- ek = rk.encrypt(cek, self.padfn)
- return {'cek': cek, 'ek': ek}
-
- def unwrap(self, key, keylen, ek, headers):
- self._check_key(key)
- rk = key.get_op_key('decrypt')
- cek = rk.decrypt(ek, self.padfn)
- if len(cek) != keylen:
- raise InvalidJWEKeyLength(keylen, len(cek))
- return cek
-
-
-class _Rsa15(_RSA):
- def __init__(self):
- super(_Rsa15, self).__init__(padding.PKCS1v15())
-
- @property
- def name(self):
- return 'RSA1_5'
-
-
-class _RsaOaep(_RSA):
- def __init__(self):
- super(_RsaOaep, self).__init__(
- padding.OAEP(padding.MGF1(hashes.SHA1()),
- hashes.SHA1(), None))
-
- @property
- def name(self):
- return 'RSA-OAEP'
-
-
-class _RsaOaep256(_RSA): # noqa: ignore=N801
- def __init__(self):
- super(_RsaOaep256, self).__init__(
- padding.OAEP(padding.MGF1(hashes.SHA256()),
- hashes.SHA256(), None))
-
- @property
- def name(self):
- return 'RSA-OAEP-256'
-
-
-class _AesKw(_RawKeyMgmt):
-
- def __init__(self, keysize):
- self.backend = default_backend()
- self.keysize = keysize // 8
-
- def _get_key(self, key, op):
- if not isinstance(key, JWK):
- raise ValueError('key is not a JWK object')
- if key.key_type != 'oct':
- raise InvalidJWEKeyType('oct', key.key_type)
- rk = base64url_decode(key.get_op_key(op))
- if len(rk) != self.keysize:
- raise InvalidJWEKeyLength(self.keysize * 8, len(rk) * 8)
- return rk
-
- def wrap(self, key, keylen, cek, headers):
- rk = self._get_key(key, 'encrypt')
- if not cek:
- cek = os.urandom(keylen)
-
- # Implement RFC 3394 Key Unwrap - 2.2.2
- # TODO: Use cryptography once issue #1733 is resolved
- iv = 'a6a6a6a6a6a6a6a6'
- a = unhexlify(iv)
- r = [cek[i:i + 8] for i in range(0, len(cek), 8)]
- n = len(r)
- for j in range(0, 6):
- for i in range(0, n):
- e = Cipher(algorithms.AES(rk), modes.ECB(),
- backend=self.backend).encryptor()
- b = e.update(a + r[i]) + e.finalize()
- a = _encode_int(_decode_int(b[:8]) ^ ((n * j) + i + 1), 64)
- r[i] = b[-8:]
- ek = a
- for i in range(0, n):
- ek += r[i]
- return {'cek': cek, 'ek': ek}
-
- def unwrap(self, key, keylen, ek, headers):
- rk = self._get_key(key, 'decrypt')
-
- # Implement RFC 3394 Key Unwrap - 2.2.3
- # TODO: Use cryptography once issue #1733 is resolved
- iv = 'a6a6a6a6a6a6a6a6'
- aiv = unhexlify(iv)
-
- r = [ek[i:i + 8] for i in range(0, len(ek), 8)]
- a = r.pop(0)
- n = len(r)
- for j in range(5, -1, -1):
- for i in range(n - 1, -1, -1):
- da = _decode_int(a)
- atr = _encode_int((da ^ ((n * j) + i + 1)), 64) + r[i]
- d = Cipher(algorithms.AES(rk), modes.ECB(),
- backend=self.backend).decryptor()
- b = d.update(atr) + d.finalize()
- a = b[:8]
- r[i] = b[-8:]
-
- if a != aiv:
- raise InvalidJWEData('Decryption Failed')
-
- cek = b''.join(r)
- if len(cek) != keylen:
- raise InvalidJWEKeyLength(keylen, len(cek))
- return cek
-
-
-class _A128KW(_AesKw):
- def __init__(self):
- super(_A128KW, self).__init__(128)
-
- @property
- def name(self):
- return 'A128KW'
-
-
-class _A192KW(_AesKw):
- def __init__(self):
- super(_A192KW, self).__init__(192)
-
- @property
- def name(self):
- return 'A192KW'
-
-
-class _A256KW(_AesKw):
- def __init__(self):
- super(_A256KW, self).__init__(256)
-
- @property
- def name(self):
- return 'A256KW'
-
-
-class _AesGcmKw(_RawKeyMgmt):
-
- def __init__(self, keysize):
- self.backend = default_backend()
- self.keysize = keysize // 8
-
- def _get_key(self, key, op):
- if not isinstance(key, JWK):
- raise ValueError('key is not a JWK object')
- if key.key_type != 'oct':
- raise InvalidJWEKeyType('oct', key.key_type)
- rk = base64url_decode(key.get_op_key(op))
- if len(rk) != self.keysize:
- raise InvalidJWEKeyLength(self.keysize * 8, len(rk) * 8)
- return rk
-
- def wrap(self, key, keylen, cek, headers):
- rk = self._get_key(key, 'encrypt')
- if not cek:
- cek = os.urandom(keylen)
-
- iv = os.urandom(96 // 8)
- cipher = Cipher(algorithms.AES(rk), modes.GCM(iv),
- backend=self.backend)
- encryptor = cipher.encryptor()
- ek = encryptor.update(cek) + encryptor.finalize()
-
- tag = encryptor.tag
- return {'cek': cek, 'ek': ek,
- 'header': {'iv': base64url_encode(iv),
- 'tag': base64url_encode(tag)}}
-
- def unwrap(self, key, keylen, ek, headers):
- rk = self._get_key(key, 'decrypt')
-
- if 'iv' not in headers:
- raise InvalidJWEData('Invalid Header, missing "iv" parameter')
- iv = base64url_decode(headers['iv'])
- if 'tag' not in headers:
- raise InvalidJWEData('Invalid Header, missing "tag" parameter')
- tag = base64url_decode(headers['tag'])
-
- cipher = Cipher(algorithms.AES(rk), modes.GCM(iv, tag),
- backend=self.backend)
- decryptor = cipher.decryptor()
- cek = decryptor.update(ek) + decryptor.finalize()
- if len(cek) != keylen:
- raise InvalidJWEKeyLength(keylen, len(cek))
- return cek
-
-
-class _A128GcmKw(_AesGcmKw):
- def __init__(self):
- super(_A128GcmKw, self).__init__(128)
-
- @property
- def name(self):
- return 'A128GCMKW'
-
-
-class _A192GcmKw(_AesGcmKw):
- def __init__(self):
- super(_A192GcmKw, self).__init__(192)
-
- @property
- def name(self):
- return 'A192GCMKW'
-
-
-class _A256GcmKw(_AesGcmKw):
- def __init__(self):
- super(_A256GcmKw, self).__init__(256)
-
- @property
- def name(self):
- return 'A256GCMKW'
-
-
-class _Pbes2HsAesKw(_RawKeyMgmt):
-
- @property
- def name(self):
- raise NotImplementedError
-
- def __init__(self, hashsize, keysize):
- self.backend = default_backend()
- self.hashsize = hashsize
- self.keysize = keysize // 8
-
- def _get_key(self, alg, key, p2s, p2c):
- if isinstance(key, bytes):
- plain = key
- else:
- plain = key.encode('utf8')
- salt = bytes(self.name.encode('utf8')) + b'\x00' + p2s
-
- if self.hashsize == 256:
- hashalg = hashes.SHA256()
- elif self.hashsize == 384:
- hashalg = hashes.SHA384()
- elif self.hashsize == 512:
- hashalg = hashes.SHA512()
- else:
- raise InvalidJWEData('Unknown Hash Size')
-
- kdf = PBKDF2HMAC(algorithm=hashalg, length=self.keysize, salt=salt,
- iterations=p2c, backend=self.backend)
- rk = kdf.derive(plain)
- if len(rk) != self.keysize:
- raise InvalidJWEKeyLength(self.keysize * 8, len(rk) * 8)
- return JWK(kty="oct", use="enc", k=base64url_encode(rk))
-
- def wrap(self, key, keylen, cek, headers):
- p2s = os.urandom(16)
- p2c = 8192
- kek = self._get_key(headers['alg'], key, p2s, p2c)
-
- aeskw = _AesKw(self.keysize * 8)
- ret = aeskw.wrap(kek, keylen, cek, headers)
- ret['header'] = {'p2s': base64url_encode(p2s), 'p2c': p2c}
- return ret
-
- def unwrap(self, key, keylen, ek, headers):
- if 'p2s' not in headers:
- raise InvalidJWEData('Invalid Header, missing "p2s" parameter')
- if 'p2c' not in headers:
- raise InvalidJWEData('Invalid Header, missing "p2c" parameter')
- p2s = base64url_decode(headers['p2s'])
- p2c = headers['p2c']
- kek = self._get_key(headers['alg'], key, p2s, p2c)
-
- aeskw = _AesKw(self.keysize * 8)
- return aeskw.unwrap(kek, keylen, ek, headers)
-
-
-class _Pbes2Hs256A128Kw(_Pbes2HsAesKw):
- def __init__(self):
- super(_Pbes2Hs256A128Kw, self).__init__(256, 128)
-
- @property
- def name(self):
- return 'PBES2-HS256+A128KW'
-
-
-class _Pbes2Hs384A192Kw(_Pbes2HsAesKw):
- def __init__(self):
- super(_Pbes2Hs384A192Kw, self).__init__(384, 192)
-
- @property
- def name(self):
- return 'PBES2-HS384+A192KW'
-
-
-class _Pbes2Hs512A256Kw(_Pbes2HsAesKw):
- def __init__(self):
- super(_Pbes2Hs512A256Kw, self).__init__(512, 256)
-
- @property
- def name(self):
- return 'PBES2-HS512+A256KW'
-
-
-class _Direct(_RawKeyMgmt):
-
- @property
- def name(self):
- return 'dir'
-
- def _check_key(self, key):
- if not isinstance(key, JWK):
- raise ValueError('key is not a JWK object')
- if key.key_type != 'oct':
- raise InvalidJWEKeyType('oct', key.key_type)
-
- def wrap(self, key, keylen, cek, headers):
- self._check_key(key)
- if cek:
- return (cek, None)
- k = base64url_decode(key.get_op_key('encrypt'))
- if len(k) != keylen:
- raise InvalidCEKeyLength(keylen, len(k))
- return {'cek': k}
-
- def unwrap(self, key, keylen, ek, headers):
- self._check_key(key)
- if ek != b'':
- raise InvalidJWEData('Invalid Encryption Key.')
- cek = base64url_decode(key.get_op_key('decrypt'))
- if len(cek) != keylen:
- raise InvalidJWEKeyLength(keylen, len(cek))
- return cek
-
-
-class _EcdhEs(_RawKeyMgmt):
-
- @property
- def name(self):
- return 'ECDH-ES'
-
- def __init__(self, keydatalen=None):
- self.backend = default_backend()
- self.keydatalen = keydatalen
-
- 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)
-
- def _derive(self, privkey, pubkey, alg, keydatalen, headers):
- # OtherInfo is defined in NIST SP 56A 5.8.1.2.1
-
- # AlgorithmID
- otherinfo = struct.pack('>I', len(alg))
- otherinfo += bytes(alg.encode('utf8'))
-
- # PartyUInfo
- apu = base64url_decode(headers['apu']) if 'apu' in headers else b''
- otherinfo += struct.pack('>I', len(apu))
- otherinfo += apu
-
- # PartyVInfo
- apv = base64url_decode(headers['apv']) if 'apv' in headers else b''
- otherinfo += struct.pack('>I', len(apv))
- otherinfo += apv
-
- # SuppPubInfo
- otherinfo += struct.pack('>I', keydatalen)
-
- # no SuppPrivInfo
-
- shared_key = privkey.exchange(ec.ECDH(), pubkey)
- ckdf = ConcatKDFHash(algorithm=hashes.SHA256(),
- length=keydatalen // 8,
- otherinfo=otherinfo,
- backend=self.backend)
- return ckdf.derive(shared_key)
-
- def wrap(self, key, keylen, cek, headers):
- self._check_key(key)
- if self.keydatalen is None:
- if cek is not None:
- raise InvalidJWEOperation('ECDH-ES cannot use an existing CEK')
- keydatalen = keylen * 8
- alg = headers['enc']
- else:
- keydatalen = self.keydatalen
- alg = headers['alg']
-
- epk = JWK.generate(kty=key.key_type, crv=key.key_curve)
- dk = self._derive(epk.get_op_key('unwrapKey'),
- key.get_op_key('wrapKey'),
- alg, keydatalen, headers)
-
- if self.keydatalen is None:
- ret = {'cek': dk}
- else:
- aeskw = _AesKw(keydatalen)
- kek = JWK(kty="oct", use="enc", k=base64url_encode(dk))
- ret = aeskw.wrap(kek, keydatalen // 8, cek, headers)
-
- ret['header'] = {'epk': json_decode(epk.export_public())}
- return ret
-
- def unwrap(self, key, keylen, ek, headers):
- if 'epk' not in headers:
- raise InvalidJWEData('Invalid Header, missing "epk" parameter')
- self._check_key(key)
- if self.keydatalen is None:
- keydatalen = keylen * 8
- alg = headers['enc']
- else:
- keydatalen = self.keydatalen
- alg = headers['alg']
-
- epk = JWK(**headers['epk'])
- dk = self._derive(key.get_op_key('unwrapKey'),
- epk.get_op_key('wrapKey'),
- alg, keydatalen, headers)
- if self.keydatalen is None:
- return dk
- else:
- aeskw = _AesKw(keydatalen)
- kek = JWK(kty="oct", use="enc", k=base64url_encode(dk))
- cek = aeskw.unwrap(kek, keydatalen // 8, ek, headers)
- return cek
-
-
-class _EcdhEsAes128Kw(_EcdhEs):
- def __init__(self):
- super(_EcdhEsAes128Kw, self).__init__(128)
-
- @property
- def name(self):
- return 'ECDH-ES+A128KW'
-
-
-class _EcdhEsAes192Kw(_EcdhEs):
- def __init__(self):
- super(_EcdhEsAes192Kw, self).__init__(192)
-
- @property
- def name(self):
- return 'ECDH-ES+A192KW'
-
-
-class _EcdhEsAes256Kw(_EcdhEs):
- def __init__(self):
- super(_EcdhEsAes256Kw, self).__init__(256)
-
- @property
- def name(self):
- return 'ECDH-ES+A256KW'
-
-
-class _RawJWE(object):
-
- def encrypt(self, k, a, m):
- raise NotImplementedError
-
- def decrypt(self, k, a, iv, e, t):
- raise NotImplementedError
-
-
-class _AesCbcHmacSha2(_RawJWE):
-
- def __init__(self, hashfn, keybits):
- self.backend = default_backend()
- self.hashfn = hashfn
- self.keysize = keybits // 8
- self.blocksize = algorithms.AES.block_size
-
- @property
- def key_size(self):
- return self.keysize * 2
-
- def _mac(self, k, a, iv, e):
- al = _encode_int(len(a * 8), 64)
- h = hmac.HMAC(k, self.hashfn, backend=self.backend)
- h.update(a)
- h.update(iv)
- h.update(e)
- h.update(al)
- m = h.finalize()
- return m[:self.keysize]
-
- # RFC 7518 - 5.2.2
- def encrypt(self, k, a, m):
- """ Encrypt according to the selected encryption and hashing
- functions.
-
- :param k: Encryption key (optional)
- :param a: Additional Authentication Data
- :param m: Plaintext
-
- Returns a dictionary with the computed data.
- """
- hkey = k[:self.keysize]
- ekey = k[self.keysize:]
-
- # encrypt
- iv = os.urandom(self.blocksize // 8)
- cipher = Cipher(algorithms.AES(ekey), modes.CBC(iv),
- backend=self.backend)
- encryptor = cipher.encryptor()
- padder = PKCS7(self.blocksize).padder()
- padded_data = padder.update(m) + padder.finalize()
- e = encryptor.update(padded_data) + encryptor.finalize()
-
- # mac
- t = self._mac(hkey, a, iv, e)
-
- return (iv, e, t)
-
- def decrypt(self, k, a, iv, e, t):
- """ Decrypt according to the selected encryption and hashing
- functions.
- :param k: Encryption key (optional)
- :param a: Additional Authenticated Data
- :param iv: Initialization Vector
- :param e: Ciphertext
- :param t: Authentication Tag
-
- Returns plaintext or raises an error
- """
- hkey = k[:self.keysize]
- dkey = k[self.keysize:]
-
- # verify mac
- if not constant_time.bytes_eq(t, self._mac(hkey, a, iv, e)):
- raise InvalidJWEData('Failed to verify MAC')
-
- # decrypt
- cipher = Cipher(algorithms.AES(dkey), modes.CBC(iv),
- backend=self.backend)
- decryptor = cipher.decryptor()
- d = decryptor.update(e) + decryptor.finalize()
- unpadder = PKCS7(self.blocksize).unpadder()
- return unpadder.update(d) + unpadder.finalize()
-
-
-class _A128CbcHs256(_AesCbcHmacSha2):
- def __init__(self):
- super(_A128CbcHs256, self).__init__(hashes.SHA256(), 128)
-
- @property
- def name(self):
- return 'A128CBC-HS256'
-
-
-class _A192CbcHs384(_AesCbcHmacSha2):
- def __init__(self):
- super(_A192CbcHs384, self).__init__(hashes.SHA384(), 192)
-
- @property
- def name(self):
- return 'A192CBC-HS384'
-
-
-class _A256CbcHs512(_AesCbcHmacSha2):
- def __init__(self):
- super(_A256CbcHs512, self).__init__(hashes.SHA512(), 256)
-
- @property
- def name(self):
- return 'A256CBC-HS512'
-
-
-class _AesGcm(_RawJWE):
-
- def __init__(self, keybits):
- self.backend = default_backend()
- self.keysize = keybits // 8
-
- @property
- def key_size(self):
- return self.keysize
-
- # RFC 7518 - 5.3
- def encrypt(self, k, a, m):
- """ Encrypt accoriding to the selected encryption and hashing
- functions.
-
- :param k: Encryption key (optional)
- :param a: Additional Authentication Data
- :param m: Plaintext
-
- Returns a dictionary with the computed data.
- """
- iv = os.urandom(96 // 8)
- cipher = Cipher(algorithms.AES(k), modes.GCM(iv),
- backend=self.backend)
- encryptor = cipher.encryptor()
- encryptor.authenticate_additional_data(a)
- e = encryptor.update(m) + encryptor.finalize()
-
- return (iv, e, encryptor.tag)
-
- def decrypt(self, k, a, iv, e, t):
- """ Decrypt accoriding to the selected encryption and hashing
- functions.
- :param k: Encryption key (optional)
- :param a: Additional Authenticated Data
- :param iv: Initialization Vector
- :param e: Ciphertext
- :param t: Authentication Tag
-
- Returns plaintext or raises an error
- """
- cipher = Cipher(algorithms.AES(k), modes.GCM(iv, t),
- backend=self.backend)
- decryptor = cipher.decryptor()
- decryptor.authenticate_additional_data(a)
- return decryptor.update(e) + decryptor.finalize()
-
-
-class _A128Gcm(_AesGcm):
- def __init__(self):
- super(_A128Gcm, self).__init__(128)
-
- @property
- def name(self):
- return 'A128GCM'
-
-
-class _A192Gcm(_AesGcm):
- def __init__(self):
- super(_A192Gcm, self).__init__(192)
-
- @property
- def name(self):
- return 'A192GCM'
-
-
-class _A256Gcm(_AesGcm):
- def __init__(self):
- super(_A256Gcm, self).__init__(256)
-
- @property
- def name(self):
- return 'A256GCM'
+# These have been moved to jwcrypto.common, maintain here for backwards compat
+InvalidCEKeyLength = common.InvalidCEKeyLength
+InvalidJWEKeyLength = common.InvalidJWEKeyLength
+InvalidJWEKeyType = common.InvalidJWEKeyType
+InvalidJWEOperation = common.InvalidJWEOperation
class JWE(object):
@@ -815,34 +72,8 @@ class JWE(object):
This object represent a JWE token.
"""
- jwas = {
- 'RSA1_5': _Rsa15,
- 'RSA-OAEP': _RsaOaep,
- 'RSA-OAEP-256': _RsaOaep256,
- 'A128KW': _A128KW,
- 'A192KW': _A192KW,
- 'A256KW': _A256KW,
- 'dir': _Direct,
- 'ECDH-ES': _EcdhEs,
- 'ECDH-ES+A128KW': _EcdhEsAes128Kw,
- 'ECDH-ES+A192KW': _EcdhEsAes192Kw,
- 'ECDH-ES+A256KW': _EcdhEsAes256Kw,
- 'A128GCMKW': _A128GcmKw,
- 'A192GCMKW': _A192GcmKw,
- 'A256GCMKW': _A256GcmKw,
- 'PBES2-HS256+A128KW': _Pbes2Hs256A128Kw,
- 'PBES2-HS384+A192KW': _Pbes2Hs384A192Kw,
- 'PBES2-HS512+A256KW': _Pbes2Hs512A256Kw,
- 'A128CBC-HS256': _A128CbcHs256,
- 'A192CBC-HS384': _A192CbcHs384,
- 'A256CBC-HS512': _A256CbcHs512,
- 'A128GCM': _A128Gcm,
- 'A192GCM': _A192Gcm,
- 'A256GCM': _A256Gcm
- }
-
def __init__(self, plaintext=None, protected=None, unprotected=None,
- aad=None, algs=None):
+ aad=None, algs=None, recipient=None, header=None):
"""Creates a JWE token.
:param plaintext(bytes): An arbitrary plaintext to be encrypted.
@@ -850,6 +81,8 @@ class JWE(object):
:param unprotected: A JSON string with the shared unprotected header.
:param aad(bytes): Arbitrary additional authenticated data
:param algs: An optional list of allowed algorithms
+ :param recipient: An optional, default recipient key
+ :param header: An optional header for the default recipient
"""
self._allowed_algs = None
self.objects = dict()
@@ -864,23 +97,36 @@ class JWE(object):
if aad:
self.objects['aad'] = aad
if protected:
- json_decode(protected) # check header encoding
+ if isinstance(protected, dict):
+ protected = json_encode(protected)
+ else:
+ json_decode(protected) # check header encoding
self.objects['protected'] = protected
if unprotected:
- json_decode(unprotected) # check header encoding
+ if isinstance(unprotected, dict):
+ unprotected = json_encode(unprotected)
+ else:
+ json_decode(unprotected) # check header encoding
self.objects['unprotected'] = unprotected
if algs:
- self.allowed_algs = algs
+ self._allowed_algs = algs
- def _jwa(self, name):
- try:
- cls = self.jwas[name]
- except (KeyError):
- raise InvalidJWAAlgorithm()
+ if recipient:
+ self.add_recipient(recipient, header=header)
+ elif header:
+ raise ValueError('Header is allowed only with default recipient')
+
+ def _jwa_keymgmt(self, name):
+ allowed = self._allowed_algs or default_allowed_algs
+ if name not in allowed:
+ raise InvalidJWEOperation('Algorithm not allowed')
+ return JWA.keymgmt_alg(name)
+
+ def _jwa_enc(self, name):
allowed = self._allowed_algs or default_allowed_algs
if name not in allowed:
raise InvalidJWEOperation('Algorithm not allowed')
- return cls()
+ return JWA.encryption_alg(name)
@property
def allowed_algs(self):
@@ -925,11 +171,11 @@ class JWE(object):
algname = jh.get('alg', None)
if algname is None:
raise InvalidJWEData('Missing "alg" from headers')
- alg = self._jwa(algname)
+ alg = self._jwa_keymgmt(algname)
encname = jh.get('enc', None)
if encname is None:
raise InvalidJWEData('Missing "enc" from headers')
- enc = self._jwa(encname)
+ enc = self._jwa_enc(encname)
return alg, enc
def _encrypt(self, alg, enc, jh):
@@ -968,6 +214,9 @@ class JWE(object):
if not isinstance(self.plaintext, bytes):
raise ValueError("Plaintext must be 'bytes'")
+ if isinstance(header, dict):
+ header = json_encode(header)
+
jh = self._get_jose_header(header)
alg, enc = self._get_alg_enc_from_headers(jh)
@@ -975,7 +224,7 @@ class JWE(object):
if header:
rec['header'] = header
- wrapped = alg.wrap(key, enc.key_size, self.cek, jh)
+ wrapped = alg.wrap(key, enc.wrap_key_size, self.cek, jh)
self.cek = wrapped['cek']
if 'ek' in wrapped:
@@ -995,11 +244,9 @@ class JWE(object):
self.objects['recipients'] = list()
n = dict()
if 'encrypted_key' in self.objects:
- n['encrypted_key'] = self.objects['encrypted_key']
- del self.objects['encrypted_key']
+ n['encrypted_key'] = self.objects.pop('encrypted_key')
if 'header' in self.objects:
- n['header'] = self.objects['header']
- del self.objects['header']
+ n['header'] = self.objects.pop('header')
self.objects['recipients'].append(n)
self.objects['recipients'].append(rec)
else:
@@ -1023,7 +270,19 @@ class JWE(object):
if compact:
for invalid in 'aad', 'unprotected':
if invalid in self.objects:
- raise InvalidJWEOperation("Can't use compact encoding")
+ raise InvalidJWEOperation(
+ "Can't use compact encoding when the '%s' parameter"
+ "is set" % invalid)
+ if 'protected' not in self.objects:
+ raise InvalidJWEOperation(
+ "Can't use compat encoding without protected headers")
+ else:
+ ph = json_decode(self.objects['protected'])
+ for required in 'alg', 'enc':
+ if required not in ph:
+ raise InvalidJWEOperation(
+ "Can't use compat encoding, '%s' must be in the "
+ "protected header" % required)
if 'recipients' in self.objects:
if len(self.objects['recipients']) != 1:
raise InvalidJWEOperation("Invalid number of recipients")
@@ -1095,14 +354,15 @@ class JWE(object):
# TODO: allow caller to specify list of headers it understands
self._check_crit(jh.get('crit', dict()))
- alg = self._jwa(jh.get('alg', None))
- enc = self._jwa(jh.get('enc', None))
+ alg = self._jwa_keymgmt(jh.get('alg', None))
+ enc = self._jwa_enc(jh.get('enc', None))
aad = base64url_encode(self.objects.get('protected', ''))
if 'aad' in self.objects:
aad += '.' + base64url_encode(self.objects['aad'])
- cek = alg.unwrap(key, enc.key_size, ppe.get('encrypted_key', b''), jh)
+ cek = alg.unwrap(key, enc.wrap_key_size,
+ ppe.get('encrypted_key', b''), jh)
data = enc.decrypt(cek, aad.encode('utf-8'),
self.objects['iv'],
self.objects['ciphertext'],
@@ -1176,30 +436,30 @@ class JWE(object):
try:
try:
djwe = json_decode(raw_jwe)
- o['iv'] = base64url_decode(str(djwe['iv']))
- o['ciphertext'] = base64url_decode(str(djwe['ciphertext']))
- o['tag'] = base64url_decode(str(djwe['tag']))
+ o['iv'] = base64url_decode(djwe['iv'])
+ o['ciphertext'] = base64url_decode(djwe['ciphertext'])
+ o['tag'] = base64url_decode(djwe['tag'])
if 'protected' in djwe:
- p = base64url_decode(str(djwe['protected']))
+ p = base64url_decode(djwe['protected'])
o['protected'] = p.decode('utf-8')
if 'unprotected' in djwe:
o['unprotected'] = json_encode(djwe['unprotected'])
if 'aad' in djwe:
- o['aad'] = base64url_decode(str(djwe['aad']))
+ o['aad'] = base64url_decode(djwe['aad'])
if 'recipients' in djwe:
o['recipients'] = list()
for rec in djwe['recipients']:
e = dict()
if 'encrypted_key' in rec:
e['encrypted_key'] = \
- base64url_decode(str(rec['encrypted_key']))
+ base64url_decode(rec['encrypted_key'])
if 'header' in rec:
e['header'] = json_encode(rec['header'])
o['recipients'].append(e)
else:
if 'encrypted_key' in djwe:
o['encrypted_key'] = \
- base64url_decode(str(djwe['encrypted_key']))
+ base64url_decode(djwe['encrypted_key'])
if 'header' in djwe:
o['header'] = json_encode(djwe['header'])
@@ -1207,14 +467,14 @@ class JWE(object):
c = raw_jwe.split('.')
if len(c) != 5:
raise InvalidJWEData()
- p = base64url_decode(str(c[0]))
+ p = base64url_decode(c[0])
o['protected'] = p.decode('utf-8')
- ekey = base64url_decode(str(c[1]))
- if ekey != '':
- o['encrypted_key'] = base64url_decode(str(c[1]))
- o['iv'] = base64url_decode(str(c[2]))
- o['ciphertext'] = base64url_decode(str(c[3]))
- o['tag'] = base64url_decode(str(c[4]))
+ ekey = base64url_decode(c[1])
+ if ekey != b'':
+ o['encrypted_key'] = base64url_decode(c[1])
+ o['iv'] = base64url_decode(c[2])
+ o['ciphertext'] = base64url_decode(c[3])
+ o['tag'] = base64url_decode(c[4])
self.objects = o
=====================================
jwcrypto/jwk.py
=====================================
@@ -1,16 +1,19 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
import os
-
from binascii import hexlify, unhexlify
+from collections import namedtuple
+from enum import Enum
+from cryptography import x509
from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric import rsa
from six import iteritems
+from jwcrypto.common import JWException
from jwcrypto.common import base64url_decode, base64url_encode
from jwcrypto.common import json_decode, json_encode
@@ -21,36 +24,59 @@ JWKTypesRegistry = {'EC': 'Elliptic Curve',
'oct': 'Octet sequence'}
"""Registry of valid Key Types"""
+
# RFC 7518 - 7.5
# It is part of the JWK Parameters Registry, but we want a more
# specific map for internal usage
-JWKValuesRegistry = {'EC': {'crv': ('Curve', 'Public', 'Required'),
- 'x': ('X Coordinate', 'Public', 'Required'),
- 'y': ('Y Coordinate', 'Public', 'Required'),
- 'd': ('ECC Private Key', 'Private', None)},
- 'RSA': {'n': ('Modulus', 'Public', 'Required'),
- 'e': ('Exponent', 'Public', 'Required'),
- 'd': ('Private Exponent', 'Private', None),
- 'p': ('First Prime Factor', 'Private', None),
- 'q': ('Second Prime Factor', 'Private', None),
- 'dp': ('First Factor CRT Exponent', 'Private',
- None),
- 'dq': ('Second Factor CRT Exponent', 'Private',
- None),
- 'qi': ('First CRT Coefficient', 'Private', None)},
- 'oct': {'k': ('Key Value', 'Private', 'Required')}}
+class ParmType(Enum):
+ name = 'A string with a name'
+ b64 = 'Base64url Encoded'
+ b64U = 'Base64urlUint Encoded'
+ unsupported = 'Unsupported Parameter'
+
+
+JWKParameter = namedtuple('Parameter', 'description public required type')
+JWKValuesRegistry = {
+ 'EC': {
+ 'crv': JWKParameter('Curve', True, True, ParmType.name),
+ 'x': JWKParameter('X Coordinate', True, True, ParmType.b64),
+ 'y': JWKParameter('Y Coordinate', True, True, ParmType.b64),
+ 'd': JWKParameter('ECC Private Key', False, False, ParmType.b64),
+ },
+ '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),
+ 'dp': JWKParameter('First Factor CRT Exponent',
+ False, False, ParmType.b64U),
+ 'dq': JWKParameter('Second Factor CRT Exponent',
+ False, False, ParmType.b64U),
+ 'qi': JWKParameter('First CRT Coefficient',
+ False, False, ParmType.b64U),
+ 'oth': JWKParameter('Other Primes Info',
+ False, False, ParmType.unsupported),
+ },
+ 'oct': {
+ 'k': JWKParameter('Key Value', False, True, ParmType.b64),
+ }
+}
"""Registry of valid key values"""
-JWKParamsRegistry = {'kty': ('Key Type', 'Public', ),
- 'use': ('Public Key Use', 'Public'),
- 'key_ops': ('Key Operations', 'Public'),
- 'alg': ('Algorithm', 'Public'),
- 'kid': ('Key ID', 'Public'),
- 'x5u': ('X.509 URL', 'Public'),
- 'x5c': ('X.509 Certificate Chain', 'Public'),
- 'x5t': ('X.509 Certificate SHA-1 Thumbprint', 'Public'),
- 'x5t#S256': ('X.509 Certificate SHA-256 Thumbprint',
- 'Public')}
+JWKParamsRegistry = {
+ 'kty': JWKParameter('Key Type', True, None, None),
+ 'use': JWKParameter('Public Key Use', True, None, None),
+ 'key_ops': JWKParameter('Key Operations', True, None, None),
+ 'alg': JWKParameter('Algorithm', True, None, None),
+ 'kid': JWKParameter('Key ID', True, None, None),
+ 'x5u': JWKParameter('X.509 URL', True, None, None),
+ 'x5c': JWKParameter('X.509 Certificate Chain', True, None, None),
+ 'x5t': JWKParameter('X.509 Certificate SHA-1 Thumbprint',
+ True, None, None),
+ 'x5t#S256': JWKParameter('X.509 Certificate SHA-256 Thumbprint',
+ True, None, None)
+}
"""Regstry of valid key parameters"""
# RFC 7518 - 7.6
@@ -82,7 +108,7 @@ JWKpycaCurveMap = {'secp256r1': 'P-256',
'secp521r1': 'P-521'}
-class InvalidJWKType(Exception):
+class InvalidJWKType(JWException):
"""Invalid JWK Type Exception.
This exception is raised when an invalid parameter type is used.
@@ -97,7 +123,7 @@ class InvalidJWKType(Exception):
self.value, list(JWKTypesRegistry.keys()))
-class InvalidJWKUsage(Exception):
+class InvalidJWKUsage(JWException):
"""Invalid JWK usage Exception.
This exception is raised when an invalid key usage is requested,
@@ -122,7 +148,7 @@ class InvalidJWKUsage(Exception):
valid)
-class InvalidJWKOperation(Exception):
+class InvalidJWKOperation(JWException):
"""Invalid JWK Operation Exception.
This exception is raised when an invalid key operation is requested,
@@ -149,7 +175,7 @@ class InvalidJWKOperation(Exception):
valid)
-class InvalidJWKValue(Exception):
+class InvalidJWKValue(JWException):
"""Invalid JWK Value Exception.
This exception is raised when an invalid/unknown value is used in the
@@ -165,7 +191,7 @@ class JWK(object):
This object represent a Key.
It must be instantiated by using the standard defined key/value pairs
- as arguents of the initialization function.
+ as arguments of the initialization function.
"""
def __init__(self, **kwargs):
@@ -209,6 +235,7 @@ class JWK(object):
@classmethod
def generate(cls, **kwargs):
obj = cls()
+ kty = None
try:
kty = kwargs['kty']
gen = getattr(obj, '_generate_%s' % kty)
@@ -218,20 +245,30 @@ class JWK(object):
return obj
def generate_key(self, **params):
+ kty = None
try:
- kty = params['generate']
- del params['generate']
+ kty = params.pop('generate')
gen = getattr(self, '_generate_%s' % kty)
except (KeyError, AttributeError):
raise InvalidJWKType(kty)
gen(params)
- def _generate_oct(self, params):
- size = 128
+ def _get_gen_size(self, params, default_size=None):
+ size = default_size
if 'size' in params:
- size = params['size']
- del params['size']
+ size = params.pop('size')
+ elif 'alg' in params:
+ try:
+ from jwcrypto.jwa import JWA
+ alg = JWA.instantiate_alg(params['alg'])
+ except KeyError:
+ raise ValueError("Invalid 'alg' parameter")
+ size = alg.keysize
+ return size
+
+ def _generate_oct(self, params):
+ size = self._get_gen_size(params, 128)
key = os.urandom(size // 8)
params['kty'] = 'oct'
params['k'] = base64url_encode(key)
@@ -243,13 +280,9 @@ class JWK(object):
def _generate_RSA(self, params):
pubexp = 65537
- size = 2048
+ size = self._get_gen_size(params, 2048)
if 'public_exponent' in params:
- pubexp = params['public_exponent']
- del params['public_exponent']
- if 'size' in params:
- size = params['size']
- del params['size']
+ pubexp = params.pop('public_exponent')
key = rsa.generate_private_key(pubexp, size, default_backend())
self._import_pyca_pri_rsa(key, **params)
@@ -290,13 +323,11 @@ class JWK(object):
def _generate_EC(self, params):
curve = 'P-256'
if 'curve' in params:
- curve = params['curve']
- del params['curve']
+ curve = params.pop('curve')
# 'curve' is for backwards compat, if 'crv' is defined it takes
# precedence
if 'crv' in params:
- curve = params['crv']
- del params['crv']
+ curve = params.pop('crv')
curve_name = self._get_curve_by_name(curve)
key = ec.generate_private_key(curve_name, default_backend())
self._import_pyca_pri_ec(key, **params)
@@ -342,8 +373,26 @@ class JWK(object):
names.remove(name)
for name, val in iteritems(JWKValuesRegistry[kty]):
- if val[2] == 'Required' and name not in self._key:
+ if val.required and name not in self._key:
raise InvalidJWKValue('Missing required value %s' % name)
+ if val.type == ParmType.unsupported and name in self._key:
+ raise InvalidJWKValue('Unsupported parameter %s' % name)
+ if val.type == ParmType.b64 and name in self._key:
+ # Check that the value is base64url encoded
+ try:
+ base64url_decode(self._key[name])
+ except Exception: # pylint: disable=broad-except
+ raise InvalidJWKValue(
+ '"%s" is not base64url encoded' % name
+ )
+ if val[3] == ParmType.b64U and name in self._key:
+ # Check that the value is Base64urlUInt encoded
+ try:
+ self._decode_int(self._key[name])
+ except Exception: # pylint: disable=broad-except
+ raise InvalidJWKValue(
+ '"%s" is not Base64urlUInt encoded' % name
+ )
# Unknown key parameters are allowed
# Let's just store them out of the way
@@ -381,36 +430,107 @@ class JWK(object):
' "key_ops" values specified at'
' the same time')
+ @classmethod
+ def from_json(cls, key):
+ """Creates a RFC 7517 JWK from the standard JSON format.
+
+ :param key: The RFC 7517 representation of a JWK.
+ """
+ obj = cls()
+ try:
+ jkey = json_decode(key)
+ except Exception as e: # pylint: disable=broad-except
+ raise InvalidJWKValue(e)
+ obj.import_key(**jkey)
+ return obj
+
def export(self, private_key=True):
"""Exports the key in the standard JSON format.
+ Exports the key regardless of type, if private_key is False
+ and the key is_symmetric an exceptionis raised.
:param private_key(bool): Whether to export the private key.
Defaults to True.
"""
- if private_key is not True:
+ if private_key is True:
+ # Use _export_all for backwards compatibility, as this
+ # function allows to export symmetrict keys too
+ return self._export_all()
+ else:
return self.export_public()
- d = dict()
- d.update(self._params)
- d.update(self._key)
- d.update(self._unknown)
- return json_encode(d)
def export_public(self):
"""Exports the public key in the standard JSON format.
- This function is deprecated and maintained only for
- backwards compatibility, use export(private_key=False)
- instead."""
+ It fails if one is not available like when this function
+ is called on a symmetric key.
+ """
+ pub = self._public_params()
+ return json_encode(pub)
+
+ def _public_params(self):
+ if not self.has_public:
+ raise InvalidJWKType("No public key available")
pub = {}
preg = JWKParamsRegistry
for name in preg:
- if preg[name][1] == 'Public':
+ if preg[name].public:
if name in self._params:
pub[name] = self._params[name]
reg = JWKValuesRegistry[self._params['kty']]
for param in reg:
- if reg[param][1] == 'Public':
+ if reg[param].public:
pub[param] = self._key[param]
- return json_encode(pub)
+ return pub
+
+ def _export_all(self):
+ d = dict()
+ d.update(self._params)
+ d.update(self._key)
+ d.update(self._unknown)
+ return json_encode(d)
+
+ def export_private(self):
+ """Export the private key in the standard JSON format.
+ It fails for a JWK that has only a public key or is symmetric.
+ """
+ if self.has_private:
+ return self._export_all()
+ raise InvalidJWKType("No private key available")
+
+ def export_symmetric(self):
+ if self.is_symmetric:
+ return self._export_all()
+ raise InvalidJWKType("Not a symmetric key")
+
+ def public(self):
+ pub = self._public_params()
+ return JWK(**pub)
+
+ @property
+ def has_public(self):
+ """Whether this JWK has an asymmetric Public key."""
+ if self.is_symmetric:
+ return False
+ reg = JWKValuesRegistry[self._params['kty']]
+ for value in reg:
+ if reg[value].public and value in self._key:
+ return True
+
+ @property
+ def has_private(self):
+ """Whether this JWK has an asymmetric key Private key."""
+ if self.is_symmetric:
+ return False
+ reg = JWKValuesRegistry[self._params['kty']]
+ for value in reg:
+ if not reg[value].public and value in self._key:
+ return True
+ return False
+
+ @property
+ def is_symmetric(self):
+ """Whether this JWK is a symmetric key."""
+ return self.key_type == 'oct'
@property
def key_type(self):
@@ -556,12 +676,89 @@ class JWK(object):
else:
raise InvalidJWKValue('Unknown key object %r' % key)
+ def import_from_pem(self, data, password=None):
+ """Imports a key from data loaded from a PEM file.
+ The key may be encrypted with a password.
+ Private keys (PKCS#8 format), public keys, and X509 certificate's
+ public keys can be imported with this interface.
+
+ :param data(bytes): The data contained in a PEM file.
+ :param password(bytes): An optional password to unwrap the key.
+ """
+
+ try:
+ key = serialization.load_pem_private_key(
+ data, password=password, backend=default_backend())
+ except ValueError as e:
+ if password is not None:
+ raise e
+ try:
+ key = serialization.load_pem_public_key(
+ data, backend=default_backend())
+ except ValueError:
+ try:
+ cert = x509.load_pem_x509_certificate(
+ data, backend=default_backend())
+ key = cert.public_key()
+ except ValueError:
+ raise e
+
+ self.import_from_pyca(key)
+ self._params['kid'] = self.thumbprint()
+
+ def export_to_pem(self, private_key=False, password=False):
+ """Exports keys to a data buffer suitable to be stored as a PEM file.
+ Either the public or the private key can be exported to a PEM file.
+ For private keys the PKCS#8 format is used. If a password is provided
+ the best encryption method available as determined by the cryptography
+ module is used to wrap the key.
+
+ :param private_key: Whether the private key should be exported.
+ Defaults to `False` which means the public key is exported by default.
+ :param password(bytes): A password for wrapping the private key.
+ Defaults to False which will cause the operation to fail. To avoid
+ encryption the user must explicitly pass None, otherwise the user
+ needs to provide a password in a bytes buffer.
+ """
+ e = serialization.Encoding.PEM
+ if private_key:
+ if not self.has_private:
+ raise InvalidJWKType("No private key available")
+ f = serialization.PrivateFormat.PKCS8
+ if password is None:
+ a = serialization.NoEncryption()
+ elif isinstance(password, bytes):
+ a = serialization.BestAvailableEncryption(password)
+ elif password is False:
+ raise ValueError("The password must be None or a bytes string")
+ else:
+ raise TypeError("The password string must be bytes")
+ return self._get_private_key().private_bytes(
+ encoding=e, format=f, encryption_algorithm=a)
+ else:
+ if not self.has_public:
+ raise InvalidJWKType("No public key available")
+ f = serialization.PublicFormat.SubjectPublicKeyInfo
+ return self._get_public_key().public_bytes(encoding=e, format=f)
+
@classmethod
def from_pyca(cls, key):
obj = cls()
obj.import_from_pyca(key)
return obj
+ @classmethod
+ def from_pem(cls, data, password=None):
+ """Creates a key from PKCS#8 formatted data loaded from a PEM file.
+ See the function `import_from_pem` for details.
+
+ :param data(bytes): The data contained in a PEM file.
+ :param password(bytes): An optional password to unwrap the key.
+ """
+ obj = cls()
+ obj.import_from_pem(data, password)
+ return obj
+
def thumbprint(self, hashalg=hashes.SHA256()):
"""Returns the key thumbprint as specified by RFC 7638.
@@ -570,7 +767,7 @@ class JWK(object):
t = {'kty': self._params['kty']}
for name, val in iteritems(JWKValuesRegistry[t['kty']]):
- if val[2] == 'Required':
+ if val.required:
t[name] = self._key[name]
digest = hashes.Hash(hashalg, backend=default_backend())
digest.update(bytes(json_encode(t).encode('utf8')))
@@ -603,6 +800,12 @@ class JWKSet(dict):
super(JWKSet, self).__setitem__('keys', _JWKkeys())
self.update(*args, **kwargs)
+ def __iter__(self):
+ return self['keys'].__iter__()
+
+ def __contains__(self, key):
+ return self['keys'].__contains__(key)
+
def __setitem__(self, key, val):
if key == 'keys':
self['keys'].add(val)
@@ -639,7 +842,7 @@ class JWKSet(dict):
"""
try:
jwkset = json_decode(keyset)
- except:
+ except Exception: # pylint: disable=broad-except
raise InvalidJWKValue()
if 'keys' not in jwkset:
@@ -652,8 +855,6 @@ class JWKSet(dict):
else:
self[k] = v
- return self
-
@classmethod
def from_json(cls, keyset):
"""Creates a RFC 7517 keyset from the standard JSON format.
@@ -661,7 +862,8 @@ class JWKSet(dict):
:param keyset: The RFC 7517 representation of a JOSE Keyset.
"""
obj = cls()
- return obj.import_keyset(keyset)
+ obj.import_keyset(keyset)
+ return obj
def get_key(self, kid):
"""Gets a key from the set.
=====================================
jwcrypto/jws.py
=====================================
@@ -1,34 +1,34 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
-from binascii import hexlify, unhexlify
+from collections import namedtuple
-from cryptography.exceptions import InvalidSignature
-from cryptography.hazmat.backends import default_backend
-from cryptography.hazmat.primitives import hashes, hmac
-from cryptography.hazmat.primitives.asymmetric import ec
-from cryptography.hazmat.primitives.asymmetric import padding
-from cryptography.hazmat.primitives.asymmetric import utils as ec_utils
-
-from jwcrypto.common import InvalidJWAAlgorithm
+from jwcrypto.common import JWException
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?)
-JWSHeaderRegistry = {'alg': ('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)}
+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)
+}
"""Registry of valid header parameters"""
default_allowed_algs = [
@@ -39,7 +39,7 @@ default_allowed_algs = [
"""Default allowed algorithms"""
-class InvalidJWSSignature(Exception):
+class InvalidJWSSignature(JWException):
"""Invalid JWS Signature.
This exception is raised when a signature cannot be validated.
@@ -56,7 +56,7 @@ class InvalidJWSSignature(Exception):
super(InvalidJWSSignature, self).__init__(msg)
-class InvalidJWSObject(Exception):
+class InvalidJWSObject(JWException):
"""Invalid JWS Object.
This exception is raised when the JWS Object is invalid and/or
@@ -72,7 +72,7 @@ class InvalidJWSObject(Exception):
super(InvalidJWSObject, self).__init__(msg)
-class InvalidJWSOperation(Exception):
+class InvalidJWSOperation(JWException):
"""Invalid JWS Object.
This exception is raised when a requested operation cannot
@@ -90,100 +90,6 @@ class InvalidJWSOperation(Exception):
super(InvalidJWSOperation, self).__init__(msg)
-class _RawJWS(object):
-
- def sign(self, key, payload):
- raise NotImplementedError
-
- def verify(self, key, payload, signature):
- raise NotImplementedError
-
-
-class _RawHMAC(_RawJWS):
-
- def __init__(self, hashfn):
- self.backend = default_backend()
- self.hashfn = hashfn
-
- def _hmac_setup(self, key, payload):
- h = hmac.HMAC(key, self.hashfn, backend=self.backend)
- h.update(payload)
- return h
-
- def sign(self, key, payload):
- skey = base64url_decode(key.get_op_key('sign'))
- h = self._hmac_setup(skey, payload)
- return h.finalize()
-
- def verify(self, key, payload, signature):
- vkey = base64url_decode(key.get_op_key('verify'))
- h = self._hmac_setup(vkey, payload)
- try:
- h.verify(signature)
- except InvalidSignature as e:
- raise InvalidJWSSignature(exception=e)
-
-
-class _RawRSA(_RawJWS):
- def __init__(self, padfn, hashfn):
- self.padfn = padfn
- self.hashfn = hashfn
-
- def sign(self, key, payload):
- skey = key.get_op_key('sign')
- signer = skey.signer(self.padfn, self.hashfn)
- signer.update(payload)
- return signer.finalize()
-
- def verify(self, key, payload, signature):
- pkey = key.get_op_key('verify')
- verifier = pkey.verifier(signature, self.padfn, self.hashfn)
- verifier.update(payload)
- verifier.verify()
-
-
-class _RawEC(_RawJWS):
- def __init__(self, curve, hashfn):
- self.curve = curve
- self.hashfn = hashfn
-
- def encode_int(self, n, l):
- e = hex(n).rstrip("L").lstrip("0x")
- ilen = (l + 7) // 8 # number of bytes rounded up
- e = '0' * (ilen * 2 - len(e)) + e # pad as necessary
- return unhexlify(e)
-
- def sign(self, key, payload):
- skey = key.get_op_key('sign', self.curve)
- signer = skey.signer(ec.ECDSA(self.hashfn))
- signer.update(payload)
- signature = signer.finalize()
- r, s = ec_utils.decode_rfc6979_signature(signature)
- l = key.get_curve(self.curve).key_size
- return self.encode_int(r, l) + self.encode_int(s, l)
-
- def verify(self, key, payload, signature):
- pkey = key.get_op_key('verify', self.curve)
- r = signature[:len(signature) // 2]
- s = signature[len(signature) // 2:]
- enc_signature = ec_utils.encode_rfc6979_signature(
- int(hexlify(r), 16), int(hexlify(s), 16))
- verifier = pkey.verifier(enc_signature, ec.ECDSA(self.hashfn))
- verifier.update(payload)
- verifier.verify()
-
-
-class _RawNone(_RawJWS):
-
- def sign(self, key, payload):
- return ''
-
- def verify(self, key, payload, signature):
- if signature != b'':
- raise InvalidJWSSignature('The "none" signature must be the '
- 'empty string')
-
-
class JWSCore(object):
"""The inner JWS Core object.
@@ -215,74 +121,41 @@ class JWSCore(object):
self.key = key
if header is not None:
+ if isinstance(header, dict):
+ self.header = header
+ header = json_encode(header)
+ else:
+ self.header = json_decode(header)
+
self.protected = base64url_encode(header.encode('utf-8'))
else:
+ self.header = dict()
self.protected = ''
- self.payload = base64url_encode(payload)
-
- def _jwa_HS256(self):
- return _RawHMAC(hashes.SHA256())
-
- def _jwa_HS384(self):
- return _RawHMAC(hashes.SHA384())
-
- def _jwa_HS512(self):
- return _RawHMAC(hashes.SHA512())
-
- def _jwa_RS256(self):
- return _RawRSA(padding.PKCS1v15(), hashes.SHA256())
-
- def _jwa_RS384(self):
- return _RawRSA(padding.PKCS1v15(), hashes.SHA384())
-
- def _jwa_RS512(self):
- return _RawRSA(padding.PKCS1v15(), hashes.SHA512())
-
- def _jwa_ES256(self):
- return _RawEC('P-256', hashes.SHA256())
-
- def _jwa_ES384(self):
- return _RawEC('P-384', hashes.SHA384())
-
- def _jwa_ES512(self):
- return _RawEC('P-521', hashes.SHA512())
-
- def _jwa_PS256(self):
- return _RawRSA(padding.PSS(padding.MGF1(hashes.SHA256()),
- hashes.SHA256.digest_size),
- hashes.SHA256())
-
- def _jwa_PS384(self):
- return _RawRSA(padding.PSS(padding.MGF1(hashes.SHA384()),
- hashes.SHA384.digest_size),
- hashes.SHA384())
-
- def _jwa_PS512(self):
- return _RawRSA(padding.PSS(padding.MGF1(hashes.SHA512()),
- hashes.SHA512.digest_size),
- hashes.SHA512())
-
- def _jwa_none(self):
- return _RawNone()
+ self.payload = payload
def _jwa(self, name, allowed):
if allowed is None:
allowed = default_allowed_algs
- attr = '_jwa_%s' % name
- try:
- fn = getattr(self, attr)
- except (KeyError, AttributeError):
- raise InvalidJWAAlgorithm()
if name not in allowed:
raise InvalidJWSOperation('Algorithm not allowed')
- return fn()
+ return JWA.signing_alg(name)
+
+ def _payload(self):
+ if self.header.get('b64', True):
+ return base64url_encode(self.payload).encode('utf-8')
+ else:
+ if isinstance(self.payload, bytes):
+ return self.payload
+ else:
+ return self.payload.encode('utf-8')
def sign(self):
"""Generates a signature"""
- sigin = ('.'.join([self.protected, self.payload])).encode('utf-8')
+ payload = self._payload()
+ sigin = b'.'.join([self.protected.encode('utf-8'), payload])
signature = self.engine.sign(self.key, sigin)
return {'protected': self.protected,
- 'payload': self.payload,
+ 'payload': payload,
'signature': base64url_encode(signature)}
def verify(self, signature):
@@ -291,7 +164,8 @@ class JWSCore(object):
:raises InvalidJWSSignature: if the verification fails.
"""
try:
- sigin = ('.'.join([self.protected, self.payload])).encode('utf-8')
+ payload = self._payload()
+ sigin = b'.'.join([self.protected.encode('utf-8'), payload])
self.engine.verify(self.key, sigin, signature)
except Exception as e: # pylint: disable=broad-except
raise InvalidJWSSignature('Verification failed', repr(e))
@@ -315,16 +189,6 @@ class JWS(object):
self.verifylog = None
self._allowed_algs = None
- def _check_crit(self, crit):
- for k in crit:
- if k not in JWSHeaderRegistry:
- raise InvalidJWSSignature('Unknown critical header: '
- '"%s"' % k)
- else:
- if not JWSHeaderRegistry[k][1]:
- raise InvalidJWSSignature('Unsupported critical '
- 'header: "%s"' % k)
-
@property
def allowed_algs(self):
"""Allowed algorithms.
@@ -348,31 +212,61 @@ class JWS(object):
def is_valid(self):
return self.objects.get('valid', False)
- def _merge_headers(self, h1, h2):
- for k in list(h1.keys()):
- if k in h2:
- raise InvalidJWSObject('Duplicate header: "%s"' % k)
- h1.update(h2)
- return h1
+ # TODO: allow caller to specify list of headers it understands
+ def _merge_check_headers(self, protected, *headers):
+ header = None
+ crit = []
+ if protected is not None:
+ if 'crit' in protected:
+ crit = protected['crit']
+ # Check immediately if we support these critical headers
+ for k in crit:
+ if k not in JWSHeaderRegistry:
+ raise InvalidJWSObject(
+ 'Unknown critical header: "%s"' % k)
+ else:
+ if not JWSHeaderRegistry[k][1]:
+ raise InvalidJWSObject(
+ 'Unsupported critical header: "%s"' % k)
+ header = protected
+ if 'b64' in header:
+ if not isinstance(header['b64'], bool):
+ raise InvalidJWSObject('b64 header must be a boolean')
+
+ for hn in headers:
+ if hn is None:
+ continue
+ if header is None:
+ header = dict()
+ for h in list(hn.keys()):
+ if h in JWSHeaderRegistry:
+ if JWSHeaderRegistry[h].mustprotect:
+ raise InvalidJWSObject('"%s" must be protected' % h)
+ if h in header:
+ raise InvalidJWSObject('Duplicate header: "%s"' % h)
+ header.update(hn)
+
+ for k in crit:
+ if k not in header:
+ raise InvalidJWSObject('Missing critical header "%s"' % k)
+
+ return header
# TODO: support selecting key with 'kid' and passing in multiple keys
def _verify(self, alg, key, payload, signature, protected, header=None):
- # verify it is a valid JSON object and keep a decode copy
+ p = dict()
+ # verify it is a valid JSON object and decode
if protected is not None:
p = json_decode(protected)
- else:
- p = dict()
- if not isinstance(p, dict):
- raise InvalidJWSSignature('Invalid Protected header')
+ if not isinstance(p, dict):
+ raise InvalidJWSSignature('Invalid Protected header')
# merge heders, and verify there are no duplicates
if header:
if not isinstance(header, dict):
raise InvalidJWSSignature('Invalid Unprotected header')
- p = self._merge_headers(p, header)
- # verify critical headers
- # TODO: allow caller to specify list of headers it understands
- if 'crit' in p:
- self._check_crit(p['crit'])
+
+ # Merge and check (critical) headers
+ self._merge_check_headers(p, header)
# check 'alg' is present
if alg is None and 'alg' not in p:
raise InvalidJWSSignature('No "alg" in headers')
@@ -433,6 +327,33 @@ class JWS(object):
raise InvalidJWSSignature('Verification failed for all '
'signatures' + repr(self.verifylog))
+ def _deserialize_signature(self, s):
+ o = dict()
+ o['signature'] = base64url_decode(str(s['signature']))
+ if 'protected' in s:
+ p = base64url_decode(str(s['protected']))
+ o['protected'] = p.decode('utf-8')
+ if 'header' in s:
+ o['header'] = s['header']
+ return o
+
+ def _deserialize_b64(self, o, protected):
+ if protected is None:
+ b64n = None
+ else:
+ p = json_decode(protected)
+ b64n = p.get('b64')
+ if b64n is not None:
+ if not isinstance(b64n, bool):
+ raise InvalidJWSObject('b64 header must be boolean')
+ b64 = o.get('b64')
+ if b64 == b64n:
+ return
+ elif b64 is None:
+ o['b64'] = b64n
+ else:
+ raise InvalidJWSObject('conflicting b64 values')
+
def deserialize(self, raw_jws, key=None, alg=None):
"""Deserialize a JWS token.
@@ -455,25 +376,21 @@ class JWS(object):
try:
try:
djws = json_decode(raw_jws)
- o['payload'] = base64url_decode(str(djws['payload']))
if 'signatures' in djws:
o['signatures'] = list()
for s in djws['signatures']:
- os = dict()
- os['signature'] = base64url_decode(str(s['signature']))
- if 'protected' in s:
- p = base64url_decode(str(s['protected']))
- os['protected'] = p.decode('utf-8')
- if 'header' in s:
- os['header'] = s['header']
+ os = self._deserialize_signature(s)
o['signatures'].append(os)
+ self._deserialize_b64(o, os.get('protected'))
else:
- o['signature'] = base64url_decode(str(djws['signature']))
- if 'protected' in djws:
- p = base64url_decode(str(djws['protected']))
- o['protected'] = p.decode('utf-8')
- if 'header' in djws:
- o['header'] = djws['header']
+ o = self._deserialize_signature(djws)
+ self._deserialize_b64(o, o.get('protected'))
+
+ if 'payload' in djws:
+ if o.get('b64', True):
+ o['payload'] = base64url_decode(str(djws['payload']))
+ else:
+ o['payload'] = djws['payload']
except ValueError:
c = raw_jws.split('.')
@@ -482,6 +399,7 @@ class JWS(object):
p = base64url_decode(str(c[0]))
if len(p) > 0:
o['protected'] = p.decode('utf-8')
+ self._deserialize_b64(o, o['protected'])
o['payload'] = base64url_decode(str(c[1]))
o['signature'] = base64url_decode(str(c[2]))
@@ -504,7 +422,8 @@ class JWS(object):
:param potected: The Protected Header (optional)
:param header: The Unprotected Header (optional)
- :raises InvalidJWSObject: if no payload has been set on the object.
+ :raises InvalidJWSObject: if no payload has been set on the object,
+ or invalid headers are provided.
:raises ValueError: if the key is not a :class:`JWK` object.
:raises ValueError: if the algorithm is missing or is not provided
by one of the headers.
@@ -515,16 +434,36 @@ class JWS(object):
if not self.objects.get('payload', None):
raise InvalidJWSObject('Missing Payload')
+ b64 = True
+
p = dict()
if protected:
- p = json_decode(protected)
- # TODO: allow caller to specify list of headers it understands
- if 'crit' in p:
- self._check_crit(p['crit'])
+ if isinstance(protected, dict):
+ p = protected
+ protected = json_encode(p)
+ else:
+ p = json_decode(protected)
+
+ # If b64 is present we must enforce criticality
+ if 'b64' in list(p.keys()):
+ crit = p.get('crit', [])
+ if 'b64' not in crit:
+ raise InvalidJWSObject('b64 header must always be critical')
+ b64 = p['b64']
+ if 'b64' in self.objects:
+ if b64 != self.objects['b64']:
+ raise InvalidJWSObject('Mixed b64 headers on signatures')
+
+ h = None
if header:
- h = json_decode(header)
- p = self._merge_headers(p, h)
+ if isinstance(header, dict):
+ h = header
+ header = json_encode(header)
+ else:
+ h = json_decode(header)
+
+ p = self._merge_check_headers(p, h)
if 'alg' in p:
if alg is None:
@@ -552,21 +491,18 @@ class JWS(object):
elif 'signature' in self.objects:
self.objects['signatures'] = list()
n = dict()
- n['signature'] = self.objects['signature']
- del self.objects['signature']
+ n['signature'] = self.objects.pop('signature')
if 'protected' in self.objects:
- n['protected'] = self.objects['protected']
- del self.objects['protected']
+ n['protected'] = self.objects.pop('protected')
if 'header' in self.objects:
- n['header'] = self.objects['header']
- del self.objects['header']
+ n['header'] = self.objects.pop('header')
if 'valid' in self.objects:
- n['valid'] = self.objects['valid']
- del self.objects['valid']
+ n['valid'] = self.objects.pop('valid')
self.objects['signatures'].append(n)
self.objects['signatures'].append(o)
else:
self.objects.update(o)
+ self.objects['b64'] = b64
def serialize(self, compact=False):
"""Serializes the object into a JWS token.
@@ -579,7 +515,6 @@ class JWS(object):
:raises InvalidJWSSignature: if no signature has been added
to the object, or no valid signature can be found.
"""
-
if compact:
if 'signatures' in self.objects:
raise InvalidJWSOperation("Can't use compact encoding with "
@@ -592,23 +527,40 @@ class JWS(object):
protected = base64url_encode(self.objects['protected'])
else:
protected = ''
- return '.'.join([protected,
- base64url_encode(self.objects['payload']),
+ if self.objects.get('payload', False):
+ if self.objects.get('b64', True):
+ payload = base64url_encode(self.objects['payload'])
+ else:
+ if isinstance(self.objects['payload'], bytes):
+ payload = self.objects['payload'].decode('utf-8')
+ else:
+ payload = self.objects['payload']
+ if '.' in payload:
+ raise InvalidJWSOperation(
+ "Can't use compact encoding with unencoded "
+ "payload that uses the . character")
+ else:
+ payload = ''
+ return '.'.join([protected, payload,
base64url_encode(self.objects['signature'])])
else:
obj = self.objects
+ sig = dict()
+ if self.objects.get('payload', False):
+ if self.objects.get('b64', True):
+ sig['payload'] = base64url_encode(self.objects['payload'])
+ else:
+ sig['payload'] = self.objects['payload']
if 'signature' in obj:
if not obj.get('valid', False):
raise InvalidJWSSignature("No valid signature found")
- sig = {'payload': base64url_encode(obj['payload']),
- 'signature': base64url_encode(obj['signature'])}
+ sig['signature'] = base64url_encode(obj['signature'])
if 'protected' in obj:
sig['protected'] = base64url_encode(obj['protected'])
if 'header' in obj:
sig['header'] = obj['header']
elif 'signatures' in obj:
- sig = {'payload': base64url_encode(obj['payload']),
- 'signatures': list()}
+ sig['signatures'] = list()
for o in obj['signatures']:
if not o.get('valid', False):
continue
@@ -632,24 +584,27 @@ class JWS(object):
raise InvalidJWSOperation("Payload not verified")
return self.objects['payload']
+ def detach_payload(self):
+ self.objects.pop('payload', None)
+
@property
def jose_header(self):
obj = self.objects
if 'signature' in obj:
- jh = dict()
if 'protected' in obj:
p = json_decode(obj['protected'])
- jh = self._merge_headers(jh, p)
- jh = self._merge_headers(jh, obj.get('header', dict()))
- return jh
+ else:
+ p = None
+ return self._merge_check_headers(p, obj.get('header', dict()))
elif 'signatures' in self.objects:
jhl = list()
for o in obj['signatures']:
jh = dict()
- if 'protected' in obj:
+ if 'protected' in o:
p = json_decode(o['protected'])
- jh = self._merge_headers(jh, p)
- jh = self._merge_headers(jh, o.get('header', dict()))
+ else:
+ p = None
+ jh = self._merge_check_headers(p, o.get('header', dict()))
jhl.append(jh)
return jhl
else:
=====================================
jwcrypto/jwt.py
=====================================
@@ -5,7 +5,7 @@ import uuid
from six import string_types
-from jwcrypto.common import json_decode, json_encode
+from jwcrypto.common import JWException, json_decode, json_encode
from jwcrypto.jwe import JWE
from jwcrypto.jwk import JWK, JWKSet
from jwcrypto.jws import JWS
@@ -22,7 +22,7 @@ JWTClaimsRegistry = {'iss': 'Issuer',
'jti': 'JWT ID'}
-class JWTExpired(Exception):
+class JWTExpired(JWException):
"""Json Web Token is expired.
This exception is raised when a token is expired accoring to its claims.
@@ -39,7 +39,7 @@ class JWTExpired(Exception):
super(JWTExpired, self).__init__(msg)
-class JWTNotYetValid(Exception):
+class JWTNotYetValid(JWException):
"""Json Web Token is not yet valid.
This exception is raised when a token is not valid yet according to its
@@ -57,7 +57,7 @@ class JWTNotYetValid(Exception):
super(JWTNotYetValid, self).__init__(msg)
-class JWTMissingClaim(Exception):
+class JWTMissingClaim(JWException):
"""Json Web Token claim is invalid.
This exception is raised when a claim does not match the expected value.
@@ -74,7 +74,7 @@ class JWTMissingClaim(Exception):
super(JWTMissingClaim, self).__init__(msg)
-class JWTInvalidClaimValue(Exception):
+class JWTInvalidClaimValue(JWException):
"""Json Web Token claim is invalid.
This exception is raised when a claim does not match the expected value.
@@ -91,7 +91,7 @@ class JWTInvalidClaimValue(Exception):
super(JWTInvalidClaimValue, self).__init__(msg)
-class JWTInvalidClaimFormat(Exception):
+class JWTInvalidClaimFormat(JWException):
"""Json Web Token claim format is invalid.
This exception is raised when a claim is not in a valid format.
@@ -108,7 +108,7 @@ class JWTInvalidClaimFormat(Exception):
super(JWTInvalidClaimFormat, self).__init__(msg)
-class JWTMissingKeyID(Exception):
+class JWTMissingKeyID(JWException):
"""Json Web Token is missing key id.
This exception is raised when trying to decode a JWT with a key set
@@ -126,7 +126,7 @@ class JWTMissingKeyID(Exception):
super(JWTMissingKeyID, self).__init__(msg)
-class JWTMissingKey(Exception):
+class JWTMissingKey(JWException):
"""Json Web Token is using a key not in the key set.
This exception is raised if the key that was used is not available
@@ -155,15 +155,15 @@ class JWT(object):
"""Creates a JWT object.
:param header: A dict or a JSON string with the JWT Header data.
- :param claims: A dict or a string withthe JWT Claims data.
+ :param claims: A dict or a string with the JWT Claims data.
:param jwt: a 'raw' JWT token
:param key: A (:class:`jwcrypto.jwk.JWK`) key to deserialize
- the token. A (:class:`jwcrypt.jwk.JWKSet`) can also be used.
+ the token. A (:class:`jwcrypto.jwk.JWKSet`) can also be used.
:param algs: An optional list of allowed algorithms
:param default_claims: An optional dict with default values for
registred claims. A None value for NumericDate type claims
will cause generation according to system time. Only the values
- fro RFC 7519 - 4.1 are evaluated.
+ from RFC 7519 - 4.1 are evaluated.
:param check_claims: An optional dict of claims that must be
present in the token, if the value is not None the claim must
match exactly.
@@ -191,15 +191,15 @@ class JWT(object):
if header:
self.header = header
- if claims:
- self.claims = claims
-
if default_claims is not None:
self._reg_claims = default_claims
if check_claims is not None:
self._check_claims = check_claims
+ if claims:
+ self.claims = claims
+
if jwt is not None:
self.deserialize(jwt, key)
@@ -212,9 +212,15 @@ class JWT(object):
@header.setter
def header(self, h):
if isinstance(h, dict):
- self._header = json_encode(h)
+ eh = json_encode(h)
else:
- self._header = h
+ eh = h
+ h = json_decode(eh)
+
+ if h.get('b64') is False:
+ raise ValueError("b64 header is invalid."
+ "JWTs cannot use unencoded payloads")
+ self._header = eh
@property
def claims(self):
@@ -224,6 +230,10 @@ class JWT(object):
@claims.setter
def claims(self, c):
+ if self._reg_claims and not isinstance(c, dict):
+ # decode c so we can set default claims
+ c = json_decode(c)
+
if isinstance(c, dict):
self._add_default_claims(c)
self._claims = json_encode(c)
@@ -276,7 +286,7 @@ class JWT(object):
def _add_jti_claim(self, claims):
if 'jti' in claims or 'jti' not in self._reg_claims:
return
- claims['jti'] = uuid.uuid4()
+ claims['jti'] = str(uuid.uuid4())
def _add_default_claims(self, claims):
if self._reg_claims is None:
@@ -340,7 +350,7 @@ class JWT(object):
if 'exp' in claims:
self._check_exp(claims['exp'], time.time(), self._leeway)
if 'nbf' in claims:
- self._check_exp(claims['nbf'], time.time(), self._leeway)
+ self._check_nbf(claims['nbf'], time.time(), self._leeway)
def _check_provided_claims(self):
# check_claims can be set to False to skip any check
@@ -380,8 +390,8 @@ class JWT(object):
if value in claims[name]:
continue
raise JWTInvalidClaimValue(
- "Invalid '%s' value. Expected '%s' in '%s'" % (
- name, value, claims[name]))
+ "Invalid '%s' value. Expected '%s' to be in '%s'" % (
+ name, claims[name], value))
elif name == 'exp':
if value is not None:
@@ -398,7 +408,7 @@ class JWT(object):
else:
if value is not None and value != claims[name]:
raise JWTInvalidClaimValue(
- "Invalid '%s' value. Expected '%d' got '%d'" % (
+ "Invalid '%s' value. Expected '%s' got '%s'" % (
name, value, claims[name]))
def make_signed_token(self, key):
@@ -437,7 +447,7 @@ class JWT(object):
:param jwt: a 'raw' JWT token.
:param key: A (:class:`jwcrypto.jwk.JWK`) verification or
- decryption key, or a (:class:`jwcrypt.jwk.JWKSet`) that
+ decryption key, or a (:class:`jwcrypto.jwk.JWKSet`) that
contains a key indexed by the 'kid' header.
"""
c = jwt.count('.')
=====================================
jwcrypto/tests.py
=====================================
@@ -3,13 +3,13 @@
from __future__ import unicode_literals
import copy
-
import unittest
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.asymmetric import rsa
+from jwcrypto import jwa
from jwcrypto import jwe
from jwcrypto import jwk
from jwcrypto import jws
@@ -170,6 +170,70 @@ RSAPrivateKey = {"kty": "RSA",
"eAvmj4sm-Fp0oYu_neotgQ0hzbI5gry7ajdYy9-2lNx_76aBZoOUu"
"9HCJ-UsfSOI8"}
+# From
+# vectors/cryptography_vectors/asymmetric/PEM_Serialization/rsa_private_key.pem
+RSAPrivatePEM = b"""-----BEGIN RSA PRIVATE KEY-----
+Proc-Type: 4,ENCRYPTED
+DEK-Info: DES-EDE3-CBC,B4B3C8C536E57CBE
+
+B8Lq1K/wcOr4JMspWrX3zCX14WAp3xgHsKAB4XfuCuju/HQZoWXtok1xoi5e2Ovw
+ENA99Jvb2yvBdDUfOlp1L1L+By3q+SwcdeNuEKjwGFG6MY2uZaVtLSiAFXf1N8PL
+id7FMRGPIxpTtXKMhfAq4luRb0BgKh7+ZvM7LkxkRxF7M1XVQPGhrU0OfxX9VODe
+YFH1q47os5JzHRcrRaFx6sn30e79ij2gRjzMVFuAX07n+yw3qeyNQNYdmDNP7iCZ
+x//0iN0NboTI81coNlxx7TL4bYwgESt1c2i/TCfLITjKgEny7MKqU1/jTrOJWu85
+PiK/ojaD1EMx9xxVgBCioQJVG/Jm9y+XhtGFAJUShzzsabX7KuANKRn3fgUN+yZS
+yp8hmD+R5gQHJk/8+zZ6/Imv8W/G+7fPZuSMgWeWtDReCkfzgnyIdjaIp3Pdp5yN
+WLLWADI4tHmNUqIzY7T25gVfg0P2tgQNzn3WzHxq4SfZN9Aw57woi8eSRpLBEn+C
+JjqwTxtFQ14ynG6GPsBaDcAduchmJPL7e9PuAfFyLJuM8sU8QyB2oir1M/qYFhTC
+ClXw2yylYjAy8TFw1L3UZA4hfAflINjYUY8pgAtTAjxeD/9PhiKSoMEX8Q/8Npti
+1Db5RpAClIEdB6nPywj6BzC+6El3dSGaCV0sTQ42LD+S3QH8VCwTB2AuKq7zyuD6
+wEQopcbIOGHSir875vYLmWLmqR9MCWZtKj/dWfTIQpBsPsI2ssZn/MptNqyEN9TW
+GfnWoTuzoziCS5YmEq7Mh98fwP9Krb0abo3fFvu6CY3dhvvoxPaXahyAxBfpKArB
+9nOf3gzHGReWNiFUtNZlvueYrC5CnblFzKaKB+81Imjw6RXM3QtuzbZ71zp+reL8
+JeiwE/mriwuGbxTJx5gwQX48zA5PJ342CCrl7jMeIos5KXmYkWoU5hEuGM3tK4Lx
+VAoGqcd/a4cWHuLWub8fbhFkIDcxFaMF8yQi0r2LOmvMOsv3RVpyfgJ07z5b9X1B
+w76CYkjGqgr0EdU40VTPtNhtHq7rrJSzGbapRsFUpvqgnkEwUSdbY6bRknLETmfo
+H3dPf2XQwXXPDMZTW54QsmQ9WjundqOFI2YsH6dCX/kmZK0IJVBpikL8SuM/ZJLK
+LcYJcrNGetENEKKl6hDwTTIsG1y3gx6y3wPzBkyJ2DtMx9dPoCqYhPHsIGc/td0/
+r4Ix9TWVLIl3MKq3z+/Hszd7jOnrkflfmKeA0DgJlqVJsuxP75pbdiKS/hCKRf8D
+AFJLvt6JSGBnz9ZZCB4KrjpHK/k+X7p8Y65uc/aX5BLu8vyRqFduhg98GVXJmD7k
+0ggXnqqFnies6SpnQ45cjfKSGDx/NjY0AwoGPH8n8CL6ZagU6K1utfHIMrqKkJME
+F6KcPHWrQkECojLdMoDInnRirdRb/FcAadWBSPrf+6Nln4ilbBJIi8W/yzeM/WFj
+UKKNjk4W26PGnNO6+TO5h1EpocDI4fx6UYIMmFjnyaLdLrSn1/SzuLL6I7pYZ0Um
+8qI4aWjP9RiUvGYJirfAUjL5Vp9w4+osf1sGiioe0GH/1WVuHeQ93A==
+-----END RSA PRIVATE KEY-----
+"""
+
+RSAPrivatePassword = b"123456"
+
+# From
+# vectors/cryptography_vectors/asymmetric/PEM_Serialization/rsa_public_key.pem
+RSAPublicPEM = b"""-----BEGIN PUBLIC KEY-----
+MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAnR4AZ+tgWYql+S3MaTQ6
+zeIO1fKzFIoau9Q0zGuv/1oCAewXwxeDSSxw+/Z3GL1NpuuS9CpbR5EQ3d71bD0v
+0G+Sf+mShSl0oljG7YqnNSPzKl+EQ3/KE+eEButcwas6KGof2BA4bFNCw/fPbuhk
+u/d8sIIEgdzBMiGRMdW33uci3rsdOenMZQA7uWsM/q/pu85YLAVOxq6wlUCzP4FM
+Tw/RKzayrPkn3Jfbqcy1aM2HDlFVx24vaN+RRbPSnVoQbo5EQYkUMXE8WmadSyHl
+pXGRnWsJSV9AdGyDrbU+6tcFwcIwnW22jb/OJy8swHdqKGkuR1kQ0XqokK1yGKFZ
+8wIDAQAB
+-----END PUBLIC KEY-----
+"""
+
+# From cryptography/vectors/cryptography_vectors/x509/v1_cert.pem
+PublicCert = b"""-----BEGIN CERTIFICATE-----
+MIIBWzCCAQYCARgwDQYJKoZIhvcNAQEEBQAwODELMAkGA1UEBhMCQVUxDDAKBgNV
+BAgTA1FMRDEbMBkGA1UEAxMSU1NMZWF5L3JzYSB0ZXN0IENBMB4XDTk1MDYxOTIz
+MzMxMloXDTk1MDcxNzIzMzMxMlowOjELMAkGA1UEBhMCQVUxDDAKBgNVBAgTA1FM
+RDEdMBsGA1UEAxMUU1NMZWF5L3JzYSB0ZXN0IGNlcnQwXDANBgkqhkiG9w0BAQEF
+AANLADBIAkEAqtt6qS5GTxVxGZYWa0/4u+IwHf7p2LNZbcPBp9/OfIcYAXBQn8hO
+/Re1uwLKXdCjIoaGs4DLdG88rkzfyK5dPQIDAQABMAwGCCqGSIb3DQIFBQADQQAE
+Wc7EcF8po2/ZO6kNCwK/ICH6DobgLekA5lSLr5EvuioZniZp5lFzAw4+YzPQ7XKJ
+zl9HYIMxATFyqSiD9jsx
+-----END CERTIFICATE-----
+"""
+
+PublicCertThumbprint = u'7KITkGJF74IZ9NKVvHfuJILbuIZny6j-roaNjB1vgiA'
+
class TestJWK(unittest.TestCase):
def test_create_pubKeys(self):
@@ -198,8 +262,12 @@ class TestJWK(unittest.TestCase):
jwk.JWK.generate(kty='oct', size=256)
jwk.JWK.generate(kty='RSA', size=4096)
jwk.JWK.generate(kty='EC', curve='P-521')
- k = jwk.JWK.generate(kty='oct', size=256, kid='MySymmetricKey')
+ k = jwk.JWK.generate(kty='oct', alg='A192KW', kid='MySymmetricKey')
self.assertEqual(k.key_id, 'MySymmetricKey')
+ self.assertEqual(len(base64url_decode(k.get_op_key('encrypt'))), 24)
+ jwk.JWK.generate(kty='RSA', alg='RS256')
+ k = jwk.JWK.generate(kty='RSA', size=4096, alg='RS256')
+ self.assertEqual(k.get_op_key('encrypt').key_size, 4096)
def test_export_public_keys(self):
k = jwk.JWK(**RSAPrivateKey)
@@ -243,11 +311,17 @@ class TestJWK(unittest.TestCase):
self.assertRaises(jwk.InvalidJWKValue,
jwk.JWK.from_pyca, dict())
+ def test_jwk_from_json(self):
+ k = jwk.JWK.generate(kty='oct', size=256)
+ y = jwk.JWK.from_json(k.export())
+ self.assertEqual(k.export(), y.export())
+
def test_jwkset(self):
k = jwk.JWK(**RSAPrivateKey)
ks = jwk.JWKSet()
ks.add(k)
- ks2 = jwk.JWKSet().import_keyset(ks.export())
+ ks2 = jwk.JWKSet()
+ ks2.import_keyset(ks.export())
self.assertEqual(len(ks), len(ks2))
self.assertEqual(len(ks), 1)
k1 = ks.get_key(RSAPrivateKey['kid'])
@@ -260,6 +334,15 @@ class TestJWK(unittest.TestCase):
ks3 = jwk.JWKSet.from_json(ks.export())
self.assertEqual(len(ks), len(ks3))
+ # Test Keyset with mutiple keys
+ ksm = jwk.JWKSet.from_json(json_encode(PrivateKeys))
+ num = 0
+ for item in ksm:
+ self.assertTrue(isinstance(item, jwk.JWK))
+ self.assertTrue(item in ksm)
+ num += 1
+ self.assertEqual(num, len(PrivateKeys['keys']))
+
def test_thumbprint(self):
for i in range(0, len(PublicKeys['keys'])):
k = jwk.JWK(**PublicKeys['keys'][i])
@@ -267,6 +350,67 @@ class TestJWK(unittest.TestCase):
k.thumbprint(),
PublicKeys['thumbprints'][i])
+ def test_import_from_pem(self):
+ pubk = jwk.JWK.from_pem(RSAPublicPEM)
+ self.assertEqual(pubk.export_to_pem(), RSAPublicPEM)
+ rsapub = pubk.get_op_key('verify')
+
+ prik = jwk.JWK.from_pem(RSAPrivatePEM, password=RSAPrivatePassword)
+ rsapri = prik.get_op_key('sign')
+ self.assertEqual(rsapri.public_key().public_numbers().n,
+ rsapub.public_numbers().n)
+
+ pubc = jwk.JWK.from_pem(PublicCert)
+ self.assertEqual(pubc.key_id, PublicCertThumbprint)
+
+ def test_export_symmetric(self):
+ key = jwk.JWK(**SymmetricKeys['keys'][0])
+ self.assertTrue(key.is_symmetric)
+ self.assertFalse(key.has_public)
+ self.assertFalse(key.has_private)
+ self.assertEqual(json_encode(SymmetricKeys['keys'][0]),
+ key.export_symmetric())
+
+ def test_export_public(self):
+ key = jwk.JWK.from_pem(PublicCert)
+ self.assertFalse(key.is_symmetric)
+ self.assertTrue(key.has_public)
+ self.assertFalse(key.has_private)
+ pubc = key.export_public()
+ self.assertEqual(json_decode(pubc)["kid"], PublicCertThumbprint)
+
+ def test_export_private(self):
+ key = jwk.JWK.from_pem(RSAPrivatePEM, password=RSAPrivatePassword)
+ self.assertFalse(key.is_symmetric)
+ self.assertTrue(key.has_public)
+ self.assertTrue(key.has_private)
+ pri = key.export_private()
+ prikey = jwk.JWK(**json_decode(pri))
+ self.assertTrue(prikey.has_private)
+ pub = key.export_public()
+ pubkey = jwk.JWK(**json_decode(pub))
+ self.assertFalse(pubkey.has_private)
+ self.assertEqual(prikey.key_id, pubkey.key_id)
+
+ def test_public(self):
+ key = jwk.JWK.from_pem(RSAPrivatePEM, password=RSAPrivatePassword)
+ self.assertTrue(key.has_public)
+ self.assertTrue(key.has_private)
+ pubkey = key.public()
+ self.assertTrue(pubkey.has_public)
+ self.assertFalse(pubkey.has_private)
+ # finally check public works
+ e = jwe.JWE('plaintext', '{"alg":"RSA-OAEP","enc":"A256GCM"}')
+ e.add_recipient(pubkey)
+ enc = e.serialize()
+ d = jwe.JWE()
+ d.deserialize(enc, key)
+ self.assertEqual(d.payload, b'plaintext')
+
+ def test_invalid_value(self):
+ with self.assertRaises(jwk.InvalidJWKValue):
+ jwk.JWK(kty='oct', k=b'\x01')
+
# RFC 7515 - A.1
A1_protected = \
@@ -445,7 +589,11 @@ A6_example = {
'key2': jwk.JWK(**A3_key),
'protected2': bytes(bytearray(A3_protected)).decode('utf-8'),
'header2': json_encode({"kid": "e9bc097a-ce51-4036-9562-d2ade882db0d"}),
- 'serialized': A6_serialized}
+ 'serialized': A6_serialized,
+ 'jose_header': [{"kid": "2010-12-29",
+ "alg": "RS256"},
+ {"kid": "e9bc097a-ce51-4036-9562-d2ade882db0d",
+ "alg": "ES256"}]}
A7_example = \
'{' + \
@@ -503,7 +651,8 @@ class TestJWS(unittest.TestCase):
self.check_sign, A5_example)
a5_bis = {'allowed_algs': ['none']}
a5_bis.update(A5_example)
- self.check_sign(a5_bis)
+ with self.assertRaises(jws.InvalidJWSSignature):
+ self.check_sign(a5_bis)
def test_A6(self):
s = jws.JWS(A6_example['payload'])
@@ -518,6 +667,7 @@ class TestJWS(unittest.TestCase):
sig = s.serialize()
s.deserialize(sig, A6_example['key1'])
s.deserialize(A6_serialized, A6_example['key2'])
+ self.assertEqual(A6_example['jose_header'], s.jose_header)
def test_A7(self):
s = jws.JWS(A6_example['payload'])
@@ -689,6 +839,29 @@ E_A5_ex = \
'"ciphertext":"KDlTtXchhZTGufMYmOYGS4HffxPSUrfmqCHXaI9wOGY",' \
'"tag":"Mz-VPPyU4RlcuYv1IwIvzw"}'
+Issue_136_Protected_Header_no_epk = {
+ "alg": "ECDH-ES+A256KW",
+ "enc": "A256CBC-HS512"}
+
+Issue_136_Contributed_JWE = \
+ "eyJhbGciOiJFQ0RILUVTK0ExMjhLVyIsImVuYyI6IkEyNTZDQkMtSFM1MTIiLCJr" \
+ "aWQiOiJrZXkxIiwiZXBrIjp7Imt0eSI6IkVDIiwiY3J2IjoiUC0yNTYiLCJ4Ijoi" \
+ "cDNpU241cEFSNUpYUE5aVF9SSEw2MTJMUGliWEI2WDhvTE9EOXFrN2NhTSIsInki" \
+ "OiI1Y04yQ2FqeXM3SVlDSXFEby1QUHF2bVQ1RzFvMEEtU0JicEQ5NFBOb3NNIn19" \
+ ".wG51hYE_Vma8tvFKVyeZs4lsHhXiarEw3-59eWHPmhRflDAKrMvnBw1urezo_Bz" \
+ "ZyPJ76m42ORQPbhEu5NvbJk3vgdgcp03j" \
+ ".lRttW8r6P6zM0uYDQt0EjQ.qnOnz7biCbqdLEdUH3acMamFm-cBRCSTFb83tNPrgDU" \
+ ".vZnwYpYjzrTaYritwMzaguaAMsq9rQOWe8NUHICv2hg"
+
+Issue_136_Contributed_Key = {
+ "alg": "ECDH-ES+A128KW",
+ "crv": "P-256",
+ "d": "F2PnliYin65AoIUxL1CwwzBPNeL2TyZPAKtkXOP50l8",
+ "kid": "key1",
+ "kty": "EC",
+ "x": "FPrb_xwxe8SBP3kO-e-WsofFp7n5-yc_tGgfAvqAP8g",
+ "y": "lM3HuyKMYUVsYdGqiWlkwTZbGO3Fh-hyadq8lfkTgBc"}
+
class TestJWE(unittest.TestCase):
def check_enc(self, plaintext, protected, key, vector):
@@ -731,6 +904,121 @@ class TestJWE(unittest.TestCase):
e = jwe.JWE(algs=['A256KW'])
e.deserialize(E_A5_ex, E_A4_ex['key2'])
+ def test_compact_protected_header(self):
+ """Compact representation requires a protected header"""
+ e = jwe.JWE(E_A1_ex['plaintext'])
+ e.add_recipient(E_A1_ex['key'], E_A1_ex['protected'])
+
+ with self.assertRaises(jwe.InvalidJWEOperation):
+ e.serialize(compact=True)
+
+ def test_compact_invalid_header(self):
+ with self.assertRaises(jwe.InvalidJWEOperation):
+ e = jwe.JWE(E_A1_ex['plaintext'], E_A1_ex['protected'],
+ aad='XYZ', recipient=E_A1_ex['key'])
+ e.serialize(compact=True)
+
+ with self.assertRaises(jwe.InvalidJWEOperation):
+ e = jwe.JWE(E_A1_ex['plaintext'], E_A1_ex['protected'],
+ unprotected='{"jku":"https://example.com/keys.jwks"}',
+ recipient=E_A1_ex['key'])
+ e.serialize(compact=True)
+
+ def test_JWE_Issue_136(self):
+ plaintext = "plain"
+ protected = json_encode(Issue_136_Protected_Header_no_epk)
+ key = jwk.JWK.generate(kty='EC', crv='P-521')
+ e = jwe.JWE(plaintext, protected)
+ e.add_recipient(key)
+ enc = e.serialize()
+ e.deserialize(enc, key)
+ self.assertEqual(e.payload, plaintext.encode('utf-8'))
+
+ e = jwe.JWE()
+ e.deserialize(Issue_136_Contributed_JWE,
+ jwk.JWK(**Issue_136_Contributed_Key))
+
+
+MMA_vector_key = jwk.JWK(**E_A2_key)
+MMA_vector_ok_cek = \
+ '{"protected":"eyJlbmMiOiJBMTI4Q0JDLUhTMjU2In0",' \
+ '"unprotected":{"jku":"https://server.example.com/keys.jwks"},' \
+ '"recipients":[' \
+ '{"header":{"alg":"RSA1_5","kid":"2011-04-29"},' \
+ '"encrypted_key":'\
+ '"UGhIOguC7IuEvf_NPVaXsGMoLOmwvc1GyqlIKOK1nN94nHPoltGRhWhw7Zx0-' \
+ 'kFm1NJn8LE9XShH59_i8J0PH5ZZyNfGy2xGdULU7sHNF6Gp2vPLgNZ__deLKx' \
+ 'GHZ7PcHALUzoOegEI-8E66jX2E4zyJKx-YxzZIItRzC5hlRirb6Y5Cl_p-ko3' \
+ 'YvkkysZIFNPccxRU7qve1WYPxqbb2Yw8kZqa2rMWI5ng8OtvzlV7elprCbuPh' \
+ 'cCdZ6XDP0_F8rkXds2vE4X-ncOIM8hAYHHi29NX0mcKiRaD0-D-ljQTP-cFPg' \
+ 'wCp6X-nZZd9OHBv-B3oWh2TbqmScqXMR4gp_A"}],' \
+ '"iv":"AxY8DCtDaGlsbGljb3RoZQ",' \
+ '"ciphertext":"PURPOSEFULLYBROKENYGS4HffxPSUrfmqCHXaI9wOGY",' \
+ '"tag":"Mz-VPPyU4RlcuYv1IwIvzw"}'
+MMA_vector_ko_cek = \
+ '{"protected":"eyJlbmMiOiJBMTI4Q0JDLUhTMjU2In0",' \
+ '"unprotected":{"jku":"https://server.example.com/keys.jwks"},' \
+ '"recipients":[' \
+ '{"header":{"alg":"RSA1_5","kid":"2011-04-29"},' \
+ '"encrypted_key":'\
+ '"UGhIOguC7IuEvf_NPVaYsGMoLOmwvc1GyqlIKOK1nN94nHPoltGRhWhw7Zx0-' \
+ 'kFm1NJn8LE9XShH59_i8J0PH5ZZyNfGy2xGdULU7sHNF6Gp2vPLgNZ__deLKx' \
+ 'GHZ7PcHALUzoOegEI-8E66jX2E4zyJKx-YxzZIItRzC5hlRirb6Y5Cl_p-ko3' \
+ 'YvkkysZIFNPccxRU7qve1WYPxqbb2Yw8kZqa2rMWI5ng8OtvzlV7elprCbuPh' \
+ 'cCdZ6XDP0_F8rkXds2vE4X-ncOIM8hAYHHi29NX0mcKiRaD0-D-ljQTP-cFPg' \
+ 'wCp6X-nZZd9OHBv-B3oWh2TbqmScqXMR4gp_A"}],' \
+ '"iv":"AxY8DCtDaGlsbGljb3RoZQ",' \
+ '"ciphertext":"PURPOSEFULLYBROKENYGS4HffxPSUrfmqCHXaI9wOGY",' \
+ '"tag":"Mz-VPPyU4RlcuYv1IwIvzw"}'
+
+
+class TestMMA(unittest.TestCase):
+ @classmethod
+ def setUpClass(cls):
+ import os
+ cls.enableMMA = os.environ.get('JWCRYPTO_TESTS_ENABLE_MMA', False)
+ cls.iterations = 500
+ cls.sub_iterations = 100
+
+ def test_MMA(self):
+ if self.enableMMA:
+
+ print('Testing MMA timing attacks')
+
+ ok_cek = 0
+ ok_e = jwe.JWE()
+ ok_e.deserialize(MMA_vector_ok_cek)
+ ko_cek = 0
+ ko_e = jwe.JWE()
+ ko_e.deserialize(MMA_vector_ko_cek)
+
+ import time
+ counter = getattr(time, 'perf_counter', time.time)
+
+ for _ in range(self.iterations):
+ start = counter()
+ for _ in range(self.sub_iterations):
+ with self.assertRaises(jwe.InvalidJWEData):
+ ok_e.decrypt(MMA_vector_key)
+ stop = counter()
+ ok_cek += (stop - start) / self.sub_iterations
+
+ start = counter()
+ for _ in range(self.sub_iterations):
+ with self.assertRaises(jwe.InvalidJWEData):
+ ko_e.decrypt(MMA_vector_key)
+ stop = counter()
+ ko_cek += (stop - start) / self.sub_iterations
+
+ ok_cek /= self.iterations
+ ko_cek /= self.iterations
+
+ deviation = ((ok_cek - ko_cek) / ok_cek) * 100
+ print('MMA ok cek: {}'.format(ok_cek))
+ print('MMA ko cek: {}'.format(ko_cek))
+ print('MMA deviation: {}% ({})'.format(int(deviation), deviation))
+ self.assertLess(deviation, 2)
+
# RFC 7519
A1_header = {
@@ -825,6 +1113,39 @@ class TestJWT(unittest.TestCase):
keyset.add(key)
jwt.JWT(jwt=token, key=keyset, check_claims={'exp': 1300819380})
+ def test_invalid_claim_type(self):
+ key = jwk.JWK(**E_A2_key)
+ claims = {"testclaim": "test"}
+ claims.update(A1_claims)
+ t = jwt.JWT(A1_header, claims)
+ t.make_encrypted_token(key)
+ token = t.serialize()
+
+ # Wrong string
+ self.assertRaises(jwt.JWTInvalidClaimValue, jwt.JWT, jwt=token,
+ key=key, check_claims={"testclaim": "ijgi"})
+
+ # Wrong type
+ self.assertRaises(jwt.JWTInvalidClaimValue, jwt.JWT, jwt=token,
+ key=key, check_claims={"testclaim": 123})
+
+ # Correct
+ jwt.JWT(jwt=token, key=key, check_claims={"testclaim": "test"})
+
+ def test_claim_params(self):
+ key = jwk.JWK(**E_A2_key)
+ default_claims = {"iss": "test", "exp": None}
+ string_claims = '{"string_claim":"test"}'
+ string_header = '{"alg":"RSA1_5","enc":"A128CBC-HS256"}'
+ t = jwt.JWT(string_header, string_claims,
+ default_claims=default_claims)
+ t.make_encrypted_token(key)
+ token = t.serialize()
+
+ # Check default_claims
+ jwt.JWT(jwt=token, key=key, check_claims={"iss": "test", "exp": None,
+ "string_claim": "test"})
+
class ConformanceTests(unittest.TestCase):
@@ -883,3 +1204,129 @@ class ConformanceTests(unittest.TestCase):
check.deserialize(o, jwk.JWK(kty='oct', k=base64url_encode(b'A' * 16)),
alg="HS512")
self.assertTrue(check.objects['valid'])
+
+ def test_jws_headers_as_dicts(self):
+ sign = jws.JWS(payload='message')
+ key = jwk.JWK(kty='oct', k=base64url_encode(b'A' * 16))
+ sign.add_signature(key, protected={'alg': 'HS512'},
+ header={'kid': key.thumbprint()})
+ o = sign.serialize()
+ check = jws.JWS()
+ check.deserialize(o, key, alg="HS512")
+ self.assertTrue(check.objects['valid'])
+ self.assertEqual(check.jose_header['kid'], key.thumbprint())
+
+ def test_jwe_headers_as_dicts(self):
+ enc = jwe.JWE(plaintext='message',
+ protected={"alg": "A256KW", "enc": "A256CBC-HS512"})
+ key = jwk.JWK(kty='oct', k=base64url_encode(b'A' * 32))
+ enc.add_recipient(key, {'kid': key.thumbprint()})
+ o = enc.serialize()
+ check = jwe.JWE()
+ check.deserialize(o)
+ check.decrypt(key)
+ self.assertEqual(check.payload, b'message')
+ self.assertEqual(
+ json_decode(check.objects['header'])['kid'], key.thumbprint())
+
+ def test_jwe_default_recipient(self):
+ key = jwk.JWK(kty='oct', k=base64url_encode(b'A' * (128 // 8)))
+ enc = jwe.JWE(plaintext='plain',
+ protected='{"alg":"A128KW","enc":"A128GCM"}',
+ recipient=key).serialize()
+ check = jwe.JWE()
+ check.deserialize(enc, key)
+ self.assertEqual(b'plain', check.payload)
+
+
+class JWATests(unittest.TestCase):
+ def test_jwa_create(self):
+ for name, cls in jwa.JWA.algorithms_registry.items():
+ self.assertEqual(cls.name, name)
+ self.assertIn(cls.algorithm_usage_location, {'alg', 'enc'})
+ if name == 'ECDH-ES':
+ self.assertIs(cls.keysize, None)
+ else:
+ self.assertIsInstance(cls.keysize, int)
+ self.assertGreaterEqual(cls.keysize, 0)
+
+ if cls.algorithm_use == 'sig':
+ with self.assertRaises(jwa.InvalidJWAAlgorithm):
+ jwa.JWA.encryption_alg(name)
+ with self.assertRaises(jwa.InvalidJWAAlgorithm):
+ jwa.JWA.keymgmt_alg(name)
+ inst = jwa.JWA.signing_alg(name)
+ self.assertIsInstance(inst, jwa.JWAAlgorithm)
+ self.assertEqual(inst.name, name)
+ elif cls.algorithm_use == 'kex':
+ with self.assertRaises(jwa.InvalidJWAAlgorithm):
+ jwa.JWA.encryption_alg(name)
+ with self.assertRaises(jwa.InvalidJWAAlgorithm):
+ jwa.JWA.signing_alg(name)
+ inst = jwa.JWA.keymgmt_alg(name)
+ self.assertIsInstance(inst, jwa.JWAAlgorithm)
+ self.assertEqual(inst.name, name)
+ elif cls.algorithm_use == 'enc':
+ with self.assertRaises(jwa.InvalidJWAAlgorithm):
+ jwa.JWA.signing_alg(name)
+ with self.assertRaises(jwa.InvalidJWAAlgorithm):
+ jwa.JWA.keymgmt_alg(name)
+ inst = jwa.JWA.encryption_alg(name)
+ self.assertIsInstance(inst, jwa.JWAAlgorithm)
+ self.assertEqual(inst.name, name)
+ else:
+ self.fail((name, cls))
+
+
+# RFC 7797
+
+rfc7797_e_header = '{"alg":"HS256"}'
+rfc7797_u_header = '{"alg":"HS256","b64":false,"crit":["b64"]}'
+rfc7797_payload = "$.02"
+
+
+class TestUnencodedPayload(unittest.TestCase):
+
+ def test_regular(self):
+ result = \
+ 'eyJhbGciOiJIUzI1NiJ9.JC4wMg.' + \
+ '5mvfOroL-g7HyqJoozehmsaqmvTYGEq5jTI1gVvoEoQ'
+
+ s = jws.JWS(rfc7797_payload)
+ s.add_signature(jwk.JWK(**SymmetricKeys['keys'][1]),
+ protected=rfc7797_e_header)
+ sig = s.serialize(compact=True)
+ self.assertEqual(sig, result)
+
+ def test_compat_unencoded(self):
+ result = \
+ 'eyJhbGciOiJIUzI1NiIsImI2NCI6ZmFsc2UsImNyaXQiOlsiYjY0Il19..' + \
+ 'A5dxf2s96_n5FLueVuW1Z_vh161FwXZC4YLPff6dmDY'
+
+ s = jws.JWS(rfc7797_payload)
+ s.add_signature(jwk.JWK(**SymmetricKeys['keys'][1]),
+ protected=rfc7797_u_header)
+ # check unencoded payload is in serialized form
+ sig = s.serialize()
+ self.assertEqual(json_decode(sig)['payload'], rfc7797_payload)
+ # check error raises if we try to get compact serialization
+ with self.assertRaises(jws.InvalidJWSOperation):
+ sig = s.serialize(compact=True)
+ # check compact serialization is allowed with detached payload
+ s.detach_payload()
+ sig = s.serialize(compact=True)
+ self.assertEqual(sig, result)
+
+ def test_misses_crit(self):
+ s = jws.JWS(rfc7797_payload)
+ with self.assertRaises(jws.InvalidJWSObject):
+ s.add_signature(jwk.JWK(**SymmetricKeys['keys'][1]),
+ protected={"alg": "HS256", "b64": False})
+
+ def test_mismatching_encoding(self):
+ s = jws.JWS(rfc7797_payload)
+ s.add_signature(jwk.JWK(**SymmetricKeys['keys'][0]),
+ protected=rfc7797_e_header)
+ with self.assertRaises(jws.InvalidJWSObject):
+ s.add_signature(jwk.JWK(**SymmetricKeys['keys'][1]),
+ protected=rfc7797_u_header)
=====================================
requirements.txt deleted
=====================================
@@ -1 +0,0 @@
-cryptography >= 0.7.2
=====================================
setup.cfg
=====================================
@@ -0,0 +1,6 @@
+[bdist_wheel]
+universal = 1
+
+[aliases]
+packages = clean --all egg_info bdist_wheel sdist --format=zip sdist --format=gztar
+release = packages register upload
=====================================
setup.py
=====================================
@@ -6,7 +6,7 @@ from setuptools import setup
setup(
name = 'jwcrypto',
- version = '0.3.1',
+ version = '0.6.0',
license = 'LGPLv3+',
maintainer = 'JWCrypto Project Contributors',
maintainer_email = 'simo at redhat.com',
@@ -17,10 +17,14 @@ setup(
'Programming Language :: Python :: 2.7',
'Programming Language :: Python :: 3.4',
'Programming Language :: Python :: 3.5',
+ 'Programming Language :: Python :: 3.6',
+ 'Programming Language :: Python :: 3.7',
'Intended Audience :: Developers',
'Topic :: Security',
'Topic :: Software Development :: Libraries :: Python Modules'
],
data_files = [('share/doc/jwcrypto', ['LICENSE', 'README.md'])],
+ install_requires = [
+ 'cryptography >= 1.5',
+ ],
)
-
=====================================
tox.ini
=====================================
@@ -1,5 +1,5 @@
[tox]
-envlist = lint,py27,py34,py35,pep8py2,pep8py3,doc,sphinx
+envlist = lint,py27,py34,py35,py36,py37,pep8py2,pep8py3,doc,sphinx
skip_missing_interpreters = true
[testenv]
@@ -8,17 +8,15 @@ setenv =
deps =
pytest
coverage
- -r{toxinidir}/requirements.txt
sitepackages = True
commands =
- {envpython} -m coverage run -m pytest --capture=no --strict {posargs}
+ {envpython} -bb -m coverage run -m pytest --capture=no --strict {posargs}
{envpython} -m coverage report -m
[testenv:lint]
basepython = python2.7
deps =
pylint
- -r{toxinidir}/requirements.txt
sitepackages = True
commands =
{envpython} -m pylint -d c,r,i,W0613 -r n -f colorized --notes= --disable=star-args ./jwcrypto
@@ -49,7 +47,6 @@ deps =
basepython = python2.7
commands =
doc8 --allow-long-titles README.md
- python setup.py check --restructuredtext --metadata --strict
markdown_py README.md -f {toxworkdir}/README.md.html
[testenv:sphinx]
@@ -57,7 +54,6 @@ basepython = python2.7
changedir = docs/source
deps =
sphinx < 1.3.0
- -r{toxinidir}/requirements.txt
commands =
sphinx-build -v -W -b html -d {envtmpdir}/doctrees . {envtmpdir}/html
View it on GitLab: https://salsa.debian.org/freeipa-team/python-jwcrypto/compare/b81d7ea6459b2ec42da519ba19f983e2f4416fbb...5d991eb808571ab463694ee1ec31b22c8b753c34
--
View it on GitLab: https://salsa.debian.org/freeipa-team/python-jwcrypto/compare/b81d7ea6459b2ec42da519ba19f983e2f4416fbb...5d991eb808571ab463694ee1ec31b22c8b753c34
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/20190402/7f26e540/attachment-0001.html>
More information about the Pkg-freeipa-devel
mailing list