ANY_STRING_SCHEMA = SCHEMA.AnyString()
LIST_OF_ANY_STRING_SCHEMA = SCHEMA.ListOf(ANY_STRING_SCHEMA)
ISO8601_DATETIME_SCHEMA = SCHEMA.RegularExpression(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z')
UNIX_TIMESTAMP_SCHEMA = SCHEMA.Integer(lo=0, hi=2147483647)
HEX_SCHEMA = SCHEMA.RegularExpression(r'[a-fA-F0-9]+')
HASH_SCHEMA = HEX_SCHEMA
HASHDICT_SCHEMA = SCHEMA.DictOf(
  key_schema = SCHEMA.AnyString(),
  value_schema = HASH_SCHEMA)
URL_SCHEMA = SCHEMA.AnyString()
KEYID_SCHEMA = HASH_SCHEMA
KEYIDS_SCHEMA = SCHEMA.ListOf(KEYID_SCHEMA)
SCHEME_SCHEMA = SCHEMA.AnyString()
PATH_SCHEMA = SCHEMA.AnyNonemptyString()
PATHS_SCHEMA = SCHEMA.ListOf(PATH_SCHEMA)
LOGLEVEL_SCHEMA = SCHEMA.Integer(lo=0, hi=50)
NAME_SCHEMA = SCHEMA.AnyString()
NAMES_SCHEMA = SCHEMA.ListOf(NAME_SCHEMA)
DATA_SCHEMA = SCHEMA.AnyBytes()
TEXT_SCHEMA = SCHEMA.AnyString()
HASHALGORITHMS_SCHEMA = SCHEMA.ListOf(SCHEMA.OneOf(
  [SCHEMA.String('md5'), SCHEMA.String('sha1'),
   SCHEMA.String('sha224'), SCHEMA.String('sha256'),
   SCHEMA.String('sha384'), SCHEMA.String('sha512'),
   SCHEMA.String('blake2s'), SCHEMA.String('blake2b'),
   SCHEMA.String('blake2b-256')]))
ENCRYPTEDKEY_SCHEMA = SCHEMA.AnyString()
BOOLEAN_SCHEMA = SCHEMA.Boolean()
RSAKEYBITS_SCHEMA = SCHEMA.Integer(lo=2048)
ECDSA_SCHEME_SCHEMA = SCHEMA.RegularExpression(r'ecdsa-sha2-nistp(256|384)')
PYCACRYPTOSIGNATURE_SCHEMA = SCHEMA.AnyBytes()
PEMRSA_SCHEMA = SCHEMA.AnyString()
PEMECDSA_SCHEMA = SCHEMA.AnyString()
PASSWORD_SCHEMA = SCHEMA.AnyString()
PASSWORDS_SCHEMA = SCHEMA.ListOf(PASSWORD_SCHEMA)
KEYVAL_SCHEMA = SCHEMA.Object(
  object_name = 'KEYVAL_SCHEMA',
  public = SCHEMA.AnyString(),
  private = SCHEMA.Optional(SCHEMA.AnyString()))
PUBLIC_KEYVAL_SCHEMA = SCHEMA.Object(
  object_name = 'KEYVAL_SCHEMA',
  public = SCHEMA.AnyString(),
  private = SCHEMA.Optional(SCHEMA.String("")))
KEYTYPE_SCHEMA = SCHEMA.OneOf(
  [SCHEMA.String('rsa'), SCHEMA.String('ed25519'), SCHEMA.String('ecdsa'),
   SCHEMA.RegularExpression(r'ecdsa-sha2-nistp(256|384)')])
KEY_SCHEMA = SCHEMA.Object(
  object_name = 'KEY_SCHEMA',
  keytype = SCHEMA.AnyString(),
  scheme = SCHEME_SCHEMA,
  keyval = KEYVAL_SCHEMA,
  expires = SCHEMA.Optional(ISO8601_DATETIME_SCHEMA))
PUBLIC_KEY_SCHEMA = SCHEMA.Object(
  object_name = 'PUBLIC_KEY_SCHEMA',
  keytype = SCHEMA.AnyString(),
  keyid_hash_algorithms = SCHEMA.Optional(HASHALGORITHMS_SCHEMA),
  keyval = PUBLIC_KEYVAL_SCHEMA,
  expires = SCHEMA.Optional(ISO8601_DATETIME_SCHEMA))
ANYKEY_SCHEMA = SCHEMA.Object(
  object_name = 'ANYKEY_SCHEMA',
  keytype = KEYTYPE_SCHEMA,
  scheme = SCHEME_SCHEMA,
  keyid = KEYID_SCHEMA,
  keyid_hash_algorithms = SCHEMA.Optional(HASHALGORITHMS_SCHEMA),
  keyval = KEYVAL_SCHEMA,
  expires = SCHEMA.Optional(ISO8601_DATETIME_SCHEMA))
ANYKEYLIST_SCHEMA = SCHEMA.ListOf(ANYKEY_SCHEMA)
RSA_SCHEME_SCHEMA = SCHEMA.OneOf([
  SCHEMA.RegularExpression(r'rsassa-pss-(md5|sha1|sha224|sha256|sha384|sha512)'),
  SCHEMA.RegularExpression(r'rsa-pkcs1v15-(md5|sha1|sha224|sha256|sha384|sha512)')])
RSAKEY_SCHEMA = SCHEMA.Object(
  object_name = 'RSAKEY_SCHEMA',
  keytype = SCHEMA.String('rsa'),
  scheme = RSA_SCHEME_SCHEMA,
  keyid = KEYID_SCHEMA,
  keyid_hash_algorithms = SCHEMA.Optional(HASHALGORITHMS_SCHEMA),
  keyval = KEYVAL_SCHEMA)
ECDSAKEY_SCHEMA = SCHEMA.Object(
  object_name = 'ECDSAKEY_SCHEMA',
  keytype = SCHEMA.OneOf([SCHEMA.String('ecdsa'),
                          SCHEMA.RegularExpression(r'ecdsa-sha2-nistp(256|384)')]),
  scheme = ECDSA_SCHEME_SCHEMA,
  keyid = KEYID_SCHEMA,
  keyid_hash_algorithms = SCHEMA.Optional(HASHALGORITHMS_SCHEMA),
  keyval = KEYVAL_SCHEMA)
ED25519PUBLIC_SCHEMA = SCHEMA.LengthBytes(32)
ED25519SEED_SCHEMA = SCHEMA.LengthBytes(32)
ED25519SIGNATURE_SCHEMA = SCHEMA.LengthBytes(64)
ECDSASIGNATURE_SCHEMA = SCHEMA.AnyBytes()
ED25519_SIG_SCHEMA = SCHEMA.OneOf([SCHEMA.String('ed25519')])
ED25519KEY_SCHEMA = SCHEMA.Object(
  object_name = 'ED25519KEY_SCHEMA',
  keytype = SCHEMA.String('ed25519'),
  scheme = ED25519_SIG_SCHEMA,
  keyid = KEYID_SCHEMA,
  keyid_hash_algorithms = SCHEMA.Optional(HASHALGORITHMS_SCHEMA),
  keyval = KEYVAL_SCHEMA)
GPG_HASH_ALGORITHM_STRING = "pgp+SHA2"
GPG_RSA_PUBKEY_METHOD_STRING = "pgp+rsa-pkcsv1.5"
GPG_DSA_PUBKEY_METHOD_STRING = "pgp+dsa-fips-180-2"
GPG_ED25519_PUBKEY_METHOD_STRING = "pgp+eddsa-ed25519"
GPG_RSA_PUBKEYVAL_SCHEMA = SCHEMA.Object(
  object_name = "GPG_RSA_PUBKEYVAL_SCHEMA",
  e = SCHEMA.AnyString(),
  n = HEX_SCHEMA
)
_GPG_RSA_PUBKEY_SCHEMA = SCHEMA.Object(
  object_name = "GPG_RSA_PUBKEY_SCHEMA",
  type = SCHEMA.String("rsa"),
  method = SCHEMA.String(GPG_RSA_PUBKEY_METHOD_STRING),
  hashes = SCHEMA.ListOf(SCHEMA.String(GPG_HASH_ALGORITHM_STRING)),
  creation_time = SCHEMA.Optional(UNIX_TIMESTAMP_SCHEMA),
  validity_period = SCHEMA.Optional(SCHEMA.Integer(lo=0)),
  keyid = KEYID_SCHEMA,
  keyval = SCHEMA.Object(
      public = GPG_RSA_PUBKEYVAL_SCHEMA,
      private = SCHEMA.String("")
    )
)
GPG_RSA_PUBKEY_SCHEMA = _create_gpg_pubkey_with_subkey_schema(
    _GPG_RSA_PUBKEY_SCHEMA)
GPG_DSA_PUBKEYVAL_SCHEMA = SCHEMA.Object(
  object_name = "GPG_DSA_PUBKEYVAL_SCHEMA",
  y = HEX_SCHEMA,
  p = HEX_SCHEMA,
  q = HEX_SCHEMA,
  g = HEX_SCHEMA
)
_GPG_DSA_PUBKEY_SCHEMA = SCHEMA.Object(
  object_name = "GPG_DSA_PUBKEY_SCHEMA",
  type = SCHEMA.String("dsa"),
  method = SCHEMA.String(GPG_DSA_PUBKEY_METHOD_STRING),
  hashes = SCHEMA.ListOf(SCHEMA.String(GPG_HASH_ALGORITHM_STRING)),
  creation_time = SCHEMA.Optional(UNIX_TIMESTAMP_SCHEMA),
  validity_period = SCHEMA.Optional(SCHEMA.Integer(lo=0)),
  keyid = KEYID_SCHEMA,
  keyval = SCHEMA.Object(
      public = GPG_DSA_PUBKEYVAL_SCHEMA,
      private = SCHEMA.String("")
    )
)
GPG_DSA_PUBKEY_SCHEMA = _create_gpg_pubkey_with_subkey_schema(
    _GPG_DSA_PUBKEY_SCHEMA)
GPG_ED25519_PUBKEYVAL_SCHEMA = SCHEMA.Object(
  object_name = "GPG_ED25519_PUBKEYVAL_SCHEMA",
  q = HEX_SCHEMA,
)
_GPG_ED25519_PUBKEY_SCHEMA = SCHEMA.Object(
  object_name = "GPG_ED25519_PUBKEY_SCHEMA",
  type = SCHEMA.String("eddsa"),
  method = SCHEMA.String(GPG_ED25519_PUBKEY_METHOD_STRING),
  hashes = SCHEMA.ListOf(SCHEMA.String(GPG_HASH_ALGORITHM_STRING)),
  creation_time = SCHEMA.Optional(UNIX_TIMESTAMP_SCHEMA),
  validity_period = SCHEMA.Optional(SCHEMA.Integer(lo=0)),
  keyid = KEYID_SCHEMA,
  keyval = SCHEMA.Object(
      public = GPG_ED25519_PUBKEYVAL_SCHEMA,
      private = SCHEMA.String("")
    )
)
GPG_ED25519_PUBKEY_SCHEMA = _create_gpg_pubkey_with_subkey_schema(
    _GPG_ED25519_PUBKEY_SCHEMA)
GPG_PUBKEY_SCHEMA = SCHEMA.OneOf([GPG_RSA_PUBKEY_SCHEMA,
    GPG_DSA_PUBKEY_SCHEMA, GPG_ED25519_PUBKEY_SCHEMA])
GPG_SIGNATURE_SCHEMA = SCHEMA.Object(
    object_name = "SIGNATURE_SCHEMA",
    keyid = KEYID_SCHEMA,
    short_keyid = SCHEMA.Optional(KEYID_SCHEMA),
    other_headers = HEX_SCHEMA,
    signature = HEX_SCHEMA,
    info = SCHEMA.Optional(SCHEMA.Any()),
  )
SIGNATURE_SCHEMA = SCHEMA.Object(
  object_name = 'SIGNATURE_SCHEMA',
  keyid = KEYID_SCHEMA,
  sig = HEX_SCHEMA)
KEYDICT_SCHEMA = SCHEMA.DictOf(
  key_schema = KEYID_SCHEMA,
  value_schema = KEY_SCHEMA)
ANY_SIGNATURE_SCHEMA = SCHEMA.OneOf([SIGNATURE_SCHEMA,
    GPG_SIGNATURE_SCHEMA])
SIGNATURES_SCHEMA = SCHEMA.ListOf(ANY_SIGNATURE_SCHEMA)
SIGNABLE_SCHEMA = SCHEMA.Object(
  object_name = 'SIGNABLE_SCHEMA',
  signed = SCHEMA.Any(),
  signatures = SIGNATURES_SCHEMA)
ANY_VERIFICATION_KEY_SCHEMA = SCHEMA.OneOf([ANYKEY_SCHEMA,
    GPG_PUBKEY_SCHEMA])
VERIFICATION_KEY_DICT_SCHEMA = SCHEMA.DictOf(
  key_schema = KEYID_SCHEMA,
  value_schema = ANY_VERIFICATION_KEY_SCHEMA)
ANY_KEYDICT_SCHEMA = SCHEMA.OneOf([KEYDICT_SCHEMA,
    VERIFICATION_KEY_DICT_SCHEMA])
ANY_PUBKEY_SCHEMA = SCHEMA.OneOf([PUBLIC_KEY_SCHEMA, GPG_PUBKEY_SCHEMA])
ANY_PUBKEY_DICT_SCHEMA = SCHEMA.DictOf(
  key_schema = KEYID_SCHEMA,
  value_schema = ANY_PUBKEY_SCHEMA)
def _create_gpg_pubkey_with_subkey_schema(pubkey_schema):
def datetime_to_unix_timestamp(datetime_object):
def unix_timestamp_to_datetime(unix_timestamp):
def format_base64(data):
def parse_base64(base64_string):
def _canonical_string_encoder(string):
def _encode_canonical(object, output_function):
def encode_canonical(object, output_function=None): CRYPTO = True
NO_CRYPTO_MSG = 'gpg.utils requires the cryptography library'
  CRYPTO = False
def get_mpi_length(data):
def hash_object(headers, algorithm, content):
def parse_packet_header(data, expected_type=None):
def compute_keyid(pubkey_packet_data):
def parse_subpacket_header(data):
def parse_subpackets(data):
def get_version():
def is_version_fully_supported():
def get_hashing_class(hash_algorithm_id): class PacketVersionNotSupportedError(Exception):
  pass class SignatureAlgorithmNotSupportedError(Exception):
  pass class KeyNotFoundError(Exception):
  pass class PacketParsingError(Exception):
  pass PACKET_TYPE_PRIMARY_KEY = 0x06 PACKET_TYPE_USER_ID = 0x0D PACKET_TYPE_USER_ATTR = 0x11 PACKET_TYPE_SUB_KEY = 0x0E PACKET_TYPE_SIGNATURE = 0x02 SUPPORTED_PUBKEY_PACKET_VERSIONS = {0x04} SIGNATURE_TYPE_BINARY = 0x00 SIGNATURE_TYPE_CERTIFICATES = {0x10, 0x11, 0x12, 0x13} SIGNATURE_TYPE_SUB_KEY_BINDING = 0x18 SUPPORTED_SIGNATURE_PACKET_VERSIONS = {0x04} FULL_KEYID_SUBPACKET = 0x21 PARTIAL_KEYID_SUBPACKET = 0x10 SHA1 = 0x02 SHA256 = 0x08 SHA512 = 0x0A KEY_EXPIRATION_SUBPACKET = 0x09 PRIMARY_USERID_SUBPACKET = 0x19 SIG_CREATION_SUBPACKET = 0x02 SIGNATURE_HANDLERS = {
    "rsa": rsa,
    "dsa": dsa,
    "eddsa": eddsa
} SUPPORTED_SIGNATURE_ALGORITHMS = {
    0x01: {
      "type":"rsa",
      "method": "pgp+rsa-pkcsv1.5",
      "handler": rsa
    },
    0x11: {
      "type": "dsa",
      "method": "pgp+dsa-fips-180-2",
      "handler": dsa
    },
    0x16: {
      "type": "eddsa",
      "method": "pgp+eddsa-ed25519",
      "handler": eddsa
    }
}
import struct
import binascii
import logging
import collections
from securesystemslib import formats
from securesystemslib.gpg import util as gpg_util
from securesystemslib.gpg.exceptions import (PacketVersionNotSupportedError,
    SignatureAlgorithmNotSupportedError, KeyNotFoundError, PacketParsingError)
from securesystemslib.gpg.constants import (
    PACKET_TYPE_PRIMARY_KEY, PACKET_TYPE_USER_ID, PACKET_TYPE_USER_ATTR,
    PACKET_TYPE_SUB_KEY, PACKET_TYPE_SIGNATURE,
    SUPPORTED_PUBKEY_PACKET_VERSIONS, SIGNATURE_TYPE_BINARY,
    SIGNATURE_TYPE_CERTIFICATES, SIGNATURE_TYPE_SUB_KEY_BINDING,
    SUPPORTED_SIGNATURE_PACKET_VERSIONS,
    FULL_KEYID_SUBPACKET, PARTIAL_KEYID_SUBPACKET,
    SHA1,SHA256, SHA512, KEY_EXPIRATION_SUBPACKET, PRIMARY_USERID_SUBPACKET,
    SIG_CREATION_SUBPACKET)
from securesystemslib.gpg.handlers import (
    SIGNATURE_HANDLERS, SUPPORTED_SIGNATURE_ALGORITHMS)
      # 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 _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)