feat: deprecation warning for the Kyber classes (WP-6)

CHANGELOG.md

@@ -6,6 +6,7 @@
 - Legacy freeze (issue #6): golden vectors capture the byte output of pykyber v1.0.5 (append-only `tests/vectors/legacy-kyber-v1.0.5.json`, with the generator); Rust and Python tests fail on any change to seeded keygen, seeded encapsulation, or decapsulation output.
 - CI (issue #7): a `test` job on Linux, macOS and Windows (oldest and newest Python) runs `cargo fmt --check`, `cargo clippy --all-targets -- -D warnings`, `cargo test --locked`, `maturin develop` and `pytest`; a `commits` job re-runs them on every commit of a PR via `git rebase --exec scripts/check.sh`. `requirements-dev.txt` declares maturin and pytest; wheel release now waits for `test`.
 - ML-KEM (issue #13): new classes `ML_KEM_512`, `ML_KEM_768` and `ML_KEM_1024` (FIPS 203) with the same shape as `Kyber*` (an instance is a keypair; static `encapsulate`, `decapsulate`, `from_keys`; hex/b64/dict; tuple unpacking). A 64-byte seed `d || z` gives deterministic keys and is exposed as `.seed` only for such keypairs. ML-KEM methods refuse `Kyber*` objects by type (`AlgorithmMismatchError`); raw bytes cannot be told apart. New `KyberError` subclasses: `InvalidInputError`, `InvalidEncapsulationKeyError`, `InvalidDecapsulationKeyError`, `RandomnessError`, `DecapsulationError`, `AlgorithmMismatchError`. The Kyber classes are unchanged.
+- Kyber deprecation (issue #14): the first use of any `Kyber512/768/1024` entry point (constructor, `from_keys`, static `encapsulate` or `decapsulate`) emits one `KyberDeprecationWarning` (a `DeprecationWarning` subclass, exported from `pykyber`) per process. Importing does not warn, `ML_KEM_*` never warns, and Kyber output is unchanged. Python shows `DeprecationWarning` by default only when it is attributed to `__main__`; run with `-W default::DeprecationWarning` or call `warnings.simplefilter` to see it elsewhere.
 
 ## v1.0.5
 

python/pykyber/__init__.py

@@ -12,6 +12,7 @@
     InvalidInputError,
     RandomnessError,
 )
+from ._deprecation import KyberDeprecationWarning
 from ._mlkem import ML_KEM_512, ML_KEM_768, ML_KEM_1024
 from ._kyber import (
     Kyber768,
@@ -31,6 +32,7 @@
     "ML_KEM_768",
     "ML_KEM_1024",
     "KyberError",
+    "KyberDeprecationWarning",
     "AlgorithmMismatchError",
     "DecapsulationError",
     "InvalidDecapsulationKeyError",

python/pykyber/_deprecation.py

@@ -0,0 +1,51 @@
+"""Deprecation warning for the legacy round-3 Kyber classes."""
+
+import functools
+import threading
+import warnings
+
+MIGRATION_GUIDE_URL = "https://github.com/Zikithezikit/pykyber/blob/main/docs/migration.md"
+
+# warnings.warn() frames to skip: 1 is _warn_once, 2 is the wrapper made by
+# _deprecated_kyber, 3 is the user's code that called the Kyber entry point.
+_CALLER_STACKLEVEL = 3
+
+_MESSAGE = (
+    "Kyber512, Kyber768 and Kyber1024 implement round-3 CRYSTALS-Kyber, which is "
+    "deprecated in favor of ML-KEM (FIPS 203): ML_KEM_512, ML_KEM_768 and "
+    "ML_KEM_1024. Kyber (round 3) and ML-KEM are not interoperable, so both sides "
+    "must switch, and switching changes the shared secret. "
+    f"See {MIGRATION_GUIDE_URL}"
+)
+
+_lock = threading.Lock()
+_warned = False
+
+
+class KyberDeprecationWarning(DeprecationWarning):
+    """The legacy round-3 Kyber classes are deprecated in favor of ML-KEM."""
+
+
+def _warn_once() -> None:
+    """Warn on the first call in this process, then do nothing.
+
+    The flag is our own: the default warning filter keeps one registry entry per
+    call site, so it would repeat the warning for every new location.
+    """
+    global _warned
+    with _lock:
+        if _warned:
+            return
+        _warned = True
+    warnings.warn(_MESSAGE, KyberDeprecationWarning, stacklevel=_CALLER_STACKLEVEL)
+
+
+def _deprecated_kyber(func):
+    """Decorate a Kyber entry point so its first use warns."""
+
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        _warn_once()
+        return func(*args, **kwargs)
+
+    return wrapper

python/pykyber/_kyber.py

@@ -1,6 +1,7 @@
 """Kyber post-quantum key encapsulation."""
 
 from typing import Optional
+from ._deprecation import _deprecated_kyber
 from ._error import KyberError
 from ._types import EncapsulationResult, Keypair
 from ._validation import _validate_encapsulate, _validate_decapsulate, _validate_keypair, _validate_pk
@@ -14,6 +15,7 @@ class Kyber512:
     CIPHERTEXT_SIZE = 768
     SHARED_SECRET_SIZE = 32
 
+    @_deprecated_kyber
     def __new__(cls, seed: Optional[bytes] = None) -> Keypair:
         """Generate a new key pair, optionally from a 64-byte seed."""
         from . import _pykyber
@@ -23,18 +25,21 @@ def __new__(cls, seed: Optional[bytes] = None) -> Keypair:
                        _validate_decapsulate(_pykyber._decapsulate_512, Kyber512.CIPHERTEXT_SIZE, "Kyber512"))
 
     @staticmethod
+    @_deprecated_kyber
     def encapsulate(public_key: bytes) -> EncapsulationResult:
         """Encapsulate a shared secret using a public key (no keypair needed)."""
         from . import _pykyber
         return EncapsulationResult(*_validate_encapsulate(_pykyber._encapsulate_512, Kyber512.PUBLIC_KEY_SIZE, "Kyber512.encapsulate")(public_key))
 
     @staticmethod
+    @_deprecated_kyber
     def decapsulate(ciphertext: bytes, secret_key: bytes) -> bytes:
         """Decapsulate a shared secret using ciphertext and secret key (no keypair needed)."""
         from . import _pykyber
         return _validate_decapsulate(_pykyber._decapsulate_512, Kyber512.CIPHERTEXT_SIZE, "Kyber512.decapsulate")(ciphertext, secret_key)
 
     @classmethod
+    @_deprecated_kyber
     def from_keys(cls, public_key: bytes, secret_key: bytes) -> Keypair:
         """Create a Keypair object from existing public and secret keys."""
         from . import _pykyber
@@ -52,6 +57,7 @@ class Kyber768:
     CIPHERTEXT_SIZE = 1088
     SHARED_SECRET_SIZE = 32
 
+    @_deprecated_kyber
     def __new__(cls, seed: Optional[bytes] = None) -> Keypair:
         """Generate a new key pair, optionally from a 64-byte seed."""
         from . import _pykyber
@@ -61,18 +67,21 @@ def __new__(cls, seed: Optional[bytes] = None) -> Keypair:
                        _validate_decapsulate(_pykyber._decapsulate_768, Kyber768.CIPHERTEXT_SIZE, "Kyber768"))
 
     @staticmethod
+    @_deprecated_kyber
     def encapsulate(public_key: bytes) -> EncapsulationResult:
         """Encapsulate a shared secret using a public key (no keypair needed)."""
         from . import _pykyber
         return EncapsulationResult(*_validate_encapsulate(_pykyber._encapsulate_768, Kyber768.PUBLIC_KEY_SIZE, "Kyber768.encapsulate")(public_key))
 
     @staticmethod
+    @_deprecated_kyber
     def decapsulate(ciphertext: bytes, secret_key: bytes) -> bytes:
         """Decapsulate a shared secret using ciphertext and secret key (no keypair needed)."""
         from . import _pykyber
         return _validate_decapsulate(_pykyber._decapsulate_768, Kyber768.CIPHERTEXT_SIZE, "Kyber768.decapsulate")(ciphertext, secret_key)
 
     @classmethod
+    @_deprecated_kyber
     def from_keys(cls, public_key: bytes, secret_key: bytes) -> Keypair:
         """Create a Keypair object from existing public and secret keys."""
         from . import _pykyber
@@ -90,6 +99,7 @@ class Kyber1024:
     CIPHERTEXT_SIZE = 1568
     SHARED_SECRET_SIZE = 32
 
+    @_deprecated_kyber
     def __new__(cls, seed: Optional[bytes] = None) -> Keypair:
         """Generate a new key pair, optionally from a 64-byte seed."""
         from . import _pykyber
@@ -99,18 +109,21 @@ def __new__(cls, seed: Optional[bytes] = None) -> Keypair:
                        _validate_decapsulate(_pykyber._decapsulate_1024, Kyber1024.CIPHERTEXT_SIZE, "Kyber1024"))
 
     @staticmethod
+    @_deprecated_kyber
     def encapsulate(public_key: bytes) -> EncapsulationResult:
         """Encapsulate a shared secret using a public key (no keypair needed)."""
         from . import _pykyber
         return EncapsulationResult(*_validate_encapsulate(_pykyber._encapsulate_1024, Kyber1024.PUBLIC_KEY_SIZE, "Kyber1024.encapsulate")(public_key))
 
     @staticmethod
+    @_deprecated_kyber
     def decapsulate(ciphertext: bytes, secret_key: bytes) -> bytes:
         """Decapsulate a shared secret using ciphertext and secret key (no keypair needed)."""
         from . import _pykyber
         return _validate_decapsulate(_pykyber._decapsulate_1024, Kyber1024.CIPHERTEXT_SIZE, "Kyber1024.decapsulate")(ciphertext, secret_key)
 
     @classmethod
+    @_deprecated_kyber
     def from_keys(cls, public_key: bytes, secret_key: bytes) -> Keypair:
         """Create a Keypair object from existing public and secret keys."""
         from . import _pykyber

python/tests/test_kyber_deprecation.py

@@ -0,0 +1,156 @@
+"""The Kyber* classes warn once per process (KyberDeprecationWarning) and
+nothing else changes. Counting is done in subprocesses, because the flag is
+per process and pytest would otherwise share it across the whole run."""
+
+import json
+import os
+import subprocess
+import sys
+import textwrap
+
+import pytest
+
+import pykyber
+from pykyber import _deprecation
+
+KYBER_CLASSES = ["Kyber512", "Kyber768", "Kyber1024"]
+ENTRY_POINTS = ["new", "from_keys", "encapsulate", "decapsulate"]
+MIGRATION_GUIDE_URL = "https://github.com/Zikithezikit/pykyber/blob/main/docs/migration.md"
+REPEATS = 3  # uses of every class and entry point per child process
+CHILD_TIMEOUT_SECONDS = 120
+
+# Records every warning the child sees, one JSON object per warning on stdout.
+# {body} runs under a "record everything" filter so the child reports what is
+# really emitted, whatever filters the parent passed with -W.
+CHILD_TEMPLATE = textwrap.dedent(
+    """
+    import json, sys, warnings
+    import pykyber
+    with warnings.catch_warnings(record=True) as seen:
+        warnings.simplefilter("always")
+    {body}
+    print(json.dumps([[w.category.__name__, str(w.message), w.filename] for w in seen]))
+    """
+)
+
+KYBER_BODY = """
+        for _ in range({repeats}):
+            for name in {classes!r}:
+                cls = getattr(pykyber, name)
+                kp = cls()
+                res = cls.encapsulate(kp.public_key)
+                cls.decapsulate(res.ciphertext, kp.secret_key)
+                cls.from_keys(kp.public_key, kp.secret_key)
+"""
+
+MLKEM_BODY = """
+        for name in ("ML_KEM_512", "ML_KEM_768", "ML_KEM_1024"):
+            cls = getattr(pykyber, name)
+            kp = cls()
+            res = cls.encapsulate(kp.public_key)
+            cls.decapsulate(res.ciphertext, kp.secret_key)
+            cls.from_keys(kp.public_key, kp.secret_key)
+"""
+
+
+def run_child(body, *flags):
+    """Run CHILD_TEMPLATE with body in a fresh interpreter and return its warnings."""
+    code = CHILD_TEMPLATE.replace("{body}", textwrap.indent(textwrap.dedent(body), "    "))
+    env = {k: v for k, v in os.environ.items() if k != "PYTHONWARNINGS"}
+    proc = subprocess.run(
+        [sys.executable, *flags, "-c", code],
+        capture_output=True,
+        text=True,
+        timeout=CHILD_TIMEOUT_SECONDS,
+        env=env,
+    )
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout.strip().splitlines()[-1])
+
+
+def kyber_body():
+    return KYBER_BODY.format(repeats=REPEATS, classes=KYBER_CLASSES)
+
+
+@pytest.mark.parametrize("flag", ["always::DeprecationWarning", "default"])
+def test_many_uses_warn_exactly_once(flag):
+    seen = run_child(kyber_body(), "-W", flag)
+    assert [category for category, _, _ in seen] == ["KyberDeprecationWarning"]
+
+
+def test_import_alone_warns_nothing():
+    assert run_child("pass", "-W", "always::DeprecationWarning") == []
+
+
+def test_ml_kem_never_warns():
+    assert run_child(MLKEM_BODY, "-W", "always::DeprecationWarning") == []
+
+
+def test_message_content():
+    ((_, message, _),) = run_child(kyber_body())
+    assert "not interoperable" in message
+    assert "both sides must switch" in message
+    assert "changes the shared secret" in message
+    assert MIGRATION_GUIDE_URL in message
+
+
+def test_library_caller_is_silent_by_default(tmp_path):
+    """Python hides DeprecationWarning unless it is attributed to __main__, so a
+    call made from an imported module prints nothing under the default filters."""
+    (tmp_path / "uses_kyber.py").write_text("import pykyber\npykyber.Kyber768()\n")
+    code = f"import sys; sys.path.insert(0, {str(tmp_path)!r}); import uses_kyber"
+    env = {k: v for k, v in os.environ.items() if k != "PYTHONWARNINGS"}
+    proc = subprocess.run(
+        [sys.executable, "-c", code],
+        capture_output=True,
+        text=True,
+        timeout=CHILD_TIMEOUT_SECONDS,
+        env=env,
+    )
+    assert proc.returncode == 0
+    assert "KyberDeprecationWarning" not in proc.stderr
+    assert proc.stderr == ""
+
+
+def test_main_caller_is_shown_by_default():
+    env = {k: v for k, v in os.environ.items() if k != "PYTHONWARNINGS"}
+    proc = subprocess.run(
+        [sys.executable, "-c", "import pykyber\npykyber.Kyber768()"],
+        capture_output=True,
+        text=True,
+        timeout=CHILD_TIMEOUT_SECONDS,
+        env=env,
+    )
+    assert proc.returncode == 0
+    assert "KyberDeprecationWarning" in proc.stderr
+
+
+def test_warning_is_exported_and_is_a_deprecation_warning():
+    assert "KyberDeprecationWarning" in pykyber.__all__
+    assert issubclass(pykyber.KyberDeprecationWarning, DeprecationWarning)
+    assert not issubclass(pykyber.KyberDeprecationWarning, pykyber.KyberError)
+
+
+def call_entry_point(cls, entry_point, kp, res):
+    """Make one call to a Kyber entry point from this file."""
+    if entry_point == "new":
+        return cls()
+    if entry_point == "from_keys":
+        return cls.from_keys(kp.public_key, kp.secret_key)
+    if entry_point == "encapsulate":
+        return cls.encapsulate(kp.public_key)
+    return cls.decapsulate(res.ciphertext, kp.secret_key)
+
+
+@pytest.mark.parametrize("entry_point", ENTRY_POINTS)
+@pytest.mark.parametrize("class_name", KYBER_CLASSES)
+def test_warning_is_attributed_to_the_callers_file(class_name, entry_point, monkeypatch):
+    cls = getattr(pykyber, class_name)
+    monkeypatch.setattr(_deprecation, "_warned", True)  # keep the setup quiet
+    kp = cls()
+    res = cls.encapsulate(kp.public_key)
+    monkeypatch.setattr(_deprecation, "_warned", False)
+    with pytest.warns(pykyber.KyberDeprecationWarning) as record:
+        call_entry_point(cls, entry_point, kp, res)
+    assert len(record) == 1
+    assert record[0].filename == __file__