chg: test: Unify system test key and algorithm handling

This is a followup from https://gitlab.isc.org/isc-projects/bind9/-/merge_requests/11807.

De-duplicates the key material handling and moves algorithm-related things into a dedicated shared module.

Merge branch 'nicki/pytest-key-reconcile' into 'main'

See merge request isc-projects/bind9!12255
This commit is contained in:
Nicki Křížek 2026-07-02 14:51:52 +02:00
commit cbc41caafa
17 changed files with 216 additions and 139 deletions

View file

@ -273,7 +273,7 @@ def control_port():
@pytest.fixture(scope="module")
def default_algorithm():
return isctest.vars.algorithms.Algorithm.default()
return isctest.algorithms.Algorithm.default()
@pytest.fixture(scope="module")

View file

@ -12,8 +12,8 @@
from re import compile as Re
from dnssec_py.common import DNSSEC_PY_MARK
from isctest.algorithms import Algorithm
from isctest.template import NS2, NS3, zones
from isctest.vars.algorithms import Algorithm
from isctest.zone import Zone, configure_root
import isctest

View file

@ -10,6 +10,7 @@
# information regarding copyright ownership.
from . import ( # pylint: disable=redefined-builtin
algorithms,
check,
hypothesis,
instance,
@ -29,6 +30,7 @@ from . import ( # pylint: disable=redefined-builtin
# instead.
__all__ = [
"algorithms",
"check",
"hypothesis",
"instance",

View file

@ -0,0 +1,61 @@
# Copyright (C) Internet Systems Consortium, Inc. ("ISC")
#
# SPDX-License-Identifier: MPL-2.0
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, you can obtain one at https://mozilla.org/MPL/2.0/.
#
# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
from typing import NamedTuple
import os
class Algorithm(NamedTuple):
name: str
number: int
dst: int
bits: int
@classmethod
def default(cls):
return cls(
os.environ["DEFAULT_ALGORITHM"],
int(os.environ["DEFAULT_ALGORITHM_NUMBER"]),
int(os.environ["DEFAULT_ALGORITHM_DST_NUMBER"]),
int(os.environ["DEFAULT_BITS"]),
)
RSASHA1 = Algorithm("RSASHA1", 5, 5, 2048)
NSEC3RSASHA1 = Algorithm("NSEC3RSASHA1", 7, 7, 2048)
RSASHA256 = Algorithm("RSASHA256", 8, 8, 2048)
RSASHA512 = Algorithm("RSASHA512", 10, 10, 2048)
ECDSAP256SHA256 = Algorithm("ECDSAP256SHA256", 13, 13, 256)
ECDSAP384SHA384 = Algorithm("ECDSAP384SHA384", 14, 14, 384)
ED25519 = Algorithm("ED25519", 15, 15, 256)
ED448 = Algorithm("ED448", 16, 16, 456)
RSASHA256OID = Algorithm("RSASHA256OID", 254, 256, 2048)
RSASHA512OID = Algorithm("RSASHA512OID", 254, 257, 2048)
ALL_ALGORITHMS = [
RSASHA1,
NSEC3RSASHA1,
RSASHA256,
RSASHA512,
ECDSAP256SHA256,
ECDSAP384SHA384,
ED25519,
ED448,
RSASHA256OID,
RSASHA512OID,
]
ALL_ALGORITHMS_BY_NUM = {alg.number: alg for alg in ALL_ALGORITHMS}
# Keyed by the DST identifier rather than the on-wire number: unlike `number`
# (where both private-OID variants collide at 254), `dst` is unique, so this
# map distinguishes RSASHA256OID (256) from RSASHA512OID (257).
ALL_ALGORITHMS_BY_DST = {alg.dst: alg for alg in ALL_ALGORITHMS}

View file

@ -20,22 +20,19 @@ import os
import re
import time
import dns.dnssec
import dns.exception
import dns.message
import dns.name
import dns.rcode
import dns.rdataclass
import dns.rdatatype
import dns.rrset
import dns.tsig
import dns.zone
import dns.zonefile
from isctest.algorithms import ALL_ALGORITHMS_BY_DST, ECDSAP256SHA256, Algorithm
from isctest.instance import NamedInstance
from isctest.run import EnvCmd
from isctest.template import TrustAnchor
from isctest.vars.algorithms import ALL_ALGORITHMS_BY_NUM, Algorithm
from isctest.zone import FileZoneKey
import isctest.log
import isctest.query
@ -218,7 +215,7 @@ class KeyProperties:
@staticmethod
def default(with_state=True) -> "KeyProperties":
metadata = {
"Algorithm": isctest.vars.algorithms.ECDSAP256SHA256.number,
"Algorithm": ECDSAP256SHA256.number,
"Length": 256,
"Lifetime": 0,
"KSK": "yes",
@ -380,26 +377,18 @@ class SettimeOptions:
@total_ordering
class Key:
class Key(FileZoneKey):
"""
Represent a key from a keyfile.
A FileZoneKey specialized with KASP timing and state-file operations.
This object keeps track of its origin (keydir + name), can be used to
retrieve metadata from the underlying files and supports convenience
operations for KASP tests.
Inherits the key-material accessors (dnskey, into_ta, ...) from FileZoneKey
and adds the metadata reads, signing-state derivation, and timing
convenience operations used by the KASP/rollover tests.
"""
def __init__(self, name: str, keydir: str | Path | None = None):
self.name = name
if keydir is None:
self.keydir = Path()
else:
self.keydir = Path(keydir)
self.path = str(self.keydir / name)
self.privatefile = f"{self.path}.private"
self.keyfile = f"{self.path}.key"
super().__init__(name, keydir)
self.statefile = f"{self.path}.state"
self.tag = int(self.name[-5:])
self.external = False
def get_timing(
@ -498,52 +487,9 @@ class Key:
return ksigning, zsigning
def get_dnsalg(self) -> int:
alg = int(self.get_metadata("Algorithm"))
if alg == isctest.vars.algorithms.RSASHA256OID.dst:
return isctest.vars.algorithms.RSASHA256OID.number
if alg == isctest.vars.algorithms.RSASHA512OID.dst:
return isctest.vars.algorithms.RSASHA512OID.number
return alg
def ttl(self) -> int:
with open(self.keyfile, "r", encoding="utf-8") as file:
for line in file:
if line.startswith(";"):
continue
return int(line.split()[1])
return 0
@property
def dnskey(self) -> dns.rrset.RRset:
with open(self.keyfile, "r", encoding="utf-8") as file:
rrsets = dns.zonefile.read_rrsets(
file.read(),
rdclass=None, # read rdclass from the file
default_ttl=DEFAULT_TTL, # use this TTL if not present
)
assert len(rrsets) == 1, f"{self.keyfile} has multiple RRsets"
dnskey_rr = rrsets[0]
assert len(dnskey_rr) == 1, f"{self.keyfile} has multiple RRs"
assert (
dnskey_rr.rdtype == dns.rdatatype.DNSKEY
), f"DNSKEY not found in {self.keyfile}"
return dnskey_rr
def into_ta(self, ta_type: str, dsdigest=dns.dnssec.DSDigest.SHA256) -> TrustAnchor:
dnskey = self.dnskey
if ta_type in ["static-ds", "initial-ds"]:
ds = dns.dnssec.make_ds(dnskey.name, dnskey[0], dsdigest)
parts = str(ds).split()
contents = " ".join(parts[:3]) + f' "{parts[3]}"'
elif ta_type in ["static-key", "initial-key"]:
parts = str(dnskey).split()
contents = " ".join(parts[4:7]) + f' "{"".join(parts[7:])}"'
else:
raise ValueError(f"invalid trust anchor type: {ta_type}")
return TrustAnchor(str(dnskey.name), ta_type, contents)
def is_ksk(self) -> bool:
# KASP role follows the .state KSK metadata, not the DNSKEY SEP flag:
# a CSK may be configured without SEP (see the csk-nosep test).
return self.get_metadata("KSK") == "yes"
def is_zsk(self) -> bool:
@ -558,8 +504,15 @@ class Key:
@property
def algorithm(self) -> Algorithm:
num = int(self.get_metadata("Algorithm"))
return ALL_ALGORITHMS_BY_NUM[num]
"""
Obtain the algorithm from key metadata.
The .state Algorithm field holds the DST identifier, which (unlike the
on-wire DNSKEY algorithm number) distinguishes the private-OID variants
RSASHA256OID and RSASHA512OID, both of which appear as 254 on the wire.
"""
dst = int(self.get_metadata("Algorithm"))
return ALL_ALGORITHMS_BY_DST[dst]
def dnskey_equals(self, value, cdnskey=False):
dnskey = value.split()
@ -588,7 +541,7 @@ class Key:
dsfromkey_command = [
os.environ.get("DSFROMKEY"),
"-T",
str(self.ttl()),
str(self.dnskey.ttl),
"-a",
alg,
"-C",
@ -997,7 +950,7 @@ def _check_signatures(
offline_ksk=offline_ksk, zsk_missing=zsk_missing, smooth=smooth
)
alg = key.get_dnsalg()
alg = key.algorithm.number
rtype = dns.rdatatype.to_text(covers)
expect = rf"IN RRSIG {rtype} {alg} (\d) (\d+) (\d+) (\d+) {key.tag} {signer}"
@ -1762,10 +1715,10 @@ def private_type_record(zone: str, key: Key, rrtype: int = 65534) -> str:
indicating that the signing process for this key is completed.
"""
keyid = key.tag
algorithm = key.get_dnsalg()
secalg = int(key.get_metadata("Algorithm"))
wire_alg = key.algorithm.number
dst_alg = key.algorithm.dst
if algorithm < 256:
return f"{zone}. 0 IN TYPE{rrtype} \\# 5 {secalg:02x}{keyid:04x}0000"
if dst_alg < 256:
return f"{zone}. 0 IN TYPE{rrtype} \\# 5 {wire_alg:02x}{keyid:04x}0000"
return f"{zone}. 0 IN TYPE{rrtype} \\# 7 {secalg:02x}{keyid:04x}0000{algorithm:04x}"
return f"{zone}. 0 IN TYPE{rrtype} \\# 7 {wire_alg:02x}{keyid:04x}0000{dst_alg:04x}"

View file

@ -19,6 +19,16 @@ import tempfile
import time
from .. import log
from ..algorithms import (
ALL_ALGORITHMS,
ECDSAP256SHA256,
ECDSAP384SHA384,
ED448,
ED25519,
RSASHA256,
RSASHA512,
Algorithm,
)
from .basic import BASIC_VARS
# Algorithms are selected randomly at runtime from a list of supported
@ -55,22 +65,6 @@ STABLE_PERIOD = 3600 * 3
"""number of secs during which algorithm selection remains stable"""
class Algorithm(NamedTuple):
name: str
number: int
dst: int
bits: int
@classmethod
def default(cls):
return cls(
os.environ["DEFAULT_ALGORITHM"],
int(os.environ["DEFAULT_ALGORITHM_NUMBER"]),
int(os.environ["DEFAULT_ALGORITHM_DST_NUMBER"]),
int(os.environ["DEFAULT_BITS"]),
)
class AlgorithmSet(NamedTuple):
"""Collection of DEFAULT, ALTERNATIVE and DISABLED algorithms"""
@ -86,30 +80,6 @@ class AlgorithmSet(NamedTuple):
"disable-algorithms" configuration option."""
RSASHA1 = Algorithm("RSASHA1", 5, 5, 2048)
RSASHA256 = Algorithm("RSASHA256", 8, 8, 2048)
RSASHA512 = Algorithm("RSASHA512", 10, 10, 2048)
ECDSAP256SHA256 = Algorithm("ECDSAP256SHA256", 13, 13, 256)
ECDSAP384SHA384 = Algorithm("ECDSAP384SHA384", 14, 14, 384)
ED25519 = Algorithm("ED25519", 15, 15, 256)
ED448 = Algorithm("ED448", 16, 16, 456)
RSASHA256OID = Algorithm("RSASHA256OID", 254, 256, 2048)
RSASHA512OID = Algorithm("RSASHA512OID", 254, 257, 2048)
ALL_ALGORITHMS = [
RSASHA1,
RSASHA256,
RSASHA512,
ECDSAP256SHA256,
ECDSAP384SHA384,
ED25519,
ED448,
RSASHA256OID,
RSASHA512OID,
]
ALL_ALGORITHMS_BY_NUM = {alg.number: alg for alg in ALL_ALGORITHMS}
ALGORITHM_SETS = {
"stable": AlgorithmSet(
default=ECDSAP256SHA256, alternative=RSASHA256, disabled=ECDSAP384SHA384

View file

@ -16,6 +16,7 @@ from collections.abc import Callable
from pathlib import Path
from typing import TypeAlias
import base64
import shutil
from cryptography.hazmat.primitives import serialization
@ -27,12 +28,19 @@ import dns.name
import dns.rdataclass
import dns.rdatatype
import dns.rrset
import dns.zonefile
from .kasp import Key
from .algorithms import (
ALL_ALGORITHMS_BY_NUM,
ECDSAP256SHA256,
ECDSAP384SHA384,
ED448,
ED25519,
Algorithm,
)
from .log import debug
from .run import EnvCmd
from .template import NS1, Nameserver, TemplateEngine, TrustAnchor
from .vars.algorithms import Algorithm
KEYDIR = "keys"
DNSKEY_TTL = 3600
@ -50,7 +58,7 @@ class ZoneKey(ABC):
Abstract base for a DNSSEC key attached to a Zone.
Two concrete implementations exist:
FileZoneKey wraps a dnssec-keygen-managed key file pair (kasp.Key).
FileZoneKey reads a dnssec-keygen-managed key file pair.
PythonZoneKey holds a Python-native (private_key, dnskey_rdata) pair
required for dnspython-based operations and signing.
@ -61,7 +69,16 @@ class ZoneKey(ABC):
@property
@abstractmethod
def dnskey(self) -> dns.rrset.RRset:
"""The DNSKEY RRset for this key (single-record, with TTL)."""
"""
The DNSKEY RRset for this key (single-record, with TTL).
"""
@property
@abstractmethod
def private_key(self) -> PrivateKey:
"""
The dnspython private-key object, for use with dns.dnssec signing.
"""
def is_ksk(self) -> bool:
return bool(self.dnskey[0].flags & int(Flag.SEP))
@ -107,18 +124,87 @@ class FileZoneKey(ZoneKey):
"""
A ZoneKey backed by dnssec-keygen-managed key files.
Constructed by FileZoneKey.generate(); callers normally do not
instantiate this directly. The underlying kasp.Key is accessible via
.key for working with timing metadata, state files, etc.
Reads key material directly from the .key/.private file pair. Normally
constructed via FileZoneKey.generate() (or Zone.add_keys());
isctest.kasp.Key subclasses it to add KASP timing/state operations.
"""
def __init__(self, key: Key, zone: Zone) -> None:
self.key = key
def __init__(
self,
name: str,
keydir: str | Path | None = None,
zone: Zone | None = None,
) -> None:
self.name = name
self.keydir = Path() if keydir is None else Path(keydir)
self.path = str(self.keydir / name)
self.privatefile = f"{self.path}.private"
self.keyfile = f"{self.path}.key"
self.tag = int(self.name[-5:])
self.zone = zone
@property
def dnskey(self) -> dns.rrset.RRset:
return self.key.dnskey
with open(self.keyfile, "r", encoding="utf-8") as file:
rrsets = dns.zonefile.read_rrsets(
file.read(),
rdclass=None, # read rdclass from the file
default_ttl=DNSKEY_TTL, # use this TTL if not present
)
assert len(rrsets) == 1, f"{self.keyfile} has multiple RRsets"
dnskey_rr = rrsets[0]
assert len(dnskey_rr) == 1, f"{self.keyfile} has multiple RRs"
assert (
dnskey_rr.rdtype == dns.rdatatype.DNSKEY
), f"DNSKEY not found in {self.keyfile}"
return dnskey_rr
@property
def private_key(self) -> PrivateKey:
"""
Read the .private file and build the corresponding cryptography
private-key object, dispatching on the DNSKEY algorithm.
Supports the RSA, ECDSA and EdDSA families; other algorithms raise
NotImplementedError.
"""
alg = ALL_ALGORITHMS_BY_NUM[self.dnskey[0].algorithm]
fields = {}
with open(self.privatefile, "r", encoding="utf-8") as file:
for line in file:
name, sep, value = line.partition(":")
if sep:
fields[name.strip()] = value.strip()
def field(name: str) -> bytes:
return base64.b64decode(fields[name])
if "Modulus" in fields: # RSA family, including the private-OID variants
public = rsa.RSAPublicNumbers(
int.from_bytes(field("PublicExponent"), "big"),
int.from_bytes(field("Modulus"), "big"),
)
return rsa.RSAPrivateNumbers(
p=int.from_bytes(field("Prime1"), "big"),
q=int.from_bytes(field("Prime2"), "big"),
d=int.from_bytes(field("PrivateExponent"), "big"),
dmp1=int.from_bytes(field("Exponent1"), "big"),
dmq1=int.from_bytes(field("Exponent2"), "big"),
iqmp=int.from_bytes(field("Coefficient"), "big"),
public_numbers=public,
).private_key()
secret = field("PrivateKey")
if alg == ECDSAP256SHA256:
return ec.derive_private_key(int.from_bytes(secret, "big"), ec.SECP256R1())
if alg == ECDSAP384SHA384:
return ec.derive_private_key(int.from_bytes(secret, "big"), ec.SECP384R1())
if alg == ED25519:
return ed25519.Ed25519PrivateKey.from_private_bytes(secret)
if alg == ED448:
return ed448.Ed448PrivateKey.from_private_bytes(secret)
raise NotImplementedError(f"dnskey algorithm {alg.name} not implemented")
def write_dsset(
self,
@ -134,6 +220,7 @@ class FileZoneKey(ZoneKey):
PythonZoneKey KSKs (PythonZoneKey appends to the same file);
Zone.copy_dssets enforces this.
"""
assert self.zone is not None, "write_dsset requires a zone-attached key"
src = Path(self.zone.ns.name) / f"dsset-{self.zone.name}."
shutil.copy(src, Path(target_dir))
debug(f"{self.zone.name}: dsset copied to {target_dir}")
@ -160,7 +247,7 @@ class FileZoneKey(ZoneKey):
"KEYGEN", f"-q -a {alg.number} -b {alg.bits} -K {KEYDIR} -L {DNSKEY_TTL}"
)
key_name = keygen(f"{params} {zone.name}", cwd=zone.ns.name).out.strip()
return FileZoneKey(Key(key_name, keydir=keydir), zone=zone)
return FileZoneKey(key_name, keydir=keydir, zone=zone)
class PythonZoneKey(ZoneKey):
@ -192,10 +279,14 @@ class PythonZoneKey(ZoneKey):
ttl: int = DNSKEY_TTL,
) -> None:
self.zone = zone
self.private_key = private_key
self._private_key = private_key
self._dnskey_rdata = dnskey_rdata
self.ttl = ttl
@property
def private_key(self) -> PrivateKey:
return self._private_key
@property
def dnskey(self) -> dns.rrset.RRset:
rrset = dns.rrset.RRset(

View file

@ -25,9 +25,9 @@ import dns.tsig
import dns.update
import pytest
from isctest.algorithms import ECDSAP256SHA256, ECDSAP384SHA384, Algorithm
from isctest.kasp import KeyProperties, KeyTimingMetadata, SettimeOptions
from isctest.util import param
from isctest.vars.algorithms import ECDSAP256SHA256, ECDSAP384SHA384, Algorithm
import isctest
import isctest.mark

View file

@ -18,8 +18,8 @@ import time
import pytest
from isctest.algorithms import Algorithm
from isctest.kasp import KeyTimingMetadata
from isctest.vars.algorithms import Algorithm
from rollover.common import TIMEDELTA
import isctest
@ -318,7 +318,7 @@ def check_rrsig_bundle(bundle_keys, bundle_lines, zone, rrtype, sigend, sigstart
count = 0
for key in bundle_keys:
found = False
alg = key.get_dnsalg()
alg = key.algorithm.number
expect = f"{zone}. {ttl} IN RRSIG {rrtype} {alg} 2 {ttl} {sigend} {sigstart} {key.tag} {zone}."
# there must be a signature of this ksk
for line in bundle_lines:

View file

@ -15,7 +15,7 @@ import os
import pytest
from isctest.vars.algorithms import Algorithm
from isctest.algorithms import Algorithm
import isctest

View file

@ -17,7 +17,7 @@ import dns.rdataclass
import dns.rdatatype
import pytest
from isctest.vars.algorithms import Algorithm
from isctest.algorithms import Algorithm
from nsec3.common import NSEC3_MARK, check_nsec3_case
import isctest

View file

@ -17,7 +17,7 @@ import dns.rcode
import dns.update
import pytest
from isctest.vars.algorithms import RSASHA1, Algorithm
from isctest.algorithms import RSASHA1, Algorithm
from nsec3.common import NSEC3_MARK, check_nsec3_case
import isctest

View file

@ -18,7 +18,7 @@ import dns.rdataclass
import dns.rdatatype
import pytest
from isctest.vars.algorithms import RSASHA1, Algorithm
from isctest.algorithms import RSASHA1, Algorithm
from nsec3.common import NSEC3_MARK, check_nsec3_case
import isctest

View file

@ -14,7 +14,7 @@ import os
import dns.rdatatype
import pytest
from isctest.vars.algorithms import Algorithm
from isctest.algorithms import Algorithm
from nsec3.common import NSEC3_MARK, check_nsec3_case, check_nsec3param
import isctest

View file

@ -16,7 +16,7 @@ import os
import dns.rcode
import dns.rdatatype
from isctest.vars.algorithms import RSASHA256
from isctest.algorithms import RSASHA256
from nsec3.common import NSEC3_MARK, check_auth_nsec3, check_nsec3param
import isctest

View file

@ -13,10 +13,10 @@ from pathlib import Path
import shutil
from isctest.algorithms import Algorithm
from isctest.kasp import SettimeOptions, private_type_record
from isctest.run import EnvCmd
from isctest.template import NS2, NS3, TrustAnchor, Zone
from isctest.vars.algorithms import Algorithm
import isctest

View file

@ -11,6 +11,7 @@
from datetime import timedelta
from isctest.algorithms import Algorithm
from isctest.kasp import (
Ipub,
Iret,
@ -20,7 +21,6 @@ from isctest.kasp import (
)
from isctest.run import EnvCmd
from isctest.template import NS3, Zone
from isctest.vars.algorithms import Algorithm
from rollover.setup import configure_root, configure_tld, setkeytimes
import isctest