From 49ac757442501ba4e52ba9721df28262bd6d095f Mon Sep 17 00:00:00 2001 From: AJITOMI Daisuke Date: Mon, 9 Nov 2026 20:21:50 +0900 Subject: [PATCH] refactor: modernize type annotations --- .pre-commit-config.yaml | 6 + CHANGES.rst | 1 + README.md | 2 +- cwt/algs/asymmetric.py | 8 +- cwt/algs/ec2.py | 30 ++-- cwt/algs/okp.py | 70 ++++----- cwt/algs/raw.py | 6 +- cwt/algs/rsa.py | 6 +- cwt/algs/symmetric.py | 44 +++--- cwt/cbor_processor.py | 4 +- cwt/claims.py | 58 +++---- cwt/cose.py | 194 ++++++++++++------------ cwt/cose_key.py | 60 ++++---- cwt/cose_key_interface.py | 32 ++-- cwt/cose_message.py | 48 +++--- cwt/cwt.py | 108 ++++++------- cwt/encrypted_cose_key.py | 12 +- cwt/helpers/hcert.py | 6 +- cwt/recipient.py | 38 ++--- cwt/recipient_algs/aes_key_wrap.py | 12 +- cwt/recipient_algs/direct.py | 8 +- cwt/recipient_algs/direct_hkdf.py | 12 +- cwt/recipient_algs/direct_key.py | 8 +- cwt/recipient_algs/ecdh_aes_key_wrap.py | 18 +-- cwt/recipient_algs/ecdh_direct_hkdf.py | 18 +-- cwt/recipient_algs/hpke.py | 18 +-- cwt/recipient_interface.py | 44 +++--- cwt/recipients.py | 14 +- cwt/signer.py | 44 +++--- cwt/utils.py | 38 ++--- samples/eudcc/swedish_verifier.py | 4 +- samples/eudcc/verifier.py | 4 +- tests/test_cose.py | 6 +- 33 files changed, 493 insertions(+), 488 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 5069b939..fd0f396c 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,4 +1,10 @@ repos: + - repo: https://github.com/asottile/pyupgrade + rev: v3.22.0 + hooks: + - id: pyupgrade + args: [--py311-plus] + - repo: https://github.com/psf/black rev: 25.1.0 hooks: diff --git a/CHANGES.rst b/CHANGES.rst index cc7c7c14..63a9114f 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -7,6 +7,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. +- Modernize type annotations to PEP 604/585 syntax (e.g. ``X | None``, ``dict[str, Any]``) and add pyupgrade (``--py311-plus``) to pre-commit. - 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. diff --git a/README.md b/README.md index 042c8459..2087c9d3 100644 --- a/README.md +++ b/README.md @@ -1323,7 +1323,7 @@ except Exception as err: print(err) ``` -A raw CWT structure (Dict[int, Any]) can also be used as follows: +A raw CWT structure (dict[int, Any]) can also be used as follows: ```py import cwt diff --git a/cwt/algs/asymmetric.py b/cwt/algs/asymmetric.py index fb8d2442..c26f24cc 100644 --- a/cwt/algs/asymmetric.py +++ b/cwt/algs/asymmetric.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List +from typing import Any from cryptography.x509 import Certificate, DNSName, load_der_x509_certificate from cryptography.x509.oid import NameOID @@ -9,12 +9,12 @@ class AsymmetricKey(COSEKeyInterface): - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) self._key: Any = b"" self._cert: Certificate = None - self._intermediates: List[Certificate] = [] + self._intermediates: list[Certificate] = [] if 33 in params: if not isinstance(params[33], (bytes, list)): @@ -26,7 +26,7 @@ def __init__(self, params: Dict[int, Any]): self._intermediates.append(load_der_x509_certificate(c)) return - def validate_certificate(self, ca_certs: List[Certificate]) -> bool: + def validate_certificate(self, ca_certs: list[Certificate]) -> bool: if not ca_certs: raise ValueError("ca_certs should be set.") if not self._cert: diff --git a/cwt/algs/ec2.py b/cwt/algs/ec2.py index 12ee8cec..da4982da 100644 --- a/cwt/algs/ec2.py +++ b/cwt/algs/ec2.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, Optional, Union +from typing import Any import cryptography from cryptography.hazmat.primitives import hashes @@ -38,7 +38,7 @@ class EC2Key(AsymmetricKey): COSE_KEY_OPERATION_VALUES["deriveBits"], ] - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) self._public_key: Any = None self._private_key: Any = None @@ -72,7 +72,7 @@ def __init__(self, params: Dict[int, Any]): # Validate alg and key_ops. if self._key_ops: - if set(self._key_ops) & set([3, 4, 5, 6, 9, 10]): + if set(self._key_ops) & {3, 4, 5, 6, 9, 10}: raise ValueError("Unknown or not permissible key_ops(4) for EC2.") if not self._alg and not self._key_ops: @@ -99,9 +99,9 @@ def __init__(self, params: Dict[int, Any]): if self._key_ops: if -4 in params: # private key for signing. - if not (set(self._key_ops) & set([1, 2])): + if not (set(self._key_ops) & {1, 2}): raise ValueError("Invalid key_ops for signing key.") - if set(self._key_ops) & set([7, 8]): + if set(self._key_ops) & {7, 8}: raise ValueError("Signing key should not be used for key derivation.") else: # public key for signing. @@ -119,9 +119,9 @@ def __init__(self, params: Dict[int, Any]): if self._key_ops: if -4 in params: # private key for key derivation. - if not (set(self._key_ops) & set([7, 8])): + if not (set(self._key_ops) & {7, 8}): raise ValueError("Invalid key_ops for key derivation.") - if set(self._key_ops) & set([1, 2]): + if set(self._key_ops) & {1, 2}: raise ValueError("ECDHE key should not be used for signing.") else: # public key for key derivation. @@ -147,9 +147,9 @@ def __init__(self, params: Dict[int, Any]): else: if -4 in params: # private key. - if set(self._key_ops) & set([1, 2]): + if set(self._key_ops) & {1, 2}: # private key for signing. - if set(self._key_ops) & set([7, 8]): + if set(self._key_ops) & {7, 8}: raise ValueError("EC2 Private key should not be used for both signing and key derivation.") if self._crv == 1: self._alg = -7 # ES256 @@ -164,7 +164,7 @@ def __init__(self, params: Dict[int, Any]): if 2 in self._key_ops: if len(self._key_ops) > 1: raise ValueError("Invalid key_ops for public key.") - elif not (set(self._key_ops) & set([7, 8])): + elif not (set(self._key_ops) & {7, 8}): raise ValueError("Invalid key_ops for public key.") if self._alg in COSE_ALGORITHMS_CKDM_KEY_AGREEMENT_ES.values(): @@ -223,9 +223,9 @@ def __init__(self, params: Dict[int, Any]): return @staticmethod - def to_cose_key(k: Union[EllipticCurvePrivateKey, EllipticCurvePublicKey]) -> Dict[int, Any]: + def to_cose_key(k: EllipticCurvePrivateKey | EllipticCurvePublicKey) -> dict[int, Any]: key_len: int = 32 - cose_key: Dict[int, Any] = {} + cose_key: dict[int, Any] = {} cose_key[1] = COSE_KEY_TYPES["EC2"] if not hasattr(k, "curve"): @@ -250,14 +250,14 @@ def to_cose_key(k: Union[EllipticCurvePrivateKey, EllipticCurvePublicKey]) -> Di return cose_key @property - def key(self) -> Union[EllipticCurvePublicKey, EllipticCurvePrivateKey]: + def key(self) -> EllipticCurvePublicKey | EllipticCurvePrivateKey: return self._key @property def crv(self) -> int: return self._crv - def to_dict(self) -> Dict[int, Any]: + def to_dict(self) -> dict[int, Any]: res = super().to_dict() res[-1] = self._crv res[-2] = self._x @@ -288,7 +288,7 @@ def verify(self, msg: bytes, sig: bytes): except ValueError as err: raise VerifyError("Invalid signature.") from err - def derive_bytes(self, length: int, material: bytes = b"", info: bytes = b"", public_key: Optional[Any] = None) -> bytes: + def derive_bytes(self, length: int, material: bytes = b"", info: bytes = b"", public_key: Any | None = None) -> bytes: if self._public_key: raise ValueError("Public key cannot be used for key derivation.") if not public_key: diff --git a/cwt/algs/okp.py b/cwt/algs/okp.py index ccd762b1..57c5e368 100644 --- a/cwt/algs/okp.py +++ b/cwt/algs/okp.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, Optional, Union +from typing import Any import cryptography from cryptography.hazmat.primitives import hashes @@ -46,7 +46,7 @@ class OKPKey(AsymmetricKey): COSE_KEY_OPERATION_VALUES["deriveBits"], ] - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) self._public_key: Any = None self._private_key: Any = None @@ -90,16 +90,16 @@ def __init__(self, params: Dict[int, Any]): # Validate alg and key_ops. if self._key_ops: - if set(self._key_ops) & set([3, 4, 5, 6, 9, 10]): + if set(self._key_ops) & {3, 4, 5, 6, 9, 10}: raise ValueError("Unknown or not permissible key_ops(4) for OKP.") if self._alg: if self._alg in COSE_ALGORITHMS_SIG_OKP.values(): if self._key_ops: if -4 in params: # private key for signing. - if not (set(self._key_ops) & set([1, 2])): + if not (set(self._key_ops) & {1, 2}): raise ValueError("Invalid key_ops for signing key.") - if set(self._key_ops) & set([7, 8]): + if set(self._key_ops) & {7, 8}: raise ValueError("Signing key should not be used for key derivation.") else: # public key for signing. @@ -124,9 +124,9 @@ def __init__(self, params: Dict[int, Any]): if self._key_ops: if -4 in params: # private key for key derivation. - if not (set(self._key_ops) & set([7, 8])): + if not (set(self._key_ops) & {7, 8}): raise ValueError("Invalid key_ops for key derivation.") - if set(self._key_ops) & set([1, 2]): + if set(self._key_ops) & {1, 2}: raise ValueError("Private key for ECDHE should not be used for signing.") else: # public key for key derivation. @@ -140,17 +140,17 @@ def __init__(self, params: Dict[int, Any]): if self._crv in [4, 5]: # X25519/X448 if self._key_ops: # private key for key derivation. - if not (set(self._key_ops) & set([7, 8])): + if not (set(self._key_ops) & {7, 8}): raise ValueError("Invalid key_ops for X25519/448 private key.") - if set(self._key_ops) & set([1, 2]): + if set(self._key_ops) & {1, 2}: raise ValueError("Invalid key_ops for X25519/448 private key.") else: self._key_ops = [7, 8] else: # Ed25519/Ed448 if self._key_ops: - if not (set(self._key_ops) & set([1, 2])): + if not (set(self._key_ops) & {1, 2}): raise ValueError("Invalid key_ops for Ed25519/448 private key.") - if set(self._key_ops) & set([7, 8]): + if set(self._key_ops) & {7, 8}: raise ValueError("Invalid key_ops for Ed25519/448 private key.") else: self._key_ops = [1, 2] @@ -213,18 +213,18 @@ def __init__(self, params: Dict[int, Any]): @staticmethod def to_cose_key( - k: Union[ - Ed448PrivateKey, - Ed448PublicKey, - Ed25519PrivateKey, - Ed25519PublicKey, - X448PrivateKey, - X448PublicKey, - X25519PrivateKey, - X25519PublicKey, - ], - ) -> Dict[int, Any]: - cose_key: Dict[int, Any] = {} + k: ( + Ed448PrivateKey + | Ed448PublicKey + | Ed25519PrivateKey + | Ed25519PublicKey + | X448PrivateKey + | X448PublicKey + | X25519PrivateKey + | X25519PublicKey + ), + ) -> dict[int, Any]: + cose_key: dict[int, Any] = {} cose_key[1] = COSE_KEY_TYPES["OKP"] cose_key[-1] = 6 # Ed25519 @@ -263,23 +263,23 @@ def to_cose_key( @property def key( self, - ) -> Union[ - Ed448PrivateKey, - Ed448PublicKey, - Ed25519PrivateKey, - Ed25519PublicKey, - X448PrivateKey, - X448PublicKey, - X25519PrivateKey, - X25519PublicKey, - ]: + ) -> ( + Ed448PrivateKey + | Ed448PublicKey + | Ed25519PrivateKey + | Ed25519PublicKey + | X448PrivateKey + | X448PublicKey + | X25519PrivateKey + | X25519PublicKey + ): return self._key @property def crv(self) -> int: return self._crv - def to_dict(self) -> Dict[int, Any]: + def to_dict(self) -> dict[int, Any]: res = super().to_dict() res[-1] = self._crv res[-2] = self._x @@ -304,7 +304,7 @@ def verify(self, msg: bytes, sig: bytes): except cryptography.exceptions.InvalidSignature as err: raise VerifyError("Failed to verify.") from err - def derive_bytes(self, length: int, material: bytes = b"", info: bytes = b"", public_key: Optional[Any] = None) -> bytes: + def derive_bytes(self, length: int, material: bytes = b"", info: bytes = b"", public_key: Any | None = None) -> bytes: if self._public_key: raise ValueError("Public key cannot be used for key derivation.") if not public_key: diff --git a/cwt/algs/raw.py b/cwt/algs/raw.py index d7726176..2403876e 100644 --- a/cwt/algs/raw.py +++ b/cwt/algs/raw.py @@ -1,10 +1,10 @@ -from typing import Any, Dict +from typing import Any from ..cose_key_interface import COSEKeyInterface class RawKey(COSEKeyInterface): - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) self._key: bytes = b"" @@ -28,7 +28,7 @@ def key(self) -> bytes: def to_bytes(self) -> bytes: return self._key - def to_dict(self) -> Dict[int, Any]: + def to_dict(self) -> dict[int, Any]: res = super().to_dict() res[-1] = self._key return res diff --git a/cwt/algs/rsa.py b/cwt/algs/rsa.py index be9048ea..afaaa34e 100644 --- a/cwt/algs/rsa.py +++ b/cwt/algs/rsa.py @@ -1,4 +1,4 @@ -from typing import Any, Dict +from typing import Any from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding @@ -23,7 +23,7 @@ class RSAKey(AsymmetricKey): COSE_KEY_OPERATION_VALUES["verify"], ] - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) self._key: Any = None @@ -115,7 +115,7 @@ def __init__(self, params: Dict[int, Any]): self._key = private_numbers.private_key() return - def to_dict(self) -> Dict[int, Any]: + def to_dict(self) -> dict[int, Any]: return self._dict def sign(self, msg: bytes) -> bytes: diff --git a/cwt/algs/symmetric.py b/cwt/algs/symmetric.py index 07e7dee4..1ae0bf72 100644 --- a/cwt/algs/symmetric.py +++ b/cwt/algs/symmetric.py @@ -1,7 +1,7 @@ import hashlib import hmac from secrets import token_bytes -from typing import Any, Dict, Optional +from typing import Any from cryptography.hazmat.primitives.ciphers.aead import AESCCM, AESGCM, ChaCha20Poly1305 from cryptography.hazmat.primitives.keywrap import aes_key_unwrap, aes_key_wrap @@ -20,7 +20,7 @@ class SymmetricKey(COSEKeyInterface): - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) self._key: bytes = b"" @@ -46,7 +46,7 @@ def key(self) -> bytes: def to_bytes(self) -> bytes: return self._key - def to_dict(self) -> Dict[int, Any]: + def to_dict(self) -> dict[int, Any]: res = super().to_dict() res[-1] = self._key return res @@ -58,7 +58,7 @@ class MACAuthenticationKey(SymmetricKey): COSE_KEY_OPERATION_VALUES["MAC verify"], ] - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) # Validate key_opt. @@ -78,7 +78,7 @@ class ContentEncryptionKey(SymmetricKey): COSE_KEY_OPERATION_VALUES["unwrap key"], ] - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) # Validate key_opt. @@ -93,7 +93,7 @@ def __init__(self, params: Dict[int, Any]): class HMACKey(MACAuthenticationKey): """ """ - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): """ """ super().__init__(params) @@ -139,7 +139,7 @@ def verify(self, msg: bytes, sig: bytes): class AESCCMKey(ContentEncryptionKey): """ """ - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): """ """ super().__init__(params) @@ -209,7 +209,7 @@ def __init__(self, params: Dict[int, Any]): def generate_nonce(self): return token_bytes(self._nonce_len) - def encrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def encrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ if len(nonce) != self._nonce_len: raise ValueError("The length of nonce should be %d bytes." % self._nonce_len) @@ -218,7 +218,7 @@ def encrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> byte except Exception as err: raise EncodeError("Failed to encrypt.") from err - def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def decrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ if len(nonce) != self._nonce_len: raise ValueError("The length of nonce should be %d bytes." % self._nonce_len) @@ -231,7 +231,7 @@ def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> byte class AESGCMKey(ContentEncryptionKey): """ """ - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): """ """ super().__init__(params) @@ -262,14 +262,14 @@ def __init__(self, params: Dict[int, Any]): def generate_nonce(self): return token_bytes(_CWT_NONCE_SIZE_AESGCM) - def encrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def encrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ try: return self._cipher.encrypt(nonce, msg, aad) except Exception as err: raise EncodeError("Failed to encrypt.") from err - def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def decrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ try: return self._cipher.decrypt(nonce, msg, aad) @@ -278,7 +278,7 @@ def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> byte class ChaCha20Key(ContentEncryptionKey): - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) # Validate alg. @@ -295,13 +295,13 @@ def __init__(self, params: Dict[int, Any]): def generate_nonce(self): return token_bytes(_CWT_NONCE_SIZE_CHACHA20_POLY1305) - def encrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def encrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: try: return self._cipher.encrypt(nonce, msg, aad) except Exception as err: raise EncodeError("Failed to encrypt.") from err - def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def decrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: try: return self._cipher.decrypt(nonce, msg, aad) except Exception as err: @@ -314,7 +314,7 @@ class AESKeyWrap(SymmetricKey): COSE_KEY_OPERATION_VALUES["unwrapKey"], ] - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): super().__init__(params) # Validate alg. @@ -360,7 +360,7 @@ def unwrap_key(self, wrapped_key: bytes) -> bytes: class AESCTRKey(ContentEncryptionKey): """ """ - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): """ """ super().__init__(params) @@ -391,14 +391,14 @@ def __init__(self, params: Dict[int, Any]): def generate_nonce(self): return token_bytes(_CWT_NONCE_SIZE_AES) - def encrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def encrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ try: return self._cipher.encrypt(nonce, msg) except Exception as err: raise EncodeError("Failed to encrypt.") from err - def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def decrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ try: return self._cipher.decrypt(nonce, msg) @@ -409,7 +409,7 @@ def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> byte class AESCBCKey(ContentEncryptionKey): """ """ - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): """ """ super().__init__(params) @@ -440,7 +440,7 @@ def __init__(self, params: Dict[int, Any]): def generate_nonce(self): return token_bytes(_CWT_NONCE_SIZE_AES) - def encrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def encrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ try: # Add padding (see RFC 9459 and 5652) @@ -451,7 +451,7 @@ def encrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> byte except Exception as err: raise EncodeError("Failed to encrypt.") from err - def decrypt(self, msg: bytes, nonce: bytes, aad: Optional[bytes] = None) -> bytes: + def decrypt(self, msg: bytes, nonce: bytes, aad: bytes | None = None) -> bytes: """ """ try: decrypted = self._cipher.decrypt(nonce, msg) diff --git a/cwt/cbor_processor.py b/cwt/cbor_processor.py index 67118a46..db58d3c0 100644 --- a/cwt/cbor_processor.py +++ b/cwt/cbor_processor.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, Union +from typing import Any from cbor2 import dumps, loads @@ -12,7 +12,7 @@ def _dumps(self, obj: Any) -> bytes: except Exception as err: raise EncodeError("Failed to encode.") from err - def _loads(self, s: bytes) -> Dict[Union[str, int], Any]: + def _loads(self, s: bytes) -> dict[str | int, Any]: try: return loads(s) except Exception as err: diff --git a/cwt/claims.py b/cwt/claims.py index b77350d9..8ff96a55 100644 --- a/cwt/claims.py +++ b/cwt/claims.py @@ -1,5 +1,5 @@ import json -from typing import Any, Dict, List, Self, Union +from typing import Any, Self from .const import CWT_CLAIM_NAMES from .cose_key import COSEKey @@ -12,8 +12,8 @@ class Claims: def __init__( self, - claims: Dict[int, Any], - claim_names: Dict[str, int] = CWT_CLAIM_NAMES, + claims: dict[int, Any], + claim_names: dict[str, int] = CWT_CLAIM_NAMES, ): if -260 in claims and not isinstance(claims[-260], dict): raise ValueError("hcert(-260) should be map.") @@ -61,13 +61,13 @@ def __init__( return @classmethod - def new(cls, claims: Dict[int, Any], private_claim_names: Dict[str, int] = {}) -> Self: + 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. + Creates a Claims object from a CBOR-like(dict[int, Any]) claim object. Args: - claims (Dict[str, Any]): A CBOR-like(Dict[int, Any]) claim object. - private_claim_names (Dict[str, int]): A set of private claim definitions which + claims (dict[str, Any]): A CBOR-like(dict[int, Any]) claim object. + private_claim_names (dict[str, int]): A set of private claim definitions which consist of a readable claim name(str) and a claim key(int). The claim key should be less than -65536 but you can use the numbers other than pre-registered numbers listed in @@ -90,8 +90,8 @@ def new(cls, claims: Dict[int, Any], private_claim_names: Dict[str, int] = {}) - @classmethod def from_json( cls, - claims: Union[str, bytes, Dict[str, Any]], - private_claim_names: Dict[str, int] = {}, + claims: 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 @@ -99,9 +99,9 @@ def from_json( it will be skipped. Args: - claims (Union[str, bytes, Dict[str, Any]]): A JWT claims object + claims (str | bytes | dict[str, Any]): A JWT claims object to be converted. - private_claim_names (Dict[str, int]): A set of private claim definitions which + private_claim_names (dict[str, int]): A set of private claim definitions which consist of a readable claim name(str) and a claim key(int). The claim key should be less than -65536 but you can use the numbers other than pre-registered numbers listed in @@ -113,7 +113,7 @@ def from_json( Raises: ValueError: Invalid arguments. """ - json_claims: Dict[str, Any] = {} + json_claims: dict[str, Any] = {} if isinstance(claims, str) or isinstance(claims, bytes): json_claims = json.loads(claims) else: @@ -125,7 +125,7 @@ def from_json( raise ValueError("It is already CBOR-like format.") # Convert JSON to CBOR (Convert the type of key from str to int). - cbor_claims: Dict[int, Any] = {} + cbor_claims: dict[int, Any] = {} for k, v in json_claims.items(): if k not in CWT_CLAIM_NAMES: if k in private_claim_names: @@ -155,12 +155,12 @@ def from_json( return cls.new(cbor_claims, private_claim_names) @classmethod - def validate(cls, claims: Dict[int, Any]): + def validate(cls, claims: dict[int, Any]): """ Validates a CWT claims object. Args: - claims (Dict[int, Any]): A CWT claims object to be validated. + claims (dict[int, Any]): A CWT claims object to be validated. Raises: ValueError: Failed to verify. @@ -169,58 +169,58 @@ def validate(cls, claims: Dict[int, Any]): return @property - def iss(self) -> Union[str, None]: + def iss(self) -> str | None: return self._claims.get(1, None) @property - def sub(self) -> Union[str, None]: + def sub(self) -> str | None: return self._claims.get(2, None) @property - def aud(self) -> Union[str, None]: + def aud(self) -> str | None: return self._claims.get(3, None) @property - def exp(self) -> Union[int, None]: + def exp(self) -> int | None: return self._claims.get(4, None) @property - def nbf(self) -> Union[int, None]: + def nbf(self) -> int | None: return self._claims.get(5, None) @property - def iat(self) -> Union[int, None]: + def iat(self) -> int | None: return self._claims.get(6, None) @property - def cti(self) -> Union[str, None]: + def cti(self) -> str | None: if 7 not in self._claims: return None return self._claims[7].decode("utf-8") @property - def hcert(self) -> Union[dict, None]: + def hcert(self) -> dict | None: return self._claims.get(-260, None) @property - def cnf(self) -> Union[Dict[int, Any], List[Any], str, None]: + def cnf(self) -> dict[int, Any] | list[Any] | str | None: if 8 not in self._claims: return None if 1 in self._claims[8]: - key: Dict[int, Any] = self._claims[8][1] + key: dict[int, Any] = self._claims[8][1] return key if 2 in self._claims[8]: - eck: List[Any] = self._claims[8][2] + eck: list[Any] = self._claims[8][2] return eck kid: bytes = self._claims[8][3] return kid.decode("utf-8") - def get(self, key: Union[str, int]) -> Any: + def get(self, key: str | int) -> Any: """ Gets a claim value with a claim key. Args: - key (Union[str, int]): A claim key. + key (str | int): A claim key. Returns: Any: The value of the claim. """ @@ -231,7 +231,7 @@ def get(self, key: Union[str, int]) -> Any: int_key = key return self._claims.get(int_key, None) if int_key != 0 else None - def to_dict(self) -> Dict[int, Any]: + def to_dict(self) -> dict[int, Any]: """ Returns a raw claim object. diff --git a/cwt/cose.py b/cwt/cose.py index b00e08a3..99076153 100644 --- a/cwt/cose.py +++ b/cwt/cose.py @@ -1,5 +1,5 @@ from collections.abc import Mapping -from typing import Any, Dict, List, Optional, Self, Tuple, Union +from typing import Any, Self from cbor2 import CBORTag from cryptography.x509 import load_pem_x509_certificates @@ -139,15 +139,15 @@ def verify_kid(self, verify_kid: bool): def encode( self, payload: bytes, - key: Optional[COSEKeyInterface] = None, - protected: Optional[Union[dict, ResolvedHeader]] = None, - unprotected: Optional[Union[dict, ResolvedHeader]] = None, - recipients: List[RecipientInterface] = [], - signers: List[Signer] = [], + key: COSEKeyInterface | None = None, + protected: dict | ResolvedHeader | None = None, + unprotected: dict | ResolvedHeader | None = None, + recipients: list[RecipientInterface] = [], + signers: list[Signer] = [], external_aad: bytes = b"", out: str = "", enable_non_aead: bool = False, - hpke_psk: Optional[bytes] = None, + hpke_psk: bytes | None = None, hpke_info: bytes = b"", ) -> bytes: """ @@ -155,11 +155,11 @@ def encode( Args: payload (bytes): A content to be MACed, signed or encrypted. - key (Optional[COSEKeyInterface]): A content encryption key as COSEKey. - protected (Optional[Union[dict, ResolvedHeader]]): Parameters that are to be cryptographically protected. - unprotected (Optional[Union[dict, ResolvedHeader]]): Parameters that are not cryptographically protected. - recipients (List[RecipientInterface]): A list of recipient information structures. - signers (List[Signer]): A list of signer information objects for + key (COSEKeyInterface | None): A content encryption key as COSEKey. + protected (dict | ResolvedHeader | None): Parameters that are to be cryptographically protected. + unprotected (dict | ResolvedHeader | None): Parameters that are not cryptographically protected. + recipients (list[RecipientInterface]): A list of recipient information structures. + signers (list[Signer]): A list of signer information objects for multiple signer cases. external_aad(bytes): External additional authenticated data supplied by application. @@ -176,7 +176,7 @@ def encode( in conjunction with an authentication and integrity mechanisms, such as a digital signature. Returns: - Union[bytes, CBORTag]: A byte string of the encoded COSE or a + bytes | CBORTag: A byte string of the encoded COSE or a cbor2.CBORTag object. Raises: ValueError: Invalid arguments. @@ -194,14 +194,14 @@ def encode( def encode_and_encrypt( self, payload: bytes, - key: Optional[COSEKeyInterface] = None, - protected: Optional[dict] = None, - unprotected: Optional[dict] = None, - recipients: List[RecipientInterface] = [], + key: COSEKeyInterface | None = None, + protected: dict | None = None, + unprotected: dict | None = None, + recipients: list[RecipientInterface] = [], external_aad: bytes = b"", out: str = "", enable_non_aead: bool = False, - hpke_psk: Optional[bytes] = None, + hpke_psk: bytes | None = None, hpke_info: bytes = b"", ) -> bytes: """ @@ -209,10 +209,10 @@ def encode_and_encrypt( Args: payload (bytes): A content to be encrypted. - key (Optional[COSEKeyInterface]): A content encryption key as COSEKey. - protected (Optional[dict]): Parameters that are to be cryptographically protected. - unprotected (Optional[dict]): Parameters that are not cryptographically protected. - recipients (List[RecipientInterface]): A list of recipient information structures. + key (COSEKeyInterface | None): A content encryption key as COSEKey. + protected (dict | None): Parameters that are to be cryptographically protected. + unprotected (dict | None): Parameters that are not cryptographically protected. + recipients (list[RecipientInterface]): A list of recipient information structures. external_aad(bytes): External additional authenticated data supplied by application. out(str): An output format. Only ``"cbor2/CBORTag"`` can be used. If @@ -228,7 +228,7 @@ def encode_and_encrypt( in conjunction with an authentication and integrity mechanisms, such as a digital signature. Returns: - Union[bytes, CBORTag]: A byte string of the encoded COSE or a + bytes | CBORTag: A byte string of the encoded COSE or a cbor2.CBORTag object. Raises: ValueError: Invalid arguments. @@ -243,29 +243,29 @@ def encode_and_encrypt( def encode_and_mac( self, payload: bytes, - key: Optional[COSEKeyInterface] = None, - protected: Optional[dict] = None, - unprotected: Optional[dict] = None, - recipients: List[RecipientInterface] = [], + key: COSEKeyInterface | None = None, + protected: dict | None = None, + unprotected: dict | None = None, + recipients: list[RecipientInterface] = [], external_aad: bytes = b"", out: str = "", - ) -> Union[bytes, CBORTag]: + ) -> bytes | CBORTag: """ Encodes data with MAC. Args: payload (bytes): A content to be MACed. key (COSEKeyInterface): A COSE key as a MAC Authentication key. - protected (Optional[dict]): Parameters that are to be cryptographically protected. - unprotected (Optional[dict]): Parameters that are not cryptographically protected. - recipients (List[RecipientInterface]): A list of recipient information structures. + protected (dict | None): Parameters that are to be cryptographically protected. + unprotected (dict | None): Parameters that are not cryptographically protected. + recipients (list[RecipientInterface]): A list of recipient information structures. external_aad(bytes): External additional authenticated data supplied by application. out(str): An output format. Only ``"cbor2/CBORTag"`` can be used. If ``"cbor2/CBORTag"`` is specified. This function will return encoded data as `cbor2 `_'s ``CBORTag`` object. If any other value is specified, it will return encoded data as bytes. Returns: - Union[bytes, CBORTag]: A byte string of the encoded COSE or a cbor2.CBORTag object. + bytes | CBORTag: A byte string of the encoded COSE or a cbor2.CBORTag object. Raises: ValueError: Invalid arguments. EncodeError: Failed to encode data. @@ -279,28 +279,28 @@ def encode_and_mac( def encode_and_sign( self, payload: bytes, - key: Optional[COSEKeyInterface] = None, - protected: Optional[Union[Dict[Any, Any], ResolvedHeader]] = None, - unprotected: Optional[Union[Dict[Any, Any], ResolvedHeader]] = None, - signers: List[Signer] = [], + key: COSEKeyInterface | None = None, + protected: dict[Any, Any] | ResolvedHeader | None = None, + unprotected: dict[Any, Any] | ResolvedHeader | None = None, + signers: list[Signer] = [], external_aad: bytes = b"", out: str = "", - ) -> Union[bytes, CBORTag]: + ) -> bytes | CBORTag: """ Encodes data with signing. Args: payload (bytes): A content to be signed. - key (Optional[COSEKeyInterface]): A signing key for single signer + key (COSEKeyInterface | None): A signing key for single signer cases. When the ``signers`` parameter is set, this ``key`` will be ignored and should not be set. - protected (Optional[Union[Dict[Any, Any], ResolvedHeader]]): Parameters that are + protected (dict[Any, Any] | ResolvedHeader | None): Parameters that are to be cryptographically protected. Use ``ResolvedHeader`` to bypass named-header resolution and value conversion. - unprotected (Optional[Union[Dict[Any, Any], ResolvedHeader]]): Parameters that are + unprotected (dict[Any, Any] | ResolvedHeader | None): Parameters that are not cryptographically protected. Use ``ResolvedHeader`` to bypass named-header resolution and value conversion. - signers (List[Signer]): A list of signer information objects for + signers (list[Signer]): A list of signer information objects for multiple signer cases. external_aad(bytes): External additional authenticated data supplied by application. @@ -310,7 +310,7 @@ def encode_and_sign( ``CBORTag`` object. If any other value is specified, it will return encoded data as bytes. Returns: - Union[bytes, CBORTag]: A byte string of the encoded COSE or a + bytes | CBORTag: A byte string of the encoded COSE or a cbor2.CBORTag object. Raises: ValueError: Invalid arguments. @@ -324,13 +324,13 @@ def encode_and_sign( def decode( self, - data: Union[bytes, CBORTag], - keys: Union[COSEKeyInterface, List[COSEKeyInterface]], - context: Optional[Union[Dict[str, Any], List[Any]]] = None, + data: bytes | CBORTag, + keys: COSEKeyInterface | list[COSEKeyInterface], + context: dict[str, Any] | list[Any] | None = None, external_aad: bytes = b"", - detached_payload: Optional[bytes] = None, + detached_payload: bytes | None = None, enable_non_aead: bool = False, - hpke_psk: Optional[bytes] = None, + hpke_psk: bytes | None = None, hpke_info: bytes = b"", hpke_aad: bytes = b"", extra_info: bytes = b"", @@ -339,15 +339,15 @@ def decode( Verifies and decodes COSE data, and returns only payload. Args: - data (Union[bytes, CBORTag]): A byte string or cbor2.CBORTag of an + data (bytes | CBORTag): A byte string or cbor2.CBORTag of an encoded data. - keys (Union[COSEKeyInterface, List[COSEKeyInterface]]): COSE key(s) + keys (COSEKeyInterface | list[COSEKeyInterface]): COSE key(s) to verify and decrypt the encoded data. - context (Optional[Union[Dict[str, Any], List[Any]]]): A context information + context (dict[str, Any] | list[Any] | None): A context information structure for key deriviation functions. external_aad(bytes): External additional authenticated data supplied by application. - detached_payload (Optional[bytes]): The detached payload that should be verified with data. + detached_payload (bytes | None): The detached payload that should be verified with data. enable_non_aead (bool): Enable non-AEAD content ecnryption algorithms (False = disabled by default). Before enable non-AEAD ciphers, read and understand Security considerations of RFC 9459 carefully. @@ -367,37 +367,37 @@ def decode( def decode_with_headers( self, - data: Union[bytes, CBORTag], - keys: Union[COSEKeyInterface, List[COSEKeyInterface]], - context: Optional[Union[Dict[str, Any], List[Any]]] = None, + data: bytes | CBORTag, + keys: COSEKeyInterface | list[COSEKeyInterface], + context: dict[str, Any] | list[Any] | None = None, external_aad: bytes = b"", - detached_payload: Optional[bytes] = None, + detached_payload: bytes | None = None, enable_non_aead: bool = False, - hpke_psk: Optional[bytes] = None, + hpke_psk: bytes | None = None, hpke_info: bytes = b"", hpke_aad: bytes = b"", extra_info: bytes = b"", - ) -> Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any], bytes]: + ) -> tuple[dict[str | int, Any], dict[str | int, Any], bytes]: """ Verifies and decodes COSE data, and returns protected headers, unprotected headers and payload. Args: - data (Union[bytes, CBORTag]): A byte string or cbor2.CBORTag of an + data (bytes | CBORTag): A byte string or cbor2.CBORTag of an encoded data. - keys (Union[COSEKeyInterface, List[COSEKeyInterface]]): COSE key(s) + keys (COSEKeyInterface | list[COSEKeyInterface]): COSE key(s) to verify and decrypt the encoded data. - context (Optional[Union[Dict[str, Any], List[Any]]]): A context information + context (dict[str, Any] | list[Any] | None): A context information structure for key deriviation functions. external_aad(bytes): External additional authenticated data supplied by application. - detached_payload (Optional[bytes]): The detached payload that should be verified with data. + detached_payload (bytes | None): The detached payload that should be verified with data. enable_non_aead (bool): Enable non-AEAD content ecnryption algorithms (False = disabled by default). Before enable non-AEAD ciphers, read and understand Security considerations of RFC 9459 carefully. Since non-AEAD ciphers DO NOT provide neither authentication nor integrity of decrypted message, make sure to validate them outside of this library. Returns: - Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any], bytes]: A dictionary data of decoded protected headers, and a dictionary data of unprotected headers, and a byte string of decoded payload. + tuple[dict[str | int, Any], dict[str | int, Any], bytes]: A dictionary data of decoded protected headers, and a dictionary data of unprotected headers, and a byte string of decoded payload. Raises: ValueError: Invalid arguments. DecodeError: Failed to decode data. @@ -449,7 +449,7 @@ def decode_with_headers( if payload is None: raise ValueError("detached_payload should be set.") - # protected: Union[Dict[int, Any], bytes] = self._loads(value[0]) if value[0] else b"" + # protected: dict[int, Any] | bytes = self._loads(value[0]) if value[0] else b"" # unprotected = value[1] # if not isinstance(unprotected, dict): # raise ValueError("unprotected header should be dict.") @@ -619,11 +619,11 @@ def decode_with_headers( def _encode_headers( self, - key: Optional[COSEKeyInterface], - protected: Optional[Union[dict, ResolvedHeader]], - unprotected: Optional[Union[dict, ResolvedHeader]], + key: COSEKeyInterface | None, + protected: dict | ResolvedHeader | None, + unprotected: dict | ResolvedHeader | None, enable_non_aead: bool, - ) -> Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any]]: + ) -> tuple[dict[str | int, Any], dict[str | int, Any]]: p = to_cose_header(protected) u = to_cose_header(unprotected) if key is not None: @@ -650,10 +650,8 @@ def _encode_headers( raise ValueError("protected header MUST be zero-length") return p, u - def _decode_headers( - self, protected: Any, unprotected: Any - ) -> Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any]]: - p: Union[Dict[Union[str, int], Any], bytes] + def _decode_headers(self, protected: Any, unprotected: Any) -> tuple[dict[str | int, Any], dict[str | int, Any]]: + p: dict[str | int, Any] | bytes p = self._loads(protected) if protected else {} if isinstance(p, bytes): if len(p) > 0: @@ -661,21 +659,21 @@ def _decode_headers( p = {} if not isinstance(unprotected, Mapping): raise ValueError("unprotected header should be dict.") - u: Dict[Union[str, int], Any] = to_mutable(unprotected) + u: dict[str | int, Any] = to_mutable(unprotected) return p, u def _validate_cose_message( self, - key: Optional[COSEKeyInterface], - p: Dict[Union[str, int], Any], - u: Dict[Union[str, int], Any], - recipients: List[RecipientInterface], - signers: List[Signer], + key: COSEKeyInterface | None, + p: dict[str | int, Any], + u: dict[str | int, Any], + recipients: list[RecipientInterface], + signers: list[Signer], ) -> int: if len(recipients) > 0 and len(signers) > 0: raise ValueError("Both recipients and signers are specified.") - h: Dict[Union[str, int], Any] = {} + h: dict[str | int, Any] = {} iv_count: int = 0 for k, v in p.items(): if k == 2: # crit @@ -795,13 +793,13 @@ def _validate_cose_message( def _encode_and_encrypt( self, payload: bytes, - key: Optional[COSEKeyInterface], - p: Dict[Union[str, int], Any], - u: Dict[Union[str, int], Any], - recipients: List[RecipientInterface], + key: COSEKeyInterface | None, + p: dict[str | int, Any], + u: dict[str | int, Any], + recipients: list[RecipientInterface], external_aad: bytes, out: str, - hpke_psk: Optional[bytes], + hpke_psk: bytes | None, hpke_info: bytes = b"", ) -> bytes: b_protected = self._dumps(p) if p else b"" @@ -837,7 +835,7 @@ def _encode_and_encrypt( content_alg = p.get(1, 0) or u.get(1, 0) recs = [] b_key = key.to_bytes() if isinstance(key, COSEKeyInterface) else b"" - cek: Optional[COSEKeyInterface] = None + cek: COSEKeyInterface | None = None for rec in recipients: if isinstance(rec, HPKE): rec.content_alg = content_alg @@ -856,7 +854,7 @@ def _encode_and_encrypt( enc_structure = ["Encrypt", b_protected, external_aad] aad = self._dumps(enc_structure) ciphertext = cek.encrypt(payload, u[5], aad) - cose_enc: List[Any] = [b_protected, u, ciphertext] + cose_enc: list[Any] = [b_protected, u, ciphertext] cose_enc.append(recs) res = CBORTag(96, cose_enc) return res if out == "cbor2/CBORTag" else self._dumps(res) @@ -864,13 +862,13 @@ def _encode_and_encrypt( def _encode_and_mac( self, payload: bytes, - key: Optional[COSEKeyInterface], - p: Dict[Union[str, int], Any], - u: Dict[Union[str, int], Any], - recipients: List[RecipientInterface], + key: COSEKeyInterface | None, + p: dict[str | int, Any], + u: dict[str | int, Any], + recipients: list[RecipientInterface], external_aad: bytes, out: str, - ) -> Union[bytes, CBORTag]: + ) -> bytes | CBORTag: b_protected = self._dumps(p) if p else b"" # MAC0 @@ -902,7 +900,7 @@ def _encode_and_mac( if key is None: raise ValueError("key should be set.") tag = key.sign(self._dumps(mac_structure)) - cose_mac: List[Any] = [b_protected, u, payload, tag] + cose_mac: list[Any] = [b_protected, u, payload, tag] cose_mac.append(recs) res = CBORTag(97, cose_mac) return res if out == "cbor2/CBORTag" else self._dumps(res) @@ -910,13 +908,13 @@ def _encode_and_mac( def _encode_and_sign( self, payload: bytes, - key: Optional[COSEKeyInterface], - p: Dict[Union[str, int], Any], - u: Dict[Union[str, int], Any], - signers: List[Signer], + key: COSEKeyInterface | None, + p: dict[str | int, Any], + u: dict[str | int, Any], + signers: list[Signer], external_aad: bytes, out: str, - ) -> Union[bytes, CBORTag]: + ) -> bytes | CBORTag: b_protected = self._dumps(p) if p else b"" # Signature1 @@ -935,8 +933,8 @@ def _encode_and_sign( res = CBORTag(98, [b_protected, u, payload, sigs]) return res if out == "cbor2/CBORTag" else self._dumps(res) - def _filter_by_key_ops(self, keys: List[COSEKeyInterface], op: int) -> List[COSEKeyInterface]: - res: List[COSEKeyInterface] = [] + def _filter_by_key_ops(self, keys: list[COSEKeyInterface], op: int) -> list[COSEKeyInterface]: + res: list[COSEKeyInterface] = [] for k in keys: if op in k.key_ops: res.append(k) diff --git a/cwt/cose_key.py b/cwt/cose_key.py index 35a3032e..80278739 100644 --- a/cwt/cose_key.py +++ b/cwt/cose_key.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Union +from typing import Any import cbor2 from cryptography import x509 @@ -45,12 +45,12 @@ class COSEKey: """ @staticmethod - def new(params: Dict[int, Any]) -> COSEKeyInterface: + def new(params: dict[int, Any]) -> COSEKeyInterface: """ Creates a COSE_Key object from a COSE_Key structure, which is a dictionary with numeric keys. Args: - params (Dict[int, Any]): A dictionary with numeric keys of a COSE key. + params (dict[int, Any]): A dictionary with numeric keys of a COSE key. Returns: COSEKeyInterface: A COSE key object. Raises: @@ -93,19 +93,19 @@ def new(params: Dict[int, Any]) -> COSEKeyInterface: @classmethod def generate_symmetric_key( cls, - alg: Union[int, str] = "", - kid: Union[bytes, str] = b"", - key_ops: Optional[Union[List[int], List[str]]] = None, + alg: int | str = "", + kid: bytes | str = b"", + key_ops: list[int] | list[str] | None = None, ) -> COSEKeyInterface: """ Generates a symmetric COSE key from from a randomly genarated byte string. Args: - alg (Union[int, str]): An algorithm label(int) or name(str). + alg (int | str): An algorithm label(int) or name(str). Supported ``alg`` are listed in `Supported COSE Algorithms `_. - kid (Union[bytes, str]): A key identifier. - key_ops (Union[List[int], List[str]]): A list of key operation values. + kid (bytes | str): A key identifier. + key_ops (list[int] | list[str]): A list of key operation values. Following values can be used: ``1("sign")``, ``2("verify")``, ``3("encrypt")``, ``4("decrypt")``, ``5("wrap key")``, ``6("unwrap key")``, ``7("derive key")``, ``8("derive bits")``, @@ -120,21 +120,21 @@ def generate_symmetric_key( @classmethod def from_symmetric_key( cls, - key: Union[bytes, str] = b"", - alg: Union[int, str] = "", - kid: Union[bytes, str] = b"", - key_ops: Optional[Union[List[int], List[str]]] = None, + key: bytes | str = b"", + alg: int | str = "", + kid: bytes | str = b"", + key_ops: list[int] | list[str] | None = None, ) -> COSEKeyInterface: """ Creates a COSE key from a symmetric key. Args: - key (Union[bytes, str]): A key bytes or string. - alg (Union[int, str]): An algorithm label(int) or name(str). + key (bytes | str): A key bytes or string. + alg (int | str): An algorithm label(int) or name(str). Supported ``alg`` are listed in `Supported COSE Algorithms `_. - kid (Union[bytes, str]): A key identifier. - key_ops (Union[List[int], List[str]]): A list of key operation values. + kid (bytes | str): A key identifier. + key_ops (list[int] | list[str]): A list of key operation values. Following values can be used: ``1("sign")``, ``2("verify")``, ``3("encrypt")``, ``4("decrypt")``, ``5("wrap key")``, ``6("unwrap key")``, ``7("derive key")``, ``8("derive bits")``, @@ -152,7 +152,7 @@ def from_symmetric_key( if alg_id == 0: raise ValueError(f"Unsupported or unknown alg(3): {alg}.") - params: Dict[int, Any] = { + params: dict[int, Any] = { COSEKeyParams.KTY: 4, # kty: 'Symmetric' COSEKeyParams.ALG: alg_id, # alg: int COSEKeyParams.K: key, # k: bstr @@ -162,7 +162,7 @@ def from_symmetric_key( if kid: params[2] = kid - key_ops_labels: List[int] = [] + key_ops_labels: list[int] = [] if key_ops and isinstance(key_ops, list): try: for ops in key_ops: @@ -192,12 +192,12 @@ def from_bytes(cls, key_data: bytes) -> COSEKeyInterface: return cls.new(params) @classmethod - def from_jwk(cls, data: Union[str, bytes, Dict[str, Any]]) -> COSEKeyInterface: + def from_jwk(cls, data: str | bytes | dict[str, Any]) -> COSEKeyInterface: """ Creates a COSE key from JWK (JSON Web Key). Args: - jwk (Union[str, bytes, Dict[str, Any]]): JWK-formatted key data. + jwk (str | bytes | dict[str, Any]): JWK-formatted key data. Returns: COSEKeyInterface: A COSE key object. Raises: @@ -209,21 +209,21 @@ def from_jwk(cls, data: Union[str, bytes, Dict[str, Any]]) -> COSEKeyInterface: @classmethod def from_pem( cls, - key_data: Union[str, bytes], - alg: Union[int, str] = "", - kid: Union[bytes, str] = b"", - key_ops: Optional[Union[List[int], List[str]]] = None, + key_data: str | bytes, + alg: int | str = "", + kid: bytes | str = b"", + key_ops: list[int] | list[str] | None = None, ) -> COSEKeyInterface: """ Creates a COSE key from PEM-formatted key data. Args: key_data (bytes): A PEM-formatted key data. - alg (Union[int, str]): An algorithm label(int) or name(str). + alg (int | str): An algorithm label(int) or name(str). Different from ::func::`cwt.COSEKey.from_symmetric_key`, it is only used when an algorithm cannot be specified by the PEM data, such as RSA family algorithms. - kid (Union[bytes, str]): A key identifier. - key_ops (Union[List[int], List[str]]): A list of key operation values. Following values can be used: + kid (bytes | str): A key identifier. + key_ops (list[int] | list[str]): A list of key operation values. Following values can be used: ``1("sign")``, ``2("verify")``, ``3("encrypt")``, ``4("decrypt")``, ``5("wrap key")``, ``6("unwrap key")``, ``7("derive key")``, ``8("derive bits")``, ``9("MAC create")``, ``10("MAC verify")`` @@ -248,13 +248,13 @@ def from_pem( else: raise ValueError("Failed to decode PEM.") - params: Dict[int, Any] = {} + params: dict[int, Any] = {} if isinstance(kid, str): kid = kid.encode("utf-8") if kid: params[2] = kid - key_ops_labels: List[int] = [] + key_ops_labels: list[int] = [] if key_ops and isinstance(key_ops, list): try: for ops in key_ops: diff --git a/cwt/cose_key_interface.py b/cwt/cose_key_interface.py index 5502f471..22fbc90e 100644 --- a/cwt/cose_key_interface.py +++ b/cwt/cose_key_interface.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Union +from typing import Any from .cbor_processor import CBORProcessor from .const import ( @@ -13,13 +13,13 @@ class COSEKeyInterface(CBORProcessor): The interface class for a COSE Key used for MAC, signing/verifying and encryption/decryption. """ - def __init__(self, params: Dict[int, Any]): + def __init__(self, params: dict[int, Any]): """ Constructor. Args: - params (Dict[int, Any]): A COSE key common parameter object formatted to CBOR-like - structure ((Dict[int, Any])). + params (dict[int, Any]): A COSE key common parameter object formatted to CBOR-like + structure ((dict[int, Any])). """ # kty @@ -50,7 +50,7 @@ def __init__(self, params: Dict[int, Any]): # key_ops if 4 in params and not isinstance(params[4], list): raise ValueError("key_ops(4) should be list.") - self._key_ops: List[int] = params[4] if 4 in params else [] + self._key_ops: list[int] = params[4] if 4 in params else [] for v in self._key_ops: if v not in COSE_KEY_OPERATION_VALUES.values(): raise ValueError(f"key_ops(4) includes invalid value: {v}.") @@ -69,28 +69,28 @@ def kty(self) -> int: return self._kty @property - def kid(self) -> Union[bytes, None]: + def kid(self) -> bytes | None: """ The key identifier. """ return self._kid @property - def alg(self) -> Union[int, None]: + def alg(self) -> int | None: """ The algorithm that is used with the key. """ return self._alg @property - def key_ops(self) -> List[int]: + def key_ops(self) -> list[int]: """ A set of permissible operations that the key is to be used for. """ return self._key_ops @property - def base_iv(self) -> Union[bytes, None]: + def base_iv(self) -> bytes | None: """ Base IV to be xor-ed with Partial IVs. """ @@ -117,14 +117,14 @@ def to_bytes(self) -> bytes: """ raise NotImplementedError - def to_dict(self) -> Dict[int, Any]: + def to_dict(self) -> dict[int, Any]: """ - Returns the CBOR-like structure (Dict[int, Any]) of the COSE key. + Returns the CBOR-like structure (dict[int, Any]) of the COSE key. Returns: - Dict[int, Any]: The CBOR-like structure of the COSE key. + dict[int, Any]: The CBOR-like structure of the COSE key. """ - res: Dict[int, Any] = {1: self._kty} + res: dict[int, Any] = {1: self._kty} if self._kid: res[2] = self._kid if self._alg: @@ -185,13 +185,13 @@ def verify(self, msg: bytes, sig: bytes): """ raise NotImplementedError - def validate_certificate(self, ca_certs: List[bytes]) -> bool: + def validate_certificate(self, ca_certs: list[bytes]) -> bool: """ Validate a certificate bound to the key with given trusted CA certificates if the key has `x5c` parameter. Args: - ca_certs(List[bytes]): A list of DER-formatted trusted root CA + ca_certs(list[bytes]): A list of DER-formatted trusted root CA certificates which contains a concatenated list of trusted root certificates. You should specify private CA certificates in your target system. There should be no need to use the public CA @@ -274,7 +274,7 @@ def derive_bytes( length: int, material: bytes = b"", info: bytes = b"", - public_key: Optional[Any] = None, + public_key: Any | None = None, ) -> bytes: """ Derives a byte string with a key material or key exchange. diff --git a/cwt/cose_message.py b/cwt/cose_message.py index 17dfd451..969b912f 100644 --- a/cwt/cose_message.py +++ b/cwt/cose_message.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Any, Dict, List, Optional, Self, Tuple, Union +from typing import Any, Self from cbor2 import CBORTag, loads @@ -18,13 +18,13 @@ class COSEMessage(CBORProcessor): The COSE message. """ - def __init__(self, type: COSETypes, msg: Union[List[Any], Tuple[Any, ...]]): + def __init__(self, type: COSETypes, msg: list[Any] | tuple[Any, ...]): """ Constructor. Args: - type (List[Any]): A type of the COSE message. - msg (Union[List[Any], Tuple[Any, ...]]): A COSE message as a CBOR array. + type (list[Any]): A type of the COSE message. + msg (list[Any] | tuple[Any, ...]): A COSE message as a CBOR array. """ self._validate_cose_message(msg) self._msg = msg if isinstance(msg, list) else list(msg) @@ -32,9 +32,9 @@ def __init__(self, type: COSETypes, msg: Union[List[Any], Tuple[Any, ...]]): self._protected = msg[0] self._unprotected = msg[1] # self._payload = msg[2] # msg[2] is mutable and has no readable alias to avoid complexity. - self._other_fields: List[bytes] = [] - self._recipients: List[List[Any]] = [] - self._signatures: List[List[Any]] = [] + self._other_fields: list[bytes] = [] + self._recipients: list[list[Any]] = [] + self._signatures: list[list[Any]] = [] if self._type == COSETypes.ENCRYPT0: if len(self._msg) != 3: @@ -125,11 +125,11 @@ def loads(cls, msg: bytes) -> Self: return cls(type, tagged.value) @classmethod - def from_cose_signature(cls, signature: Union[List[Any], Tuple[Any, ...]]) -> Self: + def from_cose_signature(cls, signature: list[Any] | tuple[Any, ...]) -> Self: return cls(COSETypes.SIGNATURE, signature) @classmethod - def from_cose_recipient(cls, recipient: Union[List[Any], Tuple[Any, ...]]) -> Self: + def from_cose_recipient(cls, recipient: list[Any] | tuple[Any, ...]) -> Self: return cls(COSETypes.RECIPIENT, recipient) @property @@ -140,14 +140,14 @@ def type(self) -> COSETypes: return self._type @property - def protected(self) -> Dict[Union[str, int], Any]: + def protected(self) -> dict[str | int, Any]: """ The protected headers as a CBOR object. """ return self._loads(self._protected) @property - def unprotected(self) -> Dict[Union[str, int], Any]: + def unprotected(self) -> dict[str | int, Any]: """ The unprotected headers as a CBOR object. """ @@ -161,21 +161,21 @@ def payload(self) -> bytes: return self._msg[2] @property - def other_fields(self) -> List[bytes]: + def other_fields(self) -> list[bytes]: """ The list of other fields of the COSE message. """ return self._other_fields @property - def signatures(self) -> List[List[Any]]: + def signatures(self) -> list[list[Any]]: """ The list of signatures of the COSE message. """ return self._signatures @property - def recipients(self) -> List[List[Any]]: + def recipients(self) -> list[list[Any]]: """ The list of recipients of the COSE message. """ @@ -194,7 +194,7 @@ def countersign( aad: bytes = b"", abbreviated: bool = False, tagged: bool = False, - detached_payload: Optional[bytes] = None, + detached_payload: bytes | None = None, ) -> COSEMessage: """ Countersigns to the COSE message with the signer specified. @@ -204,7 +204,7 @@ def countersign( aad (bytes): The application supplied additional authenticated data. abbreviated(bool): The type of the countersignature (abbreviated or not). tagged(bool): The indicator whether the countersignature is tagged or not. - detached_payload (Optional[bytes]): The detached payload that should be + detached_payload (bytes | None): The detached payload that should be countersigned with the COSEMessage. Returns: COSEMessage: The COSE message (self). @@ -255,18 +255,18 @@ def counterverify( self, key: COSEKeyInterface, aad: bytes = b"", - detached_payload: Optional[bytes] = None, - ) -> Optional[List[Any]]: + detached_payload: bytes | None = None, + ) -> list[Any] | None: """ Verifies a countersignature in the COSE message with the verification key specified. Args: key(COSEKeyInterface): A COSEKey that is used to verify a signature in the COSE message. aad (bytes): The application supplied additional authenticated data. - detached_payload (Optional[bytes]): The detached payload that should be + detached_payload (bytes | None): The detached payload that should be counterverified with the COSEMessage. Returns: - Optional[List[Any]]: The COSE signature verified. + list[Any] | None: The COSE signature verified. Raises: ValueError: Invalid arguments. VerifyError: Failed to verify. @@ -325,7 +325,7 @@ def counterverify( err = e raise err - def _validate_cose_message(self, msg: Union[List[Any], Tuple[Any, ...]]): + def _validate_cose_message(self, msg: list[Any] | tuple[Any, ...]): if not isinstance(msg, (list, tuple)) or len(msg) < 3: raise ValueError("Invalid COSE message.") if not isinstance(msg[0], bytes): @@ -350,16 +350,16 @@ def _validate_cose_message(self, msg: Union[List[Any], Tuple[Any, ...]]): self._validate_cose_message(cs) return - def _get_kid(self, sig: list) -> Optional[bytes]: + def _get_kid(self, sig: list) -> bytes | None: kid = sig[1].get(4, None) return kid if kid else self._loads(sig[0]).get(4, None) - def detach_payload(self) -> Tuple[COSEMessage, bytes]: + def detach_payload(self) -> tuple[COSEMessage, bytes]: """ Detach a payload from the COSE message Returns: - Tuple[COSEMessage, bytes]: The COSE message (self), + tuple[COSEMessage, bytes]: The COSE message (self), and a byte string of the detached payload. Raises: ValueError: The payload does not exist. diff --git a/cwt/cwt.py b/cwt/cwt.py index 4689bdce..1e799688 100644 --- a/cwt/cwt.py +++ b/cwt/cwt.py @@ -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, Self, Tuple, Union +from typing import Any, Self from cbor2 import CBORTag, dumps @@ -53,7 +53,7 @@ def __init__( verify_kid=True, ca_certs=ca_certs, ) - self._claim_names: Dict[str, int] = {} + self._claim_names: dict[str, int] = {} @classmethod def new( @@ -134,11 +134,11 @@ def cose(self) -> COSE: def encode( self, - claims: Union[Claims, Dict[str, Any], Dict[int, Any], bytes], + claims: Claims | dict[str, Any] | dict[int, Any] | bytes, key: COSEKeyInterface, nonce: bytes = b"", - recipients: List[RecipientInterface] = [], - signers: List[Signer] = [], + recipients: list[RecipientInterface] = [], + signers: list[Signer] = [], tagged: bool = False, ) -> bytes: """ @@ -154,12 +154,12 @@ def encode( parameter set to identify the usage. Args: - claims (Union[Claims, Dict[str, Any], Dict[int, Any], bytes]): A CWT + claims (Claims | dict[str, Any] | dict[int, Any] | bytes): A CWT claims object, or a JWT claims object, text string or byte string. key (COSEKeyInterface): A COSE key used to generate a MAC for the claims. nonce (bytes): A nonce for encryption. - recipients (List[RecipientInterface]): A list of recipient information structures. - signers (List[Signer]): A list of signer information structures for + recipients (list[RecipientInterface]): A list of recipient information structures. + signers (list[Signer]): A list of signer information structures for multiple signer cases. tagged (bool): An indicator whether the response is wrapped by CWT tag(61) or not. @@ -185,7 +185,7 @@ def encode( # claims = Claims.from_json(claims) # break # To avoid the error: - json_claims: Dict[str, Any] = {} + json_claims: dict[str, Any] = {} for k, v in claims.items(): if isinstance(k, str): json_claims[k] = v @@ -195,19 +195,19 @@ def encode( def encode_and_mac( self, - claims: Union[Claims, Dict[int, Any], bytes], + claims: Claims | dict[int, Any] | bytes, key: COSEKeyInterface, - recipients: List[RecipientInterface] = [], + recipients: list[RecipientInterface] = [], tagged: bool = False, ) -> bytes: """ Encodes with MAC. Args: - claims (Union[Claims, Dict[int, Any], bytes]): A CWT claims object or byte + claims (Claims | dict[int, Any] | bytes): A CWT claims object or byte string. key (COSEKeyInterface): A COSE key used to generate a MAC for the claims. - recipients (Optional[List[RecipientInterface]]): A list of recipient + recipients (list[RecipientInterface] | None): A list of recipient information structures. tagged (bool): An indicator whether the response is wrapped by CWT tag(61) or not. @@ -230,21 +230,21 @@ def encode_and_mac( def encode_and_sign( self, - claims: Union[Claims, Dict[int, Any], bytes], - key: Optional[COSEKeyInterface] = None, - signers: List[Signer] = [], + claims: Claims | dict[int, Any] | bytes, + key: COSEKeyInterface | None = None, + signers: list[Signer] = [], tagged: bool = False, ) -> bytes: """ Encodes CWT with signing. Args: - claims (Claims, Union[Dict[int, Any], bytes]): A CWT claims object or + claims (Claims, dict[int, Any] | bytes): A CWT claims object or byte string. - key (Optional[COSEKeyInterface]): A COSE key or a list of the keys + key (COSEKeyInterface | None): A COSE key or a list of the keys used to sign claims. When the ``signers`` parameter is set, this ``key`` parameter will be ignored and should not be set. - signers (List[Signer]): A list of signer information structures for + signers (list[Signer]): A list of signer information structures for multiple signer cases. tagged (bool): An indicator whether the response is wrapped by CWT tag(61) or not. @@ -267,21 +267,21 @@ def encode_and_sign( def encode_and_encrypt( self, - claims: Union[Claims, Dict[int, Any], bytes], + claims: Claims | dict[int, Any] | bytes, key: COSEKeyInterface, nonce: bytes = b"", - recipients: List[RecipientInterface] = [], + recipients: list[RecipientInterface] = [], tagged: bool = False, ) -> bytes: """ Encodes CWT with encryption. Args: - claims (Claims, Union[Dict[int, Any], bytes]): A CWT claims object or + claims (Claims, dict[int, Any] | bytes): A CWT claims object or byte string. key (COSEKeyInterface): A COSE key used to encrypt the claims. nonce (bytes): A nonce for encryption. - recipients (List[RecipientInterface]): A list of recipient information + recipients (list[RecipientInterface]): A list of recipient information structures. tagged (bool): An indicator whether the response is wrapped by CWT tag(61) or not. @@ -316,9 +316,9 @@ def encode_and_encrypt( def decode( self, data: bytes, - keys: Union[COSEKeyInterface, List[COSEKeyInterface]], + keys: COSEKeyInterface | list[COSEKeyInterface], no_verify: bool = False, - ) -> Union[Dict[Union[str, int], Any], bytes]: + ) -> dict[str | int, Any] | bytes: """ Verifies and decodes CWT. @@ -330,22 +330,22 @@ def decode( Args: data (bytes): A byte string of an encoded CWT. - keys (Union[COSEKeyInterface, List[COSEKeyInterface]]): A COSE key + keys (COSEKeyInterface | list[COSEKeyInterface]): A COSE key or a list of the keys used to verify and decrypt the encoded CWT. no_verify (bool): An indicator whether token verification is skiped or not. Returns: - Union[Dict[int, Any], bytes]: A byte string of the decoded CWT. + dict[int, Any] | bytes: A byte string of the decoded CWT. Raises: ValueError: Invalid arguments. DecodeError: Failed to decode the CWT. VerifyError: Failed to verify the CWT. """ - cwt: Union[bytes, CBORTag, Dict[Union[str, int], Any]] = self._loads(data) + cwt: bytes | CBORTag | dict[str | int, Any] = self._loads(data) if isinstance(cwt, CBORTag) and cwt.tag == CWT.CBOR_TAG: cwt = cwt.value keys = [keys] if isinstance(keys, COSEKeyInterface) else keys - headers: List[Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any]]] = [] + headers: list[tuple[dict[str | int, Any], dict[str | int, Any]]] = [] while isinstance(cwt, CBORTag): p, u, cwt = self._cose.decode_with_headers(cwt, keys) headers.append((p, u)) @@ -354,13 +354,13 @@ def decode( self._verify(cwt, headers) return cwt - def set_private_claim_names(self, claim_names: Dict[str, int]): + def set_private_claim_names(self, claim_names: dict[str, int]): """ Sets private claim definitions. The definitions will be used in :func:`encode ` when it is called with JSON-based claims. Args: - claim_names (Dict[str, int]): A set of private claim definitions which + claim_names (dict[str, int]): A set of private claim definitions which consist of a readable claim name(str) and a claim key(int). The claim key should be less than -65536 but you can use the numbers other than pre-registered numbers listed in @@ -373,11 +373,11 @@ def set_private_claim_names(self, claim_names: Dict[str, int]): def _encode( self, - claims: Union[Claims, Dict[Any, Any], bytes], + claims: Claims | dict[Any, Any] | bytes, key: COSEKeyInterface, nonce: bytes = b"", - recipients: List[RecipientInterface] = [], - signers: List[Signer] = [], + recipients: list[RecipientInterface] = [], + signers: list[Signer] = [], tagged: bool = False, ) -> bytes: if COSE_KEY_OPERATION_VALUES["sign"] in key.key_ops: @@ -394,7 +394,7 @@ def _encode( return self.encode_and_mac(claims, key, recipients, tagged) raise ValueError("The key operation could not be specified.") - def _validate(self, claims: Union[Dict[int, Any], bytes]): + def _validate(self, claims: dict[int, Any] | bytes): if isinstance(claims, bytes): try: nested = self._loads(claims) @@ -410,8 +410,8 @@ def _validate(self, claims: Union[Dict[int, Any], bytes]): def _verify( self, - claims: Union[Dict[Union[str, int], Any], bytes], - headers: Optional[List[Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any]]]] = None, + claims: dict[str | int, Any] | bytes, + headers: list[tuple[dict[str | int, Any], dict[str | int, Any]]] | None = None, ): if not isinstance(claims, dict): raise DecodeError("Failed to decode.") @@ -437,9 +437,9 @@ def _verify( def _verify_cwt_claims_header( self, - claims: Dict[Union[str, int], Any], - protected: Dict[Union[str, int], Any], - unprotected: Dict[Union[str, int], Any], + claims: dict[str | int, Any], + protected: dict[str | int, Any], + unprotected: dict[str | int, Any], ): # RFC 9597 Section 2 label = COSEHeaders.CWT_CLAIMS @@ -458,7 +458,7 @@ def _verify_cwt_claims_header( raise VerifyError(f"The CWT claim({k}) value in {where} header does not match the values in the payload.") return - def _set_default_value(self, claims: Union[Dict[int, Any], bytes]): + def _set_default_value(self, claims: dict[int, Any] | bytes): if isinstance(claims, bytes): return now = timegm(datetime.now(tz=timezone.utc).utctimetuple()) @@ -476,39 +476,39 @@ def _set_default_value(self, claims: Union[Dict[int, Any], bytes]): def encode( - claims: Union[Claims, Dict[str, Any], Dict[int, Any], bytes], + claims: Claims | dict[str, Any] | dict[int, Any] | bytes, key: COSEKeyInterface, nonce: bytes = b"", - recipients: List[RecipientInterface] = [], - signers: List[Signer] = [], + recipients: list[RecipientInterface] = [], + signers: list[Signer] = [], tagged: bool = False, ) -> bytes: return _cwt.encode(claims, key, nonce, recipients, signers, tagged) def encode_and_mac( - claims: Union[Claims, Dict[int, Any], bytes], + claims: Claims | dict[int, Any] | bytes, key: COSEKeyInterface, - recipients: List[RecipientInterface] = [], + recipients: list[RecipientInterface] = [], tagged: bool = False, ) -> bytes: return _cwt.encode_and_mac(claims, key, recipients, tagged) def encode_and_sign( - claims: Union[Claims, Dict[int, Any], bytes], - key: Optional[COSEKeyInterface] = None, - signers: List[Signer] = [], + claims: Claims | dict[int, Any] | bytes, + key: COSEKeyInterface | None = None, + signers: list[Signer] = [], tagged: bool = False, ) -> bytes: return _cwt.encode_and_sign(claims, key, signers, tagged) def encode_and_encrypt( - claims: Union[Claims, Dict[int, Any], bytes], + claims: Claims | dict[int, Any] | bytes, key: COSEKeyInterface, nonce: bytes = b"", - recipients: List[RecipientInterface] = [], + recipients: list[RecipientInterface] = [], tagged: bool = False, ) -> bytes: return _cwt.encode_and_encrypt(claims, key, nonce, recipients, tagged) @@ -516,11 +516,11 @@ def encode_and_encrypt( def decode( data: bytes, - keys: Union[COSEKeyInterface, List[COSEKeyInterface]], + keys: COSEKeyInterface | list[COSEKeyInterface], no_verify: bool = False, -) -> Union[Dict[Union[str, int], Any], bytes]: +) -> dict[str | int, Any] | bytes: return _cwt.decode(data, keys, no_verify) -def set_private_claim_names(claim_names: Dict[str, int]): +def set_private_claim_names(claim_names: dict[str, int]): return _cwt.set_private_claim_names(claim_names) diff --git a/cwt/encrypted_cose_key.py b/cwt/encrypted_cose_key.py index f5ef9860..9b35c212 100644 --- a/cwt/encrypted_cose_key.py +++ b/cwt/encrypted_cose_key.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Union +from typing import Any import cbor2 from cbor2 import CBORTag @@ -20,7 +20,7 @@ def from_cose_key( encryption_key: COSEKeyInterface, nonce: bytes = b"", tagged: bool = False, - ) -> Union[List[Any], bytes]: + ) -> list[Any] | bytes: """ Returns an encrypted COSE key formatted to COSE_Encrypt0 structure. @@ -32,13 +32,13 @@ def from_cose_key( tagged (bool): An indicator whether the response is wrapped by CWT tag(61) or not. Returns: - Union[List[Any], bytes]: A COSE_Encrypt0 structure of the target COSE key. + list[Any] | bytes: A COSE_Encrypt0 structure of the target COSE key. Raises: ValueError: Invalid arguments. EncodeError: Failed to encrypt the COSE key. """ - protected: Dict[int, Any] = {1: encryption_key.alg} - unprotected: Dict[int, Any] = {4: encryption_key.kid} if encryption_key.kid else {} + protected: dict[int, Any] = {1: encryption_key.alg} + unprotected: dict[int, Any] = {4: encryption_key.kid} if encryption_key.kid else {} if not nonce: try: nonce = encryption_key.generate_nonce() @@ -56,7 +56,7 @@ def from_cose_key( return res.value @staticmethod - def to_cose_key(key: List[Any], encryption_key: COSEKeyInterface) -> COSEKeyInterface: + def to_cose_key(key: list[Any], encryption_key: COSEKeyInterface) -> COSEKeyInterface: """ Returns an decrypted COSE key. diff --git a/cwt/helpers/hcert.py b/cwt/helpers/hcert.py index 3065c649..3fe733f9 100644 --- a/cwt/helpers/hcert.py +++ b/cwt/helpers/hcert.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, Union +from typing import Any from cryptography import x509 from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey @@ -18,7 +18,7 @@ def _generate_kid(cert: bytes) -> bytes: return fp[0:8] -def load_pem_hcert_dsc(cert: Union[str, bytes]) -> COSEKeyInterface: +def load_pem_hcert_dsc(cert: str | bytes) -> COSEKeyInterface: """ Loads PEM-formatted DSC (Digital Signing Certificate) issued by CSCA (Certificate Signing Certificate Authority) as a COSEKey. At this time, @@ -38,7 +38,7 @@ def load_pem_hcert_dsc(cert: Union[str, bytes]) -> COSEKeyInterface: k = x509.load_pem_x509_certificate(cert).public_key() else: raise ValueError("Invalid PEM data.") - params: Dict[int, Any] = {} + params: dict[int, Any] = {} params[2] = _generate_kid(cert) if isinstance(k, RSAPublicKey): diff --git a/cwt/recipient.py b/cwt/recipient.py index 78105666..630b6207 100644 --- a/cwt/recipient.py +++ b/cwt/recipient.py @@ -1,5 +1,5 @@ from collections.abc import Mapping -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any import cbor2 @@ -32,14 +32,14 @@ class Recipient: @classmethod def new( cls, - protected: Union[dict, ResolvedHeader] = {}, - unprotected: Union[dict, ResolvedHeader] = {}, + protected: dict | ResolvedHeader = {}, + unprotected: dict | ResolvedHeader = {}, ciphertext: bytes = b"", - recipients: List[Any] = [], - sender_key: Optional[COSEKeyInterface] = None, - recipient_key: Optional[COSEKeyInterface] = None, - hpke_psk: Optional[bytes] = None, - context: Optional[Union[List[Any], Dict[str, Any]]] = None, + recipients: list[Any] = [], + sender_key: COSEKeyInterface | None = None, + recipient_key: COSEKeyInterface | None = None, + hpke_psk: bytes | None = None, + context: list[Any] | dict[str, Any] | None = None, extra_info: bytes = b"", hpke_aad: bytes = b"", ) -> RecipientInterface: @@ -47,12 +47,12 @@ def new( Creates a recipient from a CBOR-like dictionary with numeric keys. Args: - protected (Union[dict, ResolvedHeader]): Parameters that are to be cryptographically protected. - unprotected (Union[dict, ResolvedHeader]): Parameters that are not cryptographically protected. - ciphertext (List[Any]): A cipher text. - sender_key (Optional[COSEKeyInterface]): A sender private key as COSEKey. - recipient_key (Optional[COSEKeyInterface]): A recipient public key as COSEKey. - context (Optional[Union[List[Any], Dict[str, Any]]]): Context + protected (dict | ResolvedHeader): Parameters that are to be cryptographically protected. + unprotected (dict | ResolvedHeader): Parameters that are not cryptographically protected. + ciphertext (list[Any]): A cipher text. + sender_key (COSEKeyInterface | None): A sender private key as COSEKey. + recipient_key (COSEKeyInterface | None): A recipient public key as COSEKey. + context (list[Any] | dict[str, Any] | None): Context information structure. Returns: RecipientInterface: A recipient object. @@ -109,9 +109,9 @@ def new( @classmethod def from_list( cls, - recipient: Union[List[Any], Tuple[Any, ...]], - context: Optional[Union[List[Any], Dict[str, Any]]] = None, - hpke_psk: Optional[bytes] = None, + recipient: list[Any] | tuple[Any, ...], + context: list[Any] | dict[str, Any] | None = None, + hpke_psk: bytes | None = None, extra_info: bytes = b"", hpke_aad: bytes = b"", ) -> RecipientInterface: @@ -119,7 +119,7 @@ def from_list( Creates a recipient from a raw COSE array data. Args: - recipient (Union[List[Any], Tuple[Any, ...]]): A raw COSE recipient array. + recipient (list[Any] | tuple[Any, ...]): A raw COSE recipient array. Returns: RecipientInterface: A recipient object. Raises: @@ -150,7 +150,7 @@ def from_list( return rec if not isinstance(recipient[3], (list, tuple)): raise ValueError("recipients should be list.") - recipients: List[RecipientInterface] = [] + recipients: list[RecipientInterface] = [] for r in recipient[3]: recipients.append(cls.from_list(r, hpke_psk=hpke_psk, extra_info=extra_info, hpke_aad=hpke_aad)) rec = cls.new( diff --git a/cwt/recipient_algs/aes_key_wrap.py b/cwt/recipient_algs/aes_key_wrap.py index 3a6733a8..9e73b4b8 100644 --- a/cwt/recipient_algs/aes_key_wrap.py +++ b/cwt/recipient_algs/aes_key_wrap.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from ..const import COSE_KEY_OPERATION_VALUES from ..cose_key import COSEKey @@ -15,10 +15,10 @@ class AESKeyWrap(RecipientInterface): def __init__( self, - unprotected: Dict[Union[str, int], Any], + unprotected: dict[str | int, Any], ciphertext: bytes = b"", - recipients: List[Any] = [], - sender_key: Optional[COSEKeyInterface] = None, + recipients: list[Any] = [], + sender_key: COSEKeyInterface | None = None, ): if sender_key is None: raise ValueError("sender_key should be set.") @@ -38,13 +38,13 @@ def __init__( ) self._sender_key: COSEKeyInterface = sender_key - def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], Optional[COSEKeyInterface]]: + def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> tuple[list[Any], COSEKeyInterface | None]: self._ciphertext = self._sender_key.wrap_key(plaintext) return self.to_list(), None def decode( self, key: COSEKeyInterface, aad: bytes = b"", alg: int = 0, as_cose_key: bool = False - ) -> Union[bytes, COSEKeyInterface]: + ) -> bytes | COSEKeyInterface: try: unwrapped = key.unwrap_key(self._ciphertext) except Exception as err: diff --git a/cwt/recipient_algs/direct.py b/cwt/recipient_algs/direct.py index 65df3e94..24b12f60 100644 --- a/cwt/recipient_algs/direct.py +++ b/cwt/recipient_algs/direct.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Union +from typing import Any from ..recipient_interface import RecipientInterface @@ -6,10 +6,10 @@ class Direct(RecipientInterface): def __init__( self, - protected: Dict[Union[str, int], Any], - unprotected: Dict[Union[str, int], Any], + protected: dict[str | int, Any], + unprotected: dict[str | int, Any], ciphertext: bytes = b"", - recipients: List[Any] = [], + recipients: list[Any] = [], ): super().__init__(protected, unprotected, ciphertext, recipients) diff --git a/cwt/recipient_algs/direct_hkdf.py b/cwt/recipient_algs/direct_hkdf.py index 2759285e..89fde60e 100644 --- a/cwt/recipient_algs/direct_hkdf.py +++ b/cwt/recipient_algs/direct_hkdf.py @@ -1,5 +1,5 @@ from secrets import token_bytes -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.hkdf import HKDF @@ -19,9 +19,9 @@ class DirectHKDF(Direct): def __init__( self, - protected: Dict[Union[str, int], Any] = {}, - unprotected: Dict[Union[str, int], Any] = {}, - context: List[Any] = [], + protected: dict[str | int, Any] = {}, + unprotected: dict[str | int, Any] = {}, + context: list[Any] = [], ): super().__init__(protected, unprotected, b"", []) @@ -68,7 +68,7 @@ def verify_key( raise VerifyError("Failed to verify key.") from err return - def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], Optional[COSEKeyInterface]]: + def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> tuple[list[Any], COSEKeyInterface | None]: try: hkdf = HKDF( algorithm=self._hash_alg, @@ -83,7 +83,7 @@ def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], O def decode( self, key: COSEKeyInterface, aad: bytes = b"", alg: int = 0, as_cose_key: bool = False - ) -> Union[bytes, COSEKeyInterface]: + ) -> bytes | COSEKeyInterface: try: hkdf = HKDF( algorithm=self._hash_alg, diff --git a/cwt/recipient_algs/direct_key.py b/cwt/recipient_algs/direct_key.py index 4afd1018..95c44c8a 100644 --- a/cwt/recipient_algs/direct_key.py +++ b/cwt/recipient_algs/direct_key.py @@ -1,23 +1,23 @@ -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from ..cose_key_interface import COSEKeyInterface from .direct import Direct class DirectKey(Direct): - def __init__(self, protected: Dict[Union[str, int], Any] = {}, unprotected: Dict[Union[str, int], Any] = {}): + def __init__(self, protected: dict[str | int, Any] = {}, unprotected: dict[str | int, Any] = {}): super().__init__(protected, unprotected, b"", []) if self._alg != -6: raise ValueError("alg(1) should be direct(-6).") return - def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], Optional[COSEKeyInterface]]: + def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> tuple[list[Any], COSEKeyInterface | None]: return self.to_list(), None def decode( self, key: COSEKeyInterface, aad: bytes = b"", alg: int = 0, as_cose_key: bool = False - ) -> Union[bytes, COSEKeyInterface]: + ) -> bytes | COSEKeyInterface: if not as_cose_key: return b"" return key diff --git a/cwt/recipient_algs/ecdh_aes_key_wrap.py b/cwt/recipient_algs/ecdh_aes_key_wrap.py index 2849be65..1965ff20 100644 --- a/cwt/recipient_algs/ecdh_aes_key_wrap.py +++ b/cwt/recipient_algs/ecdh_aes_key_wrap.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from cryptography.hazmat.primitives.keywrap import aes_key_unwrap, aes_key_wrap @@ -18,13 +18,13 @@ class ECDH_AESKeyWrap(RecipientInterface): def __init__( self, - protected: Dict[Union[str, int], Any], - unprotected: Dict[Union[str, int], Any], + protected: dict[str | int, Any], + unprotected: dict[str | int, Any], ciphertext: bytes = b"", - recipients: List[Any] = [], - sender_key: Optional[COSEKeyInterface] = None, - recipient_key: Optional[COSEKeyInterface] = None, - context: List[Any] = [], + recipients: list[Any] = [], + sender_key: COSEKeyInterface | None = None, + recipient_key: COSEKeyInterface | None = None, + context: list[Any] = [], ): super().__init__(protected, unprotected, ciphertext, recipients) self._sender_public_key: Any = None @@ -43,7 +43,7 @@ def __init__( else: raise ValueError(f"Unknown alg(1) for ECDH with key wrap: {self._alg}.") - def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], Optional[COSEKeyInterface]]: + def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> tuple[list[Any], COSEKeyInterface | None]: if not self._recipient_key: raise ValueError("recipient_key should be set in advance.") if not self._context: @@ -77,7 +77,7 @@ def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], O def decode( self, key: COSEKeyInterface, aad: bytes = b"", alg: int = 0, as_cose_key: bool = False - ) -> Union[bytes, COSEKeyInterface]: + ) -> bytes | COSEKeyInterface: if not self._context: raise ValueError("context should be set.") if not self._sender_public_key: diff --git a/cwt/recipient_algs/ecdh_direct_hkdf.py b/cwt/recipient_algs/ecdh_direct_hkdf.py index cad79290..e719471e 100644 --- a/cwt/recipient_algs/ecdh_direct_hkdf.py +++ b/cwt/recipient_algs/ecdh_direct_hkdf.py @@ -1,5 +1,5 @@ from secrets import token_bytes -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from ..algs.ec2 import EC2Key from ..algs.okp import OKPKey @@ -18,13 +18,13 @@ class ECDH_DirectHKDF(Direct): def __init__( self, - protected: Dict[Union[str, int], Any], - unprotected: Dict[Union[str, int], Any], + protected: dict[str | int, Any], + unprotected: dict[str | int, Any], ciphertext: bytes = b"", - recipients: List[Any] = [], - sender_key: Optional[COSEKeyInterface] = None, - recipient_key: Optional[COSEKeyInterface] = None, - context: List[Any] = [], + recipients: list[Any] = [], + sender_key: COSEKeyInterface | None = None, + recipient_key: COSEKeyInterface | None = None, + context: list[Any] = [], ): super().__init__(protected, unprotected, ciphertext, recipients) self._sender_public_key: Any = None @@ -60,7 +60,7 @@ def __init__( if self._context[2][1]: self._unprotected[-25] = self._context[2][1] - def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], Optional[COSEKeyInterface]]: + def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> tuple[list[Any], COSEKeyInterface | None]: if not self._recipient_key: raise ValueError("recipient_key should be set in advance.") @@ -97,7 +97,7 @@ def decode( aad: bytes = b"", alg: int = 0, as_cose_key: bool = False, - ) -> Union[bytes, COSEKeyInterface]: + ) -> bytes | COSEKeyInterface: if not self._sender_public_key: raise ValueError("sender_public_key should be set.") try: diff --git a/cwt/recipient_algs/hpke.py b/cwt/recipient_algs/hpke.py index c2afa2fa..7f0a8ce7 100644 --- a/cwt/recipient_algs/hpke.py +++ b/cwt/recipient_algs/hpke.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from pyhpke import AEADId, CipherSuite, KDFId, KEMId, KEMKey, KEMKeyInterface @@ -35,7 +35,7 @@ } -def to_hpke_ciphersuites(alg: int) -> Tuple[int, int, int]: +def to_hpke_ciphersuites(alg: int) -> tuple[int, int, int]: base_alg = _KE_TO_BASE.get(alg, alg) if base_alg in _HPKE_SUITES: return _HPKE_SUITES[base_alg] @@ -49,12 +49,12 @@ def is_hpke_ke(alg: int) -> bool: class HPKE(RecipientInterface): def __init__( self, - protected: Dict[Union[str, int], Any], - unprotected: Dict[Union[str, int], Any], + protected: dict[str | int, Any], + unprotected: dict[str | int, Any], ciphertext: bytes = b"", - recipients: List[Any] = [], - recipient_key: Optional[COSEKeyInterface] = None, - psk: Optional[bytes] = None, + recipients: list[Any] = [], + recipient_key: COSEKeyInterface | None = None, + psk: bytes | None = None, content_alg: int = 0, extra_info: bytes = b"", hpke_info: bytes = b"", @@ -97,7 +97,7 @@ def _build_recipient_info(self, content_alg: int) -> bytes: ] return self._dumps(recipient_structure) - def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> Tuple[List[Any], Optional[COSEKeyInterface]]: + def encode(self, plaintext: bytes = b"", aad: bytes = b"") -> tuple[list[Any], COSEKeyInterface | None]: if self._recipient_key is None: raise ValueError("recipient_key should be set in advance.") self._kem_key = self._to_kem_key(self._recipient_key) @@ -139,7 +139,7 @@ def decode( aad: bytes = b"", alg: int = 0, as_cose_key: bool = False, - ) -> Union[bytes, COSEKeyInterface]: + ) -> bytes | COSEKeyInterface: ek = self._unprotected.get(-4, None) if ek is None and isinstance(self._protected, dict): ek = self._protected.get(-4, None) diff --git a/cwt/recipient_interface.py b/cwt/recipient_interface.py index ae58f2ff..eed32e64 100644 --- a/cwt/recipient_interface.py +++ b/cwt/recipient_interface.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Tuple, Union +from typing import Any from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey from cryptography.hazmat.primitives.asymmetric.x448 import X448PublicKey @@ -17,20 +17,20 @@ class RecipientInterface(CBORProcessor): def __init__( self, - protected: Optional[Dict[Union[str, int], Any]] = None, - unprotected: Optional[Dict[Union[str, int], Any]] = None, + protected: dict[str | int, Any] | None = None, + unprotected: dict[str | int, Any] | None = None, ciphertext: bytes = b"", - recipients: List[Any] = [], - key_ops: List[int] = [], + recipients: list[Any] = [], + key_ops: list[int] = [], key: bytes = b"", ): """ Constructor. Args: - protected (Optional[Dict[int, Any]]): Parameters that are to be cryptographically + protected (dict[int, Any] | None): Parameters that are to be cryptographically protected. - unprotected (Optional[Dict[int, Any]]): Parameters that are not cryptographically + unprotected (dict[int, Any] | None): Parameters that are not cryptographically protected. ciphertext: A ciphertext encoded as bytes. recipients: A list of recipient information structures. @@ -74,15 +74,15 @@ def __init__( if not isinstance(unprotected[5], bytes): raise ValueError("unprotected[5](iv) should be bytes.") - self._b_protected: Optional[bytes] = None + self._b_protected: bytes | None = None self._protected = protected self._unprotected = unprotected self._ciphertext = ciphertext self._key = key - self._context: List[Any] = [0, [None, None, None], [None, None, None], [None, None]] + self._context: list[Any] = [0, [None, None, None], [None, None, None], [None, None]] # Validate recipients - self._recipients: List[RecipientInterface] = [] + self._recipients: list[RecipientInterface] = [] if not recipients: return for recipient in recipients: @@ -106,7 +106,7 @@ def alg(self) -> int: return self._alg @property - def protected(self) -> Dict[Union[str, int], Any]: + def protected(self) -> dict[str | int, Any]: """ The parameters that are to be cryptographically protected. """ @@ -122,7 +122,7 @@ def b_protected(self) -> bytes: return self._b_protected @property - def unprotected(self) -> Dict[Union[str, int], Any]: + def unprotected(self) -> dict[str | int, Any]: """ The parameters that are not cryptographically protected. """ @@ -136,29 +136,29 @@ def ciphertext(self) -> bytes: return self._ciphertext @property - def recipients(self) -> List[Any]: + def recipients(self) -> list[Any]: """ The list of recipient information structures. """ return self._recipients @property - def context(self) -> List[Any]: + def context(self) -> list[Any]: """ The recipient context information. """ return self._context - def to_list(self) -> List[Any]: + def to_list(self) -> list[Any]: """ Returns the recipient information as a COSE recipient structure. Returns: - List[Any]: The recipient structure. + list[Any]: The recipient structure. """ b_protected = self._dumps(self._protected) if self._protected else b"" b_ciphertext = self._ciphertext if self._ciphertext else b"" - res: List[Any] = [b_protected, self._unprotected, b_ciphertext] + res: list[Any] = [b_protected, self._unprotected, b_ciphertext] if not self._recipients: return res @@ -172,7 +172,7 @@ def encode( self, plaintext: bytes = b"", aad: bytes = b"", - ) -> Tuple[List[Any], Optional[COSEKeyInterface]]: + ) -> tuple[list[Any], COSEKeyInterface | None]: """ Encrypts a specified plaintext to the ciphertext in the COSE_Recipient structure with the recipient-specific method (e.g., key wrapping, key @@ -189,7 +189,7 @@ def encode( aad_context (bytes): An additional authenticated data context to build an Enc_structure internally. Returns: - Tuple[List[Any], Optional[COSEKeyInterface]]: The encoded COSE_Recipient structure + tuple[list[Any], COSEKeyInterface | None]: The encoded COSE_Recipient structure and a derived key. Raises: ValueError: Invalid arguments. @@ -203,7 +203,7 @@ def decode( aad: bytes = b"", alg: int = 0, as_cose_key: bool = False, - ) -> Union[bytes, COSEKeyInterface]: + ) -> bytes | COSEKeyInterface: """ Decrypts the ciphertext in the COSE_Recipient structure with the recipient-specific method (e.g., key wrapping, key agreement, @@ -222,7 +222,7 @@ def decode( as_cose_key (bool): The indicator whether the output will be returned as a COSEKey or not. Returns: - Union[bytes, COSEKeyInterface]: The decrypted ciphertext field or The COSEKey + bytes | COSEKeyInterface: The decrypted ciphertext field or The COSEKey converted from the decrypted ciphertext. Raises: ValueError: Invalid arguments. @@ -230,7 +230,7 @@ def decode( """ raise NotImplementedError - def _to_cose_key(self, k: Union[EllipticCurvePublicKey, X25519PublicKey, X448PublicKey]) -> Dict[int, Any]: + def _to_cose_key(self, k: EllipticCurvePublicKey | X25519PublicKey | X448PublicKey) -> dict[int, Any]: if isinstance(k, EllipticCurvePublicKey): return EC2Key.to_cose_key(k) return OKPKey.to_cose_key(k) diff --git a/cwt/recipients.py b/cwt/recipients.py index aa299a0a..7fc133c8 100644 --- a/cwt/recipients.py +++ b/cwt/recipients.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List, Optional, Self, Tuple, Union +from typing import Any, Self from .cbor_processor import CBORProcessor from .cose_key import COSEKey @@ -12,7 +12,7 @@ class Recipients(CBORProcessor): A Set of COSE Recipients. """ - def __init__(self, recipients: List[RecipientInterface], verify_kid: bool = False): + def __init__(self, recipients: list[RecipientInterface], verify_kid: bool = False): self._recipients = recipients self._verify_kid = verify_kid return @@ -20,22 +20,22 @@ def __init__(self, recipients: List[RecipientInterface], verify_kid: bool = Fals @classmethod def from_list( cls, - recipients: Union[List[Any], Tuple[Any, ...]], + recipients: list[Any] | tuple[Any, ...], verify_kid: bool = False, - context: Optional[Union[List[Any], Dict[str, Any]]] = None, - hpke_psk: Optional[bytes] = None, + context: list[Any] | dict[str, Any] | None = None, + hpke_psk: bytes | None = None, extra_info: bytes = b"", hpke_aad: bytes = b"", ) -> Self: """ Create Recipients from a CBOR-like list. """ - res: List[RecipientInterface] = [] + res: list[RecipientInterface] = [] for r in recipients: 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: str) -> 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``. """ diff --git a/cwt/signer.py b/cwt/signer.py index 33a9ebee..8a14beda 100644 --- a/cwt/signer.py +++ b/cwt/signer.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, Self, Union +from typing import Any, Self from .cbor_processor import CBORProcessor from .const import COSE_ALGORITHMS_SIGNATURE @@ -15,8 +15,8 @@ class Signer(CBORProcessor): def __init__( self, cose_key: COSEKeyInterface, - protected: Union[Dict[Union[str, int], Any], bytes], - unprotected: Dict[Union[str, int], Any], + protected: dict[str | int, Any] | bytes, + unprotected: dict[str | int, Any], signature: bytes = b"", ): self._cose_key = cose_key @@ -43,7 +43,7 @@ def protected(self) -> bytes: return self._protected @property - def unprotected(self) -> Dict[Union[str, int], Any]: + def unprotected(self) -> dict[str | int, Any]: """ The parameters that are not cryptographically protected. """ @@ -60,8 +60,8 @@ def signature(self) -> bytes: def new( cls, cose_key: COSEKeyInterface, - protected: Union[dict, bytes, ResolvedHeader] = {}, - unprotected: Union[dict, ResolvedHeader] = {}, + protected: dict | bytes | ResolvedHeader = {}, + unprotected: dict | ResolvedHeader = {}, signature: bytes = b"", ) -> Self: """ @@ -69,23 +69,23 @@ def new( Args: cose_key (COSEKey): A signature key for the signer. - protected (Union[dict, bytes, ResolvedHeader]): Parameters that are to be cryptographically + protected (dict | bytes | ResolvedHeader): Parameters that are to be cryptographically protected. - unprotected (Union[dict, bytes, ResolvedHeader]): Parameters that are not cryptographically protected. + unprotected (dict | bytes | ResolvedHeader): Parameters that are not cryptographically protected. signature (bytes): A signature as bytes. Returns: Signer: A signer information object. Raises: ValueError: Invalid arguments. """ - p: Union[Dict[Union[str, int], Any], bytes] = ( + p: dict[str | int, Any] | bytes = ( protected if isinstance(protected, bytes) else to_cose_header(protected, algs=COSE_ALGORITHMS_SIGNATURE) ) - u: Dict[Union[str, int], Any] = to_cose_header(unprotected, algs=COSE_ALGORITHMS_SIGNATURE) + u: dict[str | int, Any] = to_cose_header(unprotected, algs=COSE_ALGORITHMS_SIGNATURE) return cls(cose_key, p, u, signature) @classmethod - def from_jwk(cls, data: Union[str, bytes, Dict[str, Any]]) -> Self: + def from_jwk(cls, data: 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, @@ -94,15 +94,15 @@ def from_jwk(cls, data: Union[str, bytes, Dict[str, Any]]) -> Self: header, you have to use :func:`Signer.new `. Args: - data (Union[str, bytes, Dict[str, Any]]): A JWK. + data (str | bytes | dict[str, Any]): A JWK. Returns: Signer: A signer information object. Raises: ValueError: Invalid arguments. DecodeError: Failed to decode the key data. """ - protected: Dict[Union[str, int], Any] = {} - unprotected: Dict[Union[str, int], Any] = {} + protected: dict[str | int, Any] = {} + unprotected: dict[str | int, Any] = {} cose_key = COSEKey.from_jwk(data) @@ -119,9 +119,9 @@ def from_jwk(cls, data: Union[str, bytes, Dict[str, Any]]) -> Self: @classmethod def from_pem( cls, - data: Union[str, bytes], - alg: Union[int, str] = "", - kid: Union[bytes, str] = b"", + data: str | bytes, + alg: int | str = "", + kid: bytes | str = b"", ) -> Self: """ Creates a signer information object (COSE_Signature) from PEM-formatted key. @@ -131,19 +131,19 @@ def from_pem( header, you have to use :func:`Signer.new `. Args: - data (Union[str, bytes]): A PEM-formatted key. - alg (Union[int, str]): An algorithm label(int) or name(str). It is only + data (str | bytes): A PEM-formatted key. + alg (int | str): An algorithm label(int) or name(str). It is only used when an algorithm cannot be specified by the PEM data, such as RSA family algorithms. - kid (Union[bytes, str]): A key identifier. + kid (bytes | str): A key identifier. Returns: Signer: A signer information object. Raises: ValueError: Invalid arguments. DecodeError: Failed to decode the key data. """ - protected: Dict[Union[str, int], Any] = {} - unprotected: Dict[Union[str, int], Any] = {} + protected: dict[str | int, Any] = {} + unprotected: dict[str | int, Any] = {} cose_key = COSEKey.from_pem(data, alg=alg, kid=kid) diff --git a/cwt/utils.py b/cwt/utils.py index cef19300..fab059c4 100644 --- a/cwt/utils.py +++ b/cwt/utils.py @@ -2,7 +2,7 @@ import copy import json from collections.abc import Mapping -from typing import Any, Dict, List, Optional, Union +from typing import Any import cbor2 @@ -25,7 +25,7 @@ def to_mutable(obj: Any, *, preserve_mutable: bool = False) -> Any: """Convert containers, preserving shared references, byte strings and map keys.""" - memo: Dict[int, Any] = {} + memo: dict[int, Any] = {} def convert(value: Any) -> Any: if not isinstance(value, (Mapping, list, tuple)): @@ -35,13 +35,13 @@ def convert(value: Any) -> Any: return memo[id(value)] if isinstance(value, Mapping): - result: Dict[Any, Any] = value if preserve_mutable and isinstance(value, dict) else {} + result: dict[Any, Any] = value if preserve_mutable and isinstance(value, dict) else {} memo[id(value)] = result for key, item in value.items(): result[key] = convert(item) return result - items: List[Any] = value if preserve_mutable and isinstance(value, list) else [] + items: list[Any] = value if preserve_mutable and isinstance(value, list) else [] memo[id(value)] = items for i, item in enumerate(value): if items is value: @@ -100,7 +100,7 @@ def base64url_decode(v: str) -> bytes: def parse_apu(context: dict) -> list: - apu: List[Any] = [None, None, None] + apu: list[Any] = [None, None, None] if "apu" not in context: return apu if not isinstance(context["apu"], dict): @@ -124,7 +124,7 @@ def parse_apu(context: dict) -> list: def parse_apv(context: dict) -> list: - apv: List[Any] = [None, None, None] + apv: list[Any] = [None, None, None] if "apv" not in context: return apv if not isinstance(context["apv"], dict): @@ -147,8 +147,8 @@ def parse_apv(context: dict) -> list: return apv -def to_cis(context: Dict[str, Any], recipient_alg: Optional[int] = None) -> List[Any]: - res: List[Any] = [] +def to_cis(context: dict[str, Any], recipient_alg: int | None = None) -> list[Any]: + res: list[Any] = [] # AlgorithmID if "alg" not in context: @@ -170,7 +170,7 @@ def to_cis(context: Dict[str, Any], recipient_alg: Optional[int] = None) -> List res.append(parse_apv(context)) # SuppPubInfo - supp_pub: List[Any] = [None, None] + supp_pub: list[Any] = [None, None] protected = {} if "supp_pub" in context: if not isinstance(context["supp_pub"], dict): @@ -205,7 +205,7 @@ class ResolvedHeader: nor value encoding to bstr, and can be passed directly to the CBOR encoding logic. """ - def __init__(self, params: Dict[Union[str, int], Any]): + def __init__(self, params: dict[str | int, Any]): self.params = params @@ -213,14 +213,14 @@ def __init__(self, params: Dict[Union[str, int], Any]): _COSE_HEADER_TSTR_PARAMS = {"cty", "content type", "x5u"} -def to_cose_header(data: Optional[Union[dict, ResolvedHeader]] = None, algs: Dict[str, int] = {}) -> Dict[Union[str, int], Any]: +def to_cose_header(data: dict | ResolvedHeader | None = None, algs: dict[str, int] = {}) -> dict[str | int, Any]: if data is None: return {} if isinstance(data, ResolvedHeader): return data.params if len(data) == 0 or not isinstance(list(data.keys())[0], str): return data - res: Dict[Union[str, int], Any] = {} + res: dict[str | int, Any] = {} if not algs: algs = COSE_NAMED_ALGORITHMS_SUPPORTED for k, v in data.items(): @@ -250,11 +250,11 @@ def to_cose_header(data: Optional[Union[dict, ResolvedHeader]] = None, algs: Dic return res -def jwk_to_cose_key_params(data: Union[str, bytes, Dict[str, Any]]) -> Dict[int, Any]: - cose_key: Dict[int, Any] = {} +def jwk_to_cose_key_params(data: str | bytes | dict[str, Any]) -> dict[int, Any]: + cose_key: dict[int, Any] = {} # kty - jwk: Dict[str, Any] + jwk: dict[str, Any] if not isinstance(data, dict): jwk = json.loads(data) else: @@ -356,7 +356,7 @@ def jwk_to_cose_key_params(data: Union[str, bytes, Dict[str, Any]]) -> Dict[int, return cose_key -def _validate_context(context: List[Any]) -> List[Any]: +def _validate_context(context: list[Any]) -> list[Any]: if len(context) != 4 and len(context) != 5: raise ValueError("Invalid context information.") # AlgorithmID @@ -376,8 +376,8 @@ def _validate_context(context: List[Any]) -> List[Any]: return context -def to_recipient_context(alg: int, u: Dict[Union[str, int], Any], context: Union[List[Any], Dict[str, Any]]) -> List[Any]: - ctx: List[Any] = [ +def to_recipient_context(alg: int, u: dict[str | int, Any], context: list[Any] | dict[str, Any]) -> list[Any]: + ctx: list[Any] = [ None, [ u[-21] if -21 in u else None, @@ -406,5 +406,5 @@ def to_recipient_context(alg: int, u: Dict[Union[str, int], Any], context: Union return ctx -def sort_keys_for_deterministic_encoding(d: Dict[Union[str, int], Any]) -> Dict[Union[str, int], Any]: +def sort_keys_for_deterministic_encoding(d: dict[str | int, Any]) -> dict[str | int, Any]: return {k: v for k, v in sorted(d.items(), key=lambda kv: cbor2.dumps(kv[0]))} diff --git a/samples/eudcc/swedish_verifier.py b/samples/eudcc/swedish_verifier.py index 9627541c..f6acc8a0 100644 --- a/samples/eudcc/swedish_verifier.py +++ b/samples/eudcc/swedish_verifier.py @@ -1,7 +1,7 @@ import json import os import zlib -from typing import Any, Dict, Union +from typing import Any import jwt import requests @@ -59,7 +59,7 @@ def refresh_trustlist(self): json.dump(self._trustlist, f, indent=4) return - def verify_and_decode(self, eudcc: bytes) -> Union[Dict[Union[str, int], Any], bytes]: + def verify_and_decode(self, eudcc: bytes) -> dict[str | int, Any] | bytes: if eudcc.startswith(b"HC1:"): # Decode Base45 data. eudcc = b45decode(eudcc[4:]) diff --git a/samples/eudcc/verifier.py b/samples/eudcc/verifier.py index 71e87241..18eb18e0 100644 --- a/samples/eudcc/verifier.py +++ b/samples/eudcc/verifier.py @@ -1,7 +1,7 @@ import json import os import zlib -from typing import Any, Dict, Union +from typing import Any import requests from base45 import b45decode @@ -64,7 +64,7 @@ def refresh_trustlist(self): json.dump([v for v in self._trustlist if v["x_kid"] in active_kids], f, indent=4) return - def verify_and_decode(self, eudcc: bytes) -> Union[Dict[Union[str, int], Any], bytes]: + def verify_and_decode(self, eudcc: bytes) -> dict[str | int, Any] | bytes: if eudcc.startswith(b"HC1:"): # Decode Base45 data. eudcc = b45decode(eudcc[4:]) diff --git a/tests/test_cose.py b/tests/test_cose.py index 556370ea..7c6bb789 100644 --- a/tests/test_cose.py +++ b/tests/test_cose.py @@ -10,7 +10,7 @@ import base64 import datetime from secrets import token_bytes -from typing import Any, Dict, Union +from typing import Any import cbor2 import pytest @@ -187,8 +187,8 @@ def test_cose_signing_resolved_headers(self, ctx, wrap_protected, wrap_unprotect private_key = COSEKey.from_pem(key_file.read()) with open(key_path("public_key_ed25519.pem")) as key_file: public_key = COSEKey.from_pem(key_file.read(), kid="01") - protected: Dict[Union[str, int], Any] = {1: -8, -65537: "protected text", -65538: b"protected bytes"} - unprotected: Dict[Union[str, int], Any] = {4: b"01", -65539: "unprotected text", -65540: b"unprotected bytes"} + protected: dict[str | int, Any] = {1: -8, -65537: "protected text", -65538: b"protected bytes"} + unprotected: dict[str | int, Any] = {4: b"01", -65539: "unprotected text", -65540: b"unprotected bytes"} if wrap_protected: protected["application"] = "protected application text" if wrap_unprotected: