summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorKali Kaneko <kali@leap.se>2017-06-26 15:06:41 +0200
committerKali Kaneko <kali@leap.se>2017-06-27 10:33:48 +0200
commit59504c7ddf7aab71614d691e705d386f58b5100d (patch)
treeea6afce37890a23cd088eac50587c3293035e78c /src
parente82b8143d3e9b2f62650e06798eee262885036c2 (diff)
[pkg] vendor pgpy 0.4.1
Diffstat (limited to 'src')
-rw-r--r--src/leap/mx/mail_receiver.py5
-rw-r--r--src/leap/mx/tests/test_mail_receiver.py2
-rw-r--r--src/leap/mx/vendor/__init__.py0
-rw-r--r--src/leap/mx/vendor/pgpy/README30
-rw-r--r--src/leap/mx/vendor/pgpy/__init__.py21
-rw-r--r--src/leap/mx/vendor/pgpy/_author.py18
-rw-r--r--src/leap/mx/vendor/pgpy/_curves.py54
-rw-r--r--src/leap/mx/vendor/pgpy/constants.py484
-rw-r--r--src/leap/mx/vendor/pgpy/decorators.py131
-rw-r--r--src/leap/mx/vendor/pgpy/errors.py39
-rw-r--r--src/leap/mx/vendor/pgpy/memoryview.py128
-rw-r--r--src/leap/mx/vendor/pgpy/packet/__init__.py11
-rw-r--r--src/leap/mx/vendor/pgpy/packet/fields.py1514
-rw-r--r--src/leap/mx/vendor/pgpy/packet/packets.py1605
-rw-r--r--src/leap/mx/vendor/pgpy/packet/subpackets/__init__.py7
-rw-r--r--src/leap/mx/vendor/pgpy/packet/subpackets/signature.py887
-rw-r--r--src/leap/mx/vendor/pgpy/packet/subpackets/types.py136
-rw-r--r--src/leap/mx/vendor/pgpy/packet/subpackets/userattribute.py109
-rw-r--r--src/leap/mx/vendor/pgpy/packet/types.py288
-rw-r--r--src/leap/mx/vendor/pgpy/pgp.py2527
-rw-r--r--src/leap/mx/vendor/pgpy/symenc.py58
-rw-r--r--src/leap/mx/vendor/pgpy/types.py729
22 files changed, 8780 insertions, 3 deletions
diff --git a/src/leap/mx/mail_receiver.py b/src/leap/mx/mail_receiver.py
index e376554..276ae13 100644
--- a/src/leap/mx/mail_receiver.py
+++ b/src/leap/mx/mail_receiver.py
@@ -42,8 +42,6 @@ import email.utils
from datetime import datetime, timedelta
from email import message_from_string
-from pgpy import PGPKey, PGPMessage
-from pgpy.errors import PGPEncryptionError
from twisted.application.service import Service, IService
from twisted.internet import inotify, defer, task, reactor
@@ -59,6 +57,9 @@ from leap.soledad.common.document import ServerDocument
from leap.mx.bounce import bounce_message
from leap.mx.bounce import InvalidReturnPathError
+from leap.mx.vendor.pgpy import PGPKey, PGPMessage
+from leap.mx.vendor.pgpy.errors import PGPEncryptionError
+
class MailReceiver(Service):
"""
diff --git a/src/leap/mx/tests/test_mail_receiver.py b/src/leap/mx/tests/test_mail_receiver.py
index abe5a71..0d03806 100644
--- a/src/leap/mx/tests/test_mail_receiver.py
+++ b/src/leap/mx/tests/test_mail_receiver.py
@@ -26,11 +26,11 @@ import shutil
import tempfile
from email.message import Message
-from pgpy import PGPKey, PGPMessage
from twisted.internet import defer, reactor
from twisted.trial import unittest
from leap.mx.mail_receiver import MailReceiver
+from leap.mx.vendor.pgpy import PGPKey, PGPMessage
BOUNCE_ADDRESS = "bounce@leap.se"
diff --git a/src/leap/mx/vendor/__init__.py b/src/leap/mx/vendor/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/src/leap/mx/vendor/__init__.py
diff --git a/src/leap/mx/vendor/pgpy/README b/src/leap/mx/vendor/pgpy/README
new file mode 100644
index 0000000..275a496
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/README
@@ -0,0 +1,30 @@
+This submodule is a copy of PGPy 0.4.1
+We're vendoring it here because a package for python2 is not readily available.
+If you're thinking about packaging it, you probably could port leap.mx to py3 instead.
+
+pgpy is Copyright (c) 2014 Michael Greene - All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+ list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+ this list of conditions and the following disclaimer in the documentation
+ and/or other materials provided with the distribution.
+
+* Neither the name of the {organization} nor the names of its
+ contributors may be used to endorse or promote products derived from
+ this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/src/leap/mx/vendor/pgpy/__init__.py b/src/leap/mx/vendor/pgpy/__init__.py
new file mode 100644
index 0000000..b80740d
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/__init__.py
@@ -0,0 +1,21 @@
+""" PGPy :: Pretty Good Privacy for Python
+"""
+from ._author import *
+
+from .pgp import PGPKey
+from .pgp import PGPKeyring
+from .pgp import PGPMessage
+from .pgp import PGPSignature
+from .pgp import PGPUID
+
+__all__ = ['__author__',
+ '__copyright__',
+ '__license__',
+ '__version__',
+ 'constants',
+ 'errors',
+ 'PGPKey',
+ 'PGPKeyring',
+ 'PGPMessage',
+ 'PGPSignature',
+ 'PGPUID', ]
diff --git a/src/leap/mx/vendor/pgpy/_author.py b/src/leap/mx/vendor/pgpy/_author.py
new file mode 100644
index 0000000..ae17b45
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/_author.py
@@ -0,0 +1,18 @@
+"""_author.py
+
+Canonical location for authorship information
+__version__ is a PEP-386 compliant version string,
+making use of distutils.version.LooseVersion
+"""
+
+from distutils.version import LooseVersion
+
+__all__ = ['__author__',
+ '__copyright__',
+ '__license__',
+ '__version__']
+
+__author__ = "Michael Greene"
+__copyright__ = "Copyright (c) 2014 Michael Greene"
+__license__ = "BSD"
+__version__ = str(LooseVersion("0.4.1"))
diff --git a/src/leap/mx/vendor/pgpy/_curves.py b/src/leap/mx/vendor/pgpy/_curves.py
new file mode 100644
index 0000000..9503075
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/_curves.py
@@ -0,0 +1,54 @@
+""" _curves.py
+specify some additional curves that OpenSSL provides but cryptography doesn't explicitly expose
+"""
+
+from cryptography import utils
+
+from cryptography.hazmat.primitives.asymmetric import ec
+
+from cryptography.hazmat.bindings.openssl.binding import Binding
+
+__all__ = tuple()
+
+# TODO: investigate defining additional curves using EC_GROUP_new_curve
+# https://wiki.openssl.org/index.php/Elliptic_Curve_Cryptography#Defining_Curves
+
+
+def _openssl_get_supported_curves():
+ if hasattr(_openssl_get_supported_curves, '_curves'):
+ return _openssl_get_supported_curves._curves
+
+ # use cryptography's cffi bindings to get an array of curve names
+ b = Binding()
+ cn = b.lib.EC_get_builtin_curves(b.ffi.NULL, 0)
+ cs = b.ffi.new('EC_builtin_curve[]', cn)
+ b.lib.EC_get_builtin_curves(cs, cn)
+
+ # store the result so we don't have to do all of this every time
+ curves = { b.ffi.string(b.lib.OBJ_nid2sn(c.nid)).decode('utf-8') for c in cs }
+ _openssl_get_supported_curves._curves = curves
+ return curves
+
+
+@utils.register_interface(ec.EllipticCurve)
+class BrainpoolP256R1(object):
+ name = 'brainpoolP256r1'
+ key_size = 256
+
+
+@utils.register_interface(ec.EllipticCurve)
+class BrainpoolP384R1(object):
+ name = 'brainpoolP384r1'
+ key_size = 384
+
+
+@utils.register_interface(ec.EllipticCurve)
+class BrainpoolP512R1(object):
+ name = 'brainpoolP512r1'
+ key_size = 512
+
+
+# add these curves to the _CURVE_TYPES list
+for curve in [BrainpoolP256R1, BrainpoolP384R1, BrainpoolP512R1]:
+ if curve.name not in ec._CURVE_TYPES and curve.name in _openssl_get_supported_curves():
+ ec._CURVE_TYPES[curve.name] = curve
diff --git a/src/leap/mx/vendor/pgpy/constants.py b/src/leap/mx/vendor/pgpy/constants.py
new file mode 100644
index 0000000..271b0d0
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/constants.py
@@ -0,0 +1,484 @@
+""" constants.py
+"""
+import bz2
+import hashlib
+import imghdr
+import os
+import time
+import zlib
+
+from collections import namedtuple
+from enum import Enum
+from enum import IntEnum
+from pyasn1.type.univ import ObjectIdentifier
+
+import six
+
+from cryptography.hazmat.backends import openssl
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.ciphers import algorithms
+
+from .decorators import classproperty
+from .types import FlagEnum
+from ._curves import BrainpoolP256R1, BrainpoolP384R1, BrainpoolP512R1
+
+__all__ = ['Backend',
+ 'EllipticCurveOID',
+ 'PacketTag',
+ 'SymmetricKeyAlgorithm',
+ 'PubKeyAlgorithm',
+ 'CompressionAlgorithm',
+ 'HashAlgorithm',
+ 'RevocationReason',
+ 'ImageEncoding',
+ 'SignatureType',
+ 'KeyServerPreferences',
+ 'String2KeyType',
+ 'TrustLevel',
+ 'KeyFlags',
+ 'Features',
+ 'RevocationKeyClass',
+ 'NotationDataFlags',
+ 'TrustFlags']
+
+
+# this is 50 KiB
+_hashtunedata = bytearray([10, 11, 12, 13, 14, 15, 16, 17] * 128 * 50)
+
+
+class Backend(Enum):
+ OpenSSL = openssl.backend
+
+
+class EllipticCurveOID(Enum):
+ # these are specified as:
+ # id = (oid, curve)
+ Invalid = ('', )
+ #: DJB's fast elliptic curve
+ #:
+ #: .. warning::
+ #: This curve is not currently usable by PGPy
+ Curve25519 = ('1.3.6.1.4.1.3029.1.5.1', )
+ #: Twisted Edwards variant of Curve25519
+ #:
+ #: .. warning::
+ #: This curve is not currently usable by PGPy
+ Ed25519 = ('1.3.6.1.4.1.11591.15.1', )
+ #: NIST P-256, also known as SECG curve secp256r1
+ NIST_P256 = ('1.2.840.10045.3.1.7', ec.SECP256R1)
+ #: NIST P-384, also known as SECG curve secp384r1
+ NIST_P384 = ('1.3.132.0.34', ec.SECP384R1)
+ #: NIST P-521, also known as SECG curve secp521r1
+ NIST_P521 = ('1.3.132.0.35', ec.SECP521R1)
+ #: Brainpool Standard Curve, 256-bit
+ #:
+ #: .. note::
+ #: Requires OpenSSL >= 1.0.2
+ Brainpool_P256 = ('1.3.36.3.3.2.8.1.1.7', BrainpoolP256R1)
+ #: Brainpool Standard Curve, 384-bit
+ #:
+ #: .. note::
+ #: Requires OpenSSL >= 1.0.2
+ Brainpool_P384 = ('1.3.36.3.3.2.8.1.1.11', BrainpoolP384R1)
+ #: Brainpool Standard Curve, 512-bit
+ #:
+ #: .. note::
+ #: Requires OpenSSL >= 1.0.2
+ Brainpool_P512 = ('1.3.36.3.3.2.8.1.1.13', BrainpoolP512R1)
+ #: SECG curve secp256k1
+ SECP256K1 = ('1.3.132.0.10', ec.SECP256K1)
+
+ def __new__(cls, oid, curve=None):
+ # preprocessing stage for enum members:
+ # - set enum_member.value to ObjectIdentifier(oid)
+ # - if curve is not None and curve.name is in ec._CURVE_TYPES, set enum_member.curve to curve
+ # - otherwise, set enum_member.curve to None
+ obj = object.__new__(cls)
+ obj._value_ = ObjectIdentifier(oid)
+ obj.curve = None
+
+ if curve is not None and curve.name in ec._CURVE_TYPES:
+ obj.curve = curve
+
+ return obj
+
+ @property
+ def can_gen(self):
+ return self.curve is not None
+
+ @property
+ def key_size(self):
+ if self.curve is not None:
+ return self.curve.key_size
+
+ @property
+ def kdf_halg(self):
+ # return the hash algorithm to specify in the KDF fields when generating a key
+ algs = {256: HashAlgorithm.SHA256,
+ 384: HashAlgorithm.SHA384,
+ 512: HashAlgorithm.SHA512,
+ 521: HashAlgorithm.SHA512}
+
+ return algs.get(self.key_size, None)
+
+ @property
+ def kek_alg(self):
+ # return the AES algorithm to specify in the KDF fields when generating a key
+ algs = {256: SymmetricKeyAlgorithm.AES128,
+ 384: SymmetricKeyAlgorithm.AES192,
+ 512: SymmetricKeyAlgorithm.AES256,
+ 521: SymmetricKeyAlgorithm.AES256}
+
+ return algs.get(self.key_size, None)
+
+
+class PacketTag(IntEnum):
+ Invalid = 0
+ PublicKeyEncryptedSessionKey = 1
+ Signature = 2
+ SymmetricKeyEncryptedSessionKey = 3
+ OnePassSignature = 4
+ SecretKey = 5
+ PublicKey = 6
+ SecretSubKey = 7
+ CompressedData = 8
+ SymmetricallyEncryptedData = 9
+ Marker = 10
+ LiteralData = 11
+ Trust = 12
+ UserID = 13
+ PublicSubKey = 14
+ UserAttribute = 17
+ SymmetricallyEncryptedIntegrityProtectedData = 18
+ ModificationDetectionCode = 19
+
+
+class SymmetricKeyAlgorithm(IntEnum):
+ """Supported symmetric key algorithms."""
+ Plaintext = 0x00
+ #: .. warning::
+ #: IDEA is insecure. PGPy only allows it to be used for decryption, not encryption!
+ IDEA = 0x01
+ #: Triple-DES with 168-bit key derived from 192
+ TripleDES = 0x02
+ #: CAST5 (or CAST-128) with 128-bit key
+ CAST5 = 0x03
+ #: Blowfish with 128-bit key and 16 rounds
+ Blowfish = 0x04
+ #: AES with 128-bit key
+ AES128 = 0x07
+ #: AES with 192-bit key
+ AES192 = 0x08
+ #: AES with 256-bit key
+ AES256 = 0x09
+ # Twofish with 256-bit key - not currently supported
+ Twofish256 = 0x0A
+ #: Camellia with 128-bit key
+ Camellia128 = 0x0B
+ #: Camellia with 192-bit key
+ Camellia192 = 0x0C
+ #: Camellia with 256-bit key
+ Camellia256 = 0x0D
+
+ @property
+ def cipher(self):
+ bs = {SymmetricKeyAlgorithm.IDEA: algorithms.IDEA,
+ SymmetricKeyAlgorithm.TripleDES: algorithms.TripleDES,
+ SymmetricKeyAlgorithm.CAST5: algorithms.CAST5,
+ SymmetricKeyAlgorithm.Blowfish: algorithms.Blowfish,
+ SymmetricKeyAlgorithm.AES128: algorithms.AES,
+ SymmetricKeyAlgorithm.AES192: algorithms.AES,
+ SymmetricKeyAlgorithm.AES256: algorithms.AES,
+ SymmetricKeyAlgorithm.Twofish256: namedtuple('Twofish256', ['block_size'])(block_size=128),
+ SymmetricKeyAlgorithm.Camellia128: algorithms.Camellia,
+ SymmetricKeyAlgorithm.Camellia192: algorithms.Camellia,
+ SymmetricKeyAlgorithm.Camellia256: algorithms.Camellia}
+
+ if self in bs:
+ return bs[self]
+
+ raise NotImplementedError(repr(self))
+
+ @property
+ def is_insecure(self):
+ insecure_ciphers = {SymmetricKeyAlgorithm.IDEA}
+ return self in insecure_ciphers
+
+ @property
+ def block_size(self):
+ return self.cipher.block_size
+
+ @property
+ def key_size(self):
+ ks = {SymmetricKeyAlgorithm.IDEA: 128,
+ SymmetricKeyAlgorithm.TripleDES: 192,
+ SymmetricKeyAlgorithm.CAST5: 128,
+ SymmetricKeyAlgorithm.Blowfish: 128,
+ SymmetricKeyAlgorithm.AES128: 128,
+ SymmetricKeyAlgorithm.AES192: 192,
+ SymmetricKeyAlgorithm.AES256: 256,
+ SymmetricKeyAlgorithm.Twofish256: 256,
+ SymmetricKeyAlgorithm.Camellia128: 128,
+ SymmetricKeyAlgorithm.Camellia192: 192,
+ SymmetricKeyAlgorithm.Camellia256: 256}
+
+ if self in ks:
+ return ks[self]
+
+ raise NotImplementedError(repr(self))
+
+ def gen_iv(self):
+ return os.urandom(self.block_size // 8)
+
+ def gen_key(self):
+ return os.urandom(self.key_size // 8)
+
+
+class PubKeyAlgorithm(IntEnum):
+ Invalid = 0x00
+ #: Signifies that a key is an RSA key.
+ RSAEncryptOrSign = 0x01
+ RSAEncrypt = 0x02 # deprecated
+ RSASign = 0x03 # deprecated
+ #: Signifies that a key is an ElGamal key.
+ ElGamal = 0x10
+ #: Signifies that a key is a DSA key.
+ DSA = 0x11
+ #: Signifies that a key is an ECDH key.
+ ECDH = 0x12
+ #: Signifies that a key is an ECDSA key.
+ ECDSA = 0x13
+ FormerlyElGamalEncryptOrSign = 0x14 # deprecated - do not generate
+ # DiffieHellman = 0x15 # X9.42
+
+ @property
+ def can_gen(self):
+ return self in {PubKeyAlgorithm.RSAEncryptOrSign,
+ PubKeyAlgorithm.DSA,
+ PubKeyAlgorithm.ECDSA,
+ PubKeyAlgorithm.ECDH}
+
+ @property
+ def can_encrypt(self): # pragma: no cover
+ return self in {PubKeyAlgorithm.RSAEncryptOrSign, PubKeyAlgorithm.ElGamal, PubKeyAlgorithm.ECDH}
+
+ @property
+ def can_sign(self):
+ return self in {PubKeyAlgorithm.RSAEncryptOrSign, PubKeyAlgorithm.DSA, PubKeyAlgorithm.ECDSA}
+
+ @property
+ def deprecated(self):
+ return self in {PubKeyAlgorithm.RSAEncrypt,
+ PubKeyAlgorithm.RSASign,
+ PubKeyAlgorithm.FormerlyElGamalEncryptOrSign}
+
+
+class CompressionAlgorithm(IntEnum):
+ #: No compression
+ Uncompressed = 0x00
+ #: ZIP DEFLATE
+ ZIP = 0x01
+ #: ZIP DEFLATE with zlib headers
+ ZLIB = 0x02
+ #: Bzip2
+ BZ2 = 0x03
+
+ def compress(self, data):
+ if self is CompressionAlgorithm.Uncompressed:
+ return data
+
+ if self is CompressionAlgorithm.ZIP:
+ return zlib.compress(data)[2:-4]
+
+ if self is CompressionAlgorithm.ZLIB:
+ return zlib.compress(data)
+
+ if self is CompressionAlgorithm.BZ2:
+ return bz2.compress(data)
+
+ raise NotImplementedError(self)
+
+ def decompress(self, data):
+ if six.PY2:
+ data = bytes(data)
+
+ if self is CompressionAlgorithm.Uncompressed:
+ return data
+
+ if self is CompressionAlgorithm.ZIP:
+ return zlib.decompress(data, -15)
+
+ if self is CompressionAlgorithm.ZLIB:
+ return zlib.decompress(data)
+
+ if self is CompressionAlgorithm.BZ2:
+ return bz2.decompress(data)
+
+ raise NotImplementedError(self)
+
+
+class HashAlgorithm(IntEnum):
+ Invalid = 0x00
+ MD5 = 0x01
+ SHA1 = 0x02
+ RIPEMD160 = 0x03
+ _reserved_1 = 0x04
+ _reserved_2 = 0x05
+ _reserved_3 = 0x06
+ _reserved_4 = 0x07
+ SHA256 = 0x08
+ SHA384 = 0x09
+ SHA512 = 0x0A
+ SHA224 = 0x0B
+
+ def __init__(self, *args):
+ super(self.__class__, self).__init__()
+ self._tuned_count = 0
+
+ @property
+ def hasher(self):
+ return hashlib.new(self.name)
+
+ @property
+ def digest_size(self):
+ return self.hasher.digest_size
+
+ @property
+ def tuned_count(self):
+ if self._tuned_count == 0:
+ self.tune_count()
+
+ return self._tuned_count
+
+ def tune_count(self):
+ start = end = 0
+ htd = _hashtunedata[:]
+
+ while start == end:
+ # potentially do this multiple times in case the resolution of time.time is low enough that
+ # hashing 100 KiB isn't enough time to produce a measurable difference
+ # (e.g. if the timer for time.time doesn't have enough precision)
+ htd = htd + htd
+ h = self.hasher
+
+ start = time.time()
+ h.update(htd)
+ end = time.time()
+
+ # now calculate how many bytes need to be hashed to reach our expected time period
+ # GnuPG tunes for about 100ms, so we'll do that as well
+ _TIME = 0.100
+ ct = int(len(htd) * (_TIME / (end - start)))
+ c1 = ((ct >> (ct.bit_length() - 5)) - 16)
+ c2 = (ct.bit_length() - 11)
+ c = ((c2 << 4) + c1)
+
+ # constrain self._tuned_count to be between 0 and 255
+ self._tuned_count = max(min(c, 255), 0)
+
+
+class RevocationReason(IntEnum):
+ #: No reason was specified. This is the default reason.
+ NotSpecified = 0x00
+ #: The key was superseded by a new key. Only meaningful when revoking a key.
+ Superseded = 0x01
+ #: Key material has been compromised. Only meaningful when revoking a key.
+ Compromised = 0x02
+ #: Key is retired and no longer used. Only meaningful when revoking a key.
+ Retired = 0x03
+ #: User ID information is no longer valid. Only meaningful when revoking a certification of a user id.
+ UserID = 0x20
+
+
+class ImageEncoding(IntEnum):
+ Unknown = 0x00
+ JPEG = 0x01
+
+ @classmethod
+ def encodingof(cls, imagebytes):
+ type = imghdr.what(None, h=imagebytes)
+ if type == 'jpeg':
+ return ImageEncoding.JPEG
+ return ImageEncoding.Unknown # pragma: no cover
+
+
+class SignatureType(IntEnum):
+ BinaryDocument = 0x00
+ CanonicalDocument = 0x01
+ Standalone = 0x02
+ Generic_Cert = 0x10
+ Persona_Cert = 0x11
+ Casual_Cert = 0x12
+ Positive_Cert = 0x13
+ Subkey_Binding = 0x18
+ PrimaryKey_Binding = 0x19
+ DirectlyOnKey = 0x1F
+ KeyRevocation = 0x20
+ SubkeyRevocation = 0x28
+ CertRevocation = 0x30
+ Timestamp = 0x40
+ ThirdParty_Confirmation = 0x50
+
+
+class KeyServerPreferences(IntEnum):
+ Unknown = 0x00
+ NoModify = 0x80
+
+
+class String2KeyType(IntEnum):
+ Simple = 0
+ Salted = 1
+ Reserved = 2
+ Iterated = 3
+
+
+class TrustLevel(IntEnum):
+ Unknown = 0
+ Expired = 1
+ Undefined = 2
+ Never = 3
+ Marginal = 4
+ Fully = 5
+ Ultimate = 6
+
+
+class KeyFlags(FlagEnum):
+ #: Signifies that a key may be used to certify keys and user ids. Primary keys always have this, even if it is not specified.
+ Certify = 0x01
+ #: Signifies that a key may be used to sign messages and documents.
+ Sign = 0x02
+ #: Signifies that a key may be used to encrypt messages.
+ EncryptCommunications = 0x04
+ #: Signifies that a key may be used to encrypt storage. Currently equivalent to :py:obj:`~pgpy.constants.EncryptCommunications`.
+ EncryptStorage = 0x08
+ #: Signifies that the private component of a given key may have been split by a secret-sharing mechanism. Split
+ #: keys are not currently supported by PGPy.
+ Split = 0x10
+ #: Signifies that a key may be used for authentication.
+ Authentication = 0x20
+ #: Signifies that the private component of a key may be in the possession of more than one person.
+ MultiPerson = 0x80
+
+
+class Features(FlagEnum):
+ ModificationDetection = 0x01
+
+ @classproperty
+ def pgpy_features(cls):
+ return Features.ModificationDetection
+
+
+class RevocationKeyClass(FlagEnum):
+ Sensitive = 0x40
+ Normal = 0x80
+
+
+class NotationDataFlags(FlagEnum):
+ HumanReadable = 0x80
+
+
+class TrustFlags(FlagEnum):
+ Revoked = 0x20
+ SubRevoked = 0x40
+ Disabled = 0x80
+ PendingCheck = 0x100
diff --git a/src/leap/mx/vendor/pgpy/decorators.py b/src/leap/mx/vendor/pgpy/decorators.py
new file mode 100644
index 0000000..d2b9926
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/decorators.py
@@ -0,0 +1,131 @@
+""" decorators.py
+"""
+import contextlib
+import functools
+import six
+import warnings
+
+try:
+ from singledispatch import singledispatch
+
+except ImportError: # pragma: no cover
+ from functools import singledispatch
+
+
+from .errors import PGPError
+
+__all__ = ['classproperty',
+ 'sdmethod',
+ 'sdproperty',
+ 'KeyAction']
+
+
+def classproperty(fget):
+ class ClassProperty(object):
+ def __init__(self, fget):
+ self.fget = fget
+ self.__doc__ = fget.__doc__
+
+ def __get__(self, cls, owner):
+ return self.fget(owner)
+
+ def __set__(self, obj, value): # pragma: no cover
+ raise AttributeError("Read-only attribute")
+
+ def __delete__(self, obj): # pragma: no cover
+ raise AttributeError("Read-only attribute")
+
+ return ClassProperty(fget)
+
+
+def sdmethod(meth):
+ """
+ This is a hack to monkey patch sdproperty to work as expected with instance methods.
+ """
+ sd = singledispatch(meth)
+
+ def wrapper(obj, *args, **kwargs):
+ return sd.dispatch(args[0].__class__)(obj, *args, **kwargs)
+
+ wrapper.register = sd.register
+ wrapper.dispatch = sd.dispatch
+ wrapper.registry = sd.registry
+ wrapper._clear_cache = sd._clear_cache
+ functools.update_wrapper(wrapper, meth)
+ return wrapper
+
+
+def sdproperty(fget):
+ def defset(obj, val): # pragma: no cover
+ raise TypeError(str(val.__class__))
+
+ class SDProperty(property):
+ def register(self, cls=None, fset=None):
+ return self.fset.register(cls, fset)
+
+ def setter(self, fset):
+ self.register(object, fset)
+ return type(self)(self.fget, self.fset, self.fdel, self.__doc__)
+
+ return SDProperty(fget, sdmethod(defset))
+
+
+class KeyAction(object):
+ def __init__(self, *usage, **conditions):
+ super(KeyAction, self).__init__()
+ self.flags = set(usage)
+ self.conditions = conditions
+
+ @contextlib.contextmanager
+ def usage(self, key, user):
+ def _preiter(first, iterable):
+ yield first
+ for item in iterable:
+ yield item
+
+ em = {}
+ em['keyid'] = key.fingerprint.keyid
+ em['flags'] = ', '.join(flag.name for flag in self.flags)
+
+ if len(self.flags):
+ for _key in _preiter(key, key.subkeys.values()):
+ if self.flags & set(_key._get_key_flags(user)):
+ break
+
+ else: # pragma: no cover
+ raise PGPError("Key {keyid:s} does not have the required usage flag {flags:s}".format(**em))
+
+ else:
+ _key = key
+
+ if _key is not key:
+ em['subkeyid'] = _key.fingerprint.keyid
+ warnings.warn("Key {keyid:s} does not have the required usage flag {flags:s}; using subkey {subkeyid:s}"
+ "".format(**em), stacklevel=4)
+
+ yield _key
+
+ def check_attributes(self, key):
+ for attr, expected in self.conditions.items():
+ if getattr(key, attr) != expected:
+ raise PGPError("Expected: {attr:s} == {eval:s}. Got: {got:s}"
+ "".format(attr=attr, eval=str(expected), got=str(getattr(key, attr))))
+
+ def __call__(self, action):
+ # @functools.wraps(action)
+ @six.wraps(action)
+ def _action(key, *args, **kwargs):
+ if key._key is None:
+ raise PGPError("No key!")
+
+ # if a key is in the process of being created, it needs to be allowed to certify its own user id
+ if len(key._uids) == 0 and key.is_primary and action is not key.certify.__wrapped__:
+ raise PGPError("Key is not complete - please add a User ID!")
+
+ with self.usage(key, kwargs.get('user', None)) as _key:
+ self.check_attributes(key)
+
+ # do the thing
+ return action(_key, *args, **kwargs)
+
+ return _action
diff --git a/src/leap/mx/vendor/pgpy/errors.py b/src/leap/mx/vendor/pgpy/errors.py
new file mode 100644
index 0000000..6af0612
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/errors.py
@@ -0,0 +1,39 @@
+""" errors.py
+"""
+
+__all__ = ('PGPError',
+ 'PGPEncryptionError',
+ 'PGPDecryptionError',
+ 'PGPOpenSSLCipherNotSupported',
+ 'PGPInsecureCipher',
+ 'WontImplementError',)
+
+
+class PGPError(Exception):
+ """Raised as a general error in PGPy"""
+ pass
+
+
+class PGPEncryptionError(Exception):
+ """Raised when encryption fails"""
+ pass
+
+
+class PGPDecryptionError(Exception):
+ """Raised when decryption fails"""
+ pass
+
+
+class PGPOpenSSLCipherNotSupported(Exception):
+ """Raised when OpenSSL does not support the requested cipher"""
+ pass
+
+
+class PGPInsecureCipher(Exception):
+ """Raised when a cipher known to be insecure is attempted to be used to encrypt data"""
+ pass
+
+
+class WontImplementError(NotImplementedError):
+ """Raised when something that is not implemented, will not be implemented"""
+ pass
diff --git a/src/leap/mx/vendor/pgpy/memoryview.py b/src/leap/mx/vendor/pgpy/memoryview.py
new file mode 100644
index 0000000..35f3801
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/memoryview.py
@@ -0,0 +1,128 @@
+""" util.py
+"""
+import six
+
+__all__ = ('memoryview', )
+
+memoryview = memoryview
+
+if six.PY2:
+ # because Python2's memoryview can't be released directly, nor can it be used as a context manager
+ # this wrapper object should hopefully make the behavior more uniform to python 3's
+ import __builtin__
+ import functools
+
+ # this decorator will raise a ValueError if the wrapped memoryview object has been "released"
+ def notreleased(meth):
+ @functools.wraps(meth)
+ def _inner(self, *args, **kwargs):
+ if self._mem is None:
+ raise ValueError("operation forbidden on released memoryview object")
+ return meth(self, *args, **kwargs)
+
+ return _inner
+
+ class memoryview(object): # flake8: noqa
+ @property
+ @notreleased
+ def obj(self):
+ """The underlying object of the memoryview."""
+ return self._obj
+
+ @property
+ @notreleased
+ def nbytes(self):
+ # nbytes == product(shape) * itemsize == len(m.tobytes())
+ nb = 1
+ for dim in self.shape:
+ nb *= dim
+ return nb * self.itemsize
+
+ # TODO: c_contiguous -> (self.ndim == 0 or ???)
+ # TODO: f_contiguous -> (self.ndim == 0 or ???)
+ # TODO: contiguous -> return self.c_contiguous or self.f_contiguous
+
+ def __new__(cls, obj, parent=None):
+ memview = object.__new__(cls)
+ memview._obj = obj if parent is None else parent.obj
+ return memview
+
+ def __init__(self, obj):
+ if not hasattr(self, '_mem'):
+ if not isinstance(obj, __builtin__.memoryview):
+ obj = __builtin__.memoryview(obj)
+ self._mem = obj
+
+ def __dir__(self):
+ # so dir(...) looks like a memoryview object, and also
+ # contains our additional methods and properties, but not our instance members
+ return sorted(set(self.__class__.__dict__) | set(dir(self._mem)))
+
+ @notreleased
+ def __getitem__(self, item):
+ # if this is a slice, it'll return another real memoryview object
+ # we'll need to wrap that subview in another memoryview wrapper
+ if isinstance(item, slice):
+ return memoryview(self._mem.__getitem__(item))
+
+ return self._mem.__getitem__(item)
+
+ @notreleased
+ def __setitem__(self, key, value):
+ self._mem.__setitem__(key, value)
+
+ @notreleased
+ def __delitem__(self, key):
+ raise TypeError("cannot delete memory")
+
+ def __getattribute__(self, item):
+ try:
+ return object.__getattribute__(self, item)
+
+ except AttributeError:
+ if object.__getattribute__(self, '_mem') is None:
+ raise ValueError("operation forbidden on released memoryview object")
+
+ return object.__getattribute__(self, '_mem').__getattribute__(item)
+
+ def __setattr__(self, key, value):
+ if key not in self.__dict__ and hasattr(__builtin__.memoryview, key):
+ # there are no writable attributes on memoryview objects
+ # changing indexed values is handled by __setitem__
+ raise AttributeError("attribute '{}' of 'memoryview' objects is not writable".format(key))
+
+ else:
+ object.__setattr__(self, key, value)
+
+ @notreleased
+ def __len__(self):
+ return len(self._mem)
+
+ def __eq__(self, other):
+ if isinstance(other, memoryview):
+ return self._mem == other._mem
+
+ return self._mem == other
+
+ @notreleased
+ def __enter__(self):
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ self.release()
+
+ def __repr__(self):
+ return '<{}memory at 0x{:02X}>'.format('' if self._mem else 'released ', id(self))
+
+ def release(self):
+ """Release the underlying buffer exposed by the memoryview object"""
+ # this should effectively do the same job as memoryview.release() in Python 3
+ self._mem = None
+ self._obj = None
+
+ @notreleased
+ def hex(self):
+ """Return the data in the buffer as a string of hexadecimal numbers."""
+ return ''.join(('{:02X}'.format(ord(c)) for c in self._mem))
+
+ # TODO: cast
diff --git a/src/leap/mx/vendor/pgpy/packet/__init__.py b/src/leap/mx/vendor/pgpy/packet/__init__.py
new file mode 100644
index 0000000..36120dd
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/__init__.py
@@ -0,0 +1,11 @@
+from .types import Key
+from .types import Opaque
+from .types import Packet
+from .types import Primary
+from .types import Private
+from .types import Public
+from .types import Sub
+
+from .packets import * # NOQA
+
+__all__ = ['Key', 'Opaque', 'Packet', 'Primary', 'Private', 'Public', 'Sub']
diff --git a/src/leap/mx/vendor/pgpy/packet/fields.py b/src/leap/mx/vendor/pgpy/packet/fields.py
new file mode 100644
index 0000000..b5b5f2e
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/fields.py
@@ -0,0 +1,1514 @@
+""" fields.py
+"""
+from __future__ import absolute_import, division
+
+import abc
+import binascii
+import collections
+import copy
+import hashlib
+import itertools
+import math
+import os
+
+from pyasn1.codec.der import decoder
+from pyasn1.codec.der import encoder
+from pyasn1.type.univ import Integer
+from pyasn1.type.univ import Sequence
+
+from cryptography.exceptions import InvalidSignature
+
+from cryptography.hazmat.backends import default_backend
+
+from cryptography.hazmat.primitives import hashes
+
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric import dsa
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.asymmetric import padding
+
+from cryptography.hazmat.primitives.kdf.concatkdf import ConcatKDFHash
+
+from cryptography.hazmat.primitives.keywrap import aes_key_wrap
+from cryptography.hazmat.primitives.keywrap import aes_key_unwrap
+
+from cryptography.hazmat.primitives.padding import PKCS7
+
+from .subpackets import Signature as SignatureSP
+from .subpackets import UserAttribute
+from .subpackets import signature
+from .subpackets import userattribute
+
+from .types import MPI
+from .types import MPIs
+
+from ..constants import EllipticCurveOID
+from ..constants import HashAlgorithm
+from ..constants import PubKeyAlgorithm
+from ..constants import String2KeyType
+from ..constants import SymmetricKeyAlgorithm
+
+from ..decorators import sdproperty
+
+from ..errors import PGPDecryptionError
+from ..errors import PGPError
+
+from ..symenc import _decrypt
+from ..symenc import _encrypt
+
+from ..types import Field
+
+__all__ = ['SubPackets',
+ 'UserAttributeSubPackets',
+ 'Signature',
+ 'RSASignature',
+ 'DSASignature',
+ 'ECDSASignature',
+ 'PubKey',
+ 'OpaquePubKey',
+ 'RSAPub',
+ 'DSAPub',
+ 'ElGPub',
+ 'ECDSAPub',
+ 'ECDHPub',
+ 'String2Key',
+ 'ECKDF',
+ 'PrivKey',
+ 'OpaquePrivKey',
+ 'RSAPriv',
+ 'DSAPriv',
+ 'ElGPriv',
+ 'ECDSAPriv',
+ 'ECDHPriv',
+ 'CipherText',
+ 'RSACipherText',
+ 'ElGCipherText',
+ 'ECDHCipherText', ]
+
+
+class SubPackets(collections.MutableMapping, Field):
+ _spmodule = signature
+
+ def __init__(self):
+ super(SubPackets, self).__init__()
+ self._hashed_sp = collections.OrderedDict()
+ self._unhashed_sp = collections.OrderedDict()
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += self.__hashbytearray__()
+ _bytes += self.__unhashbytearray__()
+ return _bytes
+
+ def __hashbytearray__(self):
+ _bytes = bytearray()
+ _bytes += self.int_to_bytes(sum(len(sp) for sp in self._hashed_sp.values()), 2)
+ for hsp in self._hashed_sp.values():
+ _bytes += hsp.__bytearray__()
+ return _bytes
+
+ def __unhashbytearray__(self):
+ _bytes = bytearray()
+ _bytes += self.int_to_bytes(sum(len(sp) for sp in self._unhashed_sp.values()), 2)
+ for uhsp in self._unhashed_sp.values():
+ _bytes += uhsp.__bytearray__()
+ return _bytes
+
+ def __len__(self): # pragma: no cover
+ return sum(sp.header.length for sp in itertools.chain(self._hashed_sp.values(), self._unhashed_sp.values())) + 4
+
+ def __iter__(self):
+ for sp in itertools.chain(self._hashed_sp.values(), self._unhashed_sp.values()):
+ yield sp
+
+ def __setitem__(self, key, val):
+ # the key provided should always be the classname for the subpacket
+ # but, there can be multiple subpackets of the same type
+ # so, it should be stored in the format: [h_]<key>_<seqid>
+ # where:
+ # - <key> is the classname of val
+ # - <seqid> is a sequence id, starting at 0, for a given classname
+
+ i = 0
+ if isinstance(key, tuple): # pragma: no cover
+ key, i = key
+
+ d = self._unhashed_sp
+ if key.startswith('h_'):
+ d, key = self._hashed_sp, key[2:]
+
+ while (key, i) in d:
+ i += 1
+
+ d[(key, i)] = val
+
+ def __getitem__(self, key):
+ if isinstance(key, tuple): # pragma: no cover
+ return self._hashed_sp.get(key, self._unhashed_sp.get(key))
+
+ if key.startswith('h_'):
+ return [v for k, v in self._hashed_sp.items() if key[2:] == k[0]]
+
+ else:
+ return [v for k, v in itertools.chain(self._hashed_sp.items(), self._unhashed_sp.items()) if key == k[0]]
+
+ def __delitem__(self, key):
+ ##TODO: this
+ raise NotImplementedError
+
+ def __contains__(self, key):
+ return key in set(k for k, _ in itertools.chain(self._hashed_sp, self._unhashed_sp))
+
+ def __copy__(self):
+ sp = SubPackets()
+ sp._hashed_sp = self._hashed_sp.copy()
+ sp._unhashed_sp = self._unhashed_sp.copy()
+
+ return sp
+
+ def addnew(self, spname, hashed=False, **kwargs):
+ nsp = getattr(self._spmodule, spname)()
+ for p, v in kwargs.items():
+ if hasattr(nsp, p):
+ setattr(nsp, p, v)
+ nsp.update_hlen()
+ if hashed:
+ self['h_' + spname] = nsp
+
+ else:
+ self[spname] = nsp
+
+ def update_hlen(self):
+ for sp in self:
+ sp.update_hlen()
+
+ def parse(self, packet):
+ hl = self.bytes_to_int(packet[:2])
+ del packet[:2]
+
+ # we do it this way because we can't ensure that subpacket headers are sized appropriately
+ # for their contents, but we can at least output that correctly
+ # so instead of tracking how many bytes we can now output, we track how many bytes have we parsed so far
+ plen = len(packet)
+ while plen - len(packet) < hl:
+ sp = SignatureSP(packet)
+ self['h_' + sp.__class__.__name__] = sp
+
+ uhl = self.bytes_to_int(packet[:2])
+ del packet[:2]
+
+ plen = len(packet)
+ while plen - len(packet) < uhl:
+ sp = SignatureSP(packet)
+ self[sp.__class__.__name__] = sp
+
+
+class UserAttributeSubPackets(SubPackets):
+ """
+ This is nearly the same as just the unhashed subpackets from above,
+ except that there isn't a length specifier. So, parse will only parse one packet,
+ appending that one packet to self.__unhashed_sp.
+ """
+ _spmodule = userattribute
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ for uhsp in self._unhashed_sp.values():
+ _bytes += uhsp.__bytearray__()
+ return _bytes
+
+ def __len__(self): # pragma: no cover
+ return sum(len(sp) for sp in self._unhashed_sp.values())
+
+ def parse(self, packet):
+ # parse just one packet and add it to the unhashed subpacket ordereddict
+ # I actually have yet to come across a User Attribute packet with more than one subpacket
+ # which makes sense, given that there is only one defined subpacket
+ sp = UserAttribute(packet)
+ self[sp.__class__.__name__] = sp
+
+
+class Signature(MPIs):
+ def __init__(self):
+ for i in self.__mpis__:
+ setattr(self, i, MPI(0))
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ for i in self:
+ _bytes += i.to_mpibytes()
+ return _bytes
+
+ @abc.abstractproperty
+ def __sig__(self):
+ """return the signature bytes in a format that can be understood by the signature verifier"""
+
+ @abc.abstractmethod
+ def from_signer(self, sig):
+ """create and parse a concrete Signature class instance"""
+
+
+class RSASignature(Signature):
+ __mpis__ = ('md_mod_n', )
+
+ def __sig__(self):
+ return self.md_mod_n.to_mpibytes()[2:]
+
+ def parse(self, packet):
+ self.md_mod_n = MPI(packet)
+
+ def from_signer(self, sig):
+ self.md_mod_n = MPI(self.bytes_to_int(sig))
+
+
+class DSASignature(Signature):
+ __mpis__ = ('r', 's')
+
+ def __sig__(self):
+ # return the signature data into an ASN.1 sequence of integers in DER format
+ seq = Sequence()
+ for i in self:
+ seq.setComponentByPosition(len(seq), Integer(i))
+
+ return encoder.encode(seq)
+
+ def from_signer(self, sig):
+ ##TODO: just use pyasn1 for this
+ def _der_intf(_asn):
+ if _asn[0] != 0x02: # pragma: no cover
+ raise ValueError("Expected: Integer (0x02). Got: 0x{:02X}".format(_asn[0]))
+ del _asn[0]
+
+ if _asn[0] & 0x80: # pragma: no cover
+ llen = _asn[0] & 0x7F
+ del _asn[0]
+
+ flen = self.bytes_to_int(_asn[:llen])
+ del _asn[:llen]
+
+ else:
+ flen = _asn[0] & 0x7F
+ del _asn[0]
+
+ i = self.bytes_to_int(_asn[:flen])
+ del _asn[:flen]
+ return i
+
+ if isinstance(sig, bytes):
+ sig = bytearray(sig)
+
+ # this is a very limited asn1 decoder - it is only intended to decode a DER encoded sequence of integers
+ if not sig[0] == 0x30:
+ raise NotImplementedError("Expected: Sequence (0x30). Got: 0x{:02X}".format(sig[0]))
+ del sig[0]
+
+ # skip the sequence length field
+ if sig[0] & 0x80: # pragma: no cover
+ llen = sig[0] & 0x7F
+ del sig[:llen + 1]
+
+ else:
+ del sig[0]
+
+ self.r = MPI(_der_intf(sig))
+ self.s = MPI(_der_intf(sig))
+
+ def parse(self, packet):
+ self.r = MPI(packet)
+ self.s = MPI(packet)
+
+
+class ECDSASignature(DSASignature):
+ def from_signer(self, sig):
+ seq, _ = decoder.decode(sig)
+ self.r = MPI(seq[0])
+ self.s = MPI(seq[1])
+
+
+class PubKey(MPIs):
+ __pubfields__ = ()
+
+ @property
+ def __mpis__(self):
+ for i in self.__pubfields__:
+ yield i
+
+ def __init__(self):
+ super(PubKey, self).__init__()
+ for field in self.__pubfields__:
+ if isinstance(field, tuple): # pragma: no cover
+ field, val = field
+
+ else:
+ val = MPI(0)
+
+ setattr(self, field, val)
+
+ @abc.abstractmethod
+ def __pubkey__(self):
+ """return the requisite *PublicKey class from the cryptography library"""
+
+ def __len__(self):
+ return sum(len(getattr(self, i)) for i in self.__pubfields__)
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ for field in self.__pubfields__:
+ _bytes += getattr(self, field).to_mpibytes()
+
+ return _bytes
+
+ def publen(self):
+ return len(self)
+
+ def verify(self, subj, sigbytes, hash_alg):
+ return NotImplemented # pragma: no cover
+
+
+class OpaquePubKey(PubKey): # pragma: no cover
+ def __init__(self):
+ super(OpaquePubKey, self).__init__()
+ self.data = bytearray()
+
+ def __iter__(self):
+ yield self.data
+
+ def __pubkey__(self):
+ return NotImplemented
+
+ def __bytearray__(self):
+ return self.data
+
+ def parse(self, packet):
+ ##TODO: this needs to be length-bounded to the end of the packet
+ self.data = packet
+
+
+class RSAPub(PubKey):
+ __pubfields__ = ('n', 'e')
+
+ def __pubkey__(self):
+ return rsa.RSAPublicNumbers(self.e, self.n).public_key(default_backend())
+
+ def verify(self, subj, sigbytes, hash_alg):
+ # zero-pad sigbytes if necessary
+ sigbytes = (b'\x00' * (self.n.byte_length() - len(sigbytes))) + sigbytes
+ verifier = self.__pubkey__().verifier(sigbytes, padding.PKCS1v15(), hash_alg)
+ verifier.update(subj)
+
+ try:
+ verifier.verify()
+
+ except InvalidSignature:
+ return False
+
+ return True
+
+ def parse(self, packet):
+ self.n = MPI(packet)
+ self.e = MPI(packet)
+
+
+class DSAPub(PubKey):
+ __pubfields__ = ('p', 'q', 'g', 'y')
+
+ def __pubkey__(self):
+ params = dsa.DSAParameterNumbers(self.p, self.q, self.g)
+ return dsa.DSAPublicNumbers(self.y, params).public_key(default_backend())
+
+ def verify(self, subj, sigbytes, hash_alg):
+ verifier = self.__pubkey__().verifier(sigbytes, hash_alg)
+ verifier.update(subj)
+
+ try:
+ verifier.verify()
+
+ except InvalidSignature:
+ return False
+
+ return True
+
+ def parse(self, packet):
+ self.p = MPI(packet)
+ self.q = MPI(packet)
+ self.g = MPI(packet)
+ self.y = MPI(packet)
+
+
+class ElGPub(PubKey):
+ __pubfields__ = ('p', 'g', 'y')
+
+ def __pubkey__(self):
+ raise NotImplementedError()
+
+ def parse(self, packet):
+ self.p = MPI(packet)
+ self.g = MPI(packet)
+ self.y = MPI(packet)
+
+
+class ECDSAPub(PubKey):
+ __pubfields__ = ('x', 'y')
+
+ def __init__(self):
+ super(ECDSAPub, self).__init__()
+ self.oid = None
+
+ def __len__(self):
+ return sum([len(getattr(self, i)) - 2 for i in self.__pubfields__] +
+ [3, len(encoder.encode(self.oid.value)) - 1])
+
+ def __pubkey__(self):
+ return ec.EllipticCurvePublicNumbers(self.x, self.y, self.oid.curve()).public_key(default_backend())
+
+ def __bytearray__(self):
+ _b = bytearray()
+ _b += encoder.encode(self.oid.value)[1:]
+ # 0x04 || x || y
+ # where x and y are the same length
+ _xy = b'\x04' + self.x.to_mpibytes()[2:] + self.y.to_mpibytes()[2:]
+ _b += MPI(self.bytes_to_int(_xy, 'big')).to_mpibytes()
+
+ return _b
+
+ def __copy__(self):
+ pkt = super(ECDSAPub, self).__copy__()
+ pkt.oid = self.oid
+ return pkt
+
+ def verify(self, subj, sigbytes, hash_alg):
+ verifier = self.__pubkey__().verifier(sigbytes, ec.ECDSA(hash_alg))
+ verifier.update(subj)
+
+ try:
+ verifier.verify()
+
+ except InvalidSignature:
+ return False
+
+ return True
+
+ def parse(self, packet):
+ oidlen = packet[0]
+ del packet[0]
+ _oid = bytearray(b'\x06')
+ _oid.append(oidlen)
+ _oid += bytearray(packet[:oidlen])
+ # try:
+ oid, _ = decoder.decode(bytes(_oid))
+
+ # except:
+ # raise PGPError("Bad OID octet stream: b'{:s}'".format(''.join(['\\x{:02X}'.format(c) for c in _oid])))
+ self.oid = EllipticCurveOID(oid)
+ del packet[:oidlen]
+
+ # flen = (self.oid.bit_length // 8)
+ xy = bytearray(MPI(packet).to_mpibytes()[2:])
+ # xy = bytearray(MPI(packet).to_bytes(flen, 'big'))
+ # the first byte is just \x04
+ del xy[:1]
+ # now xy needs to be separated into x, y
+ xylen = len(xy)
+ x, y = xy[:xylen // 2], xy[xylen // 2:]
+ self.x = MPI(self.bytes_to_int(x))
+ self.y = MPI(self.bytes_to_int(y))
+
+
+class ECDHPub(PubKey):
+ __pubfields__ = ('x', 'y')
+
+ def __init__(self):
+ super(ECDHPub, self).__init__()
+ self.oid = None
+ self.kdf = ECKDF()
+
+ def __len__(self):
+ return sum([len(getattr(self, i)) - 2 for i in self.__pubfields__] +
+ [3,
+ len(self.kdf),
+ len(encoder.encode(self.oid.value)) - 1])
+
+ def __pubkey__(self):
+ return ec.EllipticCurvePublicNumbers(self.x, self.y, self.oid.curve()).public_key(default_backend())
+
+ def __bytearray__(self):
+ _b = bytearray()
+ _b += encoder.encode(self.oid.value)[1:]
+ # 0x04 || x || y
+ # where x and y are the same length
+ _xy = b'\x04' + self.x.to_mpibytes()[2:] + self.y.to_mpibytes()[2:]
+ _b += MPI(self.bytes_to_int(_xy, 'big')).to_mpibytes()
+ _b += self.kdf.__bytearray__()
+
+ return _b
+
+ def __copy__(self):
+ pkt = super(ECDHPub, self).__copy__()
+ pkt.oid = self.oid
+ pkt.kdf = copy.copy(self.kdf)
+ return pkt
+
+ def parse(self, packet):
+ """
+ Algorithm-Specific Fields for ECDH keys:
+
+ o a variable-length field containing a curve OID, formatted
+ as follows:
+
+ - a one-octet size of the following field; values 0 and
+ 0xFF are reserved for future extensions
+
+ - the octets representing a curve OID, defined in
+ Section 11
+
+ - MPI of an EC point representing a public key
+
+ o a variable-length field containing KDF parameters,
+ formatted as follows:
+
+ - a one-octet size of the following fields; values 0 and
+ 0xff are reserved for future extensions
+
+ - a one-octet value 01, reserved for future extensions
+
+ - a one-octet hash function ID used with a KDF
+
+ - a one-octet algorithm ID for the symmetric algorithm
+ used to wrap the symmetric key used for the message
+ encryption; see Section 8 for details
+ """
+ oidlen = packet[0]
+ del packet[0]
+ _oid = bytearray(b'\x06')
+ _oid.append(oidlen)
+ _oid += bytearray(packet[:oidlen])
+ # try:
+ oid, _ = decoder.decode(bytes(_oid))
+
+ # except:
+ # raise PGPError("Bad OID octet stream: b'{:s}'".format(''.join(['\\x{:02X}'.format(c) for c in _oid])))
+ self.oid = EllipticCurveOID(oid)
+ del packet[:oidlen]
+
+ # flen = (self.oid.bit_length // 8)
+ xy = bytearray(MPI(packet).to_mpibytes()[2:])
+ # xy = bytearray(MPI(packet).to_bytes(flen, 'big'))
+ # the first byte is just \x04
+ del xy[:1]
+ # now xy needs to be separated into x, y
+ xylen = len(xy)
+ x, y = xy[:xylen // 2], xy[xylen // 2:]
+ self.x = MPI(self.bytes_to_int(x))
+ self.y = MPI(self.bytes_to_int(y))
+
+ self.kdf.parse(packet)
+
+
+class String2Key(Field):
+ """
+ 3.7. String-to-Key (S2K) Specifiers
+
+ String-to-key (S2K) specifiers are used to convert passphrase strings
+ into symmetric-key encryption/decryption keys. They are used in two
+ places, currently: to encrypt the secret part of private keys in the
+ private keyring, and to convert passphrases to encryption keys for
+ symmetrically encrypted messages.
+
+ 3.7.1. String-to-Key (S2K) Specifier Types
+
+ There are three types of S2K specifiers currently supported, and
+ some reserved values:
+
+ ID S2K Type
+ -- --------
+ 0 Simple S2K
+ 1 Salted S2K
+ 2 Reserved value
+ 3 Iterated and Salted S2K
+ 100 to 110 Private/Experimental S2K
+
+ These are described in Sections 3.7.1.1 - 3.7.1.3.
+
+ 3.7.1.1. Simple S2K
+
+ This directly hashes the string to produce the key data. See below
+ for how this hashing is done.
+
+ Octet 0: 0x00
+ Octet 1: hash algorithm
+
+ Simple S2K hashes the passphrase to produce the session key. The
+ manner in which this is done depends on the size of the session key
+ (which will depend on the cipher used) and the size of the hash
+ algorithm's output. If the hash size is greater than the session key
+ size, the high-order (leftmost) octets of the hash are used as the
+ key.
+
+ If the hash size is less than the key size, multiple instances of the
+ hash context are created -- enough to produce the required key data.
+ These instances are preloaded with 0, 1, 2, ... octets of zeros (that
+ is to say, the first instance has no preloading, the second gets
+ preloaded with 1 octet of zero, the third is preloaded with two
+ octets of zeros, and so forth).
+
+ As the data is hashed, it is given independently to each hash
+ context. Since the contexts have been initialized differently, they
+ will each produce different hash output. Once the passphrase is
+ hashed, the output data from the multiple hashes is concatenated,
+ first hash leftmost, to produce the key data, with any excess octets
+ on the right discarded.
+
+ 3.7.1.2. Salted S2K
+
+ This includes a "salt" value in the S2K specifier -- some arbitrary
+ data -- that gets hashed along with the passphrase string, to help
+ prevent dictionary attacks.
+
+ Octet 0: 0x01
+ Octet 1: hash algorithm
+ Octets 2-9: 8-octet salt value
+
+ Salted S2K is exactly like Simple S2K, except that the input to the
+ hash function(s) consists of the 8 octets of salt from the S2K
+ specifier, followed by the passphrase.
+
+ 3.7.1.3. Iterated and Salted S2K
+
+ This includes both a salt and an octet count. The salt is combined
+ with the passphrase and the resulting value is hashed repeatedly.
+ This further increases the amount of work an attacker must do to try
+ dictionary attacks.
+
+ Octet 0: 0x03
+ Octet 1: hash algorithm
+ Octets 2-9: 8-octet salt value
+ Octet 10: count, a one-octet, coded value
+
+ The count is coded into a one-octet number using the following
+ formula:
+
+ #define EXPBIAS 6
+ count = ((Int32)16 + (c & 15)) << ((c >> 4) + EXPBIAS);
+
+ The above formula is in C, where "Int32" is a type for a 32-bit
+ integer, and the variable "c" is the coded count, Octet 10.
+
+ Iterated-Salted S2K hashes the passphrase and salt data multiple
+ times. The total number of octets to be hashed is specified in the
+ encoded count in the S2K specifier. Note that the resulting count
+ value is an octet count of how many octets will be hashed, not an
+ iteration count.
+
+ Initially, one or more hash contexts are set up as with the other S2K
+ algorithms, depending on how many octets of key data are needed.
+ Then the salt, followed by the passphrase data, is repeatedly hashed
+ until the number of octets specified by the octet count has been
+ hashed. The one exception is that if the octet count is less than
+ the size of the salt plus passphrase, the full salt plus passphrase
+ will be hashed even though that is greater than the octet count.
+ After the hashing is done, the data is unloaded from the hash
+ context(s) as with the other S2K algorithms.
+ """
+ @sdproperty
+ def encalg(self):
+ return self._encalg
+
+ @encalg.register(int)
+ @encalg.register(SymmetricKeyAlgorithm)
+ def encalg_int(self, val):
+ self._encalg = SymmetricKeyAlgorithm(val)
+
+ @sdproperty
+ def specifier(self):
+ return self._specifier
+
+ @specifier.register(int)
+ @specifier.register(String2KeyType)
+ def specifier_int(self, val):
+ self._specifier = String2KeyType(val)
+
+ @sdproperty
+ def halg(self):
+ return self._halg
+
+ @halg.register(int)
+ @halg.register(HashAlgorithm)
+ def halg_int(self, val):
+ self._halg = HashAlgorithm(val)
+
+ @sdproperty
+ def count(self):
+ return (16 + (self._count & 15)) << ((self._count >> 4) + 6)
+
+ @count.register(int)
+ def count_int(self, val):
+ if val < 0 or val > 255: # pragma: no cover
+ raise ValueError("count must be between 0 and 256")
+ self._count = val
+
+ def __init__(self):
+ super(String2Key, self).__init__()
+ self.usage = 0
+ self.encalg = 0
+ self.specifier = 0
+ self.iv = None
+
+ # specifier-specific fields
+ # simple, salted, iterated
+ self.halg = 0
+
+ # salted, iterated
+ self.salt = bytearray()
+
+ # iterated
+ self.count = 0
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes.append(self.usage)
+ if bool(self):
+ _bytes.append(self.encalg)
+ _bytes.append(self.specifier)
+ if self.specifier >= String2KeyType.Simple:
+ _bytes.append(self.halg)
+ if self.specifier >= String2KeyType.Salted:
+ _bytes += self.salt
+ if self.specifier == String2KeyType.Iterated:
+ _bytes.append(self._count)
+ if self.iv is not None:
+ _bytes += self.iv
+ return _bytes
+
+ def __len__(self):
+ return len(self.__bytearray__())
+
+ def __bool__(self):
+ return self.usage in [254, 255]
+
+ def __nonzero__(self):
+ return self.__bool__()
+
+ def __copy__(self):
+ s2k = String2Key()
+ s2k.usage = self.usage
+ s2k.encalg = self.encalg
+ s2k.specifier = self.specifier
+ s2k.iv = self.iv
+ s2k.halg = self.halg
+ s2k.salt = copy.copy(self.salt)
+ s2k.count = self._count
+ return s2k
+
+ def parse(self, packet, iv=True):
+ self.usage = packet[0]
+ del packet[0]
+
+ if bool(self):
+ self.encalg = packet[0]
+ del packet[0]
+
+ self.specifier = packet[0]
+ del packet[0]
+
+ if self.specifier >= String2KeyType.Simple:
+ # this will always be true
+ self.halg = packet[0]
+ del packet[0]
+
+ if self.specifier >= String2KeyType.Salted:
+ self.salt = packet[:8]
+ del packet[:8]
+
+ if self.specifier == String2KeyType.Iterated:
+ self.count = packet[0]
+ del packet[0]
+
+ if iv:
+ self.iv = packet[:(self.encalg.block_size // 8)]
+ del packet[:(self.encalg.block_size // 8)]
+
+ def derive_key(self, passphrase):
+ ##TODO: raise an exception if self.usage is not 254 or 255
+ keylen = self.encalg.key_size
+ hashlen = self.halg.digest_size * 8
+
+ ctx = int(math.ceil((keylen / hashlen)))
+
+ # Simple S2K - always done
+ hsalt = b''
+ hpass = passphrase.encode('latin-1')
+
+ # salted, iterated S2K
+ if self.specifier >= String2KeyType.Salted:
+ hsalt = bytes(self.salt)
+
+ count = len(hsalt + hpass)
+ if self.specifier == String2KeyType.Iterated and self.count > len(hsalt + hpass):
+ count = self.count
+
+ hcount = (count // len(hsalt + hpass))
+ hleft = count - (hcount * len(hsalt + hpass))
+
+ hashdata = ((hsalt + hpass) * hcount) + (hsalt + hpass)[:hleft]
+
+ h = []
+ for i in range(0, ctx):
+ _h = self.halg.hasher
+ _h.update(b'\x00' * i)
+ _h.update(hashdata)
+ h.append(_h)
+
+ # GC some stuff
+ del hsalt
+ del hpass
+ del hashdata
+
+ # and return the key!
+ return b''.join(hc.digest() for hc in h)[:(keylen // 8)]
+
+
+class ECKDF(Field):
+ """
+ o a variable-length field containing KDF parameters,
+ formatted as follows:
+
+ - a one-octet size of the following fields; values 0 and
+ 0xff are reserved for future extensions
+
+ - a one-octet value 01, reserved for future extensions
+
+ - a one-octet hash function ID used with a KDF
+
+ - a one-octet algorithm ID for the symmetric algorithm
+ used to wrap the symmetric key used for the message
+ encryption; see Section 8 for details
+ """
+ @sdproperty
+ def halg(self):
+ return self._halg
+
+ @halg.register(int)
+ @halg.register(HashAlgorithm)
+ def halg_int(self, val):
+ self._halg = HashAlgorithm(val)
+
+ @sdproperty
+ def encalg(self):
+ return self._encalg
+
+ @encalg.register(int)
+ @encalg.register(SymmetricKeyAlgorithm)
+ def encalg_int(self, val):
+ self._encalg = SymmetricKeyAlgorithm(val)
+
+ def __init__(self):
+ super(ECKDF, self).__init__()
+ self.halg = 0
+ self.encalg = 0
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes.append(len(self) - 1)
+ _bytes.append(0x01)
+ _bytes.append(self.halg)
+ _bytes.append(self.encalg)
+ return _bytes
+
+ def __len__(self):
+ return 4
+
+ def parse(self, packet):
+ # packet[0] should always be 3
+ # packet[1] should always be 1
+ # TODO: this assert is likely not necessary, but we should raise some kind of exception
+ # if parsing fails due to these fields being incorrect
+ assert packet[:2] == b'\x03\x01'
+ del packet[:2]
+
+ self.halg = packet[0]
+ del packet[0]
+
+ self.encalg = packet[0]
+ del packet[0]
+
+ def derive_key(self, s, curve, pkalg, fingerprint):
+ # wrapper around the Concatenation KDF method provided by cryptography
+ # assemble the additional data as defined in RFC 6637:
+ # Param = curve_OID_len || curve_OID || public_key_alg_ID || 03 || 01 || KDF_hash_ID || KEK_alg_ID for AESKeyWrap || "Anonymous
+ data = bytearray()
+ data += encoder.encode(curve.value)[1:]
+ data.append(pkalg)
+ data += b'\x03\x01'
+ data.append(self.halg)
+ data.append(self.encalg)
+ data += b'Anonymous Sender '
+ data += binascii.unhexlify(fingerprint.replace(' ', ''))
+
+ ckdf = ConcatKDFHash(algorithm=getattr(hashes, self.halg.name)(), length=self.encalg.key_size // 8, otherinfo=bytes(data), backend=default_backend())
+ return ckdf.derive(s)
+
+
+class PrivKey(PubKey):
+ __privfields__ = ()
+
+ @property
+ def __mpis__(self):
+ for i in super(PrivKey, self).__mpis__:
+ yield i
+
+ for i in self.__privfields__:
+ yield i
+
+ def __init__(self):
+ super(PrivKey, self).__init__()
+
+ self.s2k = String2Key()
+ self.encbytes = bytearray()
+ self.chksum = bytearray()
+
+ for field in self.__privfields__:
+ setattr(self, field, MPI(0))
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(PrivKey, self).__bytearray__()
+
+ _bytes += self.s2k.__bytearray__()
+ if self.s2k:
+ _bytes += self.encbytes
+
+ else:
+ for field in self.__privfields__:
+ _bytes += getattr(self, field).to_mpibytes()
+
+ if self.s2k.usage == 0:
+ _bytes += self.chksum
+
+ return _bytes
+
+ def __len__(self):
+ l = super(PrivKey, self).__len__() + len(self.s2k) + len(self.chksum)
+ if self.s2k:
+ l += len(self.encbytes)
+
+ else:
+ l += sum(len(getattr(self, i)) for i in self.__privfields__)
+
+ return l
+
+ def __copy__(self):
+ pk = super(PrivKey, self).__copy__()
+ pk.s2k = copy.copy(self.s2k)
+ pk.encbytes = copy.copy(self.encbytes)
+ pk.chksum = copy.copy(self.chksum)
+ return pk
+
+ @abc.abstractmethod
+ def __privkey__(self):
+ """return the requisite *PrivateKey class from the cryptography library"""
+
+ @abc.abstractmethod
+ def _generate(self, key_size):
+ """Generate a new PrivKey"""
+
+ def _compute_chksum(self):
+ chs = sum(sum(bytearray(c.to_mpibytes())) for c in self) % 65536
+ self.chksum = bytearray(self.int_to_bytes(chs, 2))
+
+ def publen(self):
+ return super(PrivKey, self).__len__()
+
+ def encrypt_keyblob(self, passphrase, enc_alg, hash_alg):
+ # PGPy will only ever use iterated and salted S2k mode
+ self.s2k.usage = 254
+ self.s2k.encalg = enc_alg
+ self.s2k.specifier = String2KeyType.Iterated
+ self.s2k.iv = enc_alg.gen_iv()
+ self.s2k.halg = hash_alg
+ self.s2k.salt = bytearray(os.urandom(8))
+ self.s2k.count = hash_alg.tuned_count
+
+ # now that String-to-Key is ready to go, derive sessionkey from passphrase
+ # and then unreference passphrase
+ sessionkey = self.s2k.derive_key(passphrase)
+ del passphrase
+
+ pt = bytearray()
+ for pf in self.__privfields__:
+ pt += getattr(self, pf).to_mpibytes()
+
+ # append a SHA-1 hash of the plaintext so far to the plaintext
+ pt += hashlib.new('sha1', pt).digest()
+
+ # encrypt
+ self.encbytes = bytearray(_encrypt(bytes(pt), bytes(sessionkey), enc_alg, bytes(self.s2k.iv)))
+
+ # delete pt and clear self
+ del pt
+ self.clear()
+
+ @abc.abstractmethod
+ def decrypt_keyblob(self, passphrase):
+ if not self.s2k: # pragma: no cover
+ # not encrypted
+ return
+
+ # Encryption/decryption of the secret data is done in CFB mode using
+ # the key created from the passphrase and the Initial Vector from the
+ # packet. A different mode is used with V3 keys (which are only RSA)
+ # than with other key formats. (...)
+ #
+ # With V4 keys, a simpler method is used. All secret MPI values are
+ # encrypted in CFB mode, including the MPI bitcount prefix.
+
+ # derive the session key from our passphrase, and then unreference passphrase
+ sessionkey = self.s2k.derive_key(passphrase)
+ del passphrase
+
+ # attempt to decrypt this key
+ pt = _decrypt(bytes(self.encbytes), bytes(sessionkey), self.s2k.encalg, bytes(self.s2k.iv))
+
+ # check the hash to see if we decrypted successfully or not
+ if self.s2k.usage == 254 and not pt[-20:] == hashlib.new('sha1', pt[:-20]).digest():
+ # if the usage byte is 254, key material is followed by a 20-octet sha-1 hash of the rest
+ # of the key material block
+ raise PGPDecryptionError("Passphrase was incorrect!")
+
+ if self.s2k.usage == 255 and not self.bytes_to_int(pt[-2:]) == (sum(bytearray(pt[:-2])) % 65536): # pragma: no cover
+ # if the usage byte is 255, key material is followed by a 2-octet checksum of the rest
+ # of the key material block
+ raise PGPDecryptionError("Passphrase was incorrect!")
+
+ return bytearray(pt)
+
+ def sign(self, sigdata, hash_alg):
+ return NotImplemented # pragma: no cover
+
+ def clear(self):
+ """delete and re-initialize all private components to zero"""
+ for field in self.__privfields__:
+ delattr(self, field)
+ setattr(self, field, MPI(0))
+
+
+class OpaquePrivKey(PrivKey, OpaquePubKey): # pragma: no cover
+ def __privkey__(self):
+ return NotImplemented
+
+ def _generate(self, key_size):
+ # return NotImplemented
+ raise NotImplementedError()
+
+ def decrypt_keyblob(self, passphrase):
+ return NotImplemented
+
+
+class RSAPriv(PrivKey, RSAPub):
+ __privfields__ = ('d', 'p', 'q', 'u')
+
+ def __privkey__(self):
+ return rsa.RSAPrivateNumbers(self.p, self.q, self.d,
+ rsa.rsa_crt_dmp1(self.d, self.p),
+ rsa.rsa_crt_dmq1(self.d, self.q),
+ rsa.rsa_crt_iqmp(self.p, self.q),
+ rsa.RSAPublicNumbers(self.e, self.n)).private_key(default_backend())
+
+ def _generate(self, key_size):
+ if any(c != 0 for c in self): # pragma: no cover
+ raise PGPError("key is already populated")
+
+ # generate some big numbers!
+ pk = rsa.generate_private_key(65537, key_size, default_backend())
+ pkn = pk.private_numbers()
+
+ self.n = MPI(pkn.public_numbers.n)
+ self.e = MPI(pkn.public_numbers.e)
+ self.d = MPI(pkn.d)
+ self.p = MPI(pkn.p)
+ self.q = MPI(pkn.q)
+ # from the RFC:
+ # "- MPI of u, the multiplicative inverse of p, mod q."
+ # or, simply, p^-1 mod p
+ # rsa.rsa_crt_iqmp(p, q) normally computes q^-1 mod p,
+ # so if we swap the values around we get the answer we want
+ self.u = MPI(rsa.rsa_crt_iqmp(pkn.q, pkn.p))
+
+ del pkn
+ del pk
+
+ self._compute_chksum()
+
+ def parse(self, packet):
+ super(RSAPriv, self).parse(packet)
+ self.s2k.parse(packet)
+
+ if not self.s2k:
+ self.d = MPI(packet)
+ self.p = MPI(packet)
+ self.q = MPI(packet)
+ self.u = MPI(packet)
+
+ if self.s2k.usage == 0:
+ self.chksum = packet[:2]
+ del packet[:2]
+
+ else:
+ ##TODO: this needs to be bounded to the length of the encrypted key material
+ self.encbytes = packet
+
+ def decrypt_keyblob(self, passphrase):
+ kb = super(RSAPriv, self).decrypt_keyblob(passphrase)
+ del passphrase
+
+ self.d = MPI(kb)
+ self.p = MPI(kb)
+ self.q = MPI(kb)
+ self.u = MPI(kb)
+
+ if self.s2k.usage in [254, 255]:
+ self.chksum = kb
+ del kb
+
+ def sign(self, sigdata, hash_alg):
+ signer = self.__privkey__().signer(padding.PKCS1v15(), hash_alg)
+ signer.update(sigdata)
+ return signer.finalize()
+
+
+class DSAPriv(PrivKey, DSAPub):
+ __privfields__ = ('x',)
+
+ def __privkey__(self):
+ params = dsa.DSAParameterNumbers(self.p, self.q, self.g)
+ pn = dsa.DSAPublicNumbers(self.y, params)
+ return dsa.DSAPrivateNumbers(self.x, pn).private_key(default_backend())
+
+ def _generate(self, key_size):
+ if any(c != 0 for c in self): # pragma: no cover
+ raise PGPError("key is already populated")
+
+ # generate some big numbers!
+ pk = dsa.generate_private_key(key_size, default_backend())
+ pkn = pk.private_numbers()
+
+ self.p = MPI(pkn.public_numbers.parameter_numbers.p)
+ self.q = MPI(pkn.public_numbers.parameter_numbers.q)
+ self.g = MPI(pkn.public_numbers.parameter_numbers.g)
+ self.y = MPI(pkn.public_numbers.y)
+ self.x = MPI(pkn.x)
+
+ del pkn
+ del pk
+
+ self._compute_chksum()
+
+ def parse(self, packet):
+ super(DSAPriv, self).parse(packet)
+ self.s2k.parse(packet)
+
+ if not self.s2k:
+ self.x = MPI(packet)
+
+ else:
+ self.encbytes = packet
+
+ if self.s2k.usage in [0, 255]:
+ self.chksum = packet[:2]
+ del packet[:2]
+
+ def decrypt_keyblob(self, passphrase):
+ kb = super(DSAPriv, self).decrypt_keyblob(passphrase)
+ del passphrase
+
+ self.x = MPI(kb)
+
+ if self.s2k.usage in [254, 255]:
+ self.chksum = kb
+ del kb
+
+ def sign(self, sigdata, hash_alg):
+ signer = self.__privkey__().signer(hash_alg)
+ signer.update(sigdata)
+ return signer.finalize()
+
+
+class ElGPriv(PrivKey, ElGPub):
+ __privfields__ = ('x', )
+
+ def __privkey__(self):
+ raise NotImplementedError()
+
+ def _generate(self, key_size):
+ raise NotImplementedError(PubKeyAlgorithm.ElGamal)
+
+ def parse(self, packet):
+ super(ElGPriv, self).parse(packet)
+ self.s2k.parse(packet)
+
+ if not self.s2k:
+ self.x = MPI(packet)
+
+ else:
+ self.encbytes = packet
+
+ if self.s2k.usage in [0, 255]:
+ self.chksum = packet[:2]
+ del packet[:2]
+
+ def decrypt_keyblob(self, passphrase):
+ kb = super(ElGPriv, self).decrypt_keyblob(passphrase)
+ del passphrase
+
+ self.x = MPI(kb)
+
+ if self.s2k.usage in [254, 255]:
+ self.chksum = kb
+ del kb
+
+
+class ECDSAPriv(PrivKey, ECDSAPub):
+ __privfields__ = ('s', )
+
+ def __privkey__(self):
+ ecp = ec.EllipticCurvePublicNumbers(self.x, self.y, self.oid.curve())
+ return ec.EllipticCurvePrivateNumbers(self.s, ecp).private_key(default_backend())
+
+ def _generate(self, oid):
+ if any(c != 0 for c in self): # pragma: no cover
+ raise PGPError("Key is already populated!")
+
+ self.oid = EllipticCurveOID(oid)
+
+ if not self.oid.can_gen:
+ raise ValueError("Curve not currently supported: {}".format(oid.name))
+
+ pk = ec.generate_private_key(self.oid.curve(), default_backend())
+ pubn = pk.public_key().public_numbers()
+ self.x = MPI(pubn.x)
+ self.y = MPI(pubn.y)
+ self.s = MPI(pk.private_numbers().private_value)
+
+ def parse(self, packet):
+ super(ECDSAPriv, self).parse(packet)
+ self.s2k.parse(packet)
+
+ if not self.s2k:
+ self.s = MPI(packet)
+
+ if self.s2k.usage == 0:
+ self.chksum = packet[:2]
+ del packet[:2]
+
+ else:
+ ##TODO: this needs to be bounded to the length of the encrypted key material
+ self.encbytes = packet
+
+ def decrypt_keyblob(self, passphrase):
+ kb = super(ECDSAPriv, self).decrypt_keyblob(passphrase)
+ del passphrase
+
+ self.s = MPI(kb)
+
+ def sign(self, sigdata, hash_alg):
+ signer = self.__privkey__().signer(ec.ECDSA(hash_alg))
+ signer.update(sigdata)
+ return signer.finalize()
+
+
+class ECDHPriv(ECDSAPriv, ECDHPub):
+ def __bytearray__(self):
+ _b = ECDHPub.__bytearray__(self)
+ _b += self.s2k.__bytearray__()
+ if not self.s2k:
+ _b += self.s.to_mpibytes()
+
+ if self.s2k.usage == 0:
+ _b += self.chksum
+
+ else:
+ _b += self.encbytes
+
+ return _b
+
+ def __len__(self):
+ # because of the inheritance used for this, ECDSAPub.__len__ is called instead of ECDHPub.__len__
+ # the only real difference is self.kdf, so we can just add that
+ return super(ECDHPriv, self).__len__() + len(self.kdf)
+
+ def _generate(self, oid):
+ ECDSAPriv._generate(self, oid)
+ self.kdf.halg = self.oid.kdf_halg
+ self.kdf.encalg = self.oid.kek_alg
+
+ def publen(self):
+ return ECDHPub.__len__(self)
+
+ def parse(self, packet):
+ ECDHPub.parse(self, packet)
+ self.s2k.parse(packet)
+
+ if not self.s2k:
+ self.s = MPI(packet)
+
+ if self.s2k.usage == 0:
+ self.chksum = packet[:2]
+ del packet[:2]
+
+ else:
+ ##TODO: this needs to be bounded to the length of the encrypted key material
+ self.encbytes = packet
+
+
+class CipherText(MPIs):
+ def __init__(self):
+ super(CipherText, self).__init__()
+ for i in self.__mpis__:
+ setattr(self, i, MPI(0))
+
+ @classmethod
+ @abc.abstractmethod
+ def encrypt(cls, encfn, *args):
+ """create and populate a concrete CipherText class instance"""
+
+ @abc.abstractmethod
+ def decrypt(self, decfn, *args):
+ """decrypt the ciphertext contained in this CipherText instance"""
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ for i in self:
+ _bytes += i.to_mpibytes()
+ return _bytes
+
+
+class RSACipherText(CipherText):
+ __mpis__ = ('me_mod_n', )
+
+ @classmethod
+ def encrypt(cls, encfn, *args):
+ ct = cls()
+ ct.me_mod_n = MPI(cls.bytes_to_int(encfn(*args)))
+ return ct
+
+ def decrypt(self, decfn, *args):
+ return decfn(*args)
+
+ def parse(self, packet):
+ self.me_mod_n = MPI(packet)
+
+
+class ElGCipherText(CipherText):
+ __mpis__ = ('gk_mod_p', 'myk_mod_p')
+
+ @classmethod
+ def encrypt(cls, encfn, *args):
+ raise NotImplementedError()
+
+ def decrypt(self, decfn, *args):
+ raise NotImplementedError()
+
+ def parse(self, packet):
+ self.gk_mod_p = MPI(packet)
+ self.myk_mod_p = MPI(packet)
+
+
+class ECDHCipherText(CipherText):
+ __mpis__ = ('vX', 'vY')
+
+ @classmethod
+ def encrypt(cls, pk, *args):
+ """
+ For convenience, the synopsis of the encoding method is given below;
+ however, this section, [NIST-SP800-56A], and [RFC3394] are the
+ normative sources of the definition.
+
+ Obtain the authenticated recipient public key R
+ Generate an ephemeral key pair {v, V=vG}
+ Compute the shared point S = vR;
+ m = symm_alg_ID || session key || checksum || pkcs5_padding;
+ curve_OID_len = (byte)len(curve_OID);
+ Param = curve_OID_len || curve_OID || public_key_alg_ID || 03
+ || 01 || KDF_hash_ID || KEK_alg_ID for AESKeyWrap || "Anonymous
+ Sender " || recipient_fingerprint;
+ Z_len = the key size for the KEK_alg_ID used with AESKeyWrap
+ Compute Z = KDF( S, Z_len, Param );
+ Compute C = AESKeyWrap( Z, m ) as per [RFC3394]
+ VB = convert point V to the octet string
+ Output (MPI(VB) || len(C) || C).
+
+ The decryption is the inverse of the method given. Note that the
+ recipient obtains the shared secret by calculating
+ """
+ # *args should be:
+ # - m
+ #
+ _m, = args
+
+ # m may need to be PKCS5-padded
+ padder = PKCS7(64).padder()
+ m = padder.update(_m) + padder.finalize()
+
+ km = pk.keymaterial
+
+ ct = cls()
+
+ # generate ephemeral key pair, then store it in ct
+ v = ec.generate_private_key(km.oid.curve(), default_backend())
+ ct.vX = MPI(v.public_key().public_numbers().x)
+ ct.vY = MPI(v.public_key().public_numbers().y)
+
+ # compute the shared point S
+ s = v.exchange(ec.ECDH(), km.__pubkey__())
+
+ # derive the wrapping key
+ z = km.kdf.derive_key(s, km.oid, PubKeyAlgorithm.ECDH, pk.fingerprint)
+
+ # compute C
+ ct.c = aes_key_wrap(z, m, default_backend())
+
+ return ct
+
+ def decrypt(self, pk, *args):
+ km = pk.keymaterial
+ # assemble the public component of ephemeral key v
+ v = ec.EllipticCurvePublicNumbers(self.vX, self.vY, km.oid.curve()).public_key(default_backend())
+
+ # compute s using the inverse of how it was derived during encryption
+ s = km.__privkey__().exchange(ec.ECDH(), v)
+
+ # derive the wrapping key
+ z = km.kdf.derive_key(s, km.oid, PubKeyAlgorithm.ECDH, pk.fingerprint)
+
+ # unwrap and unpad m
+ _m = aes_key_unwrap(z, self.c, default_backend())
+
+ padder = PKCS7(64).unpadder()
+ return padder.update(_m) + padder.finalize()
+
+ def __init__(self):
+ super(ECDHCipherText, self).__init__()
+ self.c = bytearray(0)
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _xy = b'\x04' + self.vX.to_mpibytes()[2:] + self.vY.to_mpibytes()[2:]
+ _bytes += MPI(self.bytes_to_int(_xy, 'big')).to_mpibytes()
+ _bytes.append(len(self.c))
+ _bytes += self.c
+
+ return _bytes
+
+ def parse(self, packet):
+ # self.v = MPI(packet)
+ xy = bytearray(MPI(packet).to_mpibytes()[2:])
+ del xy[:1]
+ xylen = len(xy)
+ x, y = xy[:xylen // 2], xy[xylen // 2:]
+ self.vX = MPI(self.bytes_to_int(x))
+ self.vY = MPI(self.bytes_to_int(y))
+
+ clen = packet[0]
+ del packet[0]
+
+ self.c += packet[:clen]
+ del packet[:clen]
diff --git a/src/leap/mx/vendor/pgpy/packet/packets.py b/src/leap/mx/vendor/pgpy/packet/packets.py
new file mode 100644
index 0000000..38a7200
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/packets.py
@@ -0,0 +1,1605 @@
+""" packet.py
+"""
+import abc
+import binascii
+import calendar
+import copy
+import hashlib
+import os
+import re
+
+from datetime import datetime
+
+import six
+
+from cryptography.hazmat.primitives import constant_time
+from cryptography.hazmat.primitives.asymmetric import padding
+
+from .fields import DSAPriv, DSAPub, DSASignature
+from .fields import ECDSAPub, ECDSAPriv, ECDSASignature
+from .fields import ECDHPub, ECDHPriv, ECDHCipherText
+from .fields import ElGCipherText, ElGPriv, ElGPub
+from .fields import OpaquePubKey
+from .fields import OpaquePrivKey
+from .fields import RSACipherText, RSAPriv, RSAPub, RSASignature
+from .fields import String2Key
+from .fields import SubPackets
+from .fields import UserAttributeSubPackets
+
+from .types import Packet
+from .types import Primary
+from .types import Private
+from .types import Public
+from .types import Sub
+from .types import VersionedPacket
+
+from ..constants import CompressionAlgorithm
+from ..constants import HashAlgorithm
+from ..constants import PubKeyAlgorithm
+from ..constants import SignatureType
+from ..constants import SymmetricKeyAlgorithm
+from ..constants import TrustFlags
+from ..constants import TrustLevel
+
+from ..decorators import sdproperty
+
+from ..errors import PGPDecryptionError
+
+from ..symenc import _decrypt
+from ..symenc import _encrypt
+
+from ..types import Fingerprint
+
+__all__ = ['PKESessionKey',
+ 'PKESessionKeyV3',
+ 'Signature',
+ 'SignatureV4',
+ 'SKESessionKey',
+ 'SKESessionKeyV4',
+ 'OnePassSignature',
+ 'OnePassSignatureV3',
+ 'PrivKey',
+ 'PubKey',
+ 'PubKeyV4',
+ 'PrivKeyV4',
+ 'PrivSubKey',
+ 'PrivSubKeyV4',
+ 'CompressedData',
+ 'SKEData',
+ 'Marker',
+ 'LiteralData',
+ 'Trust',
+ 'UserID',
+ 'PubSubKey',
+ 'PubSubKeyV4',
+ 'UserAttribute',
+ 'IntegrityProtectedSKEData',
+ 'IntegrityProtectedSKEDataV1',
+ 'MDC']
+
+
+class PKESessionKey(VersionedPacket):
+ __typeid__ = 0x01
+ __ver__ = 0
+
+ @abc.abstractmethod
+ def decrypt_sk(self, pk):
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def encrypt_sk(self, pk, symalg, symkey):
+ raise NotImplementedError()
+
+
+class PKESessionKeyV3(PKESessionKey):
+ """
+ 5.1. Public-Key Encrypted Session Key Packets (Tag 1)
+
+ A Public-Key Encrypted Session Key packet holds the session key used
+ to encrypt a message. Zero or more Public-Key Encrypted Session Key
+ packets and/or Symmetric-Key Encrypted Session Key packets may
+ precede a Symmetrically Encrypted Data Packet, which holds an
+ encrypted message. The message is encrypted with the session key,
+ and the session key is itself encrypted and stored in the Encrypted
+ Session Key packet(s). The Symmetrically Encrypted Data Packet is
+ preceded by one Public-Key Encrypted Session Key packet for each
+ OpenPGP key to which the message is encrypted. The recipient of the
+ message finds a session key that is encrypted to their public key,
+ decrypts the session key, and then uses the session key to decrypt
+ the message.
+
+ The body of this packet consists of:
+
+ - A one-octet number giving the version number of the packet type.
+ The currently defined value for packet version is 3.
+
+ - An eight-octet number that gives the Key ID of the public key to
+ which the session key is encrypted. If the session key is
+ encrypted to a subkey, then the Key ID of this subkey is used
+ here instead of the Key ID of the primary key.
+
+ - A one-octet number giving the public-key algorithm used.
+
+ - A string of octets that is the encrypted session key. This
+ string takes up the remainder of the packet, and its contents are
+ dependent on the public-key algorithm used.
+
+ Algorithm Specific Fields for RSA encryption
+
+ - multiprecision integer (MPI) of RSA encrypted value m**e mod n.
+
+ Algorithm Specific Fields for Elgamal encryption:
+
+ - MPI of Elgamal (Diffie-Hellman) value g**k mod p.
+
+ - MPI of Elgamal (Diffie-Hellman) value m * y**k mod p.
+
+ The value "m" in the above formulas is derived from the session key
+ as follows. First, the session key is prefixed with a one-octet
+ algorithm identifier that specifies the symmetric encryption
+ algorithm used to encrypt the following Symmetrically Encrypted Data
+ Packet. Then a two-octet checksum is appended, which is equal to the
+ sum of the preceding session key octets, not including the algorithm
+ identifier, modulo 65536. This value is then encoded as described in
+ PKCS#1 block encoding EME-PKCS1-v1_5 in Section 7.2.1 of [RFC3447] to
+ form the "m" value used in the formulas above. See Section 13.1 of
+ this document for notes on OpenPGP's use of PKCS#1.
+
+ Note that when an implementation forms several PKESKs with one
+ session key, forming a message that can be decrypted by several keys,
+ the implementation MUST make a new PKCS#1 encoding for each key.
+
+ An implementation MAY accept or use a Key ID of zero as a "wild card"
+ or "speculative" Key ID. In this case, the receiving implementation
+ would try all available private keys, checking for a valid decrypted
+ session key. This format helps reduce traffic analysis of messages.
+ """
+ __ver__ = 3
+
+ @sdproperty
+ def encrypter(self):
+ return self._encrypter
+
+ @encrypter.register(bytearray)
+ def encrypter_bin(self, val):
+ self._encrypter = binascii.hexlify(val).upper().decode('latin-1')
+
+ @sdproperty
+ def pkalg(self):
+ return self._pkalg
+
+ @pkalg.register(int)
+ @pkalg.register(PubKeyAlgorithm)
+ def pkalg_int(self, val):
+ self._pkalg = PubKeyAlgorithm(val)
+
+ _c = {PubKeyAlgorithm.RSAEncryptOrSign: RSACipherText,
+ PubKeyAlgorithm.RSAEncrypt: RSACipherText,
+ PubKeyAlgorithm.ElGamal: ElGCipherText,
+ PubKeyAlgorithm.FormerlyElGamalEncryptOrSign: ElGCipherText,
+ PubKeyAlgorithm.ECDH: ECDHCipherText}
+
+ ct = _c.get(self._pkalg, None)
+ self.ct = ct() if ct is not None else ct
+
+ def __init__(self):
+ super(PKESessionKeyV3, self).__init__()
+ self.encrypter = bytearray(8)
+ self.pkalg = 0
+ self.ct = None
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(PKESessionKeyV3, self).__bytearray__()
+ _bytes += binascii.unhexlify(self.encrypter.encode())
+ _bytes += bytearray([self.pkalg])
+ _bytes += self.ct.__bytearray__() if self.ct is not None else b'\x00' * (self.header.length - 10)
+ return _bytes
+
+ def __copy__(self):
+ sk = self.__class__()
+ sk.header = copy.copy(self.header)
+ sk._encrypter = self._encrypter
+ sk.pkalg = self.pkalg
+ if self.ct is not None:
+ sk.ct = copy.copy(self.ct)
+
+ return sk
+
+ def decrypt_sk(self, pk):
+ if self.pkalg == PubKeyAlgorithm.RSAEncryptOrSign:
+ # pad up ct with null bytes if necessary
+ ct = self.ct.me_mod_n.to_mpibytes()[2:]
+ ct = b'\x00' * ((pk.keymaterial.__privkey__().key_size // 8) - len(ct)) + ct
+
+ decrypter = pk.keymaterial.__privkey__().decrypt
+ decargs = (ct, padding.PKCS1v15(),)
+
+ elif self.pkalg == PubKeyAlgorithm.ECDH:
+ decrypter = pk
+ decargs = ()
+
+ else:
+ raise NotImplementedError(self.pkalg)
+
+ m = bytearray(self.ct.decrypt(decrypter, *decargs))
+
+ """
+ The value "m" in the above formulas is derived from the session key
+ as follows. First, the session key is prefixed with a one-octet
+ algorithm identifier that specifies the symmetric encryption
+ algorithm used to encrypt the following Symmetrically Encrypted Data
+ Packet. Then a two-octet checksum is appended, which is equal to the
+ sum of the preceding session key octets, not including the algorithm
+ identifier, modulo 65536. This value is then encoded as described in
+ PKCS#1 block encoding EME-PKCS1-v1_5 in Section 7.2.1 of [RFC3447] to
+ form the "m" value used in the formulas above. See Section 13.1 of
+ this document for notes on OpenPGP's use of PKCS#1.
+ """
+
+ symalg = SymmetricKeyAlgorithm(m[0])
+ del m[0]
+
+ symkey = m[:symalg.key_size // 8]
+ del m[:symalg.key_size // 8]
+
+ checksum = self.bytes_to_int(m[:2])
+ del m[:2]
+
+ if not sum(symkey) % 65536 == checksum: # pragma: no cover
+ raise PGPDecryptionError("{:s} decryption failed".format(self.pkalg.name))
+
+ return (symalg, symkey)
+
+ def encrypt_sk(self, pk, symalg, symkey):
+ m = bytearray(self.int_to_bytes(symalg) + symkey)
+ m += self.int_to_bytes(sum(bytearray(symkey)) % 65536, 2)
+
+ if self.pkalg == PubKeyAlgorithm.RSAEncryptOrSign:
+ encrypter = pk.keymaterial.__pubkey__().encrypt
+ encargs = (bytes(m), padding.PKCS1v15(),)
+
+ elif self.pkalg == PubKeyAlgorithm.ECDH:
+ encrypter = pk
+ encargs = (bytes(m),)
+
+ else:
+ raise NotImplementedError(self.pkalg)
+
+ self.ct = self.ct.encrypt(encrypter, *encargs)
+ self.update_hlen()
+
+ def parse(self, packet):
+ super(PKESessionKeyV3, self).parse(packet)
+ self.encrypter = packet[:8]
+ del packet[:8]
+
+ self.pkalg = packet[0]
+ del packet[0]
+
+ if self.ct is not None:
+ self.ct.parse(packet)
+
+ else: # pragma: no cover
+ del packet[:(self.header.length - 18)]
+
+
+class Signature(VersionedPacket):
+ __typeid__ = 0x02
+ __ver__ = 0
+
+
+class SignatureV4(Signature):
+ """
+ 5.2.3. Version 4 Signature Packet Format
+
+ The body of a version 4 Signature packet contains:
+
+ - One-octet version number (4).
+
+ - One-octet signature type.
+
+ - One-octet public-key algorithm.
+
+ - One-octet hash algorithm.
+
+ - Two-octet scalar octet count for following hashed subpacket data.
+ Note that this is the length in octets of all of the hashed
+ subpackets; a pointer incremented by this number will skip over
+ the hashed subpackets.
+
+ - Hashed subpacket data set (zero or more subpackets).
+
+ - Two-octet scalar octet count for the following unhashed subpacket
+ data. Note that this is the length in octets of all of the
+ unhashed subpackets; a pointer incremented by this number will
+ skip over the unhashed subpackets.
+
+ - Unhashed subpacket data set (zero or more subpackets).
+
+ - Two-octet field holding the left 16 bits of the signed hash
+ value.
+
+ - One or more multiprecision integers comprising the signature.
+ This portion is algorithm specific, as described above.
+
+ The concatenation of the data being signed and the signature data
+ from the version number through the hashed subpacket data (inclusive)
+ is hashed. The resulting hash value is what is signed. The left 16
+ bits of the hash are included in the Signature packet to provide a
+ quick test to reject some invalid signatures.
+
+ There are two fields consisting of Signature subpackets. The first
+ field is hashed with the rest of the signature data, while the second
+ is unhashed. The second set of subpackets is not cryptographically
+ protected by the signature and should include only advisory
+ information.
+
+ The algorithms for converting the hash function result to a signature
+ are described in a section below.
+ """
+ __ver__ = 4
+
+ @sdproperty
+ def sigtype(self):
+ return self._sigtype
+
+ @sigtype.register(int)
+ @sigtype.register(SignatureType)
+ def sigtype_int(self, val):
+ self._sigtype = SignatureType(val)
+
+ @sdproperty
+ def pubalg(self):
+ return self._pubalg
+
+ @pubalg.register(int)
+ @pubalg.register(PubKeyAlgorithm)
+ def pubalg_int(self, val):
+ self._pubalg = PubKeyAlgorithm(val)
+
+ sigs = {PubKeyAlgorithm.RSAEncryptOrSign: RSASignature,
+ PubKeyAlgorithm.RSAEncrypt: RSASignature,
+ PubKeyAlgorithm.RSASign: RSASignature,
+ PubKeyAlgorithm.DSA: DSASignature,
+ PubKeyAlgorithm.ECDSA: ECDSASignature, }
+
+ if self.pubalg in sigs:
+ self.signature = sigs[self.pubalg]()
+
+ @sdproperty
+ def halg(self):
+ return self._halg
+
+ @halg.register(int)
+ @halg.register(HashAlgorithm)
+ def halg_int(self, val):
+ try:
+ self._halg = HashAlgorithm(val)
+
+ except ValueError: # pragma: no cover
+ self._halg = val
+
+ @property
+ def signature(self):
+ return self._signature
+
+ @signature.setter
+ def signature(self, val):
+ self._signature = val
+
+ @property
+ def signer(self):
+ return self.subpackets['Issuer'][-1].issuer
+
+ def __init__(self):
+ super(Signature, self).__init__()
+ self._sigtype = None
+ self._pubalg = None
+ self._halg = None
+ self.subpackets = SubPackets()
+ self.hash2 = bytearray(2)
+ self.signature = None
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(Signature, self).__bytearray__()
+ _bytes += self.int_to_bytes(self.sigtype)
+ _bytes += self.int_to_bytes(self.pubalg)
+ _bytes += self.int_to_bytes(self.halg)
+ _bytes += self.subpackets.__bytearray__()
+ _bytes += self.hash2
+ _bytes += self.signature.__bytearray__()
+
+ return _bytes
+
+ def __copy__(self):
+ spkt = SignatureV4()
+ spkt.header = copy.copy(self.header)
+ spkt._sigtype = self._sigtype
+ spkt._pubalg = self._pubalg
+ spkt._halg = self._halg
+
+ spkt.subpackets = copy.copy(self.subpackets)
+ spkt.hash2 = copy.copy(self.hash2)
+ spkt.signature = copy.copy(self.signature)
+
+ return spkt
+
+ def update_hlen(self):
+ self.subpackets.update_hlen()
+ super(SignatureV4, self).update_hlen()
+
+ def parse(self, packet):
+ super(Signature, self).parse(packet)
+ self.sigtype = packet[0]
+ del packet[0]
+
+ self.pubalg = packet[0]
+ del packet[0]
+
+ self.halg = packet[0]
+ del packet[0]
+
+ self.subpackets.parse(packet)
+
+ self.hash2 = packet[:2]
+ del packet[:2]
+
+ self.signature.parse(packet)
+
+
+class SKESessionKey(VersionedPacket):
+ __typeid__ = 0x03
+ __ver__ = 0
+
+ @abc.abstractmethod
+ def decrypt_sk(self, passphrase):
+ raise NotImplementedError()
+
+ @abc.abstractmethod
+ def encrypt_sk(self, passphrase, sk):
+ raise NotImplementedError()
+
+
+class SKESessionKeyV4(SKESessionKey):
+ """
+ 5.3. Symmetric-Key Encrypted Session Key Packets (Tag 3)
+
+ The Symmetric-Key Encrypted Session Key packet holds the
+ symmetric-key encryption of a session key used to encrypt a message.
+ Zero or more Public-Key Encrypted Session Key packets and/or
+ Symmetric-Key Encrypted Session Key packets may precede a
+ Symmetrically Encrypted Data packet that holds an encrypted message.
+ The message is encrypted with a session key, and the session key is
+ itself encrypted and stored in the Encrypted Session Key packet or
+ the Symmetric-Key Encrypted Session Key packet.
+
+ If the Symmetrically Encrypted Data packet is preceded by one or
+ more Symmetric-Key Encrypted Session Key packets, each specifies a
+ passphrase that may be used to decrypt the message. This allows a
+ message to be encrypted to a number of public keys, and also to one
+ or more passphrases. This packet type is new and is not generated
+ by PGP 2.x or PGP 5.0.
+
+ The body of this packet consists of:
+
+ - A one-octet version number. The only currently defined version
+ is 4.
+
+ - A one-octet number describing the symmetric algorithm used.
+
+ - A string-to-key (S2K) specifier, length as defined above.
+
+ - Optionally, the encrypted session key itself, which is decrypted
+ with the string-to-key object.
+
+ If the encrypted session key is not present (which can be detected
+ on the basis of packet length and S2K specifier size), then the S2K
+ algorithm applied to the passphrase produces the session key for
+ decrypting the file, using the symmetric cipher algorithm from the
+ Symmetric-Key Encrypted Session Key packet.
+
+ If the encrypted session key is present, the result of applying the
+ S2K algorithm to the passphrase is used to decrypt just that
+ encrypted session key field, using CFB mode with an IV of all zeros.
+ The decryption result consists of a one-octet algorithm identifier
+ that specifies the symmetric-key encryption algorithm used to
+ encrypt the following Symmetrically Encrypted Data packet, followed
+ by the session key octets themselves.
+
+ Note: because an all-zero IV is used for this decryption, the S2K
+ specifier MUST use a salt value, either a Salted S2K or an
+ Iterated-Salted S2K. The salt value will ensure that the decryption
+ key is not repeated even if the passphrase is reused.
+ """
+ __ver__ = 4
+
+ @property
+ def symalg(self):
+ return self.s2k.encalg
+
+ def __init__(self):
+ super(SKESessionKeyV4, self).__init__()
+ self.s2k = String2Key()
+ self.ct = bytearray()
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(SKESessionKeyV4, self).__bytearray__()
+ _bytes += self.s2k.__bytearray__()[1:]
+ _bytes += self.ct
+ return _bytes
+
+ def __copy__(self):
+ sk = self.__class__()
+ sk.header = copy.copy(self.header)
+ sk.s2k = copy.copy(self.s2k)
+ sk.ct = self.ct[:]
+
+ return sk
+
+ def parse(self, packet):
+ super(SKESessionKeyV4, self).parse(packet)
+ # prepend a valid usage identifier so this parses correctly
+ packet.insert(0, 255)
+ self.s2k.parse(packet, iv=False)
+
+ ctend = self.header.length - len(self.s2k)
+ self.ct = packet[:ctend]
+ del packet[:ctend]
+
+ def decrypt_sk(self, passphrase):
+ # derive the first session key from our passphrase
+ sk = self.s2k.derive_key(passphrase)
+ del passphrase
+
+ # if there is no ciphertext, then the first session key is the session key being used
+ if len(self.ct) == 0:
+ return self.symalg, sk
+
+ # otherwise, we now need to decrypt the encrypted session key
+ m = bytearray(_decrypt(bytes(self.ct), sk, self.symalg))
+ del sk
+
+ symalg = SymmetricKeyAlgorithm(m[0])
+ del m[0]
+
+ return symalg, bytes(m)
+
+ def encrypt_sk(self, passphrase, sk):
+ # generate the salt and derive the key to encrypt sk with from it
+ self.s2k.salt = bytearray(os.urandom(8))
+ esk = self.s2k.derive_key(passphrase)
+ del passphrase
+
+ self.ct = _encrypt(self.int_to_bytes(self.symalg) + sk, esk, self.symalg)
+
+ # update header length and return sk
+ self.update_hlen()
+
+
+class OnePassSignature(VersionedPacket):
+ __typeid__ = 0x04
+ __ver__ = 0
+
+
+class OnePassSignatureV3(OnePassSignature):
+ """
+ 5.4. One-Pass Signature Packets (Tag 4)
+
+ The One-Pass Signature packet precedes the signed data and contains
+ enough information to allow the receiver to begin calculating any
+ hashes needed to verify the signature. It allows the Signature
+ packet to be placed at the end of the message, so that the signer
+ can compute the entire signed message in one pass.
+
+ A One-Pass Signature does not interoperate with PGP 2.6.x or
+ earlier.
+
+ The body of this packet consists of:
+
+ - A one-octet version number. The current version is 3.
+
+ - A one-octet signature type. Signature types are described in
+ Section 5.2.1.
+
+ - A one-octet number describing the hash algorithm used.
+
+ - A one-octet number describing the public-key algorithm used.
+
+ - An eight-octet number holding the Key ID of the signing key.
+
+ - A one-octet number holding a flag showing whether the signature
+ is nested. A zero value indicates that the next packet is
+ another One-Pass Signature packet that describes another
+ signature to be applied to the same message data.
+
+ Note that if a message contains more than one one-pass signature,
+ then the Signature packets bracket the message; that is, the first
+ Signature packet after the message corresponds to the last one-pass
+ packet and the final Signature packet corresponds to the first
+ one-pass packet.
+ """
+ __ver__ = 3
+
+ @sdproperty
+ def sigtype(self):
+ return self._sigtype
+
+ @sigtype.register(int)
+ @sigtype.register(SignatureType)
+ def sigtype_int(self, val):
+ self._sigtype = SignatureType(val)
+
+ @sdproperty
+ def pubalg(self):
+ return self._pubalg
+
+ @pubalg.register(int)
+ @pubalg.register(PubKeyAlgorithm)
+ def pubalg_int(self, val):
+ self._pubalg = PubKeyAlgorithm(val)
+ if self._pubalg in [PubKeyAlgorithm.RSAEncryptOrSign, PubKeyAlgorithm.RSAEncrypt, PubKeyAlgorithm.RSASign]:
+ self.signature = RSASignature()
+
+ elif self._pubalg == PubKeyAlgorithm.DSA:
+ self.signature = DSASignature()
+
+ @sdproperty
+ def halg(self):
+ return self._halg
+
+ @halg.register(int)
+ @halg.register(HashAlgorithm)
+ def halg_int(self, val):
+ try:
+ self._halg = HashAlgorithm(val)
+
+ except ValueError: # pragma: no cover
+ self._halg = val
+
+ @sdproperty
+ def signer(self):
+ return self._signer
+
+ @signer.register(str)
+ @signer.register(six.text_type)
+ def signer_str(self, val):
+ self._signer = val
+
+ @signer.register(bytearray)
+ def signer_bin(self, val):
+ self._signer = binascii.hexlify(val).upper().decode('latin-1')
+
+ def __init__(self):
+ super(OnePassSignatureV3, self).__init__()
+ self._sigtype = None
+ self._halg = None
+ self._pubalg = None
+ self._signer = b'\x00' * 8
+ self.nested = False
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(OnePassSignatureV3, self).__bytearray__()
+ _bytes += bytearray([self.sigtype])
+ _bytes += bytearray([self.halg])
+ _bytes += bytearray([self.pubalg])
+ _bytes += binascii.unhexlify(six.b(self.signer))
+ _bytes += bytearray([int(self.nested)])
+ return _bytes
+
+ def parse(self, packet):
+ super(OnePassSignatureV3, self).parse(packet)
+ self.sigtype = packet[0]
+ del packet[0]
+
+ self.halg = packet[0]
+ del packet[0]
+
+ self.pubalg = packet[0]
+ del packet[0]
+
+ self.signer = packet[:8]
+ del packet[:8]
+
+ self.nested = (packet[0] == 1)
+ del packet[0]
+
+
+class PrivKey(VersionedPacket, Primary, Private):
+ __typeid__ = 0x05
+ __ver__ = 0
+
+
+class PubKey(VersionedPacket, Primary, Public):
+ __typeid__ = 0x06
+ __ver__ = 0
+
+ @abc.abstractproperty
+ def fingerprint(self):
+ """compute and return the fingerprint of the key"""
+
+
+class PubKeyV4(PubKey):
+ __ver__ = 4
+
+ @sdproperty
+ def created(self):
+ return self._created
+
+ @created.register(datetime)
+ def created_datetime(self, val):
+ self._created = val
+
+ @created.register(int)
+ def created_int(self, val):
+ self.created = datetime.utcfromtimestamp(val)
+
+ @created.register(bytes)
+ @created.register(bytearray)
+ def created_bin(self, val):
+ self.created = self.bytes_to_int(val)
+
+ @sdproperty
+ def pkalg(self):
+ return self._pkalg
+
+ @pkalg.register(int)
+ @pkalg.register(PubKeyAlgorithm)
+ def pkalg_int(self, val):
+ self._pkalg = PubKeyAlgorithm(val)
+
+ _c = {
+ # True means public
+ (True, PubKeyAlgorithm.RSAEncryptOrSign): RSAPub,
+ (True, PubKeyAlgorithm.RSAEncrypt): RSAPub,
+ (True, PubKeyAlgorithm.RSASign): RSAPub,
+ (True, PubKeyAlgorithm.DSA): DSAPub,
+ (True, PubKeyAlgorithm.ElGamal): ElGPub,
+ (True, PubKeyAlgorithm.FormerlyElGamalEncryptOrSign): ElGPub,
+ (True, PubKeyAlgorithm.ECDSA): ECDSAPub,
+ (True, PubKeyAlgorithm.ECDH): ECDHPub,
+ # False means private
+ (False, PubKeyAlgorithm.RSAEncryptOrSign): RSAPriv,
+ (False, PubKeyAlgorithm.RSAEncrypt): RSAPriv,
+ (False, PubKeyAlgorithm.RSASign): RSAPriv,
+ (False, PubKeyAlgorithm.DSA): DSAPriv,
+ (False, PubKeyAlgorithm.ElGamal): ElGPriv,
+ (False, PubKeyAlgorithm.FormerlyElGamalEncryptOrSign): ElGPriv,
+ (False, PubKeyAlgorithm.ECDSA): ECDSAPriv,
+ (False, PubKeyAlgorithm.ECDH): ECDHPriv,
+ }
+
+ k = (self.public, self.pkalg)
+ km = _c.get(k, None)
+
+ self.keymaterial = (km or (OpaquePubKey if self.public else OpaquePrivKey))()
+
+ # km = _c.get(k, None)
+ # self.keymaterial = km() if km is not None else km
+
+ @property
+ def public(self):
+ return isinstance(self, PubKey) and not isinstance(self, PrivKey)
+
+ @property
+ def fingerprint(self):
+ # A V4 fingerprint is the 160-bit SHA-1 hash of the octet 0x99, followed by the two-octet packet length,
+ # followed by the entire Public-Key packet starting with the version field. The Key ID is the
+ # low-order 64 bits of the fingerprint.
+ fp = hashlib.new('sha1')
+
+ plen = self.keymaterial.publen()
+ bcde_len = self.int_to_bytes(6 + plen, 2)
+
+ # a.1) 0x99 (1 octet)
+ # a.2) high-order length octet
+ # a.3) low-order length octet
+ fp.update(b'\x99' + bcde_len[:1] + bcde_len[-1:])
+ # b) version number = 4 (1 octet);
+ fp.update(b'\x04')
+ # c) timestamp of key creation (4 octets);
+ fp.update(self.int_to_bytes(calendar.timegm(self.created.timetuple()), 4))
+ # d) algorithm (1 octet): 17 = DSA (example);
+ fp.update(self.int_to_bytes(self.pkalg))
+ # e) Algorithm-specific fields.
+ fp.update(self.keymaterial.__bytearray__()[:plen])
+
+ # and return the digest
+ return Fingerprint(fp.hexdigest().upper())
+
+ def __init__(self):
+ super(PubKeyV4, self).__init__()
+ self.created = datetime.utcnow()
+ self.pkalg = 0
+ self.keymaterial = None
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(PubKeyV4, self).__bytearray__()
+ _bytes += self.int_to_bytes(calendar.timegm(self.created.timetuple()), 4)
+ _bytes += self.int_to_bytes(self.pkalg)
+ _bytes += self.keymaterial.__bytearray__()
+ return _bytes
+
+ def __copy__(self):
+ pk = self.__class__()
+ pk.header = copy.copy(self.header)
+ pk.created = self.created
+ pk.pkalg = self.pkalg
+ pk.keymaterial = copy.copy(self.keymaterial)
+
+ return pk
+
+ def verify(self, subj, sigbytes, hash_alg):
+ return self.keymaterial.verify(subj, sigbytes, hash_alg)
+
+ def parse(self, packet):
+ super(PubKeyV4, self).parse(packet)
+
+ self.created = packet[:4]
+ del packet[:4]
+
+ self.pkalg = packet[0]
+ del packet[0]
+
+ # bound keymaterial to the remaining length of the packet
+ pend = self.header.length - 6
+ self.keymaterial.parse(packet[:pend])
+ del packet[:pend]
+
+
+class PrivKeyV4(PrivKey, PubKeyV4):
+ __ver__ = 4
+
+ @classmethod
+ def new(cls, key_algorithm, key_size):
+ # build a key packet
+ pk = PrivKeyV4()
+ pk.pkalg = key_algorithm
+ if pk.keymaterial is None:
+ raise NotImplementedError(key_algorithm)
+ pk.keymaterial._generate(key_size)
+ pk.update_hlen()
+ return pk
+
+ def pubkey(self):
+ # return a copy of ourselves, but just the public half
+ pk = PubKeyV4() if not isinstance(self, PrivSubKeyV4) else PubSubKeyV4()
+ pk.created = self.created
+ pk.pkalg = self.pkalg
+
+ # copy over MPIs
+ for pm in self.keymaterial.__pubfields__:
+ setattr(pk.keymaterial, pm, copy.copy(getattr(self.keymaterial, pm)))
+
+ if self.pkalg == PubKeyAlgorithm.ECDSA:
+ pk.keymaterial.oid = self.keymaterial.oid
+
+ if self.pkalg == PubKeyAlgorithm.ECDH:
+ pk.keymaterial.oid = self.keymaterial.oid
+ pk.keymaterial.kdf = copy.copy(self.keymaterial.kdf)
+
+ pk.update_hlen()
+ return pk
+
+ @property
+ def protected(self):
+ return bool(self.keymaterial.s2k)
+
+ @property
+ def unlocked(self):
+ if self.protected:
+ return 0 not in list(self.keymaterial)
+ return True # pragma: no cover
+
+ def protect(self, passphrase, enc_alg, hash_alg):
+ self.keymaterial.encrypt_keyblob(passphrase, enc_alg, hash_alg)
+ del passphrase
+ self.update_hlen()
+
+ def unprotect(self, passphrase):
+ self.keymaterial.decrypt_keyblob(passphrase)
+ del passphrase
+
+ def sign(self, sigdata, hash_alg):
+ return self.keymaterial.sign(sigdata, hash_alg)
+
+
+class PrivSubKey(VersionedPacket, Sub, Private):
+ __typeid__ = 0x07
+ __ver__ = 0
+
+
+class PrivSubKeyV4(PrivSubKey, PrivKeyV4):
+ __ver__ = 4
+
+
+class CompressedData(Packet):
+ """
+ 5.6. Compressed Data Packet (Tag 8)
+
+ The Compressed Data packet contains compressed data. Typically, this
+ packet is found as the contents of an encrypted packet, or following
+ a Signature or One-Pass Signature packet, and contains a literal data
+ packet.
+
+ The body of this packet consists of:
+
+ - One octet that gives the algorithm used to compress the packet.
+
+ - Compressed data, which makes up the remainder of the packet.
+
+ A Compressed Data Packet's body contains an block that compresses
+ some set of packets. See section "Packet Composition" for details on
+ how messages are formed.
+
+ ZIP-compressed packets are compressed with raw RFC 1951 [RFC1951]
+ DEFLATE blocks. Note that PGP V2.6 uses 13 bits of compression. If
+ an implementation uses more bits of compression, PGP V2.6 cannot
+ decompress it.
+
+ ZLIB-compressed packets are compressed with RFC 1950 [RFC1950] ZLIB-
+ style blocks.
+
+ BZip2-compressed packets are compressed using the BZip2 [BZ2]
+ algorithm.
+ """
+ __typeid__ = 0x08
+
+ @sdproperty
+ def calg(self):
+ return self._calg
+
+ @calg.register(int)
+ @calg.register(CompressionAlgorithm)
+ def calg_int(self, val):
+ self._calg = CompressionAlgorithm(val)
+
+ def __init__(self):
+ super(CompressedData, self).__init__()
+ self._calg = None
+ self.packets = []
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(CompressedData, self).__bytearray__()
+ _bytes += bytearray([self.calg])
+
+ _pb = bytearray()
+ for pkt in self.packets:
+ _pb += pkt.__bytearray__()
+ _bytes += self.calg.compress(bytes(_pb))
+
+ return _bytes
+
+ def parse(self, packet):
+ super(CompressedData, self).parse(packet)
+ self.calg = packet[0]
+ del packet[0]
+
+ cdata = bytearray(self.calg.decompress(packet[:self.header.length - 1]))
+ del packet[:self.header.length - 1]
+
+ while len(cdata) > 0:
+ self.packets.append(Packet(cdata))
+
+
+class SKEData(Packet):
+ """
+ 5.7. Symmetrically Encrypted Data Packet (Tag 9)
+
+ The Symmetrically Encrypted Data packet contains data encrypted with
+ a symmetric-key algorithm. When it has been decrypted, it contains
+ other packets (usually a literal data packet or compressed data
+ packet, but in theory other Symmetrically Encrypted Data packets or
+ sequences of packets that form whole OpenPGP messages).
+
+ The body of this packet consists of:
+
+ - Encrypted data, the output of the selected symmetric-key cipher
+ operating in OpenPGP's variant of Cipher Feedback (CFB) mode.
+
+ The symmetric cipher used may be specified in a Public-Key or
+ Symmetric-Key Encrypted Session Key packet that precedes the
+ Symmetrically Encrypted Data packet. In that case, the cipher
+ algorithm octet is prefixed to the session key before it is
+ encrypted. If no packets of these types precede the encrypted data,
+ the IDEA algorithm is used with the session key calculated as the MD5
+ hash of the passphrase, though this use is deprecated.
+
+ The data is encrypted in CFB mode, with a CFB shift size equal to the
+ cipher's block size. The Initial Vector (IV) is specified as all
+ zeros. Instead of using an IV, OpenPGP prefixes a string of length
+ equal to the block size of the cipher plus two to the data before it
+ is encrypted. The first block-size octets (for example, 8 octets for
+ a 64-bit block length) are random, and the following two octets are
+ copies of the last two octets of the IV. For example, in an 8-octet
+ block, octet 9 is a repeat of octet 7, and octet 10 is a repeat of
+ octet 8. In a cipher of length 16, octet 17 is a repeat of octet 15
+ and octet 18 is a repeat of octet 16. As a pedantic clarification,
+ in both these examples, we consider the first octet to be numbered 1.
+
+ After encrypting the first block-size-plus-two octets, the CFB state
+ is resynchronized. The last block-size octets of ciphertext are
+ passed through the cipher and the block boundary is reset.
+
+ The repetition of 16 bits in the random data prefixed to the message
+ allows the receiver to immediately check whether the session key is
+ incorrect. See the "Security Considerations" section for hints on
+ the proper use of this "quick check".
+ """
+ __typeid__ = 0x09
+
+ def __init__(self):
+ super(SKEData, self).__init__()
+ self.ct = bytearray()
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(SKEData, self).__bytearray__()
+ _bytes += self.ct
+ return _bytes
+
+ def __copy__(self):
+ skd = self.__class__()
+ skd.ct = self.ct[:]
+ return skd
+
+ def parse(self, packet):
+ super(SKEData, self).parse(packet)
+ self.ct = packet[:self.header.length]
+ del packet[:self.header.length]
+
+ def decrypt(self, key, alg): # pragma: no cover
+ pt = _decrypt(bytes(self.ct), bytes(key), alg)
+
+ iv = bytes(pt[:alg.block_size // 8])
+ del pt[:alg.block_size // 8]
+
+ ivl2 = bytes(pt[:2])
+ del pt[:2]
+
+ if not constant_time.bytes_eq(iv[-2:], ivl2):
+ raise PGPDecryptionError("Decryption failed")
+
+ return pt
+
+
+class Marker(Packet):
+ __typeid__ = 0x0a
+
+ def __init__(self):
+ super(Marker, self).__init__()
+ self.data = b'PGP'
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(Marker, self).__bytearray__()
+ _bytes += self.data
+ return _bytes
+
+ def parse(self, packet):
+ super(Marker, self).parse(packet)
+ self.data = packet[:self.header.length]
+ del packet[:self.header.length]
+
+
+class LiteralData(Packet):
+ """
+ 5.9. Literal Data Packet (Tag 11)
+
+ A Literal Data packet contains the body of a message; data that is
+ not to be further interpreted.
+
+ The body of this packet consists of:
+
+ - A one-octet field that describes how the data is formatted.
+
+ If it is a 'b' (0x62), then the Literal packet contains binary data.
+ If it is a 't' (0x74), then it contains text data, and thus may need
+ line ends converted to local form, or other text-mode changes. The
+ tag 'u' (0x75) means the same as 't', but also indicates that
+ implementation believes that the literal data contains UTF-8 text.
+
+ Early versions of PGP also defined a value of 'l' as a 'local' mode
+ for machine-local conversions. RFC 1991 [RFC1991] incorrectly stated
+ this local mode flag as '1' (ASCII numeral one). Both of these local
+ modes are deprecated.
+
+ - File name as a string (one-octet length, followed by a file
+ name). This may be a zero-length string. Commonly, if the
+ source of the encrypted data is a file, this will be the name of
+ the encrypted file. An implementation MAY consider the file name
+ in the Literal packet to be a more authoritative name than the
+ actual file name.
+
+ If the special name "_CONSOLE" is used, the message is considered to
+ be "for your eyes only". This advises that the message data is
+ unusually sensitive, and the receiving program should process it more
+ carefully, perhaps avoiding storing the received data to disk, for
+ example.
+
+ - A four-octet number that indicates a date associated with the
+ literal data. Commonly, the date might be the modification date
+ of a file, or the time the packet was created, or a zero that
+ indicates no specific time.
+
+ - The remainder of the packet is literal data.
+
+ Text data is stored with <CR><LF> text endings (i.e., network-
+ normal line endings). These should be converted to native line
+ endings by the receiving software.
+ """
+ __typeid__ = 0x0B
+
+ @sdproperty
+ def mtime(self):
+ return self._mtime
+
+ @mtime.register(datetime)
+ def mtime_datetime(self, val):
+ self._mtime = val
+
+ @mtime.register(int)
+ def mtime_int(self, val):
+ self.mtime = datetime.utcfromtimestamp(val)
+
+ @mtime.register(bytes)
+ @mtime.register(bytearray)
+ def mtime_bin(self, val):
+ self.mtime = self.bytes_to_int(val)
+
+ @property
+ def contents(self):
+ if self.format == 't':
+ return self._contents.decode('latin-1')
+
+ if self.format == 'u':
+ return self._contents.decode('utf-8')
+
+ return self._contents
+
+ def __init__(self):
+ super(LiteralData, self).__init__()
+ self.format = 'b'
+ self.filename = ''
+ self.mtime = datetime.utcnow()
+ self._contents = bytearray()
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(LiteralData, self).__bytearray__()
+ _bytes += self.format.encode('latin-1')
+ _bytes += bytearray([len(self.filename)])
+ _bytes += self.filename.encode('latin-1')
+ _bytes += self.int_to_bytes(calendar.timegm(self.mtime.timetuple()), 4)
+ _bytes += self._contents
+ return _bytes
+
+ def __copy__(self):
+ pkt = LiteralData()
+ pkt.header = copy.copy(self.header)
+ pkt.format = self.format
+ pkt.filename = self.filename
+ pkt.mtime = self.mtime
+ pkt._contents = self._contents[:]
+
+ return pkt
+
+ def parse(self, packet):
+ super(LiteralData, self).parse(packet)
+ self.format = chr(packet[0])
+ del packet[0]
+
+ fnl = packet[0]
+ del packet[0]
+
+ self.filename = packet[:fnl].decode()
+ del packet[:fnl]
+
+ self.mtime = packet[:4]
+ del packet[:4]
+
+ self._contents = packet[:self.header.length - (6 + fnl)]
+ del packet[:self.header.length - (6 + fnl)]
+
+
+class Trust(Packet):
+ """
+ 5.10. Trust Packet (Tag 12)
+
+ The Trust packet is used only within keyrings and is not normally
+ exported. Trust packets contain data that record the user's
+ specifications of which key holders are trustworthy introducers,
+ along with other information that implementing software uses for
+ trust information. The format of Trust packets is defined by a given
+ implementation.
+
+ Trust packets SHOULD NOT be emitted to output streams that are
+ transferred to other users, and they SHOULD be ignored on any input
+ other than local keyring files.
+ """
+ __typeid__ = 0x0C
+
+ @sdproperty
+ def trustlevel(self):
+ return self._trustlevel
+
+ @trustlevel.register(int)
+ @trustlevel.register(TrustLevel)
+ def trustlevel_int(self, val):
+ self._trustlevel = TrustLevel(val & 0x0F)
+
+ @sdproperty
+ def trustflags(self):
+ return self._trustflags
+
+ @trustflags.register(list)
+ def trustflags_list(self, val):
+ self._trustflags = val
+
+ @trustflags.register(int)
+ def trustflags_int(self, val):
+ self._trustflags = TrustFlags & val
+
+ def __init__(self):
+ super(Trust, self).__init__()
+ self.trustlevel = TrustLevel.Unknown
+ self.trustflags = []
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(Trust, self).__bytearray__()
+ _bytes += self.int_to_bytes(self.trustlevel + sum(self.trustflags), 2)
+ return _bytes
+
+ def parse(self, packet):
+ super(Trust, self).parse(packet)
+ # self.trustlevel = packet[0] & 0x1f
+ t = self.bytes_to_int(packet[:2])
+ del packet[:2]
+
+ self.trustlevel = t
+ self.trustflags = t
+
+
+class UserID(Packet):
+ """
+ 5.11. User ID Packet (Tag 13)
+
+ A User ID packet consists of UTF-8 text that is intended to represent
+ the name and email address of the key holder. By convention, it
+ includes an RFC 2822 [RFC2822] mail name-addr, but there are no
+ restrictions on its content. The packet length in the header
+ specifies the length of the User ID.
+ """
+ __typeid__ = 0x0D
+
+ def __init__(self):
+ super(UserID, self).__init__()
+ self.name = ""
+ self.comment = ""
+ self.email = ""
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(UserID, self).__bytearray__()
+ _bytes += self.text_to_bytes(self.name)
+ if self.comment:
+ _bytes += b' (' + self.text_to_bytes(self.comment) + b')'
+
+ if self.email:
+ _bytes += b' <' + self.text_to_bytes(self.email) + b'>'
+
+ return _bytes
+
+ def __copy__(self):
+ uid = UserID()
+ uid.header = copy.copy(self.header)
+ uid.name = self.name
+ uid.comment = self.comment
+ uid.email = self.email
+ return uid
+
+ def parse(self, packet):
+ super(UserID, self).parse(packet)
+
+ uid_text = packet[:self.header.length].decode('latin-1')
+ del packet[:self.header.length]
+
+ # came across a UID packet with no payload. If that happens, don't bother trying to parse anything!
+ if self.header.length > 0:
+ uid = re.match(r"""^
+ # name should always match something
+ (?P<name>.+?)
+ # comment *optionally* matches text in parens following name
+ # this should never come after email and must be followed immediately by
+ # either the email field, or the end of the packet.
+ (\ \((?P<comment>.+?)\)(?=(\ <|$)))?
+ # email *optionally* matches text in angle brackets following name or comment
+ # this should never come before a comment, if comment exists,
+ # but can immediately follow name if comment does not exist
+ (\ <(?P<email>.+)>)?
+ $
+ """, uid_text, flags=re.VERBOSE).groupdict()
+
+ self.name = uid['name']
+ self.comment = uid['comment'] or ""
+ self.email = uid['email'] or ""
+
+
+class PubSubKey(VersionedPacket, Sub, Public):
+ __typeid__ = 0x0E
+ __ver__ = 0
+
+
+class PubSubKeyV4(PubSubKey, PubKeyV4):
+ __ver__ = 4
+
+
+class UserAttribute(Packet):
+ """
+ 5.12. User Attribute Packet (Tag 17)
+
+ The User Attribute packet is a variation of the User ID packet. It
+ is capable of storing more types of data than the User ID packet,
+ which is limited to text. Like the User ID packet, a User Attribute
+ packet may be certified by the key owner ("self-signed") or any other
+ key owner who cares to certify it. Except as noted, a User Attribute
+ packet may be used anywhere that a User ID packet may be used.
+
+ While User Attribute packets are not a required part of the OpenPGP
+ standard, implementations SHOULD provide at least enough
+ compatibility to properly handle a certification signature on the
+ User Attribute packet. A simple way to do this is by treating the
+ User Attribute packet as a User ID packet with opaque contents, but
+ an implementation may use any method desired.
+
+ The User Attribute packet is made up of one or more attribute
+ subpackets. Each subpacket consists of a subpacket header and a
+ body. The header consists of:
+
+ - the subpacket length (1, 2, or 5 octets)
+
+ - the subpacket type (1 octet)
+
+ and is followed by the subpacket specific data.
+
+ The only currently defined subpacket type is 1, signifying an image.
+ An implementation SHOULD ignore any subpacket of a type that it does
+ not recognize. Subpacket types 100 through 110 are reserved for
+ private or experimental use.
+ """
+ __typeid__ = 0x11
+
+ @property
+ def image(self):
+ if 'Image' not in self.subpackets:
+ self.subpackets.addnew('Image')
+ return next(iter(self.subpackets['Image']))
+
+ def __init__(self):
+ super(UserAttribute, self).__init__()
+ self.subpackets = UserAttributeSubPackets()
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(UserAttribute, self).__bytearray__()
+ _bytes += self.subpackets.__bytearray__()
+ return _bytes
+
+ def parse(self, packet):
+ super(UserAttribute, self).parse(packet)
+
+ plen = len(packet)
+ while self.header.length > (plen - len(packet)):
+ self.subpackets.parse(packet)
+
+ def update_hlen(self):
+ self.subpackets.update_hlen()
+ super(UserAttribute, self).update_hlen()
+
+
+class IntegrityProtectedSKEData(VersionedPacket):
+ __typeid__ = 0x12
+ __ver__ = 0
+
+
+class IntegrityProtectedSKEDataV1(IntegrityProtectedSKEData):
+ """
+ 5.13. Sym. Encrypted Integrity Protected Data Packet (Tag 18)
+
+ The Symmetrically Encrypted Integrity Protected Data packet is a
+ variant of the Symmetrically Encrypted Data packet. It is a new
+ feature created for OpenPGP that addresses the problem of detecting a
+ modification to encrypted data. It is used in combination with a
+ Modification Detection Code packet.
+
+ There is a corresponding feature in the features Signature subpacket
+ that denotes that an implementation can properly use this packet
+ type. An implementation MUST support decrypting these packets and
+ SHOULD prefer generating them to the older Symmetrically Encrypted
+ Data packet when possible. Since this data packet protects against
+ modification attacks, this standard encourages its proliferation.
+ While blanket adoption of this data packet would create
+ interoperability problems, rapid adoption is nevertheless important.
+ An implementation SHOULD specifically denote support for this packet,
+ but it MAY infer it from other mechanisms.
+
+ For example, an implementation might infer from the use of a cipher
+ such as Advanced Encryption Standard (AES) or Twofish that a user
+ supports this feature. It might place in the unhashed portion of
+ another user's key signature a Features subpacket. It might also
+ present a user with an opportunity to regenerate their own self-
+ signature with a Features subpacket.
+
+ This packet contains data encrypted with a symmetric-key algorithm
+ and protected against modification by the SHA-1 hash algorithm. When
+ it has been decrypted, it will typically contain other packets (often
+ a Literal Data packet or Compressed Data packet). The last decrypted
+ packet in this packet's payload MUST be a Modification Detection Code
+ packet.
+
+ The body of this packet consists of:
+
+ - A one-octet version number. The only currently defined value is
+ 1.
+
+ - Encrypted data, the output of the selected symmetric-key cipher
+ operating in Cipher Feedback mode with shift amount equal to the
+ block size of the cipher (CFB-n where n is the block size).
+
+ The symmetric cipher used MUST be specified in a Public-Key or
+ Symmetric-Key Encrypted Session Key packet that precedes the
+ Symmetrically Encrypted Data packet. In either case, the cipher
+ algorithm octet is prefixed to the session key before it is
+ encrypted.
+
+ The data is encrypted in CFB mode, with a CFB shift size equal to the
+ cipher's block size. The Initial Vector (IV) is specified as all
+ zeros. Instead of using an IV, OpenPGP prefixes an octet string to
+ the data before it is encrypted. The length of the octet string
+ equals the block size of the cipher in octets, plus two. The first
+ octets in the group, of length equal to the block size of the cipher,
+ are random; the last two octets are each copies of their 2nd
+ preceding octet. For example, with a cipher whose block size is 128
+ bits or 16 octets, the prefix data will contain 16 random octets,
+ then two more octets, which are copies of the 15th and 16th octets,
+ respectively. Unlike the Symmetrically Encrypted Data Packet, no
+ special CFB resynchronization is done after encrypting this prefix
+ data. See "OpenPGP CFB Mode" below for more details.
+
+ The repetition of 16 bits in the random data prefixed to the message
+ allows the receiver to immediately check whether the session key is
+ incorrect.
+
+ The plaintext of the data to be encrypted is passed through the SHA-1
+ hash function, and the result of the hash is appended to the
+ plaintext in a Modification Detection Code packet. The input to the
+ hash function includes the prefix data described above; it includes
+ all of the plaintext, and then also includes two octets of values
+ 0xD3, 0x14. These represent the encoding of a Modification Detection
+ Code packet tag and length field of 20 octets.
+
+ The resulting hash value is stored in a Modification Detection Code
+ (MDC) packet, which MUST use the two octet encoding just given to
+ represent its tag and length field. The body of the MDC packet is
+ the 20-octet output of the SHA-1 hash.
+
+ The Modification Detection Code packet is appended to the plaintext
+ and encrypted along with the plaintext using the same CFB context.
+
+ During decryption, the plaintext data should be hashed with SHA-1,
+ including the prefix data as well as the packet tag and length field
+ of the Modification Detection Code packet. The body of the MDC
+ packet, upon decryption, is compared with the result of the SHA-1
+ hash.
+
+ Any failure of the MDC indicates that the message has been modified
+ and MUST be treated as a security problem. Failures include a
+ difference in the hash values, but also the absence of an MDC packet,
+ or an MDC packet in any position other than the end of the plaintext.
+ Any failure SHOULD be reported to the user.
+
+ Note: future designs of new versions of this packet should consider
+ rollback attacks since it will be possible for an attacker to change
+ the version back to 1.
+ """
+ __ver__ = 1
+
+ def __init__(self):
+ super(IntegrityProtectedSKEDataV1, self).__init__()
+ self.ct = bytearray()
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ _bytes += super(IntegrityProtectedSKEDataV1, self).__bytearray__()
+ _bytes += self.ct
+ return _bytes
+
+ def __copy__(self):
+ skd = self.__class__()
+ skd.ct = self.ct[:]
+ return skd
+
+ def parse(self, packet):
+ super(IntegrityProtectedSKEDataV1, self).parse(packet)
+ self.ct = packet[:self.header.length - 1]
+ del packet[:self.header.length - 1]
+
+ def encrypt(self, key, alg, data):
+ iv = alg.gen_iv()
+ data = iv + iv[-2:] + data
+
+ mdc = MDC()
+ mdc.mdc = binascii.hexlify(hashlib.new('SHA1', data + b'\xd3\x14').digest())
+ mdc.update_hlen()
+
+ data += mdc.__bytes__()
+ self.ct = _encrypt(data, key, alg)
+ self.update_hlen()
+
+ def decrypt(self, key, alg):
+ # iv, ivl2, pt = super(IntegrityProtectedSKEDataV1, self).decrypt(key, alg)
+ pt = _decrypt(bytes(self.ct), bytes(key), alg)
+
+ # do the MDC checks
+ _expected_mdcbytes = b'\xd3\x14' + hashlib.new('SHA1', pt[:-20]).digest()
+ if not constant_time.bytes_eq(bytes(pt[-22:]), _expected_mdcbytes):
+ raise PGPDecryptionError("Decryption failed") # pragma: no cover
+
+ iv = bytes(pt[:alg.block_size // 8])
+ del pt[:alg.block_size // 8]
+
+ ivl2 = bytes(pt[:2])
+ del pt[:2]
+
+ if not constant_time.bytes_eq(iv[-2:], ivl2):
+ raise PGPDecryptionError("Decryption failed") # pragma: no cover
+
+ return pt
+
+
+class MDC(Packet):
+ """
+ 5.14. Modification Detection Code Packet (Tag 19)
+
+ The Modification Detection Code packet contains a SHA-1 hash of
+ plaintext data, which is used to detect message modification. It is
+ only used with a Symmetrically Encrypted Integrity Protected Data
+ packet. The Modification Detection Code packet MUST be the last
+ packet in the plaintext data that is encrypted in the Symmetrically
+ Encrypted Integrity Protected Data packet, and MUST appear in no
+ other place.
+
+ A Modification Detection Code packet MUST have a length of 20 octets.
+ The body of this packet consists of:
+
+ - A 20-octet SHA-1 hash of the preceding plaintext data of the
+ Symmetrically Encrypted Integrity Protected Data packet,
+ including prefix data, the tag octet, and length octet of the
+ Modification Detection Code packet.
+
+ Note that the Modification Detection Code packet MUST always use a
+ new format encoding of the packet tag, and a one-octet encoding of
+ the packet length. The reason for this is that the hashing rules for
+ modification detection include a one-octet tag and one-octet length
+ in the data hash. While this is a bit restrictive, it reduces
+ complexity.
+ """
+ __typeid__ = 0x13
+
+ def __init__(self):
+ super(MDC, self).__init__()
+ self.mdc = ''
+
+ def __bytearray__(self):
+ return super(MDC, self).__bytearray__() + binascii.unhexlify(self.mdc)
+
+ def parse(self, packet):
+ super(MDC, self).parse(packet)
+ self.mdc = binascii.hexlify(packet[:20])
+ del packet[:20]
diff --git a/src/leap/mx/vendor/pgpy/packet/subpackets/__init__.py b/src/leap/mx/vendor/pgpy/packet/subpackets/__init__.py
new file mode 100644
index 0000000..b1e90a5
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/subpackets/__init__.py
@@ -0,0 +1,7 @@
+from .types import Signature as Signature
+from .types import UserAttribute as UserAttribute
+
+from .signature import * # NOQA
+from .userattribute import * # NOQA
+
+__all__ = ['Signature', 'UserAttribute']
diff --git a/src/leap/mx/vendor/pgpy/packet/subpackets/signature.py b/src/leap/mx/vendor/pgpy/packet/subpackets/signature.py
new file mode 100644
index 0000000..8993009
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/subpackets/signature.py
@@ -0,0 +1,887 @@
+""" signature.py
+
+Signature SubPackets
+"""
+import binascii
+import calendar
+
+from datetime import datetime
+from datetime import timedelta
+
+import six
+
+from .types import EmbeddedSignatureHeader
+from .types import Signature
+
+from ...constants import CompressionAlgorithm
+from ...constants import Features as _Features
+from ...constants import HashAlgorithm
+from ...constants import KeyFlags as _KeyFlags
+from ...constants import KeyServerPreferences as _KeyServerPreferences
+from ...constants import NotationDataFlags
+from ...constants import PubKeyAlgorithm
+from ...constants import RevocationKeyClass
+from ...constants import RevocationReason
+from ...constants import SymmetricKeyAlgorithm
+
+from ...decorators import sdproperty
+
+from ...types import Fingerprint
+
+
+__all__ = ['URI',
+ 'FlagList',
+ 'ByteFlag',
+ 'Boolean',
+ 'CreationTime',
+ 'SignatureExpirationTime',
+ 'ExportableCertification',
+ 'TrustSignature',
+ 'RegularExpression',
+ 'Revocable',
+ 'KeyExpirationTime',
+ 'PreferredSymmetricAlgorithms',
+ 'RevocationKey',
+ 'Issuer',
+ 'NotationData',
+ 'PreferredHashAlgorithms',
+ 'PreferredCompressionAlgorithms',
+ 'KeyServerPreferences',
+ 'PreferredKeyServer',
+ 'PrimaryUserID',
+ 'Policy',
+ 'KeyFlags',
+ 'SignersUserID',
+ 'ReasonForRevocation',
+ 'Features',
+ 'EmbeddedSignature']
+
+
+class URI(Signature):
+ @sdproperty
+ def uri(self):
+ return self._uri
+
+ @uri.register(str)
+ @uri.register(six.text_type)
+ def uri_str(self, val):
+ self._uri = val
+
+ @uri.register(bytearray)
+ def uri_bytearray(self, val):
+ self.uri = val.decode('latin-1')
+
+ def __init__(self):
+ super(URI, self).__init__()
+ self.uri = ""
+
+ def __bytearray__(self):
+ _bytes = super(URI, self).__bytearray__()
+ _bytes += self.uri.encode()
+ return _bytes
+
+ def parse(self, packet):
+ super(URI, self).parse(packet)
+ self.uri = packet[:(self.header.length - 1)]
+ del packet[:(self.header.length - 1)]
+
+
+class FlagList(Signature):
+ __flags__ = None
+
+ @sdproperty
+ def flags(self):
+ return self._flags
+
+ @flags.register(list)
+ @flags.register(tuple)
+ def flags_list(self, val):
+ self._flags = list(val)
+
+ @flags.register(int)
+ @flags.register(CompressionAlgorithm)
+ @flags.register(HashAlgorithm)
+ @flags.register(PubKeyAlgorithm)
+ @flags.register(SymmetricKeyAlgorithm)
+ def flags_int(self, val):
+ if self.__flags__ is None: # pragma: no cover
+ raise AttributeError("Error: __flags__ not set!")
+
+ self._flags.append(self.__flags__(val))
+
+ @flags.register(bytearray)
+ def flags_bytearray(self, val):
+ self.flags = self.bytes_to_int(val)
+
+ def __init__(self):
+ super(FlagList, self).__init__()
+ self.flags = []
+
+ def __bytearray__(self):
+ _bytes = super(FlagList, self).__bytearray__()
+ _bytes += b''.join(self.int_to_bytes(b) for b in self.flags)
+ return _bytes
+
+ def parse(self, packet):
+ super(FlagList, self).parse(packet)
+ for i in range(0, self.header.length - 1):
+ self.flags = packet[:1]
+ del packet[:1]
+
+
+class ByteFlag(Signature):
+ __flags__ = None
+
+ @sdproperty
+ def flags(self):
+ return self._flags
+
+ @flags.register(set)
+ @flags.register(list)
+ def flags_seq(self, val):
+ self._flags = set(val)
+
+ @flags.register(int)
+ @flags.register(_KeyFlags)
+ @flags.register(_Features)
+ def flags_int(self, val):
+ if self.__flags__ is None: # pragma: no cover
+ raise AttributeError("Error: __flags__ not set!")
+
+ self._flags |= (self.__flags__ & val)
+
+ @flags.register(bytearray)
+ def flags_bytearray(self, val):
+ self.flags = self.bytes_to_int(val)
+
+ def __init__(self):
+ super(ByteFlag, self).__init__()
+ self.flags = []
+
+ def __bytearray__(self):
+ _bytes = super(ByteFlag, self).__bytearray__()
+ _bytes += self.int_to_bytes(sum(self.flags))
+ # null-pad _bytes if they are not up to the end now
+ if len(_bytes) < len(self):
+ _bytes += b'\x00' * (len(self) - len(_bytes))
+ return _bytes
+
+ def parse(self, packet):
+ super(ByteFlag, self).parse(packet)
+ for i in range(0, self.header.length - 1):
+ self.flags = packet[:1]
+ del packet[:1]
+
+
+class Boolean(Signature):
+ @sdproperty
+ def bflag(self):
+ return self._bool
+
+ @bflag.register(bool)
+ def bflag_bool(self, val):
+ self._bool = val
+
+ @bflag.register(bytearray)
+ def bflag_bytearray(self, val):
+ self.bool = bool(self.bytes_to_int(val))
+
+ def __init__(self):
+ super(Boolean, self).__init__()
+ self.bflag = False
+
+ def __bytearray__(self):
+ _bytes = super(Boolean, self).__bytearray__()
+ _bytes += self.int_to_bytes(int(self.bflag))
+ return _bytes
+
+ def __bool__(self):
+ return self.bflag
+
+ def __nonzero__(self):
+ return self.__bool__()
+
+ def parse(self, packet):
+ super(Boolean, self).parse(packet)
+ self.bflag = packet[:1]
+ del packet[:1]
+
+
+class CreationTime(Signature):
+ """
+ 5.2.3.4. Signature Creation Time
+
+ (4-octet time field)
+
+ The time the signature was made.
+
+ MUST be present in the hashed area.
+ """
+ __typeid__ = 0x02
+
+ @sdproperty
+ def created(self):
+ return self._created
+
+ @created.register(datetime)
+ def created_datetime(self, val):
+ self._created = val
+
+ @created.register(int)
+ def created_int(self, val):
+ self.created = datetime.utcfromtimestamp(val)
+
+ @created.register(bytearray)
+ def created_bytearray(self, val):
+ self.created = self.bytes_to_int(val)
+
+ def __init__(self):
+ super(CreationTime, self).__init__()
+ self.created = datetime.utcnow()
+
+ def __bytearray__(self):
+ _bytes = super(CreationTime, self).__bytearray__()
+ _bytes += self.int_to_bytes(calendar.timegm(self.created.timetuple()), 4)
+ return _bytes
+
+ def parse(self, packet):
+ super(CreationTime, self).parse(packet)
+ self.created = packet[:4]
+ del packet[:4]
+
+
+class SignatureExpirationTime(Signature):
+ """
+ 5.2.3.10. Signature Expiration Time
+
+ (4-octet time field)
+
+ The validity period of the signature. This is the number of seconds
+ after the signature creation time that the signature expires. If
+ this is not present or has a value of zero, it never expires.
+ """
+ __typeid__ = 0x03
+
+ @sdproperty
+ def expires(self):
+ return self._expires
+
+ @expires.register(timedelta)
+ def expires_timedelta(self, val):
+ self._expires = val
+
+ @expires.register(int)
+ def expires_int(self, val):
+ self.expires = timedelta(seconds=val)
+
+ @expires.register(bytearray)
+ def expires_bytearray(self, val):
+ self.expires = self.bytes_to_int(val)
+
+ def __init__(self):
+ super(SignatureExpirationTime, self).__init__()
+ self.expires = 0
+
+ def __bytearray__(self):
+ _bytes = super(SignatureExpirationTime, self).__bytearray__()
+ _bytes += self.int_to_bytes(int(self.expires.total_seconds()), 4)
+ return _bytes
+
+ def parse(self, packet):
+ super(SignatureExpirationTime, self).parse(packet)
+ self.expires = packet[:4]
+ del packet[:4]
+
+
+class ExportableCertification(Boolean):
+ """
+ 5.2.3.11. Exportable Certification
+
+ (1 octet of exportability, 0 for not, 1 for exportable)
+
+ This subpacket denotes whether a certification signature is
+ "exportable", to be used by other users than the signature's issuer.
+ The packet body contains a Boolean flag indicating whether the
+ signature is exportable. If this packet is not present, the
+ certification is exportable; it is equivalent to a flag containing a
+ 1.
+
+ Non-exportable, or "local", certifications are signatures made by a
+ user to mark a key as valid within that user's implementation only.
+ Thus, when an implementation prepares a user's copy of a key for
+ transport to another user (this is the process of "exporting" the
+ key), any local certification signatures are deleted from the key.
+
+ The receiver of a transported key "imports" it, and likewise trims
+ any local certifications. In normal operation, there won't be any,
+ assuming the import is performed on an exported key. However, there
+ are instances where this can reasonably happen. For example, if an
+ implementation allows keys to be imported from a key database in
+ addition to an exported key, then this situation can arise.
+
+ Some implementations do not represent the interest of a single user
+ (for example, a key server). Such implementations always trim local
+ certifications from any key they handle.
+ """
+ __typeid__ = 0x04
+
+
+class TrustSignature(Signature):
+ """
+ 5.2.3.13. Trust Signature
+
+ (1 octet "level" (depth), 1 octet of trust amount)
+
+ Signer asserts that the key is not only valid but also trustworthy at
+ the specified level. Level 0 has the same meaning as an ordinary
+ validity signature. Level 1 means that the signed key is asserted to
+ be a valid trusted introducer, with the 2nd octet of the body
+ specifying the degree of trust. Level 2 means that the signed key is
+ asserted to be trusted to issue level 1 trust signatures, i.e., that
+ it is a "meta introducer". Generally, a level n trust signature
+ asserts that a key is trusted to issue level n-1 trust signatures.
+ The trust amount is in a range from 0-255, interpreted such that
+ values less than 120 indicate partial trust and values of 120 or
+ greater indicate complete trust. Implementations SHOULD emit values
+ of 60 for partial trust and 120 for complete trust.
+ """
+ __typeid__ = 0x05
+
+ @sdproperty
+ def level(self):
+ return self._level
+
+ @level.register(int)
+ def level_int(self, val):
+ self._level = val
+
+ @level.register(bytearray)
+ def level_bytearray(self, val):
+ self.level = self.bytes_to_int(val)
+
+ @sdproperty
+ def amount(self):
+ return self._amount
+
+ @amount.register(int)
+ def amount_int(self, val):
+ # clamp 'val' to the range 0-255
+ self._amount = max(0, min(val, 255))
+
+ @amount.register(bytearray)
+ def amount_bytearray(self, val):
+ self.amount = self.bytes_to_int(val)
+
+ def __init__(self):
+ super(TrustSignature, self).__init__()
+ self.level = 0
+ self.amount = 0
+
+ def __bytearray__(self):
+ _bytes = super(TrustSignature, self).__bytearray__()
+ _bytes += self.int_to_bytes(self.level)
+ _bytes += self.int_to_bytes(self.amount)
+ return _bytes
+
+ def parse(self, packet):
+ super(TrustSignature, self).parse(packet)
+ self.level = packet[:1]
+ del packet[:1]
+ self.amount = packet[:1]
+ del packet[:1]
+
+
+class RegularExpression(Signature):
+ """
+ 5.2.3.14. Regular Expression
+
+ (null-terminated regular expression)
+
+ Used in conjunction with trust Signature packets (of level > 0) to
+ limit the scope of trust that is extended. Only signatures by the
+ target key on User IDs that match the regular expression in the body
+ of this packet have trust extended by the trust Signature subpacket.
+ The regular expression uses the same syntax as the Henry Spencer's
+ "almost public domain" regular expression [REGEX] package. A
+ description of the syntax is found in Section 8 below.
+ """
+ __typeid__ = 0x06
+
+ @sdproperty
+ def regex(self):
+ return self._regex
+
+ @regex.register(str)
+ @regex.register(six.text_type)
+ def regex_str(self, val):
+ self._regex = val
+
+ @regex.register(bytearray)
+ def regex_bytearray(self, val):
+ self.regex = val.decode('latin-1')
+
+ def __init__(self):
+ super(RegularExpression, self).__init__()
+ self.regex = r''
+
+ def __bytearray__(self):
+ _bytes = super(RegularExpression, self).__bytearray__()
+ _bytes += self.regex.encode()
+ return _bytes
+
+ def parse(self, packet):
+ super(RegularExpression, self).parse(packet)
+ self.regex = packet[:(self.header.length - 1)]
+ del packet[:(self.header.length - 1)]
+
+
+class Revocable(Boolean):
+ """
+ 5.2.3.12. Revocable
+
+ (1 octet of revocability, 0 for not, 1 for revocable)
+
+ Signature's revocability status. The packet body contains a Boolean
+ flag indicating whether the signature is revocable. Signatures that
+ are not revocable have any later revocation signatures ignored. They
+ represent a commitment by the signer that he cannot revoke his
+ signature for the life of his key. If this packet is not present,
+ the signature is revocable.
+ """
+ __typeid__ = 0x07
+
+
+class KeyExpirationTime(SignatureExpirationTime):
+ """
+ 5.2.3.6. Key Expiration Time
+
+ (4-octet time field)
+
+ The validity period of the key. This is the number of seconds after
+ the key creation time that the key expires. If this is not present
+ or has a value of zero, the key never expires. This is found only on
+ a self-signature.
+ """
+ __typeid__ = 0x09
+
+
+class PreferredSymmetricAlgorithms(FlagList):
+ """
+ 5.2.3.7. Preferred Symmetric Algorithms
+
+ (array of one-octet values)
+
+ Symmetric algorithm numbers that indicate which algorithms the key
+ holder prefers to use. The subpacket body is an ordered list of
+ octets with the most preferred listed first. It is assumed that only
+ algorithms listed are supported by the recipient's software.
+ Algorithm numbers are in Section 9. This is only found on a self-
+ signature.
+ """
+ __typeid__ = 0x0B
+ __flags__ = SymmetricKeyAlgorithm
+
+
+class RevocationKey(Signature):
+ """
+ 5.2.3.15. Revocation Key
+
+ (1 octet of class, 1 octet of public-key algorithm ID, 20 octets of
+ fingerprint)
+
+ Authorizes the specified key to issue revocation signatures for this
+ key. Class octet must have bit 0x80 set. If the bit 0x40 is set,
+ then this means that the revocation information is sensitive. Other
+ bits are for future expansion to other kinds of authorizations. This
+ is found on a self-signature.
+
+ If the "sensitive" flag is set, the keyholder feels this subpacket
+ contains private trust information that describes a real-world
+ sensitive relationship. If this flag is set, implementations SHOULD
+ NOT export this signature to other users except in cases where the
+ data needs to be available: when the signature is being sent to the
+ designated revoker, or when it is accompanied by a revocation
+ signature from that revoker. Note that it may be appropriate to
+ isolate this subpacket within a separate signature so that it is not
+ combined with other subpackets that need to be exported.
+ """
+ __typeid__ = 0x0C
+
+ @sdproperty
+ def keyclass(self):
+ return self._keyclass
+
+ @keyclass.register(list)
+ def keyclass_list(self, val):
+ self._keyclass = val
+
+ @keyclass.register(int)
+ @keyclass.register(RevocationKeyClass)
+ def keyclass_int(self, val):
+ self._keyclass += RevocationKeyClass & val
+
+ @keyclass.register(bytearray)
+ def keyclass_bytearray(self, val):
+ self.keyclass = self.bytes_to_int(val)
+
+ @sdproperty
+ def algorithm(self):
+ return self._algorithm
+
+ @algorithm.register(int)
+ @algorithm.register(PubKeyAlgorithm)
+ def algorithm_int(self, val):
+ self._algorithm = PubKeyAlgorithm(val)
+
+ @algorithm.register(bytearray)
+ def algorithm_bytearray(self, val):
+ self.algorithm = self.bytes_to_int(val)
+
+ @sdproperty
+ def fingerprint(self):
+ return self._fingerprint
+
+ @fingerprint.register(str)
+ @fingerprint.register(six.text_type)
+ @fingerprint.register(Fingerprint)
+ def fingerprint_str(self, val):
+ self._fingerprint = Fingerprint(val)
+
+ @fingerprint.register(bytearray)
+ def fingerprint_bytearray(self, val):
+ self.fingerprint = ''.join('{:02x}'.format(c) for c in val).upper()
+
+ def __init__(self):
+ super(RevocationKey, self).__init__()
+ self.keyclass = []
+ self.algorithm = PubKeyAlgorithm.Invalid
+ self._fingerprint = ""
+
+ def __bytearray__(self):
+ _bytes = super(RevocationKey, self).__bytearray__()
+ _bytes += self.int_to_bytes(sum(self.keyclass))
+ _bytes += self.int_to_bytes(self.algorithm.value)
+ _bytes += self.fingerprint.__bytes__()
+ return _bytes
+
+ def parse(self, packet):
+ super(RevocationKey, self).parse(packet)
+ self.keyclass = packet[:1]
+ del packet[:1]
+ self.algorithm = packet[:1]
+ del packet[:1]
+ self.fingerprint = packet[:20]
+ del packet[:20]
+
+
+class Issuer(Signature):
+ __typeid__ = 0x10
+
+ @sdproperty
+ def issuer(self):
+ return self._issuer
+
+ @issuer.register(bytearray)
+ def issuer_bytearray(self, val):
+ self._issuer = binascii.hexlify(val).upper().decode('latin-1')
+
+ def __init__(self):
+ super(Issuer, self).__init__()
+ self.issuer = bytearray()
+
+ def __bytearray__(self):
+ _bytes = super(Issuer, self).__bytearray__()
+ _bytes += binascii.unhexlify(self._issuer.encode())
+ return _bytes
+
+ def parse(self, packet):
+ super(Issuer, self).parse(packet)
+ self.issuer = packet[:8]
+ del packet[:8]
+
+
+class NotationData(Signature):
+ __typeid__ = 0x14
+
+ @sdproperty
+ def flags(self):
+ return self._flags
+
+ @flags.register(list)
+ def flags_list(self, val):
+ self._flags = val
+
+ @flags.register(int)
+ @flags.register(NotationDataFlags)
+ def flags_int(self, val):
+ self.flags += NotationDataFlags & val
+
+ @flags.register(bytearray)
+ def flags_bytearray(self, val):
+ self.flags = self.bytes_to_int(val)
+
+ @sdproperty
+ def name(self):
+ return self._name
+
+ @name.register(str)
+ @name.register(six.text_type)
+ def name_str(self, val):
+ self._name = val
+
+ @name.register(bytearray)
+ def name_bytearray(self, val):
+ self.name = val.decode('latin-1')
+
+ @sdproperty
+ def value(self):
+ return self._value
+
+ @value.register(str)
+ @value.register(six.text_type)
+ def value_str(self, val):
+ self._value = val
+
+ @value.register(bytearray)
+ def value_bytearray(self, val):
+ if NotationDataFlags.HumanReadable in self.flags:
+ self.value = val.decode('latin-1')
+
+ else: # pragma: no cover
+ self._value = val
+
+ def __init__(self):
+ super(NotationData, self).__init__()
+ self.flags = [0, 0, 0, 0]
+ self.name = ""
+ self.value = ""
+
+ def __bytearray__(self):
+ _bytes = super(NotationData, self).__bytearray__()
+ _bytes += self.int_to_bytes(sum(self.flags)) + b'\x00\x00\x00'
+ _bytes += self.int_to_bytes(len(self.name), 2)
+ _bytes += self.int_to_bytes(len(self.value), 2)
+ _bytes += self.name.encode()
+ _bytes += self.value if isinstance(self.value, bytearray) else self.value.encode()
+ return bytes(_bytes)
+
+ def parse(self, packet):
+ super(NotationData, self).parse(packet)
+ self.flags = packet[:1]
+ del packet[:4]
+ nlen = self.bytes_to_int(packet[:2])
+ del packet[:2]
+ vlen = self.bytes_to_int(packet[:2])
+ del packet[:2]
+ self.name = packet[:nlen]
+ del packet[:nlen]
+ self.value = packet[:vlen]
+ del packet[:vlen]
+
+
+class PreferredHashAlgorithms(FlagList):
+ __typeid__ = 0x15
+ __flags__ = HashAlgorithm
+
+
+class PreferredCompressionAlgorithms(FlagList):
+ __typeid__ = 0x16
+ __flags__ = CompressionAlgorithm
+
+
+class KeyServerPreferences(FlagList):
+ __typeid__ = 0x17
+ __flags__ = _KeyServerPreferences
+
+
+class PreferredKeyServer(URI):
+ __typeid__ = 0x18
+
+
+class PrimaryUserID(Signature):
+ __typeid__ = 0x19
+
+ @sdproperty
+ def primary(self):
+ return self._primary
+
+ @primary.register(bool)
+ def primary_bool(self, val):
+ self._primary = val
+
+ @primary.register(bytearray)
+ def primary_byrearray(self, val):
+ self.primary = bool(self.bytes_to_int(val))
+
+ def __init__(self):
+ super(PrimaryUserID, self).__init__()
+ self.primary = True
+
+ def __bytearray__(self):
+ _bytes = super(PrimaryUserID, self).__bytearray__()
+ _bytes += self.int_to_bytes(int(self.primary))
+ return _bytes
+
+ def __bool__(self):
+ return self.primary
+
+ def __nonzero__(self):
+ return self.__bool__()
+
+ def parse(self, packet):
+ super(PrimaryUserID, self).parse(packet)
+ self.primary = packet[:1]
+ del packet[:1]
+
+
+class Policy(URI):
+ __typeid__ = 0x1a
+
+
+class KeyFlags(ByteFlag):
+ __typeid__ = 0x1B
+ __flags__ = _KeyFlags
+
+
+class SignersUserID(Signature):
+ __typeid__ = 0x1C
+
+ @sdproperty
+ def userid(self):
+ return self._userid
+
+ @userid.register(str)
+ @userid.register(six.text_type)
+ def userid_str(self, val):
+ self._userid = val
+
+ @userid.register(bytearray)
+ def userid_bytearray(self, val):
+ self.userid = val.decode('latin-1')
+
+ def __init__(self):
+ super(SignersUserID, self).__init__()
+ self.userid = ""
+
+ def __bytearray__(self):
+ _bytes = super(SignersUserID, self).__bytearray__()
+ _bytes += self.userid.encode()
+ return _bytes
+
+ def parse(self, packet):
+ super(SignersUserID, self).parse(packet)
+ self.userid = packet[:(self.header.length - 1)]
+ del packet[:(self.header.length - 1)]
+
+
+class ReasonForRevocation(Signature):
+ __typeid__ = 0x1D
+
+ @sdproperty
+ def code(self):
+ return self._code
+
+ @code.register(int)
+ @code.register(RevocationReason)
+ def code_int(self, val):
+ self._code = RevocationReason(val)
+
+ @code.register(bytearray)
+ def code_bytearray(self, val):
+ self.code = self.bytes_to_int(val)
+
+ @sdproperty
+ def string(self):
+ return self._string
+
+ @string.register(str)
+ @string.register(six.text_type)
+ def string_str(self, val):
+ self._string = val
+
+ @string.register(bytearray)
+ def string_bytearray(self, val):
+ self.string = val.decode('latin-1')
+
+ def __init__(self):
+ super(ReasonForRevocation, self).__init__()
+ self.code = 0x00
+ self.string = ""
+
+ def __bytearray__(self):
+ _bytes = super(ReasonForRevocation, self).__bytearray__()
+ _bytes += self.int_to_bytes(self.code)
+ _bytes += self.string.encode()
+ return _bytes
+
+ def parse(self, packet):
+ super(ReasonForRevocation, self).parse(packet)
+ self.code = packet[:1]
+ del packet[:1]
+ self.string = packet[:(self.header.length - 2)]
+ del packet[:(self.header.length - 2)]
+
+
+class Features(ByteFlag):
+ __typeid__ = 0x1E
+ __flags__ = _Features
+
+
+##TODO: obtain subpacket type 0x1F - Signature Target
+
+
+class EmbeddedSignature(Signature):
+ __typeid__ = 0x20
+
+ @sdproperty
+ def _sig(self):
+ return self._sigpkt
+
+ @_sig.setter
+ def _sig(self, val):
+ esh = EmbeddedSignatureHeader()
+ esh.version = val.header.version
+ val.header = esh
+ val.update_hlen()
+ self._sigpkt = val
+
+ @property
+ def sigtype(self):
+ return self._sig.sigtype
+
+ @property
+ def pubalg(self):
+ return self._sig.pubalg
+
+ @property
+ def halg(self):
+ return self._sig.halg
+
+ @property
+ def subpackets(self):
+ return self._sig.subpackets
+
+ @property
+ def hash2(self): # pragma: no cover
+ return self._sig.hash2
+
+ @property
+ def signature(self):
+ return self._sig.signature
+
+ @property
+ def signer(self):
+ return self._sig.signer
+
+ def __init__(self):
+ super(EmbeddedSignature, self).__init__()
+ from ..packets import SignatureV4
+ self._sigpkt = SignatureV4()
+ self._sigpkt.header = EmbeddedSignatureHeader()
+
+ def __bytearray__(self):
+ return super(EmbeddedSignature, self).__bytearray__() + self._sigpkt.__bytearray__()
+
+ def parse(self, packet):
+ super(EmbeddedSignature, self).parse(packet)
+ self._sig.parse(packet)
diff --git a/src/leap/mx/vendor/pgpy/packet/subpackets/types.py b/src/leap/mx/vendor/pgpy/packet/subpackets/types.py
new file mode 100644
index 0000000..7cf58d1
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/subpackets/types.py
@@ -0,0 +1,136 @@
+""" subpacket.py
+"""
+import abc
+
+from ..types import VersionedHeader
+
+from ...decorators import sdproperty
+
+from ...types import Dispatchable
+from ...types import Header as _Header
+
+__all__ = ['Header',
+ 'EmbeddedSignatureHeader',
+ 'SubPacket',
+ 'Signature',
+ 'UserAttribute',
+ 'Opaque']
+
+
+class Header(_Header):
+ @sdproperty
+ def critical(self):
+ return self._critical
+
+ @critical.register(bool)
+ def critical_bool(self, val):
+ self._critical = val
+
+ @sdproperty
+ def typeid(self):
+ return self._typeid
+
+ @typeid.register(int)
+ def typeid_int(self, val):
+ self._typeid = val & 0x7f
+
+ @typeid.register(bytes)
+ @typeid.register(bytearray)
+ def typeid_bin(self, val):
+ v = self.bytes_to_int(val)
+ self.typeid = v
+ self.critical = bool(v & 0x80)
+
+ def __init__(self):
+ super(Header, self).__init__()
+ self.typeid = b'\x00'
+ self.critical = False
+
+ def parse(self, packet):
+ self.length = packet
+
+ self.typeid = packet[:1]
+ del packet[:1]
+
+ def __len__(self):
+ return self.llen + 1
+
+ def __bytearray__(self):
+ _bytes = bytearray(self.encode_length(self.length))
+ _bytes += self.int_to_bytes((int(self.critical) << 7) + self.typeid)
+ return _bytes
+
+
+class EmbeddedSignatureHeader(VersionedHeader):
+ def __bytearray__(self):
+ return bytearray([self.version])
+
+ def parse(self, packet):
+ self.tag = 2
+ super(EmbeddedSignatureHeader, self).parse(packet)
+
+
+class SubPacket(Dispatchable):
+ __headercls__ = Header
+
+ def __init__(self):
+ super(SubPacket, self).__init__()
+ self.header = Header()
+
+ # if self.__typeid__ not in [-1, None]:
+ if (self.header.typeid == 0x00 and
+ (not hasattr(self.__typeid__, '__abstractmethod__')) and
+ (self.__typeid__ not in [-1, None])):
+ self.header.typeid = self.__typeid__
+
+ def __bytearray__(self):
+ return self.header.__bytearray__()
+
+ def __len__(self):
+ return (self.header.llen + self.header.length)
+
+ def __repr__(self):
+ return "<{} [0x{:02x}] at 0x{:x}>".format(self.__class__.__name__, self.header.typeid, id(self))
+
+ def update_hlen(self):
+ self.header.length = (len(self.__bytearray__()) - len(self.header)) + 1
+
+ @abc.abstractmethod
+ def parse(self, packet): # pragma: no cover
+ if self.header._typeid == 0:
+ self.header.parse(packet)
+
+
+class Signature(SubPacket):
+ __typeid__ = -1
+
+
+class UserAttribute(SubPacket):
+ __typeid__ = -1
+
+
+class Opaque(Signature, UserAttribute):
+ __typeid__ = None
+
+ @sdproperty
+ def payload(self):
+ return self._payload
+
+ @payload.register(bytes)
+ @payload.register(bytearray)
+ def payload_bin(self, val):
+ self._payload = bytearray(val)
+
+ def __init__(self):
+ super(Opaque, self).__init__()
+ self.payload = b''
+
+ def __bytearray__(self):
+ _bytes = super(Opaque, self).__bytearray__()
+ _bytes += self.payload
+ return _bytes
+
+ def parse(self, packet):
+ super(Opaque, self).parse(packet)
+ self.payload = packet[:(self.header.length - 1)]
+ del packet[:(self.header.length - 1)]
diff --git a/src/leap/mx/vendor/pgpy/packet/subpackets/userattribute.py b/src/leap/mx/vendor/pgpy/packet/subpackets/userattribute.py
new file mode 100644
index 0000000..32f7410
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/subpackets/userattribute.py
@@ -0,0 +1,109 @@
+""" userattribute.py
+"""
+import struct
+
+from .types import UserAttribute
+
+from ...constants import ImageEncoding
+
+from ...decorators import sdproperty
+
+from ...memoryview import memoryview
+
+
+__all__ = ('Image',)
+
+
+class Image(UserAttribute):
+ """
+ 5.12.1. The Image Attribute Subpacket
+
+ The Image Attribute subpacket is used to encode an image, presumably
+ (but not required to be) that of the key owner.
+
+ The Image Attribute subpacket begins with an image header. The first
+ two octets of the image header contain the length of the image
+ header. Note that unlike other multi-octet numerical values in this
+ document, due to a historical accident this value is encoded as a
+ little-endian number. The image header length is followed by a
+ single octet for the image header version. The only currently
+ defined version of the image header is 1, which is a 16-octet image
+ header. The first three octets of a version 1 image header are thus
+ 0x10, 0x00, 0x01.
+
+ The fourth octet of a version 1 image header designates the encoding
+ format of the image. The only currently defined encoding format is
+ the value 1 to indicate JPEG. Image format types 100 through 110 are
+ reserved for private or experimental use. The rest of the version 1
+ image header is made up of 12 reserved octets, all of which MUST be
+ set to 0.
+
+ The rest of the image subpacket contains the image itself. As the
+ only currently defined image type is JPEG, the image is encoded in
+ the JPEG File Interchange Format (JFIF), a standard file format for
+ JPEG images [JFIF].
+
+ An implementation MAY try to determine the type of an image by
+ examination of the image data if it is unable to handle a particular
+ version of the image header or if a specified encoding format value
+ is not recognized.
+ """
+ __typeid__ = 0x01
+
+ @sdproperty
+ def version(self):
+ return self._version
+
+ @version.register(int)
+ def version_int(self, val):
+ self._version = val
+
+ @sdproperty
+ def iencoding(self):
+ return self._iencoding
+
+ @iencoding.register(int)
+ @iencoding.register(ImageEncoding)
+ def iencoding_int(self, val):
+ try:
+ self._iencoding = ImageEncoding(val)
+
+ except ValueError: # pragma: no cover
+ self._iencoding = val
+
+ @sdproperty
+ def image(self):
+ return self._image
+
+ @image.register(bytes)
+ @image.register(bytearray)
+ def image_bin(self, val):
+ self._image = bytearray(val)
+
+ def __init__(self):
+ super(Image, self).__init__()
+ self.version = 1
+ self.iencoding = 1
+ self.image = bytearray()
+
+ def __bytearray__(self):
+ _bytes = super(Image, self).__bytearray__()
+
+ if self.version == 1:
+ # v1 image header length is always 16 bytes
+ # and stored little-endian due to an 'historical accident'
+ _bytes += struct.pack('<hbbiii', 16, self.version, self.iencoding, 0, 0, 0)
+
+ _bytes += self.image
+ return _bytes
+
+ def parse(self, packet):
+ super(Image, self).parse(packet)
+
+ # on Python 2, this will be the wrapper object from memoryview.py
+ with memoryview(packet) as _head:
+ _, self.version, self.iencoding, _, _, _ = struct.unpack_from('<hbbiii', _head[:16].tobytes())
+ del packet[:16]
+
+ self.image = packet[:(self.header.length - 17)]
+ del packet[:(self.header.length - 17)]
diff --git a/src/leap/mx/vendor/pgpy/packet/types.py b/src/leap/mx/vendor/pgpy/packet/types.py
new file mode 100644
index 0000000..e84ba67
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/packet/types.py
@@ -0,0 +1,288 @@
+""" types.py
+"""
+from __future__ import division
+
+import abc
+import copy
+
+import six
+
+from ..constants import PacketTag
+
+from ..decorators import sdproperty
+
+from ..types import Dispatchable
+from ..types import Field
+from ..types import Header as _Header
+
+__all__ = ['Header',
+ 'VersionedHeader',
+ 'Packet',
+ 'VersionedPacket',
+ 'Opaque',
+ 'Key',
+ 'Public',
+ 'Private',
+ 'Primary',
+ 'Sub',
+ 'MPI',
+ 'MPIs', ]
+
+
+class Header(_Header):
+ @sdproperty
+ def tag(self):
+ return self._tag
+
+ @tag.register(int)
+ @tag.register(PacketTag)
+ def tag_int(self, val):
+ _tag = (val & 0x3F) if self._lenfmt else ((val & 0x3C) >> 2)
+ try:
+ self._tag = PacketTag(_tag)
+
+ except ValueError: # pragma: no cover
+ self._tag = _tag
+
+ @property
+ def typeid(self):
+ return self.tag
+
+ def __init__(self):
+ super(Header, self).__init__()
+ self.tag = 0x00
+
+ def __bytearray__(self):
+ tag = 0x80 | (self._lenfmt << 6)
+ tag |= (self.tag) if self._lenfmt else ((self.tag << 2) | {1: 0, 2: 1, 4: 2, 0: 3}[self.llen])
+
+ _bytes = bytearray(self.int_to_bytes(tag))
+ _bytes += self.encode_length(self.length, self._lenfmt, self.llen)
+ return _bytes
+
+ def __len__(self):
+ return 1 + self.llen
+
+ def parse(self, packet):
+ """
+ There are two formats for headers
+
+ old style
+ ---------
+
+ Old style headers can be 1, 2, 3, or 6 octets long and are composed of a Tag and a Length.
+ If the header length is 1 octet (length_type == 3), then there is no Length field.
+
+ new style
+ ---------
+
+ New style headers can be 2, 3, or 6 octets long and are also composed of a Tag and a Length.
+
+
+ Packet Tag
+ ----------
+
+ The packet tag is the first byte, comprising the following fields:
+
+ +-------------+----------+---------------+---+---+---+---+----------+----------+
+ | byte | 1 |
+ +-------------+----------+---------------+---+---+---+---+----------+----------+
+ | bit | 7 | 6 | 5 | 4 | 3 | 2 | 1 | 0 |
+ +-------------+----------+---------------+---+---+---+---+----------+----------+
+ | old-style | always 1 | packet format | packet tag | length type |
+ | description | | 0 = old-style | | 0 = 1 octet |
+ | | | 1 = new-style | | 1 = 2 octets |
+ | | | | | 2 = 5 octets |
+ | | | | | 3 = no length field |
+ +-------------+ + +---------------+---------------------+
+ | new-style | | | packet tag |
+ | description | | | |
+ +-------------+----------+---------------+-------------------------------------+
+
+ :param packet: raw packet bytes
+ """
+ self._lenfmt = ((packet[0] & 0x40) >> 6)
+ self.tag = packet[0]
+ if self._lenfmt == 0:
+ self.llen = (packet[0] & 0x03)
+ del packet[0]
+
+ if (self._lenfmt == 0 and self.llen > 0) or self._lenfmt == 1:
+ self.length = packet
+
+ else:
+ # indeterminate packet length
+ self.length = len(packet)
+
+
+class VersionedHeader(Header):
+ @sdproperty
+ def version(self):
+ return self._version
+
+ @version.register(int)
+ def version_int(self, val):
+ self._version = val
+
+ def __init__(self):
+ super(VersionedHeader, self).__init__()
+ self.version = 0
+
+ def __bytearray__(self):
+ _bytes = bytearray(super(VersionedHeader, self).__bytearray__())
+ _bytes += bytearray([self.version])
+ return _bytes
+
+ def parse(self, packet): # pragma: no cover
+ if self.tag == 0:
+ super(VersionedHeader, self).parse(packet)
+
+ if self.version == 0:
+ self.version = packet[0]
+ del packet[0]
+
+
+class Packet(Dispatchable):
+ __typeid__ = -1
+ __headercls__ = Header
+
+ def __init__(self):
+ super(Packet, self).__init__()
+ self.header = self.__headercls__()
+ if isinstance(self.__typeid__, six.integer_types):
+ self.header.tag = self.__typeid__
+
+ @abc.abstractmethod
+ def __bytearray__(self):
+ return self.header.__bytearray__()
+
+ def __len__(self):
+ return len(self.header) + self.header.length
+
+ def __repr__(self):
+ return "<{cls:s} [tag 0x{tag:02d}] at 0x{id:x}>".format(cls=self.__class__.__name__, tag=self.header.tag, id=id(self))
+
+ def update_hlen(self):
+ self.header.length = len(self.__bytearray__()) - len(self.header)
+
+ @abc.abstractmethod
+ def parse(self, packet):
+ if self.header.tag == 0:
+ self.header.parse(packet)
+
+
+class VersionedPacket(Packet):
+ __headercls__ = VersionedHeader
+
+ def __init__(self):
+ super(VersionedPacket, self).__init__()
+ if isinstance(self.__ver__, six.integer_types):
+ self.header.version = self.__ver__
+
+ def __repr__(self):
+ return "<{cls:s} [tag 0x{tag:02d}][v{ver:d}] at 0x{id:x}>".format(cls=self.__class__.__name__, tag=self.header.tag,
+ ver=self.header.version, id=id(self))
+
+
+class Opaque(Packet):
+ __typeid__ = None
+
+ @sdproperty
+ def payload(self):
+ return self._payload
+
+ @payload.register(bytearray)
+ @payload.register(bytes)
+ def payload_bin(self, val):
+ self._payload = val
+
+ def __init__(self):
+ super(Opaque, self).__init__()
+ self.payload = b''
+
+ def __bytearray__(self):
+ _bytes = super(Opaque, self).__bytearray__()
+ _bytes += self.payload
+ return _bytes
+
+ def parse(self, packet): # pragma: no cover
+ super(Opaque, self).parse(packet)
+ pend = self.header.length
+ if hasattr(self.header, 'version'):
+ pend -= 1
+
+ self.payload = packet[:pend]
+ del packet[:pend]
+
+
+# key marker classes for convenience
+class Key(object):
+ pass
+
+
+class Public(Key):
+ pass
+
+
+class Private(Key):
+ pass
+
+
+class Primary(Key):
+ pass
+
+
+class Sub(Key):
+ pass
+
+
+# This is required for class MPI to work in both Python 2 and 3
+if not six.PY2:
+ long = int
+
+
+class MPI(long):
+ def __new__(cls, num):
+ mpi = num
+
+ if isinstance(num, (bytes, bytearray)):
+ if isinstance(num, bytes): # pragma: no cover
+ num = bytearray(num)
+
+ fl = ((MPIs.bytes_to_int(num[:2]) + 7) // 8)
+ del num[:2]
+
+ mpi = MPIs.bytes_to_int(num[:fl])
+ del num[:fl]
+
+ return super(MPI, cls).__new__(cls, mpi)
+
+ def byte_length(self):
+ return ((self.bit_length() + 7) // 8)
+
+ def to_mpibytes(self):
+ return MPIs.int_to_bytes(self.bit_length(), 2) + MPIs.int_to_bytes(self, self.byte_length())
+
+ def __len__(self):
+ return self.byte_length() + 2
+
+
+class MPIs(Field):
+ # this differs from MPI in that it's subclasses hold/parse several MPI fields
+ # and, in the case of v4 private keys, also a String2Key specifier/information.
+ __mpis__ = ()
+
+ def __len__(self):
+ return sum(len(i) for i in self)
+
+ def __iter__(self):
+ """yield all components of an MPI so it can be iterated over"""
+ for i in self.__mpis__:
+ yield getattr(self, i)
+
+ def __copy__(self):
+ pk = self.__class__()
+ for m in self.__mpis__:
+ setattr(pk, m, copy.copy(getattr(self, m)))
+
+ return pk
diff --git a/src/leap/mx/vendor/pgpy/pgp.py b/src/leap/mx/vendor/pgpy/pgp.py
new file mode 100644
index 0000000..82b3a0b
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/pgp.py
@@ -0,0 +1,2527 @@
+""" pgp.py
+
+this is where the armorable PGP block objects live
+"""
+import binascii
+import calendar
+import collections
+import contextlib
+import copy
+import functools
+import itertools
+import operator
+import os
+import re
+import warnings
+import weakref
+
+import six
+
+from datetime import datetime
+
+from cryptography.hazmat.primitives import hashes
+
+from .constants import CompressionAlgorithm
+from .constants import Features
+from .constants import HashAlgorithm
+from .constants import ImageEncoding
+from .constants import KeyFlags
+from .constants import NotationDataFlags
+from .constants import PacketTag
+from .constants import PubKeyAlgorithm
+from .constants import RevocationKeyClass
+from .constants import RevocationReason
+from .constants import SignatureType
+from .constants import SymmetricKeyAlgorithm
+
+from .decorators import KeyAction
+
+from .errors import PGPDecryptionError
+from .errors import PGPError
+
+from .packet import Key
+from .packet import MDC
+from .packet import Packet
+from .packet import Primary
+from .packet import Private
+from .packet import PubKeyV4
+from .packet import PubSubKeyV4
+from .packet import PrivKeyV4
+from .packet import PrivSubKeyV4
+from .packet import Public
+from .packet import Sub
+from .packet import UserID
+from .packet import UserAttribute
+
+from .packet.packets import CompressedData
+from .packet.packets import IntegrityProtectedSKEData
+from .packet.packets import IntegrityProtectedSKEDataV1
+from .packet.packets import LiteralData
+from .packet.packets import OnePassSignature
+from .packet.packets import OnePassSignatureV3
+from .packet.packets import PKESessionKey
+from .packet.packets import PKESessionKeyV3
+from .packet.packets import Signature
+from .packet.packets import SignatureV4
+from .packet.packets import SKEData
+from .packet.packets import Marker
+from .packet.packets import SKESessionKey
+from .packet.packets import SKESessionKeyV4
+
+from .packet.types import Opaque
+
+from .types import Armorable
+from .types import Fingerprint
+from .types import ParentRef
+from .types import PGPObject
+from .types import SignatureVerification
+from .types import SorteDeque
+
+__all__ = ['PGPSignature',
+ 'PGPUID',
+ 'PGPMessage',
+ 'PGPKey',
+ 'PGPKeyring']
+
+
+class PGPSignature(Armorable, ParentRef, PGPObject):
+ @property
+ def __sig__(self):
+ return self._signature.signature.__sig__()
+
+ @property
+ def cipherprefs(self):
+ """
+ A ``list`` of preferred symmetric algorithms specified in this signature, if any. Otherwise, an empty ``list``.
+ """
+ if 'PreferredSymmetricAlgorithms' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['h_PreferredSymmetricAlgorithms'])).flags
+ return []
+
+ @property
+ def compprefs(self):
+ """
+ A ``list`` of preferred compression algorithms specified in this signature, if any. Otherwise, an empty ``list``.
+ """
+ if 'PreferredCompressionAlgorithms' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['h_PreferredCompressionAlgorithms'])).flags
+ return []
+
+ @property
+ def created(self):
+ """
+ A :py:obj:`~datetime.datetime` of when this signature was created.
+ """
+ return self._signature.subpackets['h_CreationTime'][-1].created
+
+ @property
+ def embedded(self):
+ return self.parent is not None
+
+ @property
+ def expires_at(self):
+ """
+ A :py:obj:`~datetime.datetime` of when this signature expires, if a signature expiration date is specified.
+ Otherwise, ``None``
+ """
+ if 'SignatureExpirationTime' in self._signature.subpackets:
+ expd = next(iter(self._signature.subpackets['SignatureExpirationTime'])).expires
+ return self.created + expd
+ return None
+
+ @property
+ def exportable(self):
+ """
+ ``False`` if this signature is marked as being not exportable. Otherwise, ``True``.
+ """
+ if 'ExportableCertification' in self._signature.subpackets:
+ return bool(next(iter(self._signature.subpackets['ExportableCertification'])))
+
+ return True
+
+ @property
+ def features(self):
+ """
+ A ``set`` of implementation features specified in this signature, if any. Otherwise, an empty ``set``.
+ """
+ if 'Features' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['Features'])).flags
+ return set()
+
+ @property
+ def hash2(self):
+ return self._signature.hash2
+
+ @property
+ def hashprefs(self):
+ """
+ A ``list`` of preferred hash algorithms specified in this signature, if any. Otherwise, an empty ``list``.
+ """
+ if 'PreferredHashAlgorithms' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['h_PreferredHashAlgorithms'])).flags
+ return []
+
+ @property
+ def hash_algorithm(self):
+ """
+ The :py:obj:`~constants.HashAlgorithm` used when computing this signature.
+ """
+ return self._signature.halg
+
+ @property
+ def is_expired(self):
+ """
+ ``True`` if the signature has an expiration date, and is expired. Otherwise, ``False``
+ """
+ expires_at = self.expires_at
+ if expires_at is not None and expires_at != self.created:
+ return expires_at < datetime.utcnow()
+
+ return False
+
+ @property
+ def key_algorithm(self):
+ """
+ The :py:obj:`~constants.PubKeyAlgorithm` of the key that generated this signature.
+ """
+ return self._signature.pubalg
+
+ @property
+ def key_expiration(self):
+ if 'KeyExpirationTime' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['KeyExpirationTime'])).expires
+ return None
+
+ @property
+ def key_flags(self):
+ """
+ A ``set`` of :py:obj:`~constants.KeyFlags` specified in this signature, if any. Otherwise, an empty ``set``.
+ """
+ if 'KeyFlags' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['h_KeyFlags'])).flags
+ return set()
+
+ @property
+ def keyserver(self):
+ """
+ The preferred key server specified in this signature, if any. Otherwise, an empty ``str``.
+ """
+ if 'PreferredKeyServer' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['h_PreferredKeyServer'])).uri
+ return ''
+
+ @property
+ def keyserverprefs(self):
+ """
+ A ``list`` of :py:obj:`~constants.KeyServerPreferences` in this signature, if any. Otherwise, an empty ``list``.
+ """
+ if 'KeyServerPreferences' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['h_KeyServerPreferences'])).flags
+ return []
+
+ @property
+ def magic(self):
+ return "SIGNATURE"
+
+ @property
+ def notation(self):
+ """
+ A ``dict`` of notation data in this signature, if any. Otherwise, an empty ``dict``.
+ """
+ return dict((nd.name, nd.value) for nd in self._signature.subpackets['NotationData'])
+
+ @property
+ def policy_uri(self):
+ """
+ The policy URI specified in this signature, if any. Otherwise, an empty ``str``.
+ """
+ if 'Policy' in self._signature.subpackets:
+ return next(iter(self._signature.subpackets['Policy'])).uri
+ return ''
+
+ @property
+ def revocable(self):
+ """
+ ``False`` if this signature is marked as being not revocable. Otherwise, ``True``.
+ """
+ if 'Revocable' in self._signature.subpackets:
+ return bool(next(iter(self._signature.subpackets['Revocable'])))
+ return True
+
+ @property
+ def revocation_key(self):
+ if 'RevocationKey' in self._signature.subpackets:
+ raise NotImplementedError()
+ return None
+
+ @property
+ def signer(self):
+ """
+ The 16-character Key ID of the key that generated this signature.
+ """
+ return self._signature.signer
+
+ @property
+ def target_signature(self):
+ return NotImplemented
+
+ @property
+ def type(self):
+ """
+ The :py:obj:`~constants.SignatureType` of this signature.
+ """
+ return self._signature.sigtype
+
+ @classmethod
+ def new(cls, sigtype, pkalg, halg, signer):
+ sig = PGPSignature()
+
+ sigpkt = SignatureV4()
+ sigpkt.header.tag = 2
+ sigpkt.header.version = 4
+ sigpkt.subpackets.addnew('CreationTime', hashed=True, created=datetime.utcnow())
+ sigpkt.subpackets.addnew('Issuer', _issuer=signer)
+
+ sigpkt.sigtype = sigtype
+ sigpkt.pubalg = pkalg
+
+ if halg is not None:
+ sigpkt.halg = halg
+
+ sig._signature = sigpkt
+ return sig
+
+ def __init__(self):
+ """
+ PGPSignature objects represent OpenPGP compliant signatures.
+
+ PGPSignature implements the ``__str__`` method, the output of which will be the signature object in
+ OpenPGP-compliant ASCII-armored format.
+
+ PGPSignature implements the ``__bytes__`` method, the output of which will be the signature object in
+ OpenPGP-compliant binary format.
+ """
+ super(PGPSignature, self).__init__()
+ self._signature = None
+
+ def __bytearray__(self):
+ return self._signature.__bytearray__()
+
+ def __repr__(self):
+ return "<PGPSignature [{:s}] object at 0x{:02x}>".format(self.type.name, id(self))
+
+ def __lt__(self, other):
+ return self.created < other.created
+
+ def __or__(self, other):
+ if isinstance(other, Signature):
+ if self._signature is None:
+ self._signature = other
+ return self
+
+ ##TODO: this is not a great way to do this
+ if other.__class__.__name__ == 'EmbeddedSignature':
+ self._signature = other
+ return self
+
+ raise TypeError
+
+ def __copy__(self):
+ # because the default shallow copy isn't actually all that useful,
+ # and deepcopy does too much work
+ sig = super(PGPSignature, self).__copy__()
+ # sig = PGPSignature()
+ # sig.ascii_headers = self.ascii_headers.copy()
+ sig |= copy.copy(self._signature)
+ return sig
+
+ def hashdata(self, subject):
+ _data = bytearray()
+
+ if isinstance(subject, six.string_types):
+ subject = subject.encode('latin-1')
+
+ """
+ All signatures are formed by producing a hash over the signature
+ data, and then using the resulting hash in the signature algorithm.
+ """
+
+ if self.type == SignatureType.BinaryDocument:
+ """
+ For binary document signatures (type 0x00), the document data is
+ hashed directly.
+ """
+ _data += bytearray(subject)
+
+ if self.type == SignatureType.CanonicalDocument:
+ """
+ For text document signatures (type 0x01), the
+ document is canonicalized by converting line endings to <CR><LF>,
+ and the resulting data is hashed.
+ """
+ _data += re.subn(br'\r?\n', b'\r\n', subject)[0]
+
+ if self.type in {SignatureType.Generic_Cert, SignatureType.Persona_Cert, SignatureType.Casual_Cert,
+ SignatureType.Positive_Cert, SignatureType.CertRevocation, SignatureType.Subkey_Binding,
+ SignatureType.PrimaryKey_Binding}:
+ """
+ When a signature is made over a key, the hash data starts with the
+ octet 0x99, followed by a two-octet length of the key, and then body
+ of the key packet. (Note that this is an old-style packet header for
+ a key packet with two-octet length.) ...
+ Key revocation signatures (types 0x20 and 0x28)
+ hash only the key being revoked.
+ """
+ _s = b''
+ if isinstance(subject, PGPUID):
+ _s = subject._parent.hashdata
+
+ elif isinstance(subject, PGPKey) and not subject.is_primary:
+ _s = subject._parent.hashdata
+
+ elif isinstance(subject, PGPKey) and subject.is_primary:
+ _s = subject.hashdata
+
+ if len(_s) > 0:
+ _data += b'\x99' + self.int_to_bytes(len(_s), 2) + _s
+
+ if self.type in {SignatureType.Subkey_Binding, SignatureType.PrimaryKey_Binding}:
+ """
+ A subkey binding signature
+ (type 0x18) or primary key binding signature (type 0x19) then hashes
+ the subkey using the same format as the main key (also using 0x99 as
+ the first octet).
+ """
+ if subject.is_primary:
+ _s = subject.subkeys[self.signer].hashdata
+
+ else:
+ _s = subject.hashdata
+
+ _data += b'\x99' + self.int_to_bytes(len(_s), 2) + _s
+
+ if self.type in {SignatureType.KeyRevocation, SignatureType.SubkeyRevocation, SignatureType.DirectlyOnKey}:
+ """
+ The signature is calculated directly on the key being revoked. A
+ revoked key is not to be used. Only revocation signatures by the
+ key being revoked, or by an authorized revocation key, should be
+ considered valid revocation signatures.
+
+ Subkey revocation signature
+ The signature is calculated directly on the subkey being revoked.
+ A revoked subkey is not to be used. Only revocation signatures
+ by the top-level signature key that is bound to this subkey, or
+ by an authorized revocation key, should be considered valid
+ revocation signatures.
+
+ Signature directly on a key
+ This signature is calculated directly on a key. It binds the
+ information in the Signature subpackets to the key, and is
+ appropriate to be used for subpackets that provide information
+ about the key, such as the Revocation Key subpacket. It is also
+ appropriate for statements that non-self certifiers want to make
+ about the key itself, rather than the binding between a key and a
+ name.
+ """
+ _s = subject.hashdata
+ _data += b'\x99' + self.int_to_bytes(len(_s), 2) + _s
+
+ if self.type in {SignatureType.Generic_Cert, SignatureType.Persona_Cert, SignatureType.Casual_Cert,
+ SignatureType.Positive_Cert, SignatureType.CertRevocation}:
+ """
+ A certification signature (type 0x10 through 0x13) hashes the User
+ ID being bound to the key into the hash context after the above
+ data. ... A V4 certification
+ hashes the constant 0xB4 for User ID certifications or the constant
+ 0xD1 for User Attribute certifications, followed by a four-octet
+ number giving the length of the User ID or User Attribute data, and
+ then the User ID or User Attribute data.
+
+ ...
+
+ The [certificate revocation] signature
+ is computed over the same data as the certificate that it
+ revokes, and should have a later creation date than that
+ certificate.
+ """
+
+ _s = subject.hashdata
+ if subject.is_uid:
+ _data += b'\xb4'
+
+ else:
+ _data += b'\xd1'
+
+ _data += self.int_to_bytes(len(_s), 4) + _s
+
+ # if this is a new signature, do update_hlen
+ if 0 in list(self._signature.signature):
+ self._signature.update_hlen()
+
+ """
+ Once the data body is hashed, then a trailer is hashed. (...)
+ A V4 signature hashes the packet body
+ starting from its first field, the version number, through the end
+ of the hashed subpacket data. Thus, the fields hashed are the
+ signature version, the signature type, the public-key algorithm, the
+ hash algorithm, the hashed subpacket length, and the hashed
+ subpacket body.
+
+ V4 signatures also hash in a final trailer of six octets: the
+ version of the Signature packet, i.e., 0x04; 0xFF; and a four-octet,
+ big-endian number that is the length of the hashed data from the
+ Signature packet (note that this number does not include these final
+ six octets).
+ """
+
+ hcontext = bytearray()
+ hcontext.append(self._signature.header.version if not self.embedded else self._signature._sig.header.version)
+ hcontext.append(self.type)
+ hcontext.append(self.key_algorithm)
+ hcontext.append(self.hash_algorithm)
+ hcontext += self._signature.subpackets.__hashbytearray__()
+ hlen = len(hcontext)
+ _data += hcontext
+ _data += b'\x04\xff'
+ _data += self.int_to_bytes(hlen, 4)
+ return bytes(_data)
+
+ def make_onepass(self):
+ onepass = OnePassSignatureV3()
+ onepass.sigtype = self.type
+ onepass.halg = self.hash_algorithm
+ onepass.pubalg = self.key_algorithm
+ onepass.signer = self.signer
+ onepass.update_hlen()
+ return onepass
+
+ def parse(self, packet):
+ unarmored = self.ascii_unarmor(packet)
+ data = unarmored['body']
+
+ if unarmored['magic'] is not None and unarmored['magic'] != 'SIGNATURE':
+ raise ValueError('Expected: SIGNATURE. Got: {}'.format(str(unarmored['magic'])))
+
+ if unarmored['headers'] is not None:
+ self.ascii_headers = unarmored['headers']
+
+ # load *one* packet from data
+ pkt = Packet(data)
+ if pkt.header.tag == PacketTag.Signature and not isinstance(pkt, Opaque):
+ self._signature = pkt
+
+ else:
+ raise ValueError('Expected: Signature. Got: {:s}'.format(pkt.__class__.__name__))
+
+
+class PGPUID(ParentRef):
+ @property
+ def __sig__(self):
+ return list(self._signatures)
+
+ @property
+ def name(self):
+ """If this is a User ID, the stored name. If this is not a User ID, this will be an empty string."""
+ return self._uid.name if isinstance(self._uid, UserID) else ""
+
+ @property
+ def comment(self):
+ """
+ If this is a User ID, this will be the stored comment. If this is not a User ID, or there is no stored comment,
+ this will be an empty string.,
+ """
+
+ return self._uid.comment if isinstance(self._uid, UserID) else ""
+
+ @property
+ def email(self):
+ """
+ If this is a User ID, this will be the stored email address. If this is not a User ID, or there is no stored
+ email address, this will be an empty string.
+ """
+ return self._uid.email if isinstance(self._uid, UserID) else ""
+
+ @property
+ def image(self):
+ """
+ If this is a User Attribute, this will be the stored image. If this is not a User Attribute, this will be ``None``.
+ """
+ return self._uid.image.image if isinstance(self._uid, UserAttribute) else None
+
+ @property
+ def is_primary(self):
+ """
+ If the most recent, valid self-signature specifies this as being primary, this will be True. Otherwise, Faqlse.
+ """
+ return bool(next(iter(self.selfsig._signature.subpackets['h_PrimaryUserID']), False))
+
+ @property
+ def is_uid(self):
+ """
+ ``True`` if this is a User ID, otherwise False.
+ """
+ return isinstance(self._uid, UserID)
+
+ @property
+ def is_ua(self):
+ """
+ ``True`` if this is a User Attribute, otherwise False.
+ """
+ return isinstance(self._uid, UserAttribute)
+
+ @property
+ def selfsig(self):
+ """
+ This will be the most recent, self-signature of this User ID or Attribute. If there isn't one, this will be ``None``.
+ """
+ if self.parent is not None:
+ return next((sig for sig in reversed(self._signatures) if sig.signer == self.parent.fingerprint.keyid), None)
+
+ @property
+ def signers(self):
+ """
+ This will be a set of all of the key ids which have signed this User ID or Attribute.
+ """
+ return set(s.signer for s in self.__sig__)
+
+ @property
+ def hashdata(self):
+ if self.is_uid:
+ return self._uid.__bytearray__()[len(self._uid.header):]
+
+ if self.is_ua:
+ return self._uid.subpackets.__bytearray__()
+
+ @classmethod
+ def new(cls, pn, comment="", email=""):
+ """
+ Create a new User ID or photo.
+
+ :param pn: User ID name, or photo. If this is a ``bytearray``, it will be loaded as a photo.
+ Otherwise, it will be used as the name field for a User ID.
+ :type pn: ``bytearray``, ``str``, ``unicode``
+ :param comment: The comment field for a User ID. Ignored if this is a photo.
+ :type comment: ``str``, ``unicode``
+ :param email: The email address field for a User ID. Ignored if this is a photo.
+ :type email: ``str``, ``unicode``
+ :returns: :py:obj:`PGPUID`
+ """
+ uid = PGPUID()
+ if isinstance(pn, bytearray):
+ uid._uid = UserAttribute()
+ uid._uid.image.image = pn
+ uid._uid.image.iencoding = ImageEncoding.encodingof(pn)
+ uid._uid.update_hlen()
+
+ else:
+ uid._uid = UserID()
+ uid._uid.name = pn
+ uid._uid.comment = comment
+ uid._uid.email = email
+ uid._uid.update_hlen()
+
+ return uid
+
+ def __init__(self):
+ """
+ PGPUID objects represent User IDs and User Attributes for keys.
+
+ PGPUID implements the ``__format__`` method for User IDs, returning a string in the format
+ 'name (comment) <email>', leaving out any comment or email fields that are not present.
+ """
+ super(PGPUID, self).__init__()
+ self._uid = None
+ self._signatures = SorteDeque()
+
+ def __repr__(self):
+ if self.selfsig is not None:
+ return "<PGPUID [{:s}][{}] at 0x{:02X}>".format(self._uid.__class__.__name__, self.selfsig.created, id(self))
+ return "<PGPUID [{:s}] at 0x{:02X}>".format(self._uid.__class__.__name__, id(self))
+
+ def __lt__(self, other): # pragma: no cover
+ if self.is_uid == other.is_uid:
+ if self.is_primary == other.is_primary:
+ return self.selfsig > other.selfsig
+
+ if self.is_primary:
+ return True
+
+ return False
+
+ if self.is_uid and other.is_ua:
+ return True
+
+ if self.is_ua and other.is_uid:
+ return False
+
+ def __or__(self, other):
+ if isinstance(other, PGPSignature):
+ self._signatures.insort(other)
+ if self.parent is not None and self in self.parent._uids:
+ self.parent._uids.resort(self)
+
+ return self
+
+ if isinstance(other, UserID) and self._uid is None:
+ self._uid = other
+ return self
+
+ if isinstance(other, UserAttribute) and self._uid is None:
+ self._uid = other
+ return self
+
+ raise TypeError("unsupported operand type(s) for |: '{:s}' and '{:s}'"
+ "".format(self.__class__.__name__, other.__class__.__name__))
+
+ def __copy__(self):
+ # because the default shallow copy isn't actually all that useful,
+ # and deepcopy does too much work
+ uid = PGPUID()
+ uid |= copy.copy(self._uid)
+ for sig in self._signatures:
+ uid |= copy.copy(sig)
+ return uid
+
+ def __format__(self, format_spec):
+ if self.is_uid:
+ comment = six.u("") if self.comment == "" else six.u(" ({:s})").format(self.comment)
+ email = six.u("") if self.email == "" else six.u(" <{:s}>").format(self.email)
+ return six.u("{:s}{:s}{:s}").format(self.name, comment, email)
+
+ raise NotImplementedError
+
+
+class PGPMessage(Armorable, PGPObject):
+ @staticmethod
+ def dash_unescape(text):
+ return re.subn(r'^- -', '-', text, flags=re.MULTILINE)[0]
+
+ @staticmethod
+ def dash_escape(text):
+ return re.subn(r'^-', '- -', text, flags=re.MULTILINE)[0]
+
+ @property
+ def encrypters(self):
+ """A ``set`` containing all key ids (if any) to which this message was encrypted."""
+ return set(m.encrypter for m in self._sessionkeys if isinstance(m, PKESessionKey))
+
+ @property
+ def filename(self):
+ """If applicable, returns the original filename of the message. Otherwise, returns an empty string."""
+ if self.type == 'literal':
+ return self._message.filename
+ return ''
+
+ @property
+ def is_compressed(self):
+ """``True`` if this message will be compressed when exported"""
+ return self._compression != CompressionAlgorithm.Uncompressed
+
+ @property
+ def is_encrypted(self):
+ """``True`` if this message is encrypted; otherwise, ``False``"""
+ return isinstance(self._message, (SKEData, IntegrityProtectedSKEData))
+
+ @property
+ def is_sensitive(self):
+ """``True`` if this message is marked sensitive; otherwise ``False``"""
+ return self.type == 'literal' and self._message.filename == '_CONSOLE'
+
+ @property
+ def is_signed(self):
+ """
+ ``True`` if this message is signed; otherwise, ``False``.
+ Should always be ``False`` if the message is encrypted.
+ """
+ return len(self._signatures) > 0
+
+ @property
+ def issuers(self):
+ """A ``set`` containing all key ids (if any) which have signed or encrypted this message."""
+ return self.encrypters | self.signers
+
+ @property
+ def magic(self):
+ if self.type == 'cleartext':
+ return "SIGNATURE"
+ return "MESSAGE"
+
+ @property
+ def message(self):
+ """The message contents"""
+ if self.type == 'cleartext':
+ return self.bytes_to_text(self._message)
+
+ if self.type == 'literal':
+ return self._message.contents
+
+ if self.type == 'encrypted':
+ return self._message
+
+ @property
+ def signatures(self):
+ """A ``set`` containing all key ids (if any) which have signed this message."""
+ return list(self._signatures)
+
+ @property
+ def signers(self):
+ """A ``set`` containing all key ids (if any) which have signed this message."""
+ return set(m.signer for m in self._signatures)
+
+ @property
+ def type(self):
+ ##TODO: it might be better to use an Enum for the output of this
+ if isinstance(self._message, (six.string_types, six.binary_type, bytearray)):
+ return 'cleartext'
+
+ if isinstance(self._message, LiteralData):
+ return 'literal'
+
+ if isinstance(self._message, (SKEData, IntegrityProtectedSKEData)):
+ return 'encrypted'
+
+ raise NotImplementedError
+
+ def __init__(self):
+ """
+ PGPMessage objects represent OpenPGP message compositions.
+
+ PGPMessage implements the `__str__` method, the output of which will be the message composition in
+ OpenPGP-compliant ASCII-armored format.
+
+ PGPMessage implements the `__bytes__` method, the output of which will be the message composition in
+ OpenPGP-compliant binary format.
+
+ Any signatures within the PGPMessage that are marked as being non-exportable will not be included in the output
+ of either of those methods.
+ """
+ super(PGPMessage, self).__init__()
+ self._compression = CompressionAlgorithm.Uncompressed
+ self._message = None
+ self._mdc = None
+ self._signatures = SorteDeque()
+ self._sessionkeys = []
+
+ def __bytearray__(self):
+ if self.is_compressed:
+ comp = CompressedData()
+ comp.calg = self._compression
+ comp.packets = [pkt for pkt in self]
+ comp.update_hlen()
+ return comp.__bytearray__()
+
+ _bytes = bytearray()
+ for pkt in self:
+ _bytes += pkt.__bytearray__()
+ return _bytes
+
+ def __str__(self):
+ if self.type == 'cleartext':
+ tmpl = u"-----BEGIN PGP SIGNED MESSAGE-----\n" \
+ u"{hhdr:s}\n" \
+ u"{cleartext:s}\n" \
+ u"{signature:s}"
+
+ # only add a Hash: header if we actually have at least one signature
+ hashes = set(s.hash_algorithm.name for s in self.signatures)
+ hhdr = 'Hash: {hashes:s}\n'.format(hashes=','.join(sorted(hashes))) if hashes else ''
+
+ return tmpl.format(hhdr=hhdr,
+ cleartext=self.dash_escape(self.bytes_to_text(self._message)),
+ signature=super(PGPMessage, self).__str__())
+
+ return super(PGPMessage, self).__str__()
+
+ def __iter__(self):
+ if self.type == 'cleartext':
+ for sig in self._signatures:
+ yield sig
+
+ elif self.is_encrypted:
+ for pkt in self._sessionkeys:
+ yield pkt
+ yield self.message
+
+ else:
+ ##TODO: is it worth coming up with a way of disabling one-pass signing?
+ for sig in self._signatures:
+ ops = sig.make_onepass()
+ if sig is not self._signatures[-1]:
+ ops.nested = True
+ yield ops
+
+ yield self._message
+ if self._mdc is not None: # pragma: no cover
+ yield self._mdc
+
+ for sig in self._signatures:
+ yield sig
+
+ def __or__(self, other):
+ if isinstance(other, Marker):
+ return self
+
+ if isinstance(other, CompressedData):
+ self._compression = other.calg
+ for pkt in other.packets:
+ self |= pkt
+ return self
+
+ if isinstance(other, (six.string_types, six.binary_type, bytearray)):
+ if self._message is None:
+ self._message = self.text_to_bytes(other)
+ return self
+
+ if isinstance(other, (LiteralData, SKEData, IntegrityProtectedSKEData)):
+ if self._message is None:
+ self._message = other
+ return self
+
+ if isinstance(other, MDC):
+ if self._mdc is None:
+ self._mdc = other
+ return self
+
+ if isinstance(other, OnePassSignature):
+ # these are "generated" on the fly during composition
+ return self
+
+ if isinstance(other, Signature):
+ other = PGPSignature() | other
+
+ if isinstance(other, PGPSignature):
+ self._signatures.insort(other)
+ return self
+
+ if isinstance(other, (PKESessionKey, SKESessionKey)):
+ self._sessionkeys.append(other)
+ return self
+
+ if isinstance(other, PGPMessage):
+ self._message = other._message
+ self._mdc = other._mdc
+ self._compression = other._compression
+ self._sessionkeys += other._sessionkeys
+ self._signatures += other._signatures
+ return self
+
+ raise NotImplementedError(str(type(other)))
+
+ def __copy__(self):
+ msg = super(PGPMessage, self).__copy__()
+ msg._compression = self._compression
+ msg._message = copy.copy(self._message)
+ msg._mdc = copy.copy(self._mdc)
+
+ for sig in self._signatures:
+ msg |= copy.copy(sig)
+
+ for sk in self._sessionkeys:
+ msg |= copy.copy(sk)
+
+ return msg
+
+ @classmethod
+ def new(cls, message, **kwargs):
+ """
+ Create a new PGPMessage object.
+
+ :param message: The message to be stored.
+ :type message: ``str``, ``unicode``, ``bytes``, ``bytearray``
+ :returns: :py:obj:`PGPMessage`
+
+ The following optional keyword arguments can be used with :py:meth:`PGPMessage.new`:
+
+ :keyword file: if True, ``message`` should be a path to a file. The contents of that file will be read and used
+ as the contents of the message.
+ :type file: ``bool``
+ :keyword cleartext: if True, the message will be cleartext with inline signatures.
+ :type cleartext: ``bool``
+ :keyword sensitive: if True, the filename will be set to '_CONSOLE' to signal other OpenPGP clients to treat
+ this message as being 'for your eyes only'. Ignored if cleartext is True.
+ :type sensitive: ``bool``
+ :keyword format: Set the message format identifier. Ignored if cleartext is True.
+ :type format: ``str``
+ :keyword compression: Set the compression algorithm for the new message.
+ Defaults to :py:obj:`CompressionAlgorithm.ZIP`. Ignored if cleartext is True.
+ :keyword encoding: Set the Charset header for the message.
+ :type encoding: ``str`` representing a valid codec in codecs
+ """
+ # TODO: have 'codecs' above (in :type encoding:) link to python documentation page on codecs
+ cleartext = kwargs.pop('cleartext', False)
+ format = kwargs.pop('format', None)
+ sensitive = kwargs.pop('sensitive', False)
+ compression = kwargs.pop('compression', CompressionAlgorithm.ZIP)
+ file = kwargs.pop('file', False)
+ charset = kwargs.pop('encoding', None)
+
+ filename = ''
+ mtime = datetime.utcnow()
+
+ msg = PGPMessage()
+
+ if charset:
+ msg.charset = charset
+
+ # if format in 'tu' and isinstance(message, (six.binary_type, bytearray)):
+ # # if message format is text or unicode and we got binary data, we'll need to transcode it to UTF-8
+ # message =
+
+ if file and os.path.isfile(message):
+ filename = message
+ message = bytearray(os.path.getsize(filename))
+ mtime = datetime.utcfromtimestamp(os.path.getmtime(filename))
+
+ with open(filename, 'rb') as mf:
+ mf.readinto(message)
+
+ # if format is None, we can try to detect it
+ if format is None:
+ if isinstance(message, six.text_type):
+ # message is definitely UTF-8 already
+ format = 'u'
+
+ elif cls.is_ascii(message):
+ # message is probably text
+ format = 't'
+
+ else:
+ # message is probably binary
+ format = 'b'
+
+ # if message is a binary type and we're building a textual message, we need to transcode the bytes to UTF-8
+ if isinstance(message, (six.binary_type, bytearray)) and (cleartext or format in 'tu'):
+ message = message.decode(charset or 'utf-8')
+
+ if cleartext:
+ msg |= message
+
+ else:
+ # load literal data
+ lit = LiteralData()
+ lit._contents = bytearray(msg.text_to_bytes(message))
+ lit.filename = '_CONSOLE' if sensitive else os.path.basename(filename)
+ lit.mtime = mtime
+ lit.format = format
+
+ # if cls.is_ascii(message):
+ # lit.format = 't'
+
+ lit.update_hlen()
+
+ msg |= lit
+ msg._compression = compression
+
+ return msg
+
+ def encrypt(self, passphrase, sessionkey=None, **prefs):
+ """
+ Encrypt the contents of this message using a passphrase.
+ :param passphrase: The passphrase to use for encrypting this message.
+ :type passphrase: ``str``, ``unicode``, ``bytes``
+
+ :optional param sessionkey: Provide a session key to use when encrypting something. Default is ``None``.
+ If ``None``, a session key of the appropriate length will be generated randomly.
+
+ .. warning::
+
+ Care should be taken when making use of this option! Session keys *absolutely need*
+ to be unpredictable! Use the ``gen_key()`` method on the desired
+ :py:obj:`~constants.SymmetricKeyAlgorithm` to generate the session key!
+
+ :type sessionkey: ``bytes``, ``str``
+ :raises: :py:exc:`~errors.PGPEncryptionError`
+ :returns: A new :py:obj:`PGPMessage` containing the encrypted contents of this message.
+ """
+ cipher_algo = prefs.pop('cipher', SymmetricKeyAlgorithm.AES256)
+ hash_algo = prefs.pop('hash', HashAlgorithm.SHA256)
+
+ # set up a new SKESessionKeyV4
+ skesk = SKESessionKeyV4()
+ skesk.s2k.usage = 255
+ skesk.s2k.specifier = 3
+ skesk.s2k.halg = hash_algo
+ skesk.s2k.encalg = cipher_algo
+ skesk.s2k.count = skesk.s2k.halg.tuned_count
+
+ if sessionkey is None:
+ sessionkey = cipher_algo.gen_key()
+ skesk.encrypt_sk(passphrase, sessionkey)
+ del passphrase
+
+ msg = PGPMessage() | skesk
+
+ if not self.is_encrypted:
+ skedata = IntegrityProtectedSKEDataV1()
+ skedata.encrypt(sessionkey, cipher_algo, self.__bytes__())
+ msg |= skedata
+
+ else:
+ msg |= self
+
+ return msg
+
+ def decrypt(self, passphrase):
+ """
+ Attempt to decrypt this message using a passphrase.
+
+ :param passphrase: The passphrase to use to attempt to decrypt this message.
+ :type passphrase: ``str``, ``unicode``, ``bytes``
+ :raises: :py:exc:`~errors.PGPDecryptionError` if decryption failed for any reason.
+ :returns: A new :py:obj:`PGPMessage` containing the decrypted contents of this message
+ """
+ if not self.is_encrypted:
+ raise PGPError("This message is not encrypted!")
+
+ for skesk in iter(sk for sk in self._sessionkeys if isinstance(sk, SKESessionKey)):
+ try:
+ symalg, key = skesk.decrypt_sk(passphrase)
+ decmsg = PGPMessage()
+ decmsg.parse(self.message.decrypt(key, symalg))
+
+ except (TypeError, ValueError, NotImplementedError, PGPDecryptionError):
+ continue
+
+ else:
+ del passphrase
+ break
+
+ else:
+ raise PGPDecryptionError("Decryption failed")
+
+ return decmsg
+
+ def parse(self, packet):
+ unarmored = self.ascii_unarmor(packet)
+ data = unarmored['body']
+
+ if unarmored['magic'] is not None and unarmored['magic'] not in ['MESSAGE', 'SIGNATURE']:
+ raise ValueError('Expected: MESSAGE. Got: {}'.format(str(unarmored['magic'])))
+
+ if unarmored['headers'] is not None:
+ self.ascii_headers = unarmored['headers']
+
+ # cleartext signature
+ if unarmored['magic'] == 'SIGNATURE':
+ # the composition for this will be the 'cleartext' as a str,
+ # followed by one or more signatures (each one loaded into a PGPSignature)
+ self |= self.dash_unescape(unarmored['cleartext'])
+ while len(data) > 0:
+ pkt = Packet(data)
+ if not isinstance(pkt, Signature): # pragma: no cover
+ warnings.warn("Discarded unexpected packet: {:s}".format(pkt.__class__.__name__), stacklevel=2)
+ continue
+ self |= PGPSignature() | pkt
+
+ else:
+ while len(data) > 0:
+ self |= Packet(data)
+
+
+class PGPKey(Armorable, ParentRef, PGPObject):
+ """
+ 11.1. Transferable Public Keys
+
+ OpenPGP users may transfer public keys. The essential elements of a
+ transferable public key are as follows:
+
+ - One Public-Key packet
+
+ - Zero or more revocation signatures
+ - One or more User ID packets
+
+ - After each User ID packet, zero or more Signature packets
+ (certifications)
+
+ - Zero or more User Attribute packets
+
+ - After each User Attribute packet, zero or more Signature packets
+ (certifications)
+
+ - Zero or more Subkey packets
+
+ - After each Subkey packet, one Signature packet, plus optionally a
+ revocation
+
+ The Public-Key packet occurs first. Each of the following User ID
+ packets provides the identity of the owner of this public key. If
+ there are multiple User ID packets, this corresponds to multiple
+ means of identifying the same unique individual user; for example, a
+ user may have more than one email address, and construct a User ID
+ for each one.
+
+ Immediately following each User ID packet, there are zero or more
+ Signature packets. Each Signature packet is calculated on the
+ immediately preceding User ID packet and the initial Public-Key
+ packet. The signature serves to certify the corresponding public key
+ and User ID. In effect, the signer is testifying to his or her
+ belief that this public key belongs to the user identified by this
+ User ID.
+
+ Within the same section as the User ID packets, there are zero or
+ more User Attribute packets. Like the User ID packets, a User
+ Attribute packet is followed by zero or more Signature packets
+ calculated on the immediately preceding User Attribute packet and the
+ initial Public-Key packet.
+
+ User Attribute packets and User ID packets may be freely intermixed
+ in this section, so long as the signatures that follow them are
+ maintained on the proper User Attribute or User ID packet.
+
+ After the User ID packet or Attribute packet, there may be zero or
+ more Subkey packets. In general, subkeys are provided in cases where
+ the top-level public key is a signature-only key. However, any V4
+ key may have subkeys, and the subkeys may be encryption-only keys,
+ signature-only keys, or general-purpose keys. V3 keys MUST NOT have
+ subkeys.
+
+ Each Subkey packet MUST be followed by one Signature packet, which
+ should be a subkey binding signature issued by the top-level key.
+ For subkeys that can issue signatures, the subkey binding signature
+ MUST contain an Embedded Signature subpacket with a primary key
+ binding signature (0x19) issued by the subkey on the top-level key.
+
+ Subkey and Key packets may each be followed by a revocation Signature
+ packet to indicate that the key is revoked. Revocation signatures
+ are only accepted if they are issued by the key itself, or by a key
+ that is authorized to issue revocations via a Revocation Key
+ subpacket in a self-signature by the top-level key.
+
+ Transferable public-key packet sequences may be concatenated to allow
+ transferring multiple public keys in one operation.
+
+ 11.2. Transferable Secret Keys
+
+ OpenPGP users may transfer secret keys. The format of a transferable
+ secret key is the same as a transferable public key except that
+ secret-key and secret-subkey packets are used instead of the public
+ key and public-subkey packets. Implementations SHOULD include self-
+ signatures on any user IDs and subkeys, as this allows for a complete
+ public key to be automatically extracted from the transferable secret
+ key. Implementations MAY choose to omit the self-signatures,
+ especially if a transferable public key accompanies the transferable
+ secret key.
+ """
+ @property
+ def __key__(self):
+ return self._key.keymaterial
+
+ @property
+ def __sig__(self):
+ return list(self._signatures)
+
+ @property
+ def created(self):
+ """A :py:obj:`~datetime.datetime` object of the creation date and time of the key, in UTC."""
+ return self._key.created
+
+ @property
+ def expires_at(self):
+ """A :py:obj:`~datetime.datetime` object of when this key is to be considered expired, if any. Otherwise, ``None``"""
+ try:
+ expires = min(sig.key_expiration for sig in itertools.chain(iter(uid.selfsig for uid in self.userids), self.self_signatures)
+ if sig.key_expiration is not None)
+
+ except ValueError:
+ return None
+
+ else:
+ return (self.created + expires)
+
+ @property
+ def fingerprint(self):
+ """The fingerprint of this key, as a :py:obj:`~pgpy.types.Fingerprint` object."""
+ if self._key:
+ return self._key.fingerprint
+
+ @property
+ def hashdata(self):
+ # when signing a key, only the public portion of the keys is hashed
+ # if this is a private key, the private components of the key material need to be left out
+ if self.is_public:
+ return self._key.__bytearray__()[len(self._key.header):]
+
+ pub = self._key.pubkey()
+ return pub.__bytearray__()[len(pub.header):]
+
+ @property
+ def is_expired(self):
+ """``True`` if this key is expired, otherwise ``False``"""
+ expires = self.expires_at
+ if expires is not None:
+ return expires <= datetime.utcnow()
+
+ return False
+
+ @property
+ def is_primary(self):
+ """``True`` if this is a primary key; ``False`` if this is a subkey"""
+ return isinstance(self._key, Primary) and not isinstance(self._key, Sub)
+
+ @property
+ def is_protected(self):
+ """``True`` if this is a private key that is protected with a passphrase, otherwise ``False``"""
+ if self.is_public:
+ return False
+
+ return self._key.protected
+
+ @property
+ def is_public(self):
+ """``True`` if this is a public key, otherwise ``False``"""
+ return isinstance(self._key, Public) and not isinstance(self._key, Private)
+
+ @property
+ def is_unlocked(self):
+ """``False`` if this is a private key that is protected with a passphrase and has not yet been unlocked, otherwise ``True``"""
+ if self.is_public:
+ return True
+
+ if not self.is_protected:
+ return True
+
+ return self._key.unlocked
+
+ @property
+ def key_algorithm(self):
+ """The :py:obj:`constants.PubKeyAlgorithm` pertaining to this key"""
+ return self._key.pkalg
+
+ @property
+ def key_size(self):
+ """*new in 0.4.1*
+ The size pertaining to this key. ``int`` for non-EC key algorithms; :py:obj:`constants.EllipticCurveOID` for EC keys.
+ """
+ if self.key_algorithm in {PubKeyAlgorithm.ECDSA, PubKeyAlgorithm.ECDH}:
+ return self._key.keymaterial.oid
+ return next(iter(self._key.keymaterial)).bit_length()
+
+ @property
+ def magic(self):
+ return '{:s} KEY BLOCK'.format('PUBLIC' if (isinstance(self._key, Public) and not isinstance(self._key, Private)) else
+ 'PRIVATE' if isinstance(self._key, Private) else '')
+
+ @property
+ def pubkey(self):
+ """If the :py:obj:`PGPKey` object is a private key, this method returns a corresponding public key object with
+ all the trimmings. Otherwise, returns ``None``
+ """
+ if not self.is_public:
+ if self._sibling is None or isinstance(self._sibling, weakref.ref):
+ # create a new key shell
+ pub = PGPKey()
+ pub.ascii_headers = self.ascii_headers.copy()
+
+ # get the public half of the primary key
+ pub._key = self._key.pubkey()
+
+ # get the public half of each subkey
+ for skid, subkey in self.subkeys.items():
+ pub.subkeys[skid] = subkey.pubkey
+
+ # copy user ids and user attributes
+ for uid in self._uids:
+ pub |= copy.copy(uid)
+
+ # copy signatures that weren't copied with uids
+ for sig in self._signatures:
+ if sig.parent is None:
+ pub |= copy.copy(sig)
+
+ # keep connect the two halves using a weak reference
+ self._sibling = weakref.ref(pub)
+ pub._sibling = weakref.ref(self)
+
+ return self._sibling()
+ return None
+
+ @pubkey.setter
+ def pubkey(self, pubkey):
+ if self.is_public:
+ raise TypeError("cannot add public sibling to pubkey")
+
+ if not pubkey.is_public:
+ raise TypeError("sibling must be public")
+
+ if self._sibling is not None and self._sibling() is not None:
+ raise ValueError("public key reference already set")
+
+ if pubkey.fingerprint != self.fingerprint:
+ raise ValueError("key fingerprint mismatch")
+
+ # TODO: sync packets with sibling
+ self._sibling = weakref.ref(pubkey)
+ pubkey._sibling = weakref.ref(self)
+
+ @property
+ def self_signatures(self):
+ keyid, keytype = (self.fingerprint.keyid, SignatureType.DirectlyOnKey) if self.is_primary \
+ else (self.parent.fingerprint.keyid, SignatureType.Subkey_Binding)
+
+ ##TODO: filter out revoked signatures as well
+ for sig in iter(sig for sig in self._signatures
+ if all([sig.type == keytype, sig.signer == keyid, not sig.is_expired])):
+ yield sig
+
+ @property
+ def signers(self):
+ """A ``set`` of key ids of keys that were used to sign this key"""
+ return {sig.signer for sig in self.__sig__}
+
+ @property
+ def subkeys(self):
+ """An :py:obj:`~collections.OrderedDict` of subkeys bound to this primary key, if applicable,
+ selected by 16-character keyid."""
+ return self._children
+
+ @property
+ def userids(self):
+ """A ``list`` of :py:obj:`PGPUID` objects containing User ID information about this key"""
+ return [ u for u in self._uids if u.is_uid ]
+
+ @property
+ def userattributes(self):
+ """A ``list`` of :py:obj:`PGPUID` objects containing one or more images associated with this key"""
+ return [u for u in self._uids if u.is_ua]
+
+ @classmethod
+ def new(cls, key_algorithm, key_size):
+ """
+ Generate a new PGP key
+
+ :param key_algorithm: Key algorithm to use.
+ :type key_algorithm: A :py:obj:`~constants.PubKeyAlgorithm`
+ :param key_size: Key size in bits, unless `key_algorithm` is :py:obj:`~constants.PubKeyAlgorithm.ECDSA` or
+ :py:obj:`~constants.PubKeyAlgorithm.ECDH`, in which case it should be the Curve OID to use.
+ :type key_size: ``int`` or :py:obj:`~constants.EllipticCurveOID`
+ :return: A newly generated :py:obj:`PGPKey`
+ """
+ # new private key shell first
+ key = PGPKey()
+
+ if key_algorithm in {PubKeyAlgorithm.RSAEncrypt, PubKeyAlgorithm.RSASign}: # pragma: no cover
+ warnings.warn('{:s} is deprecated - generating key using RSAEncryptOrSign'.format(key_algorithm.name))
+ key_algorithm = PubKeyAlgorithm.RSAEncryptOrSign
+
+ # generate some key data to match key_algorithm and key_size
+ key._key = PrivKeyV4.new(key_algorithm, key_size)
+
+ return key
+
+ def __init__(self):
+ """
+ PGPKey objects represent OpenPGP compliant keys along with all of their associated data.
+
+ PGPKey implements the `__str__` method, the output of which will be the key composition in
+ OpenPGP-compliant ASCII-armored format.
+
+ PGPKey implements the `__bytes__` method, the output of which will be the key composition in
+ OpenPGP-compliant binary format.
+
+ Any signatures within the PGPKey that are marked as being non-exportable will not be included in the output
+ of either of those methods.
+ """
+ super(PGPKey, self).__init__()
+ self._key = None
+ self._children = collections.OrderedDict()
+ self._signatures = SorteDeque()
+ self._uids = SorteDeque()
+ self._sibling = None
+
+ def __bytearray__(self):
+ _bytes = bytearray()
+ # us
+ _bytes += self._key.__bytearray__()
+ # our signatures; ignore embedded signatures
+ for sig in iter(s for s in self._signatures if not s.embedded and s.exportable):
+ _bytes += sig.__bytearray__()
+ # one or more User IDs, followed by their signatures
+ for uid in self._uids:
+ _bytes += uid._uid.__bytearray__()
+ for s in [s for s in uid._signatures if s.exportable]:
+ _bytes += s.__bytearray__()
+ # subkeys
+ for sk in self._children.values():
+ _bytes += sk.__bytearray__()
+
+ return _bytes
+
+ def __repr__(self):
+ if self._key is not None:
+ return "<PGPKey [{:s}][0x{:s}] at 0x{:02X}>" \
+ "".format(self._key.__class__.__name__, self.fingerprint.keyid, id(self))
+
+ return "<PGPKey [unknown] at 0x{:02X}>" \
+ "".format(id(self))
+
+ def __contains__(self, item):
+ if isinstance(item, PGPKey): # pragma: no cover
+ return item.fingerprint.keyid in self.subkeys
+
+ if isinstance(item, Fingerprint): # pragma: no cover
+ return item.keyid in self.subkeys
+
+ if isinstance(item, PGPUID):
+ return item in self._uids
+
+ if isinstance(item, PGPSignature):
+ return item in self._signatures
+
+ raise TypeError
+
+ def __or__(self, other, from_sib=False):
+ if isinstance(other, Key) and self._key is None:
+ self._key = other
+
+ elif isinstance(other, PGPKey) and not other.is_primary and other.is_public == self.is_public:
+ other._parent = self
+ self._children[other.fingerprint.keyid] = other
+
+ elif isinstance(other, PGPSignature):
+ self._signatures.insort(other)
+
+ # if this is a subkey binding signature that has embedded primary key binding signatures, add them to parent
+ if other.type == SignatureType.Subkey_Binding:
+ for es in iter(pkb for pkb in other._signature.subpackets['EmbeddedSignature']):
+ esig = PGPSignature() | es
+ esig._parent = other
+ self._signatures.insort(esig)
+
+ elif isinstance(other, PGPUID):
+ other._parent = weakref.ref(self)
+ self._uids.insort(other)
+
+ else:
+ raise TypeError("unsupported operand type(s) for |: '{:s}' and '{:s}'"
+ "".format(self.__class__.__name__, other.__class__.__name__))
+
+ if isinstance(self._sibling, weakref.ref) and not from_sib:
+ sib = self._sibling()
+ if sib is None:
+ self._sibling = None
+
+ else: # pragma: no cover
+ sib.__or__(copy.copy(other), True)
+
+ return self
+
+ def __copy__(self):
+ key = super(PGPKey, self).__copy__()
+ key._key = copy.copy(self._key)
+
+ for uid in self._uids:
+ key |= copy.copy(uid)
+
+ for id, subkey in self._children.items():
+ key |= copy.copy(subkey)
+
+ for sig in self._signatures:
+ if sig.embedded:
+ # embedded signatures don't need to be explicitly copied
+ continue
+
+ print(len(key._signatures))
+ key |= copy.copy(sig)
+
+ return key
+
+ def protect(self, passphrase, enc_alg, hash_alg):
+ """
+ Add a passphrase to a private key. If the key is already passphrase protected, it should be unlocked before
+ a new passphrase can be specified.
+
+ Has no effect on public keys.
+
+ :param passphrase: A passphrase to protect the key with
+ :type passphrase: ``str``, ``unicode``
+ :param enc_alg: Symmetric encryption algorithm to use to protect the key
+ :type enc_alg: :py:obj:`~constants.SymmetricKeyAlgorithm`
+ :param hash_alg: Hash algorithm to use in the String-to-Key specifier
+ :type hash_alg: :py:obj:`~constants.HashAlgorithm`
+ """
+ ##TODO: specify strong defaults for enc_alg and hash_alg
+ if self.is_public:
+ # we can't protect public keys because only private key material is ever protected
+ warnings.warn("Public keys cannot be passphrase-protected", stacklevel=2)
+ return
+
+ if self.is_protected and not self.is_unlocked:
+ # we can't protect a key that is already protected unless it is unlocked first
+ warnings.warn("This key is already protected with a passphrase - "
+ "please unlock it before attempting to specify a new passphrase", stacklevel=2)
+ return
+
+ for sk in itertools.chain([self], self.subkeys.values()):
+ sk._key.protect(passphrase, enc_alg, hash_alg)
+
+ del passphrase
+
+ @contextlib.contextmanager
+ def unlock(self, passphrase):
+ """
+ Context manager method for unlocking passphrase-protected private keys. Has no effect if the key is not both
+ private and passphrase-protected.
+
+ When the context managed block is exited, the unprotected private key material is removed.
+
+ Example::
+
+ privkey = PGPKey()
+ privkey.parse(keytext)
+
+ assert privkey.is_protected
+ assert privkey.is_unlocked is False
+ # privkey.sign("some text") <- this would raise an exception
+
+ with privkey.unlock("TheCorrectPassphrase"):
+ # privkey is now unlocked
+ assert privkey.is_unlocked
+ # so you can do things with it
+ sig = privkey.sign("some text")
+
+ # privkey is no longer unlocked
+ assert privkey.is_unlocked is False
+
+ Emits a :py:obj:`~warnings.UserWarning` if the key is public or not passphrase protected.
+
+ :param str passphrase: The passphrase to be used to unlock this key.
+ :raises: :py:exc:`~pgpy.errors.PGPDecryptionError` if the passphrase is incorrect
+ """
+ if self.is_public:
+ # we can't unprotect public keys because only private key material is ever protected
+ warnings.warn("Public keys cannot be passphrase-protected", stacklevel=3)
+ yield self
+ return
+
+ if not self.is_protected:
+ # we can't unprotect private keys that are not protected, because there is no ciphertext to decrypt
+ warnings.warn("This key is not protected with a passphrase", stacklevel=3)
+ yield self
+ return
+
+ try:
+ for sk in itertools.chain([self], self.subkeys.values()):
+ sk._key.unprotect(passphrase)
+ del passphrase
+ yield self
+
+ finally:
+ # clean up here by deleting the previously decrypted secret key material
+ for sk in itertools.chain([self], self.subkeys.values()):
+ sk._key.keymaterial.clear()
+
+ def add_uid(self, uid, selfsign=True, **prefs):
+ """
+ Add a User ID to this key.
+
+ :param uid: The user id to add
+ :type uid: :py:obj:`~pgpy.PGPUID`
+ :param selfsign: Whether or not to self-sign the user id before adding it
+ :type selfsign: ``bool``
+
+ Valid optional keyword arguments are identical to those of self-signatures for :py:meth:`PGPKey.certify`.
+ Any such keyword arguments are ignored if selfsign is ``False``
+ """
+ uid._parent = self
+ if selfsign:
+ uid |= self.certify(uid, SignatureType.Positive_Cert, **prefs)
+
+ self |= uid
+
+ def get_uid(self, search):
+ """
+ Find and return a User ID that matches the search string given.
+
+ :param search: A text string to match name, comment, or email address against
+ :type search: ``str``, ``unicode``
+ :return: The first matching :py:obj:`~pgpy.PGPUID`, or ``None`` if no matches were found.
+ """
+ if self.is_primary:
+ return next((u for u in self._uids if search in filter(lambda a: a is not None, (u.name, u.comment, u.email))), None)
+ return self.parent.get_uid(search)
+
+ def del_uid(self, search):
+ """
+ Find and remove a user id that matches the search string given. This method does not modify the corresponding
+ :py:obj:`~pgpy.PGPUID` object; it only removes it from the list of user ids on the key.
+
+ :param search: A text string to match name, comment, or email address against
+ :type search: ``str``, ``unicode``
+ """
+ u = self.get_uid(search)
+
+ if u is None:
+ raise KeyError("uid '{:s}' not found".format(search))
+
+ u._parent = None
+ self._uids.remove(u)
+
+ def add_subkey(self, key, **prefs):
+ """
+ Add a key as a subkey to this key.
+ :param key: A private :py:obj:`~pgpy.PGPKey` that does not have any subkeys of its own
+
+ :keyword usage: A ``set`` of key usage flags, as :py:obj:`~constants.KeyFlags` for the subkey to be added.
+ :type usage: ``set``
+
+ Other valid optional keyword arguments are identical to those of self-signatures for :py:meth:`PGPKey.certify`
+ """
+ if self.is_public:
+ raise PGPError("Cannot add a subkey to a public key. Add the subkey to the private component first!")
+
+ if key.is_public:
+ raise PGPError("Cannot add a public key as a subkey to this key")
+
+ if key.is_primary:
+ if len(key._children) > 0:
+ raise PGPError("Cannot add a key that already has subkeys as a subkey!")
+
+ # convert key into a subkey
+ npk = PrivSubKeyV4()
+ npk.pkalg = key._key.pkalg
+ npk.created = key._key.created
+ npk.keymaterial = key._key.keymaterial
+ key._key = npk
+ key._key.update_hlen()
+
+ self._children[key.fingerprint.keyid] = key
+ key._parent = self
+
+ ##TODO: skip this step if the key already has a subkey binding signature
+ bsig = self.bind(key, **prefs)
+ key |= bsig
+
+ def _get_key_flags(self, user=None):
+ if self.is_primary:
+ if user is not None:
+ user = self.get_uid(user)
+
+ elif len(self._uids) == 0:
+ return {KeyFlags.Certify}
+
+ else:
+ user = next(iter(self.userids))
+
+ # RFC 4880 says that primary keys *must* be capable of certification
+ return {KeyFlags.Certify} | user.selfsig.key_flags
+
+ return next(self.self_signatures).key_flags
+
+ def _sign(self, subject, sig, **prefs):
+ """
+ The actual signing magic happens here.
+ :param subject: The subject to sign
+ :param sig: The :py:obj:`PGPSignature` object the new signature is to be encapsulated within
+ :returns: ``sig``, after the signature is added to it.
+ """
+ user = prefs.pop('user', None)
+ uid = None
+ if user is not None:
+ uid = self.get_uid(user)
+
+ else:
+ uid = next(iter(self.userids), None)
+ if uid is None and self.parent is not None:
+ uid = next(iter(self.parent.userids), None)
+
+ if sig.hash_algorithm is None:
+ sig._signature.halg = uid.selfsig.hashprefs[0]
+
+ if uid is not None and sig.hash_algorithm not in uid.selfsig.hashprefs:
+ warnings.warn("Selected hash algorithm not in key preferences", stacklevel=4)
+
+ # signature options that can be applied at any level
+ expires = prefs.pop('expires', None)
+ notation = prefs.pop('notation', None)
+ revocable = prefs.pop('revocable', True)
+ policy_uri = prefs.pop('policy_uri', None)
+
+ if expires is not None:
+ # expires should be a timedelta, so if it's a datetime, turn it into a timedelta
+ if isinstance(expires, datetime):
+ expires = expires - self.created
+
+ sig._signature.subpackets.addnew('SignatureExpirationTime', hashed=True, expires=expires)
+
+ if revocable is False:
+ sig._signature.subpackets.addnew('Revocable', hashed=True, bflag=revocable)
+
+ if notation is not None:
+ for name, value in notation.items():
+ # mark all notations as human readable unless value is a bytearray
+ flags = NotationDataFlags.HumanReadable
+ if isinstance(value, bytearray):
+ flags = 0x00
+
+ sig._signature.subpackets.addnew('NotationData', hashed=True, flags=flags, name=name, value=value)
+
+ if policy_uri is not None:
+ sig._signature.subpackets.addnew('Policy', hashed=True, uri=policy_uri)
+
+ if user is not None and uid is not None:
+ signers_uid = "{:s}".format(uid)
+ sig._signature.subpackets.addnew('SignersUserID', hashed=True, userid=signers_uid)
+
+ # handle an edge case for timestamp signatures vs standalone signatures
+ if sig.type == SignatureType.Timestamp and len(sig._signature.subpackets._hashed_sp) > 1:
+ sig._signature.sigtype = SignatureType.Standalone
+
+ sigdata = sig.hashdata(subject)
+ h2 = sig.hash_algorithm.hasher
+ h2.update(sigdata)
+ sig._signature.hash2 = bytearray(h2.digest()[:2])
+
+ _sig = self._key.sign(sigdata, getattr(hashes, sig.hash_algorithm.name)())
+ if _sig is NotImplemented:
+ raise NotImplementedError(self.key_algorithm)
+
+ sig._signature.signature.from_signer(_sig)
+ sig._signature.update_hlen()
+
+ return sig
+
+ @KeyAction(KeyFlags.Sign, is_unlocked=True, is_public=False)
+ def sign(self, subject, **prefs):
+ """
+ Sign text, a message, or a timestamp using this key.
+
+ :param subject: The text to be signed
+ :type subject: ``str``, :py:obj:`~pgpy.PGPMessage`, ``None``
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is passphrase-protected and has not been unlocked
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is public
+ :returns: :py:obj:`PGPSignature`
+
+ The following optional keyword arguments can be used with :py:meth:`PGPKey.sign`, as well as
+ :py:meth:`PGPKey.certify`, :py:meth:`PGPKey.revoke`, and :py:meth:`PGPKey.bind`:
+
+ :keyword expires: Set an expiration date for this signature
+ :type expires: :py:obj:`~datetime.datetime`, :py:obj:`~datetime.timedelta`
+ :keyword notation: Add arbitrary notation data to this signature.
+ :type notation: ``dict``
+ :keyword policy_uri: Add a URI to the signature that should describe the policy under which the signature
+ was issued.
+ :type policy_uri: ``str``
+ :keyword revocable: If ``False``, this signature will be marked non-revocable
+ :type revocable: ``bool``
+ :keyword user: Specify which User ID to use when creating this signature. Also adds a "Signer's User ID"
+ to the signature.
+ :type user: ``str``
+ """
+ sig_type = SignatureType.BinaryDocument
+ hash_algo = prefs.pop('hash', None)
+
+ if subject is None:
+ sig_type = SignatureType.Timestamp
+
+ if isinstance(subject, PGPMessage):
+ if subject.type == 'cleartext':
+ sig_type = SignatureType.CanonicalDocument
+
+ subject = subject.message
+
+ sig = PGPSignature.new(sig_type, self.key_algorithm, hash_algo, self.fingerprint.keyid)
+
+ return self._sign(subject, sig, **prefs)
+
+ @KeyAction(KeyFlags.Certify, is_unlocked=True, is_public=False)
+ def certify(self, subject, level=SignatureType.Generic_Cert, **prefs):
+ """
+ Sign a key or a user id within a key.
+
+ :param subject: The user id or key to be certified.
+ :type subject: :py:obj:`PGPKey`, :py:obj:`PGPUID`
+ :param level: :py:obj:`~constants.SignatureType.Generic_Cert`, :py:obj:`~constants.SignatureType.Persona_Cert`,
+ :py:obj:`~constants.SignatureType.Casual_Cert`, or :py:obj:`~constants.SignatureType.Positive_Cert`.
+ Only used if subject is a :py:obj:`PGPUID`; otherwise, it is ignored.
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is passphrase-protected and has not been unlocked
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is public
+ :returns: :py:obj:`PGPSignature`
+
+ In addition to the optional keyword arguments accepted by :py:meth:`PGPKey.sign`, the following optional
+ keyword arguments can be used with :py:meth:`PGPKey.certify`.
+
+ These optional keywords only make sense, and thus only have an effect, when self-signing a key or User ID:
+
+ :keyword usage: A ``set`` of key usage flags, as :py:obj:`~constants.KeyFlags`.
+ This keyword is ignored for non-self-certifications.
+ :type usage: ``set``
+ :keyword ciphers: A list of preferred symmetric ciphers, as :py:obj:`~constants.SymmetricKeyAlgorithm`.
+ This keyword is ignored for non-self-certifications.
+ :type ciphers: ``list``
+ :keyword hashes: A list of preferred hash algorithms, as :py:obj:`~constants.HashAlgorithm`.
+ This keyword is ignored for non-self-certifications.
+ :type hashes: ``list``
+ :keyword compression: A list of preferred compression algorithms, as :py:obj:`~constants.CompressionAlgorithm`.
+ This keyword is ignored for non-self-certifications.
+ :type compression: ``list``
+ :keyword key_expiration: Specify a key expiration date for when this key should expire, or a
+ :py:obj:`~datetime.timedelta` of how long after the key was created it should expire.
+ This keyword is ignored for non-self-certifications.
+ :type key_expiration: :py:obj:`datetime.datetime`, :py:obj:`datetime.timedelta`
+ :keyword keyserver: Specify the URI of the preferred key server of the user.
+ This keyword is ignored for non-self-certifications.
+ :type keyserver: ``str``, ``unicode``, ``bytes``
+ :keyword primary: Whether or not to consider the certified User ID as the primary one.
+ This keyword is ignored for non-self-certifications, and any certifications directly on keys.
+ :type primary: ``bool``
+
+ These optional keywords only make sense, and thus only have an effect, when signing another key or User ID:
+
+ :keyword trust: Specify the level and amount of trust to assert when certifying a public key. Should be a tuple
+ of two ``int`` s, specifying the trust level and trust amount. See
+ `RFC 4880 Section 5.2.3.13. Trust Signature <http://tools.ietf.org/html/rfc4880#section-5.2.3.13>`_
+ for more on what these values mean.
+ :type trust: ``tuple`` of two ``int`` s
+ :keyword regex: Specify a regular expression to constrain the specified trust signature in the resulting signature.
+ Symbolically signifies that the specified trust signature only applies to User IDs which match
+ this regular expression.
+ This is meaningless without also specifying trust level and amount.
+ :type regex: ``str``
+ """
+ hash_algo = prefs.pop('hash', None)
+ sig_type = level
+ if isinstance(subject, PGPKey):
+ sig_type = SignatureType.DirectlyOnKey
+
+ sig = PGPSignature.new(sig_type, self.key_algorithm, hash_algo, self.fingerprint.keyid)
+
+ # signature options that only make sense in certifications
+ usage = prefs.pop('usage', None)
+ exportable = prefs.pop('exportable', None)
+
+ if usage is not None:
+ sig._signature.subpackets.addnew('KeyFlags', hashed=True, flags=usage)
+
+ if exportable is not None:
+ sig._signature.subpackets.addnew('ExportableCertification', hashed=True, bflag=exportable)
+
+ keyfp = self.fingerprint
+ if isinstance(subject, PGPKey):
+ keyfp = subject.fingerprint
+ if isinstance(subject, PGPUID) and subject._parent is not None:
+ keyfp = subject._parent.fingerprint
+
+ if keyfp == self.fingerprint:
+ # signature options that only make sense in self-certifications
+ cipher_prefs = prefs.pop('ciphers', None)
+ hash_prefs = prefs.pop('hashes', None)
+ compression_prefs = prefs.pop('compression', None)
+ key_expires = prefs.pop('key_expiration', None)
+ keyserver_flags = prefs.pop('keyserver_flags', None)
+ keyserver = prefs.pop('keyserver', None)
+ primary_uid = prefs.pop('primary', None)
+
+ if key_expires is not None:
+ # key expires should be a timedelta, so if it's a datetime, turn it into a timedelta
+ if isinstance(key_expires, datetime):
+ key_expires = key_expires - self.created
+
+ sig._signature.subpackets.addnew('KeyExpirationTime', hashed=True, expires=key_expires)
+
+ if cipher_prefs is not None:
+ sig._signature.subpackets.addnew('PreferredSymmetricAlgorithms', hashed=True, flags=cipher_prefs)
+
+ if hash_prefs is not None:
+ sig._signature.subpackets.addnew('PreferredHashAlgorithms', hashed=True, flags=hash_prefs)
+ if sig.hash_algorithm is None:
+ sig._signature.halg = hash_prefs[0]
+
+ if compression_prefs is not None:
+ sig._signature.subpackets.addnew('PreferredCompressionAlgorithms', hashed=True, flags=compression_prefs)
+
+ if keyserver_flags is not None:
+ sig._signature.subpackets.addnew('KeyServerPreferences', hashed=True, flags=keyserver_flags)
+
+ if keyserver is not None:
+ sig._signature.subpackets.addnew('PreferredKeyServer', hashed=True, uri=keyserver)
+
+ if primary_uid is not None:
+ sig._signature.subpackets.addnew('PrimaryUserID', hashed=True, primary=primary_uid)
+
+ # Features is always set on self-signatures
+ sig._signature.subpackets.addnew('Features', hashed=True, flags=Features.pgpy_features)
+
+ else:
+ # signature options that only make sense in non-self-certifications
+ trust = prefs.pop('trust', None)
+ regex = prefs.pop('regex', None)
+
+ if trust is not None:
+ sig._signature.subpackets.addnew('TrustSignature', hashed=True, level=trust[0], amount=trust[1])
+
+ if regex is not None:
+ sig._signature.subpackets.addnew('RegularExpression', hashed=True, regex=regex)
+
+ return self._sign(subject, sig, **prefs)
+
+ @KeyAction(KeyFlags.Certify, is_unlocked=True, is_public=False)
+ def revoke(self, target, **prefs):
+ """
+ Revoke a key, a subkey, or all current certification signatures of a User ID that were generated by this key so far.
+
+ :param target: The key to revoke
+ :type target: :py:obj:`PGPKey`, :py:obj:`PGPUID`
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is passphrase-protected and has not been unlocked
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is public
+ :returns: :py:obj:`PGPSignature`
+
+ In addition to the optional keyword arguments accepted by :py:meth:`PGPKey.sign`, the following optional
+ keyword arguments can be used with :py:meth:`PGPKey.revoke`.
+
+ :keyword reason: Defaults to :py:obj:`constants.RevocationReason.NotSpecified`
+ :type reason: One of :py:obj:`constants.RevocationReason`.
+ :keyword comment: Defaults to an empty string.
+ :type comment: ``str``
+ """
+ hash_algo = prefs.pop('hash', None)
+ if isinstance(target, PGPUID):
+ sig_type = SignatureType.CertRevocation
+
+ elif isinstance(target, PGPKey):
+ ##TODO: check to make sure that the key that is being revoked:
+ # - is this key
+ # - is one of this key's subkeys
+ # - specifies this key as its revocation key
+ if target.is_primary:
+ sig_type = SignatureType.KeyRevocation
+
+ else:
+ sig_type = SignatureType.SubkeyRevocation
+
+ else: # pragma: no cover
+ raise TypeError
+
+ sig = PGPSignature.new(sig_type, self.key_algorithm, hash_algo, self.fingerprint.keyid)
+
+ # signature options that only make sense when revoking
+ reason = prefs.pop('reason', RevocationReason.NotSpecified)
+ comment = prefs.pop('comment', "")
+ sig._signature.subpackets.addnew('ReasonForRevocation', hashed=True, code=reason, string=comment)
+
+ return self._sign(target, sig, **prefs)
+
+ @KeyAction(is_unlocked=True, is_public=False)
+ def revoker(self, revoker, **prefs):
+ """
+ Generate a signature that specifies another key as being valid for revoking this key.
+
+ :param revoker: The :py:obj:`PGPKey` to specify as a valid revocation key.
+ :type revoker: :py:obj:`PGPKey`
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is passphrase-protected and has not been unlocked
+ :raises: :py:exc:`~pgpy.errors.PGPError` if the key is public
+ :returns: :py:obj:`PGPSignature`
+
+ In addition to the optional keyword arguments accepted by :py:meth:`PGPKey.sign`, the following optional
+ keyword arguments can be used with :py:meth:`PGPKey.revoker`.
+
+ :keyword sensitive: If ``True``, this sets the sensitive flag on the RevocationKey subpacket. Currently,
+ this has no other effect.
+ :type sensitive: ``bool``
+ """
+ hash_algo = prefs.pop('hash', None)
+
+ sig = PGPSignature.new(SignatureType.DirectlyOnKey, self.key_algorithm, hash_algo, self.fingerprint.keyid)
+
+ # signature options that only make sense when adding a revocation key
+ sensitive = prefs.pop('sensitive', False)
+ keyclass = RevocationKeyClass.Normal | (RevocationKeyClass.Sensitive if sensitive else 0x00)
+
+ sig._signature.subpackets.addnew('RevocationKey',
+ hashed=True,
+ algorithm=revoker.key_algorithm,
+ fingerprint=revoker.fingerprint,
+ keyclass=keyclass)
+
+ # revocation keys should really not be revocable themselves
+ prefs['revocable'] = False
+ return self._sign(self, sig, **prefs)
+
+ @KeyAction(is_unlocked=True, is_public=False)
+ def bind(self, key, **prefs):
+ """
+ Bind a subkey to this key.
+
+ Valid optional keyword arguments are identical to those of self-signatures for :py:meth:`PGPkey.certify`
+ """
+ hash_algo = prefs.pop('hash', None)
+
+ if self.is_primary and not key.is_primary:
+ sig_type = SignatureType.Subkey_Binding
+
+ elif key.is_primary and not self.is_primary:
+ sig_type = SignatureType.PrimaryKey_Binding
+
+ else: # pragma: no cover
+ raise PGPError
+
+ sig = PGPSignature.new(sig_type, self.key_algorithm, hash_algo, self.fingerprint.keyid)
+
+ if sig_type == SignatureType.Subkey_Binding:
+ # signature options that only make sense in subkey binding signatures
+ usage = prefs.pop('usage', None)
+
+ if usage is not None:
+ sig._signature.subpackets.addnew('KeyFlags', hashed=True, flags=usage)
+
+ # if possible, have the subkey create a primary key binding signature
+ if key.key_algorithm.can_sign:
+ subkeyid = key.fingerprint.keyid
+ esig = None
+
+ if not key.is_public:
+ esig = key.bind(self)
+
+ elif subkeyid in self.subkeys: # pragma: no cover
+ esig = self.subkeys[subkeyid].bind(self)
+
+ if esig is not None:
+ sig._signature.subpackets.addnew('EmbeddedSignature', hashed=False, _sig=esig._signature)
+
+ return self._sign(key, sig, **prefs)
+
+ def verify(self, subject, signature=None):
+ """
+ Verify a subject with a signature using this key.
+
+ :param subject: The subject to verify
+ :type subject: ``str``, ``unicode``, ``None``, :py:obj:`PGPMessage`, :py:obj:`PGPKey`, :py:obj:`PGPUID`
+ :param signature: If the signature is detached, it should be specified here.
+ :type signature: :py:obj:`PGPSignature`
+ :returns: :py:obj:`~pgpy.types.SignatureVerification`
+ """
+ sspairs = []
+
+ # some type checking
+ if not isinstance(subject, (type(None), PGPMessage, PGPKey, PGPUID, PGPSignature, six.string_types, bytes, bytearray)):
+ raise TypeError("Unexpected subject value: {:s}".format(str(type(subject))))
+ if not isinstance(signature, (type(None), PGPSignature)):
+ raise TypeError("Unexpected signature value: {:s}".format(str(type(signature))))
+
+ def _filter_sigs(sigs):
+ _ids = {self.fingerprint.keyid} | set(self.subkeys)
+ return [ sig for sig in sigs if sig.signer in _ids ]
+
+ # collect signature(s)
+ if signature is None:
+ if isinstance(subject, PGPMessage):
+ sspairs += [ (sig, subject.message) for sig in _filter_sigs(subject.signatures) ]
+
+ if isinstance(subject, (PGPUID, PGPKey)):
+ sspairs += [ (sig, subject) for sig in _filter_sigs(subject.__sig__) ]
+
+ if isinstance(subject, PGPKey):
+ # user ids
+ sspairs += [ (sig, uid) for uid in subject.userids for sig in _filter_sigs(uid.__sig__) ]
+ # user attributes
+ sspairs += [ (sig, ua) for ua in subject.userattributes for sig in _filter_sigs(ua.__sig__) ]
+ # subkey binding signatures
+ sspairs += [ (sig, subkey) for subkey in subject.subkeys.values() for sig in _filter_sigs(subkey.__sig__) ]
+
+ elif signature.signer in {self.fingerprint.keyid} | set(self.subkeys):
+ sspairs += [(signature, subject)]
+
+ if len(sspairs) == 0:
+ raise PGPError("No signatures to verify")
+
+ # finally, start verifying signatures
+ sigv = SignatureVerification()
+ for sig, subj in sspairs:
+ if self.fingerprint.keyid != sig.signer and sig.signer in self.subkeys:
+ warnings.warn("Signature was signed with this key's subkey: {:s}. "
+ "Verifying with subkey...".format(sig.signer),
+ stacklevel=2)
+ sigv &= self.subkeys[sig.signer].verify(subj, sig)
+
+ else:
+ verified = self._key.verify(sig.hashdata(subj), sig.__sig__, getattr(hashes, sig.hash_algorithm.name)())
+ if verified is NotImplemented:
+ raise NotImplementedError(sig.key_algorithm)
+
+ sigv.add_sigsubj(sig, self.fingerprint.keyid, subj, verified)
+
+ return sigv
+
+ @KeyAction(KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage, is_public=True)
+ def encrypt(self, message, sessionkey=None, **prefs):
+ """
+ Encrypt a PGPMessage using this key.
+
+ :param message: The message to encrypt.
+ :type message: :py:obj:`PGPMessage`
+ :optional param sessionkey: Provide a session key to use when encrypting something. Default is ``None``.
+ If ``None``, a session key of the appropriate length will be generated randomly.
+
+ .. warning::
+
+ Care should be taken when making use of this option! Session keys *absolutely need*
+ to be unpredictable! Use the ``gen_key()`` method on the desired
+ :py:obj:`~constants.SymmetricKeyAlgorithm` to generate the session key!
+ :type sessionkey: ``bytes``, ``str``
+
+ :raises: :py:exc:`~errors.PGPEncryptionError` if encryption failed for any reason.
+ :returns: A new :py:obj:`PGPMessage` with the encrypted contents of ``message``
+
+ The following optional keyword arguments can be used with :py:meth:`PGPKey.encrypt`:
+
+ :keyword cipher: Specifies the symmetric block cipher to use when encrypting the message.
+ :type cipher: :py:obj:`~constants.SymmetricKeyAlgorithm`
+ :keyword user: Specifies the User ID to use as the recipient for this encryption operation, for the purposes of
+ preference defaults and selection validation.
+ :type user: ``str``, ``unicode``
+ """
+ user = prefs.pop('user', None)
+ uid = None
+ if user is not None:
+ uid = self.get_uid(user)
+ else:
+ uid = next(iter(self.userids), None)
+ if uid is None and self.parent is not None:
+ uid = next(iter(self.parent.userids), None)
+ cipher_algo = prefs.pop('cipher', uid.selfsig.cipherprefs[0])
+
+ if cipher_algo not in uid.selfsig.cipherprefs:
+ warnings.warn("Selected symmetric algorithm not in key preferences", stacklevel=3)
+
+ if message.is_compressed and message._compression not in uid.selfsig.compprefs:
+ warnings.warn("Selected compression algorithm not in key preferences", stacklevel=3)
+
+ if sessionkey is None:
+ sessionkey = cipher_algo.gen_key()
+
+ # set up a new PKESessionKeyV3
+ pkesk = PKESessionKeyV3()
+ pkesk.encrypter = bytearray(binascii.unhexlify(self.fingerprint.keyid.encode('latin-1')))
+ pkesk.pkalg = self.key_algorithm
+ # pkesk.encrypt_sk(self.__key__, cipher_algo, sessionkey)
+ pkesk.encrypt_sk(self._key, cipher_algo, sessionkey)
+
+ if message.is_encrypted: # pragma: no cover
+ _m = message
+
+ else:
+ _m = PGPMessage()
+ skedata = IntegrityProtectedSKEDataV1()
+ skedata.encrypt(sessionkey, cipher_algo, message.__bytes__())
+ _m |= skedata
+
+ _m |= pkesk
+
+ return _m
+
+ def decrypt(self, message):
+ """
+ Decrypt a PGPMessage using this key.
+
+ :param message: An encrypted :py:obj:`PGPMessage`
+ :returns: A new :py:obj:`PGPMessage` with the decrypted contents of ``message``
+ """
+ if not message.is_encrypted:
+ warnings.warn("This message is not encrypted", stacklevel=2)
+ return message
+
+ if self.fingerprint.keyid not in message.issuers:
+ sks = set(self.subkeys)
+ mis = set(message.issuers)
+ if sks & mis:
+ skid = list(sks & mis)[0]
+ warnings.warn("Message was encrypted with this key's subkey: {:s}. "
+ "Decrypting with that...".format(skid),
+ stacklevel=2)
+ return self.subkeys[skid].decrypt(message)
+
+ raise PGPError("Cannot decrypt the provided message with this key")
+
+ pkesk = next(pk for pk in message._sessionkeys if pk.pkalg == self.key_algorithm and pk.encrypter == self.fingerprint.keyid)
+ alg, key = pkesk.decrypt_sk(self._key)
+
+ # now that we have the symmetric cipher used and the key, we can decrypt the actual message
+ decmsg = PGPMessage()
+ decmsg.parse(message.message.decrypt(key, alg))
+
+ return decmsg
+
+ def parse(self, data):
+ unarmored = self.ascii_unarmor(data)
+ data = unarmored['body']
+
+ if unarmored['magic'] is not None and 'KEY' not in unarmored['magic']:
+ raise ValueError('Expected: KEY. Got: {}'.format(str(unarmored['magic'])))
+
+ if unarmored['headers'] is not None:
+ self.ascii_headers = unarmored['headers']
+
+ # parse packets
+ # keys will hold other keys parsed here
+ keys = collections.OrderedDict()
+ # orphaned will hold all non-opaque orphaned packets
+ orphaned = []
+ # last holds the last non-signature thing processed
+
+ ##TODO: see issue #141 and fix this better
+ getpkt = lambda d: Packet(d) if len(d) > 0 else None # flake8: noqa
+ # some packets are filtered out
+ getpkt = filter(lambda p: p.header.tag != PacketTag.Trust, iter(functools.partial(getpkt, data), None))
+
+ def pktgrouper():
+ class PktGrouper(object):
+ def __init__(self):
+ self.last = None
+
+ def __call__(self, pkt):
+ if pkt.header.tag != PacketTag.Signature:
+ self.last = '{:02X}_{:s}'.format(id(pkt), pkt.__class__.__name__)
+ return self.last
+ return PktGrouper()
+
+ while True:
+ # print(type(p) for p in getpkt)
+ for group in iter(group for _, group in itertools.groupby(getpkt, key=pktgrouper()) if not _.endswith('Opaque')):
+ pkt = next(group)
+
+ # deal with pkt first
+ if isinstance(pkt, Key):
+ pgpobj = (self if self._key is None else PGPKey()) | pkt
+
+ elif isinstance(pkt, (UserID, UserAttribute)):
+ pgpobj = PGPUID() | pkt
+
+ else: # pragma: no cover
+ break
+
+ # add signatures to whatever we got
+ [ operator.ior(pgpobj, PGPSignature() | sig) for sig in group if not isinstance(sig, Opaque) ]
+
+ # and file away pgpobj
+ if isinstance(pgpobj, PGPKey):
+ if pgpobj.is_primary:
+ keys[(pgpobj.fingerprint.keyid, pgpobj.is_public)] = pgpobj
+
+ else:
+ keys[next(reversed(keys))] |= pgpobj
+
+ elif isinstance(pgpobj, PGPUID):
+ # parent is likely the most recently parsed primary key
+ keys[next(reversed(keys))] |= pgpobj
+
+ else: # pragma: no cover
+ break
+ else:
+ # finished normally
+ break
+
+ # this will only be reached called if the inner loop hit a break
+ warnings.warn("Warning: Orphaned packet detected! {:s}".format(repr(pkt)), stacklevel=2) # pragma: no cover
+ orphaned.append(pkt) # pragma: no cover
+ for pkt in group: # pragma: no cover
+ orphaned.append(pkt)
+
+ # remove the reference to self from keys
+ [ keys.pop((getattr(self, 'fingerprint.keyid', '~'), None), t) for t in (True, False) ]
+ # return {'keys': keys, 'orphaned': orphaned}
+ return keys
+
+
+class PGPKeyring(collections.Container, collections.Iterable, collections.Sized):
+ def __init__(self, *args):
+ """
+ PGPKeyring objects represent in-memory keyrings that can contain any combination of supported private and public
+ keys. It can not currently be conveniently exported to a format that can be understood by GnuPG.
+ """
+ super(PGPKeyring, self).__init__()
+ self._keys = {}
+ self._pubkeys = collections.deque()
+ self._privkeys = collections.deque()
+ self._aliases = collections.deque([{}])
+ self.load(*args)
+
+ def __contains__(self, alias):
+ aliases = set().union(*self._aliases)
+
+ if isinstance(alias, six.string_types):
+ return alias in aliases or alias.replace(' ', '') in aliases
+
+ return alias in aliases # pragma: no cover
+
+ def __len__(self):
+ return len(self._keys)
+
+ def __iter__(self): # pragma: no cover
+ for pgpkey in itertools.chain(self._pubkeys, self._privkeys):
+ yield pgpkey
+
+ def _get_key(self, alias):
+ for m in self._aliases:
+ if alias in m:
+ return self._keys[m[alias]]
+
+ if alias.replace(' ', '') in m:
+ return self._keys[m[alias.replace(' ', '')]]
+
+ raise KeyError(alias)
+
+ def _get_keys(self, alias):
+ return [self._keys[m[alias]] for m in self._aliases if alias in m]
+
+ def _sort_alias(self, alias):
+ # remove alias from all levels of _aliases, and sort by created time and key half
+ # so the order of _aliases from left to right:
+ # - newer keys come before older ones
+ # - private keys come before public ones
+ #
+ # this list is sorted in the opposite direction from that, because they will be placed into self._aliases
+ # from right to left.
+ pkids = sorted(list(set().union(m.pop(alias) for m in self._aliases if alias in m)),
+ key=lambda pkid: (self._keys[pkid].created, self._keys[pkid].is_public))
+
+ # drop the now-sorted aliases into place
+ for depth, pkid in enumerate(pkids):
+ self._aliases[depth][alias] = pkid
+
+ # finally, remove any empty dicts left over
+ while {} in self._aliases: # pragma: no cover
+ self._aliases.remove({})
+
+ def _add_alias(self, alias, pkid):
+ # brand new alias never seen before!
+ if alias not in self:
+ self._aliases[-1][alias] = pkid
+
+ # this is a duplicate alias->key link; ignore it
+ elif alias in self and pkid in set(m[alias] for m in self._aliases if alias in m):
+ pass # pragma: no cover
+
+ # this is an alias that already exists, but points to a key that is not already referenced by it
+ else:
+ adepth = len(self._aliases) - len([None for m in self._aliases if alias in m]) - 1
+ # all alias maps have this alias, so increase total depth by 1
+ if adepth == -1:
+ self._aliases.appendleft({})
+ adepth = 0
+
+ self._aliases[adepth][alias] = pkid
+ self._sort_alias(alias)
+
+ def _add_key(self, pgpkey):
+ pkid = id(pgpkey)
+ if pkid not in self._keys:
+ self._keys[pkid] = pgpkey
+
+ # add to _{pub,priv}keys if this is either a primary key, or a subkey without one
+ if pgpkey.parent is None:
+ if pgpkey.is_public:
+ self._pubkeys.append(pkid)
+
+ else:
+ self._privkeys.append(pkid)
+
+ # aliases
+ self._add_alias(pgpkey.fingerprint, pkid)
+ self._add_alias(pgpkey.fingerprint.keyid, pkid)
+ self._add_alias(pgpkey.fingerprint.shortid, pkid)
+ for uid in pgpkey.userids:
+ self._add_alias(uid.name, pkid)
+ if uid.comment:
+ self._add_alias(uid.comment, pkid)
+
+ if uid.email:
+ self._add_alias(uid.email, pkid)
+
+ # subkeys
+ for subkey in pgpkey.subkeys.values():
+ self._add_key(subkey)
+
+ def load(self, *args):
+ """
+ Load all keys provided into this keyring object.
+
+ :param \*args: Each arg in ``args`` can be any of the formats supported by :py:meth:`PGPKey.from_path` and
+ :py:meth:`PGPKey.from_blob`, or a ``list`` or ``tuple`` of these.
+ :type \*args: ``list``, ``tuple``, ``str``, ``unicode``, ``bytes``, ``bytearray``
+ :returns: a ``set`` containing the unique fingerprints of all of the keys that were loaded during this operation.
+ """
+ def _preiter(first, iterable):
+ yield first
+ for item in iterable:
+ yield item
+
+ loaded = set()
+ for key in iter(item for ilist in iter(ilist if isinstance(ilist, (tuple, list)) else [ilist] for ilist in args)
+ for item in ilist):
+ if os.path.isfile(key):
+ _key, keys = PGPKey.from_file(key)
+
+ else:
+ _key, keys = PGPKey.from_blob(key)
+
+ for ik in _preiter(_key, keys.values()):
+ self._add_key(ik)
+ loaded |= {ik.fingerprint} | {isk.fingerprint for isk in ik.subkeys.values()}
+
+ return list(loaded)
+
+ @contextlib.contextmanager
+ def key(self, identifier):
+ """
+ A context-manager method. Yields the first :py:obj:`PGPKey` object that matches the provided identifier.
+
+ :param identifier: The identifier to use to select a loaded key.
+ :type identifier: :py:exc:`PGPMessage`, :py:exc:`PGPSignature`, ``str``
+ :raises: :py:exc:`KeyError` if there is no loaded key that satisfies the identifier.
+ """
+ if isinstance(identifier, PGPMessage):
+ for issuer in identifier.issuers:
+ if issuer in self:
+ identifier = issuer
+ break
+
+ if isinstance(identifier, PGPSignature):
+ identifier = identifier.signer
+
+ yield self._get_key(identifier)
+
+ def fingerprints(self, keyhalf='any', keytype='any'):
+ """
+ List loaded fingerprints with some optional filtering.
+
+ :param str keyhalf: Can be 'any', 'public', or 'private'. If 'public', or 'private', the fingerprints of keys of the
+ the other type will not be included in the results.
+ :param str keytype: Can be 'any', 'primary', or 'sub'. If 'primary' or 'sub', the fingerprints of keys of the
+ the other type will not be included in the results.
+ :returns: a ``set`` of fingerprints of keys matching the filters specified.
+ """
+ return {pk.fingerprint for pk in self._keys.values()
+ if pk.is_primary in [True if keytype in ['primary', 'any'] else None,
+ False if keytype in ['sub', 'any'] else None]
+ if pk.is_public in [True if keyhalf in ['public', 'any'] else None,
+ False if keyhalf in ['private', 'any'] else None]}
+
+ def unload(self, key):
+ """
+ Unload a loaded key and its subkeys.
+
+ The easiest way to do this is to select a key using :py:meth:`PGPKeyring.key` first::
+
+ with keyring.key("DSA von TestKey") as key:
+ keyring.unload(key)
+
+ :param key: The key to unload.
+ :type key: :py:obj:`PGPKey`
+ """
+ assert isinstance(key, PGPKey)
+ pkid = id(key)
+ if pkid in self._keys:
+ # remove references
+ [ kd.remove(pkid) for kd in [self._pubkeys, self._privkeys] if pkid in kd ]
+ # remove the key
+ self._keys.pop(pkid)
+
+ # remove aliases
+ for m, a in [ (m, a) for m in self._aliases for a, p in m.items() if p == pkid ]:
+ m.pop(a)
+ # do a re-sort of this alias if it was not unique
+ if a in self:
+ self._sort_alias(a)
+
+ # if key is a primary key, unload its subkeys as well
+ if key.is_primary:
+ [ self.unload(sk) for sk in key.subkeys.values() ]
diff --git a/src/leap/mx/vendor/pgpy/symenc.py b/src/leap/mx/vendor/pgpy/symenc.py
new file mode 100644
index 0000000..3d81a7c
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/symenc.py
@@ -0,0 +1,58 @@
+""" symenc.py
+"""
+import six
+
+from cryptography.exceptions import UnsupportedAlgorithm
+
+from cryptography.hazmat.backends import default_backend
+
+from cryptography.hazmat.primitives.ciphers import Cipher
+from cryptography.hazmat.primitives.ciphers import modes
+
+from .errors import PGPDecryptionError
+from .errors import PGPEncryptionError
+from .errors import PGPInsecureCipher
+
+__all__ = ['_encrypt',
+ '_decrypt']
+
+
+def _encrypt(pt, key, alg, iv=None):
+ if iv is None:
+ iv = b'\x00' * (alg.block_size // 8)
+
+ if alg.is_insecure:
+ raise PGPInsecureCipher("{:s} is not secure. Do not use it for encryption!".format(alg.name))
+
+ if not callable(alg.cipher):
+ raise PGPEncryptionError("Cipher {:s} not supported".format(alg.name))
+
+ try:
+ encryptor = Cipher(alg.cipher(key), modes.CFB(iv), default_backend()).encryptor()
+
+ except UnsupportedAlgorithm as ex: # pragma: no cover
+ six.raise_from(PGPEncryptionError, ex)
+
+ else:
+ return bytearray(encryptor.update(pt) + encryptor.finalize())
+
+
+def _decrypt(ct, key, alg, iv=None):
+ if iv is None:
+ """
+ Instead of using an IV, OpenPGP prefixes a string of length
+ equal to the block size of the cipher plus two to the data before it
+ is encrypted. The first block-size octets (for example, 8 octets for
+ a 64-bit block length) are random, and the following two octets are
+ copies of the last two octets of the IV.
+ """
+ iv = b'\x00' * (alg.block_size // 8)
+
+ try:
+ decryptor = Cipher(alg.cipher(key), modes.CFB(iv), default_backend()).decryptor()
+
+ except UnsupportedAlgorithm as ex: # pragma: no cover
+ six.raise_from(PGPDecryptionError, ex)
+
+ else:
+ return bytearray(decryptor.update(ct) + decryptor.finalize())
diff --git a/src/leap/mx/vendor/pgpy/types.py b/src/leap/mx/vendor/pgpy/types.py
new file mode 100644
index 0000000..b549434
--- /dev/null
+++ b/src/leap/mx/vendor/pgpy/types.py
@@ -0,0 +1,729 @@
+""" types.py
+"""
+from __future__ import division
+
+import abc
+import base64
+import binascii
+import bisect
+import codecs
+import collections
+import operator
+import os
+import re
+import warnings
+import weakref
+
+from enum import EnumMeta
+from enum import IntEnum
+
+import six
+
+from ._author import __version__
+
+from .decorators import sdproperty
+
+from .errors import PGPError
+
+__all__ = ['Armorable',
+ 'ParentRef',
+ 'PGPObject',
+ 'Field',
+ 'Header',
+ 'MetaDispatchable',
+ 'Dispatchable',
+ 'SignatureVerification',
+ 'FlagEnumMeta',
+ 'FlagEnum',
+ 'Fingerprint',
+ 'SorteDeque']
+
+if six.PY2:
+ FileNotFoundError = IOError
+ re.ASCII = 0
+
+
+class Armorable(six.with_metaclass(abc.ABCMeta)):
+ __crc24_init__ = 0x0B704CE
+ __crc24_poly__ = 0x1864CFB
+
+ __armor_fmt__ = '-----BEGIN PGP {block_type}-----\n' \
+ '{headers}\n' \
+ '{packet}\n' \
+ '={crc}\n' \
+ '-----END PGP {block_type}-----\n'
+
+ @property
+ def charset(self):
+ return self.ascii_headers.get('Charset', 'utf-8')
+
+ @charset.setter
+ def charset(self, encoding):
+ self.ascii_headers['Charset'] = codecs.lookup(encoding).name
+
+ @staticmethod
+ def is_ascii(text):
+ if isinstance(text, six.string_types):
+ return bool(re.match(r'^[ -~\r\n]+$', text, flags=re.ASCII))
+
+ if isinstance(text, (bytes, bytearray)):
+ return bool(re.match(br'^[ -~\r\n]+$', text, flags=re.ASCII))
+
+ raise TypeError("Expected: ASCII input of type str, bytes, or bytearray") # pragma: no cover
+
+ @staticmethod
+ def ascii_unarmor(text):
+ """
+ Takes an ASCII-armored PGP block and returns the decoded byte value.
+
+ :param text: An ASCII-armored PGP block, to un-armor.
+ :raises: :py:exc:`ValueError` if ``text`` did not contain an ASCII-armored PGP block.
+ :raises: :py:exc:`TypeError` if ``text`` is not a ``str``, ``bytes``, or ``bytearray``
+ :returns: A ``dict`` containing information from ``text``, including the de-armored data.
+ """
+ m = {'magic': None, 'headers': None, 'body': bytearray(), 'crc': None}
+ if not Armorable.is_ascii(text):
+ m['body'] = bytearray(text)
+ return m
+
+ if isinstance(text, (bytes, bytearray)): # pragma: no cover
+ text = text.decode('latin-1')
+
+ # the re.VERBOSE flag allows for:
+ # - whitespace is ignored except when in a character class or escaped
+ # - anything after a '#' that is not escaped or in a character class is ignored, allowing for comments
+ m = re.search(r"""# This capture group is optional because it will only be present in signed cleartext messages
+ (^-{5}BEGIN\ PGP\ SIGNED\ MESSAGE-{5}(?:\r?\n)
+ (Hash:\ (?P<hashes>[A-Za-z0-9\-,]+)(?:\r?\n){2})?
+ (?P<cleartext>(.*\n)+)(?:\r?\n)
+ )?
+ # armor header line; capture the variable part of the magic text
+ ^-{5}BEGIN\ PGP\ (?P<magic>[A-Z0-9 ,]+)-{5}(?:\r?\n)
+ # try to capture all the headers into one capture group
+ # if this doesn't match, m['headers'] will be None
+ (?P<headers>(^.+:\ .+(?:\r?\n))+)?(?:\r?\n)?
+ # capture all lines of the body, up to 76 characters long,
+ # including the newline, and the pad character(s)
+ (?P<body>([A-Za-z0-9+/]{1,75}={,2}(?:\r?\n))+)
+ # capture the armored CRC24 value
+ ^=(?P<crc>[A-Za-z0-9+/]{4})(?:\r?\n)
+ # finally, capture the armor tail line, which must match the armor header line
+ ^-{5}END\ PGP\ (?P=magic)-{5}(?:\r?\n)?
+ """, text, flags=re.MULTILINE | re.VERBOSE)
+
+ if m is None: # pragma: no cover
+ raise ValueError("Expected: ASCII-armored PGP data")
+
+ m = m.groupdict()
+
+ if m['hashes'] is not None:
+ m['hashes'] = m['hashes'].split(',')
+
+ if m['headers'] is not None:
+ m['headers'] = collections.OrderedDict(re.findall('^(?P<key>.+): (?P<value>.+)$\n?', m['headers'], flags=re.MULTILINE))
+
+ if m['body'] is not None:
+ try:
+ m['body'] = bytearray(base64.b64decode(m['body'].encode()))
+
+ except (binascii.Error, TypeError) as ex:
+ six.raise_from(PGPError, ex)
+
+ if m['crc'] is not None:
+ m['crc'] = Header.bytes_to_int(base64.b64decode(m['crc'].encode()))
+ if Armorable.crc24(m['body']) != m['crc']:
+ warnings.warn('Incorrect crc24', stacklevel=3)
+
+ return m
+
+ @staticmethod
+ def crc24(data):
+ # CRC24 computation, as described in the RFC 4880 section on Radix-64 Conversions
+ #
+ # The checksum is a 24-bit Cyclic Redundancy Check (CRC) converted to
+ # four characters of radix-64 encoding by the same MIME base64
+ # transformation, preceded by an equal sign (=). The CRC is computed
+ # by using the generator 0x864CFB and an initialization of 0xB704CE.
+ # The accumulation is done on the data before it is converted to
+ # radix-64, rather than on the converted data.
+ crc = Armorable.__crc24_init__
+
+ if not isinstance(data, bytearray):
+ data = six.iterbytes(data)
+
+ for b in data:
+ crc ^= b << 16
+
+ for i in range(8):
+ crc <<= 1
+ if crc & 0x1000000:
+ crc ^= Armorable.__crc24_poly__
+
+ return crc & 0xFFFFFF
+
+ @abc.abstractproperty
+ def magic(self):
+ """The magic string identifier for the current PGP type"""
+
+ @classmethod
+ def from_file(cls, filename):
+ with open(filename, 'rb') as file:
+ obj = cls()
+ data = bytearray(os.path.getsize(filename))
+ file.readinto(data)
+
+ po = obj.parse(data)
+
+ if po is not None:
+ return (obj, po)
+
+ return obj # pragma: no cover
+
+ @classmethod
+ def from_blob(cls, blob):
+ obj = cls()
+ if (not isinstance(blob, six.binary_type)) and (not isinstance(blob, bytearray)):
+ po = obj.parse(bytearray(blob, 'latin-1'))
+
+ else:
+ po = obj.parse(bytearray(blob))
+
+ if po is not None:
+ return (obj, po)
+
+ return obj # pragma: no cover
+
+ def __init__(self):
+ super(Armorable, self).__init__()
+ self.ascii_headers = collections.OrderedDict()
+ self.ascii_headers['Version'] = 'PGPy v' + __version__ # Default value
+
+ def __str__(self):
+ payload = base64.b64encode(self.__bytes__()).decode('latin-1')
+ payload = '\n'.join(payload[i:(i + 64)] for i in range(0, len(payload), 64))
+
+ return self.__armor_fmt__.format(
+ block_type=self.magic,
+ headers=''.join('{key}: {val}\n'.format(key=key, val=val) for key, val in self.ascii_headers.items()),
+ packet=payload,
+ crc=base64.b64encode(PGPObject.int_to_bytes(self.crc24(self.__bytes__()), 3)).decode('latin-1')
+ )
+
+ def __copy__(self):
+ obj = self.__class__()
+ obj.ascii_headers = self.ascii_headers.copy()
+
+ return obj
+
+
+class ParentRef(object):
+ # mixin class to handle weak-referencing a parent object
+ @property
+ def _parent(self):
+ if isinstance(self.__parent, weakref.ref):
+ return self.__parent()
+ return self.__parent
+
+ @_parent.setter
+ def _parent(self, parent):
+ try:
+ self.__parent = weakref.ref(parent)
+
+ except TypeError:
+ self.__parent = parent
+
+ @property
+ def parent(self):
+ return self._parent
+
+ def __init__(self):
+ super(ParentRef, self).__init__()
+ self._parent = None
+
+
+class PGPObject(six.with_metaclass(abc.ABCMeta, object)):
+ __metaclass__ = abc.ABCMeta
+
+ @staticmethod
+ def int_byte_len(i):
+ return (i.bit_length() + 7) // 8
+
+ @staticmethod
+ def bytes_to_int(b, order='big'): # pragma: no cover
+ """convert bytes to integer"""
+ if six.PY2:
+ # save the original type of b without having to copy any data
+ _b = b.__class__()
+ if order != 'little':
+ b = reversed(b)
+
+ if not isinstance(_b, bytearray):
+ b = six.iterbytes(b)
+
+ return sum(c << (i * 8) for i, c in enumerate(b))
+
+ return int.from_bytes(b, order)
+
+ @staticmethod
+ def int_to_bytes(i, minlen=1, order='big'): # pragma: no cover
+ """convert integer to bytes"""
+ blen = max(minlen, PGPObject.int_byte_len(i), 1)
+
+ if six.PY2:
+ r = iter(_ * 8 for _ in (range(blen) if order == 'little' else range(blen - 1, -1, -1)))
+ return bytes(bytearray((i >> c) & 0xff for c in r))
+
+ return i.to_bytes(blen, order)
+
+ @staticmethod
+ def text_to_bytes(text):
+ if text is None:
+ return text
+
+ # if we got bytes, just return it
+ if isinstance(text, (bytearray, six.binary_type)):
+ return text
+
+ # if we were given a unicode string, or if we translated the string into utf-8,
+ # we know that Python already has it in utf-8 encoding, so we can now just encode it to bytes
+ return text.encode('utf-8')
+
+ @staticmethod
+ def bytes_to_text(text):
+ if text is None or isinstance(text, six.text_type):
+ return text
+
+ return text.decode('utf-8')
+
+ @abc.abstractmethod
+ def parse(self, packet):
+ """this method is too abstract to understand"""
+
+ @abc.abstractmethod
+ def __bytearray__(self):
+ """
+ Returns the contents of concrete subclasses in a binary format that can be understood by other OpenPGP
+ implementations
+ """
+
+ def __bytes__(self):
+ """
+ Return the contents of concrete subclasses in a binary format that can be understood by other OpenPGP
+ implementations
+ """
+ # this is what all subclasses will do anyway, so doing this here we can reduce code duplication significantly
+ return bytes(self.__bytearray__())
+
+
+class Field(PGPObject):
+ @abc.abstractmethod
+ def __len__(self):
+ """Return the length of the output of __bytes__"""
+
+
+class Header(Field):
+ @staticmethod
+ def encode_length(l, nhf=True, llen=1):
+ def _new_length(l):
+ if 192 > l:
+ return Header.int_to_bytes(l)
+
+ elif 8384 > l:
+ elen = ((l & 0xFF00) + (192 << 8)) + ((l & 0xFF) - 192)
+ return Header.int_to_bytes(elen, 2)
+
+ return b'\xFF' + Header.int_to_bytes(l, 4)
+
+ def _old_length(l, llen):
+ return Header.int_to_bytes(l, llen) if llen > 0 else b''
+
+ return _new_length(l) if nhf else _old_length(l, llen)
+
+ @sdproperty
+ def length(self):
+ return self._len
+
+ @length.register(int)
+ def length_int(self, val):
+ self._len = val
+
+ @length.register(six.binary_type)
+ @length.register(bytearray)
+ def length_bin(self, val):
+ def _new_len(b):
+ fo = b[0]
+
+ if 192 > fo:
+ self._len = self.bytes_to_int(b[:1])
+ del b[:1]
+
+ elif 224 > fo: # >= 192 is implied
+ dlen = self.bytes_to_int(b[:2])
+ self._len = ((dlen - (192 << 8)) & 0xFF00) + ((dlen & 0xFF) + 192)
+ del b[:2]
+
+ elif 255 > fo: # pragma: no cover
+ # not testable until partial body lengths actually work
+ # >= 224 is implied
+ # this is a partial-length header
+ self._partial = True
+ self._len = 1 << (fo & 0x1f)
+
+ elif 255 == fo:
+ self._len = self.bytes_to_int(b[1:5])
+ del b[:5]
+
+ else: # pragma: no cover
+ raise ValueError("Malformed length: 0x{:02x}".format(fo))
+
+ def _old_len(b):
+ if self.llen > 0:
+ self._len = self.bytes_to_int(b[:self.llen])
+ del b[:self.llen]
+
+ else: # pragma: no cover
+ self._len = 0
+
+ _new_len(val) if self._lenfmt == 1 else _old_len(val)
+
+ @sdproperty
+ def llen(self):
+ l = self.length
+ lf = self._lenfmt
+
+ if lf == 1:
+ # new-format length
+ if 192 > l:
+ return 1
+
+ elif 8384 > self.length: # >= 192 is implied
+ return 2
+
+ else:
+ return 5
+
+ else:
+ # old-format length
+ ##TODO: what if _llen needs to be (re)computed?
+ return self._llen
+
+ @llen.register(int)
+ def llen_int(self, val):
+ if self._lenfmt == 0:
+ self._llen = {0: 1, 1: 2, 2: 4, 3: 0}[val]
+
+ def __init__(self):
+ super(Header, self).__init__()
+ self._len = 1
+ self._llen = 1
+ self._lenfmt = 1
+ self._partial = False
+
+
+class MetaDispatchable(abc.ABCMeta):
+ """
+ MetaDispatchable is a metaclass for objects that subclass Dispatchable
+ """
+
+ _roots = set()
+ """
+ _roots is a set of all currently registered RootClass class objects
+
+ A RootClass is successfully registered if the following things are true:
+ - it inherits (directly or indirectly) from Dispatchable
+ - __typeid__ == -1
+ """
+ _registry = {}
+ """
+ _registry is the Dispatchable class registry. It uses the following format:
+
+ { (RootClass, None): OpaqueClass }:
+ denotes the default ("opaque") for a given RootClass.
+
+ An OpaqueClass is successfully registered as such provided the following conditions are met:
+ - it inherits directly from a RootClass
+ - __typeid__ is None
+
+ { (RootClass, TypeID): SubClass }:
+ denotes the class that handles the type given in TypeID
+
+ a SubClass is successfully registered as such provided the following conditions are met:
+ - it inherits (directly or indirectly) from a RootClass
+ - __typeid__ is a positive int
+ - the given typeid is not already registered
+
+ { (RootClass, TypeID): VerSubClass }:
+ denotes that a given TypeID has multiple versions, and that this is class' subclasses handle those.
+ A VerSubClass is registered identically to a normal SubClass.
+
+ { (RootClass, TypeID, Ver): VerSubClass }:
+ denotes the class that handles the type given in TypeID and the version of that type given in Ver
+
+ a Versioned SubClass is successfully registered as such provided the following conditions are met:
+ - it inherits from a VerSubClass
+ - __ver__ > 0
+ - the given typeid/ver combination is not already registered
+ """
+
+ def __new__(mcs, name, bases, attrs): # NOQA
+ ncls = super(MetaDispatchable, mcs).__new__(mcs, name, bases, attrs)
+
+ if not hasattr(ncls.__typeid__, '__isabstractmethod__'):
+ if ncls.__typeid__ == -1 and not issubclass(ncls, tuple(MetaDispatchable._roots)):
+ # this is a root class
+ MetaDispatchable._roots.add(ncls)
+
+ elif issubclass(ncls, tuple(MetaDispatchable._roots)) and ncls.__typeid__ != -1:
+ for rcls in [ root for root in MetaDispatchable._roots if issubclass(ncls, root) ]:
+ if (rcls, ncls.__typeid__) not in MetaDispatchable._registry:
+ MetaDispatchable._registry[(rcls, ncls.__typeid__)] = ncls
+
+ if (ncls.__ver__ is not None and ncls.__ver__ > 0 and
+ (rcls, ncls.__typeid__, ncls.__ver__) not in MetaDispatchable._registry):
+ MetaDispatchable._registry[(rcls, ncls.__typeid__, ncls.__ver__)] = ncls
+
+ # finally, return the new class object
+ return ncls
+
+ def __call__(cls, packet=None): # NOQA
+ def _makeobj(cls):
+ obj = object.__new__(cls)
+ obj.__init__()
+ return obj
+
+ if packet is not None:
+ if cls in MetaDispatchable._roots:
+ rcls = cls
+
+ elif issubclass(cls, tuple(MetaDispatchable._roots)): # pragma: no cover
+ rcls = next(root for root in MetaDispatchable._roots if issubclass(cls, root))
+
+ ##TODO: else raise an exception of some kind, but this should never happen
+
+ header = rcls.__headercls__()
+ header.parse(packet)
+
+ ncls = None
+ if (rcls, header.typeid) in MetaDispatchable._registry:
+ ncls = MetaDispatchable._registry[(rcls, header.typeid)]
+
+ if ncls.__ver__ == 0:
+ if header.__class__ != ncls.__headercls__:
+ nh = ncls.__headercls__()
+ nh.__dict__.update(header.__dict__)
+ try:
+ nh.parse(packet)
+
+ except Exception as ex:
+ six.raise_from(PGPError, ex)
+
+ header = nh
+
+ if (rcls, header.typeid, header.version) in MetaDispatchable._registry:
+ ncls = MetaDispatchable._registry[(rcls, header.typeid, header.version)]
+
+ else: # pragma: no cover
+ ncls = None
+
+ if ncls is None:
+ ncls = MetaDispatchable._registry[(rcls, None)]
+
+ obj = _makeobj(ncls)
+ obj.header = header
+
+ try:
+ obj.parse(packet)
+
+ except Exception as ex:
+ six.raise_from(PGPError, ex)
+
+ else:
+ obj = _makeobj(cls)
+
+ return obj
+
+
+class Dispatchable(six.with_metaclass(MetaDispatchable, PGPObject)):
+ __metaclass__ = MetaDispatchable
+
+ @abc.abstractproperty
+ def __headercls__(self): # pragma: no cover
+ return False
+
+ @abc.abstractproperty
+ def __typeid__(self): # pragma: no cover
+ return False
+
+ __ver__ = None
+
+
+class SignatureVerification(object):
+ _sigsubj = collections.namedtuple('sigsubj', ['verified', 'by', 'signature', 'subject'])
+
+ @property
+ def good_signatures(self):
+ """
+ A generator yielding namedtuples of all signatures that were successfully verified
+ in the operation that returned this instance. The namedtuple has the following attributes:
+
+ ``sigsubj.verified`` - ``bool`` of whether the signature verified successfully or not.
+
+ ``sigsubj.by`` - the :py:obj:`~pgpy.PGPKey` that was used in this verify operation.
+
+ ``sigsubj.signature`` - the :py:obj:`~pgpy.PGPSignature` that was verified.
+
+ ``sigsubj.subject`` - the subject that was verified using the signature.
+ """
+ for s in [ i for i in self._subjects if i.verified ]:
+ yield s
+
+ @property
+ def bad_signatures(self): # pragma: no cover
+ """
+ A generator yielding namedtuples of all signatures that were not verified
+ in the operation that returned this instance. The namedtuple has the following attributes:
+
+ ``sigsubj.verified`` - ``bool`` of whether the signature verified successfully or not.
+
+ ``sigsubj.by`` - the :py:obj:`~pgpy.PGPKey` that was used in this verify operation.
+
+ ``sigsubj.signature`` - the :py:obj:`~pgpy.PGPSignature` that was verified.
+
+ ``sigsubj.subject`` - the subject that was verified using the signature.
+ """
+ for s in [ i for i in self._subjects if not i.verified ]:
+ yield s
+
+ def __init__(self):
+ """
+ Returned by :py:meth:`PGPKey.verify`
+
+ Can be compared directly as a boolean to determine whether or not the specified signature verified.
+ """
+ super(SignatureVerification, self).__init__()
+ self._subjects = []
+
+ def __contains__(self, item):
+ return item in {ii for i in self._subjects for ii in [i.signature, i.subject]}
+
+ def __len__(self):
+ return len(self._subjects)
+
+ def __bool__(self):
+ return all(s.verified for s in self._subjects)
+
+ def __nonzero__(self):
+ return self.__bool__()
+
+ def __and__(self, other):
+ if not isinstance(other, SignatureVerification):
+ raise TypeError(type(other))
+
+ self._subjects += other._subjects
+ return self
+
+ def __repr__(self):
+ return "<SignatureVerification({verified})>".format(verified=str(bool(self)))
+
+ def add_sigsubj(self, signature, by, subject=None, verified=False):
+ self._subjects.append(self._sigsubj(verified, by, signature, subject))
+
+
+class FlagEnumMeta(EnumMeta):
+ def __and__(self, other):
+ return { f for f in iter(self) if f.value & other }
+
+ def __rand__(self, other): # pragma: no cover
+ return self & other
+
+
+if six.PY2:
+ class FlagEnum(IntEnum):
+ __metaclass__ = FlagEnumMeta
+
+else:
+ namespace = FlagEnumMeta.__prepare__('FlagEnum', (IntEnum,))
+ FlagEnum = FlagEnumMeta('FlagEnum', (IntEnum,), namespace)
+
+
+class Fingerprint(str):
+ """
+ A subclass of ``str``. Can be compared using == and != to ``str``, ``unicode``, and other :py:obj:`Fingerprint` instances.
+
+ Primarily used as a key for internal dictionaries, so it ignores spaces when comparing and hashing
+ """
+ @property
+ def keyid(self):
+ return str(self).replace(' ', '')[-16:]
+
+ @property
+ def shortid(self):
+ return str(self).replace(' ', '')[-8:]
+
+ def __new__(cls, content):
+ if isinstance(content, Fingerprint):
+ return content
+
+ # validate input before continuing: this should be a string of 40 hex digits
+ content = content.upper().replace(' ', '')
+ if not bool(re.match(r'^[A-F0-9]{40}$', content)):
+ raise ValueError("Expected: String of 40 hex digits")
+
+ # store in the format: "AAAA BBBB CCCC DDDD EEEE FFFF 0000 1111 2222 3333"
+ # ^^ note 2 spaces here
+ spaces = [ ' ' if i != 4 else ' ' for i in range(10) ]
+ chunks = [ ''.join(g) for g in six.moves.zip_longest(*[iter(content)] * 4) ]
+ content = ''.join(j for i in six.moves.zip_longest(chunks, spaces, fillvalue='') for j in i).strip()
+
+ return str.__new__(cls, content)
+
+ def __eq__(self, other):
+ if isinstance(other, Fingerprint):
+ return str(self) == str(other)
+
+ if isinstance(other, (six.text_type, bytes, bytearray)):
+ if isinstance(other, (bytes, bytearray)): # pragma: no cover
+ other = other.decode('latin-1')
+
+ other = str(other).replace(' ', '')
+ return any([self.replace(' ', '') == other,
+ self.keyid == other,
+ self.shortid == other])
+
+ return False # pragma: no cover
+
+ def __ne__(self, other):
+ return not (self == other)
+
+ def __hash__(self):
+ return hash(str(self.replace(' ', '')))
+
+ def __bytes__(self):
+ return binascii.unhexlify(six.b(self.replace(' ', '')))
+
+
+class SorteDeque(collections.deque):
+ """A deque subclass that tries to maintain sorted ordering using bisect"""
+ def insort(self, item):
+ i = bisect.bisect_left(self, item)
+ self.rotate(- i)
+ self.appendleft(item)
+ self.rotate(i)
+
+ def resort(self, item): # pragma: no cover
+ if item in self:
+ # if item is already in self, see if it is still in sorted order.
+ # if not, re-sort it by removing it and then inserting it into its sorted order
+ i = bisect.bisect_left(self, item)
+ if i == len(self) or self[i] is not item:
+ self.remove(item)
+ self.insort(item)
+
+ else:
+ # if item is not in self, just insert it in sorted order
+ self.insort(item)
+
+ def check(self): # pragma: no cover
+ """re-sort any items in self that are not sorted"""
+ for unsorted in iter(self[i] for i in range(len(self) - 2) if not operator.le(self[i], self[i + 1])):
+ self.resort(unsorted)