diff --git a/CHANGELOG.md b/CHANGELOG.md index 083e7de..8fae603 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/python/pykyber/__init__.py b/python/pykyber/__init__.py index c62c361..47d3e24 100644 --- a/python/pykyber/__init__.py +++ b/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", diff --git a/python/pykyber/_deprecation.py b/python/pykyber/_deprecation.py new file mode 100644 index 0000000..fcf6776 --- /dev/null +++ b/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 diff --git a/python/pykyber/_kyber.py b/python/pykyber/_kyber.py index 877a06d..74d4678 100644 --- a/python/pykyber/_kyber.py +++ b/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 diff --git a/python/tests/test_kyber_deprecation.py b/python/tests/test_kyber_deprecation.py new file mode 100644 index 0000000..40a427b --- /dev/null +++ b/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__