|
1 | 1 | # Aligns with https://github.com/eu-digital-identity-wallet/pyMDOC-CBOR |
2 | 2 | import json |
3 | 3 | import random |
| 4 | +from collections.abc import Mapping, Sequence |
| 5 | +from typing import Any, Union |
4 | 6 |
|
5 | 7 | import cbor2 |
6 | 8 | from cbor2.tool import DefaultEncoder, key_to_str |
7 | 9 | from pycose.messages import Sign1Message |
8 | 10 |
|
9 | 11 |
|
| 12 | +def thaw_cbor(obj: Any) -> Any: |
| 13 | + """ |
| 14 | + Convert cbor2 6.x immutable decode results into mutable structures. |
| 15 | +
|
| 16 | + cbor2 >= 6 returns tuple for arrays and frozendict for maps. pycose 1.x |
| 17 | + requires a mutable list (it uses .pop) and a plain dict for COSE headers. |
| 18 | + """ |
| 19 | + if isinstance(obj, cbor2.CBORTag): |
| 20 | + return cbor2.CBORTag(obj.tag, thaw_cbor(obj.value)) |
| 21 | + if isinstance(obj, Mapping): |
| 22 | + return {thaw_cbor(k): thaw_cbor(v) for k, v in obj.items()} |
| 23 | + if isinstance(obj, tuple): |
| 24 | + return [thaw_cbor(item) for item in obj] |
| 25 | + if isinstance(obj, list): |
| 26 | + return [thaw_cbor(item) for item in obj] |
| 27 | + return obj |
| 28 | + |
| 29 | + |
| 30 | +def _cose_sign1_from_obj(cose_obj: Any) -> Sign1Message: |
| 31 | + """Build a Sign1Message from a decoded COSE_Sign1 array (list or tuple).""" |
| 32 | + cose_obj = thaw_cbor(cose_obj) |
| 33 | + if not isinstance(cose_obj, list): |
| 34 | + raise TypeError("Bytes cannot be decoded as COSE message") |
| 35 | + return Sign1Message.from_cose_obj(cose_obj, True) |
| 36 | + |
| 37 | + |
10 | 38 | def bytes2CoseSign1(data: bytes) -> Sign1Message: |
11 | 39 | """ |
12 | 40 | Gets bytes and return a COSE_Sign1 object |
13 | 41 |
|
14 | | - :param data: bytes: the COSE Sign1 as bytes |
| 42 | + :param data: bytes: the COSE Sign1 as bytes (optionally CBOR-tagged 18) |
15 | 43 | :return: Sign1Message: the COSE Sign1 object |
16 | 44 | """ |
17 | | - decoded = Sign1Message.decode(cbor2.loads(data).value) |
| 45 | + loaded = cbor2.loads(data) |
| 46 | + if isinstance(loaded, cbor2.CBORTag): |
| 47 | + cose_obj = loaded.value |
| 48 | + else: |
| 49 | + cose_obj = loaded |
| 50 | + |
| 51 | + # Tag value may itself be the encoded array as bytes |
| 52 | + if isinstance(cose_obj, (bytes, bytearray)): |
| 53 | + cose_obj = cbor2.loads(cose_obj) |
18 | 54 |
|
19 | | - return decoded |
| 55 | + return _cose_sign1_from_obj(cose_obj) |
20 | 56 |
|
21 | 57 |
|
22 | | -def cborlist2CoseSign1(data: list) -> Sign1Message: |
| 58 | +def cborlist2CoseSign1(data: Union[list, tuple, Sequence]) -> Sign1Message: |
23 | 59 | """ |
24 | | - Gets cbor2 decoded COSE Sign1 as a list and return a COSE_Sign1 object |
| 60 | + Gets cbor2 decoded COSE Sign1 as a list/tuple and return a COSE_Sign1 object |
25 | 61 |
|
26 | | - :param data: list: the COSE Sign1 as a list |
| 62 | + :param data: list | tuple: the COSE Sign1 as a decoded array |
27 | 63 | :return: Sign1Message: the COSE Sign1 object |
28 | 64 | """ |
29 | | - decoded = Sign1Message.decode( |
30 | | - cbor2.dumps( |
31 | | - cbor2.CBORTag(18, value=data) |
32 | | - ) |
33 | | - ) |
34 | | - |
35 | | - return decoded |
| 65 | + return _cose_sign1_from_obj(data) |
36 | 66 |
|
37 | 67 |
|
38 | 68 | def pretty_print(cbor_loaded: dict) -> None: |
|
0 commit comments