Fix quadratic ReDoS in PEM import block scanning

Replace the backtracking _PEM_BLOCK regex with a forward-only scanner
that indexes END markers by label, so repeated unterminated headers do
not cause the regex to rescan the remaining input.

Impact: 80 KiB payload went from 8.34 s to 0.001 s.

server/packages/kca-core/src/kca/core/imports.py

@@ -3,8 +3,9 @@ from __future__ import annotations
 
 import base64
 import re
+from collections import defaultdict, deque
 from dataclasses import dataclass, field
-from typing import Any, Dict, Iterable, List, Optional, Sequence
+from typing import Any, DefaultDict, Dict, Iterable, List, Optional, Sequence
 
 from cryptography import x509
 from cryptography.hazmat.primitives.serialization import pkcs7, pkcs12
@@ -16,10 +17,8 @@ from .names import entries_from_name
 from .requests import request_info
 from .templates import template_from_xca_data
 
-_PEM_BLOCK = re.compile(
+_PEM_BEGIN = re.compile(r"-----BEGIN ([^-]+)-----")
-    r"-----BEGIN ([^-]+)-----\s*(.*?)\s*-----END \1-----",
+_PEM_END = re.compile(r"-----END ([^-]+)-----")
-    re.DOTALL,
-)
 
 
 @dataclass
@@ -161,15 +160,38 @@ def parse_pkcs7(data: bytes, filename: str = "") -> List[ParsedItem]:
     return [_cert_item(cert, filename, f"{_guess_name(cert, filename)}_{index}" if len(certs) > 1 else None) for index, cert in enumerate(certs)]
 
 
+def _iter_pem_blocks(text: str) -> Iterable[tuple[str, str]]:
+    """Yield complete PEM blocks without rescanning unmatched headers."""
+    endings: DefaultDict[str, deque[tuple[int, int]]] = defaultdict(deque)
+    for match in _PEM_END.finditer(text):
+        endings[match.group(1)].append((match.start(), match.end()))
+
+    position = 0
+    while True:
+        begin = _PEM_BEGIN.search(text, position)
+        if begin is None:
+            return
+        candidates = endings.get(begin.group(1))
+        if candidates:
+            while candidates and candidates[0][0] < begin.end():
+                candidates.popleft()
+            if candidates:
+                _, block_end = candidates.popleft()
+                yield begin.group(1), text[begin.start():block_end]
+                position = block_end
+                continue
+        position = begin.end()
+
+
 def parse_pem_bundle(data: bytes, filename: str = "", password: Optional[str] = None) -> List[ParsedItem]:
     text = data.decode("utf-8", "replace")
-    blocks = list(_PEM_BLOCK.finditer(text))
+    blocks = list(_iter_pem_blocks(text))
     if not blocks:
         return []
     items: List[ParsedItem] = []
-    for match in blocks:
+    for raw_label, text_block in blocks:
-        label = match.group(1).strip().upper()
+        label = raw_label.strip().upper()
-        block = match.group(0).encode("utf-8")
+        block = text_block.encode("utf-8")
         try:
             if label == "CERTIFICATE":
                 items.append(_cert_item(x509.load_pem_x509_certificate(block), filename))

server/packages/kca-core/tests/test_core.py

@@ -1,6 +1,7 @@
 """Core domain tests: names, extensions, keys, certificates, CRLs and exports."""
 from __future__ import annotations
 
+import time
 import unittest
 from datetime import datetime, timedelta, timezone
 
@@ -273,6 +274,30 @@ class ExportImportTest(unittest.TestCase):
         items = imports.parse_file(pem.data, "export.crt")
         self.assertEqual(items[0].kind, "certificate")
 
+    def test_pem_bundle_ignores_incomplete_blocks(self) -> None:
+        certificate = exports.export_certificate(self.cert, "pem", name="export").data
+        private_key = keys.private_pem(self.key)
+        pem = (
+            b"before\n-----BEGIN INCOMPLETE-----\nignored\n"
+            + certificate
+            + b"\n"
+            + private_key
+            + b"\nafter\n"
+        )
+        items = imports.parse_pem_bundle(pem)
+        self.assertEqual([item.kind for item in items], ["certificate", "key"])
+        mismatched = b"-----BEGIN CERTIFICATE-----\nignored\n-----END PRIVATE KEY-----"
+        self.assertEqual(imports.parse_pem_bundle(mismatched), [])
+
+    def test_unterminated_pem_headers_complete_quickly(self) -> None:
+        header = b"-----BEGIN UNTERMINATED-----\n"
+        data = header * ((80 * 1024) // len(header))
+        started = time.perf_counter()
+        items = imports.parse_pem_bundle(data)
+        elapsed = time.perf_counter() - started
+        self.assertEqual(items, [])
+        self.assertLess(elapsed, 2.0, f"PEM scanning took {elapsed:.3f}s for {len(data)} bytes")
+
     def test_key_exports(self) -> None:
         for fmt in ("pem-public", "pem-private", "pkcs8", "ssh-public", "jwk-public", "pvk"):
             result = exports.export_key(self.key, fmt, name="export")