Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES.rst
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ Unreleased

- **Breaking change**: Drop support for Python 3.10.
- Correct COSE signing header annotations to accept dictionaries and ResolvedHeader in strict type checkers.
- Annotate factory class methods (``COSE.new``, ``CWT.new``, ``Claims.new``, ``Claims.from_json``, ``Signer.new``, ``Signer.from_jwk``, ``Signer.from_pem``, ``COSEMessage.loads``, etc.) to return ``Self`` for strict type checkers.
- Follow RFC 9597 (CWT Claims in COSE Headers), which supersedes draft-ietf-cose-cwt-claims-in-headers-06.
- **Breaking change**: ``COSEHeaders.CWT_CLAIMS`` is now ``15``, the label assigned by IANA in RFC 9597 (it was ``13``, the label requested by draft-06). IANA assigned ``13`` to the unrelated ``kcwt`` header parameter (RFC 9528), so ``13`` is no longer interpreted as CWT Claims. Tokens using the draft label are no longer cross-checked by ``CWT.decode()``.
- ``CWT.decode()`` now checks the CWT Claims header parameter in unprotected headers and in every COSE layer of a nested CWT, not only in the protected header of the innermost layer.
Expand Down
6 changes: 3 additions & 3 deletions cwt/claims.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import json
from typing import Any, Dict, List, Union
from typing import Any, Dict, List, Self, Union

from .const import CWT_CLAIM_NAMES
from .cose_key import COSEKey
Expand Down Expand Up @@ -61,7 +61,7 @@ def __init__(
return

@classmethod
def new(cls, claims: Dict[int, Any], private_claim_names: Dict[str, int] = {}):
def new(cls, claims: Dict[int, Any], private_claim_names: Dict[str, int] = {}) -> Self:
"""
Creates a Claims object from a CBOR-like(Dict[int, Any]) claim object.

Expand Down Expand Up @@ -92,7 +92,7 @@ def from_json(
cls,
claims: Union[str, bytes, Dict[str, Any]],
private_claim_names: Dict[str, int] = {},
):
) -> Self:
"""
Converts a JWT claims object into a CWT claims object which has numeric
keys. If a key string in JSON data cannot be mapped to a numeric key,
Expand Down
4 changes: 2 additions & 2 deletions cwt/cose.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from collections.abc import Mapping
from typing import Any, Dict, List, Optional, Tuple, Union
from typing import Any, Dict, List, Optional, Self, Tuple, Union

from cbor2 import CBORTag
from cryptography.x509 import load_pem_x509_certificates
Expand Down Expand Up @@ -76,7 +76,7 @@ def new(
verify_kid: bool = False,
ca_certs: str = "",
deterministic_header: bool = False,
):
) -> Self:
"""
Constructor.

Expand Down
8 changes: 4 additions & 4 deletions cwt/cose_message.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from __future__ import annotations

from collections.abc import Mapping
from typing import Any, Dict, List, Optional, Tuple, Union
from typing import Any, Dict, List, Optional, Self, Tuple, Union

from cbor2 import CBORTag, loads

Expand Down Expand Up @@ -115,7 +115,7 @@ def __ne__(self: COSEMessage, other: object) -> bool:
return not self.__eq__(other)

@classmethod
def loads(cls, msg: bytes):
def loads(cls, msg: bytes) -> Self:
tagged = loads(msg)
if not isinstance(tagged, CBORTag):
raise ValueError("Invalid COSE format.")
Expand All @@ -125,11 +125,11 @@ def loads(cls, msg: bytes):
return cls(type, tagged.value)

@classmethod
def from_cose_signature(cls, signature: Union[List[Any], Tuple[Any, ...]]):
def from_cose_signature(cls, signature: Union[List[Any], Tuple[Any, ...]]) -> Self:
return cls(COSETypes.SIGNATURE, signature)

@classmethod
def from_cose_recipient(cls, recipient: Union[List[Any], Tuple[Any, ...]]):
def from_cose_recipient(cls, recipient: Union[List[Any], Tuple[Any, ...]]) -> Self:
return cls(COSETypes.RECIPIENT, recipient)

@property
Expand Down
4 changes: 2 additions & 2 deletions cwt/cwt.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from calendar import timegm
from collections.abc import Mapping
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple, Union
from typing import Any, Dict, List, Optional, Self, Tuple, Union

from cbor2 import CBORTag, dumps

Expand Down Expand Up @@ -61,7 +61,7 @@ def new(
expires_in: int = CWT_DEFAULT_EXPIRES_IN,
leeway: int = CWT_DEFAULT_LEEWAY,
ca_certs: str = "",
):
) -> Self:
"""
Constructor.

Expand Down
6 changes: 3 additions & 3 deletions cwt/recipients.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Any, Dict, List, Optional, Tuple, Union
from typing import Any, Dict, List, Optional, Self, Tuple, Union

from .cbor_processor import CBORProcessor
from .cose_key import COSEKey
Expand Down Expand Up @@ -26,7 +26,7 @@ def from_list(
hpke_psk: Optional[bytes] = None,
extra_info: bytes = b"",
hpke_aad: bytes = b"",
):
) -> Self:
"""
Create Recipients from a CBOR-like list.
"""
Expand All @@ -35,7 +35,7 @@ def from_list(
res.append(Recipient.from_list(r, context, hpke_psk=hpke_psk, extra_info=extra_info, hpke_aad=hpke_aad))
return cls(res, verify_kid)

def derive_key(self, keys: List[COSEKeyInterface], alg: int, external_aad: bytes, content_aad: bytes) -> COSEKeyInterface:
def derive_key(self, keys: List[COSEKeyInterface], alg: int, external_aad: bytes, content_aad: str) -> COSEKeyInterface:
"""
Decodes an appropriate key from recipients or keys provided as a parameter ``keys``.
"""
Expand Down
8 changes: 4 additions & 4 deletions cwt/signer.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Any, Dict, Union
from typing import Any, Dict, Self, Union

from .cbor_processor import CBORProcessor
from .const import COSE_ALGORITHMS_SIGNATURE
Expand Down Expand Up @@ -63,7 +63,7 @@ def new(
protected: Union[dict, bytes, ResolvedHeader] = {},
unprotected: Union[dict, ResolvedHeader] = {},
signature: bytes = b"",
):
) -> Self:
"""
Creates a signer information object (COSE_Signature).

Expand All @@ -85,7 +85,7 @@ def new(
return cls(cose_key, p, u, signature)

@classmethod
def from_jwk(cls, data: Union[str, bytes, Dict[str, Any]]):
def from_jwk(cls, data: Union[str, bytes, Dict[str, Any]]) -> Self:
"""
Creates a signer information object (COSE_Signature) from JWK.
The ``alg`` in the JWK will be included in the protected header,
Expand Down Expand Up @@ -122,7 +122,7 @@ def from_pem(
data: Union[str, bytes],
alg: Union[int, str] = "",
kid: Union[bytes, str] = b"",
):
) -> Self:
"""
Creates a signer information object (COSE_Signature) from PEM-formatted key.
The ``alg`` in the JWK will be included in the protected header,
Expand Down
12 changes: 6 additions & 6 deletions tests/test_recipient.py
Original file line number Diff line number Diff line change
Expand Up @@ -573,15 +573,15 @@ def test_recipients_constructor(self):
def test_recipients_constructor_with_recipient_alg_direct(self):
key = COSEKey.from_symmetric_key("mysecret", alg="HMAC 256/64", kid="our-secret")
r = Recipients([Recipient.new(unprotected={COSEHeaders.ALG: COSEAlgs.DIRECT, COSEHeaders.KID: b"our-secret"})])
key = r.derive_key([key], key.alg, b"", b"")
key = r.derive_key([key], key.alg, b"", "Enc_Recipient")
assert key.kty == 4
assert key.alg == COSEAlgs.HS256_64
assert key.kid == b"our-secret"

def test_recipients_derive_key_without_key(self):
r = Recipients([RecipientInterface(unprotected={COSEHeaders.ALG: COSEAlgs.DIRECT, COSEHeaders.KID: b"our-secret"})])
with pytest.raises(ValueError) as err:
r.derive_key([], 0, b"", b"")
r.derive_key([], 0, b"", "Enc_Recipient")
pytest.fail("derive_key() should fail.")
assert "key is not found." in str(err.value)

Expand All @@ -602,7 +602,7 @@ def test_recipients_derive_key_without_key(self):
def test_recipients_derive_key_with_empty_recipients(self, material, context):
r = Recipients([])
with pytest.raises(ValueError) as err:
r.derive_key([material], 0, b"", b"")
r.derive_key([material], 0, b"", "Enc_Recipient")
pytest.fail("derive_key() should fail.")
assert "No recipients." in str(err.value)

Expand Down Expand Up @@ -671,7 +671,7 @@ def test_recipients_derive_key_with_different_kid(self):
key = COSEKey.from_symmetric_key("mysecret", alg="HMAC 256/64", kid="our-secret")
r = Recipients([RecipientInterface(unprotected={COSEHeaders.ALG: COSEAlgs.DIRECT, COSEHeaders.KID: b"your-secret"})])
with pytest.raises(ValueError) as err:
r.derive_key([key], key.alg, b"", b"")
r.derive_key([key], key.alg, b"", "Enc_Recipient")
pytest.fail("derive_key() should fail.")
assert "key is not found." in str(err.value)

Expand Down Expand Up @@ -747,14 +747,14 @@ def test_recipients_open_without_key(self):
)
rs = Recipients([r])
with pytest.raises(ValueError) as err:
rs.derive_key([], 0, b"", b"")
rs.derive_key([], 0, b"", "Enc_Recipient")
pytest.fail("open() should fail.")
assert "key is not found." in str(err.value)

def test_recipients_open_with_empty_recipients(self, rsk1):
rs = Recipients([])
with pytest.raises(ValueError) as err:
rs.derive_key([rsk1], 0, b"", b"")
rs.derive_key([rsk1], 0, b"", "Enc_Recipient")
pytest.fail("open() should fail.")
assert "No recipients." in str(err.value)

Expand Down
92 changes: 92 additions & 0 deletions tests/test_typing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""
Tests for static return types of factory class methods.

``typing.assert_type`` is a no-op at runtime, but mypy (run via pre-commit)
checks that each factory is annotated to return ``Self``.
"""

from typing import assert_type

from cwt import COSE, CWT, Claims, COSEKey, COSEMessage, Signer
from cwt.recipients import Recipients

from .utils import key_path


class MyCOSE(COSE):
pass


class MyCWT(CWT):
pass


class MyClaims(Claims):
pass


class MySigner(Signer):
pass


class MyRecipients(Recipients):
pass


class MyCOSEMessage(COSEMessage):
pass


class TestTyping:
"""
Tests for return types of factory class methods.
"""

def test_typing_cose_and_cwt_new(self) -> None:
assert isinstance(assert_type(COSE.new(), COSE), COSE)
assert isinstance(assert_type(MyCOSE.new(), MyCOSE), MyCOSE)
assert isinstance(assert_type(CWT.new(), CWT), CWT)
assert isinstance(assert_type(MyCWT.new(), MyCWT), MyCWT)

def test_typing_claims(self) -> None:
assert isinstance(assert_type(Claims.new({1: "coaps://as.example"}), Claims), Claims)
assert isinstance(assert_type(MyClaims.new({1: "coaps://as.example"}), MyClaims), MyClaims)
assert isinstance(assert_type(Claims.from_json({"iss": "coaps://as.example"}), Claims), Claims)
assert isinstance(assert_type(MyClaims.from_json({"iss": "coaps://as.example"}), MyClaims), MyClaims)

def test_typing_signer(self) -> None:
with open(key_path("private_key_ed25519.pem")) as key_file:
pem = key_file.read()
jwk = {
"kty": "OKP",
"d": "L8JS08VsFZoZxGa9JvzYmCWOwg7zaKcei3KZmYsj7dc",
"use": "sig",
"crv": "Ed25519",
"kid": "01",
"x": "2E6dX83gqD_D0eAmqnaHe1TC1xuld6iAKXfw2OVATr0",
"alg": "EdDSA",
}
key = COSEKey.from_pem(pem, kid="01")
assert isinstance(assert_type(Signer.new(cose_key=key), Signer), Signer)
assert isinstance(assert_type(MySigner.new(cose_key=key), MySigner), MySigner)
assert isinstance(assert_type(Signer.from_jwk(jwk), Signer), Signer)
assert isinstance(assert_type(MySigner.from_jwk(jwk), MySigner), MySigner)
assert isinstance(assert_type(Signer.from_pem(pem, kid="01"), Signer), Signer)
assert isinstance(assert_type(MySigner.from_pem(pem, kid="01"), MySigner), MySigner)

def test_typing_recipients(self) -> None:
assert isinstance(assert_type(Recipients.from_list([]), Recipients), Recipients)
assert isinstance(assert_type(MyRecipients.from_list([]), MyRecipients), MyRecipients)

def test_typing_cose_message(self) -> None:
with open(key_path("private_key_ed25519.pem")) as key_file:
key = COSEKey.from_pem(key_file.read(), kid="01")
encoded = COSE.new(alg_auto_inclusion=True).encode_and_sign(b"Hello world!", key)
assert isinstance(encoded, bytes)
signature = [b"", {}, b""]
assert isinstance(assert_type(COSEMessage.loads(encoded), COSEMessage), COSEMessage)
assert isinstance(assert_type(MyCOSEMessage.loads(encoded), MyCOSEMessage), MyCOSEMessage)
assert isinstance(assert_type(COSEMessage.from_cose_signature(signature), COSEMessage), COSEMessage)
assert isinstance(assert_type(MyCOSEMessage.from_cose_signature(signature), MyCOSEMessage), MyCOSEMessage)
assert isinstance(assert_type(COSEMessage.from_cose_recipient(signature), COSEMessage), COSEMessage)
assert isinstance(assert_type(MyCOSEMessage.from_cose_recipient(signature), MyCOSEMessage), MyCOSEMessage)
Loading