Import python-jwcrypto_0.6.0.orig.tar.gz

[dgit import orig python-jwcrypto_0.6.0.orig.tar.gz]
This commit is contained in:
Timo Aaltonen 2019-04-02 08:05:15 +02:00
commit cbdf4089fb
27 changed files with 7464 additions and 0 deletions

23
.coveragerc Normal file
View File

@ -0,0 +1,23 @@
[run]
branch = True
source =
jwcrypto
tests
[paths]
source =
jwcrypto
.tox/*/lib/python*/site-packages/jwcrypto
[report]
ignore_errors = False
precision = 1
exclude_lines =
pragma: no cover
raise AssertionError
raise NotImplementedError
if 0:
if False:
if __name__ == .__main__.:
if PY3
if not PY3

8
.gitignore vendored Normal file
View File

@ -0,0 +1,8 @@
build/
dist/
*.pyc
*.pyo
cscope.out
.tox
.coverage
*.egg-info

35
.travis.yml Normal file
View File

@ -0,0 +1,35 @@
sudo: false
language: python
cache: pip
matrix:
include:
- python: 2.7
env: TOXENV=py27
- python: 3.4
env: TOXENV=py34
- python: 3.5
env: TOXENV=py35
- python: 3.6
env: TOXENV=py36
- python: 3.6
env: TOXENV=doc
- python: 3.6
env: TOXENV=sphinx
- python: 3.6
env: TOXENV=lint
- python: 2.7
env: TOXENV=pep8py2
- python: 3.6
env: TOXENV=pep8py3
install:
- pip install --upgrade pip setuptools
- pip --version
- pip install tox
- tox --version
script:
- tox

165
LICENSE Normal file
View File

@ -0,0 +1,165 @@
GNU LESSER GENERAL PUBLIC LICENSE
Version 3, 29 June 2007
Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>
Everyone is permitted to copy and distribute verbatim copies
of this license document, but changing it is not allowed.
This version of the GNU Lesser General Public License incorporates
the terms and conditions of version 3 of the GNU General Public
License, supplemented by the additional permissions listed below.
0. Additional Definitions.
As used herein, "this License" refers to version 3 of the GNU Lesser
General Public License, and the "GNU GPL" refers to version 3 of the GNU
General Public License.
"The Library" refers to a covered work governed by this License,
other than an Application or a Combined Work as defined below.
An "Application" is any work that makes use of an interface provided
by the Library, but which is not otherwise based on the Library.
Defining a subclass of a class defined by the Library is deemed a mode
of using an interface provided by the Library.
A "Combined Work" is a work produced by combining or linking an
Application with the Library. The particular version of the Library
with which the Combined Work was made is also called the "Linked
Version".
The "Minimal Corresponding Source" for a Combined Work means the
Corresponding Source for the Combined Work, excluding any source code
for portions of the Combined Work that, considered in isolation, are
based on the Application, and not on the Linked Version.
The "Corresponding Application Code" for a Combined Work means the
object code and/or source code for the Application, including any data
and utility programs needed for reproducing the Combined Work from the
Application, but excluding the System Libraries of the Combined Work.
1. Exception to Section 3 of the GNU GPL.
You may convey a covered work under sections 3 and 4 of this License
without being bound by section 3 of the GNU GPL.
2. Conveying Modified Versions.
If you modify a copy of the Library, and, in your modifications, a
facility refers to a function or data to be supplied by an Application
that uses the facility (other than as an argument passed when the
facility is invoked), then you may convey a copy of the modified
version:
a) under this License, provided that you make a good faith effort to
ensure that, in the event an Application does not supply the
function or data, the facility still operates, and performs
whatever part of its purpose remains meaningful, or
b) under the GNU GPL, with none of the additional permissions of
this License applicable to that copy.
3. Object Code Incorporating Material from Library Header Files.
The object code form of an Application may incorporate material from
a header file that is part of the Library. You may convey such object
code under terms of your choice, provided that, if the incorporated
material is not limited to numerical parameters, data structure
layouts and accessors, or small macros, inline functions and templates
(ten or fewer lines in length), you do both of the following:
a) Give prominent notice with each copy of the object code that the
Library is used in it and that the Library and its use are
covered by this License.
b) Accompany the object code with a copy of the GNU GPL and this license
document.
4. Combined Works.
You may convey a Combined Work under terms of your choice that,
taken together, effectively do not restrict modification of the
portions of the Library contained in the Combined Work and reverse
engineering for debugging such modifications, if you also do each of
the following:
a) Give prominent notice with each copy of the Combined Work that
the Library is used in it and that the Library and its use are
covered by this License.
b) Accompany the Combined Work with a copy of the GNU GPL and this license
document.
c) For a Combined Work that displays copyright notices during
execution, include the copyright notice for the Library among
these notices, as well as a reference directing the user to the
copies of the GNU GPL and this license document.
d) Do one of the following:
0) Convey the Minimal Corresponding Source under the terms of this
License, and the Corresponding Application Code in a form
suitable for, and under terms that permit, the user to
recombine or relink the Application with a modified version of
the Linked Version to produce a modified Combined Work, in the
manner specified by section 6 of the GNU GPL for conveying
Corresponding Source.
1) Use a suitable shared library mechanism for linking with the
Library. A suitable mechanism is one that (a) uses at run time
a copy of the Library already present on the user's computer
system, and (b) will operate properly with a modified version
of the Library that is interface-compatible with the Linked
Version.
e) Provide Installation Information, but only if you would otherwise
be required to provide such information under section 6 of the
GNU GPL, and only to the extent that such information is
necessary to install and execute a modified version of the
Combined Work produced by recombining or relinking the
Application with a modified version of the Linked Version. (If
you use option 4d0, the Installation Information must accompany
the Minimal Corresponding Source and Corresponding Application
Code. If you use option 4d1, you must provide the Installation
Information in the manner specified by section 6 of the GNU GPL
for conveying Corresponding Source.)
5. Combined Libraries.
You may place library facilities that are a work based on the
Library side by side in a single library together with other library
facilities that are not Applications and are not covered by this
License, and convey such a combined library under terms of your
choice, if you do both of the following:
a) Accompany the combined library with a copy of the same work based
on the Library, uncombined with any other library facilities,
conveyed under the terms of this License.
b) Give prominent notice with the combined library that part of it
is a work based on the Library, and explaining where to find the
accompanying uncombined form of the same work.
6. Revised Versions of the GNU Lesser General Public License.
The Free Software Foundation may publish revised and/or new versions
of the GNU Lesser General Public License from time to time. Such new
versions will be similar in spirit to the present version, but may
differ in detail to address new problems or concerns.
Each version is given a distinguishing version number. If the
Library as you received it specifies that a certain numbered version
of the GNU Lesser General Public License "or any later version"
applies to it, you have the option of following the terms and
conditions either of that published version or of any later version
published by the Free Software Foundation. If the Library as you
received it does not specify a version number of the GNU Lesser
General Public License, you may choose any version of the GNU Lesser
General Public License ever published by the Free Software Foundation.
If the Library as you received it specifies that a proxy can decide
whether future versions of the GNU Lesser General Public License shall
apply, that proxy's public statement of acceptance of any version is
permanent authorization for you to choose that version for the
Library.

2
MANIFEST.in Normal file
View File

@ -0,0 +1,2 @@
include LICENSE README.md
include tox.ini setup.cfg

38
Makefile Normal file
View File

@ -0,0 +1,38 @@
all: lint pep8 docs test
echo "All tests passed"
lint:
# Pylint checks
tox -e lint
pep8:
# Check style consistency
tox -e pep8py2
tox -e pep8py3
clean:
rm -fr build dist *.egg-info
find ./ -name '*.pyc' -exec rm -f {} \;
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
docs:
$(MAKE) -C $(DOCS_DIR) html

18
README.md Normal file
View File

@ -0,0 +1,18 @@
[![Build Status](https://travis-ci.org/latchset/jwcrypto.svg?branch=master)](https://travis-ci.org/latchset/jwcrypto)
JWCrypto
========
An implementation of the JOSE Working Group documents:
RFC 7515 - JSON Web Signature (JWS)
RFC 7516 - JSON Web Encryption (JWE)
RFC 7517 - JSON Web Key (JWK)
RFC 7518 - JSON Web Algorithms (JWA)
RFC 7519 - JSON Web Token (JWT)
RFC 7520 - Examples of Protecting Content Using JSON Object Signing and
Encryption (JOSE)
Documentation
=============
http://jwcrypto.readthedocs.org

177
docs/Makefile Normal file
View File

@ -0,0 +1,177 @@
# Makefile for Sphinx documentation
#
# You can set these variables from the command line.
SPHINXOPTS =
SPHINXBUILD = sphinx-build
PAPER =
BUILDDIR = build
# User-friendly check for sphinx-build
ifeq ($(shell which $(SPHINXBUILD) >/dev/null 2>&1; echo $$?), 1)
$(error The '$(SPHINXBUILD)' command was not found. Make sure you have Sphinx installed, then set the SPHINXBUILD environment variable to point to the full path of the '$(SPHINXBUILD)' executable. Alternatively you can add the directory with the executable to your PATH. If you don't have Sphinx installed, grab it from http://sphinx-doc.org/)
endif
# Internal variables.
PAPEROPT_a4 = -D latex_paper_size=a4
PAPEROPT_letter = -D latex_paper_size=letter
ALLSPHINXOPTS = -d $(BUILDDIR)/doctrees $(PAPEROPT_$(PAPER)) $(SPHINXOPTS) source
# the i18n builder cannot share the environment and doctrees with the others
I18NSPHINXOPTS = $(PAPEROPT_$(PAPER)) $(SPHINXOPTS) source
.PHONY: help clean html dirhtml singlehtml pickle json htmlhelp qthelp devhelp epub latex latexpdf text man changes linkcheck doctest gettext
help:
@echo "Please use \`make <target>' where <target> is one of"
@echo " html to make standalone HTML files"
@echo " dirhtml to make HTML files named index.html in directories"
@echo " singlehtml to make a single large HTML file"
@echo " pickle to make pickle files"
@echo " json to make JSON files"
@echo " htmlhelp to make HTML files and a HTML help project"
@echo " qthelp to make HTML files and a qthelp project"
@echo " devhelp to make HTML files and a Devhelp project"
@echo " epub to make an epub"
@echo " latex to make LaTeX files, you can set PAPER=a4 or PAPER=letter"
@echo " latexpdf to make LaTeX files and run them through pdflatex"
@echo " latexpdfja to make LaTeX files and run them through platex/dvipdfmx"
@echo " text to make text files"
@echo " man to make manual pages"
@echo " texinfo to make Texinfo files"
@echo " info to make Texinfo files and run them through makeinfo"
@echo " gettext to make PO message catalogs"
@echo " changes to make an overview of all changed/added/deprecated items"
@echo " xml to make Docutils-native XML files"
@echo " pseudoxml to make pseudoxml-XML files for display purposes"
@echo " linkcheck to check all external links for integrity"
@echo " doctest to run all doctests embedded in the documentation (if enabled)"
clean:
rm -rf $(BUILDDIR)/*
html:
$(SPHINXBUILD) -b html $(ALLSPHINXOPTS) $(BUILDDIR)/html
@echo
@echo "Build finished. The HTML pages are in $(BUILDDIR)/html."
dirhtml:
$(SPHINXBUILD) -b dirhtml $(ALLSPHINXOPTS) $(BUILDDIR)/dirhtml
@echo
@echo "Build finished. The HTML pages are in $(BUILDDIR)/dirhtml."
singlehtml:
$(SPHINXBUILD) -b singlehtml $(ALLSPHINXOPTS) $(BUILDDIR)/singlehtml
@echo
@echo "Build finished. The HTML page is in $(BUILDDIR)/singlehtml."
pickle:
$(SPHINXBUILD) -b pickle $(ALLSPHINXOPTS) $(BUILDDIR)/pickle
@echo
@echo "Build finished; now you can process the pickle files."
json:
$(SPHINXBUILD) -b json $(ALLSPHINXOPTS) $(BUILDDIR)/json
@echo
@echo "Build finished; now you can process the JSON files."
htmlhelp:
$(SPHINXBUILD) -b htmlhelp $(ALLSPHINXOPTS) $(BUILDDIR)/htmlhelp
@echo
@echo "Build finished; now you can run HTML Help Workshop with the" \
".hhp project file in $(BUILDDIR)/htmlhelp."
qthelp:
$(SPHINXBUILD) -b qthelp $(ALLSPHINXOPTS) $(BUILDDIR)/qthelp
@echo
@echo "Build finished; now you can run "qcollectiongenerator" with the" \
".qhcp project file in $(BUILDDIR)/qthelp, like this:"
@echo "# qcollectiongenerator $(BUILDDIR)/qthelp/JWCrypto.qhcp"
@echo "To view the help file:"
@echo "# assistant -collectionFile $(BUILDDIR)/qthelp/JWCrypto.qhc"
devhelp:
$(SPHINXBUILD) -b devhelp $(ALLSPHINXOPTS) $(BUILDDIR)/devhelp
@echo
@echo "Build finished."
@echo "To view the help file:"
@echo "# mkdir -p $$HOME/.local/share/devhelp/JWCrypto"
@echo "# ln -s $(BUILDDIR)/devhelp $$HOME/.local/share/devhelp/JWCrypto"
@echo "# devhelp"
epub:
$(SPHINXBUILD) -b epub $(ALLSPHINXOPTS) $(BUILDDIR)/epub
@echo
@echo "Build finished. The epub file is in $(BUILDDIR)/epub."
latex:
$(SPHINXBUILD) -b latex $(ALLSPHINXOPTS) $(BUILDDIR)/latex
@echo
@echo "Build finished; the LaTeX files are in $(BUILDDIR)/latex."
@echo "Run \`make' in that directory to run these through (pdf)latex" \
"(use \`make latexpdf' here to do that automatically)."
latexpdf:
$(SPHINXBUILD) -b latex $(ALLSPHINXOPTS) $(BUILDDIR)/latex
@echo "Running LaTeX files through pdflatex..."
$(MAKE) -C $(BUILDDIR)/latex all-pdf
@echo "pdflatex finished; the PDF files are in $(BUILDDIR)/latex."
latexpdfja:
$(SPHINXBUILD) -b latex $(ALLSPHINXOPTS) $(BUILDDIR)/latex
@echo "Running LaTeX files through platex and dvipdfmx..."
$(MAKE) -C $(BUILDDIR)/latex all-pdf-ja
@echo "pdflatex finished; the PDF files are in $(BUILDDIR)/latex."
text:
$(SPHINXBUILD) -b text $(ALLSPHINXOPTS) $(BUILDDIR)/text
@echo
@echo "Build finished. The text files are in $(BUILDDIR)/text."
man:
$(SPHINXBUILD) -b man $(ALLSPHINXOPTS) $(BUILDDIR)/man
@echo
@echo "Build finished. The manual pages are in $(BUILDDIR)/man."
texinfo:
$(SPHINXBUILD) -b texinfo $(ALLSPHINXOPTS) $(BUILDDIR)/texinfo
@echo
@echo "Build finished. The Texinfo files are in $(BUILDDIR)/texinfo."
@echo "Run \`make' in that directory to run these through makeinfo" \
"(use \`make info' here to do that automatically)."
info:
$(SPHINXBUILD) -b texinfo $(ALLSPHINXOPTS) $(BUILDDIR)/texinfo
@echo "Running Texinfo files through makeinfo..."
make -C $(BUILDDIR)/texinfo info
@echo "makeinfo finished; the Info files are in $(BUILDDIR)/texinfo."
gettext:
$(SPHINXBUILD) -b gettext $(I18NSPHINXOPTS) $(BUILDDIR)/locale
@echo
@echo "Build finished. The message catalogs are in $(BUILDDIR)/locale."
changes:
$(SPHINXBUILD) -b changes $(ALLSPHINXOPTS) $(BUILDDIR)/changes
@echo
@echo "The overview file is in $(BUILDDIR)/changes."
linkcheck:
$(SPHINXBUILD) -b linkcheck $(ALLSPHINXOPTS) $(BUILDDIR)/linkcheck
@echo
@echo "Link check complete; look for any errors in the above output " \
"or in $(BUILDDIR)/linkcheck/output.txt."
doctest:
$(SPHINXBUILD) -b doctest $(ALLSPHINXOPTS) $(BUILDDIR)/doctest
@echo "Testing of doctests in the sources finished, look at the " \
"results in $(BUILDDIR)/doctest/output.txt."
xml:
$(SPHINXBUILD) -b xml $(ALLSPHINXOPTS) $(BUILDDIR)/xml
@echo
@echo "Build finished. The XML files are in $(BUILDDIR)/xml."
pseudoxml:
$(SPHINXBUILD) -b pseudoxml $(ALLSPHINXOPTS) $(BUILDDIR)/pseudoxml
@echo
@echo "Build finished. The pseudo-XML files are in $(BUILDDIR)/pseudoxml."

1
docs/source/.static/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
# empty

263
docs/source/conf.py Normal file
View File

@ -0,0 +1,263 @@
# -*- coding: utf-8 -*-
#
# JWCrypto documentation build configuration file, created by
# sphinx-quickstart on Tue Apr 14 10:25:10 2015.
#
# This file is execfile()d with the current directory set to its
# containing dir.
#
# Note that not all possible configuration values are present in this
# autogenerated file.
#
# All configuration values have a default; values that are commented out
# serve to show the default.
import sys
import os
# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
#sys.path.insert(0, os.path.abspath('.'))
sys.path.insert(0, os.path.abspath('../..'))
# -- General configuration ------------------------------------------------
# If your documentation needs a minimal Sphinx version, state it here.
#needs_sphinx = '1.0'
# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = ['sphinx.ext.autodoc', 'sphinx.ext.intersphinx']
# Add any paths that contain templates here, relative to this directory.
templates_path = ['.templates']
# The suffix of source filenames.
source_suffix = '.rst'
# The encoding of source files.
#source_encoding = 'utf-8-sig'
# The master toctree document.
master_doc = 'index'
# General information about the project.
project = u'JWCrypto'
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.6'
# The full version, including alpha/beta/rc tags.
release = '0.6'
# The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages.
#language = None
# There are two options for replacing |today|: either, you set today to some
# non-false value, then it is used:
#today = ''
# Else, today_fmt is used as the format for a strftime call.
#today_fmt = '%B %d, %Y'
# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
exclude_patterns = []
# The reST default role (used for this markup: `text`) to use for all
# documents.
#default_role = None
# If true, '()' will be appended to :func: etc. cross-reference text.
#add_function_parentheses = True
# If true, the current module name will be prepended to all description
# unit titles (such as .. function::).
#add_module_names = True
# If true, sectionauthor and moduleauthor directives will be shown in the
# output. They are ignored by default.
#show_authors = False
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = 'sphinx'
# A list of ignored prefixes for module index sorting.
#modindex_common_prefix = []
# If true, keep warnings as "system message" paragraphs in the built documents.
#keep_warnings = False
# -- Options for HTML output ----------------------------------------------
# The theme to use for HTML and HTML Help pages. See the documentation for
# a list of builtin themes.
html_theme = 'default'
# Theme options are theme-specific and customize the look and feel of a theme
# further. For a list of options available for each theme, see the
# documentation.
#html_theme_options = {}
# Add any paths that contain custom themes here, relative to this directory.
#html_theme_path = []
# The name for this set of Sphinx documents. If None, it defaults to
# "<project> v<release> documentation".
#html_title = None
# A shorter title for the navigation bar. Default is the same as html_title.
#html_short_title = None
# The name of an image file (relative to this directory) to place at the top
# of the sidebar.
#html_logo = None
# The name of an image file (within the static path) to use as favicon of the
# docs. This file should be a Windows icon file (.ico) being 16x16 or 32x32
# pixels large.
#html_favicon = None
# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ['.static']
# Add any extra paths that contain custom files (such as robots.txt or
# .htaccess) here, relative to this directory. These files are copied
# directly to the root of the documentation.
#html_extra_path = []
# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
# using the given strftime format.
#html_last_updated_fmt = '%b %d, %Y'
# If true, SmartyPants will be used to convert quotes and dashes to
# typographically correct entities.
#html_use_smartypants = True
# Custom sidebar templates, maps document names to template names.
#html_sidebars = {}
# Additional templates that should be rendered to pages, maps page names to
# template names.
#html_additional_pages = {}
# If false, no module index is generated.
#html_domain_indices = True
# If false, no index is generated.
#html_use_index = True
# If true, the index is split into individual pages for each letter.
#html_split_index = False
# If true, links to the reST sources are added to the pages.
#html_show_sourcelink = True
# If true, "Created using Sphinx" is shown in the HTML footer. Default is True.
#html_show_sphinx = True
# If true, "(C) Copyright ..." is shown in the HTML footer. Default is True.
#html_show_copyright = True
# If true, an OpenSearch description file will be output, and all pages will
# contain a <link> tag referring to it. The value of this option must be the
# base URL from which the finished HTML is served.
#html_use_opensearch = ''
# This is the file name suffix for HTML files (e.g. ".xhtml").
#html_file_suffix = None
# Output file base name for HTML help builder.
htmlhelp_basename = 'JWCryptodoc'
# -- Options for LaTeX output ---------------------------------------------
latex_elements = {
# The paper size ('letterpaper' or 'a4paper').
#'papersize': 'letterpaper',
# The font size ('10pt', '11pt' or '12pt').
#'pointsize': '10pt',
# Additional stuff for the LaTeX preamble.
#'preamble': '',
}
# Grouping the document tree into LaTeX files. List of tuples
# (source start file, target name, title,
# author, documentclass [howto, manual, or own class]).
latex_documents = [
('index', 'JWCrypto.tex', u'JWCrypto Documentation',
u'JWCrypto Contributors', 'manual'),
]
# The name of an image file (relative to this directory) to place at the top of
# the title page.
#latex_logo = None
# For "manual" documents, if this is true, then toplevel headings are parts,
# not chapters.
#latex_use_parts = False
# If true, show page references after internal links.
#latex_show_pagerefs = False
# If true, show URL addresses after external links.
#latex_show_urls = False
# Documents to append as an appendix to all manuals.
#latex_appendices = []
# If false, no module index is generated.
#latex_domain_indices = True
# -- Options for manual page output ---------------------------------------
# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
man_pages = [
('index', 'jwcrypto', u'JWCrypto Documentation',
[u'JWCrypto Contributors'], 1)
]
# If true, show URL addresses after external links.
#man_show_urls = False
# -- Options for Texinfo output -------------------------------------------
# Grouping the document tree into Texinfo files. List of tuples
# (source start file, target name, title, author,
# dir menu entry, description, category)
texinfo_documents = [
('index', 'JWCrypto', u'JWCrypto Documentation',
u'JWCrypto Contributors', 'JWCrypto', 'One line description of project.',
'Miscellaneous'),
]
# Documents to append as an appendix to all manuals.
#texinfo_appendices = []
# If false, no module index is generated.
#texinfo_domain_indices = True
# How to display URL addresses: 'footnote', 'no', or 'inline'.
#texinfo_show_urls = 'footnote'
# If true, do not generate a @detailmenu in the "Top" node's menu.
#texinfo_no_detailmenu = False
autoclass_content = 'both'
autodoc_member_order = 'groupwise'

36
docs/source/index.rst Normal file
View File

@ -0,0 +1,36 @@
.. JWCrypto documentation master file, created by
sphinx-quickstart on Tue Apr 14 10:25:10 2015.
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
Welcome to JWCrypto's documentation!
====================================
JWCrypto is an implementation of the Javascript Object Signing and
Encryption (JOSE) Web Standards as they are being developed in the
JOSE_ IETF Working Group and related technology.
JWCrypto is Python2 and Python3 compatible and uses the Cryptography_
package for all the crypto functions.
.. _JOSE: https://datatracker.ietf.org/wg/jose/charter/
.. _Cryptography: https://cryptography.io/
Contents:
.. toctree::
:maxdepth: 2
jwk
jws
jwe
jwt
Indices and tables
==================
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`

98
docs/source/jwe.rst Normal file
View File

@ -0,0 +1,98 @@
JSON Web Encryption (JWE)
=========================
The jwe Module implements the `JSON Web Encryption`_ standard.
A JSON Web Encryption is represented by a JWE object, related utility
classes and functions are availbale in this module too.
.. _JSON Web Encryption: https://tools.ietf.org/html/rfc7516
Classes
-------
.. autoclass:: jwcrypto.jwe.JWE
:members:
:show-inheritance:
Variables
---------
.. autodata:: jwcrypto.jwe.default_allowed_algs
Exceptions
----------
.. autoclass:: jwcrypto.jwe.InvalidJWEOperation
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwe.InvalidJWEData
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwe.InvalidJWEKeyType
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwe.InvalidJWEKeyLength
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwe.InvalidCEKeyLength
:members:
:show-inheritance:
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

90
docs/source/jwk.rst Normal file
View File

@ -0,0 +1,90 @@
JSON Web Key (JWK)
==================
The jwk Module implements the `JSON Web Key`_ standard.
A JSON Web Key is represented by a JWK object, related utility classes and
functions are availbale in this module too.
.. _JSON Web Key: http://tools.ietf.org/html/rfc7517
Classes
-------
.. autoclass:: jwcrypto.jwk.JWK
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwk.JWKSet
:members:
:show-inheritance:
Exceptions
----------
.. autoclass:: jwcrypto.jwk.InvalidJWKType
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwk.InvalidJWKValue
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwk.InvalidJWKOperation
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jwk.InvalidJWKUsage
:members:
:show-inheritance:
Registries
----------
.. autodata:: jwcrypto.jwk.JWKTypesRegistry
:annotation:
.. autodata:: jwcrypto.jwk.JWKValuesRegistry
:annotation:
.. autodata:: jwcrypto.jwk.JWKParamsRegistry
:annotation:
.. autodata:: jwcrypto.jwk.JWKEllipticCurveRegistry
:annotation:
.. autodata:: jwcrypto.jwk.JWKUseRegistry
:annotation:
.. autodata:: jwcrypto.jwk.JWKOperationsRegistry
:annotation:
Examples
--------
Create a 256bit symmetric key::
>>> from jwcrypto import jwk
>>> key = jwk.JWK.generate(kty='oct', size=256)
Export the key with::
>>> key.export()
'{"k":"X6TBlwY2so8EwKZ2TFXM7XHSgWBKQJhcspzYydp5Y-o","kty":"oct"}'
Create a 2048bit RSA keypair::
>>> jwk.JWK.generate(kty='RSA', size=2048)
Create a P-256 EC keypair and export the public key::
>>> key = jwk.JWK.generate(kty='EC', crv='P-256')
>>> key.export(private_key=False)
'{"y":"VYlYwBfOTIICojCPfdUjnmkpN-g-lzZKxzjAoFmDRm8",
"x":"3mdE0rODWRju6qqU01Kw5oPYdNxBOMisFvJFH1vEu9Q",
"crv":"P-256","kty":"EC"}'
Import a P-256 Public Key::
>>> expkey = {"y":"VYlYwBfOTIICojCPfdUjnmkpN-g-lzZKxzjAoFmDRm8",
"x":"3mdE0rODWRju6qqU01Kw5oPYdNxBOMisFvJFH1vEu9Q",
"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())

65
docs/source/jws.rst Normal file
View File

@ -0,0 +1,65 @@
JSON Web Signature (JWS)
========================
The jws Module implements the `JSON Web Signature`_ standard.
A JSON Web Signature is represented by a JWS object, related utility
classes and functions are available in this module too.
.. _JSON Web Signature: http://tools.ietf.org/html/rfc7515
Classes
-------
.. autoclass:: jwcrypto.jws.JWS
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jws.JWSCore
:members:
:show-inheritance:
Variables
---------
.. autodata:: jwcrypto.jws.default_allowed_algs
Exceptions
----------
.. autoclass:: jwcrypto.jws.InvalidJWSSignature
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jws.InvalidJWSObject
:members:
:show-inheritance:
.. autoclass:: jwcrypto.jws.InvalidJWSOperation
:members:
:show-inheritance:
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

48
docs/source/jwt.rst Normal file
View File

@ -0,0 +1,48 @@
JSON Web Token (JWT)
====================
The jwt Module implements the `JSON Web Token`_ standard.
A JSON Web Token is represented by a JWT object, related utility classes and
functions are availbale in this module too.
.. _JSON Web Token: http://tools.ietf.org/html/rfc7519
Classes
-------
.. autoclass:: jwcrypto.jwt.JWT
:members:
:show-inheritance:
Examples
--------
Create a symmetric key::
>>> from jwcrypto import jwt, jwk
>>> key = jwk.JWK(generate='oct', size=256)
>>> key.export()
'{"k":"Wal4ZHCBsml0Al_Y8faoNTKsXCkw8eefKXYFuwTBOpA","kty":"oct"}'
Create a signed token with the generated key::
>>> Token = jwt.JWT(header={"alg": "HS256"},
claims={"info": "I'm a signed token"})
>>> Token.make_signed_token(key)
>>> Token.serialize()
u'eyJhbGciOiJIUzI1NiJ9.eyJpbmZvIjoiSSdtIGEgc2lnbmVkIHRva2VuIn0.rjnRMAKcaRamEHnENhg0_Fqv7Obo-30U4bcI_v-nfEM'
Further encrypt the token with the same key::
>>> Etoken = jwt.JWT(header={"alg": "A256KW", "enc": "A256CBC-HS512"},
claims=Token.serialize())
>>> Etoken.make_encrypted_token(key)
>>> Etoken.serialize()
u'eyJhbGciOiJBMjU2S1ciLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIn0.ST5RmjqDLj696xo7YFTFuKUhcd3naCrm6yMjBM3cqWiFD6U8j2JIsbclsF7ryNg8Ktmt1kQJRKavV6DaTl1T840tP3sIs1qz.wSxVhZH5GyzbJnPBAUMdzQ.6uiVYwrRBzAm7Uge9rEUjExPWGbgerF177A7tMuQurJAqBhgk3_5vee5DRH84kHSapFOxcEuDdMBEQLI7V2E0F57-d01TFStHzwtgtSmeZRQ6JSIL5XlgJouwHfSxn9Z_TGl5xxq4TksORHED1vnRA.5jPyPWanJVqlOohApEbHmxi3JHp1MXbmvQe2_dVd8FI'
Now decrypt and verify::
>>> from jwcrypto import jwt, jwk
>>> k = {"k": "Wal4ZHCBsml0Al_Y8faoNTKsXCkw8eefKXYFuwTBOpA", "kty": "oct"}
>>> key = jwk.JWK(**k)
>>> e = u'eyJhbGciOiJBMjU2S1ciLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIn0.ST5RmjqDLj696xo7YFTFuKUhcd3naCrm6yMjBM3cqWiFD6U8j2JIsbclsF7ryNg8Ktmt1kQJRKavV6DaTl1T840tP3sIs1qz.wSxVhZH5GyzbJnPBAUMdzQ.6uiVYwrRBzAm7Uge9rEUjExPWGbgerF177A7tMuQurJAqBhgk3_5vee5DRH84kHSapFOxcEuDdMBEQLI7V2E0F57-d01TFStHzwtgtSmeZRQ6JSIL5XlgJouwHfSxn9Z_TGl5xxq4TksORHED1vnRA.5jPyPWanJVqlOohApEbHmxi3JHp1MXbmvQe2_dVd8FI'
>>> ET = jwt.JWT(key=key, jwt=e)
>>> ST = jwt.JWT(key=key, jwt=ET.claims)
>>> ST.claims
u'{"info":"I\'m a signed token"}'

0
jwcrypto/__init__.py Normal file
View File

106
jwcrypto/common.py Normal file
View File

@ -0,0 +1,106 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
import json
from base64 import urlsafe_b64decode, urlsafe_b64encode
# Padding stripping versions as described in
# RFC 7515 Appendix C
def base64url_encode(payload):
if not isinstance(payload, bytes):
payload = payload.encode('utf-8')
encode = urlsafe_b64encode(payload)
return encode.decode('utf-8').rstrip('=')
def base64url_decode(payload):
size = len(payload) % 4
if size == 2:
payload += '=='
elif size == 3:
payload += '='
elif size != 0:
raise ValueError('Invalid base64 string')
return urlsafe_b64decode(payload.encode('utf-8'))
# JSON encoding/decoding helpers with good defaults
def json_encode(string):
if isinstance(string, bytes):
string = string.decode('utf-8')
return json.dumps(string, separators=(',', ':'), sort_keys=True)
def json_decode(string):
if isinstance(string, bytes):
string = string.decode('utf-8')
return json.loads(string)
class JWException(Exception):
pass
class InvalidJWAAlgorithm(JWException):
def __init__(self, message=None):
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)

1072
jwcrypto/jwa.py Normal file

File diff suppressed because it is too large Load Diff

498
jwcrypto/jwe.py Normal file
View File

@ -0,0 +1,498 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
import zlib
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.jwa import JWA
# RFC 7516 - 4.1
# name: (description, supported?)
JWEHeaderRegistry = {'alg': ('Algorithm', True),
'enc': ('Encryption Algorithm', True),
'zip': ('Compression Algorithm', True),
'jku': ('JWK Set URL', False),
'jwk': ('JSON Web Key', False),
'kid': ('Key ID', True),
'x5u': ('X.509 URL', False),
'x5c': ('X.509 Certificate Chain', False),
'x5t': ('X.509 Certificate SHA-1 Thumbprint', False),
'x5t#S256': ('X.509 Certificate SHA-256 Thumbprint',
False),
'typ': ('Type', True),
'cty': ('Content Type', True),
'crit': ('Critical', True)}
"""Registry of valid header parameters"""
default_allowed_algs = [
# Key Management Algorithms
'RSA1_5', 'RSA-OAEP', 'RSA-OAEP-256',
'A128KW', 'A192KW', 'A256KW',
'dir',
'ECDH-ES', 'ECDH-ES+A128KW', 'ECDH-ES+A192KW', 'ECDH-ES+A256KW',
'A128GCMKW', 'A192GCMKW', 'A256GCMKW',
'PBES2-HS256+A128KW', 'PBES2-HS384+A192KW', 'PBES2-HS512+A256KW',
# Content Encryption Algoritms
'A128CBC-HS256', 'A192CBC-HS384', 'A256CBC-HS512',
'A128GCM', 'A192GCM', 'A256GCM']
"""Default allowed algorithms"""
class InvalidJWEData(JWException):
"""Invalid JWE Object.
This exception is raised when the JWE Object is invalid and/or
improperly formatted.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = message
else:
msg = 'Unknown Data Verification Failure'
if exception:
msg += ' {%s}' % str(exception)
super(InvalidJWEData, self).__init__(msg)
# 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):
"""JSON Web Encryption object
This object represent a JWE token.
"""
def __init__(self, plaintext=None, protected=None, unprotected=None,
aad=None, algs=None, recipient=None, header=None):
"""Creates a JWE token.
:param plaintext(bytes): An arbitrary plaintext to be encrypted.
:param protected: A JSON string with the protected header.
: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()
self.plaintext = None
if plaintext is not None:
if isinstance(plaintext, bytes):
self.plaintext = plaintext
else:
self.plaintext = plaintext.encode('utf-8')
self.cek = None
self.decryptlog = None
if aad:
self.objects['aad'] = aad
if protected:
if isinstance(protected, dict):
protected = json_encode(protected)
else:
json_decode(protected) # check header encoding
self.objects['protected'] = protected
if unprotected:
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
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 JWA.encryption_alg(name)
@property
def allowed_algs(self):
"""Allowed algorithms.
The list of allowed algorithms.
Can be changed by setting a list of algorithm names.
"""
if self._allowed_algs:
return self._allowed_algs
else:
return default_allowed_algs
@allowed_algs.setter
def allowed_algs(self, algs):
if not isinstance(algs, list):
raise TypeError('Allowed Algs must be a list')
self._allowed_algs = algs
def _merge_headers(self, h1, h2):
for k in list(h1.keys()):
if k in h2:
raise InvalidJWEData('Duplicate header: "%s"' % k)
h1.update(h2)
return h1
def _get_jose_header(self, header=None):
jh = dict()
if 'protected' in self.objects:
ph = json_decode(self.objects['protected'])
jh = self._merge_headers(jh, ph)
if 'unprotected' in self.objects:
uh = json_decode(self.objects['unprotected'])
jh = self._merge_headers(jh, uh)
if header:
rh = json_decode(header)
jh = self._merge_headers(jh, rh)
return jh
def _get_alg_enc_from_headers(self, jh):
algname = jh.get('alg', None)
if algname is None:
raise InvalidJWEData('Missing "alg" from headers')
alg = self._jwa_keymgmt(algname)
encname = jh.get('enc', None)
if encname is None:
raise InvalidJWEData('Missing "enc" from headers')
enc = self._jwa_enc(encname)
return alg, enc
def _encrypt(self, alg, enc, jh):
aad = base64url_encode(self.objects.get('protected', ''))
if 'aad' in self.objects:
aad += '.' + base64url_encode(self.objects['aad'])
aad = aad.encode('utf-8')
compress = jh.get('zip', None)
if compress == 'DEF':
data = zlib.compress(self.plaintext)[2:-4]
elif compress is None:
data = self.plaintext
else:
raise ValueError('Unknown compression')
iv, ciphertext, tag = enc.encrypt(self.cek, aad, data)
self.objects['iv'] = iv
self.objects['ciphertext'] = ciphertext
self.objects['tag'] = tag
def add_recipient(self, key, header=None):
"""Encrypt the plaintext with the given key.
:param key: A JWK key or password of appropriate type for the 'alg'
provided in the JOSE Headers.
:param header: A JSON string representing the per-recipient header.
:raises ValueError: if the plaintext is missing or not of type bytes.
:raises ValueError: if the compression type is unknown.
:raises InvalidJWAAlgorithm: if the 'alg' provided in the JOSE
headers is missing or unknown, or otherwise not implemented.
"""
if self.plaintext is None:
raise ValueError('Missing plaintext')
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)
rec = dict()
if header:
rec['header'] = header
wrapped = alg.wrap(key, enc.wrap_key_size, self.cek, jh)
self.cek = wrapped['cek']
if 'ek' in wrapped:
rec['encrypted_key'] = wrapped['ek']
if 'header' in wrapped:
h = json_decode(rec.get('header', '{}'))
nh = self._merge_headers(h, wrapped['header'])
rec['header'] = json_encode(nh)
if 'ciphertext' not in self.objects:
self._encrypt(alg, enc, jh)
if 'recipients' in self.objects:
self.objects['recipients'].append(rec)
elif 'encrypted_key' in self.objects or 'header' in self.objects:
self.objects['recipients'] = list()
n = dict()
if 'encrypted_key' in self.objects:
n['encrypted_key'] = self.objects.pop('encrypted_key')
if 'header' in self.objects:
n['header'] = self.objects.pop('header')
self.objects['recipients'].append(n)
self.objects['recipients'].append(rec)
else:
self.objects.update(rec)
def serialize(self, compact=False):
"""Serializes the object into a JWE token.
:param compact(boolean): if True generates the compact
representation, otherwise generates a standard JSON format.
:raises InvalidJWEOperation: if the object cannot serialized
with the compact representation and `compact` is True.
:raises InvalidJWEOperation: if no recipients have been added
to the object.
"""
if 'ciphertext' not in self.objects:
raise InvalidJWEOperation("No available ciphertext")
if compact:
for invalid in 'aad', 'unprotected':
if invalid in self.objects:
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")
rec = self.objects['recipients'][0]
else:
rec = self.objects
if 'header' in rec:
# The AESGCMKW algorithm generates data (iv, tag) we put in the
# per-recipient unpotected header by default. Move it to the
# protected header and re-encrypt the payload, as the protected
# header is used as additional authenticated data.
h = json_decode(rec['header'])
ph = json_decode(self.objects['protected'])
nph = self._merge_headers(h, ph)
self.objects['protected'] = json_encode(nph)
jh = self._get_jose_header()
alg, enc = self._get_alg_enc_from_headers(jh)
self._encrypt(alg, enc, jh)
del rec['header']
return '.'.join([base64url_encode(self.objects['protected']),
base64url_encode(rec.get('encrypted_key', '')),
base64url_encode(self.objects['iv']),
base64url_encode(self.objects['ciphertext']),
base64url_encode(self.objects['tag'])])
else:
obj = self.objects
enc = {'ciphertext': base64url_encode(obj['ciphertext']),
'iv': base64url_encode(obj['iv']),
'tag': base64url_encode(self.objects['tag'])}
if 'protected' in obj:
enc['protected'] = base64url_encode(obj['protected'])
if 'unprotected' in obj:
enc['unprotected'] = json_decode(obj['unprotected'])
if 'aad' in obj:
enc['aad'] = base64url_encode(obj['aad'])
if 'recipients' in obj:
enc['recipients'] = list()
for rec in obj['recipients']:
e = dict()
if 'encrypted_key' in rec:
e['encrypted_key'] = \
base64url_encode(rec['encrypted_key'])
if 'header' in rec:
e['header'] = json_decode(rec['header'])
enc['recipients'].append(e)
else:
if 'encrypted_key' in obj:
enc['encrypted_key'] = \
base64url_encode(obj['encrypted_key'])
if 'header' in obj:
enc['header'] = json_decode(obj['header'])
return json_encode(enc)
def _check_crit(self, crit):
for k in crit:
if k not in JWEHeaderRegistry:
raise InvalidJWEData('Unknown critical header: "%s"' % k)
else:
if not JWEHeaderRegistry[k][1]:
raise InvalidJWEData('Unsupported critical header: '
'"%s"' % k)
# FIXME: allow to specify which algorithms to accept as valid
def _decrypt(self, key, ppe):
jh = self._get_jose_header(ppe.get('header', None))
# TODO: allow caller to specify list of headers it understands
self._check_crit(jh.get('crit', dict()))
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.wrap_key_size,
ppe.get('encrypted_key', b''), jh)
data = enc.decrypt(cek, aad.encode('utf-8'),
self.objects['iv'],
self.objects['ciphertext'],
self.objects['tag'])
self.decryptlog.append('Success')
self.cek = cek
compress = jh.get('zip', None)
if compress == 'DEF':
self.plaintext = zlib.decompress(data, -zlib.MAX_WBITS)
elif compress is None:
self.plaintext = data
else:
raise ValueError('Unknown compression')
def decrypt(self, key):
"""Decrypt a JWE token.
:param key: The (:class:`jwcrypto.jwk.JWK`) decryption key.
:param key: A (:class:`jwcrypto.jwk.JWK`) decryption key or a password
string (optional).
:raises InvalidJWEOperation: if the key is not a JWK object.
:raises InvalidJWEData: if the ciphertext can't be decrypted or
the object is otherwise malformed.
"""
if 'ciphertext' not in self.objects:
raise InvalidJWEOperation("No available ciphertext")
self.decryptlog = list()
if 'recipients' in self.objects:
for rec in self.objects['recipients']:
try:
self._decrypt(key, rec)
except Exception as e: # pylint: disable=broad-except
self.decryptlog.append('Failed: [%s]' % repr(e))
else:
try:
self._decrypt(key, self.objects)
except Exception as e: # pylint: disable=broad-except
self.decryptlog.append('Failed: [%s]' % repr(e))
if not self.plaintext:
raise InvalidJWEData('No recipient matched the provided '
'key' + repr(self.decryptlog))
def deserialize(self, raw_jwe, key=None):
"""Deserialize a JWE token.
NOTE: Destroys any current status and tries to import the raw
JWE provided.
:param raw_jwe: a 'raw' JWE token (JSON Encoded or Compact
notation) string.
:param key: A (:class:`jwcrypto.jwk.JWK`) decryption key or a password
string (optional).
If a key is provided a decryption step will be attempted after
the object is successfully deserialized.
:raises InvalidJWEData: if the raw object is an invaid JWE token.
:raises InvalidJWEOperation: if the decryption fails.
"""
self.objects = dict()
self.plaintext = None
self.cek = None
o = dict()
try:
try:
djwe = json_decode(raw_jwe)
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(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(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(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(djwe['encrypted_key'])
if 'header' in djwe:
o['header'] = json_encode(djwe['header'])
except ValueError:
c = raw_jwe.split('.')
if len(c) != 5:
raise InvalidJWEData()
p = base64url_decode(c[0])
o['protected'] = p.decode('utf-8')
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
except Exception as e: # pylint: disable=broad-except
raise InvalidJWEData('Invalid format', repr(e))
if key:
self.decrypt(key)
@property
def payload(self):
if not self.plaintext:
raise InvalidJWEOperation("Plaintext not available")
return self.plaintext
@property
def jose_header(self):
jh = self._get_jose_header()
if len(jh) == 0:
raise InvalidJWEOperation("JOSE Header not available")
return jh

875
jwcrypto/jwk.py Normal file
View File

@ -0,0 +1,875 @@
# 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, 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
# RFC 7518 - 7.4
JWKTypesRegistry = {'EC': 'Elliptic Curve',
'RSA': 'RSA',
'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
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': 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
JWKEllipticCurveRegistry = {'P-256': 'P-256 curve',
'P-384': 'P-384 curve',
'P-521': 'P-521 curve'}
"""Registry of allowed Elliptic Curves"""
# RFC 7517 - 8.2
JWKUseRegistry = {'sig': 'Digital Signature or MAC',
'enc': 'Encryption'}
"""Registry of allowed uses"""
# RFC 7517 - 8.3
JWKOperationsRegistry = {'sign': 'Compute digital Signature or MAC',
'verify': 'Verify digital signature or MAC',
'encrypt': 'Encrypt content',
'decrypt': 'Decrypt content and validate'
' decryption, if applicable',
'wrapKey': 'Encrypt key',
'unwrapKey': 'Decrypt key and validate'
' decryption, if applicable',
'deriveKey': 'Derive key',
'deriveBits': 'Derive bits not to be used as a key'}
"""Registry of allowed operations"""
JWKpycaCurveMap = {'secp256r1': 'P-256',
'secp384r1': 'P-384',
'secp521r1': 'P-521'}
class InvalidJWKType(JWException):
"""Invalid JWK Type Exception.
This exception is raised when an invalid parameter type is used.
"""
def __init__(self, value=None):
super(InvalidJWKType, self).__init__()
self.value = value
def __str__(self):
return 'Unknown type "%s", valid types are: %s' % (
self.value, list(JWKTypesRegistry.keys()))
class InvalidJWKUsage(JWException):
"""Invalid JWK usage Exception.
This exception is raised when an invalid key usage is requested,
based on the key type and declared usage constraints.
"""
def __init__(self, use, value):
super(InvalidJWKUsage, self).__init__()
self.value = value
self.use = use
def __str__(self):
if self.use in list(JWKUseRegistry.keys()):
usage = JWKUseRegistry[self.use]
else:
usage = 'Unknown(%s)' % self.use
if self.value in list(JWKUseRegistry.keys()):
valid = JWKUseRegistry[self.value]
else:
valid = 'Unknown(%s)' % self.value
return 'Invalid usage requested: "%s". Valid for: "%s"' % (usage,
valid)
class InvalidJWKOperation(JWException):
"""Invalid JWK Operation Exception.
This exception is raised when an invalid key operation is requested,
based on the key type and declared usage constraints.
"""
def __init__(self, operation, values):
super(InvalidJWKOperation, self).__init__()
self.op = operation
self.values = values
def __str__(self):
if self.op in list(JWKOperationsRegistry.keys()):
op = JWKOperationsRegistry[self.op]
else:
op = 'Unknown(%s)' % self.op
valid = list()
for v in self.values:
if v in list(JWKOperationsRegistry.keys()):
valid.append(JWKOperationsRegistry[v])
else:
valid.append('Unknown(%s)' % v)
return 'Invalid operation requested: "%s". Valid for: "%s"' % (op,
valid)
class InvalidJWKValue(JWException):
"""Invalid JWK Value Exception.
This exception is raised when an invalid/unknown value is used in the
context of an operation that requires specific values to be used based
on the key type or other constraints.
"""
pass
class JWK(object):
"""JSON Web Key object
This object represent a Key.
It must be instantiated by using the standard defined key/value pairs
as arguments of the initialization function.
"""
def __init__(self, **kwargs):
"""Creates a new JWK object.
The function arguments must be valid parameters as defined in the
'IANA JSON Web Key Set Parameters registry' and specified in
the :data:`JWKParamsRegistry` variable. The 'kty' parameter must
always be provided and its value must be a valid one as defined
by the 'IANA JSON Web Key Types registry' and specified in the
:data:`JWKTypesRegistry` variable. The valid key parameters per
key type are defined in the :data:`JWKValuesregistry` variable.
To generate a new random key call the class method generate() with
the appropriate 'kty' parameter, and other parameters as needed (key
size, public exponents, curve types, etc..)
Valid options per type, when generating new keys:
* oct: size(int)
* RSA: public_exponent(int), size(int)
* EC: curve(str) (one of P-256, P-384, P-521)
Deprecated:
Alternatively if the 'generate' parameter is provided, with a
valid key type as value then a new key will be generated according
to the defaults or provided key strenght options (type specific).
:raises InvalidJWKType: if the key type is invalid
:raises InvalidJWKValue: if incorrect or inconsistent parameters
are provided.
"""
self._params = dict()
self._key = dict()
self._unknown = dict()
if 'generate' in kwargs:
self.generate_key(**kwargs)
elif kwargs:
self.import_key(**kwargs)
@classmethod
def generate(cls, **kwargs):
obj = cls()
kty = None
try:
kty = kwargs['kty']
gen = getattr(obj, '_generate_%s' % kty)
except (KeyError, AttributeError):
raise InvalidJWKType(kty)
gen(kwargs)
return obj
def generate_key(self, **params):
kty = None
try:
kty = params.pop('generate')
gen = getattr(self, '_generate_%s' % kty)
except (KeyError, AttributeError):
raise InvalidJWKType(kty)
gen(params)
def _get_gen_size(self, params, default_size=None):
size = default_size
if 'size' in params:
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)
self.import_key(**params)
def _encode_int(self, i):
intg = hex(i).rstrip("L").lstrip("0x")
return base64url_encode(unhexlify((len(intg) % 2) * '0' + intg))
def _generate_RSA(self, params):
pubexp = 65537
size = self._get_gen_size(params, 2048)
if 'public_exponent' in params:
pubexp = params.pop('public_exponent')
key = rsa.generate_private_key(pubexp, size, default_backend())
self._import_pyca_pri_rsa(key, **params)
def _import_pyca_pri_rsa(self, key, **params):
pn = key.private_numbers()
params.update(
kty='RSA',
n=self._encode_int(pn.public_numbers.n),
e=self._encode_int(pn.public_numbers.e),
d=self._encode_int(pn.d),
p=self._encode_int(pn.p),
q=self._encode_int(pn.q),
dp=self._encode_int(pn.dmp1),
dq=self._encode_int(pn.dmq1),
qi=self._encode_int(pn.iqmp)
)
self.import_key(**params)
def _import_pyca_pub_rsa(self, key, **params):
pn = key.public_numbers()
params.update(
kty='RSA',
n=self._encode_int(pn.n),
e=self._encode_int(pn.e)
)
self.import_key(**params)
def _get_curve_by_name(self, name):
if name == 'P-256':
return ec.SECP256R1()
elif name == 'P-384':
return ec.SECP384R1()
elif name == 'P-521':
return ec.SECP521R1()
else:
raise InvalidJWKValue('Unknown Elliptic Curve Type')
def _generate_EC(self, params):
curve = 'P-256'
if 'curve' in params:
curve = params.pop('curve')
# 'curve' is for backwards compat, if 'crv' is defined it takes
# precedence
if 'crv' in params:
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)
def _import_pyca_pri_ec(self, key, **params):
pn = key.private_numbers()
params.update(
kty='EC',
crv=JWKpycaCurveMap[key.curve.name],
x=self._encode_int(pn.public_numbers.x),
y=self._encode_int(pn.public_numbers.y),
d=self._encode_int(pn.private_value)
)
self.import_key(**params)
def _import_pyca_pub_ec(self, key, **params):
pn = key.public_numbers()
params.update(
kty='EC',
crv=JWKpycaCurveMap[key.curve.name],
x=self._encode_int(pn.x),
y=self._encode_int(pn.y),
)
self.import_key(**params)
def import_key(self, **kwargs):
names = list(kwargs.keys())
for name in list(JWKParamsRegistry.keys()):
if name in kwargs:
self._params[name] = kwargs[name]
while name in names:
names.remove(name)
kty = self._params.get('kty', None)
if kty not in JWKTypesRegistry:
raise InvalidJWKType(kty)
for name in list(JWKValuesRegistry[kty].keys()):
if name in kwargs:
self._key[name] = kwargs[name]
while name in names:
names.remove(name)
for name, val in iteritems(JWKValuesRegistry[kty]):
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
for name in names:
self._unknown[name] = kwargs[name]
if len(self._key) == 0:
raise InvalidJWKValue('No Key Values found')
# check key_ops
if 'key_ops' in self._params:
for ko in self._params['key_ops']:
c = 0
for cko in self._params['key_ops']:
if ko == cko:
c += 1
if c != 1:
raise InvalidJWKValue('Duplicate values in "key_ops"')
# check use/key_ops consistency
if 'use' in self._params and 'key_ops' in self._params:
sigl = ['sign', 'verify']
encl = ['encrypt', 'decrypt', 'wrapKey', 'unwrapKey',
'deriveKey', 'deriveBits']
if self._params['use'] == 'sig':
for op in encl:
if op in self._params['key_ops']:
raise InvalidJWKValue('Incompatible "use" and'
' "key_ops" values specified at'
' the same time')
elif self._params['use'] == 'enc':
for op in sigl:
if op in self._params['key_ops']:
raise InvalidJWKValue('Incompatible "use" and'
' "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 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()
def export_public(self):
"""Exports the public key in the standard JSON format.
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].public:
if name in self._params:
pub[name] = self._params[name]
reg = JWKValuesRegistry[self._params['kty']]
for param in reg:
if reg[param].public:
pub[param] = self._key[param]
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):
"""The Key type"""
return self._params.get('kty', None)
@property
def key_id(self):
"""The Key ID.
Provided by the kid parameter if present, otherwise returns None.
"""
return self._params.get('kid', None)
@property
def key_curve(self):
"""The Curve Name."""
if self._params['kty'] != 'EC':
raise InvalidJWKType('Not an EC key')
return self._key['crv']
def get_curve(self, arg):
"""Gets the Elliptic Curve associated with the key.
:param arg: an optional curve name
:raises InvalidJWKType: the key is not an EC key.
:raises InvalidJWKValue: if the curve names is invalid.
"""
k = self._key
if self._params['kty'] != 'EC':
raise InvalidJWKType('Not an EC key')
if arg and k['crv'] != arg:
raise InvalidJWKValue('Curve requested is "%s", but '
'key curve is "%s"' % (arg, k['crv']))
return self._get_curve_by_name(k['crv'])
def _check_constraints(self, usage, operation):
use = self._params.get('use', None)
if use and use != usage:
raise InvalidJWKUsage(usage, use)
ops = self._params.get('key_ops', None)
if ops:
if not isinstance(ops, list):
ops = [ops]
if operation not in ops:
raise InvalidJWKOperation(operation, ops)
# TODO: check alg ?
def _decode_int(self, n):
return int(hexlify(base64url_decode(n)), 16)
def _rsa_pub(self, k):
return rsa.RSAPublicNumbers(self._decode_int(k['e']),
self._decode_int(k['n']))
def _rsa_pri(self, k):
return rsa.RSAPrivateNumbers(self._decode_int(k['p']),
self._decode_int(k['q']),
self._decode_int(k['d']),
self._decode_int(k['dp']),
self._decode_int(k['dq']),
self._decode_int(k['qi']),
self._rsa_pub(k))
def _ec_pub(self, k, curve):
return ec.EllipticCurvePublicNumbers(self._decode_int(k['x']),
self._decode_int(k['y']),
self.get_curve(curve))
def _ec_pri(self, k, curve):
return ec.EllipticCurvePrivateNumbers(self._decode_int(k['d']),
self._ec_pub(k, curve))
def _get_public_key(self, arg=None):
if self._params['kty'] == 'oct':
return self._key['k']
elif self._params['kty'] == 'RSA':
return self._rsa_pub(self._key).public_key(default_backend())
elif self._params['kty'] == 'EC':
return self._ec_pub(self._key, arg).public_key(default_backend())
else:
raise NotImplementedError
def _get_private_key(self, arg=None):
if self._params['kty'] == 'oct':
return self._key['k']
elif self._params['kty'] == 'RSA':
return self._rsa_pri(self._key).private_key(default_backend())
elif self._params['kty'] == 'EC':
return self._ec_pri(self._key, arg).private_key(default_backend())
else:
raise NotImplementedError
def get_op_key(self, operation=None, arg=None):
"""Get the key object associated to the requested opration.
For example the public RSA key for the 'verify' operation or
the private EC key for the 'decrypt' operation.
:param operation: The requested operation.
The valid set of operations is availble in the
:data:`JWKOperationsRegistry` registry.
:param arg: an optional, context specific, argument
For example a curve name.
:raises InvalidJWKOperation: if the operation is unknown or
not permitted with this key.
:raises InvalidJWKUsage: if the use constraints do not permit
the operation.
"""
validops = self._params.get('key_ops',
list(JWKOperationsRegistry.keys()))
if validops is not list:
validops = [validops]
if operation is None:
if self._params['kty'] == 'oct':
return self._key['k']
raise InvalidJWKOperation(operation, validops)
elif operation == 'sign':
self._check_constraints('sig', operation)
return self._get_private_key(arg)
elif operation == 'verify':
self._check_constraints('sig', operation)
return self._get_public_key(arg)
elif operation == 'encrypt' or operation == 'wrapKey':
self._check_constraints('enc', operation)
return self._get_public_key(arg)
elif operation == 'decrypt' or operation == 'unwrapKey':
self._check_constraints('enc', operation)
return self._get_private_key(arg)
else:
raise NotImplementedError
def import_from_pyca(self, key):
if isinstance(key, rsa.RSAPrivateKey):
self._import_pyca_pri_rsa(key)
elif isinstance(key, rsa.RSAPublicKey):
self._import_pyca_pub_rsa(key)
elif isinstance(key, ec.EllipticCurvePrivateKey):
self._import_pyca_pri_ec(key)
elif isinstance(key, ec.EllipticCurvePublicKey):
self._import_pyca_pub_ec(key)
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.
:param hashalg: A hash function (defaults to SHA256)
"""
t = {'kty': self._params['kty']}
for name, val in iteritems(JWKValuesRegistry[t['kty']]):
if val.required:
t[name] = self._key[name]
digest = hashes.Hash(hashalg, backend=default_backend())
digest.update(bytes(json_encode(t).encode('utf8')))
return base64url_encode(digest.finalize())
class _JWKkeys(set):
def add(self, elem):
"""Adds a JWK object to the set
:param elem: the JWK object to add.
:raises TypeError: if the object is not a JWK.
"""
if not isinstance(elem, JWK):
raise TypeError('Only JWK objects are valid elements')
set.add(self, elem)
class JWKSet(dict):
"""A set of JWK objects.
Inherits from the standard 'dict' bultin type.
Creates a special key 'keys' that is of a type derived from 'set'
The 'keys' attribute accepts only :class:`jwcrypto.jwk.JWK` elements.
"""
def __init__(self, *args, **kwargs):
super(JWKSet, self).__init__()
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)
else:
super(JWKSet, self).__setitem__(key, val)
def update(self, *args, **kwargs):
for k, v in iteritems(dict(*args, **kwargs)):
self.__setitem__(k, v)
def add(self, elem):
self['keys'].add(elem)
def export(self, private_keys=True):
"""Exports a RFC 7517 keyset using the standard JSON format
:param private_key(bool): Whether to export private keys.
Defaults to True.
"""
exp_dict = dict()
for k, v in iteritems(self):
if k == 'keys':
keys = list()
for jwk in v:
keys.append(json_decode(jwk.export(private_keys)))
v = keys
exp_dict[k] = v
return json_encode(exp_dict)
def import_keyset(self, keyset):
"""Imports a RFC 7517 keyset using the standard JSON format.
:param keyset: The RFC 7517 representation of a JOSE Keyset.
"""
try:
jwkset = json_decode(keyset)
except Exception: # pylint: disable=broad-except
raise InvalidJWKValue()
if 'keys' not in jwkset:
raise InvalidJWKValue()
for k, v in iteritems(jwkset):
if k == 'keys':
for jwk in v:
self['keys'].add(JWK(**jwk))
else:
self[k] = v
@classmethod
def from_json(cls, keyset):
"""Creates a RFC 7517 keyset from the standard JSON format.
:param keyset: The RFC 7517 representation of a JOSE Keyset.
"""
obj = cls()
obj.import_keyset(keyset)
return obj
def get_key(self, kid):
"""Gets a key from the set.
:param kid: the 'kid' key identifier.
"""
for jwk in self['keys']:
if jwk.key_id == kid:
return jwk
return None

611
jwcrypto/jws.py Normal file
View File

@ -0,0 +1,611 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
from collections import namedtuple
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?)
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 = [
'HS256', 'HS384', 'HS512',
'RS256', 'RS384', 'RS512',
'ES256', 'ES384', 'ES512',
'PS256', 'PS384', 'PS512']
"""Default allowed algorithms"""
class InvalidJWSSignature(JWException):
"""Invalid JWS Signature.
This exception is raised when a signature cannot be validated.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Unknown Signature Verification Failure'
if exception:
msg += ' {%s}' % str(exception)
super(InvalidJWSSignature, self).__init__(msg)
class InvalidJWSObject(JWException):
"""Invalid JWS Object.
This exception is raised when the JWS Object is invalid and/or
improperly formatted.
"""
def __init__(self, message=None, exception=None):
msg = 'Invalid JWS Object'
if message:
msg += ' [%s]' % message
if exception:
msg += ' {%s}' % str(exception)
super(InvalidJWSObject, self).__init__(msg)
class InvalidJWSOperation(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}' % str(exception)
super(InvalidJWSOperation, self).__init__(msg)
class JWSCore(object):
"""The inner JWS Core object.
This object SHOULD NOT be used directly, the JWS object should be
used instead as JWS perform necessary checks on the validity of
the object and requested operations.
"""
def __init__(self, alg, key, header, payload, algs=None):
"""Core JWS token handling.
:param alg: The algorithm used to produce the signature.
See RFC 7518
:param key: A (:class:`jwcrypto.jwk.JWK`) key of appropriate
type for the "alg" provided in the 'protected' json string.
:param header: A JSON string representing the protected header.
:param payload(bytes): An arbitrary value
:param algs: An optional list of allowed algorithms
:raises ValueError: if the key is not a :class:`JWK` object
:raises InvalidJWAAlgorithm: if the algorithm is not valid, is
unknown or otherwise not yet implemented.
"""
self.alg = alg
self.engine = self._jwa(alg, algs)
if not isinstance(key, JWK):
raise ValueError('key is not a JWK 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 = payload
def _jwa(self, name, allowed):
if allowed is None:
allowed = default_allowed_algs
if name not in allowed:
raise InvalidJWSOperation('Algorithm not allowed')
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"""
payload = self._payload()
sigin = b'.'.join([self.protected.encode('utf-8'), payload])
signature = self.engine.sign(self.key, sigin)
return {'protected': self.protected,
'payload': payload,
'signature': base64url_encode(signature)}
def verify(self, signature):
"""Verifies a signature
:raises InvalidJWSSignature: if the verification fails.
"""
try:
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))
return True
class JWS(object):
"""JSON Web Signature object
This object represent a JWS token.
"""
def __init__(self, payload=None):
"""Creates a JWS object.
:param payload(bytes): An arbitrary value (optional).
"""
self.objects = dict()
if payload:
self.objects['payload'] = payload
self.verifylog = None
self._allowed_algs = None
@property
def allowed_algs(self):
"""Allowed algorithms.
The list of allowed algorithms.
Can be changed by setting a list of algorithm names.
"""
if self._allowed_algs:
return self._allowed_algs
else:
return default_allowed_algs
@allowed_algs.setter
def allowed_algs(self, algs):
if not isinstance(algs, list):
raise TypeError('Allowed Algs must be a list')
self._allowed_algs = algs
@property
def is_valid(self):
return self.objects.get('valid', False)
# 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):
p = dict()
# verify it is a valid JSON object and decode
if protected is not None:
p = json_decode(protected)
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')
# 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')
if alg:
if 'alg' in p and alg != p['alg']:
raise InvalidJWSSignature('"alg" mismatch, requested '
'"%s", found "%s"' % (alg,
p['alg']))
a = alg
else:
a = p['alg']
# the following will verify the "alg" is supported and the signature
# verifies
c = JWSCore(a, key, protected, payload, self._allowed_algs)
c.verify(signature)
def verify(self, key, alg=None):
"""Verifies a JWS token.
:param key: The (:class:`jwcrypto.jwk.JWK`) verification key.
:param alg: The signing algorithm (optional). usually the algorithm
is known as it is provided with the JOSE Headers of the token.
:raises InvalidJWSSignature: if the verification fails.
"""
self.verifylog = list()
self.objects['valid'] = False
obj = self.objects
if 'signature' in obj:
try:
self._verify(alg, key,
obj['payload'],
obj['signature'],
obj.get('protected', None),
obj.get('header', None))
obj['valid'] = True
except Exception as e: # pylint: disable=broad-except
self.verifylog.append('Failed: [%s]' % repr(e))
elif 'signatures' in obj:
for o in obj['signatures']:
try:
self._verify(alg, key,
obj['payload'],
o['signature'],
o.get('protected', None),
o.get('header', None))
# Ok if at least one verifies
obj['valid'] = True
except Exception as e: # pylint: disable=broad-except
self.verifylog.append('Failed: [%s]' % repr(e))
else:
raise InvalidJWSSignature('No signatures availble')
if not self.is_valid:
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.
NOTE: Destroys any current status and tries to import the raw
JWS provided.
:param raw_jws: a 'raw' JWS token (JSON Encoded or Compact
notation) string.
:param key: A (:class:`jwcrypto.jwk.JWK`) verification key (optional).
If a key is provided a verification step will be attempted after
the object is successfully deserialized.
:param alg: The signing algorithm (optional). usually the algorithm
is known as it is provided with the JOSE Headers of the token.
:raises InvalidJWSObject: if the raw object is an invaid JWS token.
:raises InvalidJWSSignature: if the verification fails.
"""
self.objects = dict()
o = dict()
try:
try:
djws = json_decode(raw_jws)
if 'signatures' in djws:
o['signatures'] = list()
for s in djws['signatures']:
os = self._deserialize_signature(s)
o['signatures'].append(os)
self._deserialize_b64(o, os.get('protected'))
else:
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('.')
if len(c) != 3:
raise InvalidJWSObject('Unrecognized representation')
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]))
self.objects = o
except Exception as e: # pylint: disable=broad-except
raise InvalidJWSObject('Invalid format', repr(e))
if key:
self.verify(key, alg)
def add_signature(self, key, alg=None, protected=None, header=None):
"""Adds a new signature to the object.
:param key: A (:class:`jwcrypto.jwk.JWK`) key of appropriate for
the "alg" provided.
:param alg: An optional algorithm name. If already provided as an
element of the protected or unprotected header it can be safely
omitted.
:param potected: The Protected Header (optional)
:param header: The Unprotected Header (optional)
: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.
:raises InvalidJWAAlgorithm: if the algorithm is not valid, is
unknown or otherwise not yet implemented.
"""
if not self.objects.get('payload', None):
raise InvalidJWSObject('Missing Payload')
b64 = True
p = dict()
if protected:
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:
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:
alg = p['alg']
elif alg != p['alg']:
raise ValueError('"alg" value mismatch, specified "alg" '
'does not match JOSE header value')
if alg is None:
raise ValueError('"alg" not specified')
c = JWSCore(alg, key, protected, self.objects['payload'])
sig = c.sign()
o = dict()
o['signature'] = base64url_decode(sig['signature'])
if protected:
o['protected'] = protected
if header:
o['header'] = h
o['valid'] = True
if 'signatures' in self.objects:
self.objects['signatures'].append(o)
elif 'signature' in self.objects:
self.objects['signatures'] = list()
n = dict()
n['signature'] = self.objects.pop('signature')
if 'protected' in self.objects:
n['protected'] = self.objects.pop('protected')
if 'header' in self.objects:
n['header'] = self.objects.pop('header')
if 'valid' in self.objects:
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.
:param compact(boolean): if True generates the compact
representation, otherwise generates a standard JSON format.
:raises InvalidJWSOperation: if the object cannot serialized
with the compact representation and `compat` is True.
: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 "
"multiple signatures")
if 'signature' not in self.objects:
raise InvalidJWSSignature("No available signature")
if not self.objects.get('valid', False):
raise InvalidJWSSignature("No valid signature found")
if 'protected' in self.objects:
protected = base64url_encode(self.objects['protected'])
else:
protected = ''
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['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['signatures'] = list()
for o in obj['signatures']:
if not o.get('valid', False):
continue
s = {'signature': base64url_encode(o['signature'])}
if 'protected' in o:
s['protected'] = base64url_encode(o['protected'])
if 'header' in o:
s['header'] = o['header']
sig['signatures'].append(s)
if len(sig['signatures']) == 0:
raise InvalidJWSSignature("No valid signature found")
else:
raise InvalidJWSSignature("No available signature")
return json_encode(sig)
@property
def payload(self):
if 'payload' not in self.objects:
raise InvalidJWSOperation("Payload not available")
if not self.is_valid:
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:
if 'protected' in obj:
p = json_decode(obj['protected'])
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 o:
p = json_decode(o['protected'])
else:
p = None
jh = self._merge_check_headers(p, o.get('header', dict()))
jhl.append(jh)
return jhl
else:
raise InvalidJWSOperation("JOSE Header(s) not available")

506
jwcrypto/jwt.py Normal file
View File

@ -0,0 +1,506 @@
# Copyright (C) 2015 JWCrypto Project Contributors - see LICENSE file
import time
import uuid
from six import string_types
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
# RFC 7519 - 4.1
# name: description
JWTClaimsRegistry = {'iss': 'Issuer',
'sub': 'Subject',
'aud': 'Audience',
'exp': 'Expiration Time',
'nbf': 'Not Before',
'iat': 'Issued At',
'jti': 'JWT ID'}
class JWTExpired(JWException):
"""Json Web Token is expired.
This exception is raised when a token is expired accoring to its claims.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Token expired'
if exception:
msg += ' {%s}' % str(exception)
super(JWTExpired, self).__init__(msg)
class JWTNotYetValid(JWException):
"""Json Web Token is not yet valid.
This exception is raised when a token is not valid yet according to its
claims.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Token not yet valid'
if exception:
msg += ' {%s}' % str(exception)
super(JWTNotYetValid, self).__init__(msg)
class JWTMissingClaim(JWException):
"""Json Web Token claim is invalid.
This exception is raised when a claim does not match the expected value.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Invalid Claim Value'
if exception:
msg += ' {%s}' % str(exception)
super(JWTMissingClaim, self).__init__(msg)
class JWTInvalidClaimValue(JWException):
"""Json Web Token claim is invalid.
This exception is raised when a claim does not match the expected value.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Invalid Claim Value'
if exception:
msg += ' {%s}' % str(exception)
super(JWTInvalidClaimValue, self).__init__(msg)
class JWTInvalidClaimFormat(JWException):
"""Json Web Token claim format is invalid.
This exception is raised when a claim is not in a valid format.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Invalid Claim Format'
if exception:
msg += ' {%s}' % str(exception)
super(JWTInvalidClaimFormat, self).__init__(msg)
class JWTMissingKeyID(JWException):
"""Json Web Token is missing key id.
This exception is raised when trying to decode a JWT with a key set
that does not have a kid value in its header.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Missing Key ID'
if exception:
msg += ' {%s}' % str(exception)
super(JWTMissingKeyID, self).__init__(msg)
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
in the passed key set.
"""
def __init__(self, message=None, exception=None):
msg = None
if message:
msg = str(message)
else:
msg = 'Missing Key'
if exception:
msg += ' {%s}' % str(exception)
super(JWTMissingKey, self).__init__(msg)
class JWT(object):
"""JSON Web token object
This object represent a generic token.
"""
def __init__(self, header=None, claims=None, jwt=None, key=None,
algs=None, default_claims=None, check_claims=None):
"""Creates a JWT object.
:param header: A dict or a JSON string with the JWT Header 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:`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
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.
Note: either the header,claims or jwt,key parameters should be
provided as a deserialization operation (which occurs if the jwt
is provided will wipe any header os claim provided by setting
those obtained from the deserialization of the jwt token.
Note: if check_claims is not provided the 'exp' and 'nbf' claims
are checked if they are set on the token but not enforced if not
set. Any other RFC 7519 registered claims are checked only for
format conformance.
"""
self._header = None
self._claims = None
self._token = None
self._algs = algs
self._reg_claims = None
self._check_claims = None
self._leeway = 60 # 1 minute clock skew allowed
self._validity = 600 # 10 minutes validity (up to 11 with leeway)
if header:
self.header = header
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)
@property
def header(self):
if self._header is None:
raise KeyError("'header' not set")
return self._header
@header.setter
def header(self, h):
if isinstance(h, dict):
eh = json_encode(h)
else:
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):
if self._claims is None:
raise KeyError("'claims' not set")
return self._claims
@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)
else:
self._claims = c
@property
def token(self):
return self._token
@token.setter
def token(self, t):
if isinstance(t, JWS) or isinstance(t, JWE) or isinstance(t, JWT):
self._token = t
else:
raise TypeError("Invalid token type, must be one of JWS,JWE,JWT")
@property
def leeway(self):
return self._leeway
@leeway.setter
def leeway(self, l):
self._leeway = int(l)
@property
def validity(self):
return self._validity
@validity.setter
def validity(self, v):
self._validity = int(v)
def _add_optional_claim(self, name, claims):
if name in claims:
return
val = self._reg_claims.get(name, None)
if val is not None:
claims[name] = val
def _add_time_claim(self, name, claims, defval):
if name in claims:
return
if name in self._reg_claims:
if self._reg_claims[name] is None:
claims[name] = defval
else:
claims[name] = self._reg_claims[name]
def _add_jti_claim(self, claims):
if 'jti' in claims or 'jti' not in self._reg_claims:
return
claims['jti'] = str(uuid.uuid4())
def _add_default_claims(self, claims):
if self._reg_claims is None:
return
now = int(time.time())
self._add_optional_claim('iss', claims)
self._add_optional_claim('sub', claims)
self._add_optional_claim('aud', claims)
self._add_time_claim('exp', claims, now + self.validity)
self._add_time_claim('nbf', claims, now)
self._add_time_claim('iat', claims, now)
self._add_jti_claim(claims)
def _check_string_claim(self, name, claims):
if name not in claims:
return
if not isinstance(claims[name], string_types):
raise JWTInvalidClaimFormat("Claim %s is not a StringOrURI type")
def _check_array_or_string_claim(self, name, claims):
if name not in claims:
return
if isinstance(claims[name], list):
if any(not isinstance(claim, string_types) for claim in claims):
raise JWTInvalidClaimFormat(
"Claim %s contains non StringOrURI types" % (name, ))
elif not isinstance(claims[name], string_types):
raise JWTInvalidClaimFormat(
"Claim %s is not a StringOrURI type" % (name, ))
def _check_integer_claim(self, name, claims):
if name not in claims:
return
try:
int(claims[name])
except ValueError:
raise JWTInvalidClaimFormat(
"Claim %s is not an integer" % (name, ))
def _check_exp(self, claim, limit, leeway):
if claim < limit - leeway:
raise JWTExpired('Expired at %d, time: %d(leeway: %d)' % (
claim, limit, leeway))
def _check_nbf(self, claim, limit, leeway):
if claim > limit + leeway:
raise JWTNotYetValid('Valid from %d, time: %d(leeway: %d)' % (
claim, limit, leeway))
def _check_default_claims(self, claims):
self._check_string_claim('iss', claims)
self._check_string_claim('sub', claims)
self._check_array_or_string_claim('aud', claims)
self._check_integer_claim('exp', claims)
self._check_integer_claim('nbf', claims)
self._check_integer_claim('iat', claims)
self._check_string_claim('jti', claims)
if self._check_claims is None:
if 'exp' in claims:
self._check_exp(claims['exp'], time.time(), self._leeway)
if 'nbf' in claims:
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
if self._check_claims is False:
return
try:
claims = json_decode(self.claims)
if not isinstance(claims, dict):
raise ValueError()
except ValueError:
if self._check_claims is not None:
raise JWTInvalidClaimFormat(
"Claims check requested but claims is not a json dict")
return
self._check_default_claims(claims)
if self._check_claims is None:
return
for name, value in self._check_claims.items():
if name not in claims:
raise JWTMissingClaim("Claim %s is missing" % (name, ))
if name in ['iss', 'sub', 'jti']:
if value is not None and value != claims[name]:
raise JWTInvalidClaimValue(
"Invalid '%s' value. Expected '%s' got '%s'" % (
name, value, claims[name]))
elif name == 'aud':
if value is not None:
if value == claims[name]:
continue
if isinstance(claims[name], list):
if value in claims[name]:
continue
raise JWTInvalidClaimValue(
"Invalid '%s' value. Expected '%s' to be in '%s'" % (
name, claims[name], value))
elif name == 'exp':
if value is not None:
self._check_exp(claims[name], value, 0)
else:
self._check_exp(claims[name], time.time(), self._leeway)
elif name == 'nbf':
if value is not None:
self._check_nbf(claims[name], value, 0)
else:
self._check_nbf(claims[name], time.time(), self._leeway)
else:
if value is not None and value != claims[name]:
raise JWTInvalidClaimValue(
"Invalid '%s' value. Expected '%s' got '%s'" % (
name, value, claims[name]))
def make_signed_token(self, key):
"""Signs the payload.
Creates a JWS token with the header as the JWS protected header and
the claims as the payload. See (:class:`jwcrypto.jws.JWS`) for
details on the exceptions that may be reaised.
:param key: A (:class:`jwcrypto.jwk.JWK`) key.
"""
t = JWS(self.claims)
t.add_signature(key, protected=self.header)
self.token = t
def make_encrypted_token(self, key):
"""Encrypts the payload.
Creates a JWE token with the header as the JWE protected header and
the claims as the plaintext. See (:class:`jwcrypto.jwe.JWE`) for
details on the exceptions that may be reaised.
:param key: A (:class:`jwcrypto.jwk.JWK`) key.
"""
t = JWE(self.claims, self.header)
t.add_recipient(key)
self.token = t
def deserialize(self, jwt, key=None):
"""Deserialize a JWT token.
NOTE: Destroys any current status and tries to import the raw
token provided.
:param jwt: a 'raw' JWT token.
:param key: A (:class:`jwcrypto.jwk.JWK`) verification or
decryption key, or a (:class:`jwcrypto.jwk.JWKSet`) that
contains a key indexed by the 'kid' header.
"""
c = jwt.count('.')
if c == 2:
self.token = JWS()
elif c == 4:
self.token = JWE()
else:
raise ValueError("Token format unrecognized")
# Apply algs restrictions if any, before performing any operation
if self._algs:
self.token.allowed_algs = self._algs
# now deserialize and also decrypt/verify (or raise) if we
# have a key
if key is None:
self.token.deserialize(jwt, None)
elif isinstance(key, JWK):
self.token.deserialize(jwt, key)
elif isinstance(key, JWKSet):
self.token.deserialize(jwt, None)
if 'kid' not in self.token.jose_header:
raise JWTMissingKeyID('No key ID in JWT header')
token_key = key.get_key(self.token.jose_header['kid'])
if not token_key:
raise JWTMissingKey('Key ID %s not in key set'
% self.token.jose_header['kid'])
if isinstance(self.token, JWE):
self.token.decrypt(token_key)
elif isinstance(self.token, JWS):
self.token.verify(token_key)
else:
raise RuntimeError("Unknown Token Type")
else:
raise ValueError("Unrecognized Key Type")
if key is not None:
self.header = self.token.jose_header
self.claims = self.token.payload.decode('utf-8')
self._check_provided_claims()
def serialize(self, compact=True):
"""Serializes the object into a JWS token.
:param compact(boolean): must be True.
Note: the compact parameter is provided for general compatibility
with the serialize() functions of :class:`jwcrypto.jws.JWS` and
:class:`jwcrypto.jwe.JWE` so that these objects can all be used
interchangeably. However the only valid JWT representtion is the
compact representation.
"""
return self.token.serialize(compact)

1293
jwcrypto/tests-cookbook.py Normal file

File diff suppressed because it is too large Load Diff

1332
jwcrypto/tests.py Normal file

File diff suppressed because it is too large Load Diff

6
setup.cfg Normal file
View File

@ -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

30
setup.py Executable file
View File

@ -0,0 +1,30 @@
#!/usr/bin/python
#
# Copyright (C) 2015 JWCrypto Project Contributors, see LICENSE file
from setuptools import setup
setup(
name = 'jwcrypto',
version = '0.6.0',
license = 'LGPLv3+',
maintainer = 'JWCrypto Project Contributors',
maintainer_email = 'simo@redhat.com',
url='https://github.com/latchset/jwcrypto',
packages = ['jwcrypto'],
description = 'Implementation of JOSE Web standards',
classifiers = [
'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',
],
)

68
tox.ini Normal file
View File

@ -0,0 +1,68 @@
[tox]
envlist = lint,py27,py34,py35,py36,py37,pep8py2,pep8py3,doc,sphinx
skip_missing_interpreters = true
[testenv]
setenv =
PYTHONPATH = {envsitepackagesdir}
deps =
pytest
coverage
sitepackages = True
commands =
{envpython} -bb -m coverage run -m pytest --capture=no --strict {posargs}
{envpython} -m coverage report -m
[testenv:lint]
basepython = python2.7
deps =
pylint
sitepackages = True
commands =
{envpython} -m pylint -d c,r,i,W0613 -r n -f colorized --notes= --disable=star-args ./jwcrypto
[testenv:pep8py2]
basepython = python2.7
deps =
flake8
flake8-import-order
pep8-naming
commands =
{envpython} -m flake8 {posargs} jwcrypto
[testenv:pep8py3]
basepython = python3
deps =
flake8
flake8-import-order
pep8-naming
commands =
{envpython} -m flake8 {posargs} jwcrypto
[testenv:doc]
deps =
doc8
docutils
markdown
basepython = python2.7
commands =
doc8 --allow-long-titles README.md
markdown_py README.md -f {toxworkdir}/README.md.html
[testenv:sphinx]
basepython = python2.7
changedir = docs/source
deps =
sphinx < 1.3.0
commands =
sphinx-build -v -W -b html -d {envtmpdir}/doctrees . {envtmpdir}/html
[pytest]
python_files = jwcrypto/test*.py
[flake8]
exclude = .tox,*.egg,dist,build,docs/source
show-source = true
max-line-length = 79
ignore = N802
application-import-names = jwcrypto