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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion pymdoccbor/mdoc/issuer.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

from pymdoccbor.mdoc.exceptions import InvalidStatusDescriptor
from pymdoccbor.mso.issuer import MsoIssuer
from pymdoccbor.tools import thaw_cbor

logger = logging.getLogger("pymdoccbor")

Expand Down Expand Up @@ -228,7 +229,8 @@ def new(
ns: [v for k, v in dgst.items()]
for ns, dgst in msoi.disclosure_map.items()
},
"issuerAuth": cbor2.loads(mso_cbor)
# thaw_cbor: cbor2 >= 6 returns tuple/frozendict
"issuerAuth": thaw_cbor(cbor2.loads(mso_cbor))
},
}
],
Expand Down
8 changes: 6 additions & 2 deletions pymdoccbor/mdoc/issuersigned.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,12 +24,16 @@ class IssuerSigned:
]
"""

def __init__(self, nameSpaces: dict, issuerAuth: Union[cbor2.CBORTag, dict, bytes]) -> None:
def __init__(
self,
nameSpaces: dict,
issuerAuth: Union[cbor2.CBORTag, list, tuple, dict, bytes],
) -> None:
"""
Initialize the IssuerSigned object

:param nameSpaces: dict: the nameSpaces of the document
:param issuerAuth: Union[dict, bytes]: the issuerAuth info of the document
:param issuerAuth: COSE_Sign1 as list/tuple/bytes/CBORTag(18)
"""

self.namespaces: dict = nameSpaces
Expand Down
21 changes: 15 additions & 6 deletions pymdoccbor/mso/issuer.py
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,19 @@ def format_datetime_repr(self, dt: datetime.datetime) -> str:
"""
return dt.isoformat().split(".")[0] + "Z"

@staticmethod
def parse_date(date_str: str, *, end_of_day: bool = False) -> datetime.datetime:
"""
Parse a date-only string (YYYY-MM-DD) into a datetime.

Date-only expiry values are interpreted as end-of-day so that
validityInfo.signed remains within [validFrom, validUntil].
"""
dt = datetime.datetime.strptime(date_str, "%Y-%m-%d")
if end_of_day:
dt = dt.replace(hour=23, minute=59, second=59)
return dt

def sign(
self,
device_key: dict | None = None,
Expand All @@ -181,16 +194,12 @@ def sign(
"""

utcnow = datetime.datetime.utcnow()
valid_from = datetime.datetime.strptime(
self.validity["issuance_date"], "%Y-%m-%d"
)
valid_from = self.parse_date(self.validity["issuance_date"])

if settings.PYMDOC_EXP_DELTA_HOURS:
exp = utcnow + datetime.timedelta(hours=settings.PYMDOC_EXP_DELTA_HOURS)
else:
# five years
exp = datetime.datetime.strptime(self.validity["expiry_date"], "%Y-%m-%d")
# exp = utcnow + datetime.timedelta(hours=(24 * 365) * 5)
exp = self.parse_date(self.validity["expiry_date"], end_of_day=True)

if utcnow > valid_from:
valid_from = utcnow
Expand Down
18 changes: 11 additions & 7 deletions pymdoccbor/mso/verifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import hashlib
import logging
from datetime import datetime, timezone
from typing import Union
from typing import Sequence, Union

import cbor2
import cryptography
Expand Down Expand Up @@ -31,30 +31,34 @@ class MsoVerifier:
structure as defined in RFC 8152.
"""

def __init__(self, data: Union[cbor2.CBORTag, bytes, list]) -> None:
def __init__(
self, data: Union[cbor2.CBORTag, bytes, list, tuple, Sequence]
) -> None:
"""
Initialize the MsoParser object

:param data: Union[cbor2.CBORTag, bytes, list]: the data to parse
:param data: Union[cbor2.CBORTag, bytes, list, tuple]: the data to parse
"""

self._data = data

if isinstance(self._data, bytes):
self.object: Sign1Message = bytes2CoseSign1(
cbor2.dumps(cbor2.CBORTag(18, value=self._data)))
elif isinstance(self._data, list):
elif isinstance(self._data, (list, tuple)):
# cbor2 >= 6 decodes arrays as tuple
self.object: Sign1Message = cborlist2CoseSign1(self._data)
elif isinstance(self._data, cbor2.CBORTag) and self._data.tag == 18:
# COSE_Sign1 is CBOR tag 18; value can be list (decoded) or bytes
# COSE_Sign1 is CBOR tag 18; value can be list/tuple (decoded) or bytes
val = self._data.value
if isinstance(val, list):
if isinstance(val, (list, tuple)):
self.object = cborlist2CoseSign1(val)
else:
self.object = bytes2CoseSign1(cbor2.dumps(self._data))
else:
raise UnsupportedMsoDataFormat(
f"MsoParser only supports raw bytes, list, or CBORTag(18); got {type(data)}"
f"MsoParser only supports raw bytes, list, tuple, or "
f"CBORTag(18); got {type(data)}"
)

self.object.key = None
Expand Down
4 changes: 3 additions & 1 deletion pymdoccbor/tests/test_02_mdoc_issuer.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from pymdoccbor.tests.cert_data import CERT_DATA
from pymdoccbor.tests.pid_data import PID_DATA
from pymdoccbor.tests.pkey import PKEY, PKEY_ED25519, PKEY_RSA
from pymdoccbor.tools import bytes2CoseSign1


def extract_mso(mdoc: dict):
Expand Down Expand Up @@ -39,7 +40,8 @@ def test_mso_writer():

mso = msoi.sign()

Sign1Message.decode(mso.encode())
# Use our helper: pycose Sign1Message.decode breaks on cbor2 >= 6 (tuple)
assert isinstance(bytes2CoseSign1(mso.encode()), Sign1Message)


def test_mdoc_issuer():
Expand Down
33 changes: 33 additions & 0 deletions pymdoccbor/tests/test_06_mso_issuer.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,6 @@
import datetime

import cbor2
from pycose.messages import CoseMessage

from pymdoccbor.mso.issuer import MsoIssuer
Expand Down Expand Up @@ -46,3 +49,33 @@ def test_mso_issuer_sign():

mso = msoi.sign()
assert isinstance(mso, CoseMessage)


def test_mso_issuer_validity_same_day():
today = datetime.datetime.utcnow().strftime("%Y-%m-%d")
msoi = MsoIssuer(
data=MICOV_DATA,
private_key=PKEY,
validity={"issuance_date": today, "expiry_date": today},
alg="ES256",
cert_info=CERT_DATA,
)

mso = msoi.sign()
payload = cbor2.loads(mso.payload)
mso_body = cbor2.loads(payload.value)
validity = mso_body["validityInfo"]

def _as_utc(dt):
if isinstance(dt, cbor2.CBORTag):
dt = dt.value
if isinstance(dt, str):
return datetime.datetime.fromisoformat(dt.replace("Z", "+00:00"))
return dt.replace(tzinfo=datetime.timezone.utc)

signed = _as_utc(validity["signed"])
valid_from = _as_utc(validity["validFrom"])
valid_until = _as_utc(validity["validUntil"])

assert valid_from <= signed <= valid_until
assert valid_until.hour == 23 and valid_until.minute == 59 and valid_until.second == 59
59 changes: 59 additions & 0 deletions pymdoccbor/tests/test_11_cbor2_compat.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
"""Compatibility with cbor2 >= 6 (immutable tuple / frozendict decode)."""
import cbor2
from pycose.messages import Sign1Message

from pymdoccbor.mdoc.issuer import MdocCborIssuer
from pymdoccbor.mdoc.verifier import MobileDocument
from pymdoccbor.mso.verifier import MsoVerifier
from pymdoccbor.tests.cert_data import CERT_DATA
from pymdoccbor.tests.micov_data import MICOV_DATA
from pymdoccbor.tests.pkey import PKEY
from pymdoccbor.tools import cborlist2CoseSign1, thaw_cbor


def test_thaw_cbor_tuple_and_mapping():
tagged = cbor2.dumps(cbor2.CBORTag(18, [b"\xa1\x01&", {33: b"cert"}, b"pay", b"sig"]))
loaded = cbor2.loads(tagged)
# cbor2 5 -> list/dict; cbor2 6 -> tuple/frozendict
thawed = thaw_cbor(loaded)
assert isinstance(thawed, cbor2.CBORTag)
assert isinstance(thawed.value, list)
assert isinstance(thawed.value[1], dict)
assert thawed.value[1][33] == b"cert"


def test_cborlist2CoseSign1_accepts_tuple():
cose_tuple = (b"\xa1\x01&", {33: b"cert"}, b"pay", b"sig")
msg = cborlist2CoseSign1(cose_tuple)
assert isinstance(msg, Sign1Message)
assert msg.payload == b"pay"


def test_mso_verifier_with_cbor2_loads_roundtrip():
mdoc = MdocCborIssuer(private_key=PKEY, alg="ES256", cert_info=CERT_DATA)
mdoc.new(
data=MICOV_DATA,
doctype="org.micov.medical.1",
validity={"issuance_date": "2024-12-31", "expiry_date": "2050-12-31"},
)
issuer_auth = mdoc.signed["documents"][0]["issuerSigned"]["issuerAuth"]
assert isinstance(issuer_auth, list)

# Simulate cbor2 >= 6 decode of untagged COSE_Sign1
reloaded = cbor2.loads(cbor2.dumps(issuer_auth))
msov = MsoVerifier(reloaded)
assert isinstance(msov.object, Sign1Message)
assert msov.verify_signature()


def test_mobile_document_from_issuer_signed():
mdoc = MdocCborIssuer(private_key=PKEY, alg="ES256", cert_info=CERT_DATA)
mdoc.new(
data=MICOV_DATA,
doctype="org.micov.medical.1",
validity={"issuance_date": "2024-12-31", "expiry_date": "2050-12-31"},
)
document = mdoc.signed["documents"][0]
doc = MobileDocument(**document)
assert doc.doctype == "org.micov.medical.1"
assert doc.errors == {}
56 changes: 43 additions & 13 deletions pymdoccbor/tools.py
Original file line number Diff line number Diff line change
@@ -1,38 +1,68 @@
# Aligns with https://github.com/eu-digital-identity-wallet/pyMDOC-CBOR
import json
import random
from collections.abc import Mapping, Sequence
from typing import Any, Union

import cbor2
from cbor2.tool import DefaultEncoder, key_to_str
from pycose.messages import Sign1Message


def thaw_cbor(obj: Any) -> Any:
"""
Convert cbor2 6.x immutable decode results into mutable structures.

cbor2 >= 6 returns tuple for arrays and frozendict for maps. pycose 1.x
requires a mutable list (it uses .pop) and a plain dict for COSE headers.
"""
if isinstance(obj, cbor2.CBORTag):
return cbor2.CBORTag(obj.tag, thaw_cbor(obj.value))
if isinstance(obj, Mapping):
return {thaw_cbor(k): thaw_cbor(v) for k, v in obj.items()}
if isinstance(obj, tuple):
return [thaw_cbor(item) for item in obj]
if isinstance(obj, list):
return [thaw_cbor(item) for item in obj]
return obj


def _cose_sign1_from_obj(cose_obj: Any) -> Sign1Message:
"""Build a Sign1Message from a decoded COSE_Sign1 array (list or tuple)."""
cose_obj = thaw_cbor(cose_obj)
if not isinstance(cose_obj, list):
raise TypeError("Bytes cannot be decoded as COSE message")
return Sign1Message.from_cose_obj(cose_obj, True)


def bytes2CoseSign1(data: bytes) -> Sign1Message:
"""
Gets bytes and return a COSE_Sign1 object

:param data: bytes: the COSE Sign1 as bytes
:param data: bytes: the COSE Sign1 as bytes (optionally CBOR-tagged 18)
:return: Sign1Message: the COSE Sign1 object
"""
decoded = Sign1Message.decode(cbor2.loads(data).value)
loaded = cbor2.loads(data)
if isinstance(loaded, cbor2.CBORTag):
cose_obj = loaded.value
else:
cose_obj = loaded

# Tag value may itself be the encoded array as bytes
if isinstance(cose_obj, (bytes, bytearray)):
cose_obj = cbor2.loads(cose_obj)

return decoded
return _cose_sign1_from_obj(cose_obj)


def cborlist2CoseSign1(data: list) -> Sign1Message:
def cborlist2CoseSign1(data: Union[list, tuple, Sequence]) -> Sign1Message:
"""
Gets cbor2 decoded COSE Sign1 as a list and return a COSE_Sign1 object
Gets cbor2 decoded COSE Sign1 as a list/tuple and return a COSE_Sign1 object

:param data: list: the COSE Sign1 as a list
:param data: list | tuple: the COSE Sign1 as a decoded array
:return: Sign1Message: the COSE Sign1 object
"""
decoded = Sign1Message.decode(
cbor2.dumps(
cbor2.CBORTag(18, value=data)
)
)

return decoded
return _cose_sign1_from_obj(data)


def pretty_print(cbor_loaded: dict) -> None:
Expand Down
Loading