diff options
| author | alexv-smirnov <alex@ydb.tech> | 2023-12-01 12:02:50 +0300 |
|---|---|---|
| committer | alexv-smirnov <alex@ydb.tech> | 2023-12-01 13:28:10 +0300 |
| commit | 0e578a4c44d4abd539d9838347b9ebafaca41dfb (patch) | |
| tree | a0c1969c37f818c830ebeff9c077eacf30be6ef8 /contrib/python/rsa/py2 | |
| parent | 84f2d3d4cc985e63217cff149bd2e6d67ae6fe22 (diff) | |
| download | ydb-0e578a4c44d4abd539d9838347b9ebafaca41dfb.tar.gz | |
Change "ya.make"
Diffstat (limited to 'contrib/python/rsa/py2')
38 files changed, 4711 insertions, 0 deletions
diff --git a/contrib/python/rsa/py2/.dist-info/METADATA b/contrib/python/rsa/py2/.dist-info/METADATA new file mode 100644 index 0000000000..3ab66353aa --- /dev/null +++ b/contrib/python/rsa/py2/.dist-info/METADATA @@ -0,0 +1,85 @@ +Metadata-Version: 2.1 +Name: rsa +Version: 4.5 +Summary: Pure-Python RSA implementation +Home-page: https://stuvel.eu/rsa +Author: Sybren A. Stuvel +Author-email: sybren@stuvel.eu +Maintainer: Sybren A. Stuvel +Maintainer-email: sybren@stuvel.eu +License: ASL 2 +Platform: UNKNOWN +Classifier: Development Status :: 5 - Production/Stable +Classifier: Intended Audience :: Developers +Classifier: Intended Audience :: Education +Classifier: Intended Audience :: Information Technology +Classifier: License :: OSI Approved :: Apache Software License +Classifier: Operating System :: OS Independent +Classifier: Programming Language :: Python +Classifier: Programming Language :: Python :: 2 +Classifier: Programming Language :: Python :: 2.7 +Classifier: Programming Language :: Python :: 3 +Classifier: Programming Language :: Python :: 3.5 +Classifier: Programming Language :: Python :: 3.6 +Classifier: Programming Language :: Python :: 3.7 +Classifier: Programming Language :: Python :: Implementation :: CPython +Classifier: Programming Language :: Python :: Implementation :: PyPy +Classifier: Topic :: Security :: Cryptography +Requires-Python: >=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*, <4 +Description-Content-Type: text/markdown +Requires-Dist: pyasn1 (>=0.1.3) + +Pure Python RSA implementation +============================== + +[](https://pypi.org/project/rsa/) +[](https://travis-ci.org/sybrenstuvel/python-rsa) +[](https://coveralls.io/github/sybrenstuvel/python-rsa?branch=master) +[](https://codeclimate.com/github/sybrenstuvel/python-rsa) + +[Python-RSA](https://stuvel.eu/rsa) is a pure-Python RSA implementation. It supports +encryption and decryption, signing and verifying signatures, and key +generation according to PKCS#1 version 1.5. It can be used as a Python +library as well as on the commandline. The code was mostly written by +Sybren A. Stüvel. + +Documentation can be found at the [Python-RSA homepage](https://stuvel.eu/rsa). + +Download and install using: + + pip install rsa + +or download it from the [Python Package Index](https://pypi.org/project/rsa/). + +The source code is maintained at [GitHub](https://github.com/sybrenstuvel/python-rsa/) and is +licensed under the [Apache License, version 2.0](https://www.apache.org/licenses/LICENSE-2.0) + +Changes in 4.1-4.4 +------------------ + +Version 4.1 dropped support for Python 2.7, and soon after that version 4.2 was released. Neither of the two made it explicit in `setup.cfg` that Python 3.5 or newer is required. This caused issues on Python 2.7, as Pip happily upgraded to the new version. + +Version 4.3 is a re-tagged release of version 4.0. It is the last to support Python 2.7. + +Version 4.4 will be a re-tagged release of version 4.2, and explicitly require Python 3.5 or newer. + + +Major changes in 4.0 +-------------------- + +Version 3.4 was the last version in the 3.x range. Version 4.0 drops the following modules, +as they are insecure: + +- `rsa._version133` +- `rsa._version200` +- `rsa.bigfile` +- `rsa.varblock` + +Those modules were marked as deprecated in version 3.4. + +Furthermore, in 4.0 the I/O functions is streamlined to always work with bytes on all +supported versions of Python. + +Version 4.0 drops support for Python 2.6 and 3.3. + + diff --git a/contrib/python/rsa/py2/.dist-info/entry_points.txt b/contrib/python/rsa/py2/.dist-info/entry_points.txt new file mode 100644 index 0000000000..1c27571272 --- /dev/null +++ b/contrib/python/rsa/py2/.dist-info/entry_points.txt @@ -0,0 +1,8 @@ +[console_scripts] +pyrsa-decrypt = rsa.cli:decrypt +pyrsa-encrypt = rsa.cli:encrypt +pyrsa-keygen = rsa.cli:keygen +pyrsa-priv2pub = rsa.util:private_to_public +pyrsa-sign = rsa.cli:sign +pyrsa-verify = rsa.cli:verify + diff --git a/contrib/python/rsa/py2/.dist-info/top_level.txt b/contrib/python/rsa/py2/.dist-info/top_level.txt new file mode 100644 index 0000000000..703f551006 --- /dev/null +++ b/contrib/python/rsa/py2/.dist-info/top_level.txt @@ -0,0 +1 @@ +rsa diff --git a/contrib/python/rsa/py2/LICENSE b/contrib/python/rsa/py2/LICENSE new file mode 100644 index 0000000000..67589cbb86 --- /dev/null +++ b/contrib/python/rsa/py2/LICENSE @@ -0,0 +1,13 @@ +Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. diff --git a/contrib/python/rsa/py2/README.md b/contrib/python/rsa/py2/README.md new file mode 100644 index 0000000000..b7df4fbde4 --- /dev/null +++ b/contrib/python/rsa/py2/README.md @@ -0,0 +1,52 @@ +Pure Python RSA implementation +============================== + +[](https://pypi.org/project/rsa/) +[](https://travis-ci.org/sybrenstuvel/python-rsa) +[](https://coveralls.io/github/sybrenstuvel/python-rsa?branch=master) +[](https://codeclimate.com/github/sybrenstuvel/python-rsa) + +[Python-RSA](https://stuvel.eu/rsa) is a pure-Python RSA implementation. It supports +encryption and decryption, signing and verifying signatures, and key +generation according to PKCS#1 version 1.5. It can be used as a Python +library as well as on the commandline. The code was mostly written by +Sybren A. Stüvel. + +Documentation can be found at the [Python-RSA homepage](https://stuvel.eu/rsa). + +Download and install using: + + pip install rsa + +or download it from the [Python Package Index](https://pypi.org/project/rsa/). + +The source code is maintained at [GitHub](https://github.com/sybrenstuvel/python-rsa/) and is +licensed under the [Apache License, version 2.0](https://www.apache.org/licenses/LICENSE-2.0) + +Changes in 4.1-4.4 +------------------ + +Version 4.1 dropped support for Python 2.7, and soon after that version 4.2 was released. Neither of the two made it explicit in `setup.cfg` that Python 3.5 or newer is required. This caused issues on Python 2.7, as Pip happily upgraded to the new version. + +Version 4.3 is a re-tagged release of version 4.0. It is the last to support Python 2.7. + +Version 4.4 will be a re-tagged release of version 4.2, and explicitly require Python 3.5 or newer. + + +Major changes in 4.0 +-------------------- + +Version 3.4 was the last version in the 3.x range. Version 4.0 drops the following modules, +as they are insecure: + +- `rsa._version133` +- `rsa._version200` +- `rsa.bigfile` +- `rsa.varblock` + +Those modules were marked as deprecated in version 3.4. + +Furthermore, in 4.0 the I/O functions is streamlined to always work with bytes on all +supported versions of Python. + +Version 4.0 drops support for Python 2.6 and 3.3. diff --git a/contrib/python/rsa/py2/rsa/__init__.py b/contrib/python/rsa/py2/rsa/__init__.py new file mode 100644 index 0000000000..af5487f3d3 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/__init__.py @@ -0,0 +1,42 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""RSA module + +Module for calculating large primes, and RSA encryption, decryption, signing +and verification. Includes generating public and private keys. + +WARNING: this implementation does not use compression of the cleartext input to +prevent repetitions, or other common security improvements. Use with care. + +""" + +from rsa.key import newkeys, PrivateKey, PublicKey +from rsa.pkcs1 import encrypt, decrypt, sign, verify, DecryptionError, \ + VerificationError, find_signature_hash, sign_hash, compute_hash + +__author__ = "Sybren Stuvel, Barry Mead and Yesudeep Mangalapilly" +__date__ = "2020-06-12" +__version__ = '4.5' + +# Do doctest if we're run directly +if __name__ == "__main__": + import doctest + + doctest.testmod() + +__all__ = ["newkeys", "encrypt", "decrypt", "sign", "verify", 'PublicKey', + 'PrivateKey', 'DecryptionError', 'VerificationError', + 'compute_hash', 'sign_hash'] diff --git a/contrib/python/rsa/py2/rsa/_compat.py b/contrib/python/rsa/py2/rsa/_compat.py new file mode 100644 index 0000000000..71197a55b8 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/_compat.py @@ -0,0 +1,162 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Python compatibility wrappers.""" + +from __future__ import absolute_import + +import itertools +import sys +from struct import pack + +MAX_INT = sys.maxsize +MAX_INT64 = (1 << 63) - 1 +MAX_INT32 = (1 << 31) - 1 +MAX_INT16 = (1 << 15) - 1 + +PY2 = sys.version_info[0] == 2 + +# Determine the word size of the processor. +if MAX_INT == MAX_INT64: + # 64-bit processor. + MACHINE_WORD_SIZE = 64 +elif MAX_INT == MAX_INT32: + # 32-bit processor. + MACHINE_WORD_SIZE = 32 +else: + # Else we just assume 64-bit processor keeping up with modern times. + MACHINE_WORD_SIZE = 64 + +if PY2: + integer_types = (int, long) + range = xrange + zip = itertools.izip +else: + integer_types = (int, ) + range = range + zip = zip + + +def write_to_stdout(data): + """Writes bytes to stdout + + :type data: bytes + """ + if PY2: + sys.stdout.write(data) + else: + # On Py3 we must use the buffer interface to write bytes. + sys.stdout.buffer.write(data) + + +def is_bytes(obj): + """ + Determines whether the given value is a byte string. + + :param obj: + The value to test. + :returns: + ``True`` if ``value`` is a byte string; ``False`` otherwise. + """ + return isinstance(obj, bytes) + + +def is_integer(obj): + """ + Determines whether the given value is an integer. + + :param obj: + The value to test. + :returns: + ``True`` if ``value`` is an integer; ``False`` otherwise. + """ + return isinstance(obj, integer_types) + + +def byte(num): + """ + Converts a number between 0 and 255 (both inclusive) to a base-256 (byte) + representation. + + Use it as a replacement for ``chr`` where you are expecting a byte + because this will work on all current versions of Python:: + + :param num: + An unsigned integer between 0 and 255 (both inclusive). + :returns: + A single byte. + """ + return pack("B", num) + + +def xor_bytes(b1, b2): + """ + Returns the bitwise XOR result between two bytes objects, b1 ^ b2. + + Bitwise XOR operation is commutative, so order of parameters doesn't + generate different results. If parameters have different length, extra + length of the largest one is ignored. + + :param b1: + First bytes object. + :param b2: + Second bytes object. + :returns: + Bytes object, result of XOR operation. + """ + if PY2: + return ''.join(byte(ord(x) ^ ord(y)) for x, y in zip(b1, b2)) + + return bytes(x ^ y for x, y in zip(b1, b2)) + + +def get_word_alignment(num, force_arch=64, + _machine_word_size=MACHINE_WORD_SIZE): + """ + Returns alignment details for the given number based on the platform + Python is running on. + + :param num: + Unsigned integral number. + :param force_arch: + If you don't want to use 64-bit unsigned chunks, set this to + anything other than 64. 32-bit chunks will be preferred then. + Default 64 will be used when on a 64-bit machine. + :param _machine_word_size: + (Internal) The machine word size used for alignment. + :returns: + 4-tuple:: + + (word_bits, word_bytes, + max_uint, packing_format_type) + """ + max_uint64 = 0xffffffffffffffff + max_uint32 = 0xffffffff + max_uint16 = 0xffff + max_uint8 = 0xff + + if force_arch == 64 and _machine_word_size >= 64 and num > max_uint32: + # 64-bit unsigned integer. + return 64, 8, max_uint64, "Q" + elif num > max_uint16: + # 32-bit unsigned integer + return 32, 4, max_uint32, "L" + elif num > max_uint8: + # 16-bit unsigned integer. + return 16, 2, max_uint16, "H" + else: + # 8-bit unsigned integer. + return 8, 1, max_uint8, "B" diff --git a/contrib/python/rsa/py2/rsa/asn1.py b/contrib/python/rsa/py2/rsa/asn1.py new file mode 100644 index 0000000000..b724b8f53d --- /dev/null +++ b/contrib/python/rsa/py2/rsa/asn1.py @@ -0,0 +1,53 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""ASN.1 definitions. + +Not all ASN.1-handling code use these definitions, but when it does, they should be here. +""" + +from pyasn1.type import univ, namedtype, tag + + +class PubKeyHeader(univ.Sequence): + componentType = namedtype.NamedTypes( + namedtype.NamedType('oid', univ.ObjectIdentifier()), + namedtype.NamedType('parameters', univ.Null()), + ) + + +class OpenSSLPubKey(univ.Sequence): + componentType = namedtype.NamedTypes( + namedtype.NamedType('header', PubKeyHeader()), + + # This little hack (the implicit tag) allows us to get a Bit String as Octet String + namedtype.NamedType('key', univ.OctetString().subtype( + implicitTag=tag.Tag(tagClass=0, tagFormat=0, tagId=3))), + ) + + +class AsnPubKey(univ.Sequence): + """ASN.1 contents of DER encoded public key: + + RSAPublicKey ::= SEQUENCE { + modulus INTEGER, -- n + publicExponent INTEGER, -- e + """ + + componentType = namedtype.NamedTypes( + namedtype.NamedType('modulus', univ.Integer()), + namedtype.NamedType('publicExponent', univ.Integer()), + ) diff --git a/contrib/python/rsa/py2/rsa/cli.py b/contrib/python/rsa/py2/rsa/cli.py new file mode 100644 index 0000000000..6450af427f --- /dev/null +++ b/contrib/python/rsa/py2/rsa/cli.py @@ -0,0 +1,288 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Commandline scripts. + +These scripts are called by the executables defined in setup.py. +""" + +from __future__ import with_statement, print_function + +import abc +import sys +from optparse import OptionParser + +import rsa +import rsa.pkcs1 + +HASH_METHODS = sorted(rsa.pkcs1.HASH_METHODS.keys()) + + +def keygen(): + """Key generator.""" + + # Parse the CLI options + parser = OptionParser(usage='usage: %prog [options] keysize', + description='Generates a new RSA keypair of "keysize" bits.') + + parser.add_option('--pubout', type='string', + help='Output filename for the public key. The public key is ' + 'not saved if this option is not present. You can use ' + 'pyrsa-priv2pub to create the public key file later.') + + parser.add_option('-o', '--out', type='string', + help='Output filename for the private key. The key is ' + 'written to stdout if this option is not present.') + + parser.add_option('--form', + help='key format of the private and public keys - default PEM', + choices=('PEM', 'DER'), default='PEM') + + (cli, cli_args) = parser.parse_args(sys.argv[1:]) + + if len(cli_args) != 1: + parser.print_help() + raise SystemExit(1) + + try: + keysize = int(cli_args[0]) + except ValueError: + parser.print_help() + print('Not a valid number: %s' % cli_args[0], file=sys.stderr) + raise SystemExit(1) + + print('Generating %i-bit key' % keysize, file=sys.stderr) + (pub_key, priv_key) = rsa.newkeys(keysize) + + # Save public key + if cli.pubout: + print('Writing public key to %s' % cli.pubout, file=sys.stderr) + data = pub_key.save_pkcs1(format=cli.form) + with open(cli.pubout, 'wb') as outfile: + outfile.write(data) + + # Save private key + data = priv_key.save_pkcs1(format=cli.form) + + if cli.out: + print('Writing private key to %s' % cli.out, file=sys.stderr) + with open(cli.out, 'wb') as outfile: + outfile.write(data) + else: + print('Writing private key to stdout', file=sys.stderr) + rsa._compat.write_to_stdout(data) + + +class CryptoOperation(object): + """CLI callable that operates with input, output, and a key.""" + + __metaclass__ = abc.ABCMeta + + keyname = 'public' # or 'private' + usage = 'usage: %%prog [options] %(keyname)s_key' + description = None + operation = 'decrypt' + operation_past = 'decrypted' + operation_progressive = 'decrypting' + input_help = 'Name of the file to %(operation)s. Reads from stdin if ' \ + 'not specified.' + output_help = 'Name of the file to write the %(operation_past)s file ' \ + 'to. Written to stdout if this option is not present.' + expected_cli_args = 1 + has_output = True + + key_class = rsa.PublicKey + + def __init__(self): + self.usage = self.usage % self.__class__.__dict__ + self.input_help = self.input_help % self.__class__.__dict__ + self.output_help = self.output_help % self.__class__.__dict__ + + @abc.abstractmethod + def perform_operation(self, indata, key, cli_args): + """Performs the program's operation. + + Implement in a subclass. + + :returns: the data to write to the output. + """ + + def __call__(self): + """Runs the program.""" + + (cli, cli_args) = self.parse_cli() + + key = self.read_key(cli_args[0], cli.keyform) + + indata = self.read_infile(cli.input) + + print(self.operation_progressive.title(), file=sys.stderr) + outdata = self.perform_operation(indata, key, cli_args) + + if self.has_output: + self.write_outfile(outdata, cli.output) + + def parse_cli(self): + """Parse the CLI options + + :returns: (cli_opts, cli_args) + """ + + parser = OptionParser(usage=self.usage, description=self.description) + + parser.add_option('-i', '--input', type='string', help=self.input_help) + + if self.has_output: + parser.add_option('-o', '--output', type='string', help=self.output_help) + + parser.add_option('--keyform', + help='Key format of the %s key - default PEM' % self.keyname, + choices=('PEM', 'DER'), default='PEM') + + (cli, cli_args) = parser.parse_args(sys.argv[1:]) + + if len(cli_args) != self.expected_cli_args: + parser.print_help() + raise SystemExit(1) + + return cli, cli_args + + def read_key(self, filename, keyform): + """Reads a public or private key.""" + + print('Reading %s key from %s' % (self.keyname, filename), file=sys.stderr) + with open(filename, 'rb') as keyfile: + keydata = keyfile.read() + + return self.key_class.load_pkcs1(keydata, keyform) + + def read_infile(self, inname): + """Read the input file""" + + if inname: + print('Reading input from %s' % inname, file=sys.stderr) + with open(inname, 'rb') as infile: + return infile.read() + + print('Reading input from stdin', file=sys.stderr) + return sys.stdin.read() + + def write_outfile(self, outdata, outname): + """Write the output file""" + + if outname: + print('Writing output to %s' % outname, file=sys.stderr) + with open(outname, 'wb') as outfile: + outfile.write(outdata) + else: + print('Writing output to stdout', file=sys.stderr) + rsa._compat.write_to_stdout(outdata) + + +class EncryptOperation(CryptoOperation): + """Encrypts a file.""" + + keyname = 'public' + description = ('Encrypts a file. The file must be shorter than the key ' + 'length in order to be encrypted.') + operation = 'encrypt' + operation_past = 'encrypted' + operation_progressive = 'encrypting' + + def perform_operation(self, indata, pub_key, cli_args=None): + """Encrypts files.""" + + return rsa.encrypt(indata, pub_key) + + +class DecryptOperation(CryptoOperation): + """Decrypts a file.""" + + keyname = 'private' + description = ('Decrypts a file. The original file must be shorter than ' + 'the key length in order to have been encrypted.') + operation = 'decrypt' + operation_past = 'decrypted' + operation_progressive = 'decrypting' + key_class = rsa.PrivateKey + + def perform_operation(self, indata, priv_key, cli_args=None): + """Decrypts files.""" + + return rsa.decrypt(indata, priv_key) + + +class SignOperation(CryptoOperation): + """Signs a file.""" + + keyname = 'private' + usage = 'usage: %%prog [options] private_key hash_method' + description = ('Signs a file, outputs the signature. Choose the hash ' + 'method from %s' % ', '.join(HASH_METHODS)) + operation = 'sign' + operation_past = 'signature' + operation_progressive = 'Signing' + key_class = rsa.PrivateKey + expected_cli_args = 2 + + output_help = ('Name of the file to write the signature to. Written ' + 'to stdout if this option is not present.') + + def perform_operation(self, indata, priv_key, cli_args): + """Signs files.""" + + hash_method = cli_args[1] + if hash_method not in HASH_METHODS: + raise SystemExit('Invalid hash method, choose one of %s' % + ', '.join(HASH_METHODS)) + + return rsa.sign(indata, priv_key, hash_method) + + +class VerifyOperation(CryptoOperation): + """Verify a signature.""" + + keyname = 'public' + usage = 'usage: %%prog [options] public_key signature_file' + description = ('Verifies a signature, exits with status 0 upon success, ' + 'prints an error message and exits with status 1 upon error.') + operation = 'verify' + operation_past = 'verified' + operation_progressive = 'Verifying' + key_class = rsa.PublicKey + expected_cli_args = 2 + has_output = False + + def perform_operation(self, indata, pub_key, cli_args): + """Verifies files.""" + + signature_file = cli_args[1] + + with open(signature_file, 'rb') as sigfile: + signature = sigfile.read() + + try: + rsa.verify(indata, signature, pub_key) + except rsa.VerificationError: + raise SystemExit('Verification failed.') + + print('Verification OK', file=sys.stderr) + + +encrypt = EncryptOperation() +decrypt = DecryptOperation() +sign = SignOperation() +verify = VerifyOperation() diff --git a/contrib/python/rsa/py2/rsa/common.py b/contrib/python/rsa/py2/rsa/common.py new file mode 100644 index 0000000000..f7aa2d1496 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/common.py @@ -0,0 +1,188 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from rsa._compat import zip + +"""Common functionality shared by several modules.""" + + +class NotRelativePrimeError(ValueError): + def __init__(self, a, b, d, msg=None): + super(NotRelativePrimeError, self).__init__( + msg or "%d and %d are not relatively prime, divider=%i" % (a, b, d)) + self.a = a + self.b = b + self.d = d + + +def bit_size(num): + """ + Number of bits needed to represent a integer excluding any prefix + 0 bits. + + Usage:: + + >>> bit_size(1023) + 10 + >>> bit_size(1024) + 11 + >>> bit_size(1025) + 11 + + :param num: + Integer value. If num is 0, returns 0. Only the absolute value of the + number is considered. Therefore, signed integers will be abs(num) + before the number's bit length is determined. + :returns: + Returns the number of bits in the integer. + """ + + try: + return num.bit_length() + except AttributeError: + raise TypeError('bit_size(num) only supports integers, not %r' % type(num)) + + +def byte_size(number): + """ + Returns the number of bytes required to hold a specific long number. + + The number of bytes is rounded up. + + Usage:: + + >>> byte_size(1 << 1023) + 128 + >>> byte_size((1 << 1024) - 1) + 128 + >>> byte_size(1 << 1024) + 129 + + :param number: + An unsigned integer + :returns: + The number of bytes required to hold a specific long number. + """ + if number == 0: + return 1 + return ceil_div(bit_size(number), 8) + + +def ceil_div(num, div): + """ + Returns the ceiling function of a division between `num` and `div`. + + Usage:: + + >>> ceil_div(100, 7) + 15 + >>> ceil_div(100, 10) + 10 + >>> ceil_div(1, 4) + 1 + + :param num: Division's numerator, a number + :param div: Division's divisor, a number + + :return: Rounded up result of the division between the parameters. + """ + quanta, mod = divmod(num, div) + if mod: + quanta += 1 + return quanta + + +def extended_gcd(a, b): + """Returns a tuple (r, i, j) such that r = gcd(a, b) = ia + jb + """ + # r = gcd(a,b) i = multiplicitive inverse of a mod b + # or j = multiplicitive inverse of b mod a + # Neg return values for i or j are made positive mod b or a respectively + # Iterateive Version is faster and uses much less stack space + x = 0 + y = 1 + lx = 1 + ly = 0 + oa = a # Remember original a/b to remove + ob = b # negative values from return results + while b != 0: + q = a // b + (a, b) = (b, a % b) + (x, lx) = ((lx - (q * x)), x) + (y, ly) = ((ly - (q * y)), y) + if lx < 0: + lx += ob # If neg wrap modulo orignal b + if ly < 0: + ly += oa # If neg wrap modulo orignal a + return a, lx, ly # Return only positive values + + +def inverse(x, n): + """Returns the inverse of x % n under multiplication, a.k.a x^-1 (mod n) + + >>> inverse(7, 4) + 3 + >>> (inverse(143, 4) * 143) % 4 + 1 + """ + + (divider, inv, _) = extended_gcd(x, n) + + if divider != 1: + raise NotRelativePrimeError(x, n, divider) + + return inv + + +def crt(a_values, modulo_values): + """Chinese Remainder Theorem. + + Calculates x such that x = a[i] (mod m[i]) for each i. + + :param a_values: the a-values of the above equation + :param modulo_values: the m-values of the above equation + :returns: x such that x = a[i] (mod m[i]) for each i + + + >>> crt([2, 3], [3, 5]) + 8 + + >>> crt([2, 3, 2], [3, 5, 7]) + 23 + + >>> crt([2, 3, 0], [7, 11, 15]) + 135 + """ + + m = 1 + x = 0 + + for modulo in modulo_values: + m *= modulo + + for (m_i, a_i) in zip(modulo_values, a_values): + M_i = m // m_i + inv = inverse(M_i, m_i) + + x = (x + a_i * M_i * inv) % m + + return x + + +if __name__ == '__main__': + import doctest + + doctest.testmod() diff --git a/contrib/python/rsa/py2/rsa/core.py b/contrib/python/rsa/py2/rsa/core.py new file mode 100644 index 0000000000..b3114d9e9c --- /dev/null +++ b/contrib/python/rsa/py2/rsa/core.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Core mathematical operations. + +This is the actual core RSA implementation, which is only defined +mathematically on integers. +""" + +from rsa._compat import is_integer + + +def assert_int(var, name): + if is_integer(var): + return + + raise TypeError('%s should be an integer, not %s' % (name, var.__class__)) + + +def encrypt_int(message, ekey, n): + """Encrypts a message using encryption key 'ekey', working modulo n""" + + assert_int(message, 'message') + assert_int(ekey, 'ekey') + assert_int(n, 'n') + + if message < 0: + raise ValueError('Only non-negative numbers are supported') + + if message > n: + raise OverflowError("The message %i is too long for n=%i" % (message, n)) + + return pow(message, ekey, n) + + +def decrypt_int(cyphertext, dkey, n): + """Decrypts a cypher text using the decryption key 'dkey', working modulo n""" + + assert_int(cyphertext, 'cyphertext') + assert_int(dkey, 'dkey') + assert_int(n, 'n') + + message = pow(cyphertext, dkey, n) + return message diff --git a/contrib/python/rsa/py2/rsa/key.py b/contrib/python/rsa/py2/rsa/key.py new file mode 100644 index 0000000000..1e2f6fe455 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/key.py @@ -0,0 +1,798 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""RSA key generation code. + +Create new keys with the newkeys() function. It will give you a PublicKey and a +PrivateKey object. + +Loading and saving keys requires the pyasn1 module. This module is imported as +late as possible, such that other functionality will remain working in absence +of pyasn1. + +.. note:: + + Storing public and private keys via the `pickle` module is possible. + However, it is insecure to load a key from an untrusted source. + The pickle module is not secure against erroneous or maliciously + constructed data. Never unpickle data received from an untrusted + or unauthenticated source. + +""" + +import logging +import warnings + +from rsa._compat import range +import rsa.prime +import rsa.pem +import rsa.common +import rsa.randnum +import rsa.core + + +log = logging.getLogger(__name__) +DEFAULT_EXPONENT = 65537 + + +class AbstractKey(object): + """Abstract superclass for private and public keys.""" + + __slots__ = ('n', 'e') + + def __init__(self, n, e): + self.n = n + self.e = e + + @classmethod + def _load_pkcs1_pem(cls, keyfile): + """Loads a key in PKCS#1 PEM format, implement in a subclass. + + :param keyfile: contents of a PEM-encoded file that contains + the public key. + :type keyfile: bytes + + :return: the loaded key + :rtype: AbstractKey + """ + + @classmethod + def _load_pkcs1_der(cls, keyfile): + """Loads a key in PKCS#1 PEM format, implement in a subclass. + + :param keyfile: contents of a DER-encoded file that contains + the public key. + :type keyfile: bytes + + :return: the loaded key + :rtype: AbstractKey + """ + + def _save_pkcs1_pem(self): + """Saves the key in PKCS#1 PEM format, implement in a subclass. + + :returns: the PEM-encoded key. + :rtype: bytes + """ + + def _save_pkcs1_der(self): + """Saves the key in PKCS#1 DER format, implement in a subclass. + + :returns: the DER-encoded key. + :rtype: bytes + """ + + @classmethod + def load_pkcs1(cls, keyfile, format='PEM'): + """Loads a key in PKCS#1 DER or PEM format. + + :param keyfile: contents of a DER- or PEM-encoded file that contains + the key. + :type keyfile: bytes + :param format: the format of the file to load; 'PEM' or 'DER' + :type format: str + + :return: the loaded key + :rtype: AbstractKey + """ + + methods = { + 'PEM': cls._load_pkcs1_pem, + 'DER': cls._load_pkcs1_der, + } + + method = cls._assert_format_exists(format, methods) + return method(keyfile) + + @staticmethod + def _assert_format_exists(file_format, methods): + """Checks whether the given file format exists in 'methods'. + """ + + try: + return methods[file_format] + except KeyError: + formats = ', '.join(sorted(methods.keys())) + raise ValueError('Unsupported format: %r, try one of %s' % (file_format, + formats)) + + def save_pkcs1(self, format='PEM'): + """Saves the key in PKCS#1 DER or PEM format. + + :param format: the format to save; 'PEM' or 'DER' + :type format: str + :returns: the DER- or PEM-encoded key. + :rtype: bytes + """ + + methods = { + 'PEM': self._save_pkcs1_pem, + 'DER': self._save_pkcs1_der, + } + + method = self._assert_format_exists(format, methods) + return method() + + def blind(self, message, r): + """Performs blinding on the message using random number 'r'. + + :param message: the message, as integer, to blind. + :type message: int + :param r: the random number to blind with. + :type r: int + :return: the blinded message. + :rtype: int + + The blinding is such that message = unblind(decrypt(blind(encrypt(message))). + + See https://en.wikipedia.org/wiki/Blinding_%28cryptography%29 + """ + + return (message * pow(r, self.e, self.n)) % self.n + + def unblind(self, blinded, r): + """Performs blinding on the message using random number 'r'. + + :param blinded: the blinded message, as integer, to unblind. + :param r: the random number to unblind with. + :return: the original message. + + The blinding is such that message = unblind(decrypt(blind(encrypt(message))). + + See https://en.wikipedia.org/wiki/Blinding_%28cryptography%29 + """ + + return (rsa.common.inverse(r, self.n) * blinded) % self.n + + +class PublicKey(AbstractKey): + """Represents a public RSA key. + + This key is also known as the 'encryption key'. It contains the 'n' and 'e' + values. + + Supports attributes as well as dictionary-like access. Attribute access is + faster, though. + + >>> PublicKey(5, 3) + PublicKey(5, 3) + + >>> key = PublicKey(5, 3) + >>> key.n + 5 + >>> key['n'] + 5 + >>> key.e + 3 + >>> key['e'] + 3 + + """ + + __slots__ = ('n', 'e') + + def __getitem__(self, key): + return getattr(self, key) + + def __repr__(self): + return 'PublicKey(%i, %i)' % (self.n, self.e) + + def __getstate__(self): + """Returns the key as tuple for pickling.""" + return self.n, self.e + + def __setstate__(self, state): + """Sets the key from tuple.""" + self.n, self.e = state + + def __eq__(self, other): + if other is None: + return False + + if not isinstance(other, PublicKey): + return False + + return self.n == other.n and self.e == other.e + + def __ne__(self, other): + return not (self == other) + + def __hash__(self): + return hash((self.n, self.e)) + + @classmethod + def _load_pkcs1_der(cls, keyfile): + """Loads a key in PKCS#1 DER format. + + :param keyfile: contents of a DER-encoded file that contains the public + key. + :return: a PublicKey object + + First let's construct a DER encoded key: + + >>> import base64 + >>> b64der = 'MAwCBQCNGmYtAgMBAAE=' + >>> der = base64.standard_b64decode(b64der) + + This loads the file: + + >>> PublicKey._load_pkcs1_der(der) + PublicKey(2367317549, 65537) + + """ + + from pyasn1.codec.der import decoder + from rsa.asn1 import AsnPubKey + + (priv, _) = decoder.decode(keyfile, asn1Spec=AsnPubKey()) + return cls(n=int(priv['modulus']), e=int(priv['publicExponent'])) + + def _save_pkcs1_der(self): + """Saves the public key in PKCS#1 DER format. + + :returns: the DER-encoded public key. + :rtype: bytes + """ + + from pyasn1.codec.der import encoder + from rsa.asn1 import AsnPubKey + + # Create the ASN object + asn_key = AsnPubKey() + asn_key.setComponentByName('modulus', self.n) + asn_key.setComponentByName('publicExponent', self.e) + + return encoder.encode(asn_key) + + @classmethod + def _load_pkcs1_pem(cls, keyfile): + """Loads a PKCS#1 PEM-encoded public key file. + + The contents of the file before the "-----BEGIN RSA PUBLIC KEY-----" and + after the "-----END RSA PUBLIC KEY-----" lines is ignored. + + :param keyfile: contents of a PEM-encoded file that contains the public + key. + :return: a PublicKey object + """ + + der = rsa.pem.load_pem(keyfile, 'RSA PUBLIC KEY') + return cls._load_pkcs1_der(der) + + def _save_pkcs1_pem(self): + """Saves a PKCS#1 PEM-encoded public key file. + + :return: contents of a PEM-encoded file that contains the public key. + :rtype: bytes + """ + + der = self._save_pkcs1_der() + return rsa.pem.save_pem(der, 'RSA PUBLIC KEY') + + @classmethod + def load_pkcs1_openssl_pem(cls, keyfile): + """Loads a PKCS#1.5 PEM-encoded public key file from OpenSSL. + + These files can be recognised in that they start with BEGIN PUBLIC KEY + rather than BEGIN RSA PUBLIC KEY. + + The contents of the file before the "-----BEGIN PUBLIC KEY-----" and + after the "-----END PUBLIC KEY-----" lines is ignored. + + :param keyfile: contents of a PEM-encoded file that contains the public + key, from OpenSSL. + :type keyfile: bytes + :return: a PublicKey object + """ + + der = rsa.pem.load_pem(keyfile, 'PUBLIC KEY') + return cls.load_pkcs1_openssl_der(der) + + @classmethod + def load_pkcs1_openssl_der(cls, keyfile): + """Loads a PKCS#1 DER-encoded public key file from OpenSSL. + + :param keyfile: contents of a DER-encoded file that contains the public + key, from OpenSSL. + :return: a PublicKey object + :rtype: bytes + + """ + + from rsa.asn1 import OpenSSLPubKey + from pyasn1.codec.der import decoder + from pyasn1.type import univ + + (keyinfo, _) = decoder.decode(keyfile, asn1Spec=OpenSSLPubKey()) + + if keyinfo['header']['oid'] != univ.ObjectIdentifier('1.2.840.113549.1.1.1'): + raise TypeError("This is not a DER-encoded OpenSSL-compatible public key") + + return cls._load_pkcs1_der(keyinfo['key'][1:]) + + +class PrivateKey(AbstractKey): + """Represents a private RSA key. + + This key is also known as the 'decryption key'. It contains the 'n', 'e', + 'd', 'p', 'q' and other values. + + Supports attributes as well as dictionary-like access. Attribute access is + faster, though. + + >>> PrivateKey(3247, 65537, 833, 191, 17) + PrivateKey(3247, 65537, 833, 191, 17) + + exp1, exp2 and coef will be calculated: + + >>> pk = PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + >>> pk.exp1 + 55063 + >>> pk.exp2 + 10095 + >>> pk.coef + 50797 + + """ + + __slots__ = ('n', 'e', 'd', 'p', 'q', 'exp1', 'exp2', 'coef') + + def __init__(self, n, e, d, p, q): + AbstractKey.__init__(self, n, e) + self.d = d + self.p = p + self.q = q + + # Calculate exponents and coefficient. + self.exp1 = int(d % (p - 1)) + self.exp2 = int(d % (q - 1)) + self.coef = rsa.common.inverse(q, p) + + def __getitem__(self, key): + return getattr(self, key) + + def __repr__(self): + return 'PrivateKey(%(n)i, %(e)i, %(d)i, %(p)i, %(q)i)' % self + + def __getstate__(self): + """Returns the key as tuple for pickling.""" + return self.n, self.e, self.d, self.p, self.q, self.exp1, self.exp2, self.coef + + def __setstate__(self, state): + """Sets the key from tuple.""" + self.n, self.e, self.d, self.p, self.q, self.exp1, self.exp2, self.coef = state + + def __eq__(self, other): + if other is None: + return False + + if not isinstance(other, PrivateKey): + return False + + return (self.n == other.n and + self.e == other.e and + self.d == other.d and + self.p == other.p and + self.q == other.q and + self.exp1 == other.exp1 and + self.exp2 == other.exp2 and + self.coef == other.coef) + + def __ne__(self, other): + return not (self == other) + + def __hash__(self): + return hash((self.n, self.e, self.d, self.p, self.q, self.exp1, self.exp2, self.coef)) + + def _get_blinding_factor(self): + for _ in range(1000): + blind_r = rsa.randnum.randint(self.n - 1) + if rsa.prime.are_relatively_prime(self.n, blind_r): + return blind_r + raise RuntimeError('unable to find blinding factor') + + def blinded_decrypt(self, encrypted): + """Decrypts the message using blinding to prevent side-channel attacks. + + :param encrypted: the encrypted message + :type encrypted: int + + :returns: the decrypted message + :rtype: int + """ + + blind_r = self._get_blinding_factor() + blinded = self.blind(encrypted, blind_r) # blind before decrypting + decrypted = rsa.core.decrypt_int(blinded, self.d, self.n) + + return self.unblind(decrypted, blind_r) + + def blinded_encrypt(self, message): + """Encrypts the message using blinding to prevent side-channel attacks. + + :param message: the message to encrypt + :type message: int + + :returns: the encrypted message + :rtype: int + """ + + blind_r = self._get_blinding_factor() + blinded = self.blind(message, blind_r) # blind before encrypting + encrypted = rsa.core.encrypt_int(blinded, self.d, self.n) + return self.unblind(encrypted, blind_r) + + @classmethod + def _load_pkcs1_der(cls, keyfile): + """Loads a key in PKCS#1 DER format. + + :param keyfile: contents of a DER-encoded file that contains the private + key. + :type keyfile: bytes + :return: a PrivateKey object + + First let's construct a DER encoded key: + + >>> import base64 + >>> b64der = 'MC4CAQACBQDeKYlRAgMBAAECBQDHn4npAgMA/icCAwDfxwIDANcXAgInbwIDAMZt' + >>> der = base64.standard_b64decode(b64der) + + This loads the file: + + >>> PrivateKey._load_pkcs1_der(der) + PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + + """ + + from pyasn1.codec.der import decoder + (priv, _) = decoder.decode(keyfile) + + # ASN.1 contents of DER encoded private key: + # + # RSAPrivateKey ::= SEQUENCE { + # version Version, + # modulus INTEGER, -- n + # publicExponent INTEGER, -- e + # privateExponent INTEGER, -- d + # prime1 INTEGER, -- p + # prime2 INTEGER, -- q + # exponent1 INTEGER, -- d mod (p-1) + # exponent2 INTEGER, -- d mod (q-1) + # coefficient INTEGER, -- (inverse of q) mod p + # otherPrimeInfos OtherPrimeInfos OPTIONAL + # } + + if priv[0] != 0: + raise ValueError('Unable to read this file, version %s != 0' % priv[0]) + + as_ints = map(int, priv[1:6]) + key = cls(*as_ints) + + exp1, exp2, coef = map(int, priv[6:9]) + + if (key.exp1, key.exp2, key.coef) != (exp1, exp2, coef): + warnings.warn( + 'You have provided a malformed keyfile. Either the exponents ' + 'or the coefficient are incorrect. Using the correct values ' + 'instead.', + UserWarning, + ) + + return key + + def _save_pkcs1_der(self): + """Saves the private key in PKCS#1 DER format. + + :returns: the DER-encoded private key. + :rtype: bytes + """ + + from pyasn1.type import univ, namedtype + from pyasn1.codec.der import encoder + + class AsnPrivKey(univ.Sequence): + componentType = namedtype.NamedTypes( + namedtype.NamedType('version', univ.Integer()), + namedtype.NamedType('modulus', univ.Integer()), + namedtype.NamedType('publicExponent', univ.Integer()), + namedtype.NamedType('privateExponent', univ.Integer()), + namedtype.NamedType('prime1', univ.Integer()), + namedtype.NamedType('prime2', univ.Integer()), + namedtype.NamedType('exponent1', univ.Integer()), + namedtype.NamedType('exponent2', univ.Integer()), + namedtype.NamedType('coefficient', univ.Integer()), + ) + + # Create the ASN object + asn_key = AsnPrivKey() + asn_key.setComponentByName('version', 0) + asn_key.setComponentByName('modulus', self.n) + asn_key.setComponentByName('publicExponent', self.e) + asn_key.setComponentByName('privateExponent', self.d) + asn_key.setComponentByName('prime1', self.p) + asn_key.setComponentByName('prime2', self.q) + asn_key.setComponentByName('exponent1', self.exp1) + asn_key.setComponentByName('exponent2', self.exp2) + asn_key.setComponentByName('coefficient', self.coef) + + return encoder.encode(asn_key) + + @classmethod + def _load_pkcs1_pem(cls, keyfile): + """Loads a PKCS#1 PEM-encoded private key file. + + The contents of the file before the "-----BEGIN RSA PRIVATE KEY-----" and + after the "-----END RSA PRIVATE KEY-----" lines is ignored. + + :param keyfile: contents of a PEM-encoded file that contains the private + key. + :type keyfile: bytes + :return: a PrivateKey object + """ + + der = rsa.pem.load_pem(keyfile, b'RSA PRIVATE KEY') + return cls._load_pkcs1_der(der) + + def _save_pkcs1_pem(self): + """Saves a PKCS#1 PEM-encoded private key file. + + :return: contents of a PEM-encoded file that contains the private key. + :rtype: bytes + """ + + der = self._save_pkcs1_der() + return rsa.pem.save_pem(der, b'RSA PRIVATE KEY') + + +def find_p_q(nbits, getprime_func=rsa.prime.getprime, accurate=True): + """Returns a tuple of two different primes of nbits bits each. + + The resulting p * q has exacty 2 * nbits bits, and the returned p and q + will not be equal. + + :param nbits: the number of bits in each of p and q. + :param getprime_func: the getprime function, defaults to + :py:func:`rsa.prime.getprime`. + + *Introduced in Python-RSA 3.1* + + :param accurate: whether to enable accurate mode or not. + :returns: (p, q), where p > q + + >>> (p, q) = find_p_q(128) + >>> from rsa import common + >>> common.bit_size(p * q) + 256 + + When not in accurate mode, the number of bits can be slightly less + + >>> (p, q) = find_p_q(128, accurate=False) + >>> from rsa import common + >>> common.bit_size(p * q) <= 256 + True + >>> common.bit_size(p * q) > 240 + True + + """ + + total_bits = nbits * 2 + + # Make sure that p and q aren't too close or the factoring programs can + # factor n. + shift = nbits // 16 + pbits = nbits + shift + qbits = nbits - shift + + # Choose the two initial primes + log.debug('find_p_q(%i): Finding p', nbits) + p = getprime_func(pbits) + log.debug('find_p_q(%i): Finding q', nbits) + q = getprime_func(qbits) + + def is_acceptable(p, q): + """Returns True iff p and q are acceptable: + + - p and q differ + - (p * q) has the right nr of bits (when accurate=True) + """ + + if p == q: + return False + + if not accurate: + return True + + # Make sure we have just the right amount of bits + found_size = rsa.common.bit_size(p * q) + return total_bits == found_size + + # Keep choosing other primes until they match our requirements. + change_p = False + while not is_acceptable(p, q): + # Change p on one iteration and q on the other + if change_p: + p = getprime_func(pbits) + else: + q = getprime_func(qbits) + + change_p = not change_p + + # We want p > q as described on + # http://www.di-mgt.com.au/rsa_alg.html#crt + return max(p, q), min(p, q) + + +def calculate_keys_custom_exponent(p, q, exponent): + """Calculates an encryption and a decryption key given p, q and an exponent, + and returns them as a tuple (e, d) + + :param p: the first large prime + :param q: the second large prime + :param exponent: the exponent for the key; only change this if you know + what you're doing, as the exponent influences how difficult your + private key can be cracked. A very common choice for e is 65537. + :type exponent: int + + """ + + phi_n = (p - 1) * (q - 1) + + try: + d = rsa.common.inverse(exponent, phi_n) + except rsa.common.NotRelativePrimeError as ex: + raise rsa.common.NotRelativePrimeError( + exponent, phi_n, ex.d, + msg="e (%d) and phi_n (%d) are not relatively prime (divider=%i)" % + (exponent, phi_n, ex.d)) + + if (exponent * d) % phi_n != 1: + raise ValueError("e (%d) and d (%d) are not mult. inv. modulo " + "phi_n (%d)" % (exponent, d, phi_n)) + + return exponent, d + + +def calculate_keys(p, q): + """Calculates an encryption and a decryption key given p and q, and + returns them as a tuple (e, d) + + :param p: the first large prime + :param q: the second large prime + + :return: tuple (e, d) with the encryption and decryption exponents. + """ + + return calculate_keys_custom_exponent(p, q, DEFAULT_EXPONENT) + + +def gen_keys(nbits, getprime_func, accurate=True, exponent=DEFAULT_EXPONENT): + """Generate RSA keys of nbits bits. Returns (p, q, e, d). + + Note: this can take a long time, depending on the key size. + + :param nbits: the total number of bits in ``p`` and ``q``. Both ``p`` and + ``q`` will use ``nbits/2`` bits. + :param getprime_func: either :py:func:`rsa.prime.getprime` or a function + with similar signature. + :param exponent: the exponent for the key; only change this if you know + what you're doing, as the exponent influences how difficult your + private key can be cracked. A very common choice for e is 65537. + :type exponent: int + """ + + # Regenerate p and q values, until calculate_keys doesn't raise a + # ValueError. + while True: + (p, q) = find_p_q(nbits // 2, getprime_func, accurate) + try: + (e, d) = calculate_keys_custom_exponent(p, q, exponent=exponent) + break + except ValueError: + pass + + return p, q, e, d + + +def newkeys(nbits, accurate=True, poolsize=1, exponent=DEFAULT_EXPONENT): + """Generates public and private keys, and returns them as (pub, priv). + + The public key is also known as the 'encryption key', and is a + :py:class:`rsa.PublicKey` object. The private key is also known as the + 'decryption key' and is a :py:class:`rsa.PrivateKey` object. + + :param nbits: the number of bits required to store ``n = p*q``. + :param accurate: when True, ``n`` will have exactly the number of bits you + asked for. However, this makes key generation much slower. When False, + `n`` may have slightly less bits. + :param poolsize: the number of processes to use to generate the prime + numbers. If set to a number > 1, a parallel algorithm will be used. + This requires Python 2.6 or newer. + :param exponent: the exponent for the key; only change this if you know + what you're doing, as the exponent influences how difficult your + private key can be cracked. A very common choice for e is 65537. + :type exponent: int + + :returns: a tuple (:py:class:`rsa.PublicKey`, :py:class:`rsa.PrivateKey`) + + The ``poolsize`` parameter was added in *Python-RSA 3.1* and requires + Python 2.6 or newer. + + """ + + if nbits < 16: + raise ValueError('Key too small') + + if poolsize < 1: + raise ValueError('Pool size (%i) should be >= 1' % poolsize) + + # Determine which getprime function to use + if poolsize > 1: + from rsa import parallel + import functools + + getprime_func = functools.partial(parallel.getprime, poolsize=poolsize) + else: + getprime_func = rsa.prime.getprime + + # Generate the key components + (p, q, e, d) = gen_keys(nbits, getprime_func, accurate=accurate, exponent=exponent) + + # Create the key objects + n = p * q + + return ( + PublicKey(n, e), + PrivateKey(n, e, d, p, q) + ) + + +__all__ = ['PublicKey', 'PrivateKey', 'newkeys'] + +if __name__ == '__main__': + import doctest + + try: + for count in range(100): + (failures, tests) = doctest.testmod() + if failures: + break + + if (count % 10 == 0 and count) or count == 1: + print('%i times' % count) + except KeyboardInterrupt: + print('Aborted') + else: + print('Doctests done') diff --git a/contrib/python/rsa/py2/rsa/machine_size.py b/contrib/python/rsa/py2/rsa/machine_size.py new file mode 100644 index 0000000000..2a871b8f6e --- /dev/null +++ b/contrib/python/rsa/py2/rsa/machine_size.py @@ -0,0 +1,74 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Detection of 32-bit and 64-bit machines and byte alignment.""" + +import sys + +MAX_INT = sys.maxsize +MAX_INT64 = (1 << 63) - 1 +MAX_INT32 = (1 << 31) - 1 +MAX_INT16 = (1 << 15) - 1 + +# Determine the word size of the processor. +if MAX_INT == MAX_INT64: + # 64-bit processor. + MACHINE_WORD_SIZE = 64 +elif MAX_INT == MAX_INT32: + # 32-bit processor. + MACHINE_WORD_SIZE = 32 +else: + # Else we just assume 64-bit processor keeping up with modern times. + MACHINE_WORD_SIZE = 64 + + +def get_word_alignment(num, force_arch=64, + _machine_word_size=MACHINE_WORD_SIZE): + """ + Returns alignment details for the given number based on the platform + Python is running on. + + :param num: + Unsigned integral number. + :param force_arch: + If you don't want to use 64-bit unsigned chunks, set this to + anything other than 64. 32-bit chunks will be preferred then. + Default 64 will be used when on a 64-bit machine. + :param _machine_word_size: + (Internal) The machine word size used for alignment. + :returns: + 4-tuple:: + + (word_bits, word_bytes, + max_uint, packing_format_type) + """ + max_uint64 = 0xffffffffffffffff + max_uint32 = 0xffffffff + max_uint16 = 0xffff + max_uint8 = 0xff + + if force_arch == 64 and _machine_word_size >= 64 and num > max_uint32: + # 64-bit unsigned integer. + return 64, 8, max_uint64, "Q" + elif num > max_uint16: + # 32-bit unsigned integer + return 32, 4, max_uint32, "L" + elif num > max_uint8: + # 16-bit unsigned integer. + return 16, 2, max_uint16, "H" + else: + # 8-bit unsigned integer. + return 8, 1, max_uint8, "B" diff --git a/contrib/python/rsa/py2/rsa/parallel.py b/contrib/python/rsa/py2/rsa/parallel.py new file mode 100644 index 0000000000..a3fe312204 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/parallel.py @@ -0,0 +1,101 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Functions for parallel computation on multiple cores. + +Introduced in Python-RSA 3.1. + +.. note:: + + Requires Python 2.6 or newer. + +""" + +from __future__ import print_function + +import multiprocessing as mp + +from rsa._compat import range +import rsa.prime +import rsa.randnum + + +def _find_prime(nbits, pipe): + while True: + integer = rsa.randnum.read_random_odd_int(nbits) + + # Test for primeness + if rsa.prime.is_prime(integer): + pipe.send(integer) + return + + +def getprime(nbits, poolsize): + """Returns a prime number that can be stored in 'nbits' bits. + + Works in multiple threads at the same time. + + >>> p = getprime(128, 3) + >>> rsa.prime.is_prime(p-1) + False + >>> rsa.prime.is_prime(p) + True + >>> rsa.prime.is_prime(p+1) + False + + >>> from rsa import common + >>> common.bit_size(p) == 128 + True + + """ + + (pipe_recv, pipe_send) = mp.Pipe(duplex=False) + + # Create processes + try: + procs = [mp.Process(target=_find_prime, args=(nbits, pipe_send)) + for _ in range(poolsize)] + # Start processes + for p in procs: + p.start() + + result = pipe_recv.recv() + finally: + pipe_recv.close() + pipe_send.close() + + # Terminate processes + for p in procs: + p.terminate() + + return result + + +__all__ = ['getprime'] + +if __name__ == '__main__': + print('Running doctests 1000x or until failure') + import doctest + + for count in range(100): + (failures, tests) = doctest.testmod() + if failures: + break + + if count % 10 == 0 and count: + print('%i times' % count) + + print('Doctests done') diff --git a/contrib/python/rsa/py2/rsa/pem.py b/contrib/python/rsa/py2/rsa/pem.py new file mode 100644 index 0000000000..2ddfae86e2 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/pem.py @@ -0,0 +1,126 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Functions that load and write PEM-encoded files.""" + +import base64 + +from rsa._compat import is_bytes, range + + +def _markers(pem_marker): + """ + Returns the start and end PEM markers, as bytes. + """ + + if not is_bytes(pem_marker): + pem_marker = pem_marker.encode('ascii') + + return (b'-----BEGIN ' + pem_marker + b'-----', + b'-----END ' + pem_marker + b'-----') + + +def load_pem(contents, pem_marker): + """Loads a PEM file. + + :param contents: the contents of the file to interpret + :param pem_marker: the marker of the PEM content, such as 'RSA PRIVATE KEY' + when your file has '-----BEGIN RSA PRIVATE KEY-----' and + '-----END RSA PRIVATE KEY-----' markers. + + :return: the base64-decoded content between the start and end markers. + + @raise ValueError: when the content is invalid, for example when the start + marker cannot be found. + + """ + + # We want bytes, not text. If it's text, it can be converted to ASCII bytes. + if not is_bytes(contents): + contents = contents.encode('ascii') + + (pem_start, pem_end) = _markers(pem_marker) + + pem_lines = [] + in_pem_part = False + + for line in contents.splitlines(): + line = line.strip() + + # Skip empty lines + if not line: + continue + + # Handle start marker + if line == pem_start: + if in_pem_part: + raise ValueError('Seen start marker "%s" twice' % pem_start) + + in_pem_part = True + continue + + # Skip stuff before first marker + if not in_pem_part: + continue + + # Handle end marker + if in_pem_part and line == pem_end: + in_pem_part = False + break + + # Load fields + if b':' in line: + continue + + pem_lines.append(line) + + # Do some sanity checks + if not pem_lines: + raise ValueError('No PEM start marker "%s" found' % pem_start) + + if in_pem_part: + raise ValueError('No PEM end marker "%s" found' % pem_end) + + # Base64-decode the contents + pem = b''.join(pem_lines) + return base64.standard_b64decode(pem) + + +def save_pem(contents, pem_marker): + """Saves a PEM file. + + :param contents: the contents to encode in PEM format + :param pem_marker: the marker of the PEM content, such as 'RSA PRIVATE KEY' + when your file has '-----BEGIN RSA PRIVATE KEY-----' and + '-----END RSA PRIVATE KEY-----' markers. + + :return: the base64-encoded content between the start and end markers, as bytes. + + """ + + (pem_start, pem_end) = _markers(pem_marker) + + b64 = base64.standard_b64encode(contents).replace(b'\n', b'') + pem_lines = [pem_start] + + for block_start in range(0, len(b64), 64): + block = b64[block_start:block_start + 64] + pem_lines.append(block) + + pem_lines.append(pem_end) + pem_lines.append(b'') + + return b'\n'.join(pem_lines) diff --git a/contrib/python/rsa/py2/rsa/pkcs1.py b/contrib/python/rsa/py2/rsa/pkcs1.py new file mode 100644 index 0000000000..c05239afce --- /dev/null +++ b/contrib/python/rsa/py2/rsa/pkcs1.py @@ -0,0 +1,448 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Functions for PKCS#1 version 1.5 encryption and signing + +This module implements certain functionality from PKCS#1 version 1.5. For a +very clear example, read http://www.di-mgt.com.au/rsa_alg.html#pkcs1schemes + +At least 8 bytes of random padding is used when encrypting a message. This makes +these methods much more secure than the ones in the ``rsa`` module. + +WARNING: this module leaks information when decryption fails. The exceptions +that are raised contain the Python traceback information, which can be used to +deduce where in the process the failure occurred. DO NOT PASS SUCH INFORMATION +to your users. +""" + +import hashlib +import os + +from rsa._compat import range +from rsa import common, transform, core + +# ASN.1 codes that describe the hash algorithm used. +HASH_ASN1 = { + 'MD5': b'\x30\x20\x30\x0c\x06\x08\x2a\x86\x48\x86\xf7\x0d\x02\x05\x05\x00\x04\x10', + 'SHA-1': b'\x30\x21\x30\x09\x06\x05\x2b\x0e\x03\x02\x1a\x05\x00\x04\x14', + 'SHA-224': b'\x30\x2d\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x04\x05\x00\x04\x1c', + 'SHA-256': b'\x30\x31\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x01\x05\x00\x04\x20', + 'SHA-384': b'\x30\x41\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x02\x05\x00\x04\x30', + 'SHA-512': b'\x30\x51\x30\x0d\x06\x09\x60\x86\x48\x01\x65\x03\x04\x02\x03\x05\x00\x04\x40', +} + +HASH_METHODS = { + 'MD5': hashlib.md5, + 'SHA-1': hashlib.sha1, + 'SHA-224': hashlib.sha224, + 'SHA-256': hashlib.sha256, + 'SHA-384': hashlib.sha384, + 'SHA-512': hashlib.sha512, +} + + +class CryptoError(Exception): + """Base class for all exceptions in this module.""" + + +class DecryptionError(CryptoError): + """Raised when decryption fails.""" + + +class VerificationError(CryptoError): + """Raised when verification fails.""" + + +def _pad_for_encryption(message, target_length): + r"""Pads the message for encryption, returning the padded message. + + :return: 00 02 RANDOM_DATA 00 MESSAGE + + >>> block = _pad_for_encryption(b'hello', 16) + >>> len(block) + 16 + >>> block[0:2] + b'\x00\x02' + >>> block[-6:] + b'\x00hello' + + """ + + max_msglength = target_length - 11 + msglength = len(message) + + if msglength > max_msglength: + raise OverflowError('%i bytes needed for message, but there is only' + ' space for %i' % (msglength, max_msglength)) + + # Get random padding + padding = b'' + padding_length = target_length - msglength - 3 + + # We remove 0-bytes, so we'll end up with less padding than we've asked for, + # so keep adding data until we're at the correct length. + while len(padding) < padding_length: + needed_bytes = padding_length - len(padding) + + # Always read at least 8 bytes more than we need, and trim off the rest + # after removing the 0-bytes. This increases the chance of getting + # enough bytes, especially when needed_bytes is small + new_padding = os.urandom(needed_bytes + 5) + new_padding = new_padding.replace(b'\x00', b'') + padding = padding + new_padding[:needed_bytes] + + assert len(padding) == padding_length + + return b''.join([b'\x00\x02', + padding, + b'\x00', + message]) + + +def _pad_for_signing(message, target_length): + r"""Pads the message for signing, returning the padded message. + + The padding is always a repetition of FF bytes. + + :return: 00 01 PADDING 00 MESSAGE + + >>> block = _pad_for_signing(b'hello', 16) + >>> len(block) + 16 + >>> block[0:2] + b'\x00\x01' + >>> block[-6:] + b'\x00hello' + >>> block[2:-6] + b'\xff\xff\xff\xff\xff\xff\xff\xff' + + """ + + max_msglength = target_length - 11 + msglength = len(message) + + if msglength > max_msglength: + raise OverflowError('%i bytes needed for message, but there is only' + ' space for %i' % (msglength, max_msglength)) + + padding_length = target_length - msglength - 3 + + return b''.join([b'\x00\x01', + padding_length * b'\xff', + b'\x00', + message]) + + +def encrypt(message, pub_key): + """Encrypts the given message using PKCS#1 v1.5 + + :param message: the message to encrypt. Must be a byte string no longer than + ``k-11`` bytes, where ``k`` is the number of bytes needed to encode + the ``n`` component of the public key. + :param pub_key: the :py:class:`rsa.PublicKey` to encrypt with. + :raise OverflowError: when the message is too large to fit in the padded + block. + + >>> from rsa import key, common + >>> (pub_key, priv_key) = key.newkeys(256) + >>> message = b'hello' + >>> crypto = encrypt(message, pub_key) + + The crypto text should be just as long as the public key 'n' component: + + >>> len(crypto) == common.byte_size(pub_key.n) + True + + """ + + keylength = common.byte_size(pub_key.n) + padded = _pad_for_encryption(message, keylength) + + payload = transform.bytes2int(padded) + encrypted = core.encrypt_int(payload, pub_key.e, pub_key.n) + block = transform.int2bytes(encrypted, keylength) + + return block + + +def decrypt(crypto, priv_key): + r"""Decrypts the given message using PKCS#1 v1.5 + + The decryption is considered 'failed' when the resulting cleartext doesn't + start with the bytes 00 02, or when the 00 byte between the padding and + the message cannot be found. + + :param crypto: the crypto text as returned by :py:func:`rsa.encrypt` + :param priv_key: the :py:class:`rsa.PrivateKey` to decrypt with. + :raise DecryptionError: when the decryption fails. No details are given as + to why the code thinks the decryption fails, as this would leak + information about the private key. + + + >>> import rsa + >>> (pub_key, priv_key) = rsa.newkeys(256) + + It works with strings: + + >>> crypto = encrypt(b'hello', pub_key) + >>> decrypt(crypto, priv_key) + b'hello' + + And with binary data: + + >>> crypto = encrypt(b'\x00\x00\x00\x00\x01', pub_key) + >>> decrypt(crypto, priv_key) + b'\x00\x00\x00\x00\x01' + + Altering the encrypted information will *likely* cause a + :py:class:`rsa.pkcs1.DecryptionError`. If you want to be *sure*, use + :py:func:`rsa.sign`. + + + .. warning:: + + Never display the stack trace of a + :py:class:`rsa.pkcs1.DecryptionError` exception. It shows where in the + code the exception occurred, and thus leaks information about the key. + It's only a tiny bit of information, but every bit makes cracking the + keys easier. + + >>> crypto = encrypt(b'hello', pub_key) + >>> crypto = crypto[0:5] + b'X' + crypto[6:] # change a byte + >>> decrypt(crypto, priv_key) + Traceback (most recent call last): + ... + rsa.pkcs1.DecryptionError: Decryption failed + + """ + + blocksize = common.byte_size(priv_key.n) + encrypted = transform.bytes2int(crypto) + decrypted = priv_key.blinded_decrypt(encrypted) + cleartext = transform.int2bytes(decrypted, blocksize) + + # Detect leading zeroes in the crypto. These are not reflected in the + # encrypted value (as leading zeroes do not influence the value of an + # integer). This fixes CVE-2020-13757. + if len(crypto) > blocksize: + raise DecryptionError('Decryption failed') + + # If we can't find the cleartext marker, decryption failed. + if cleartext[0:2] != b'\x00\x02': + raise DecryptionError('Decryption failed') + + # Find the 00 separator between the padding and the message + try: + sep_idx = cleartext.index(b'\x00', 2) + except ValueError: + raise DecryptionError('Decryption failed') + + return cleartext[sep_idx + 1:] + + +def sign_hash(hash_value, priv_key, hash_method): + """Signs a precomputed hash with the private key. + + Hashes the message, then signs the hash with the given key. This is known + as a "detached signature", because the message itself isn't altered. + + :param hash_value: A precomputed hash to sign (ignores message). Should be set to + None if needing to hash and sign message. + :param priv_key: the :py:class:`rsa.PrivateKey` to sign with + :param hash_method: the hash method used on the message. Use 'MD5', 'SHA-1', + 'SHA-224', SHA-256', 'SHA-384' or 'SHA-512'. + :return: a message signature block. + :raise OverflowError: if the private key is too small to contain the + requested hash. + + """ + + # Get the ASN1 code for this hash method + if hash_method not in HASH_ASN1: + raise ValueError('Invalid hash method: %s' % hash_method) + asn1code = HASH_ASN1[hash_method] + + # Encrypt the hash with the private key + cleartext = asn1code + hash_value + keylength = common.byte_size(priv_key.n) + padded = _pad_for_signing(cleartext, keylength) + + payload = transform.bytes2int(padded) + encrypted = priv_key.blinded_encrypt(payload) + block = transform.int2bytes(encrypted, keylength) + + return block + + +def sign(message, priv_key, hash_method): + """Signs the message with the private key. + + Hashes the message, then signs the hash with the given key. This is known + as a "detached signature", because the message itself isn't altered. + + :param message: the message to sign. Can be an 8-bit string or a file-like + object. If ``message`` has a ``read()`` method, it is assumed to be a + file-like object. + :param priv_key: the :py:class:`rsa.PrivateKey` to sign with + :param hash_method: the hash method used on the message. Use 'MD5', 'SHA-1', + 'SHA-224', SHA-256', 'SHA-384' or 'SHA-512'. + :return: a message signature block. + :raise OverflowError: if the private key is too small to contain the + requested hash. + + """ + + msg_hash = compute_hash(message, hash_method) + return sign_hash(msg_hash, priv_key, hash_method) + + +def verify(message, signature, pub_key): + """Verifies that the signature matches the message. + + The hash method is detected automatically from the signature. + + :param message: the signed message. Can be an 8-bit string or a file-like + object. If ``message`` has a ``read()`` method, it is assumed to be a + file-like object. + :param signature: the signature block, as created with :py:func:`rsa.sign`. + :param pub_key: the :py:class:`rsa.PublicKey` of the person signing the message. + :raise VerificationError: when the signature doesn't match the message. + :returns: the name of the used hash. + + """ + + keylength = common.byte_size(pub_key.n) + encrypted = transform.bytes2int(signature) + decrypted = core.decrypt_int(encrypted, pub_key.e, pub_key.n) + clearsig = transform.int2bytes(decrypted, keylength) + + # Get the hash method + method_name = _find_method_hash(clearsig) + message_hash = compute_hash(message, method_name) + + # Reconstruct the expected padded hash + cleartext = HASH_ASN1[method_name] + message_hash + expected = _pad_for_signing(cleartext, keylength) + + if len(signature) != keylength: + raise VerificationError('Verification failed') + + # Compare with the signed one + if expected != clearsig: + raise VerificationError('Verification failed') + + return method_name + + +def find_signature_hash(signature, pub_key): + """Returns the hash name detected from the signature. + + If you also want to verify the message, use :py:func:`rsa.verify()` instead. + It also returns the name of the used hash. + + :param signature: the signature block, as created with :py:func:`rsa.sign`. + :param pub_key: the :py:class:`rsa.PublicKey` of the person signing the message. + :returns: the name of the used hash. + """ + + keylength = common.byte_size(pub_key.n) + encrypted = transform.bytes2int(signature) + decrypted = core.decrypt_int(encrypted, pub_key.e, pub_key.n) + clearsig = transform.int2bytes(decrypted, keylength) + + return _find_method_hash(clearsig) + + +def yield_fixedblocks(infile, blocksize): + """Generator, yields each block of ``blocksize`` bytes in the input file. + + :param infile: file to read and separate in blocks. + :param blocksize: block size in bytes. + :returns: a generator that yields the contents of each block + """ + + while True: + block = infile.read(blocksize) + + read_bytes = len(block) + if read_bytes == 0: + break + + yield block + + if read_bytes < blocksize: + break + + +def compute_hash(message, method_name): + """Returns the message digest. + + :param message: the signed message. Can be an 8-bit string or a file-like + object. If ``message`` has a ``read()`` method, it is assumed to be a + file-like object. + :param method_name: the hash method, must be a key of + :py:const:`HASH_METHODS`. + + """ + + if method_name not in HASH_METHODS: + raise ValueError('Invalid hash method: %s' % method_name) + + method = HASH_METHODS[method_name] + hasher = method() + + if hasattr(message, 'read') and hasattr(message.read, '__call__'): + # read as 1K blocks + for block in yield_fixedblocks(message, 1024): + hasher.update(block) + else: + # hash the message object itself. + hasher.update(message) + + return hasher.digest() + + +def _find_method_hash(clearsig): + """Finds the hash method. + + :param clearsig: full padded ASN1 and hash. + :return: the used hash method. + :raise VerificationFailed: when the hash method cannot be found + """ + + for (hashname, asn1code) in HASH_ASN1.items(): + if asn1code in clearsig: + return hashname + + raise VerificationError('Verification failed') + + +__all__ = ['encrypt', 'decrypt', 'sign', 'verify', + 'DecryptionError', 'VerificationError', 'CryptoError'] + +if __name__ == '__main__': + print('Running doctests 1000x or until failure') + import doctest + + for count in range(1000): + (failures, tests) = doctest.testmod() + if failures: + break + + if count % 100 == 0 and count: + print('%i times' % count) + + print('Doctests done') diff --git a/contrib/python/rsa/py2/rsa/pkcs1_v2.py b/contrib/python/rsa/py2/rsa/pkcs1_v2.py new file mode 100644 index 0000000000..5f9c7ddcea --- /dev/null +++ b/contrib/python/rsa/py2/rsa/pkcs1_v2.py @@ -0,0 +1,103 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Functions for PKCS#1 version 2 encryption and signing + +This module implements certain functionality from PKCS#1 version 2. Main +documentation is RFC 2437: https://tools.ietf.org/html/rfc2437 +""" + +from rsa._compat import range +from rsa import ( + common, + pkcs1, + transform, +) + + +def mgf1(seed, length, hasher='SHA-1'): + """ + MGF1 is a Mask Generation Function based on a hash function. + + A mask generation function takes an octet string of variable length and a + desired output length as input, and outputs an octet string of the desired + length. The plaintext-awareness of RSAES-OAEP relies on the random nature of + the output of the mask generation function, which in turn relies on the + random nature of the underlying hash. + + :param bytes seed: seed from which mask is generated, an octet string + :param int length: intended length in octets of the mask, at most 2^32(hLen) + :param str hasher: hash function (hLen denotes the length in octets of the hash + function output) + + :return: mask, an octet string of length `length` + :rtype: bytes + + :raise OverflowError: when `length` is too large for the specified `hasher` + :raise ValueError: when specified `hasher` is invalid + """ + + try: + hash_length = pkcs1.HASH_METHODS[hasher]().digest_size + except KeyError: + raise ValueError( + 'Invalid `hasher` specified. Please select one of: {hash_list}'.format( + hash_list=', '.join(sorted(pkcs1.HASH_METHODS.keys())) + ) + ) + + # If l > 2^32(hLen), output "mask too long" and stop. + if length > (2**32 * hash_length): + raise OverflowError( + "Desired length should be at most 2**32 times the hasher's output " + "length ({hash_length} for {hasher} function)".format( + hash_length=hash_length, + hasher=hasher, + ) + ) + + # Looping `counter` from 0 to ceil(l / hLen)-1, build `output` based on the + # hashes formed by (`seed` + C), being `C` an octet string of length 4 + # generated by converting `counter` with the primitive I2OSP + output = b''.join( + pkcs1.compute_hash( + seed + transform.int2bytes(counter, fill_size=4), + method_name=hasher, + ) + for counter in range(common.ceil_div(length, hash_length) + 1) + ) + + # Output the leading `length` octets of `output` as the octet string mask. + return output[:length] + + +__all__ = [ + 'mgf1', +] + +if __name__ == '__main__': + print('Running doctests 1000x or until failure') + import doctest + + for count in range(1000): + (failures, tests) = doctest.testmod() + if failures: + break + + if count % 100 == 0 and count: + print('%i times' % count) + + print('Doctests done') diff --git a/contrib/python/rsa/py2/rsa/prime.py b/contrib/python/rsa/py2/rsa/prime.py new file mode 100644 index 0000000000..3d63542e66 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/prime.py @@ -0,0 +1,201 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Numerical functions related to primes. + +Implementation based on the book Algorithm Design by Michael T. Goodrich and +Roberto Tamassia, 2002. +""" + +from rsa._compat import range +import rsa.common +import rsa.randnum + +__all__ = ['getprime', 'are_relatively_prime'] + + +def gcd(p, q): + """Returns the greatest common divisor of p and q + + >>> gcd(48, 180) + 12 + """ + + while q != 0: + (p, q) = (q, p % q) + return p + + +def get_primality_testing_rounds(number): + """Returns minimum number of rounds for Miller-Rabing primality testing, + based on number bitsize. + + According to NIST FIPS 186-4, Appendix C, Table C.3, minimum number of + rounds of M-R testing, using an error probability of 2 ** (-100), for + different p, q bitsizes are: + * p, q bitsize: 512; rounds: 7 + * p, q bitsize: 1024; rounds: 4 + * p, q bitsize: 1536; rounds: 3 + See: http://nvlpubs.nist.gov/nistpubs/FIPS/NIST.FIPS.186-4.pdf + """ + + # Calculate number bitsize. + bitsize = rsa.common.bit_size(number) + # Set number of rounds. + if bitsize >= 1536: + return 3 + if bitsize >= 1024: + return 4 + if bitsize >= 512: + return 7 + # For smaller bitsizes, set arbitrary number of rounds. + return 10 + + +def miller_rabin_primality_testing(n, k): + """Calculates whether n is composite (which is always correct) or prime + (which theoretically is incorrect with error probability 4**-k), by + applying Miller-Rabin primality testing. + + For reference and implementation example, see: + https://en.wikipedia.org/wiki/Miller%E2%80%93Rabin_primality_test + + :param n: Integer to be tested for primality. + :type n: int + :param k: Number of rounds (witnesses) of Miller-Rabin testing. + :type k: int + :return: False if the number is composite, True if it's probably prime. + :rtype: bool + """ + + # prevent potential infinite loop when d = 0 + if n < 2: + return False + + # Decompose (n - 1) to write it as (2 ** r) * d + # While d is even, divide it by 2 and increase the exponent. + d = n - 1 + r = 0 + + while not (d & 1): + r += 1 + d >>= 1 + + # Test k witnesses. + for _ in range(k): + # Generate random integer a, where 2 <= a <= (n - 2) + a = rsa.randnum.randint(n - 3) + 1 + + x = pow(a, d, n) + if x == 1 or x == n - 1: + continue + + for _ in range(r - 1): + x = pow(x, 2, n) + if x == 1: + # n is composite. + return False + if x == n - 1: + # Exit inner loop and continue with next witness. + break + else: + # If loop doesn't break, n is composite. + return False + + return True + + +def is_prime(number): + """Returns True if the number is prime, and False otherwise. + + >>> is_prime(2) + True + >>> is_prime(42) + False + >>> is_prime(41) + True + """ + + # Check for small numbers. + if number < 10: + return number in {2, 3, 5, 7} + + # Check for even numbers. + if not (number & 1): + return False + + # Calculate minimum number of rounds. + k = get_primality_testing_rounds(number) + + # Run primality testing with (minimum + 1) rounds. + return miller_rabin_primality_testing(number, k + 1) + + +def getprime(nbits): + """Returns a prime number that can be stored in 'nbits' bits. + + >>> p = getprime(128) + >>> is_prime(p-1) + False + >>> is_prime(p) + True + >>> is_prime(p+1) + False + + >>> from rsa import common + >>> common.bit_size(p) == 128 + True + """ + + assert nbits > 3 # the loop wil hang on too small numbers + + while True: + integer = rsa.randnum.read_random_odd_int(nbits) + + # Test for primeness + if is_prime(integer): + return integer + + # Retry if not prime + + +def are_relatively_prime(a, b): + """Returns True if a and b are relatively prime, and False if they + are not. + + >>> are_relatively_prime(2, 3) + True + >>> are_relatively_prime(2, 4) + False + """ + + d = gcd(a, b) + return d == 1 + + +if __name__ == '__main__': + print('Running doctests 1000x or until failure') + import doctest + + for count in range(1000): + (failures, tests) = doctest.testmod() + if failures: + break + + if count % 100 == 0 and count: + print('%i times' % count) + + print('Doctests done') diff --git a/contrib/python/rsa/py2/rsa/randnum.py b/contrib/python/rsa/py2/rsa/randnum.py new file mode 100644 index 0000000000..310acaa620 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/randnum.py @@ -0,0 +1,98 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Functions for generating random numbers.""" + +# Source inspired by code by Yesudeep Mangalapilly <yesudeep@gmail.com> + +import os + +from rsa import common, transform +from rsa._compat import byte + + +def read_random_bits(nbits): + """Reads 'nbits' random bits. + + If nbits isn't a whole number of bytes, an extra byte will be appended with + only the lower bits set. + """ + + nbytes, rbits = divmod(nbits, 8) + + # Get the random bytes + randomdata = os.urandom(nbytes) + + # Add the remaining random bits + if rbits > 0: + randomvalue = ord(os.urandom(1)) + randomvalue >>= (8 - rbits) + randomdata = byte(randomvalue) + randomdata + + return randomdata + + +def read_random_int(nbits): + """Reads a random integer of approximately nbits bits. + """ + + randomdata = read_random_bits(nbits) + value = transform.bytes2int(randomdata) + + # Ensure that the number is large enough to just fill out the required + # number of bits. + value |= 1 << (nbits - 1) + + return value + + +def read_random_odd_int(nbits): + """Reads a random odd integer of approximately nbits bits. + + >>> read_random_odd_int(512) & 1 + 1 + """ + + value = read_random_int(nbits) + + # Make sure it's odd + return value | 1 + + +def randint(maxvalue): + """Returns a random integer x with 1 <= x <= maxvalue + + May take a very long time in specific situations. If maxvalue needs N bits + to store, the closer maxvalue is to (2 ** N) - 1, the faster this function + is. + """ + + bit_size = common.bit_size(maxvalue) + + tries = 0 + while True: + value = read_random_int(bit_size) + if value <= maxvalue: + break + + if tries % 10 == 0 and tries: + # After a lot of tries to get the right number of bits but still + # smaller than maxvalue, decrease the number of bits by 1. That'll + # dramatically increase the chances to get a large enough number. + bit_size -= 1 + tries += 1 + + return value diff --git a/contrib/python/rsa/py2/rsa/transform.py b/contrib/python/rsa/py2/rsa/transform.py new file mode 100644 index 0000000000..628d0afb55 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/transform.py @@ -0,0 +1,215 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Data transformation functions. + +From bytes to a number, number to bytes, etc. +""" + +from __future__ import absolute_import + +import binascii +from struct import pack + +from rsa._compat import byte, is_integer +from rsa import common, machine_size + + +def bytes2int(raw_bytes): + r"""Converts a list of bytes or an 8-bit string to an integer. + + When using unicode strings, encode it to some encoding like UTF8 first. + + >>> (((128 * 256) + 64) * 256) + 15 + 8405007 + >>> bytes2int(b'\x80@\x0f') + 8405007 + + """ + + return int(binascii.hexlify(raw_bytes), 16) + + +def _int2bytes(number, block_size=None): + r"""Converts a number to a string of bytes. + + Usage:: + + >>> _int2bytes(123456789) + b'\x07[\xcd\x15' + >>> bytes2int(_int2bytes(123456789)) + 123456789 + + >>> _int2bytes(123456789, 6) + b'\x00\x00\x07[\xcd\x15' + >>> bytes2int(_int2bytes(123456789, 128)) + 123456789 + + >>> _int2bytes(123456789, 3) + Traceback (most recent call last): + ... + OverflowError: Needed 4 bytes for number, but block size is 3 + + @param number: the number to convert + @param block_size: the number of bytes to output. If the number encoded to + bytes is less than this, the block will be zero-padded. When not given, + the returned block is not padded. + + @throws OverflowError when block_size is given and the number takes up more + bytes than fit into the block. + """ + + # Type checking + if not is_integer(number): + raise TypeError("You must pass an integer for 'number', not %s" % + number.__class__) + + if number < 0: + raise ValueError('Negative numbers cannot be used: %i' % number) + + # Do some bounds checking + if number == 0: + needed_bytes = 1 + raw_bytes = [b'\x00'] + else: + needed_bytes = common.byte_size(number) + raw_bytes = [] + + # You cannot compare None > 0 in Python 3x. It will fail with a TypeError. + if block_size and block_size > 0: + if needed_bytes > block_size: + raise OverflowError('Needed %i bytes for number, but block size ' + 'is %i' % (needed_bytes, block_size)) + + # Convert the number to bytes. + while number > 0: + raw_bytes.insert(0, byte(number & 0xFF)) + number >>= 8 + + # Pad with zeroes to fill the block + if block_size and block_size > 0: + padding = (block_size - needed_bytes) * b'\x00' + else: + padding = b'' + + return padding + b''.join(raw_bytes) + + +def bytes_leading(raw_bytes, needle=b'\x00'): + """ + Finds the number of prefixed byte occurrences in the haystack. + + Useful when you want to deal with padding. + + :param raw_bytes: + Raw bytes. + :param needle: + The byte to count. Default \x00. + :returns: + The number of leading needle bytes. + """ + + leading = 0 + # Indexing keeps compatibility between Python 2.x and Python 3.x + _byte = needle[0] + for x in raw_bytes: + if x == _byte: + leading += 1 + else: + break + return leading + + +def int2bytes(number, fill_size=None, chunk_size=None, overflow=False): + """ + Convert an unsigned integer to bytes (base-256 representation):: + + Does not preserve leading zeros if you don't specify a chunk size or + fill size. + + .. NOTE: + You must not specify both fill_size and chunk_size. Only one + of them is allowed. + + :param number: + Integer value + :param fill_size: + If the optional fill size is given the length of the resulting + byte string is expected to be the fill size and will be padded + with prefix zero bytes to satisfy that length. + :param chunk_size: + If optional chunk size is given and greater than zero, pad the front of + the byte string with binary zeros so that the length is a multiple of + ``chunk_size``. + :param overflow: + ``False`` (default). If this is ``True``, no ``OverflowError`` + will be raised when the fill_size is shorter than the length + of the generated byte sequence. Instead the byte sequence will + be returned as is. + :returns: + Raw bytes (base-256 representation). + :raises: + ``OverflowError`` when fill_size is given and the number takes up more + bytes than fit into the block. This requires the ``overflow`` + argument to this function to be set to ``False`` otherwise, no + error will be raised. + """ + + if number < 0: + raise ValueError("Number must be an unsigned integer: %d" % number) + + if fill_size and chunk_size: + raise ValueError("You can either fill or pad chunks, but not both") + + # Ensure these are integers. + number & 1 + + raw_bytes = b'' + + # Pack the integer one machine word at a time into bytes. + num = number + word_bits, _, max_uint, pack_type = machine_size.get_word_alignment(num) + pack_format = ">%s" % pack_type + while num > 0: + raw_bytes = pack(pack_format, num & max_uint) + raw_bytes + num >>= word_bits + # Obtain the index of the first non-zero byte. + zero_leading = bytes_leading(raw_bytes) + if number == 0: + raw_bytes = b'\x00' + # De-padding. + raw_bytes = raw_bytes[zero_leading:] + + length = len(raw_bytes) + if fill_size and fill_size > 0: + if not overflow and length > fill_size: + raise OverflowError( + "Need %d bytes for number, but fill size is %d" % + (length, fill_size) + ) + raw_bytes = raw_bytes.rjust(fill_size, b'\x00') + elif chunk_size and chunk_size > 0: + remainder = length % chunk_size + if remainder: + padding_size = chunk_size - remainder + raw_bytes = raw_bytes.rjust(length + padding_size, b'\x00') + return raw_bytes + + +if __name__ == '__main__': + import doctest + + doctest.testmod() diff --git a/contrib/python/rsa/py2/rsa/util.py b/contrib/python/rsa/py2/rsa/util.py new file mode 100644 index 0000000000..29d5eb1218 --- /dev/null +++ b/contrib/python/rsa/py2/rsa/util.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Utility functions.""" + +from __future__ import with_statement, print_function + +import sys +from optparse import OptionParser + +import rsa.key + + +def private_to_public(): + """Reads a private key and outputs the corresponding public key.""" + + # Parse the CLI options + parser = OptionParser(usage='usage: %prog [options]', + description='Reads a private key and outputs the ' + 'corresponding public key. Both private and public keys use ' + 'the format described in PKCS#1 v1.5') + + parser.add_option('-i', '--input', dest='infilename', type='string', + help='Input filename. Reads from stdin if not specified') + parser.add_option('-o', '--output', dest='outfilename', type='string', + help='Output filename. Writes to stdout of not specified') + + parser.add_option('--inform', dest='inform', + help='key format of input - default PEM', + choices=('PEM', 'DER'), default='PEM') + + parser.add_option('--outform', dest='outform', + help='key format of output - default PEM', + choices=('PEM', 'DER'), default='PEM') + + (cli, cli_args) = parser.parse_args(sys.argv) + + # Read the input data + if cli.infilename: + print('Reading private key from %s in %s format' % + (cli.infilename, cli.inform), file=sys.stderr) + with open(cli.infilename, 'rb') as infile: + in_data = infile.read() + else: + print('Reading private key from stdin in %s format' % cli.inform, + file=sys.stderr) + in_data = sys.stdin.read().encode('ascii') + + assert type(in_data) == bytes, type(in_data) + + # Take the public fields and create a public key + priv_key = rsa.key.PrivateKey.load_pkcs1(in_data, cli.inform) + pub_key = rsa.key.PublicKey(priv_key.n, priv_key.e) + + # Save to the output file + out_data = pub_key.save_pkcs1(cli.outform) + + if cli.outfilename: + print('Writing public key to %s in %s format' % + (cli.outfilename, cli.outform), file=sys.stderr) + with open(cli.outfilename, 'wb') as outfile: + outfile.write(out_data) + else: + print('Writing public key to stdout in %s format' % cli.outform, + file=sys.stderr) + sys.stdout.write(out_data.decode('ascii')) diff --git a/contrib/python/rsa/py2/tests/__init__.py b/contrib/python/rsa/py2/tests/__init__.py new file mode 100644 index 0000000000..e69de29bb2 --- /dev/null +++ b/contrib/python/rsa/py2/tests/__init__.py diff --git a/contrib/python/rsa/py2/tests/private.pem b/contrib/python/rsa/py2/tests/private.pem new file mode 100644 index 0000000000..1a17279f23 --- /dev/null +++ b/contrib/python/rsa/py2/tests/private.pem @@ -0,0 +1,5 @@ +-----BEGIN RSA PRIVATE KEY----- +MGECAQACEQCvWovlXBvfEeOMZPEleO9NAgMBAAECEA20Y+6fDkaWvC24horBzQEC +CQDdS2PAL/tK4QIJAMratZuNnT3tAghs7iNYA0ZrgQIIQQ5nU93U4fkCCHR55el6 +/K+2 +-----END RSA PRIVATE KEY----- diff --git a/contrib/python/rsa/py2/tests/test_cli.py b/contrib/python/rsa/py2/tests/test_cli.py new file mode 100644 index 0000000000..7ce57ebd99 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_cli.py @@ -0,0 +1,296 @@ +""" +Unit tests for CLI entry points. +""" + +from __future__ import print_function + +import unittest +import sys +import functools +from contextlib import contextmanager + +import os +from io import StringIO, BytesIO + +import rsa +import rsa.cli +import rsa.util +from rsa._compat import PY2 + + +def make_buffer(): + if PY2: + return BytesIO() + buf = StringIO() + buf.buffer = BytesIO() + return buf + + +def get_bytes_out(out): + if PY2: + # Python 2.x writes 'str' to stdout + return out.getvalue() + # Python 3.x writes 'bytes' to stdout.buffer + return out.buffer.getvalue() + + +@contextmanager +def captured_output(): + """Captures output to stdout and stderr""" + + new_out, new_err = make_buffer(), make_buffer() + old_out, old_err = sys.stdout, sys.stderr + try: + sys.stdout, sys.stderr = new_out, new_err + yield new_out, new_err + finally: + sys.stdout, sys.stderr = old_out, old_err + + +@contextmanager +def cli_args(*new_argv): + """Updates sys.argv[1:] for a single test.""" + + old_args = sys.argv[:] + sys.argv[1:] = [str(arg) for arg in new_argv] + + try: + yield + finally: + sys.argv[1:] = old_args + + +def remove_if_exists(fname): + """Removes a file if it exists.""" + + if os.path.exists(fname): + os.unlink(fname) + + +def cleanup_files(*filenames): + """Makes sure the files don't exist when the test runs, and deletes them afterward.""" + + def remove(): + for fname in filenames: + remove_if_exists(fname) + + def decorator(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + remove() + try: + return func(*args, **kwargs) + finally: + remove() + + return wrapper + + return decorator + + +class AbstractCliTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + # Ensure there is a key to use + cls.pub_key, cls.priv_key = rsa.newkeys(512) + cls.pub_fname = '%s.pub' % cls.__name__ + cls.priv_fname = '%s.key' % cls.__name__ + + with open(cls.pub_fname, 'wb') as outfile: + outfile.write(cls.pub_key.save_pkcs1()) + + with open(cls.priv_fname, 'wb') as outfile: + outfile.write(cls.priv_key.save_pkcs1()) + + @classmethod + def tearDownClass(cls): + if hasattr(cls, 'pub_fname'): + remove_if_exists(cls.pub_fname) + if hasattr(cls, 'priv_fname'): + remove_if_exists(cls.priv_fname) + + def assertExits(self, status_code, func, *args, **kwargs): + try: + func(*args, **kwargs) + except SystemExit as ex: + if status_code == ex.code: + return + self.fail('SystemExit() raised by %r, but exited with code %r, expected %r' % ( + func, ex.code, status_code)) + else: + self.fail('SystemExit() not raised by %r' % func) + + +class KeygenTest(AbstractCliTest): + def test_keygen_no_args(self): + with cli_args(): + self.assertExits(1, rsa.cli.keygen) + + def test_keygen_priv_stdout(self): + with captured_output() as (out, err): + with cli_args(128): + rsa.cli.keygen() + + lines = get_bytes_out(out).splitlines() + self.assertEqual(b'-----BEGIN RSA PRIVATE KEY-----', lines[0]) + self.assertEqual(b'-----END RSA PRIVATE KEY-----', lines[-1]) + + # The key size should be shown on stderr + self.assertTrue('128-bit key' in err.getvalue()) + + @cleanup_files('test_cli_privkey_out.pem') + def test_keygen_priv_out_pem(self): + with captured_output() as (out, err): + with cli_args('--out=test_cli_privkey_out.pem', '--form=PEM', 128): + rsa.cli.keygen() + + # The key size should be shown on stderr + self.assertTrue('128-bit key' in err.getvalue()) + + # The output file should be shown on stderr + self.assertTrue('test_cli_privkey_out.pem' in err.getvalue()) + + # If we can load the file as PEM, it's good enough. + with open('test_cli_privkey_out.pem', 'rb') as pemfile: + rsa.PrivateKey.load_pkcs1(pemfile.read()) + + @cleanup_files('test_cli_privkey_out.der') + def test_keygen_priv_out_der(self): + with captured_output() as (out, err): + with cli_args('--out=test_cli_privkey_out.der', '--form=DER', 128): + rsa.cli.keygen() + + # The key size should be shown on stderr + self.assertTrue('128-bit key' in err.getvalue()) + + # The output file should be shown on stderr + self.assertTrue('test_cli_privkey_out.der' in err.getvalue()) + + # If we can load the file as der, it's good enough. + with open('test_cli_privkey_out.der', 'rb') as derfile: + rsa.PrivateKey.load_pkcs1(derfile.read(), format='DER') + + @cleanup_files('test_cli_privkey_out.pem', 'test_cli_pubkey_out.pem') + def test_keygen_pub_out_pem(self): + with captured_output() as (out, err): + with cli_args('--out=test_cli_privkey_out.pem', + '--pubout=test_cli_pubkey_out.pem', + '--form=PEM', 256): + rsa.cli.keygen() + + # The key size should be shown on stderr + self.assertTrue('256-bit key' in err.getvalue()) + + # The output files should be shown on stderr + self.assertTrue('test_cli_privkey_out.pem' in err.getvalue()) + self.assertTrue('test_cli_pubkey_out.pem' in err.getvalue()) + + # If we can load the file as PEM, it's good enough. + with open('test_cli_pubkey_out.pem', 'rb') as pemfile: + rsa.PublicKey.load_pkcs1(pemfile.read()) + + +class EncryptDecryptTest(AbstractCliTest): + def test_empty_decrypt(self): + with cli_args(): + self.assertExits(1, rsa.cli.decrypt) + + def test_empty_encrypt(self): + with cli_args(): + self.assertExits(1, rsa.cli.encrypt) + + @cleanup_files('encrypted.txt', 'cleartext.txt') + def test_encrypt_decrypt(self): + with open('cleartext.txt', 'wb') as outfile: + outfile.write(b'Hello cleartext RSA users!') + + with cli_args('-i', 'cleartext.txt', '--out=encrypted.txt', self.pub_fname): + with captured_output(): + rsa.cli.encrypt() + + with cli_args('-i', 'encrypted.txt', self.priv_fname): + with captured_output() as (out, err): + rsa.cli.decrypt() + + # We should have the original cleartext on stdout now. + output = get_bytes_out(out) + self.assertEqual(b'Hello cleartext RSA users!', output) + + @cleanup_files('encrypted.txt', 'cleartext.txt') + def test_encrypt_decrypt_unhappy(self): + with open('cleartext.txt', 'wb') as outfile: + outfile.write(b'Hello cleartext RSA users!') + + with cli_args('-i', 'cleartext.txt', '--out=encrypted.txt', self.pub_fname): + with captured_output(): + rsa.cli.encrypt() + + # Change a few bytes in the encrypted stream. + with open('encrypted.txt', 'r+b') as encfile: + encfile.seek(40) + encfile.write(b'hahaha') + + with cli_args('-i', 'encrypted.txt', self.priv_fname): + with captured_output() as (out, err): + self.assertRaises(rsa.DecryptionError, rsa.cli.decrypt) + + +class SignVerifyTest(AbstractCliTest): + def test_empty_verify(self): + with cli_args(): + self.assertExits(1, rsa.cli.verify) + + def test_empty_sign(self): + with cli_args(): + self.assertExits(1, rsa.cli.sign) + + @cleanup_files('signature.txt', 'cleartext.txt') + def test_sign_verify(self): + with open('cleartext.txt', 'wb') as outfile: + outfile.write(b'Hello RSA users!') + + with cli_args('-i', 'cleartext.txt', '--out=signature.txt', self.priv_fname, 'SHA-256'): + with captured_output(): + rsa.cli.sign() + + with cli_args('-i', 'cleartext.txt', self.pub_fname, 'signature.txt'): + with captured_output() as (out, err): + rsa.cli.verify() + + self.assertFalse(b'Verification OK' in get_bytes_out(out)) + + @cleanup_files('signature.txt', 'cleartext.txt') + def test_sign_verify_unhappy(self): + with open('cleartext.txt', 'wb') as outfile: + outfile.write(b'Hello RSA users!') + + with cli_args('-i', 'cleartext.txt', '--out=signature.txt', self.priv_fname, 'SHA-256'): + with captured_output(): + rsa.cli.sign() + + # Change a few bytes in the cleartext file. + with open('cleartext.txt', 'r+b') as encfile: + encfile.seek(6) + encfile.write(b'DSA') + + with cli_args('-i', 'cleartext.txt', self.pub_fname, 'signature.txt'): + with captured_output() as (out, err): + self.assertExits('Verification failed.', rsa.cli.verify) + + +class PrivatePublicTest(AbstractCliTest): + """Test CLI command to convert a private to a public key.""" + + @cleanup_files('test_private_to_public.pem') + def test_private_to_public(self): + + with cli_args('-i', self.priv_fname, '-o', 'test_private_to_public.pem'): + with captured_output(): + rsa.util.private_to_public() + + # Check that the key is indeed valid. + with open('test_private_to_public.pem', 'rb') as pemfile: + key = rsa.PublicKey.load_pkcs1(pemfile.read()) + + self.assertEqual(self.priv_key.n, key.n) + self.assertEqual(self.priv_key.e, key.e) diff --git a/contrib/python/rsa/py2/tests/test_common.py b/contrib/python/rsa/py2/tests/test_common.py new file mode 100644 index 0000000000..af13695a7e --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_common.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import unittest +import struct +from rsa._compat import byte +from rsa.common import byte_size, bit_size, inverse + + +class TestByte(unittest.TestCase): + def test_values(self): + self.assertEqual(byte(0), b'\x00') + self.assertEqual(byte(255), b'\xff') + + def test_struct_error_when_out_of_bounds(self): + self.assertRaises(struct.error, byte, 256) + self.assertRaises(struct.error, byte, -1) + + +class TestByteSize(unittest.TestCase): + def test_values(self): + self.assertEqual(byte_size(1 << 1023), 128) + self.assertEqual(byte_size((1 << 1024) - 1), 128) + self.assertEqual(byte_size(1 << 1024), 129) + self.assertEqual(byte_size(255), 1) + self.assertEqual(byte_size(256), 2) + self.assertEqual(byte_size(0xffff), 2) + self.assertEqual(byte_size(0xffffff), 3) + self.assertEqual(byte_size(0xffffffff), 4) + self.assertEqual(byte_size(0xffffffffff), 5) + self.assertEqual(byte_size(0xffffffffffff), 6) + self.assertEqual(byte_size(0xffffffffffffff), 7) + self.assertEqual(byte_size(0xffffffffffffffff), 8) + + def test_zero(self): + self.assertEqual(byte_size(0), 1) + + def test_bad_type(self): + self.assertRaises(TypeError, byte_size, []) + self.assertRaises(TypeError, byte_size, ()) + self.assertRaises(TypeError, byte_size, dict()) + self.assertRaises(TypeError, byte_size, "") + self.assertRaises(TypeError, byte_size, None) + + +class TestBitSize(unittest.TestCase): + def test_zero(self): + self.assertEqual(bit_size(0), 0) + + def test_values(self): + self.assertEqual(bit_size(1023), 10) + self.assertEqual(bit_size(1024), 11) + self.assertEqual(bit_size(1025), 11) + self.assertEqual(bit_size(1 << 1024), 1025) + self.assertEqual(bit_size((1 << 1024) + 1), 1025) + self.assertEqual(bit_size((1 << 1024) - 1), 1024) + + def test_negative_values(self): + self.assertEqual(bit_size(-1023), 10) + self.assertEqual(bit_size(-1024), 11) + self.assertEqual(bit_size(-1025), 11) + self.assertEqual(bit_size(-1 << 1024), 1025) + self.assertEqual(bit_size(-((1 << 1024) + 1)), 1025) + self.assertEqual(bit_size(-((1 << 1024) - 1)), 1024) + + def test_bad_type(self): + self.assertRaises(TypeError, bit_size, []) + self.assertRaises(TypeError, bit_size, ()) + self.assertRaises(TypeError, bit_size, dict()) + self.assertRaises(TypeError, bit_size, "") + self.assertRaises(TypeError, bit_size, None) + self.assertRaises(TypeError, bit_size, 0.0) + + +class TestInverse(unittest.TestCase): + def test_normal(self): + self.assertEqual(3, inverse(7, 4)) + self.assertEqual(9, inverse(5, 11)) + + def test_not_relprime(self): + self.assertRaises(ValueError, inverse, 4, 8) + self.assertRaises(ValueError, inverse, 25, 5) diff --git a/contrib/python/rsa/py2/tests/test_compat.py b/contrib/python/rsa/py2/tests/test_compat.py new file mode 100644 index 0000000000..62e933f25f --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_compat.py @@ -0,0 +1,80 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import unittest +import struct + +from rsa._compat import byte, is_bytes, range, xor_bytes + + +class TestByte(unittest.TestCase): + """Tests for single bytes.""" + + def test_byte(self): + for i in range(256): + byt = byte(i) + self.assertTrue(is_bytes(byt)) + self.assertEqual(ord(byt), i) + + def test_raises_StructError_on_overflow(self): + self.assertRaises(struct.error, byte, 256) + self.assertRaises(struct.error, byte, -1) + + def test_byte_literal(self): + self.assertIsInstance(b'abc', bytes) + + +class TestBytes(unittest.TestCase): + """Tests for bytes objects.""" + + def setUp(self): + self.b1 = b'\xff\xff\xff\xff' + self.b2 = b'\x00\x00\x00\x00' + self.b3 = b'\xf0\xf0\xf0\xf0' + self.b4 = b'\x4d\x23\xca\xe2' + self.b5 = b'\x9b\x61\x3b\xdc' + self.b6 = b'\xff\xff' + + self.byte_strings = (self.b1, self.b2, self.b3, self.b4, self.b5, self.b6) + + def test_xor_bytes(self): + self.assertEqual(xor_bytes(self.b1, self.b2), b'\xff\xff\xff\xff') + self.assertEqual(xor_bytes(self.b1, self.b3), b'\x0f\x0f\x0f\x0f') + self.assertEqual(xor_bytes(self.b1, self.b4), b'\xb2\xdc\x35\x1d') + self.assertEqual(xor_bytes(self.b1, self.b5), b'\x64\x9e\xc4\x23') + self.assertEqual(xor_bytes(self.b2, self.b3), b'\xf0\xf0\xf0\xf0') + self.assertEqual(xor_bytes(self.b2, self.b4), b'\x4d\x23\xca\xe2') + self.assertEqual(xor_bytes(self.b2, self.b5), b'\x9b\x61\x3b\xdc') + self.assertEqual(xor_bytes(self.b3, self.b4), b'\xbd\xd3\x3a\x12') + self.assertEqual(xor_bytes(self.b3, self.b5), b'\x6b\x91\xcb\x2c') + self.assertEqual(xor_bytes(self.b4, self.b5), b'\xd6\x42\xf1\x3e') + + def test_xor_bytes_length(self): + self.assertEqual(xor_bytes(self.b1, self.b6), b'\x00\x00') + self.assertEqual(xor_bytes(self.b2, self.b6), b'\xff\xff') + self.assertEqual(xor_bytes(self.b3, self.b6), b'\x0f\x0f') + self.assertEqual(xor_bytes(self.b4, self.b6), b'\xb2\xdc') + self.assertEqual(xor_bytes(self.b5, self.b6), b'\x64\x9e') + self.assertEqual(xor_bytes(self.b6, b''), b'') + + def test_xor_bytes_commutative(self): + for first in self.byte_strings: + for second in self.byte_strings: + min_length = min(len(first), len(second)) + result = xor_bytes(first, second) + + self.assertEqual(result, xor_bytes(second, first)) + self.assertEqual(len(result), min_length) diff --git a/contrib/python/rsa/py2/tests/test_integers.py b/contrib/python/rsa/py2/tests/test_integers.py new file mode 100644 index 0000000000..fb29ba41d3 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_integers.py @@ -0,0 +1,50 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests integer operations.""" + +import unittest + +import rsa +import rsa.core + + +class IntegerTest(unittest.TestCase): + def setUp(self): + (self.pub, self.priv) = rsa.newkeys(64) + + def test_enc_dec(self): + message = 42 + print("\tMessage: %d" % message) + + encrypted = rsa.core.encrypt_int(message, self.pub.e, self.pub.n) + print("\tEncrypted: %d" % encrypted) + + decrypted = rsa.core.decrypt_int(encrypted, self.priv.d, self.pub.n) + print("\tDecrypted: %d" % decrypted) + + self.assertEqual(message, decrypted) + + def test_sign_verify(self): + message = 42 + + signed = rsa.core.encrypt_int(message, self.priv.d, self.pub.n) + print("\tSigned: %d" % signed) + + verified = rsa.core.decrypt_int(signed, self.pub.e, self.pub.n) + print("\tVerified: %d" % verified) + + self.assertEqual(message, verified) diff --git a/contrib/python/rsa/py2/tests/test_key.py b/contrib/python/rsa/py2/tests/test_key.py new file mode 100644 index 0000000000..9db30cedf6 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_key.py @@ -0,0 +1,79 @@ +""" +Some tests for the rsa/key.py file. +""" + +import unittest + +import rsa.key +import rsa.core + + +class BlindingTest(unittest.TestCase): + def test_blinding(self): + """Test blinding and unblinding. + + This is basically the doctest of the PrivateKey.blind method, but then + implemented as unittest to allow running on different Python versions. + """ + + pk = rsa.key.PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + + message = 12345 + encrypted = rsa.core.encrypt_int(message, pk.e, pk.n) + + blinded = pk.blind(encrypted, 4134431) # blind before decrypting + decrypted = rsa.core.decrypt_int(blinded, pk.d, pk.n) + unblinded = pk.unblind(decrypted, 4134431) + + self.assertEqual(unblinded, message) + + +class KeyGenTest(unittest.TestCase): + def test_custom_exponent(self): + priv, pub = rsa.key.newkeys(16, exponent=3) + + self.assertEqual(3, priv.e) + self.assertEqual(3, pub.e) + + def test_default_exponent(self): + priv, pub = rsa.key.newkeys(16) + + self.assertEqual(0x10001, priv.e) + self.assertEqual(0x10001, pub.e) + + def test_exponents_coefficient_calculation(self): + pk = rsa.key.PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + + self.assertEqual(pk.exp1, 55063) + self.assertEqual(pk.exp2, 10095) + self.assertEqual(pk.coef, 50797) + + def test_custom_getprime_func(self): + # List of primes to test with, in order [p, q, p, q, ....] + # By starting with two of the same primes, we test that this is + # properly rejected. + primes = [64123, 64123, 64123, 50957, 39317, 33107] + + def getprime(_): + return primes.pop(0) + + # This exponent will cause two other primes to be generated. + exponent = 136407 + + (p, q, e, d) = rsa.key.gen_keys(64, + accurate=False, + getprime_func=getprime, + exponent=exponent) + self.assertEqual(39317, p) + self.assertEqual(33107, q) + + +class HashTest(unittest.TestCase): + """Test hashing of keys""" + + def test_hash_possible(self): + priv, pub = rsa.key.newkeys(16) + + # This raises a TypeError when hashing isn't possible. + hash(priv) + hash(pub) diff --git a/contrib/python/rsa/py2/tests/test_load_save_keys.py b/contrib/python/rsa/py2/tests/test_load_save_keys.py new file mode 100644 index 0000000000..967c946e74 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_load_save_keys.py @@ -0,0 +1,217 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unittest for saving and loading keys.""" + +import base64 +import mock +import os.path +import pickle +import unittest +import warnings + +from rsa._compat import range +import rsa.key + +B64PRIV_DER = b'MC4CAQACBQDeKYlRAgMBAAECBQDHn4npAgMA/icCAwDfxwIDANcXAgInbwIDAMZt' +PRIVATE_DER = base64.standard_b64decode(B64PRIV_DER) + +B64PUB_DER = b'MAwCBQDeKYlRAgMBAAE=' +PUBLIC_DER = base64.standard_b64decode(B64PUB_DER) + +PRIVATE_PEM = b'''\ +-----BEGIN CONFUSING STUFF----- +Cruft before the key + +-----BEGIN RSA PRIVATE KEY----- +Comment: something blah + +''' + B64PRIV_DER + b''' +-----END RSA PRIVATE KEY----- + +Stuff after the key +-----END CONFUSING STUFF----- +''' + +CLEAN_PRIVATE_PEM = b'''\ +-----BEGIN RSA PRIVATE KEY----- +''' + B64PRIV_DER + b''' +-----END RSA PRIVATE KEY----- +''' + +PUBLIC_PEM = b'''\ +-----BEGIN CONFUSING STUFF----- +Cruft before the key + +-----BEGIN RSA PUBLIC KEY----- +Comment: something blah + +''' + B64PUB_DER + b''' +-----END RSA PUBLIC KEY----- + +Stuff after the key +-----END CONFUSING STUFF----- +''' + +CLEAN_PUBLIC_PEM = b'''\ +-----BEGIN RSA PUBLIC KEY----- +''' + B64PUB_DER + b''' +-----END RSA PUBLIC KEY----- +''' + + +class DerTest(unittest.TestCase): + """Test saving and loading DER keys.""" + + def test_load_private_key(self): + """Test loading private DER keys.""" + + key = rsa.key.PrivateKey.load_pkcs1(PRIVATE_DER, 'DER') + expected = rsa.key.PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + + self.assertEqual(expected, key) + self.assertEqual(key.exp1, 55063) + self.assertEqual(key.exp2, 10095) + self.assertEqual(key.coef, 50797) + + @mock.patch('pyasn1.codec.der.decoder.decode') + def test_load_malformed_private_key(self, der_decode): + """Test loading malformed private DER keys.""" + + # Decode returns an invalid exp2 value. + der_decode.return_value = ( + [0, 3727264081, 65537, 3349121513, 65063, 57287, 55063, 0, 50797], + 0, + ) + + with warnings.catch_warnings(record=True) as w: + # Always print warnings + warnings.simplefilter('always') + + # Load 3 keys + for _ in range(3): + key = rsa.key.PrivateKey.load_pkcs1(PRIVATE_DER, 'DER') + + # Check that 3 warnings were generated. + self.assertEqual(3, len(w)) + + for warning in w: + self.assertTrue(issubclass(warning.category, UserWarning)) + self.assertIn('malformed', str(warning.message)) + + # Check that we are creating the key with correct values + self.assertEqual(key.exp1, 55063) + self.assertEqual(key.exp2, 10095) + self.assertEqual(key.coef, 50797) + + def test_save_private_key(self): + """Test saving private DER keys.""" + + key = rsa.key.PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + der = key.save_pkcs1('DER') + + self.assertIsInstance(der, bytes) + self.assertEqual(PRIVATE_DER, der) + + def test_load_public_key(self): + """Test loading public DER keys.""" + + key = rsa.key.PublicKey.load_pkcs1(PUBLIC_DER, 'DER') + expected = rsa.key.PublicKey(3727264081, 65537) + + self.assertEqual(expected, key) + + def test_save_public_key(self): + """Test saving public DER keys.""" + + key = rsa.key.PublicKey(3727264081, 65537) + der = key.save_pkcs1('DER') + + self.assertIsInstance(der, bytes) + self.assertEqual(PUBLIC_DER, der) + + +class PemTest(unittest.TestCase): + """Test saving and loading PEM keys.""" + + def test_load_private_key(self): + """Test loading private PEM files.""" + + key = rsa.key.PrivateKey.load_pkcs1(PRIVATE_PEM, 'PEM') + expected = rsa.key.PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + + self.assertEqual(expected, key) + self.assertEqual(key.exp1, 55063) + self.assertEqual(key.exp2, 10095) + self.assertEqual(key.coef, 50797) + + def test_save_private_key(self): + """Test saving private PEM files.""" + + key = rsa.key.PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + pem = key.save_pkcs1('PEM') + + self.assertIsInstance(pem, bytes) + self.assertEqual(CLEAN_PRIVATE_PEM, pem) + + def test_load_public_key(self): + """Test loading public PEM files.""" + + key = rsa.key.PublicKey.load_pkcs1(PUBLIC_PEM, 'PEM') + expected = rsa.key.PublicKey(3727264081, 65537) + + self.assertEqual(expected, key) + + def test_save_public_key(self): + """Test saving public PEM files.""" + + key = rsa.key.PublicKey(3727264081, 65537) + pem = key.save_pkcs1('PEM') + + self.assertIsInstance(pem, bytes) + self.assertEqual(CLEAN_PUBLIC_PEM, pem) + + def test_load_from_disk(self): + from yatest.common import source_path + + """Test loading a PEM file from disk.""" + + fname = source_path('contrib/python/rsa/py2/tests/private.pem') + with open(fname, mode='rb') as privatefile: + keydata = privatefile.read() + privkey = rsa.key.PrivateKey.load_pkcs1(keydata) + + self.assertEqual(15945948582725241569, privkey.p) + self.assertEqual(14617195220284816877, privkey.q) + + +class PickleTest(unittest.TestCase): + """Test saving and loading keys by pickling.""" + + def test_private_key(self): + pk = rsa.key.PrivateKey(3727264081, 65537, 3349121513, 65063, 57287) + + pickled = pickle.dumps(pk) + unpickled = pickle.loads(pickled) + self.assertEqual(pk, unpickled) + + def test_public_key(self): + pk = rsa.key.PublicKey(3727264081, 65537) + + pickled = pickle.dumps(pk) + unpickled = pickle.loads(pickled) + + self.assertEqual(pk, unpickled) diff --git a/contrib/python/rsa/py2/tests/test_parallel.py b/contrib/python/rsa/py2/tests/test_parallel.py new file mode 100644 index 0000000000..1a69e9ece6 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_parallel.py @@ -0,0 +1,20 @@ +"""Test for multiprocess prime generation.""" + +import unittest + +import rsa.prime +import rsa.parallel +import rsa.common + + +class ParallelTest(unittest.TestCase): + """Tests for multiprocess prime generation.""" + + def test_parallel_primegen(self): + p = rsa.parallel.getprime(1024, 3) + + self.assertFalse(rsa.prime.is_prime(p - 1)) + self.assertTrue(rsa.prime.is_prime(p)) + self.assertFalse(rsa.prime.is_prime(p + 1)) + + self.assertEqual(1024, rsa.common.bit_size(p)) diff --git a/contrib/python/rsa/py2/tests/test_pem.py b/contrib/python/rsa/py2/tests/test_pem.py new file mode 100644 index 0000000000..5fb96002af --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_pem.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import unittest + +from rsa._compat import is_bytes +from rsa.pem import _markers +import rsa.key + +# 512-bit key. Too small for practical purposes, but good enough for testing with. +public_key_pem = ''' +-----BEGIN PUBLIC KEY----- +MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBAKH0aYP9ZFuctlPnXhEyHjgc8ltKKx9M +0c+h4sKMXwjhjbQAZdtWIw8RRghpUJnKj+6bN2XzZDazyULxgPhtax0CAwEAAQ== +-----END PUBLIC KEY----- +''' + +private_key_pem = ''' +-----BEGIN RSA PRIVATE KEY----- +MIIBOwIBAAJBAKH0aYP9ZFuctlPnXhEyHjgc8ltKKx9M0c+h4sKMXwjhjbQAZdtW +Iw8RRghpUJnKj+6bN2XzZDazyULxgPhtax0CAwEAAQJADwR36EpNzQTqDzusCFIq +ZS+h9X8aIovgBK3RNhMIGO2ThpsnhiDTcqIvgQ56knbl6B2W4iOl54tJ6CNtf6l6 +zQIhANTaNLFGsJfOvZHcI0WL1r89+1A4JVxR+lpslJJwAvgDAiEAwsjqqZ2wY2F0 +F8p1J98BEbtjU2mEZIVCMn6vQuhWdl8CIDRL4IJl4eGKlB0QP0JJF1wpeGO/R76l +DaPF5cMM7k3NAiEAss28m/ck9BWBfFVdNjx/vsdFZkx2O9AX9EJWoBSnSgECIQCa ++sVQMUVJFGsdE/31C7wCIbE3IpB7ziABZ7mN+V3Dhg== +-----END RSA PRIVATE KEY----- +''' + +# Private key components +prime1 = 96275860229939261876671084930484419185939191875438854026071315955024109172739 +prime2 = 88103681619592083641803383393198542599284510949756076218404908654323473741407 + + +class TestMarkers(unittest.TestCase): + def test_values(self): + self.assertEqual(_markers('RSA PRIVATE KEY'), + (b'-----BEGIN RSA PRIVATE KEY-----', + b'-----END RSA PRIVATE KEY-----')) + + +class TestBytesAndStrings(unittest.TestCase): + """Test that we can use PEM in both Unicode strings and bytes.""" + + def test_unicode_public(self): + key = rsa.key.PublicKey.load_pkcs1_openssl_pem(public_key_pem) + self.assertEqual(prime1 * prime2, key.n) + + def test_bytes_public(self): + key = rsa.key.PublicKey.load_pkcs1_openssl_pem(public_key_pem.encode('ascii')) + self.assertEqual(prime1 * prime2, key.n) + + def test_unicode_private(self): + key = rsa.key.PrivateKey.load_pkcs1(private_key_pem) + self.assertEqual(prime1 * prime2, key.n) + + def test_bytes_private(self): + key = rsa.key.PrivateKey.load_pkcs1(private_key_pem.encode('ascii')) + self.assertEqual(prime1, key.p) + self.assertEqual(prime2, key.q) + + +class TestByteOutput(unittest.TestCase): + """Tests that PEM and DER are returned as bytes.""" + + def test_bytes_public(self): + key = rsa.key.PublicKey.load_pkcs1_openssl_pem(public_key_pem) + self.assertTrue(is_bytes(key.save_pkcs1(format='DER'))) + self.assertTrue(is_bytes(key.save_pkcs1(format='PEM'))) + + def test_bytes_private(self): + key = rsa.key.PrivateKey.load_pkcs1(private_key_pem) + self.assertTrue(is_bytes(key.save_pkcs1(format='DER'))) + self.assertTrue(is_bytes(key.save_pkcs1(format='PEM'))) + + +class TestByteInput(unittest.TestCase): + """Tests that PEM and DER can be loaded from bytes.""" + + def test_bytes_public(self): + key = rsa.key.PublicKey.load_pkcs1_openssl_pem(public_key_pem.encode('ascii')) + self.assertTrue(is_bytes(key.save_pkcs1(format='DER'))) + self.assertTrue(is_bytes(key.save_pkcs1(format='PEM'))) + + def test_bytes_private(self): + key = rsa.key.PrivateKey.load_pkcs1(private_key_pem.encode('ascii')) + self.assertTrue(is_bytes(key.save_pkcs1(format='DER'))) + self.assertTrue(is_bytes(key.save_pkcs1(format='PEM'))) diff --git a/contrib/python/rsa/py2/tests/test_pkcs1.py b/contrib/python/rsa/py2/tests/test_pkcs1.py new file mode 100644 index 0000000000..9f7dcea7ad --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_pkcs1.py @@ -0,0 +1,184 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests string operations.""" + +import struct +import sys +import unittest + +import rsa +from rsa import pkcs1 +from rsa._compat import byte, is_bytes + + +class BinaryTest(unittest.TestCase): + def setUp(self): + (self.pub, self.priv) = rsa.newkeys(256) + + def test_enc_dec(self): + message = struct.pack('>IIII', 0, 0, 0, 1) + print("\tMessage: %r" % message) + + encrypted = pkcs1.encrypt(message, self.pub) + print("\tEncrypted: %r" % encrypted) + + decrypted = pkcs1.decrypt(encrypted, self.priv) + print("\tDecrypted: %r" % decrypted) + + self.assertEqual(message, decrypted) + + def test_decoding_failure(self): + message = struct.pack('>IIII', 0, 0, 0, 1) + encrypted = pkcs1.encrypt(message, self.pub) + + # Alter the encrypted stream + a = encrypted[5] + if is_bytes(a): + a = ord(a) + altered_a = (a + 1) % 256 + encrypted = encrypted[:5] + byte(altered_a) + encrypted[6:] + + self.assertRaises(pkcs1.DecryptionError, pkcs1.decrypt, encrypted, + self.priv) + + def test_randomness(self): + """Encrypting the same message twice should result in different + cryptos. + """ + + message = struct.pack('>IIII', 0, 0, 0, 1) + encrypted1 = pkcs1.encrypt(message, self.pub) + encrypted2 = pkcs1.encrypt(message, self.pub) + + self.assertNotEqual(encrypted1, encrypted2) + + +class ExtraZeroesTest(unittest.TestCase): + def setUp(self): + # Key, cyphertext, and plaintext taken from https://github.com/sybrenstuvel/python-rsa/issues/146 + self.private_key = rsa.PrivateKey.load_pkcs1( + "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEAs1EKK81M5kTFtZSuUFnhKy8FS2WNXaWVmi/fGHG4CLw98+Yo\n0nkuUarVwSS0O9pFPcpc3kvPKOe9Tv+6DLS3Qru21aATy2PRqjqJ4CYn71OYtSwM\n/ZfSCKvrjXybzgu+sBmobdtYm+sppbdL+GEHXGd8gdQw8DDCZSR6+dPJFAzLZTCd\nB+Ctwe/RXPF+ewVdfaOGjkZIzDoYDw7n+OHnsYCYozkbTOcWHpjVevipR+IBpGPi\n1rvKgFnlcG6d/tj0hWRl/6cS7RqhjoiNEtxqoJzpXs/Kg8xbCxXbCchkf11STA8u\ndiCjQWuWI8rcDwl69XMmHJjIQAqhKvOOQ8rYTQIDAQABAoIBABpQLQ7qbHtp4h1Y\nORAfcFRW7Q74UvtH/iEHH1TF8zyM6wZsYtcn4y0mxYE3Mp+J0xlTJbeVJkwZXYVH\nL3UH29CWHSlR+TWiazTwrCTRVJDhEoqbcTiRW8fb+o/jljVxMcVDrpyYUHNo2c6w\njBxhmKPtp66hhaDpds1Cwi0A8APZ8Z2W6kya/L/hRBzMgCz7Bon1nYBMak5PQEwV\nF0dF7Wy4vIjvCzO6DSqA415DvJDzUAUucgFudbANNXo4HJwNRnBpymYIh8mHdmNJ\n/MQ0YLSqUWvOB57dh7oWQwe3UsJ37ZUorTugvxh3NJ7Tt5ZqbCQBEECb9ND63gxo\n/a3YR/0CgYEA7BJc834xCi/0YmO5suBinWOQAF7IiRPU+3G9TdhWEkSYquupg9e6\nK9lC5k0iP+t6I69NYF7+6mvXDTmv6Z01o6oV50oXaHeAk74O3UqNCbLe9tybZ/+F\ndkYlwuGSNttMQBzjCiVy0+y0+Wm3rRnFIsAtd0RlZ24aN3bFTWJINIsCgYEAwnQq\nvNmJe9SwtnH5c/yCqPhKv1cF/4jdQZSGI6/p3KYNxlQzkHZ/6uvrU5V27ov6YbX8\nvKlKfO91oJFQxUD6lpTdgAStI3GMiJBJIZNpyZ9EWNSvwUj28H34cySpbZz3s4Xd\nhiJBShgy+fKURvBQwtWmQHZJ3EGrcOI7PcwiyYcCgYEAlql5jSUCY0ALtidzQogW\nJ+B87N+RGHsBuJ/0cxQYinwg+ySAAVbSyF1WZujfbO/5+YBN362A/1dn3lbswCnH\nK/bHF9+fZNqvwprPnceQj5oK1n4g6JSZNsy6GNAhosT+uwQ0misgR8SQE4W25dDG\nkdEYsz+BgCsyrCcu8J5C+tUCgYAFVPQbC4f2ikVyKzvgz0qx4WUDTBqRACq48p6e\n+eLatv7nskVbr7QgN+nS9+Uz80ihR0Ev1yCAvnwmM/XYAskcOea87OPmdeWZlQM8\nVXNwINrZ6LMNBLgorfuTBK1UoRo1pPUHCYdqxbEYI2unak18mikd2WB7Fp3h0YI4\nVpGZnwKBgBxkAYnZv+jGI4MyEKdsQgxvROXXYOJZkWzsKuKxVkVpYP2V4nR2YMOJ\nViJQ8FUEnPq35cMDlUk4SnoqrrHIJNOvcJSCqM+bWHAioAsfByLbUPM8sm3CDdIk\nXVJl32HuKYPJOMIWfc7hIfxLRHnCN+coz2M6tgqMDs0E/OfjuqVZ\n-----END RSA PRIVATE KEY-----", + format='PEM') + cyphertext = "4501b4d669e01b9ef2dc800aa1b06d49196f5a09fe8fbcd037323c60eaf027bfb98432be4e4a26c567ffec718bcbea977dd26812fa071c33808b4d5ebb742d9879806094b6fbeea63d25ea3141733b60e31c6912106e1b758a7fe0014f075193faa8b4622bfd5d3013f0a32190a95de61a3604711bc62945f95a6522bd4dfed0a994ef185b28c281f7b5e4c8ed41176d12d9fc1b837e6a0111d0132d08a6d6f0580de0c9eed8ed105531799482d1e466c68c23b0c222af7fc12ac279bc4ff57e7b4586d209371b38c4c1035edd418dc5f960441cb21ea2bedbfea86de0d7861e81021b650a1de51002c315f1e7c12debe4dcebf790caaa54a2f26b149cf9e77d" + plaintext = "54657374" + + if sys.version_info < (3, 0): + self.cyphertext = cyphertext.decode("hex") + self.plaintext = plaintext.decode('hex') + else: + self.cyphertext = bytes.fromhex(cyphertext) + self.plaintext = bytes.fromhex(plaintext) + + def test_unmodified(self): + message = rsa.decrypt(self.cyphertext, self.private_key) + self.assertEqual(message, self.plaintext) + + def test_prepend_zeroes(self): + cyphertext = b'\00\00' + self.cyphertext + with self.assertRaises(rsa.DecryptionError): + rsa.decrypt(cyphertext, self.private_key) + + def test_append_zeroes(self): + cyphertext = self.cyphertext + b'\00\00' + with self.assertRaises(rsa.DecryptionError): + rsa.decrypt(cyphertext, self.private_key) + + +class SignatureTest(unittest.TestCase): + def setUp(self): + (self.pub, self.priv) = rsa.newkeys(512) + + def test_sign_verify(self): + """Test happy flow of sign and verify""" + + message = b'je moeder' + signature = pkcs1.sign(message, self.priv, 'SHA-256') + + self.assertEqual('SHA-256', pkcs1.verify(message, signature, self.pub)) + + def test_find_signature_hash(self): + """Test happy flow of sign and find_signature_hash""" + + message = b'je moeder' + signature = pkcs1.sign(message, self.priv, 'SHA-256') + + self.assertEqual('SHA-256', pkcs1.find_signature_hash(signature, self.pub)) + + def test_alter_message(self): + """Altering the message should let the verification fail.""" + + signature = pkcs1.sign(b'je moeder', self.priv, 'SHA-256') + self.assertRaises(pkcs1.VerificationError, pkcs1.verify, + b'mijn moeder', signature, self.pub) + + def test_sign_different_key(self): + """Signing with another key should let the verification fail.""" + + (otherpub, _) = rsa.newkeys(512) + + message = b'je moeder' + signature = pkcs1.sign(message, self.priv, 'SHA-256') + self.assertRaises(pkcs1.VerificationError, pkcs1.verify, + message, signature, otherpub) + + def test_multiple_signings(self): + """Signing the same message twice should return the same signatures.""" + + message = struct.pack('>IIII', 0, 0, 0, 1) + signature1 = pkcs1.sign(message, self.priv, 'SHA-1') + signature2 = pkcs1.sign(message, self.priv, 'SHA-1') + + self.assertEqual(signature1, signature2) + + def test_split_hash_sign(self): + """Hashing and then signing should match with directly signing the message. """ + + message = b'je moeder' + msg_hash = pkcs1.compute_hash(message, 'SHA-256') + signature1 = pkcs1.sign_hash(msg_hash, self.priv, 'SHA-256') + + # Calculate the signature using the unified method + signature2 = pkcs1.sign(message, self.priv, 'SHA-256') + + self.assertEqual(signature1, signature2) + + def test_hash_sign_verify(self): + """Test happy flow of hash, sign, and verify""" + + message = b'je moeder' + msg_hash = pkcs1.compute_hash(message, 'SHA-224') + signature = pkcs1.sign_hash(msg_hash, self.priv, 'SHA-224') + + self.assertTrue(pkcs1.verify(message, signature, self.pub)) + + def test_prepend_zeroes(self): + """Prepending the signature with zeroes should be detected.""" + + message = b'je moeder' + signature = pkcs1.sign(message, self.priv, 'SHA-256') + signature = b'\00\00' + signature + with self.assertRaises(rsa.VerificationError): + pkcs1.verify(message, signature, self.pub) + + def test_apppend_zeroes(self): + """Apppending the signature with zeroes should be detected.""" + + message = b'je moeder' + signature = pkcs1.sign(message, self.priv, 'SHA-256') + signature = signature + b'\00\00' + with self.assertRaises(rsa.VerificationError): + pkcs1.verify(message, signature, self.pub) diff --git a/contrib/python/rsa/py2/tests/test_pkcs1_v2.py b/contrib/python/rsa/py2/tests/test_pkcs1_v2.py new file mode 100644 index 0000000000..1d8f0010de --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_pkcs1_v2.py @@ -0,0 +1,83 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests PKCS #1 version 2 functionality. + +Most of the mocked values come from the test vectors found at: +http://www.itomorrowmag.com/emc-plus/rsa-labs/standards-initiatives/pkcs-rsa-cryptography-standard.htm +""" + +import unittest + +from rsa import pkcs1_v2 + + +class MGFTest(unittest.TestCase): + def test_oaep_int_db_mask(self): + seed = ( + b'\xaa\xfd\x12\xf6\x59\xca\xe6\x34\x89\xb4\x79\xe5\x07\x6d\xde\xc2' + b'\xf0\x6c\xb5\x8f' + ) + db = ( + b'\xda\x39\xa3\xee\x5e\x6b\x4b\x0d\x32\x55\xbf\xef\x95\x60\x18\x90' + b'\xaf\xd8\x07\x09\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' + b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\xd4\x36\xe9\x95\x69' + b'\xfd\x32\xa7\xc8\xa0\x5b\xbc\x90\xd3\x2c\x49' + ) + masked_db = ( + b'\xdc\xd8\x7d\x5c\x68\xf1\xee\xa8\xf5\x52\x67\xc3\x1b\x2e\x8b\xb4' + b'\x25\x1f\x84\xd7\xe0\xb2\xc0\x46\x26\xf5\xaf\xf9\x3e\xdc\xfb\x25' + b'\xc9\xc2\xb3\xff\x8a\xe1\x0e\x83\x9a\x2d\xdb\x4c\xdc\xfe\x4f\xf4' + b'\x77\x28\xb4\xa1\xb7\xc1\x36\x2b\xaa\xd2\x9a\xb4\x8d\x28\x69\xd5' + b'\x02\x41\x21\x43\x58\x11\x59\x1b\xe3\x92\xf9\x82\xfb\x3e\x87\xd0' + b'\x95\xae\xb4\x04\x48\xdb\x97\x2f\x3a\xc1\x4f\x7b\xc2\x75\x19\x52' + b'\x81\xce\x32\xd2\xf1\xb7\x6d\x4d\x35\x3e\x2d' + ) + + # dbMask = MGF(seed, length(DB)) + db_mask = pkcs1_v2.mgf1(seed, length=len(db)) + expected_db_mask = ( + b'\x06\xe1\xde\xb2\x36\x9a\xa5\xa5\xc7\x07\xd8\x2c\x8e\x4e\x93\x24' + b'\x8a\xc7\x83\xde\xe0\xb2\xc0\x46\x26\xf5\xaf\xf9\x3e\xdc\xfb\x25' + b'\xc9\xc2\xb3\xff\x8a\xe1\x0e\x83\x9a\x2d\xdb\x4c\xdc\xfe\x4f\xf4' + b'\x77\x28\xb4\xa1\xb7\xc1\x36\x2b\xaa\xd2\x9a\xb4\x8d\x28\x69\xd5' + b'\x02\x41\x21\x43\x58\x11\x59\x1b\xe3\x92\xf9\x82\xfb\x3e\x87\xd0' + b'\x95\xae\xb4\x04\x48\xdb\x97\x2f\x3a\xc1\x4e\xaf\xf4\x9c\x8c\x3b' + b'\x7c\xfc\x95\x1a\x51\xec\xd1\xdd\xe6\x12\x64' + ) + + self.assertEqual(db_mask, expected_db_mask) + + # seedMask = MGF(maskedDB, length(seed)) + seed_mask = pkcs1_v2.mgf1(masked_db, length=len(seed)) + expected_seed_mask = ( + b'\x41\x87\x0b\x5a\xb0\x29\xe6\x57\xd9\x57\x50\xb5\x4c\x28\x3c\x08' + b'\x72\x5d\xbe\xa9' + ) + + self.assertEqual(seed_mask, expected_seed_mask) + + def test_invalid_hasher(self): + """Tests an invalid hasher generates an exception""" + with self.assertRaises(ValueError): + pkcs1_v2.mgf1(b'\x06\xe1\xde\xb2', length=8, hasher='SHA2') + + def test_invalid_length(self): + with self.assertRaises(OverflowError): + pkcs1_v2.mgf1(b'\x06\xe1\xde\xb2', length=2**50) diff --git a/contrib/python/rsa/py2/tests/test_prime.py b/contrib/python/rsa/py2/tests/test_prime.py new file mode 100644 index 0000000000..f3bda9b486 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_prime.py @@ -0,0 +1,110 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests prime functions.""" + +import unittest + +from rsa._compat import range +import rsa.prime +import rsa.randnum + + +class PrimeTest(unittest.TestCase): + def test_is_prime(self): + """Test some common primes.""" + + # Test some trivial numbers + self.assertFalse(rsa.prime.is_prime(-1)) + self.assertFalse(rsa.prime.is_prime(0)) + self.assertFalse(rsa.prime.is_prime(1)) + self.assertTrue(rsa.prime.is_prime(2)) + self.assertFalse(rsa.prime.is_prime(42)) + self.assertTrue(rsa.prime.is_prime(41)) + + # Test some slightly larger numbers + self.assertEqual( + [907, 911, 919, 929, 937, 941, 947, 953, 967, 971, 977, 983, 991, 997], + [x for x in range(901, 1000) if rsa.prime.is_prime(x)] + ) + + # Test around the 50th millionth known prime. + self.assertTrue(rsa.prime.is_prime(982451653)) + self.assertFalse(rsa.prime.is_prime(982451653 * 961748941)) + + def test_miller_rabin_primality_testing(self): + """Uses monkeypatching to ensure certain random numbers. + + This allows us to predict/control the code path. + """ + + randints = [] + + def fake_randint(maxvalue): + return randints.pop(0) + + orig_randint = rsa.randnum.randint + rsa.randnum.randint = fake_randint + try: + # 'n is composite' + randints.append(2630484832) # causes the 'n is composite' case with n=3784949785 + self.assertEqual(False, rsa.prime.miller_rabin_primality_testing(2787998641, 7)) + self.assertEqual([], randints) + + # 'Exit inner loop and continue with next witness' + randints.extend([ + 2119139098, # causes 'Exit inner loop and continue with next witness' + # the next witnesses for the above case: + 3051067716, 3603501763, 3230895847, 3687808133, 3760099987, 4026931495, 3022471882, + ]) + self.assertEqual(True, rsa.prime.miller_rabin_primality_testing(2211417913, + len(randints))) + self.assertEqual([], randints) + finally: + rsa.randnum.randint = orig_randint + + def test_mersenne_primes(self): + """Tests first known Mersenne primes. + + Mersenne primes are prime numbers that can be written in the form + `Mn = 2**n - 1` for some integer `n`. For the list of known Mersenne + primes, see: + https://en.wikipedia.org/wiki/Mersenne_prime#List_of_known_Mersenne_primes + """ + + # List of known Mersenne exponents. + known_mersenne_exponents = [ + 2, 3, 5, 7, 13, 17, 19, 31, 61, 89, 107, 127, 521, 607, 1279, + 2203, 2281, 4423, + ] + + # Test Mersenne primes. + for exp in known_mersenne_exponents: + self.assertTrue(rsa.prime.is_prime(2**exp - 1)) + + def test_get_primality_testing_rounds(self): + """Test round calculation for primality testing.""" + + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 63), 10) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 127), 10) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 255), 10) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 511), 7) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 767), 7) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 1023), 4) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 1279), 4) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 1535), 3) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 2047), 3) + self.assertEqual(rsa.prime.get_primality_testing_rounds(1 << 4095), 3) diff --git a/contrib/python/rsa/py2/tests/test_strings.py b/contrib/python/rsa/py2/tests/test_strings.py new file mode 100644 index 0000000000..28fa091a47 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_strings.py @@ -0,0 +1,42 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Tests string operations.""" + +from __future__ import absolute_import + +import unittest + +import rsa + +unicode_string = u"Euro=\u20ac ABCDEFGHIJKLMNOPQRSTUVWXYZ" + + +class StringTest(unittest.TestCase): + def setUp(self): + (self.pub, self.priv) = rsa.newkeys(384) + + def test_enc_dec(self): + message = unicode_string.encode('utf-8') + print("\tMessage: %s" % message) + + encrypted = rsa.encrypt(message, self.pub) + print("\tEncrypted: %s" % encrypted) + + decrypted = rsa.decrypt(encrypted, self.priv) + print("\tDecrypted: %s" % decrypted) + + self.assertEqual(message, decrypted) diff --git a/contrib/python/rsa/py2/tests/test_transform.py b/contrib/python/rsa/py2/tests/test_transform.py new file mode 100644 index 0000000000..fe0970c962 --- /dev/null +++ b/contrib/python/rsa/py2/tests/test_transform.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +# +# Copyright 2011 Sybren A. Stüvel <sybren@stuvel.eu> +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import unittest +from rsa.transform import int2bytes, bytes2int, _int2bytes + + +class Test_int2bytes(unittest.TestCase): + def test_accuracy(self): + self.assertEqual(int2bytes(123456789), b'\x07[\xcd\x15') + self.assertEqual(_int2bytes(123456789), b'\x07[\xcd\x15') + + def test_codec_identity(self): + self.assertEqual(bytes2int(int2bytes(123456789, 128)), 123456789) + self.assertEqual(bytes2int(_int2bytes(123456789, 128)), 123456789) + + def test_chunk_size(self): + self.assertEqual(int2bytes(123456789, 6), b'\x00\x00\x07[\xcd\x15') + self.assertEqual(int2bytes(123456789, 7), + b'\x00\x00\x00\x07[\xcd\x15') + + self.assertEqual(_int2bytes(123456789, 6), + b'\x00\x00\x07[\xcd\x15') + self.assertEqual(_int2bytes(123456789, 7), + b'\x00\x00\x00\x07[\xcd\x15') + + def test_zero(self): + self.assertEqual(int2bytes(0, 4), b'\x00' * 4) + self.assertEqual(int2bytes(0, 7), b'\x00' * 7) + self.assertEqual(int2bytes(0), b'\x00') + + self.assertEqual(_int2bytes(0, 4), b'\x00' * 4) + self.assertEqual(_int2bytes(0, 7), b'\x00' * 7) + self.assertEqual(_int2bytes(0), b'\x00') + + def test_correctness_against_base_implementation(self): + # Slow test. + values = [ + 1 << 512, + 1 << 8192, + 1 << 77, + ] + for value in values: + self.assertEqual(int2bytes(value), _int2bytes(value), + "Boom %d" % value) + self.assertEqual(bytes2int(int2bytes(value)), + value, + "Boom %d" % value) + self.assertEqual(bytes2int(_int2bytes(value)), + value, + "Boom %d" % value) + + def test_raises_OverflowError_when_chunk_size_is_insufficient(self): + self.assertRaises(OverflowError, int2bytes, 123456789, 3) + self.assertRaises(OverflowError, int2bytes, 299999999999, 4) + + self.assertRaises(OverflowError, _int2bytes, 123456789, 3) + self.assertRaises(OverflowError, _int2bytes, 299999999999, 4) + + def test_raises_ValueError_when_negative_integer(self): + self.assertRaises(ValueError, int2bytes, -1) + self.assertRaises(ValueError, _int2bytes, -1) + + def test_raises_TypeError_when_not_integer(self): + self.assertRaises(TypeError, int2bytes, None) + self.assertRaises(TypeError, _int2bytes, None) diff --git a/contrib/python/rsa/py2/tests/ya.make b/contrib/python/rsa/py2/tests/ya.make new file mode 100644 index 0000000000..05640300fb --- /dev/null +++ b/contrib/python/rsa/py2/tests/ya.make @@ -0,0 +1,30 @@ +PY2TEST() + +PEERDIR( + contrib/python/rsa + contrib/python/mock +) + +NO_LINT() + +TEST_SRCS( + test_cli.py + test_common.py + test_compat.py + test_integers.py + test_key.py + test_load_save_keys.py + test_parallel.py + test_pem.py + test_pkcs1.py + test_pkcs1_v2.py + test_prime.py + test_strings.py + test_transform.py +) + +DATA ( + arcadia/contrib/python/rsa/py2/tests +) + +END() diff --git a/contrib/python/rsa/py2/ya.make b/contrib/python/rsa/py2/ya.make new file mode 100644 index 0000000000..f196afec21 --- /dev/null +++ b/contrib/python/rsa/py2/ya.make @@ -0,0 +1,46 @@ +# Generated by devtools/yamaker (pypi). + +PY2_LIBRARY() + +VERSION(4.5) + +LICENSE(Apache-2.0) + +PEERDIR( + contrib/python/pyasn1 +) + +NO_LINT() + +PY_SRCS( + TOP_LEVEL + rsa/__init__.py + rsa/_compat.py + rsa/asn1.py + rsa/cli.py + rsa/common.py + rsa/core.py + rsa/key.py + rsa/machine_size.py + rsa/parallel.py + rsa/pem.py + rsa/pkcs1.py + rsa/pkcs1_v2.py + rsa/prime.py + rsa/randnum.py + rsa/transform.py + rsa/util.py +) + +RESOURCE_FILES( + PREFIX contrib/python/rsa/py2/ + .dist-info/METADATA + .dist-info/entry_points.txt + .dist-info/top_level.txt +) + +END() + +RECURSE_FOR_TESTS( + tests +) |