#!/usr/bin/env python3
# python3-pkcs11-probe.py -- one interpreter start for every measurement
# 99-python3-pkcs11.test scores against the PURE-PYTHON half of python-pkcs11.
#
# ONE START, because a python3 start costs tens of seconds on this arm and the
# work here is milliseconds beside it. Every measurement is emitted as a
# `key=value' line on stdout and the suite greps for the key it wants.
#
# EVERY MEASUREMENT IS ISOLATED behind meas(), which emits
# `ERR:<type>:<first line>' rather than propagating -- so one layer that raises
# costs its own cases and not the ones after it.
#
# NOTHING HERE TOUCHES THE C EXTENSION, AND THAT IS DELIBERATE RATHER THAN
# INCIDENTAL. `pkcs11._pkcs11' is a wasm side module; on this arm a loaded side
# module traps at its first suspension with `hwjs-user-rt VIOLATION[lane 0]:
# set-over-set' and rc=132 (SIGILL), which kills the interpreter outright
# (linux-in-the-browser#184, BRINTOS-2046's remaining cause). A probe that
# imported it would take every measurement below down with it and report as a
# broken library rather than as one platform gap. The suite drives that import
# in its OWN python3 start, so the blast radius is one case.
#
# `import pkcs11' does NOT reach the extension: pkcs11/__init__.py:29 imports
# `._pkcs11' inside lib(), not at module scope. Everything this probe measures
# -- the constant tables, the exception hierarchy and the DER codecs in
# pkcs11.util -- is pure Python over asn1crypto, and is expected to work on this
# image with no token, no module and no dlopen.
#
# THE EXPECTED VALUES ARE THE CRYPTOKI SPECIFICATION'S, not this package's own
# answers. Each constant table is fingerprinted here and the fingerprint's
# expected value was computed on an x86-64 control after cross-checking the
# tables member by member against `extern/pkcs11t.h' -- the OASIS header shipped
# in this same sdist, which is a second source and not the Python file's
# paraphrase of itself. 306 of the 462 members carried by the six spec-derived
# enums are defined in that header and all 306 agreed; the remainder are vendor
# extensions and later-revision mechanisms the shipped header predates, so they
# rest on the control alone.
#
# The DER fixtures in python3-pkcs11-fixtures.b64 were produced by OpenSSL 3.0.13
# -- a different implementation -- and not by this package, so a codec that is
# merely self-consistent cannot pass them.

import base64
import binascii
import hashlib
import os
import sys

_n = 0


def emit(key, value):
    global _n
    _n += 1
    text = str(value).replace("\n", " ").replace("\r", " ")
    sys.stdout.write("%s=%s\n" % (key, text))


def meas(key, fn):
    try:
        emit(key, fn())
    except BaseException as exc:  # noqa: BLE001 -- an isolated measurement
        lines = str(exc).splitlines()
        emit(key, "ERR:%s:%s" % (type(exc).__name__, lines[0] if lines else ""))


def hx(b):
    return binascii.hexlify(b).decode("ascii")


def load_fixtures():
    here = os.path.dirname(os.path.abspath(__file__))
    blobs = {}
    with open(os.path.join(here, "python3-pkcs11-fixtures.b64")) as fh:
        for line in fh:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            name, b64 = line.split(None, 1)
            blobs[name] = base64.b64decode(b64)
    return blobs


FIX = {}
try:
    FIX = load_fixtures()
    emit("fixtures_error", "")
except BaseException as exc:  # noqa: BLE001
    emit("fixtures_error", "%s: %s" % (type(exc).__name__, exc))
emit("fixtures_count", len(FIX))


def F(name):
    if name not in FIX:
        raise KeyError("fixture %s did not ship" % name)
    return FIX[name]


# ------------------------------------------------------------ the dependency ---
# asn1crypto is a HARD runtime dependency of pkcs11.util and is another member of
# this same round (BRINTOS-2398). It is asked for first and on its own, so a
# missing dependency is one legible line rather than nine codec reds.
def _asn1_import():
    try:
        import asn1crypto
        return asn1crypto.__version__
    except BaseException as exc:  # noqa: BLE001
        return "ERR:%s:%s" % (type(exc).__name__, exc)


meas("asn1crypto_version", _asn1_import)


def _cached_property_import():
    try:
        import cached_property  # noqa: F401
        return ""
    except BaseException as exc:  # noqa: BLE001
        return "%s: %s" % (type(exc).__name__, exc)


meas("cached_property_import_error", _cached_property_import)

# ---------------------------------------------------------------- the import ---
# Guarded, so that an import failure is ONE reported line and every case below
# still emits an ERR: naming the same cause, rather than the probe dying with the
# reason nowhere.
pkcs11 = None
try:
    import pkcs11 as _pkcs11_mod
    pkcs11 = _pkcs11_mod
    emit("import_error", "")
except BaseException as exc:  # noqa: BLE001
    emit("import_error", "%s: %s" % (type(exc).__name__, exc))


def P():
    if pkcs11 is None:
        raise ImportError("pkcs11 did not import")
    return pkcs11


# ------------------------------------------------------------------ identity ---
meas("module_file", lambda: P().__file__)
meas("is_package", lambda: "yes" if hasattr(P(), "__path__") else "no")
meas("version_attr", lambda: __import__("importlib.metadata", fromlist=["x"])
     .version("python-pkcs11"))
# The lazy-extension property this whole probe rests on: `import pkcs11' must
# leave `pkcs11._pkcs11' OUT of sys.modules. If this ever says `yes', the module
# reaches compiled code at import time and every case here is at the mercy of
# linux-in-the-browser#184.
meas("ext_not_imported",
     lambda: "yes" if "pkcs11._pkcs11" not in sys.modules else "no")
meas("lib_is_callable", lambda: "yes" if callable(P().lib) else "no")


# -------------------------------------------------------- the constant tables ---
# A FINGERPRINT PER TABLE, plus spot literals below. The fingerprint is
# sha256 over "NAME=<decimal>" for every member in name order -- one case that
# catches any single value moving, any member added and any member dropped, which
# is what a table of 299 mechanisms needs. The spot cases are what make a red
# legible once the fingerprint has said something moved.
def fp(enum_cls):
    body = "\n".join("%s=%d" % (m.name, int(m.value))
                     for m in sorted(enum_cls, key=lambda m: m.name))
    return "%d:%s" % (len(list(enum_cls)),
                      hashlib.sha256(body.encode("ascii")).hexdigest()[:32])


def _table(mod, name):
    return getattr(__import__("pkcs11." + mod, fromlist=[name]), name)


for _mod, _name, _key in (
        ("constants", "Attribute", "attribute"),
        ("constants", "ObjectClass", "object_class"),
        ("constants", "CertificateType", "certificate_type"),
        ("constants", "UserType", "user_type"),
        ("constants", "MechanismFlag", "mechanism_flag"),
        ("constants", "SlotFlag", "slot_flag"),
        ("constants", "TokenFlag", "token_flag"),
        ("mechanisms", "Mechanism", "mechanism"),
        ("mechanisms", "KeyType", "key_type"),
        ("mechanisms", "KDF", "kdf"),
        ("mechanisms", "MGF", "mgf"),
):
    meas("fp_" + _key, (lambda m=_mod, n=_name: fp(_table(m, n))))

# SPOT LITERALS, quoted from PKCS#11 v2.40 (and from `extern/pkcs11t.h', which
# is the same numbers in C). These are the values a table regenerated from the
# wrong revision, or byte-swapped, gets wrong first.
meas("cka_class", lambda: hex(_table("constants", "Attribute").CLASS.value))
meas("cka_token", lambda: hex(_table("constants", "Attribute").TOKEN.value))
meas("cka_modulus", lambda: hex(_table("constants", "Attribute").MODULUS.value))
meas("cka_ec_params",
     lambda: hex(_table("constants", "Attribute").EC_PARAMS.value))
meas("cko_private_key",
     lambda: hex(_table("constants", "ObjectClass").PRIVATE_KEY.value))
meas("cko_certificate",
     lambda: hex(_table("constants", "ObjectClass").CERTIFICATE.value))
meas("ckk_rsa", lambda: hex(_table("mechanisms", "KeyType").RSA.value))
meas("ckk_ec", lambda: hex(_table("mechanisms", "KeyType").EC.value))
meas("ckm_rsa_pkcs", lambda: hex(_table("mechanisms", "Mechanism").RSA_PKCS.value))
meas("ckm_sha256_rsa_pkcs",
     lambda: hex(_table("mechanisms", "Mechanism").SHA256_RSA_PKCS.value))
meas("ckm_aes_cbc_pad",
     lambda: hex(_table("mechanisms", "Mechanism").AES_CBC_PAD.value))
meas("ckm_ecdsa", lambda: hex(_table("mechanisms", "Mechanism").ECDSA.value))
meas("ckm_vendor_defined",
     lambda: hex(_table("mechanisms", "Mechanism")._VENDOR_DEFINED.value))
meas("ckc_x_509",
     lambda: hex(_table("constants", "CertificateType").X_509.value))
meas("cku_user", lambda: hex(_table("constants", "UserType").USER.value))
# The FLAG tables are bit fields, so what matters is that each is a distinct
# power of two rather than an ordinal -- a member holding 3 where 4 belongs is
# invisible to a spot case and fatal to every mask.
meas("ckf_flags_are_bits", lambda: "yes" if all(
    int(m.value) and not (int(m.value) & (int(m.value) - 1))
    for m in _table("constants", "MechanismFlag")) else "no")
meas("ckf_encrypt",
     lambda: hex(_table("constants", "MechanismFlag").ENCRYPT.value))
meas("ckf_sign", lambda: hex(_table("constants", "MechanismFlag").SIGN.value))
meas("ckf_token_present",
     lambda: hex(_table("constants", "SlotFlag").TOKEN_PRESENT.value))
meas("ckf_login_required",
     lambda: hex(_table("constants", "TokenFlag").LOGIN_REQUIRED.value))


# ---------------------------------------------------------- the default table ---
# defaults.DEFAULT_KEY_CAPABILITIES is what every decode_* below consults when
# the caller passes no capabilities, so its content is asserted rather than left
# implicit in the codec cases.
def _caps(kt):
    d = __import__("pkcs11.defaults", fromlist=["DEFAULT_KEY_CAPABILITIES"])
    K = _table("mechanisms", "KeyType")
    return "|".join(sorted(f.name for f in _table("constants", "MechanismFlag")
                           if f in d.DEFAULT_KEY_CAPABILITIES[getattr(K, kt)]))


meas("caps_rsa", lambda: _caps("RSA"))
meas("caps_ec", lambda: _caps("EC"))
meas("caps_aes", lambda: _caps("AES"))


# ------------------------------------------------------- the exception tree ---
# PKCS#11 return codes are exposed as exceptions and the HIERARCHY is the part
# callers write `except' against; a class reparented to Exception would still
# import, still have the right name, and silently stop being caught.
def _exc(name):
    return getattr(__import__("pkcs11.exceptions", fromlist=[name]), name)


meas("exc_base_is_runtimeerror",
     lambda: "yes" if issubclass(_exc("PKCS11Error"), RuntimeError) else "no")
meas("exc_count", lambda: len([
    n for n in dir(__import__("pkcs11.exceptions", fromlist=["PKCS11Error"]))
    if isinstance(_exc(n), type) and issubclass(_exc(n), BaseException)]))
meas("exc_all_under_base", lambda: "yes" if all(
    issubclass(_exc(n), _exc("PKCS11Error"))
    for n in dir(__import__("pkcs11.exceptions", fromlist=["PKCS11Error"]))
    if isinstance(_exc(n), type) and issubclass(_exc(n), BaseException)
    and n != "PKCS11Error") else "no")
# Named members, so a rename is a red rather than a count that still adds up.
meas("exc_names_spot", lambda: ",".join(n for n in (
    "AlreadyInitialized", "ArgumentsBad", "DeviceError", "FunctionFailed",
    "NoSuchToken", "PinIncorrect", "SessionHandleInvalid", "TokenNotPresent",
    "UserNotLoggedIn") if hasattr(
        __import__("pkcs11.exceptions", fromlist=["PKCS11Error"]), n)))
# Re-exported at package level, which is where every documented example catches
# them from.
meas("exc_reexported", lambda: "yes" if all(
    getattr(P(), n, None) is _exc(n)
    for n in ("PKCS11Error", "AlreadyInitialized", "NoSuchToken")) else "no")
# The one this port's failure path lands on -- see the suite's `lib-*' cases.
meas("exc_already_initialized_is_pkcs11error",
     lambda: "yes" if issubclass(_exc("AlreadyInitialized"),
                                 _exc("PKCS11Error")) else "no")


# ------------------------------------------------------------ the DER codecs ---
# EVERY FIXTURE WAS PRODUCED BY OpenSSL 3.0.13 AND NOT BY THIS PACKAGE, so a
# decoder that is merely self-consistent fails them. Each key is decoded field by
# field, the fields are cross-checked against each other where two fixtures share
# a key, and the encode side is asserted BYTE-IDENTICAL against the same blob --
# which is the assertion a stub cannot pass.
def A(name):
    return getattr(_table("constants", "Attribute"), name)


def _u(mod, fn):
    return getattr(__import__("pkcs11.util." + mod, fromlist=[fn]), fn)


def _BIG(v):
    return __import__("pkcs11.util", fromlist=["biginteger"]).biginteger(v)


def _int(b):
    return int.from_bytes(b, byteorder="big")


# biginteger() is the shared primitive: PKCS#11 wants an unsigned big-endian
# byte string, and asn1crypto hands back a signed INTEGER. A leading zero byte
# left in, or a short value not padded, is the classic fault and every codec
# below inherits it.
meas("biginteger_type", lambda: type(
    _BIG(__import__("asn1crypto.core", fromlist=["Integer"])
                         .Integer(255))).__name__)
meas("biginteger_255",
     lambda: hx(_BIG(
         __import__("asn1crypto.core", fromlist=["Integer"]).Integer(255))))
meas("biginteger_high_bit",
     lambda: hx(_BIG(
         __import__("asn1crypto.core", fromlist=["Integer"]).Integer(0x80))))

# --- RSA -------------------------------------------------------------------
meas("rsa_pub_keys", lambda: ",".join(sorted(
    k.name for k in _u("rsa", "decode_rsa_public_key")(F("rsa-pub")))))
meas("rsa_pub_class", lambda: _u("rsa", "decode_rsa_public_key")(
    F("rsa-pub"))[A("CLASS")].name)
meas("rsa_pub_keytype", lambda: _u("rsa", "decode_rsa_public_key")(
    F("rsa-pub"))[A("KEY_TYPE")].name)
meas("rsa_pub_modulus_bits", lambda: len(_u("rsa", "decode_rsa_public_key")(
    F("rsa-pub"))[A("MODULUS")]) * 8)
meas("rsa_pub_exponent", lambda: _int(_u("rsa", "decode_rsa_public_key")(
    F("rsa-pub"))[A("PUBLIC_EXPONENT")]))
meas("rsa_pub_modulus_sha256", lambda: hashlib.sha256(
    _u("rsa", "decode_rsa_public_key")(F("rsa-pub"))[A("MODULUS")]
).hexdigest()[:32])
meas("rsa_pub_roundtrip", lambda: "yes" if _u("rsa", "encode_rsa_public_key")(
    _u("rsa", "decode_rsa_public_key")(F("rsa-pub"))) == F("rsa-pub") else "no")
meas("rsa_priv_keys", lambda: ",".join(sorted(
    k.name for k in _u("rsa", "decode_rsa_private_key")(F("rsa-priv")))))
# THE CROSS-CHECK: the two fixtures are one key pair, so the private key's
# modulus must be the public key's, and p*q must be that modulus. A decoder
# returning plausible-but-wrong integers cannot make both hold.
meas("rsa_priv_modulus_matches_pub", lambda: "yes" if
     _u("rsa", "decode_rsa_private_key")(F("rsa-priv"))[A("MODULUS")] ==
     _u("rsa", "decode_rsa_public_key")(F("rsa-pub"))[A("MODULUS")] else "no")
meas("rsa_priv_primes_multiply", lambda: "yes" if
     _int(_u("rsa", "decode_rsa_private_key")(F("rsa-priv"))[A("PRIME_1")]) *
     _int(_u("rsa", "decode_rsa_private_key")(F("rsa-priv"))[A("PRIME_2")]) ==
     _int(_u("rsa", "decode_rsa_private_key")(F("rsa-priv"))[A("MODULUS")])
     else "no")
# ...and the private exponent really inverts the public one, which is the whole
# of what an RSA key IS. This is arithmetic on the decoded fields, so it fails if
# any one of d, e, p or q came back wrong.
meas("rsa_priv_exponents_invert", lambda: "yes" if (
    lambda k: (_int(k[A("PRIVATE_EXPONENT")]) *
               _int(k[A("PUBLIC_EXPONENT")])) %
    ((_int(k[A("PRIME_1")]) - 1) * (_int(k[A("PRIME_2")]) - 1) //
     __import__("math").gcd(_int(k[A("PRIME_1")]) - 1,
                            _int(k[A("PRIME_2")]) - 1)) == 1)(
    _u("rsa", "decode_rsa_private_key")(F("rsa-priv"))) else "no")
meas("rsa_priv_class", lambda: _u("rsa", "decode_rsa_private_key")(
    F("rsa-priv"))[A("CLASS")].name)
# The capability flags the DEFAULT table produces, which is what an unqualified
# decode hands to create_object.
meas("rsa_pub_caps", lambda: ",".join(sorted(
    k.name for k in _u("rsa", "decode_rsa_public_key")(F("rsa-pub"))
    if _u("rsa", "decode_rsa_public_key")(F("rsa-pub"))[k] is True)))

# --- EC --------------------------------------------------------------------
meas("ec_pub_keys", lambda: ",".join(sorted(
    k.name for k in _u("ec", "decode_ec_public_key")(F("ec-pub")))))
meas("ec_pub_keytype", lambda: _u("ec", "decode_ec_public_key")(
    F("ec-pub"))[A("KEY_TYPE")].name)
# EC_PARAMS is the DER of the named-curve OID; prime256v1 is 1.2.840.10045.3.1.7.
meas("ec_pub_params", lambda: hx(_u("ec", "decode_ec_public_key")(
    F("ec-pub"))[A("EC_PARAMS")]))
# EC_POINT is the point wrapped in an OCTET STRING, per PKCS#11 -- 0x04 0x41 then
# the uncompressed point, itself starting 0x04. Both layers are asserted.
meas("ec_pub_point_len", lambda: len(_u("ec", "decode_ec_public_key")(
    F("ec-pub"))[A("EC_POINT")]))
meas("ec_pub_point_head", lambda: hx(_u("ec", "decode_ec_public_key")(
    F("ec-pub"))[A("EC_POINT")][:3]))
meas("ec_pub_roundtrip", lambda: "yes" if _u("ec", "encode_ec_public_key")(
    _u("ec", "decode_ec_public_key")(F("ec-pub"))) == F("ec-pub") else "no")
# encode_ec_point=False is the OTHER documented shape -- the raw point with no
# OCTET STRING wrapper, which some modules want.
meas("ec_pub_point_unwrapped_len", lambda: len(
    _u("ec", "decode_ec_public_key")(F("ec-pub"), encode_ec_point=False)[
        A("EC_POINT")]))
meas("ec_priv_keys", lambda: ",".join(sorted(
    k.name for k in _u("ec", "decode_ec_private_key")(F("ec-priv")))))
# THE CROSS-CHECK: private and public fixture are one key, so the curve must
# agree between them.
meas("ec_priv_params_match_pub", lambda: "yes" if
     _u("ec", "decode_ec_private_key")(F("ec-priv"))[A("EC_PARAMS")] ==
     _u("ec", "decode_ec_public_key")(F("ec-pub"))[A("EC_PARAMS")] else "no")
meas("ec_named_curve_secp256r1",
     lambda: hx(_u("ec", "encode_named_curve_parameters")("secp256r1")))
meas("ec_named_curve_matches_fixture", lambda: "yes" if
     _u("ec", "encode_named_curve_parameters")("secp256r1") ==
     _u("ec", "decode_ec_public_key")(F("ec-pub"))[A("EC_PARAMS")] else "no")
# ECDSA signatures: PKCS#11 wants r||s fixed-width, X.509 wants a DER SEQUENCE.
# The fixture is OpenSSL's DER, so the decode is measured against a foreign
# encoder and the encode is asserted byte-identical back to it.
meas("ecdsa_sig_p1363_len",
     lambda: len(_u("ec", "decode_ecdsa_signature")(F("ecdsa-sig"))))
meas("ecdsa_sig_p1363",
     lambda: hx(_u("ec", "decode_ecdsa_signature")(F("ecdsa-sig"))))
meas("ecdsa_sig_roundtrip", lambda: "yes" if
     _u("ec", "encode_ecdsa_signature")(
         _u("ec", "decode_ecdsa_signature")(F("ecdsa-sig"))) == F("ecdsa-sig")
     else "no")

# --- DSA -------------------------------------------------------------------
meas("dsa_params_keys", lambda: ",".join(sorted(
    k.name for k in _u("dsa", "decode_dsa_domain_parameters")(
        F("dsa-params")))))
meas("dsa_params_prime_bits", lambda: len(
    _u("dsa", "decode_dsa_domain_parameters")(F("dsa-params"))[
        A("PRIME")]) * 8)
meas("dsa_params_subprime_bits", lambda: len(
    _u("dsa", "decode_dsa_domain_parameters")(F("dsa-params"))[
        A("SUBPRIME")]) * 8)
meas("dsa_params_roundtrip", lambda: "yes" if
     _u("dsa", "encode_dsa_domain_parameters")(
         _u("dsa", "decode_dsa_domain_parameters")(F("dsa-params"))) ==
     F("dsa-params") else "no")
# The domain parameters are a group, so q | p-1 must hold and g must have order
# q. Both are arithmetic on the decoded integers and neither can survive a field
# swapped for another.
meas("dsa_params_q_divides_p1", lambda: "yes" if (
    lambda d: (_int(d[A("PRIME")]) - 1) % _int(d[A("SUBPRIME")]) == 0)(
    _u("dsa", "decode_dsa_domain_parameters")(F("dsa-params"))) else "no")
meas("dsa_params_g_order_q", lambda: "yes" if (
    lambda d: pow(_int(d[A("BASE")]), _int(d[A("SUBPRIME")]),
                  _int(d[A("PRIME")])) == 1)(
    _u("dsa", "decode_dsa_domain_parameters")(F("dsa-params"))) else "no")
meas("dsa_pub_len",
     lambda: len(_u("dsa", "decode_dsa_public_key")(F("dsa-pub-y"))))
meas("dsa_pub_roundtrip", lambda: "yes" if
     _u("dsa", "encode_dsa_public_key")(
         {A("VALUE"): _u("dsa", "decode_dsa_public_key")(F("dsa-pub-y"))}) ==
     F("dsa-pub-y") else "no")
# THE CROSS-CHECK: y is the public value in the group the parameters describe, so
# y^q mod p == 1. That ties two independently produced fixtures together.
meas("dsa_pub_in_group", lambda: "yes" if (
    lambda d, y: pow(_int(y), _int(d[A("SUBPRIME")]),
                     _int(d[A("PRIME")])) == 1)(
    _u("dsa", "decode_dsa_domain_parameters")(F("dsa-params")),
    _u("dsa", "decode_dsa_public_key")(F("dsa-pub-y"))) else "no")
meas("dsa_sig_p1363_len",
     lambda: len(_u("dsa", "decode_dsa_signature")(F("dsa-sig"))))
meas("dsa_sig_p1363",
     lambda: hx(_u("dsa", "decode_dsa_signature")(F("dsa-sig"))))
meas("dsa_sig_roundtrip", lambda: "yes" if
     _u("dsa", "encode_dsa_signature")(
         _u("dsa", "decode_dsa_signature")(F("dsa-sig"))) == F("dsa-sig")
     else "no")

# --- DH --------------------------------------------------------------------
meas("dh_params_keys", lambda: ",".join(sorted(
    k.name for k in _u("dh", "decode_dh_domain_parameters")(F("dh-params")))))
meas("dh_params_prime_bits", lambda: len(
    _u("dh", "decode_dh_domain_parameters")(F("dh-params"))[A("PRIME")]) * 8)
meas("dh_params_base", lambda: _int(
    _u("dh", "decode_dh_domain_parameters")(F("dh-params"))[A("BASE")]))
meas("dh_params_roundtrip", lambda: "yes" if
     _u("dh", "encode_dh_domain_parameters")(
         _u("dh", "decode_dh_domain_parameters")(F("dh-params"))) ==
     F("dh-params") else "no")
meas("dh_pub_len",
     lambda: len(_u("dh", "decode_dh_public_key")(F("dh-pub-y"))))
meas("dh_pub_roundtrip", lambda: "yes" if
     _u("dh", "encode_dh_public_key")(
         {A("VALUE"): _u("dh", "decode_dh_public_key")(F("dh-pub-y"))}) ==
     F("dh-pub-y") else "no")
# y < p is the one structural property a DH public value must have, and it is
# what a decode that lost or gained a byte breaks.
meas("dh_pub_below_prime", lambda: "yes" if
     _int(_u("dh", "decode_dh_public_key")(F("dh-pub-y"))) <
     _int(_u("dh", "decode_dh_domain_parameters")(F("dh-params"))[A("PRIME")])
     else "no")

# --- X.509 -----------------------------------------------------------------
meas("x509_keys", lambda: ",".join(sorted(
    k.name for k in _u("x509", "decode_x509_certificate")(F("cert")))))
meas("x509_class", lambda: _u("x509", "decode_x509_certificate")(
    F("cert"))[A("CLASS")].name)
meas("x509_cert_type", lambda: _u("x509", "decode_x509_certificate")(
    F("cert"))[A("CERTIFICATE_TYPE")].name)
# VALUE must be the certificate BYTE FOR BYTE -- PKCS#11 stores the DER, not a
# re-encoding of it.
meas("x509_value_is_der", lambda: "yes" if
     _u("x509", "decode_x509_certificate")(F("cert"))[A("VALUE")] == F("cert")
     else "no")
# SUBJECT and ISSUER are the DER of the Name, and this fixture is self-signed, so
# the two must be identical -- a cross-check no single field can fake.
meas("x509_subject_is_issuer", lambda: "yes" if
     _u("x509", "decode_x509_certificate")(F("cert"))[A("SUBJECT")] ==
     _u("x509", "decode_x509_certificate")(F("cert"))[A("ISSUER")] else "no")
meas("x509_subject_sha256", lambda: hashlib.sha256(
    _u("x509", "decode_x509_certificate")(F("cert"))[A("SUBJECT")]
).hexdigest()[:32])
meas("x509_serial_len", lambda: len(
    _u("x509", "decode_x509_certificate")(F("cert"))[A("SERIAL_NUMBER")]))
meas("x509_extended_keys", lambda: ",".join(sorted(
    k.name for k in _u("x509", "decode_x509_certificate")(
        F("cert"), extended_set=True))))
meas("x509_pub_keys", lambda: ",".join(sorted(
    k.name for k in _u("x509", "decode_x509_public_key")(F("cert")))))
# THE CROSS-CHECK THAT TIES THE WHOLE FIXTURE SET TOGETHER: the certificate was
# signed with the RSA key above, so the public key extracted from the
# certificate must be the RSA fixture's, modulus for modulus. Two codecs, two
# blobs, one answer.
meas("x509_pub_modulus_matches_rsa", lambda: "yes" if
     _u("x509", "decode_x509_public_key")(F("cert"))[A("MODULUS")] ==
     _u("rsa", "decode_rsa_public_key")(F("rsa-pub"))[A("MODULUS")] else "no")

# --------------------------------------------------------------- error paths ---
# A codec handed the wrong blob must RAISE rather than return a plausible key.
def _raises(fn):
    try:
        fn()
        return "no-raise"
    except BaseException as exc:  # noqa: BLE001
        return type(exc).__name__


meas("rsa_pub_rejects_cert",
     lambda: _raises(lambda: _u("rsa", "decode_rsa_public_key")(F("cert"))))
meas("rsa_pub_rejects_empty",
     lambda: _raises(lambda: _u("rsa", "decode_rsa_public_key")(b"")))
meas("rsa_pub_rejects_truncated",
     lambda: _raises(lambda: _u("rsa", "decode_rsa_public_key")(
         F("rsa-pub")[:20])))
meas("x509_rejects_rsa_key",
     lambda: _raises(lambda: _u("x509", "decode_x509_certificate")(
         F("rsa-pub"))))
meas("ec_named_curve_rejects_unknown",
     lambda: _raises(lambda: _u("ec", "encode_named_curve_parameters")(
         "no-such-curve-here")))

emit("emitted_count", _n + 1)
emit("probe_done", "yes")