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: