# This file is dual licensed under the terms of the Apache License, Version # 2.0, and the BSD License. See the LICENSE file in the root of this repository # for complete details. from __future__ import absolute_import, division, print_function import binascii import collections import math import re from contextlib import contextmanager import pytest import six from cryptography.exceptions import UnsupportedAlgorithm import cryptography_vectors HashVector = collections.namedtuple("HashVector", ["message", "digest"]) KeyedHashVector = collections.namedtuple( "KeyedHashVector", ["message", "digest", "key"] ) def select_backends(names, backend_list): if names is None: return backend_list split_names = [x.strip() for x in names.split(',')] selected_backends = [] for backend in backend_list: if backend.name in split_names: selected_backends.append(backend) if len(selected_backends) > 0: return selected_backends else: raise ValueError( "No backend selected. Tried to select: {0}".format(split_names) ) def skip_if_empty(backend_list, required_interfaces): if not backend_list: pytest.skip( "No backends provided supply the interface: {0}".format( ", ".join(iface.__name__ for iface in required_interfaces) ) ) def check_backend_support(item): supported = item.keywords.get("supported") if supported and "backend" in item.funcargs: for mark in supported: if not mark.kwargs["only_if"](item.funcargs["backend"]): pytest.skip("{0} ({1})".format( mark.kwargs["skip_message"], item.funcargs["backend"] )) elif supported: raise ValueError("This mark is only available on methods that take a " "backend") @contextmanager def raises_unsupported_algorithm(reason): with pytest.raises(UnsupportedAlgorithm) as exc_info: yield exc_info assert exc_info.value._reason is reason def load_vectors_from_file(filename, loader, mode="r"): with cryptography_vectors.open_vector_file(filename, mode) as vector_file: return loader(vector_file) def load_nist_vectors(vector_data): test_data = None data = [] for line in vector_data: line = line.strip() # Blank lines, comments, and section headers are ignored if not line or line.startswith("#") or (line.startswith("[") and line.endswith("]")): continue if line.strip() == "FAIL": test_data["fail"] = True continue # Build our data using a simple Key = Value format name, value = [c.strip() for c in line.split("=")] # Some tests (PBKDF2) contain \0, which should be interpreted as a # null character rather than literal. value = value.replace("\\0", "\0") # COUNT is a special token that indicates a new block of data if name.upper() == "COUNT": test_data = {} data.append(test_data) continue # For all other tokens we simply want the name, value stored in # the dictionary else: test_data[name.lower()] = value.encode("ascii") return data def load_cryptrec_vectors(vector_data): cryptrec_list = [] for line in vector_data: line = line.strip() # Blank lines and comments are ignored if not line or line.startswith("#"): continue if line.startswith("K"): key = line.split(" : ")[1].replace(" ", "").encode("ascii") elif line.startswith("P"): pt = line.split(" : ")[1].replace(" ", "").encode("ascii") elif line.startswith("C"): ct = line.split(" : ")[1].replace(" ", "").encode("ascii") # after a C is found the K+P+C tuple is complete # there are many P+C pairs for each K cryptrec_list.append({ "key": key, "plaintext": pt, "ciphertext": ct }) else: raise ValueError("Invalid line in file '{}'".format(line)) return cryptrec_list def load_hash_vectors(vector_data): vectors = [] key = None msg = None md = None for line in vector_data: line = line.strip() if not line or line.startswith("#") or line.startswith("["): continue if line.startswith("Len"): length = int(line.split(" = ")[1]) elif line.startswith("Key"): # HMAC vectors contain a key attribute. Hash vectors do not. key = line.split(" = ")[1].encode("ascii") elif line.startswith("Msg"): # In the NIST vectors they have chosen to represent an empty # string as hex 00, which is of course not actually an empty # string. So we parse the provided length and catch this edge case. msg = line.split(" = ")[1].encode("ascii") if length > 0 else b"" elif line.startswith("MD"): md = line.split(" = ")[1] # after MD is found the Msg+MD (+ potential key) tuple is complete if key is not None: vectors.append(KeyedHashVector(msg, md, key)) key = None msg = None md = None else: vectors.append(HashVector(msg, md)) msg = None md = None else: raise ValueError("Unknown line in hash vector") return vectors def load_pkcs1_vectors(vector_data): """ Loads data out of RSA PKCS #1 vector files. """ private_key_vector = None public_key_vector = None attr = None key = None example_vector = None examples = [] vectors = [] for line in vector_data: if ( line
# This file is dual licensed under the terms of the Apache License, Version
# 2.0, and the BSD License. See the LICENSE file in the root of this repository
# for complete details.

from __future__ import absolute_import, division, print_function

import os
import sys

from _cffi_src.utils import (
    build_ffi_for_binding, compiler_type, extra_link_args
)


def _get_openssl_libraries(platform):
    if os.environ.get("CRYPTOGRAPHY_SUPPRESS_LINK_FLAGS", None):
        return []
    # OpenSSL goes by a different library name on different operating systems.
    if platform == "win32" and compiler_type() == "msvc":
        windows_link_legacy_openssl = os.environ.get(
            "CRYPTOGRAPHY_WINDOWS_LINK_LEGACY_OPENSSL", None
        )
        if windows_link_legacy_openssl is None:
            # Link against the 1.1.0 names
            libs = ["libssl", "libcrypto"]
        else:
            # Link against the 1.0.2 and lower names
            libs = ["libeay32", "ssleay32"]
        return libs + ["advapi32", "crypt32", "gdi32", "user32", "ws2_32"]
    else:
        # darwin, linux, mingw all use this path
        # In some circumstances, the order in which these libs are
        # specified on the linker command-line is significant;
        # libssl must come before libcrypto
        # (http://marc.info/?l=openssl-users&m=135361825921871)
        return ["ssl", "crypto"]


ffi = build_ffi_for_binding(
    module_name="_openssl",
    module_prefix="_cffi_src.openssl.",
    modules=[
        # This goes first so we can define some cryptography-wide symbols.
        "cryptography",

        "aes",
        "asn1",
        "bignum",
        "bio",
        "cmac",
        "cms",
        "conf",
        "crypto",
        "ct",
        "dh",
        "dsa",
        "ec",
        "ecdh",
        "ecdsa",
        "engine",
        "err",
        "evp",
        "hmac",
        "nid",
        "objects",
        "ocsp",
        "opensslv",
        "osrandom_engine",
        "pem",
        "pkcs12",
        "rand",
        "rsa",
        "ssl",
        "x509",
        "x509name",
        "x509v3",
        "x509_vfy",
        "pkcs7",
        "callbacks",
    ],
    libraries=_get_openssl_libraries(sys.platform),
    extra_link_args=extra_link_args(compiler_type()),
)
-") vectors = [] data = { "fail_z": False, "fail_agree": False } for line in vector_data: line = line.strip() if not line or line.startswith("#"): continue if line.startswith("P = "): data["p"] = int(line.split("=")[1], 16) elif line.startswith("Q = "): data["q"] = int(line.split("=")[1], 16) elif line.startswith("G = "): data["g"] = int(line.split("=")[1], 16) elif line.startswith("Z = "): z_hex = line.split("=")[1].strip().encode("ascii") data["z"] = binascii.unhexlify(z_hex) elif line.startswith("XstatCAVS = "): data["x1"] = int(line.split("=")[1], 16) elif line.startswith("YstatCAVS = "): data["y1"] = int(line.split("=")[1], 16) elif line.startswith("XstatIUT = "): data["x2"] = int(line.split("=")[1], 16) elif line.startswith("YstatIUT = "): data["y2"] = int(line.split("=")[1], 16) elif line.startswith("Result = "): result_str = line.split("=")[1].strip() match = result_rx.match(result_str) if match.group(1) == "F": if int(match.group(2)) in (5, 10): data["fail_z"] = True else: data["fail_agree"] = True vectors.append(data) data = { "p": data["p"], "q": data["q"], "g": data["g"], "fail_z": False, "fail_agree": False } return vectors def load_kasvs_ecdh_vectors(vector_data): """ Loads data out of the KASVS key exchange vector data """ curve_name_map = { "P-192": "secp192r1", "P-224": "secp224r1", "P-256": "secp256r1", "P-384": "secp384r1", "P-521": "secp521r1", } result_rx = re.compile(r"([FP]) \(([0-9]+) -") tags = [] sets = {} vectors = [] # find info in header for line in vector_data: line = line.strip() if line.startswith("#"): parm = line.split("Parameter set(s) supported:") if len(parm) == 2: names = parm[1].strip().split() for n in names: tags.append("[%s]" % n) break # Sets Metadata tag = None curve = None for line in vector_data: line = line.strip() if not line or line.startswith("#"): continue if line in tags: tag = line curve = None elif line.startswith("[Curve selected:"): curve = curve_name_map[line.split(':')[1].strip()[:-1]] if tag is not None and curve is not None: sets[tag.strip("[]")] = curve tag = None if len(tags) == len(sets): break # Data data = { "CAVS": {}, "IUT": {}, } tag = None for line in vector_data: line = line.strip() if not line or line.startswith("#"): continue if line.startswith("["): tag = line.split()[0][1:] elif line.startswith("COUNT = "): data["COUNT"] = int(line.split("=")[1]) elif line.startswith("dsCAVS = "): data["CAVS"]["d"] = int(line.split("=")[1], 16) elif line.startswith("QsCAVSx = "): data["CAVS"]["x"] = int(line.split("=")[1], 16) elif line.startswith("QsCAVSy = "): data["CAVS"]["y"] = int(line.split("=")[1], 16) elif line.startswith("dsIUT = "): data["IUT"]["d"] = int(line.split("=")[1], 16) elif line.startswith("QsIUTx = "): data["IUT"]["x"] = int(line.split("=")[1], 16) elif line.startswith("QsIUTy = "): data["IUT"]["y"] = int(line.split("=")[1], 16) elif line.startswith("OI = "): data["OI"] = int(line.split("=")[1], 16) elif line.startswith("Z = "): data["Z"] = int(line.split("=")[1], 16) elif line.startswith("DKM = "): data["DKM"] = int(line.split("=")[1], 16) elif line.startswith("Result = "): result_str = line.split("=")[1].strip() match = result_rx.match(result_str) if match.group(1) == "F": data["fail"] = True else: data["fail"] = False data["errno"] = int(match.group(2)) data["curve"] = sets[tag] vectors.append(data) data = { "CAVS": {}, "IUT": {}, } return vectors def load_x963_vectors(vector_data): """ Loads data out of the X9.63 vector data """ vectors = [] # Sets Metadata hashname = None vector = {} for line in vector_data: line = line.strip() if line.startswith("[SHA"): hashname = line[1:-1] shared_secret_len = 0 shared_info_len = 0 key_data_len = 0 elif line.startswith("[shared secret length"): shared_secret_len = int(line[1:-1].split("=")[1].strip()) elif line.startswith("[SharedInfo length"): shared_info_len = int(line[1:-1].split("=")[1].strip()) elif line.startswith("[key data length"): key_data_len = int(line[1:-1].split("=")[1].strip()) elif line.startswith("COUNT"): count = int(line.split("=")[1].strip()) vector["hash"] = hashname vector["count"] = count vector["shared_secret_length"] = shared_secret_len vector["sharedinfo_length"] = shared_info_len vector["key_data_length"] = key_data_len elif line.startswith("Z"): vector["Z"] = line.split("=")[1].strip() assert math.ceil(shared_secret_len / 8) * 2 == len(vector["Z"]) elif line.startswith("SharedInfo"): if shared_info_len != 0: vector["sharedinfo"] = line.split("=")[1].strip() silen = len(vector["sharedinfo"]) assert math.ceil(shared_info_len / 8) * 2 == silen elif line.startswith("key_data"): vector["key_data"] = line.split("=")[1].strip() assert math.ceil(key_data_len / 8) * 2 == len(vector["key_data"]) vectors.append(vector) vector = {} return vectors