Skip to content

Commit e17c035

Browse files
dajiajidajiaji
committed
Add encode() to Recipients other than HPKE.
1 parent 85f0f2a commit e17c035

11 files changed

Lines changed: 206 additions & 141 deletions

‎cwt/algs/raw.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,9 @@ def __init__(self, params: Dict[int, Any]):
2525
def key(self) -> bytes:
2626
return self._key
2727

28+
def to_bytes(self) -> bytes:
29+
return self._key
30+
2831
def to_dict(self) -> Dict[int, Any]:
2932
res = super().to_dict()
3033
res[-1] = self._key

‎cwt/cose_key_interface.py‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,13 @@ def key(self) -> Any:
104104
"""
105105
raise NotImplementedError
106106

107+
@property
108+
def crv(self) -> int:
109+
"""
110+
The curve (``crv`` parameter value) of the key.
111+
"""
112+
raise NotImplementedError
113+
107114
def to_bytes(self) -> bytes:
108115
"""
109116
Serializes the body of the key as a byte string.

‎cwt/recipient.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ def new(
3636
ciphertext: bytes = b"",
3737
recipients: List[Any] = [],
3838
sender_key: Optional[COSEKeyInterface] = None,
39+
recipient_key: Optional[COSEKeyInterface] = None,
3940
) -> RecipientInterface:
4041
"""
4142
Creates a recipient from a CBOR-like dictionary with numeric keys.

‎cwt/recipient_algs/aes_key_wrap.py‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
from ..const import COSE_KEY_OPERATION_VALUES
44
from ..cose_key import COSEKey
55
from ..cose_key_interface import COSEKeyInterface
6-
from ..exceptions import DecodeError, EncodeError
6+
from ..exceptions import DecodeError
77
from ..recipient_interface import RecipientInterface
88

99

@@ -35,6 +35,19 @@ def __init__(
3535
)
3636
self._sender_key = sender_key
3737

38+
def encode(
39+
self,
40+
plaintext: bytes = b"",
41+
recipient_key: Optional[COSEKeyInterface] = None,
42+
salt: Optional[bytes] = None,
43+
context: Optional[Union[List[Any], Dict[str, Any]]] = None,
44+
external_aad: bytes = b"",
45+
aad_context: str = "Enc_Recipient",
46+
) -> Optional[COSEKeyInterface]:
47+
48+
self._ciphertext = self._sender_key.wrap_key(plaintext)
49+
return None
50+
3851
def apply(
3952
self,
4053
key: Optional[COSEKeyInterface] = None,
@@ -48,10 +61,7 @@ def apply(
4861
raise ValueError("key should be set.")
4962
if key.kid:
5063
self._protected[4] = key.kid
51-
try:
52-
self._ciphertext = self._sender_key.wrap_key(key.key)
53-
except Exception as err:
54-
raise EncodeError("Failed to wrap key.") from err
64+
self._ciphertext = self._sender_key.wrap_key(key.key)
5565
return key
5666

5767
def extract(

‎cwt/recipient_algs/direct_hkdf.py‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,55 @@ def verify_key(
8282
raise VerifyError("Failed to verify key.") from err
8383
return
8484

85+
def encode(
86+
self,
87+
plaintext: bytes = b"",
88+
recipient_key: Optional[COSEKeyInterface] = None,
89+
salt: Optional[bytes] = None,
90+
context: Optional[Union[List[Any], Dict[str, Any]]] = None,
91+
external_aad: bytes = b"",
92+
aad_context: str = "Enc_Recipient",
93+
) -> Optional[COSEKeyInterface]:
94+
95+
if not context:
96+
raise ValueError("context should be set.")
97+
ctx: list
98+
if isinstance(context, dict):
99+
alg = self._alg if isinstance(self._alg, int) else 0
100+
ctx = to_cis(context, alg)
101+
else:
102+
self._validate_context(context)
103+
ctx = context
104+
self._applied_ctx = self._apply_context(ctx)
105+
106+
# Generate a salt automatically if both of a salt and a PartyU nonce are not specified.
107+
if not salt and not self._salt and not self._applied_ctx[1][1]:
108+
self._salt = token_bytes(32) if self._alg == -10 else token_bytes(64)
109+
self._unprotected[-20] = self._salt
110+
elif salt:
111+
self._salt = salt
112+
self._unprotected[-20] = self._salt
113+
114+
# PartyU nonce
115+
if self._applied_ctx[1][1]:
116+
self._unprotected[-22] = self._applied_ctx[1][1]
117+
# PartyV nonce
118+
if self._applied_ctx[2][1]:
119+
self._unprotected[-25] = self._applied_ctx[2][1]
120+
121+
# Derive key.
122+
try:
123+
hkdf = HKDF(
124+
algorithm=self._hash_alg,
125+
length=COSE_KEY_LEN[self._applied_ctx[0]] // 8,
126+
salt=self._salt,
127+
info=self._dumps(self._applied_ctx),
128+
)
129+
derived = hkdf.derive(plaintext)
130+
return COSEKey.from_symmetric_key(derived, self._applied_ctx[0], self._kid)
131+
except Exception as err:
132+
raise EncodeError("Failed to derive key.") from err
133+
85134
def apply(
86135
self,
87136
key: Optional[COSEKeyInterface] = None,

‎cwt/recipient_algs/ecdh_aes_key_wrap.py‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
from cryptography.hazmat.primitives.keywrap import aes_key_unwrap, aes_key_wrap
44

5+
from ..algs.ec2 import EC2Key
56
from ..const import COSE_KEY_OPERATION_VALUES
67
from ..cose_key import COSEKey
78
from ..cose_key_interface import COSEKeyInterface
@@ -49,6 +50,40 @@ def __init__(
4950
else:
5051
raise ValueError(f"Unknown alg(1) for ECDH with key wrap: {self._alg}.")
5152

53+
def encode(
54+
self,
55+
plaintext: bytes = b"",
56+
recipient_key: Optional[COSEKeyInterface] = None,
57+
salt: Optional[bytes] = None,
58+
context: Optional[Union[List[Any], Dict[str, Any]]] = None,
59+
external_aad: bytes = b"",
60+
aad_context: str = "Enc_Recipient",
61+
) -> Optional[COSEKeyInterface]:
62+
63+
if not recipient_key:
64+
raise ValueError("recipient_key should be set in advance.")
65+
if not context:
66+
raise ValueError("context should be set.")
67+
if self._alg in [-29, -30, -31]:
68+
# ECDH-ES
69+
self._sender_key = EC2Key({1: 2, -1: recipient_key.crv, 3: self._alg})
70+
else:
71+
# ECDH-SS (alg=-32, -33, -34)
72+
if not self._sender_key:
73+
raise ValueError("sender_key should be set in advance.")
74+
wrapping_key = self._sender_key.derive_key(context, public_key=recipient_key)
75+
if self._alg in [-29, -30, -31]:
76+
# ECDH-ES
77+
self._unprotected[-1] = self._to_cose_key(self._sender_key.key.public_key())
78+
else:
79+
# ECDH-SS (alg=-32, -33, -34)
80+
self._unprotected[-2] = self._to_cose_key(self._sender_key.key.public_key())
81+
try:
82+
self._ciphertext = aes_key_wrap(wrapping_key.key, plaintext)
83+
except Exception as err:
84+
raise EncodeError("Failed to wrap key.") from err
85+
return None
86+
5287
def apply(
5388
self,
5489
key: Optional[COSEKeyInterface] = None,

‎cwt/recipient_algs/ecdh_direct_hkdf.py‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22
from secrets import token_bytes
33
from typing import Any, Dict, List, Optional, Union
44

5+
from ..algs.ec2 import EC2Key
6+
from ..algs.okp import OKPKey
57
from ..const import COSE_KEY_OPERATION_VALUES
68
from ..cose_key import COSEKey
79
from ..cose_key_interface import COSEKeyInterface
@@ -58,6 +60,66 @@ def __init__(
5860
else:
5961
raise ValueError(f"Unknown alg(1) for ECDH with HKDF: {self._alg}.")
6062

63+
def encode(
64+
self,
65+
plaintext: bytes = b"",
66+
recipient_key: Optional[COSEKeyInterface] = None,
67+
salt: Optional[bytes] = None,
68+
context: Optional[Union[List[Any], Dict[str, Any]]] = None,
69+
external_aad: bytes = b"",
70+
aad_context: str = "Enc_Recipient",
71+
) -> Optional[COSEKeyInterface]:
72+
73+
if not recipient_key:
74+
raise ValueError("recipient_key should be set in advance.")
75+
if not context:
76+
raise ValueError("context should be set.")
77+
ctx: list
78+
if isinstance(context, dict):
79+
alg = self._alg if isinstance(self._alg, int) else 0
80+
ctx = to_cis(context, alg)
81+
else:
82+
self._validate_context(context)
83+
ctx = context
84+
self._applied_ctx = self._apply_context(ctx)
85+
86+
# Generate a salt automatically if both of a salt and a PartyU nonce are not specified.
87+
if self._alg in [-27, -28]: # ECDH-SS
88+
if not salt and not self._salt and not self._applied_ctx[1][1]:
89+
self._salt = token_bytes(32) if self._alg == -27 else token_bytes(64)
90+
self._unprotected[-20] = self._salt
91+
elif salt:
92+
self._salt = salt
93+
self._unprotected[-20] = self._salt
94+
95+
# PartyU nonce
96+
if self._applied_ctx[1][1]:
97+
self._unprotected[-22] = self._applied_ctx[1][1]
98+
# PartyV nonce
99+
if self._applied_ctx[2][1]:
100+
self._unprotected[-25] = self._applied_ctx[2][1]
101+
102+
# Derive key.
103+
if self._alg in [-25, -26]:
104+
# ECDH-ES
105+
if recipient_key.kty == 2:
106+
self._sender_key = EC2Key({1: 2, -1: recipient_key.crv, 3: self._alg})
107+
else:
108+
# should drop this support.
109+
self._sender_key = OKPKey({1: 1, -1: recipient_key.crv, 3: self._alg})
110+
else:
111+
# ECDH-SS (alg=-27 or -28)
112+
if not self._sender_key:
113+
raise ValueError("sender_key should be set in advance.")
114+
derived_key = self._sender_key.derive_key(self._applied_ctx, public_key=recipient_key)
115+
if self._alg in [-25, -26]:
116+
# ECDH-ES
117+
self._unprotected[-1] = self._to_cose_key(self._sender_key.key.public_key())
118+
else:
119+
# ECDH-SS (alg=-27 or -28)
120+
self._unprotected[-2] = self._to_cose_key(self._sender_key.key.public_key())
121+
return derived_key
122+
61123
def apply(
62124
self,
63125
key: Optional[COSEKeyInterface] = None,

‎cwt/recipient_algs/hpke.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ def __init__(
3333

3434
def encode(
3535
self,
36-
plaintext: bytes,
36+
plaintext: bytes = b"",
3737
recipient_key: Optional[COSEKeyInterface] = None,
3838
salt: Optional[bytes] = None,
3939
context: Optional[Union[List[Any], Dict[str, Any]]] = None,

‎cwt/recipient_interface.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -156,7 +156,7 @@ def to_list(self, payload: bytes = b"", external_aad: bytes = b"", aad_context:
156156

157157
def encode(
158158
self,
159-
plaintext: bytes,
159+
plaintext: bytes = b"",
160160
recipient_key: Optional[COSEKeyInterface] = None,
161161
salt: Optional[bytes] = None,
162162
context: Optional[Union[List[Any], Dict[str, Any]]] = None,

‎tests/test_cose_key.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -626,3 +626,7 @@ def test_cose_key_interface(self):
626626
ki.to_bytes()
627627
pytest.fail("to_bytes should fail.")
628628
assert "" == str(err.value)
629+
with pytest.raises(NotImplementedError) as err:
630+
ki.crv
631+
pytest.fail("crv should fail.")
632+
assert "" == str(err.value)

0 commit comments

Comments
 (0)