DEVNULL = subprocess.DEVNULL
PIPE = subprocess.PIPE
def _default_timeout():
def run(cmd, check=True, timeout=_default_timeout(), **kwargs):
def run_duplicate_streams(cmd, timeout=_default_timeout()):
      def _duplicate_streams(): def create_signature(content, keyid=None, homedir=None):
  """
  <Purpose>
    Calls the gpg command line utility to sign the passed content with the key
    identified by the passed keyid from the gpg keyring at the passed homedir.

    The executed base command is defined in
    securesystemslib.gpgp.constants.GPG_SIGN_COMMAND.

    NOTE: On not fully supported versions of GPG, i.e. versions below
    securesystemslib.gpg.constants.FULLY_SUPPORTED_MIN_VERSION the returned
    signature does not contain the full keyid. As a work around, we export the
    public key bundle identified by the short keyid to compute the full keyid
    and add it to the returned signature.

  <Arguments>
    content:
            The content to be signed. (bytes)

    keyid: (optional)
            The keyid of the gpg signing keyid. If not passed the default
            key in the keyring is used.

    homedir: (optional)
            Path to the gpg keyring. If not passed the default keyring is used.

  <Exceptions>
    securesystemslib.exceptions.FormatError:
            If the keyid was passed and does not match
            securesystemslib.formats.KEYID_SCHEMA

    ValueError:
            If the gpg command failed to create a valid signature.

    OSError:
            If the gpg command is not present or non-executable.

    securesystemslib.exceptions.UnsupportedLibraryError:
            If the gpg command is not available, or
            the cryptography library is not installed.

    securesystemslib.gpg.exceptions.CommandError:
            If the gpg command returned a non-zero exit code

    securesystemslib.gpg.exceptions.KeyNotFoundError:
            If the used gpg version is not fully supported
            and no public key can be found for short keyid.

  <Side Effects>
    None.

  <Returns>
    The created signature in the format:
    securesystemslib.formats.GPG_SIGNATURE_SCHEMA.

  """
  if not HAVE_GPG: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_GPG_MSG)

  if not CRYPTO: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_CRYPTO_MSG)

  keyarg = ""
  if keyid:
    formats.KEYID_SCHEMA.check_match(keyid)
    keyarg = "--local-user {}".format(keyid)

  homearg = ""
  if homedir:
    homearg = "--homedir {}".format(homedir).replace("\\", "/")

  command = GPG_SIGN_COMMAND.format(keyarg=keyarg, homearg=homearg)

  gpg_process = process.run(command, input=content, check=False,
      stdout=process.PIPE, stderr=process.PIPE)

  # TODO: It's suggested to take a look at `--status-fd` for proper error
  # reporting, as there is no clear distinction between the return codes
  # https://lists.gnupg.org/pipermail/gnupg-devel/2005-December/022559.html
  if gpg_process.returncode != 0:
    raise CommandError("Command '{}' returned "
        "non-zero exit status '{}', stderr was:\n{}.".format(gpg_process.args,
        gpg_process.returncode, gpg_process.stderr.decode()))

  signature_data = gpg_process.stdout
  signature = parse_signature_packet(signature_data)

  # On GPG < 2.1 we cannot derive the full keyid from the signature data.
  # Instead we try to compute the keyid from the public part of the signing
  # key or its subkeys, identified by the short keyid.
  # parse_signature_packet is guaranteed to return at least one of keyid or
  # short_keyid.
  # Exclude the following code from coverage for consistent coverage across
  # test environments.
  if not signature["keyid"]: # pragma: no cover
    log.warning("The created signature does not include the hashed subpacket"
        " '33' (full keyid). You probably have a gpg version <{}."
        " We will export the public keys associated with the short keyid to"
        " compute the full keyid.".format(FULLY_SUPPORTED_MIN_VERSION))

    short_keyid = signature["short_keyid"]

    # Export public key bundle (master key including with optional subkeys)
    public_key_bundle = export_pubkey(short_keyid, homedir)

    # Test if the short keyid matches the master key ...
    master_key_full_keyid = public_key_bundle["keyid"]
    if master_key_full_keyid.endswith(short_keyid.lower()):
      signature["keyid"] = master_key_full_keyid

    # ... or one of the subkeys, and add the full keyid to the signature dict.
    else:
      for sub_key_full_keyid in list(
          public_key_bundle.get("subkeys", {}).keys()):

        if sub_key_full_keyid.endswith(short_keyid.lower()):
          signature["keyid"] = sub_key_full_keyid
          break

  # If there is still no full keyid something went wrong
  if not signature["keyid"]: # pragma: no cover
    raise ValueError("Full keyid could not be determined for signature '{}'".
        format(signature))

  # It is okay now to remove the optional short keyid to save space
  signature.pop("short_keyid", None)

  return signature def export_pubkey(keyid, homedir=None):
  """Exports a public key from a GnuPG keyring.

  Arguments:
    keyid: An OpenPGP keyid in KEYID_SCHEMA format.
    homedir (optional): A path to the GnuPG home directory. If not set the
        default GnuPG home directory is used.

  Raises:
    ValueError: Keyid is not a string.
    UnsupportedLibraryError: The gpg command or pyca/cryptography are not
        available.
    KeyNotFoundError: No key or subkey was found for that keyid.

  Side Effects:
    Calls system gpg command in a subprocess.

  Returns:
    An OpenPGP public key object in GPG_PUBKEY_SCHEMA format.

  """
  if not HAVE_GPG: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_GPG_MSG)

  if not CRYPTO: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_CRYPTO_MSG)

  if not formats.KEYID_SCHEMA.matches(keyid):
    # FIXME: probably needs smarter parsing of what a valid keyid is so as to
    # not export more than one pubkey packet.
    raise ValueError("we need to export an individual key. Please provide a "
        " valid keyid! Keyid was '{}'.".format(keyid))

  homearg = ""
  if homedir:
    homearg = "--homedir {}".format(homedir).replace("\\", "/")

  # TODO: Consider adopting command error handling from `create_signature`
  # above, e.g. in a common 'run gpg command' utility function
  command = GPG_EXPORT_PUBKEY_COMMAND.format(keyid=keyid, homearg=homearg)
  gpg_process = process.run(command, stdout=process.PIPE, stderr=process.PIPE)

  key_packet = gpg_process.stdout
  key_bundle = get_pubkey_bundle(key_packet, keyid)

  return key_bundle def verify_signature(signature_object, pubkey_info, content):
  """
  <Purpose>
    Verifies the passed signature against the passed content using the
    passed public key, or one of its subkeys, associated by the signature's
    keyid.

    The function selects the appropriate verification algorithm (rsa or dsa)
    based on the "type" field in the passed public key object.

  <Arguments>
    signature_object:
            A signature object in the format:
            securesystemslib.formats.GPG_SIGNATURE_SCHEMA

    pubkey_info:
            A public key object in the format:
            securesystemslib.formats.GPG_PUBKEY_SCHEMA

    content:
            The content to be verified. (bytes)

  <Exceptions>
    securesystemslib.gpg.exceptions.KeyExpirationError:
            if the passed public key has expired

    securesystemslib.exceptions.UnsupportedLibraryError:
            if the cryptography module is unavailable

  <Side Effects>
    None.

  <Returns>
    True if signature verification passes, False otherwise.

  """
  if not CRYPTO: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_CRYPTO_MSG)

  formats.GPG_PUBKEY_SCHEMA.check_match(pubkey_info)
  formats.GPG_SIGNATURE_SCHEMA.check_match(signature_object)

  handler = SIGNATURE_HANDLERS[pubkey_info['type']]
  sig_keyid = signature_object["keyid"]

  verification_key = pubkey_info

  # If the keyid on the signature matches a subkey of the passed key,
  # we use that subkey for verification instead of the master key.
  if sig_keyid in list(pubkey_info.get("subkeys", {}).keys()):
    verification_key = pubkey_info["subkeys"][sig_keyid]


  creation_time = verification_key.get("creation_time")
  validity_period = verification_key.get("validity_period")

  if creation_time and validity_period and \
      creation_time + validity_period < time.time():
    raise KeyExpirationError(verification_key)

  return handler.verify_signature(
      signature_object, verification_key, content, SHA256) def export_pubkeys(keyids, homedir=None):
  """Exports multiple public keys from a GnuPG keyring.

  Arguments:
    keyids: A list of OpenPGP keyids in KEYID_SCHEMA format.
    homedir (optional): A path to the GnuPG home directory. If not set the
        default GnuPG home directory is used.

  Raises:
    TypeError: Keyids is not iterable.
    ValueError: A Keyid is not a string.
    UnsupportedLibraryError: The gpg command or pyca/cryptography are not
        available.
    KeyNotFoundError: No key or subkey was found for that keyid.

  Side Effects:
    Calls system gpg command in a subprocess.

  Returns:
    A dict of OpenPGP public key objects in GPG_PUBKEY_SCHEMA format as values,
    and their keyids as dict keys.


  """
  public_key_dict = {}
  for gpg_keyid in keyids:
    public_key = export_pubkey(gpg_keyid, homedir=homedir)
    keyid = public_key["keyid"]
    public_key_dict[keyid] = public_key

  return public_key_dict def get_version():
  """
  <Purpose>
    Uses `gpg2 --version` to get the version info of the installed gpg2
    and extracts and returns the version number.

    The executed base command is defined in constants.GPG_VERSION_COMMAND.

  <Exceptions>
    securesystemslib.exceptions.UnsupportedLibraryError:
            If the gpg command is not available

  <Returns>
    Version number string, e.g. "2.1.22"

  """
  if not constants.HAVE_GPG: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(constants.NO_GPG_MSG)

  command = constants.GPG_VERSION_COMMAND
  gpg_process = process.run(command, stdout=process.PIPE,
      stderr=process.PIPE, universal_newlines=True)

  full_version_info = gpg_process.stdout
  version_string = re.search(r'(\d\.\d\.\d+)', full_version_info).group(1)

  return version_string def is_version_fully_supported():
  """
  <Purpose>
    Compares the version of installed gpg2 with the minimal fully supported
    gpg2 version (2.1.0).

  <Returns>
    True if the version returned by `get_version` is greater-equal
    constants.FULLY_SUPPORTED_MIN_VERSION, False otherwise.

  """

  installed_version = get_version()
  # Excluded so that coverage does not vary in different test environments
  if (StrictVersion(installed_version) >=
      StrictVersion(constants.FULLY_SUPPORTED_MIN_VERSION)): # pragma: no cover
    return True

  else: # pragma: no cover
    return False def get_hashing_class(hash_algorithm_id):
  """
  <Purpose>
    Return a pyca/cryptography hashing class reference for the passed RFC4880
    hash algorithm ID.

  <Arguments>
    hash_algorithm_id:
            one of SHA1, SHA256, SHA512 (see securesystemslib.gpg.constants)

  <Exceptions>
    ValueError
            if the passed hash_algorithm_id is not supported.

  <Returns>
    A pyca/cryptography hashing class

  """
  supported_hashing_algorithms = [constants.SHA1, constants.SHA256,
      constants.SHA512]
  corresponding_hashing_classes = [hashing.SHA1, hashing.SHA256,
      hashing.SHA512]

  # Map supported hash algorithm ids to corresponding hashing classes
  hashing_class = dict(zip(supported_hashing_algorithms,
      corresponding_hashing_classes))

  try:
    return hashing_class[hash_algorithm_id]

  except KeyError:
    raise ValueError("Hash algorithm '{}' not supported, must be one of '{}' "
        "(see RFC4880 9.4. Hash Algorithms).".format(hash_algorithm_id,
        supported_hashing_algorithms)) def parse_packet_header(data, expected_type=None):
  """
  <Purpose>
    Parse out packet type and header and body lengths from an RFC4880 packet.

  <Arguments>
    data:
            An RFC4880 packet as described in section 4.2 of the rfc.

    expected_type: (optional)
            Used to error out if the packet does not have the expected
            type. See securesystemslib.gpg.constants.PACKET_TYPE_* for
            available types.

  <Exceptions>
    securesystemslib.gpg.exceptions.PacketParsingError
            If the new format packet length encodes a partial body length
            If the old format packet length encodes an indeterminate length
            If header or body length could not be determined
            If the expected_type was passed and does not match the packet type

    IndexError
            If the passed data is incomplete

  <Side Effects>
    None.

  <Returns>
    A tuple of packet type, header length, body length and packet length.
    (see  RFC4880 4.3. for the list of available packet types)

  """
  data = bytearray(data)
  header_len = None
  body_len = None

  # If Bit 6 of 1st octet is set we parse a New Format Packet Length, and
  # an Old Format Packet Lengths otherwise
  if data[0] & 0b01000000:
    # In new format packet lengths the packet type is encoded in Bits 5-0 of
    # the 1st octet of the packet
    packet_type = data[0] & 0b00111111

    # The rest of the packet header is the body length header, which may
    # consist of one, two or five octets. To disambiguate the RFC, the first
    # octet of the body length header is the second octet of the packet.
    if data[1] < 192:
      header_len = 2
      body_len = data[1]

    elif data[1] >= 192 and data[1] <= 223:
      header_len = 3
      body_len = (data[1] - 192 << 8) + data[2] + 192

    elif data[1] >= 224 and data[1] < 255:
      raise PacketParsingError("New length "
          "format packets of partial body lengths are not supported")

    elif data[1] == 255:
      header_len = 6
      body_len = data[2] << 24 | data[3] << 16 | data[4] << 8 | data[5]

    else: # pragma: no cover
      # Unreachable: octet must be between 0 and 255
      raise PacketParsingError("Invalid new length")

  else:
    # In old format packet lengths the packet type is encoded in Bits 5-2 of
    # the 1st octet and the length type in Bits 1-0
    packet_type = (data[0] & 0b00111100) >> 2
    length_type = data[0] & 0b00000011

    # The body length is encoded using one, two, or four octets, starting
    # with the second octet of the packet
    if length_type == 0:
      body_len = data[1]
      header_len = 2

    elif length_type == 1:
      header_len = 3
      body_len = struct.unpack(">H", data[1:header_len])[0]

    elif length_type == 2:
      header_len = 5
      body_len = struct.unpack(">I", data[1:header_len])[0]

    elif length_type == 3:
      raise PacketParsingError("Old length "
          "format packets of indeterminate length are not supported")

    else: # pragma: no cover (unreachable)
      # Unreachable: bits 1-0 must be one of 0 to 3
      raise PacketParsingError("Invalid old length")

  if header_len is None or body_len is None: # pragma: no cover
    # Unreachable: One of above must have assigned lengths or raised error
    raise PacketParsingError("Could not determine packet length")

  if expected_type is not None and packet_type != expected_type:
    raise PacketParsingError("Expected packet "
        "{}, but got {} instead!".format(expected_type, packet_type))

  return packet_type, header_len, body_len, header_len + body_len def parse_subpacket_header(data):
  """ Parse out subpacket header as per RFC4880 5.2.3.1. Signature Subpacket
  Specification. """
  # NOTE: Although the RFC does not state it explicitly, the length encoded
  # in the header must be greater equal 1, as it includes the mandatory
  # subpacket type octet.
  # Hence, passed bytearrays like [0] or [255, 0, 0, 0, 0], which encode a
  # subpacket length 0  are invalid.
  # The caller has to deal with the resulting IndexError.
  if data[0] < 192:
    length_len = 1
    length = data[0]

  elif data[0] >= 192 and data[0] < 255:
    length_len = 2
    length = ((data[0] - 192 << 8) + (data[1] + 192))

  elif data[0] == 255:
    length_len = 5
    length = struct.unpack(">I", data[1:length_len])[0]

  else: # pragma: no cover (unreachable)
    raise PacketParsingError("Invalid subpacket header")

  return data[length_len], length_len + 1, length - 1, length_len + length def create_pubkey(pubkey_info):
  """
  <Purpose>
    Create and return an RSAPublicKey object from the passed pubkey_info
    using pyca/cryptography.

  <Arguments>
    pubkey_info:
            The RSA pubkey info dictionary as specified by
            securesystemslib.formats.GPG_RSA_PUBKEY_SCHEMA

  <Exceptions>
    securesystemslib.exceptions.FormatError if
      pubkey_info does not match securesystemslib.formats.GPG_RSA_PUBKEY_SCHEMA

    securesystemslib.exceptions.UnsupportedLibraryError if
      the cryptography module is unavailable

  <Returns>
    A cryptography.hazmat.primitives.asymmetric.rsa.RSAPublicKey based on the
    passed pubkey_info.

  """
  if not CRYPTO: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_CRYPTO_MSG)

  formats.GPG_RSA_PUBKEY_SCHEMA.check_match(pubkey_info)

  e = int(pubkey_info['keyval']['public']['e'], 16)
  n = int(pubkey_info['keyval']['public']['n'], 16)
  pubkey = rsa.RSAPublicNumbers(e, n).public_key(backends.default_backend())

  return pubkey def create_pubkey(pubkey_info):
  """
  <Purpose>
    Create and return a DSAPublicKey object from the passed pubkey_info
    using pyca/cryptography.

  <Arguments>
    pubkey_info:
            The DSA pubkey info dictionary as specified by
            securesystemslib.formats.GPG_DSA_PUBKEY_SCHEMA

  <Exceptions>
    securesystemslib.exceptions.FormatError if
      pubkey_info does not match securesystemslib.formats.GPG_DSA_PUBKEY_SCHEMA

    securesystemslib.exceptions.UnsupportedLibraryError if
      the cryptography module is not available

  <Returns>
    A cryptography.hazmat.primitives.asymmetric.dsa.DSAPublicKey based on the
    passed pubkey_info.

  """
  if not CRYPTO: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_CRYPTO_MSG)

  formats.GPG_DSA_PUBKEY_SCHEMA.check_match(pubkey_info)

  y = int(pubkey_info['keyval']['public']['y'], 16)
  g = int(pubkey_info['keyval']['public']['g'], 16)
  p = int(pubkey_info['keyval']['public']['p'], 16)
  q = int(pubkey_info['keyval']['public']['q'], 16)
  parameter_numbers = dsa.DSAParameterNumbers(p, q, g)
  pubkey = dsa.DSAPublicNumbers(y, parameter_numbers).public_key(
      backends.default_backend())

  return pubkey def create_pubkey(pubkey_info):
  """
  <Purpose>
    Create and return an Ed25519PublicKey object from the passed pubkey_info
    using pyca/cryptography.

  <Arguments>
    pubkey_info:
          The ED25519 public key dictionary as specified by
          securesystemslib.formats.GPG_ED25519_PUBKEY_SCHEMA

  <Exceptions>
    securesystemslib.exceptions.FormatError if
      pubkey_info does not match securesystemslib.formats.GPG_DSA_PUBKEY_SCHEMA

    securesystemslib.exceptions.UnsupportedLibraryError if
      the cryptography module is unavailable

  <Returns>
    A cryptography.hazmat.primitives.asymmetric.ed25519.Ed25519PublicKey based
    on the passed pubkey_info.

  """
  if not CRYPTO: # pragma: no cover
    raise exceptions.UnsupportedLibraryError(NO_CRYPTO_MSG)

  formats.GPG_ED25519_PUBKEY_SCHEMA.check_match(pubkey_info)

  public_bytes = binascii.unhexlify(pubkey_info["keyval"]["public"]["q"])
  public_key = pyca_ed25519.Ed25519PublicKey.from_public_bytes(public_bytes)

  return public_key ED25519_SIG_LENGTH = 64 def parse_pubkey_payload(data):
  """
  <Purpose>
    Parse the passed public-key packet (payload only) and construct a
    public key dictionary.

  <Arguments>
    data:
          An RFC4880 public key packet payload as described in section 5.5.2.
          (version 4) of the RFC.

          NOTE: The payload can be parsed from a full key packet (header +
          payload) by using securesystemslib.gpg.util.parse_packet_header.

          WARNING: this doesn't support armored pubkey packets, so use with
          care. pubkey packets are a little bit more complicated than the
          signature ones

  <Exceptions>
    ValueError
          If the passed public key data is empty.

    securesystemslib.gpg.exceptions.PacketVersionNotSupportedError
          If the packet version does not match
          securesystemslib.gpg.constants.SUPPORTED_PUBKEY_PACKET_VERSIONS

    securesystemslib.gpg.exceptions.SignatureAlgorithmNotSupportedError
          If the signature algorithm does not match one of
          securesystemslib.gpg.constants.SUPPORTED_SIGNATURE_ALGORITHMS

  <Side Effects>
    None.

  <Returns>
    A public key in the format securesystemslib.formats.GPG_PUBKEY_SCHEMA

  """
  if not data:
    raise ValueError("Could not parse empty pubkey payload.")

  ptr = 0
  keyinfo = {}
  version_number = data[ptr]
  ptr += 1
  if version_number not in SUPPORTED_PUBKEY_PACKET_VERSIONS:
    raise PacketVersionNotSupportedError(
        "Pubkey packet version '{}' not supported, must be one of {}".format(
        version_number, SUPPORTED_PUBKEY_PACKET_VERSIONS))

  # NOTE: Uncomment this line to decode the time of creation
  time_of_creation = struct.unpack(">I", data[ptr:ptr + 4])
  ptr += 4

  algorithm = data[ptr]

  ptr += 1

  # TODO: Should we only export keys with signing capabilities?
  # Section 5.5.2 of RFC4880 describes a public-key algorithm octet with one
  # of the values described in section 9.1 that could be used to determine the
  # capabilities. However, in case of RSA subkeys this field doesn't seem to
  # correctly encode the capabilities. It always has the value 1, i.e.
  # RSA (Encrypt or Sign).
  # For RSA public keys we would have to parse the subkey's signature created
  # with the master key, for the signature's key flags subpacket, identified
  # by the value 27 (see section 5.2.3.1.) containing a list of binary flags
  # as described in section 5.2.3.21.
  if algorithm not in SUPPORTED_SIGNATURE_ALGORITHMS:
    raise SignatureAlgorithmNotSupportedError("Signature algorithm '{}' not "
        "supported, please verify that your gpg configuration is creating "
        "either DSA, RSA, or EdDSA signatures (see RFC4880 9.1. Public-Key "
        "Algorithms).".format(algorithm))

  keyinfo['type'] = SUPPORTED_SIGNATURE_ALGORITHMS[algorithm]['type']
  keyinfo['method'] = SUPPORTED_SIGNATURE_ALGORITHMS[algorithm]['method']
  handler = SIGNATURE_HANDLERS[keyinfo['type']]
  keyinfo['keyid'] = gpg_util.compute_keyid(data)
  key_params = handler.get_pubkey_params(data[ptr:])

  return {
    "method": keyinfo['method'],
    "type": keyinfo['type'],
    "hashes": [formats.GPG_HASH_ALGORITHM_STRING],
    "creation_time": time_of_creation[0],
    "keyid": keyinfo['keyid'],
    "keyval" : {
      "private": "",
      "public": key_params
      }
    } def parse_pubkey_bundle(data):
  """
  <Purpose>
    Parse packets from passed gpg public key data, associating self-signatures
    with the packets they correspond to, based on the structure of V4 keys
    defined in RFC4880 12.1 Key Structures.

    The returned raw key bundle may be used to further enrich the master key,
    with certified information (e.g. key expiration date) taken from
    self-signatures, and/or to verify that the parsed subkeys are bound to the
    primary key via signatures.

  <Arguments>
    data:
          Public key data as written to stdout by GPG_EXPORT_PUBKEY_COMMAND.

  <Exceptions>
    securesystemslib.gpg.exceptions.PacketParsingError
          If data is empty.
          If data cannot be parsed.

  <Side Effects>
    None.

  <Returns>
    A raw public key bundle where self-signatures are associated with their
    corresponding packets. See `key_bundle` for details.

  """
  if not data:
    raise PacketParsingError("Cannot parse keys from empty gpg data.")

  # Temporary data structure to hold parsed gpg packets
  key_bundle = {
    PACKET_TYPE_PRIMARY_KEY: {
      "key": {},
      "packet": None,
      "signatures": []
    },
    PACKET_TYPE_USER_ID: collections.OrderedDict(),
    PACKET_TYPE_USER_ATTR: collections.OrderedDict(),
    PACKET_TYPE_SUB_KEY: collections.OrderedDict()
  }

  # Iterate over gpg data and parse out packets of different types
  position = 0
  while position < len(data):
    try:
      packet_type, header_len, body_len, packet_length = \
          gpg_util.parse_packet_header(data[position:])

      packet = data[position:position+packet_length]
      payload = packet[header_len:]
      # The first (and only the first) packet in the bundle must be the master
      # key.  See RFC4880 12.1 Key Structures, V4 version keys
      # TODO: Do we need additional key structure assertions? e.g.
      # - there must be least one User ID packet, or
      # - order and type of signatures, or
      # - disallow duplicate packets
      if packet_type != PACKET_TYPE_PRIMARY_KEY and \
          not key_bundle[PACKET_TYPE_PRIMARY_KEY]["key"]:
        raise PacketParsingError("First packet must be a primary key ('{}'), "
            "got '{}'.".format(PACKET_TYPE_PRIMARY_KEY, packet_type))

      elif packet_type == PACKET_TYPE_PRIMARY_KEY and \
          key_bundle[PACKET_TYPE_PRIMARY_KEY]["key"]:
        raise PacketParsingError("Unexpected primary key.")

      # Fully parse master key to fail early, e.g. if key is malformed
      # or not supported, but also retain original packet for subkey binding
      # signature verification
      elif packet_type == PACKET_TYPE_PRIMARY_KEY:
        key_bundle[PACKET_TYPE_PRIMARY_KEY] = {
          "key": parse_pubkey_payload(bytearray(payload)),
          "packet": packet,
          "signatures": []
        }

      # Other non-signature packets in the key bundle include User IDs and User
      # Attributes, required to verify primary key certificates, and subkey
      # packets. For each packet we create a new ordered dictionary entry. We
      # use a dictionary to aggregate signatures by packet below,
      # and it must be ordered because each signature packet belongs to the
      # most recently parsed packet of a type.
      elif packet_type in {PACKET_TYPE_USER_ID, PACKET_TYPE_USER_ATTR,
          PACKET_TYPE_SUB_KEY}:
        key_bundle[packet_type][packet] = {
          "header_len": header_len,
          "body_len": body_len,
          "signatures": []
        }

      # The remaining relevant packets are signatures, required to bind subkeys
      # to the primary key, or to gather additional information about the
      # primary key, e.g. expiration date.
      # A signature corresponds to the most recently parsed packet of a type,
      # where the type is given by the availability of respective packets.
      # We test availability and assign accordingly as per the order of packet
      # types defined in RFC4880 12.1 (bottom-up).
      elif packet_type == PACKET_TYPE_SIGNATURE:
        for _type in [PACKET_TYPE_SUB_KEY, PACKET_TYPE_USER_ATTR,
            PACKET_TYPE_USER_ID]:
          if key_bundle[_type]:
            # Add to most recently added packet's signatures of matching type
            key_bundle[_type][next(reversed(key_bundle[_type]))]\
                ["signatures"].append(packet)
            break

        else:
          # If no packets are available for any of above types (yet), the
          # signature belongs to the primary key
          key_bundle[PACKET_TYPE_PRIMARY_KEY]["signatures"].append(packet)

      else:
        log.info("Ignoring gpg key packet '{}', we only handle packets of "
            "types '{}' (see RFC4880 4.3. Packet Tags).".format(packet_type,
            [PACKET_TYPE_PRIMARY_KEY, PACKET_TYPE_USER_ID,
            PACKET_TYPE_USER_ATTR, PACKET_TYPE_SUB_KEY,
            PACKET_TYPE_SIGNATURE]))

    # Both errors might be raised in parse_packet_header and in this loop
    except (PacketParsingError, IndexError) as e:
      raise PacketParsingError("Invalid public key data at position {}: {}."
          .format(position, e))

    # Go to next packet
    position += packet_length

  return key_bundle def get_pubkey_bundle(data, keyid):
  """
  <Purpose>
    Call function to extract and verify master key and subkeys from the passed
    gpg key data, where either the master key or one of the subkeys matches the
    passed keyid.

    NOTE:
    - If the keyid matches one of the subkeys, a warning is issued to notify
      the user about potential privilege escalation
    - Subkeys with invalid key binding signatures are discarded

  <Arguments>
    data:
          Public key data as written to stdout by
          securesystemslib.gpg.constants.GPG_EXPORT_PUBKEY_COMMAND.

    keyid:
          The keyid of the master key or one of its subkeys expected to be
          contained in the passed gpg data.

  <Exceptions>
    securesystemslib.gpg.exceptions.PacketParsingError
          If the key data could not be parsed

    securesystemslib.gpg.exceptions.KeyNotFoundError
          If the passed data is empty.
          If no master key or subkeys could be found that matches the passed
          keyid.

    securesystemslib.exceptions.FormatError
          If the passed keyid does not match
          securesystemslib.formats.KEYID_SCHEMA

  <Side Effects>
    None.

  <Returns>
    A public key in the format securesystemslib.formats.GPG_PUBKEY_SCHEMA with
    optional subkeys.

  """
  formats.KEYID_SCHEMA.check_match(keyid)
  if not data:
    raise KeyNotFoundError("Could not find gpg key '{}' in empty exported key "
        "data.".format(keyid))

  # Parse out master key and subkeys (enriched and verified via certificates
  # and binding signatures)
  raw_key_bundle = parse_pubkey_bundle(data)
  master_public_key = _assign_certified_key_info(raw_key_bundle)
  sub_public_keys = _get_verified_subkeys(raw_key_bundle)

  # Since GPG returns all pubkeys associated with a keyid (master key and
  # subkeys) we check which key matches the passed keyid.
  # If the matching key is a subkey, we warn the user because we return
  # the whole bundle (master plus all subkeys) and not only the subkey.
  # If no matching key is found we raise a KeyNotFoundError.
  for idx, public_key in enumerate(
      [master_public_key] + list(sub_public_keys.values())):
    if public_key and public_key["keyid"].endswith(keyid.lower()):
      if idx > 1:
        log.warning("Exporting master key '{}' including subkeys '{}' for"
            " passed keyid '{}'.".format(master_public_key["keyid"],
            ", ".join(list(sub_public_keys.keys())), keyid))
      break

  else:
    raise KeyNotFoundError("Could not find gpg key '{}' in exported key data."
        .format(keyid))

  # Add subkeys dictionary to master pubkey "subkeys" field if subkeys exist
  if sub_public_keys:
    master_public_key["subkeys"] = sub_public_keys

  return master_public_key def _assign_certified_key_info(bundle):
  """
  <Purpose>
    Helper function to verify User ID certificates corresponding to a gpg
    master key, in order to enrich the master key with additional information
    (e.g. expiration dates). The enriched master key is returned.

    NOTE: Currently we only consider User ID certificates. We can do the same
    for User Attribute certificates by iterating over
    bundle[PACKET_TYPE_USER_ATTR] instead of bundle[PACKET_TYPE_USER_ID], and
    replacing the signed_content constant '\xb4'  with '\xd1' (see RFC4880
    section 5.2.4. paragraph 4).

  <Arguments>
    bundle:
          GPG key bundle as parsed in parse_pubkey_bundle().

  <Exceptions>
    None.

  <Side Effects>
    None.

  <Returns>
    A public key in the format securesystemslib.formats.GPG_PUBKEY_SCHEMA.

  """
  # Create handler shortcut
  handler = SIGNATURE_HANDLERS[bundle[PACKET_TYPE_PRIMARY_KEY]["key"]["type"]]

  is_primary_user = False
  validity_period = None
  sig_creation_time = None

  # Verify User ID signatures to gather information about primary key
  # (see Notes about certification signatures in RFC 4880 5.2.3.3.)
  for user_id_packet, packet_data in bundle[PACKET_TYPE_USER_ID].items():
    # Construct signed content (see RFC4880 section 5.2.4. paragraph 4)
    signed_content = (bundle[PACKET_TYPE_PRIMARY_KEY]["packet"] +
        b"\xb4\x00\x00\x00" + user_id_packet[1:])
    for signature_packet in packet_data["signatures"]:
      try:
        signature = parse_signature_packet(signature_packet,
            supported_hash_algorithms={SHA1, SHA256, SHA512},
            supported_signature_types=SIGNATURE_TYPE_CERTIFICATES,
            include_info=True)
        # verify_signature requires a "keyid" even if it is short.
        # (see parse_signature_packet for more information about keyids)
        signature["keyid"] = signature["keyid"] or signature["short_keyid"]

      # TODO: Revise exception taxonomy:
      # It's okay to ignore some exceptions (unsupported algorithms etc.) but
      # we should blow up if a signature is malformed (missing subpackets).
      except Exception as e:
        log.info(e)
        continue

      if not bundle[PACKET_TYPE_PRIMARY_KEY]["key"]["keyid"].endswith(
          signature["keyid"]):
        log.info("Ignoring User ID certificate issued by '{}'.".format(
            signature["keyid"]))
        continue

      is_valid = handler.verify_signature(signature,
          bundle[PACKET_TYPE_PRIMARY_KEY]["key"], signed_content,
          signature["info"]["hash_algorithm"])

      if not is_valid:
        log.info("Ignoring invalid User ID self-certificate issued "
            "by '{}'.".format(signature["keyid"]))
        continue

      # If the signature is valid, we try to extract subpackets relevant to
      # the primary key, i.e. expiration time.
      # NOTE: There might be multiple User IDs per primary key and multiple
      # certificates per User ID. RFC4880 5.2.3.19. and last paragraph of
      # 5.2.3.3. provides some suggestions about ambiguity, but delegates the
      # responsibility to the implementer.

      # Ambiguity resolution scheme:
      # We take the key expiration time from the most recent certificate, i.e.
      # the certificate with the highest signature creation time. Additionally,
      # we prioritize certificates with primary user id flag set True. Note
      # that, if the ultimately prioritized certificate does not have a key
      # expiration time subpacket, we don't assign one, even if there were
      # certificates of lower priority carrying that subpacket.
      tmp_validity_period = \
          signature["info"]["subpackets"].get(KEY_EXPIRATION_SUBPACKET)

      # No key expiration time, go to next certificate
      if tmp_validity_period is None:
        continue

      # Create shortcut to mandatory pre-parsed creation time subpacket
      tmp_sig_creation_time = signature["info"]["creation_time"]

      tmp_is_primary_user = \
          signature["info"]["subpackets"].get(PRIMARY_USERID_SUBPACKET)

      if tmp_is_primary_user is not None:
        tmp_is_primary_user = bool(tmp_is_primary_user[0])

      # If we already have a primary user certified expiration date and this
      # is none, we don't consider it, and go to next certificate
      if is_primary_user and not tmp_is_primary_user:
        continue

      if not sig_creation_time or sig_creation_time < tmp_sig_creation_time:
        # This is the most recent certificate that has a validity_period and
        # doesn't have lower priority in regard to the primary user id flag. We
        # accept it the keys validty_period, until we get a newer value from
        # a certificate with higher priority.
        validity_period = struct.unpack(">I", tmp_validity_period)[0]
        # We also keep track of the used certificate's primary user id flag and
        # the signature creation time, for prioritization.
        is_primary_user = tmp_is_primary_user
        sig_creation_time = tmp_sig_creation_time

  if validity_period is not None:
    bundle[PACKET_TYPE_PRIMARY_KEY]["key"]["validity_period"] = validity_period

  return bundle[PACKET_TYPE_PRIMARY_KEY]["key"] def _get_verified_subkeys(bundle):
  """
  <Purpose>
    Helper function to verify the subkey binding signature for all subkeys in
    the passed bundle in order to enrich subkeys with additional information
    (e.g. expiration dates). Only valid (i.e. parsable) subkeys that are
    verifiably bound to the the master key of the bundle are returned. All
    other subkeys are discarded.

  <Arguments>
    bundle:
          GPG key bundle as parsed in parse_pubkey_bundle().

  <Exceptions>
    None.

  <Side Effects>
    None.

  <Returns>
    A dictionary of public keys in the format
    securesystemslib.formats.GPG_PUBKEY_SCHEMA, with keyids as dict keys.

  """
  # Create handler shortcut
  handler = SIGNATURE_HANDLERS[bundle[PACKET_TYPE_PRIMARY_KEY]["key"]["type"]]

  # Verify subkey binding signatures and only keep verified keys
  # See notes about subkey binding signature in RFC4880 5.2.3.3
  verified_subkeys = {}
  for subkey_packet, packet_data in bundle[PACKET_TYPE_SUB_KEY].items():
    try:
      # Parse subkey if possible and skip if invalid (e.g. not-supported)
      subkey = parse_pubkey_payload(
          bytearray(subkey_packet[-packet_data["body_len"]:]))

    # TODO: Revise exception taxonomy
    except Exception as e:
      log.info(e)
      continue

    # Construct signed content (see RFC4880 section 5.2.4. paragraph 3)
    signed_content = (bundle[PACKET_TYPE_PRIMARY_KEY]["packet"] + b"\x99" +
        subkey_packet[1:])

    # Filter sub key binding signature from other signatures, e.g. subkey
    # binding revocation signatures
    key_binding_signatures = []
    for signature_packet in packet_data["signatures"]:
      try:
        signature = parse_signature_packet(signature_packet,
            supported_hash_algorithms={SHA1, SHA256, SHA512},
            supported_signature_types={SIGNATURE_TYPE_SUB_KEY_BINDING},
            include_info=True)
        # verify_signature requires a "keyid" even if it is short.
        # (see parse_signature_packet for more information about keyids)
        signature["keyid"] = signature["keyid"] or signature["short_keyid"]
        key_binding_signatures.append(signature)

      # TODO: Revise exception taxonomy
      except Exception as e:
        log.info(e)
        continue
    # NOTE: As per the V4 key structure diagram in RFC4880 section 12.1., a
    # subkey must be followed by exactly one Primary-Key-Binding-Signature.
    # Based on inspection of real-world keys and other parts of the RFC (e.g.
    # the paragraph below the diagram and paragraph 0x18: Subkey Binding
    # Signature in section 5.2.1.) the mandated signature is actually a
    # *subkey binding signature*, which in case of a signing subkey, must have
    # an *embedded primary key binding signature*.
    if len(key_binding_signatures) != 1:
      log.info("Ignoring subkey '{}' due to wrong amount of key binding "
          "signatures ({}), must be exactly 1.".format(subkey["keyid"],
          len(key_binding_signatures)))
      continue
    is_valid = handler.verify_signature(signature,
        bundle[PACKET_TYPE_PRIMARY_KEY]["key"], signed_content,
        signature["info"]["hash_algorithm"])

    if not is_valid:
      log.info("Ignoring subkey '{}' due to invalid key binding signature."
          .format(subkey["keyid"]))
      continue

    # If the signature is valid, we may also extract relevant information from
    # its "info" field (e.g. subkey expiration date) and assign to it to the
    # subkey here
    validity_period = \
        signature["info"]["subpackets"].get(KEY_EXPIRATION_SUBPACKET)
    if validity_period is not None:
      subkey["validity_period"] = struct.unpack(">I", validity_period)[0]

    verified_subkeys[subkey["keyid"]] = subkey

  return verified_subkeys def parse_signature_packet(data, supported_signature_types=None,
    supported_hash_algorithms=None, include_info=False):
  """
  <Purpose>
    Parse the signature information on an RFC4880-encoded binary signature data
    buffer.

    NOTE: Older gpg versions (< FULLY_SUPPORTED_MIN_VERSION) might only
    reveal the partial key id. It is the callers responsibility to determine
    the full keyid based on the partial keyid, e.g. by exporting the related
    public and replacing the partial keyid with the full keyid.

  <Arguments>
    data:
           the RFC4880-encoded binary signature data buffer as described in
           section 5.2 (and 5.2.3.1).
    supported_signature_types: (optional)
          a set of supported signature_types, the signature packet may be
          (see securesystemslib.gpg.constants for available types). If None is
          specified the signature packet must be of type SIGNATURE_TYPE_BINARY.
    supported_hash_algorithms: (optional)
          a set of supported hash algorithm ids, the signature packet
          may use. Available ids are SHA1, SHA256, SHA512 (see
          securesystemslib.gpg.constants). If None is specified, the signature
          packet must use SHA256.
    include_info: (optional)
          a boolean that indicates whether an opaque dictionary should be
          added to the returned signature under the key "info". Default is
          False.

  <Exceptions>
    ValueError: if the signature packet is not supported or the data is
      malformed
    IndexError: if the signature packet is incomplete

  <Side Effects>
    None.

  <Returns>
    A signature dictionary matching
    securesystemslib.formats.GPG_SIGNATURE_SCHEMA with the following special
    characteristics:
     - The "keyid" field is an empty string if it cannot be determined
     - The "short_keyid" is not added if it cannot be determined
     - At least one of non-empty "keyid" or "short_keyid" are part of the
       signature

  """
  if not supported_signature_types:
    supported_signature_types = {SIGNATURE_TYPE_BINARY}

  if not supported_hash_algorithms:
    supported_hash_algorithms = {SHA256}

  _, header_len, _, packet_len = gpg_util.parse_packet_header(
      data, PACKET_TYPE_SIGNATURE)

  data = bytearray(data[header_len:packet_len])

  ptr = 0

  # we get the version number, which we also expect to be v4, or we bail
  # FIXME: support v3 type signatures (which I haven't seen in the wild)
  version_number = data[ptr]
  ptr += 1
  if version_number not in SUPPORTED_SIGNATURE_PACKET_VERSIONS:
    raise ValueError("Signature version '{}' not supported, must be one of "
        "{}.".format(version_number, SUPPORTED_SIGNATURE_PACKET_VERSIONS))

  # Per default we only parse "signatures of a binary document". Other types
  # may be allowed by passing type constants via `supported_signature_types`.
  # Types include revocation signatures, key binding signatures, persona
  # certifications, etc. (see RFC 4880 section 5.2.1.).
  signature_type = data[ptr]
  ptr += 1

  if signature_type not in supported_signature_types:
    raise ValueError("Signature type '{}' not supported, must be one of {} "
        "(see RFC4880 5.2.1. Signature Types).".format(signature_type,
        supported_signature_types))

  signature_algorithm = data[ptr]
  ptr += 1

  if signature_algorithm not in SUPPORTED_SIGNATURE_ALGORITHMS:
    raise ValueError("Signature algorithm '{}' not "
        "supported, please verify that your gpg configuration is creating "
        "either DSA, RSA, or EdDSA signatures (see RFC4880 9.1. Public-Key "
        "Algorithms).".format(signature_algorithm))

  key_type = SUPPORTED_SIGNATURE_ALGORITHMS[signature_algorithm]['type']
  handler = SIGNATURE_HANDLERS[key_type]

  hash_algorithm = data[ptr]
  ptr += 1

  if hash_algorithm not in supported_hash_algorithms:
    raise ValueError("Hash algorithm '{}' not supported, must be one of {}"
        " (see RFC4880 9.4. Hash Algorithms).".format(hash_algorithm,
        supported_hash_algorithms))

  # Obtain the hashed octets
  hashed_octet_count = struct.unpack(">H", data[ptr:ptr+2])[0]
  ptr += 2
  hashed_subpackets = data[ptr:ptr+hashed_octet_count]
  hashed_subpacket_info = gpg_util.parse_subpackets(hashed_subpackets)

  # Check whether we were actually able to read this much hashed octets
  if len(hashed_subpackets) != hashed_octet_count: # pragma: no cover
    raise ValueError("This signature packet seems to be corrupted."
        "It is missing hashed octets!")

  ptr += hashed_octet_count
  other_headers_ptr = ptr

  unhashed_octet_count = struct.unpack(">H", data[ptr: ptr + 2])[0]
  ptr += 2

  unhashed_subpackets = data[ptr:ptr+unhashed_octet_count]
  unhashed_subpacket_info = gpg_util.parse_subpackets(unhashed_subpackets)

  ptr += unhashed_octet_count

  # Use the info dict to return further signature information that may be
  # needed for intermediate processing, but does not have to be on the eventual
  # signature datastructure
  info = {
    "signature_type": signature_type,
    "hash_algorithm": hash_algorithm,
    "creation_time": None,
    "subpackets": {},
  }

  keyid = ""
  short_keyid = ""


  # Parse "Issuer" (short keyid) and "Issuer Fingerprint" (full keyid) type
  # subpackets
  # Strategy: Loop over all unhashed and hashed subpackets (in that order!) and
  # store only the last of a type. Due to the order in the loop, hashed
  # subpackets are prioritized over unhashed subpackets (see NOTEs below).

  # NOTE: A subpacket may be found either in the hashed or unhashed subpacket
  # sections of a signature. If a subpacket is not hashed, then the information
  # in it cannot be considered definitive because it is not part of the
  # signature proper. (see RFC4880 5.2.3.2.)
  # NOTE: Signatures may contain conflicting information in subpackets. In most
  # cases, an implementation SHOULD use the last subpacket, but MAY use any
  # conflict resolution scheme that makes more sense. (see RFC4880 5.2.4.1.)
  for idx, subpacket_tuple in \
      enumerate(unhashed_subpacket_info + hashed_subpacket_info):

    # The idx indicates if the info is from the unhashed (first) or
    # hashed (second) of the above concatenated lists
    is_hashed = (idx >= len(unhashed_subpacket_info))
    subpacket_type, subpacket_data = subpacket_tuple

    # Warn if expiration subpacket is not hashed
    if subpacket_type == KEY_EXPIRATION_SUBPACKET:
      if not is_hashed:
        log.warning("Expiration subpacket not hashed, gpg client possibly "
            "exporting a weakly configured key.")

    # Full keyids are only available in newer signatures
    # (see RFC4880 and rfc4880bis-06 5.2.3.1.)
    if subpacket_type == FULL_KEYID_SUBPACKET: # pragma: no cover
      # Exclude from coverage for consistent results across test envs
      # NOTE: The first byte of the subpacket payload is a version number
      # (see rfc4880bis-06 5.2.3.28.)
      keyid = binascii.hexlify(subpacket_data[1:]).decode("ascii")

    # We also return the short keyid, because the full might not be available
    if subpacket_type == PARTIAL_KEYID_SUBPACKET:
      short_keyid = binascii.hexlify(subpacket_data).decode("ascii")

    if subpacket_type == SIG_CREATION_SUBPACKET:
      info["creation_time"] = struct.unpack(">I", subpacket_data)[0]

    info["subpackets"][subpacket_type] = subpacket_data


  # Fail if there is no keyid at all (this should not happen)
  if not (keyid or short_keyid): # pragma: no cover
    raise ValueError("This signature packet seems to be corrupted. It does "
        "not have an 'Issuer' or 'Issuer Fingerprint' subpacket (see RFC4880 "
        "and rfc4880bis-06 5.2.3.1. Signature Subpacket Specification).")

  # Fail if keyid and short keyid are specified but don't match
  if keyid and not keyid.endswith(short_keyid): # pragma: no cover
    raise ValueError("This signature packet seems to be corrupted. The key ID "
        "'{}' of the 'Issuer' subpacket must match the lower 64 bits of the "
        "fingerprint '{}' of the 'Issuer Fingerprint' subpacket (see RFC4880 "
        "and rfc4880bis-06 5.2.3.28. Issuer Fingerprint).".format(
        short_keyid, keyid))

  if not info["creation_time"]: # pragma: no cover
    raise ValueError("This signature packet seems to be corrupted. It does "
        "not have a 'Signature Creation Time' subpacket (see RFC4880 5.2.3.4 "
        "Signature Creation Time).")

  # Uncomment this variable to obtain the left-hash-bits information (used for
  # early rejection)
  #left_hash_bits = struct.unpack(">H", data[ptr:ptr+2])[0]
  ptr += 2

  # Finally, fetch the actual signature (as opposed to signature metadata).
  signature = handler.get_signature_params(data[ptr:])

  signature_data = {
    'keyid': "{}".format(keyid),
    'other_headers': binascii.hexlify(data[:other_headers_ptr]).decode('ascii'),
    'signature': binascii.hexlify(signature).decode('ascii')
  }

  if short_keyid: # pragma: no branch
    signature_data["short_keyid"] = short_keyid

  if include_info:
    signature_data["info"] = info

  return signature_data SHA1 = 0x02 SHA256 = 0x08 SHA512 = 0x0A GPG_EXPORT_PUBKEY_COMMAND = GPG_COMMAND + " {homearg} --export {keyid}" PACKET_TYPE_PRIMARY_KEY = 0x06 PACKET_TYPE_USER_ID = 0x0D PACKET_TYPE_USER_ATTR = 0x11 PACKET_TYPE_SUB_KEY = 0x0E HAVE_GPG = False class PacketParsingError(Exception):
  pass class PacketVersionNotSupportedError(Exception):
  pass class SignatureAlgorithmNotSupportedError(Exception):
  pass class KeyNotFoundError(Exception):
  pass class CommandError(Exception):
  pass class KeyExpirationError(Exception):
  def __init__(self, key):
    super(KeyExpirationError, self).__init__()
    self.key = key

  def __str__(self):
    creation_time = datetime.datetime.utcfromtimestamp(
        self.key["creation_time"])
    expiration_time = datetime.datetime.utcfromtimestamp(
        self.key["creation_time"] + self.key["validity_period"])
    validity_period = expiration_time - creation_time

    return ("GPG key '{}' created on '{:%Y-%m-%d %H:%M} UTC' with validity "
        "period '{}' expired on '{:%Y-%m-%d %H:%M} UTC'.".format(
        self.key["keyid"], creation_time, validity_period, expiration_time)) GPG_PUBKEY_SCHEMA = SCHEMA.OneOf([GPG_RSA_PUBKEY_SCHEMA,
    GPG_DSA_PUBKEY_SCHEMA, GPG_ED25519_PUBKEY_SCHEMA]) ANY_PUBKEY_DICT_SCHEMA = SCHEMA.DictOf(
  key_schema = KEYID_SCHEMA,
  value_schema = ANY_PUBKEY_SCHEMA)
import os
import sys
import shutil
import tempfile
import unittest
import cryptography.hazmat.primitives.serialization as serialization
import cryptography.hazmat.backends as backends
import cryptography.hazmat.primitives.hashes as hashing
  from unittest.mock import patch # pylint: disable=no-name-in-module,import-error
  from mock import patch # pylint: disable=import-error
from copy import deepcopy
from collections import OrderedDict
from securesystemslib import process
from securesystemslib.gpg.functions import (create_signature, export_pubkey,
    verify_signature, export_pubkeys)
from securesystemslib.gpg.util import (get_version, is_version_fully_supported,
    get_hashing_class, parse_packet_header, parse_subpacket_header)
from securesystemslib.gpg.rsa import create_pubkey as rsa_create_pubkey
from securesystemslib.gpg.dsa import create_pubkey as dsa_create_pubkey
from securesystemslib.gpg.eddsa import create_pubkey as eddsa_create_pubkey
from securesystemslib.gpg.eddsa import ED25519_SIG_LENGTH
from securesystemslib.gpg.common import (parse_pubkey_payload,
    parse_pubkey_bundle, get_pubkey_bundle, _assign_certified_key_info,
    _get_verified_subkeys, parse_signature_packet)
from securesystemslib.gpg.constants import (SHA1, SHA256, SHA512,
    GPG_EXPORT_PUBKEY_COMMAND, PACKET_TYPE_PRIMARY_KEY, PACKET_TYPE_USER_ID,
    PACKET_TYPE_USER_ATTR, PACKET_TYPE_SUB_KEY, HAVE_GPG)
from securesystemslib.gpg.exceptions import (PacketParsingError,
    PacketVersionNotSupportedError, SignatureAlgorithmNotSupportedError,
    KeyNotFoundError, CommandError, KeyExpirationError)
from securesystemslib.formats import (GPG_PUBKEY_SCHEMA,
    ANY_PUBKEY_DICT_SCHEMA)
#!/usr/bin/env python

"""
<Program Name>
  test_gpg.py

<Author>
  Santiago Torres-Arias <santiago@nyu.edu>
  Lukas Puehringer <lukas.puehringer@nyu.edu>

<Started>
  Nov 15, 2017

<Copyright>
  See LICENSE for licensing information.

<Purpose>
  Test gpg/pgp-related functions.

"""


if sys.version_info >= (3, 3):
else:





@unittest.skipIf(not HAVE_GPG, "gpg not found")
class TestUtil(unittest.TestCase):
  """Test util functions. """
  def test_version_utils_return_types(self):
    """Run dummy tests for coverage. """
    self.assertTrue(isinstance(get_version(), str))

    self.assertTrue(isinstance(is_version_fully_supported(), bool))