rpki/deploy/docker-installer/custom-tal-patch/tools/generate_custom_fixture.py

#!/usr/bin/env python3
"""Generate a small self-contained RPKI repository for installer tests.

The generator uses Python cryptography and the OpenSSL CLI for CMS signing.
It does not invoke Barry, Rapport, or a public RIR service.
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import ipaddress
import shutil
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat, PublicFormat
from cryptography.x509 import AccessDescription, AuthorityInformationAccess, DistributionPoint, UniformResourceIdentifier
from cryptography.x509.oid import AuthorityInformationAccessOID, NameOID, ObjectIdentifier


RRDP_HOST = "host.docker.internal"
RRDP_PORT = 18443
RSYNC_PORT = 1873
TAL_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/ta.cer"
TAL_HTTPS_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/ta/custom-ta.cer"
RRDP_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/rrdp/notification.xml"
ROOT_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/root/"
ROOT_MFT_URI = f"{ROOT_REPO_URI}root.mft"
CHILD_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/child/"
CHILD_MFT_URI = f"{CHILD_REPO_URI}child.mft"
OID_SIA = ObjectIdentifier("1.3.6.1.5.5.7.1.11")
OID_RESOURCES_IP = ObjectIdentifier("1.3.6.1.5.5.7.1.7")
OID_RESOURCES_AS = ObjectIdentifier("1.3.6.1.5.5.7.1.8")
OID_POLICY_IP_AS = ObjectIdentifier("1.3.6.1.5.5.7.14.2")
OID_ROA_ECONTENT = "1.2.840.113549.1.9.16.1.24"
OID_MANIFEST_ECONTENT = "1.2.840.113549.1.9.16.1.26"
OID_SHA256 = "2.16.840.1.101.3.4.2.1"
SIA_CA_REPOSITORY = "1.3.6.1.5.5.7.48.5"
SIA_RPKI_MANIFEST = "1.3.6.1.5.5.7.48.10"
SIA_SIGNED_OBJECT = "1.3.6.1.5.5.7.48.11"
SIA_RPKI_NOTIFY = "1.3.6.1.5.5.7.48.13"
UTC = timezone.utc
VALID_FROM = datetime(2026, 1, 1, tzinfo=UTC)
VALID_TO = datetime(2027, 12, 31, 23, 59, 59, tzinfo=UTC)
CHILD_PREFIX = ipaddress.ip_network("203.0.113.0/24")
SECOND_PREFIX = ipaddress.ip_network("203.0.113.128/25")
OUTSIDE_PREFIX = ipaddress.ip_network("203.0.114.0/24")
TEST_ASN = 64496
SESSION_ID = "11111111-2222-4333-8444-555555555555"


def tlv(tag: int, value: bytes) -> bytes:
    length = len(value)
    if length < 128:
        encoded_length = bytes([length])
    else:
        raw = length.to_bytes((length.bit_length() + 7) // 8, "big")
        encoded_length = bytes([0x80 | len(raw)]) + raw
    return bytes([tag]) + encoded_length + value


def seq(*values: bytes) -> bytes:
    return tlv(0x30, b"".join(values))


def integer(value: int) -> bytes:
    raw = value.to_bytes(max(1, (value.bit_length() + 7) // 8), "big")
    if raw[0] & 0x80:
        raw = b"\x00" + raw
    return tlv(0x02, raw)


def oid(oid_value: str) -> bytes:
    parts = [int(part) for part in oid_value.split(".")]
    encoded = bytearray([40 * parts[0] + parts[1]])
    for part in parts[2:]:
        chunks = [part & 0x7F]
        part >>= 7
        while part:
            chunks.append(0x80 | (part & 0x7F))
            part >>= 7
        encoded.extend(reversed(chunks))
    return tlv(0x06, bytes(encoded))


def octet(value: bytes) -> bytes:
    return tlv(0x04, value)


def uri(value: str) -> bytes:
    return tlv(0x86, value.encode("ascii"))


def bit_string(value: bytes, unused: int = 0) -> bytes:
    return tlv(0x03, bytes([unused]) + value)


def generalized_time(value: datetime) -> bytes:
    return tlv(0x18, value.astimezone(UTC).strftime("%Y%m%d%H%M%SZ").encode("ascii"))


def context_zero(value: bytes) -> bytes:
    return tlv(0xA0, value)


def ip_choice(networks: list[ipaddress._BaseNetwork]) -> bytes:
    entries = []
    for network in networks:
        length = (network.prefixlen + 7) // 8
        raw = network.network_address.packed[:length]
        entries.append(bit_string(raw, length * 8 - network.prefixlen))
    return seq(*entries)


def ip_resources(networks: list[ipaddress._BaseNetwork]) -> bytes:
    families: dict[int, list[ipaddress._BaseNetwork]] = {1: [], 2: []}
    for network in networks:
        families[1 if network.version == 4 else 2].append(network)
    return seq(
        *(
            seq(octet(afi.to_bytes(2, "big")), ip_choice(items))
            for afi, items in families.items()
            if items
        )
    )


def ip_inherit() -> bytes:
    return seq(
        seq(octet(b"\x00\x01"), tlv(0x05, b"")),
        seq(octet(b"\x00\x02"), tlv(0x05, b"")),
    )


def as_resources(asns: list[int] | None) -> bytes:
    if asns is None:
        return seq(context_zero(tlv(0x05, b"")))
    return seq(context_zero(seq(*(integer(asn) for asn in asns))))


def sia(entries: list[tuple[str, str]]) -> bytes:
    return seq(*(seq(oid(method), uri(location)) for method, location in entries))


def write(path: Path, data: bytes, mode: int | None = None) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)
    if mode is not None:
        path.chmod(mode)


def pem_cert(cert: x509.Certificate) -> bytes:
    return cert.public_bytes(Encoding.PEM)


def pem_key(key: rsa.RSAPrivateKey) -> bytes:
    return key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption())


def write_key(path: Path, key: rsa.RSAPrivateKey) -> None:
    write(path, pem_key(key), 0o600)


def sign_cert(
    subject: x509.Name,
    issuer: x509.Name,
    public_key,
    signing_key: rsa.RSAPrivateKey,
    serial: int,
    is_ca: bool,
    ski_key: rsa.RSAPrivateKey,
    aki_key: rsa.RSAPrivateKey | None = None,
    ip_ext: bytes | None = None,
    as_ext: bytes | None = None,
    ca_repository: str | None = None,
    manifest_uri: str | None = None,
    notify_uri: str | None = None,
    signed_object_uri: str | None = None,
    crl_uri: str | None = None,
    ca_issuer_uri: str | None = None,
    not_before: datetime = VALID_FROM,
    not_after: datetime = VALID_TO,
) -> x509.Certificate:
    builder = (
        x509.CertificateBuilder()
        .subject_name(subject)
        .issuer_name(issuer)
        .public_key(public_key)
        .serial_number(serial)
        .not_valid_before(not_before)
        .not_valid_after(not_after)
    )
    if is_ca:
        builder = builder.add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
    if is_ca:
        usage = x509.KeyUsage(False, False, False, False, False, True, True, False, False)
    else:
        usage = x509.KeyUsage(True, False, False, False, False, False, False, False, False)
    builder = builder.add_extension(usage, critical=True)
    builder = builder.add_extension(x509.SubjectKeyIdentifier.from_public_key(ski_key.public_key()), critical=False)
    if aki_key is not None:
        builder = builder.add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(aki_key.public_key()), critical=False)
    builder = builder.add_extension(
        x509.CertificatePolicies([x509.PolicyInformation(OID_POLICY_IP_AS, None)]),
        critical=True,
    )
    if crl_uri is not None:
        builder = builder.add_extension(
            x509.CRLDistributionPoints(
                [DistributionPoint([UniformResourceIdentifier(crl_uri)], None, None, None)]
            ),
            critical=False,
        )
    if ca_issuer_uri is not None:
        builder = builder.add_extension(
            AuthorityInformationAccess(
                [AccessDescription(AuthorityInformationAccessOID.CA_ISSUERS, UniformResourceIdentifier(ca_issuer_uri))]
            ),
            critical=False,
        )
    sia_entries: list[tuple[str, str]] = []
    if ca_repository is not None:
        sia_entries.append((SIA_CA_REPOSITORY, ca_repository))
    if manifest_uri is not None:
        sia_entries.append((SIA_RPKI_MANIFEST, manifest_uri))
    if notify_uri is not None:
        sia_entries.append((SIA_RPKI_NOTIFY, notify_uri))
    if signed_object_uri is not None:
        sia_entries.append((SIA_SIGNED_OBJECT, signed_object_uri))
    if sia_entries:
        builder = builder.add_extension(x509.UnrecognizedExtension(OID_SIA, sia(sia_entries)), critical=False)
    if ip_ext is not None:
        builder = builder.add_extension(x509.UnrecognizedExtension(OID_RESOURCES_IP, ip_ext), critical=True)
    if as_ext is not None:
        builder = builder.add_extension(x509.UnrecognizedExtension(OID_RESOURCES_AS, as_ext), critical=True)
    return builder.sign(signing_key, hashes.SHA256())


def make_crl(subject: x509.Name, key: rsa.RSAPrivateKey, number: int) -> x509.CertificateRevocationList:
    return (
        x509.CertificateRevocationListBuilder()
        .issuer_name(subject)
        .last_update(VALID_FROM + timedelta(days=30))
        .next_update(VALID_TO - timedelta(days=30))
        .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(key.public_key()), critical=False)
        .add_extension(x509.CRLNumber(number), critical=False)
        .sign(key, hashes.SHA256())
    )


def roa_content(prefix: ipaddress._BaseNetwork, max_length: int) -> bytes:
    length = (prefix.prefixlen + 7) // 8
    raw = prefix.network_address.packed[:length]
    address = seq(bit_string(raw, length * 8 - prefix.prefixlen), integer(max_length))
    family = seq(octet((1 if prefix.version == 4 else 2).to_bytes(2, "big")), seq(address))
    return seq(integer(TEST_ASN), seq(family))


def manifest_content(number: int, files: list[tuple[str, bytes]]) -> bytes:
    entries = [
        seq(tlv(0x16, name.encode("ascii")), bit_string(hashlib.sha256(content).digest()))
        for name, content in files
    ]
    return seq(
        integer(number),
        generalized_time(VALID_FROM + timedelta(days=30)),
        generalized_time(VALID_TO - timedelta(days=30)),
        oid(OID_SHA256),
        seq(*entries),
    )


def write_cms(content: bytes, content_type: str, signer_cert: x509.Certificate, signer_key: rsa.RSAPrivateKey, output: Path, work: Path) -> None:
    content_path = work / f"{output.name}.content.der"
    signer_path = work / f"{output.name}.signer.pem"
    key_path = work / f"{output.name}.key.pem"
    write(content_path, content)
    write(signer_path, pem_cert(signer_cert))
    write_key(key_path, signer_key)
    subprocess.run(
        [
            "openssl", "cms", "-sign", "-binary", "-in", str(content_path),
            "-signer", str(signer_path), "-inkey", str(key_path),
            "-outform", "DER", "-nodetach", "-nosmimecap", "-keyid", "-md", "sha256",
            "-econtent_type", content_type, "-out", str(output),
        ],
        check=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.PIPE,
    )


def make_roa(
    name: str,
    prefix,
    max_length: int,
    not_before: datetime,
    not_after: datetime,
    child_name,
    child_key,
    common: Path,
    work: Path,
    certificate_prefix=None,
) -> None:
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    cert = sign_cert(
        x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"#142 EE {name}")]),
        child_name,
        key.public_key(),
        child_key,
        x509.random_serial_number(),
        False,
        key,
        child_key,
        ip_ext=ip_resources([certificate_prefix or prefix]),
        signed_object_uri=f"{CHILD_REPO_URI}{name}",
        crl_uri=f"{CHILD_REPO_URI}child.crl",
        ca_issuer_uri=f"{CHILD_REPO_URI}child.cer",
        not_before=not_before,
        not_after=not_after,
    )
    write_cms(roa_content(prefix, max_length), OID_ROA_ECONTENT, cert, key, common / name, work)


def make_manifest(name: str, number: int, files: list[tuple[str, bytes]], child_name, child_key, common: Path, work: Path) -> bytes:
    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    cert = sign_cert(
        x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, f"#142 Manifest EE {name}")]),
        child_name,
        key.public_key(),
        child_key,
        x509.random_serial_number(),
        False,
        key,
        child_key,
        ip_ext=ip_inherit(),
        as_ext=as_resources(None),
        signed_object_uri=CHILD_MFT_URI if name != "root" else ROOT_MFT_URI,
        crl_uri=f"{CHILD_REPO_URI}child.crl" if name != "root" else f"{ROOT_REPO_URI}root.crl",
        ca_issuer_uri=f"{CHILD_REPO_URI}child.cer" if name != "root" else f"{ROOT_REPO_URI}ta.cer",
    )
    output = common / f"{name}.mft"
    write_cms(manifest_content(number, files), OID_MANIFEST_ECONTENT, cert, key, output, work)
    return output.read_bytes()


def publish(uri: str, path: Path) -> str:
    encoded = base64.b64encode(path.read_bytes()).decode("ascii")
    return f'  <publish uri="{uri}">{encoded}</publish>\n'


def snapshot(case_root: Path, names: list[str], serial: int, extra_root_names: list[str] | None = None) -> bytes:
    repository = case_root / "repository"
    objects = [
        (f"{ROOT_REPO_URI}root.crl", repository / "root/root.crl"),
        (f"{ROOT_REPO_URI}child.cer", repository / "root/child.cer"),
        *[
            (f"{ROOT_REPO_URI}{name}", repository / f"root/{name}")
            for name in extra_root_names or []
        ],
        (f"{CHILD_REPO_URI}child.crl", repository / "child/child.crl"),
        *[(f"{CHILD_REPO_URI}{name}", repository / f"child/{name}") for name in names],
        (ROOT_MFT_URI, repository / "root/root.mft"),
        (CHILD_MFT_URI, repository / "child/child.mft"),
    ]
    body = "".join(publish(uri, path) for uri, path in objects)
    return (
        f'<snapshot xmlns="http://www.ripe.net/rpki/rrdp" session_id="{SESSION_ID}" serial="{serial}">\n'
        f"{body}</snapshot>\n"
    ).encode()


def notification(serial: int, snap: bytes, delta: bytes | None) -> bytes:
    lines = [
        f'<notification xmlns="http://www.ripe.net/rpki/rrdp" version="1" session_id="{SESSION_ID}" serial="{serial}">',
        f'  <snapshot uri="https://{RRDP_HOST}:{RRDP_PORT}/rrdp/snapshot.xml" hash="{hashlib.sha256(snap).hexdigest()}"/>',
    ]
    if delta is not None:
        lines.append(
            f'  <delta serial="{serial}" uri="https://{RRDP_HOST}:{RRDP_PORT}/rrdp/delta-{serial}.xml" hash="{hashlib.sha256(delta).hexdigest()}"/>'
        )
    return ("\n".join(lines) + "\n</notification>\n").encode()


def corrupt_first_hash(xml: bytes) -> bytes:
    marker = b'hash="'
    start = xml.index(marker) + len(marker)
    end = xml.index(b'"', start)
    value = bytearray(xml[start:end])
    value[0] = ord("0") if value[0] != ord("0") else ord("1")
    return xml[:start] + bytes(value) + xml[end:]


def delta(add_name: str, add_path: Path, new_mft: Path) -> bytes:
    body = publish(CHILD_MFT_URI, new_mft)
    body += publish(f"{CHILD_REPO_URI}{add_name}", add_path)
    return f'<delta xmlns="http://www.ripe.net/rpki/rrdp" session_id="{SESSION_ID}" serial="2">\n{body}</delta>\n'.encode()


def build_case(
    output: Path,
    common: Path,
    tal: bytes,
    ta: bytes,
    name: str,
    number: int,
    names: list[str],
    child_mft: bytes,
    change: bytes | None = None,
    root_mft: bytes | None = None,
    extra_root_objects: list[tuple[str, bytes]] | None = None,
    notification_hash_mismatch: bool = False,
) -> None:
    case_root = output / "cases" / name
    repo = case_root / "repository"
    (repo / "root").mkdir(parents=True, exist_ok=True)
    (repo / "child").mkdir(parents=True, exist_ok=True)
    shutil.copy2(common / "ta.cer", repo / "ta.cer")
    for item in ("root.crl", "child.cer"):
        shutil.copy2(common / item, repo / "root" / item)
    (repo / "root/root.mft").write_bytes(root_mft if root_mft is not None else (common / "root.mft").read_bytes())
    for item, content in extra_root_objects or []:
        (repo / "root" / item).write_bytes(content)
    for item in (
        "child.crl", "valid.roa", "valid2.roa", "expired.roa", "max-length.roa",
        "max-length-invalid.roa", "roa-prefix-outside.roa", "out-of-resource.roa", "bad-format.roa",
    ):
        if (common / item).exists():
            shutil.copy2(common / item, repo / "child" / item)
    (repo / "child/child.mft").write_bytes(child_mft)
    snap = snapshot(case_root, names, number, [item for item, _ in extra_root_objects or []])
    (case_root / "http/rrdp").mkdir(parents=True, exist_ok=True)
    (case_root / "http/tal").mkdir(parents=True, exist_ok=True)
    (case_root / "http/ta").mkdir(parents=True, exist_ok=True)
    (case_root / "http/rrdp/snapshot.xml").write_bytes(snap)
    notification_xml = notification(number, snap, change)
    if notification_hash_mismatch:
        notification_xml = corrupt_first_hash(notification_xml)
    (case_root / "http/rrdp/notification.xml").write_bytes(notification_xml)
    if change is not None:
        (case_root / "http/rrdp/delta-2.xml").write_bytes(change)
    (case_root / "http/tal/custom.tal").write_bytes(tal)
    (case_root / "http/ta/custom-ta.cer").write_bytes(ta)
    (case_root / "CASE-MANIFEST.txt").write_text(
        f"case={name}\nserial={number}\nobjects={','.join(names)}\nsnapshot_sha256={hashlib.sha256(snap).hexdigest()}\n",
        encoding="utf-8",
    )


def make_https_certificates(output: Path) -> None:
    ca_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    ca_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 fixed RRDP test CA")])
    ca_cert = (
        x509.CertificateBuilder()
        .subject_name(ca_name).issuer_name(ca_name).public_key(ca_key.public_key())
        .serial_number(x509.random_serial_number()).not_valid_before(VALID_FROM).not_valid_after(VALID_TO)
        .add_extension(x509.BasicConstraints(ca=True, path_length=1), critical=True)
        .add_extension(x509.KeyUsage(False, False, False, False, False, True, True, False, False), critical=True)
        .sign(ca_key, hashes.SHA256())
    )
    server_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    server_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, RRDP_HOST)])
    try:
        server_host = x509.IPAddress(ipaddress.ip_address(RRDP_HOST))
    except ValueError:
        server_host = x509.DNSName(RRDP_HOST)
    san_entries = [server_host]
    if RRDP_HOST != "localhost":
        san_entries.append(x509.DNSName("localhost"))
    server_cert = (
        x509.CertificateBuilder()
        .subject_name(server_name).issuer_name(ca_cert.subject).public_key(server_key.public_key())
        .serial_number(x509.random_serial_number()).not_valid_before(VALID_FROM).not_valid_after(VALID_TO)
        .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
        .add_extension(x509.KeyUsage(True, False, False, False, False, False, False, False, False), critical=True)
        .add_extension(x509.SubjectAlternativeName(san_entries), critical=False)
        .sign(ca_key, hashes.SHA256())
    )
    write(output / "certs/rrdp-ca.pem", pem_cert(ca_cert))
    write_key(output / "certs/rrdp-ca.key", ca_key)
    write(output / "certs/rrdp-server.pem", pem_cert(server_cert))
    write_key(output / "certs/rrdp-server.key", server_key)


def main() -> None:
    global RRDP_HOST, TAL_URI, TAL_HTTPS_URI, RRDP_URI
    global ROOT_REPO_URI, ROOT_MFT_URI, CHILD_REPO_URI, CHILD_MFT_URI
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument(
        "--host",
        default=RRDP_HOST,
        help="host embedded in TAL, certificate SIA and RRDP data (default: host.docker.internal)",
    )
    args = parser.parse_args()
    RRDP_HOST = args.host
    TAL_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/ta.cer"
    TAL_HTTPS_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/ta/custom-ta.cer"
    RRDP_URI = f"https://{RRDP_HOST}:{RRDP_PORT}/rrdp/notification.xml"
    ROOT_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/root/"
    ROOT_MFT_URI = f"{ROOT_REPO_URI}root.mft"
    CHILD_REPO_URI = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/child/"
    CHILD_MFT_URI = f"{CHILD_REPO_URI}child.mft"
    output = args.output.resolve()
    if output.exists():
        shutil.rmtree(output)
    for directory in ("tal", "ta", "certs", "services", "keys"):
        (output / directory).mkdir(parents=True)

    root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    child_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    root_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 Custom RPKI Root")])
    child_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 Custom RPKI Child")])
    root_cert = sign_cert(
        root_name, root_name, root_key.public_key(), root_key, 0x14200001, True, root_key,
        ip_ext=ip_resources([CHILD_PREFIX]),
        as_ext=as_resources([TEST_ASN]), ca_repository=ROOT_REPO_URI, manifest_uri=ROOT_MFT_URI,
        notify_uri=RRDP_URI,
    )
    child_cert = sign_cert(
        child_name, root_name, child_key.public_key(), root_key, 0x14200002, True, child_key, root_key,
        ip_ext=ip_resources([CHILD_PREFIX]), as_ext=as_resources([TEST_ASN]),
        ca_repository=CHILD_REPO_URI, manifest_uri=CHILD_MFT_URI, notify_uri=RRDP_URI,
        crl_uri=f"{ROOT_REPO_URI}root.crl", ca_issuer_uri=f"{ROOT_REPO_URI}child.cer",
    )
    bad_child_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "#142 Custom RPKI Overclaiming Child")])
    bad_child_repo_uri = f"rsync://{RRDP_HOST}:{RSYNC_PORT}/custom/bad-child/"
    bad_child_cert = sign_cert(
        bad_child_name, root_name, child_key.public_key(), root_key, 0x14200003, True, child_key, root_key,
        ip_ext=ip_resources([OUTSIDE_PREFIX]), as_ext=as_resources([TEST_ASN]),
        ca_repository=bad_child_repo_uri, manifest_uri=f"{bad_child_repo_uri}bad-child.mft", notify_uri=RRDP_URI,
        crl_uri=f"{ROOT_REPO_URI}root.crl", ca_issuer_uri=f"{ROOT_REPO_URI}child.cer",
    )
    root_crl = make_crl(root_name, root_key, 1)
    child_crl = make_crl(child_name, child_key, 1)
    tal = (
        f"{TAL_URI}\n{TAL_HTTPS_URI}\n\n"
        + base64.b64encode(root_key.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)).decode()
        + "\n"
    ).encode()
    write(output / "tal/custom.tal", tal)
    write(output / "ta/custom-ta.cer", root_cert.public_bytes(Encoding.DER))
    with tempfile.TemporaryDirectory(prefix="custom-rpki-cms-") as temp:
        work = Path(temp)
        common = output / ".common"
        common.mkdir()
        write(common / "ta.cer", root_cert.public_bytes(Encoding.DER))
        write(common / "child.cer", child_cert.public_bytes(Encoding.DER))
        write(common / "bad-child.cer", bad_child_cert.public_bytes(Encoding.DER))
        write(common / "root.crl", root_crl.public_bytes(Encoding.DER))
        write(common / "child.crl", child_crl.public_bytes(Encoding.DER))
        for name, prefix, max_length, before, after, certificate_prefix in (
            ("valid.roa", CHILD_PREFIX, 24, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
            ("valid2.roa", SECOND_PREFIX, 25, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
            ("expired.roa", CHILD_PREFIX, 24, VALID_FROM - timedelta(days=30), VALID_FROM - timedelta(days=1), None),
            ("max-length.roa", CHILD_PREFIX, 25, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
            ("max-length-invalid.roa", CHILD_PREFIX, 33, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
            ("roa-prefix-outside.roa", OUTSIDE_PREFIX, 24, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), CHILD_PREFIX),
            ("out-of-resource.roa", OUTSIDE_PREFIX, 24, VALID_FROM + timedelta(days=31), VALID_TO - timedelta(days=31), None),
        ):
            make_roa(
                name, prefix, max_length, before, after, child_name, child_key, common, work,
                certificate_prefix=certificate_prefix,
            )
        bad = bytearray((common / "valid.roa").read_bytes())
        bad[-1] ^= 1
        write(common / "bad-format.roa", bytes(bad))

        root_mft = make_manifest(
            "root", 1, [
                ("root.crl", (common / "root.crl").read_bytes()),
                ("child.cer", (common / "child.cer").read_bytes()),
            ], root_name, root_key, common, work,
        )
        write(common / "root.mft", root_mft)
        bad_root_mft = make_manifest(
            "root", 1, [
                ("root.crl", (common / "root.crl").read_bytes()),
                ("child.cer", (common / "child.cer").read_bytes()),
                ("bad-child.cer", (common / "bad-child.cer").read_bytes()),
            ], root_name, root_key, common, work,
        )
        write(common / "root.mft", root_mft)
        case_specs = (
            ("baseline-v1", 1, ["child.crl", "valid.roa"]),
            ("sync-hash-mismatch", 1, ["child.crl", "valid.roa"]),
            ("baseline-v2", 2, ["child.crl", "valid.roa", "valid2.roa"]),
            ("validation-expired", 1, ["child.crl", "expired.roa"]),
            ("validation-max-length", 1, ["child.crl", "max-length.roa"]),
            ("validation-max-length-invalid", 1, ["child.crl", "max-length-invalid.roa"]),
            ("validation-roa-prefix-outside", 1, ["child.crl", "roa-prefix-outside.roa"]),
            ("validation-out-of-resource", 1, ["child.crl", "out-of-resource.roa"]),
            ("validation-ca-over-resource", 1, ["child.crl", "valid.roa"]),
            ("validation-nonstandard", 1, ["child.crl", "bad-format.roa"]),
        )
        child_mfts = {}
        for name, number, names in case_specs:
            child_mfts[name] = make_manifest(
                name, number, [(item, (common / item).read_bytes()) for item in names],
                child_name, child_key, common, work,
            )
        for name, number, names in case_specs:
            change = None
            if name == "baseline-v2":
                change = delta("valid2.roa", common / "valid2.roa", common / "baseline-v2.mft")
            build_case(
                output, common, tal, root_cert.public_bytes(Encoding.DER), name, number, names,
                child_mfts[name], change,
                root_mft=bad_root_mft if name == "validation-ca-over-resource" else None,
                extra_root_objects=[("bad-child.cer", (common / "bad-child.cer").read_bytes())]
                if name == "validation-ca-over-resource" else None,
                notification_hash_mismatch=name == "sync-hash-mismatch",
            )
    write_key(output / "keys/root.key", root_key)
    write_key(output / "keys/child.key", child_key)
    make_https_certificates(output)
    shutil.rmtree(output / ".common")
    (output / "FIXTURE-MANIFEST.txt").write_text(
        "fixture_schema_version=1\nfixture_kind=custom-tal-fixed-local-rpki\n"
        f"tal_uri={TAL_URI}\nrrdp_uri={RRDP_URI}\nrsync_module=custom\n"
        f"rrdp_port={RRDP_PORT}\nrsync_port={RSYNC_PORT}\n"
        "cases=baseline-v1,sync-hash-mismatch,baseline-v2,validation-expired,validation-max-length,validation-max-length-invalid,"
        "validation-roa-prefix-outside,validation-out-of-resource,validation-ca-over-resource,validation-nonstandard\n",
        encoding="utf-8",
    )
    print(f"generated custom fixture: {output}")


if __name__ == "__main__":
    main()