Skip to content
Open
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
17 changes: 17 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,23 @@

### Fixed

- **[SECURITY][SDK]** `attach_receipt()`, `attach_attestation()`, and
`attach_approvals()` could crash with `cbor2.CBOREncodeError` instead of
the documented `CoseError` on a malformed COSE envelope. `_decode_tagged()`
checked that the unprotected header was a CBOR map but never inspected
its contents; a stray break byte (`0xff`) in the wrong position decodes
into cbor2's internal break-marker sentinel rather than raising, and that
sentinel travelled through `attach_unprotected()` undetected until the
re-encode at the end, where cbor2 cannot serialize it. Same underlying
cbor2 quirk as the signature-slot check `_decode_tagged()` already applies
to `body[3]`; this closes the same hole for `body[1]`'s contents.
`_decode_tagged()` now recursively rejects the sentinel anywhere in the
unprotected header - inside a `Mapping`/`list`/`tuple`, inside a
`CBORTag` (an unrecognised semantic tag), and inside a `set`/`frozenset`
(cbor2 auto-decodes tag 258 to a `set`) - before any caller touches it,
and `attach_unprotected()` converts any residual `CBOREncodeError` to
`CoseStructureError` as a second layer.

- **[SECURITY][SDK]** `_strict_schema_violations()` tolerated a missing
top-level `issuer` claim for **any** manifest version, not just v0.1. The
exception was added for legacy v0.1 records, which predate the `issuer`
Expand Down
64 changes: 63 additions & 1 deletion python/src/agent_manifest/_cose.py
Original file line number Diff line number Diff line change
Expand Up @@ -409,7 +409,16 @@ def attach_unprotected(cose_bytes: bytes, label: Union[int, str], value: Any) ->
unprotected[label] = value
body = list(body)
body[1] = unprotected
return cbor2.dumps(cbor2.CBORTag(tag, body), canonical=True)
try:
return cbor2.dumps(cbor2.CBORTag(tag, body), canonical=True)
except cbor2.CBOREncodeError as exc:
# Belt-and-suspenders: _decode_tagged() already rejects the known
# undecodable sentinel before we get here. This catches anything
# else that slips through - a different cbor2 version, a value this
# function itself added - so the documented contract ("this raises
# CoseError, nothing else") holds even if the front-door check ever
# misses something.
raise CoseStructureError(f"unprotected header is not re-encodable: {exc}") from exc


def attach_receipt(cose_bytes: bytes, receipt: bytes) -> bytes:
Expand Down Expand Up @@ -516,6 +525,54 @@ def _plain(value: Any) -> Any:
return value


def _reject_cbor_sentinels(value: Any, *, what: str) -> None:
"""Recursively reject cbor2's internal break-marker sentinel.

A stray break byte (``0xff``) in the wrong position does not always make
cbor2 raise during decode: depending on where it lands, cbor2 hands back
its own internal marker object instead - a bare ``object()``, never a
subclass, never something a normal decode produces otherwise. That
marker cannot be re-encoded, so anything holding one dies later with
``cbor2.CBOREncodeError`` when a caller (``attach_unprotected`` and
everything built on it) tries to write the structure back out.

``body[3]`` (the signature slot) is checked per-tag in ``_decode_tagged``
itself; this function closes the same hole for the unprotected header,
whose *values* were never inspected past "is it a Mapping". It matches on
the exact marker type - ``type(value) is object`` - rather than
allow-listing legitimate types, so it can never reject a manifest that
decodes normally, no matter which types cbor2's tag support adds in a
future version (``UUID``, ``Decimal``, ``datetime`` and friends already
round-trip fine and are left untouched).

Every CBOR container type that can hold a nested value is walked, not
just the ones the unprotected header's own top-level shape happens to
use: ``Mapping`` (both keys and values - cbor2 6.x hands back an
immutable ``frozendict``, itself a ``Mapping``), ``list``/``tuple``
(array), ``set``/``frozenset`` (cbor2 auto-decodes semantic tag 258 to a
plain ``set``), and ``cbor2.CBORTag`` (any tag cbor2 does *not* have a
built-in decoder for is handed back as a ``CBORTag`` wrapping its
payload, e.g. an unrecognised or future semantic tag). A sentinel can be
tucked inside any of these - ``{1: CBORTag(9999, [sentinel])}`` or
``{1: {sentinel}}`` decode without error and without matching the old
Mapping/list/tuple-only check - so all of them are recursed into.
"""
if type(value) is object:
raise CoseStructureError(
f"{what} contains an undecodable CBOR value "
"(a malformed indefinite-length break byte)"
)
if isinstance(value, Mapping):
for key, item in value.items():
_reject_cbor_sentinels(key, what=what)
_reject_cbor_sentinels(item, what=what)
elif isinstance(value, cbor2.CBORTag):
_reject_cbor_sentinels(value.value, what=what)
elif isinstance(value, (list, tuple, set, frozenset)):
for item in value:
_reject_cbor_sentinels(item, what=what)


def _decode_tagged(cose_bytes: bytes) -> tuple[int, list[Any]]:
"""Decode exactly one tagged COSE object and return ``(tag, body)``.

Expand Down Expand Up @@ -563,6 +620,11 @@ def _decode_tagged(cose_bytes: bytes) -> tuple[int, list[Any]]:
# cbor2 hands back an immutable mapping for a map inside a tag.
if not isinstance(body[1], Mapping):
raise CoseStructureError("unprotected header must be a map")
# Being a Mapping only proves the outer shape; a key or value inside it
# can still be cbor2's undecodable break-marker sentinel (see
# _reject_cbor_sentinels). Rejected here, at parse time, rather than
# letting it travel into attach_unprotected() and die on re-encode.
_reject_cbor_sentinels(body[1], what="unprotected header")
if not isinstance(body[2], bytes):
raise CoseStructureError("payload must be a byte string, inline not detached")
# The signature slot was the one element never type-checked here, which let
Expand Down
161 changes: 161 additions & 0 deletions python/tests/test_cose.py
Original file line number Diff line number Diff line change
Expand Up @@ -1542,3 +1542,164 @@ def test_signature_slot_type_is_checked_at_decode():
):
with pytest.raises(CoseError, match="signature must be a byte string"):
call()


def test_unprotected_header_sentinel_is_checked_at_decode():
"""A value inside the unprotected header cannot be an undecodable sentinel.

Found by ``fuzz_cose`` (ClusterFuzzLite). Same underlying cbor2 quirk as
``test_signature_slot_type_is_checked_at_decode`` above - a stray break
byte decodes into cbor2's internal break marker, a bare ``object()`` -
but this time the marker lands as a *value inside the unprotected header
map* rather than in the signature slot. ``_decode_tagged`` checked that
the unprotected header was a ``Mapping`` but never inspected what was
inside it, so this envelope reached ``attach_unprotected``, which copies
the map and re-encodes it, and died there with ``CBOREncodeError``
instead of the ``CoseError`` every caller is written against.

The bytes below are the fuzzer's own reproducer (with the trailing byte
atheris's ``FuzzedDataProvider`` consumed for its ``choice`` selector
already stripped off), kept verbatim.
"""
envelope = bytes.fromhex(
"d28443cbffffa5a032d825500000cbffffa5a032d825500000ff407fff"
"0000000041a0a04040"
)

for call in (
lambda: attach_receipt(envelope, b"\xa0"),
lambda: attach_unprotected(envelope, 1, b"x"),
lambda: attach_attestation(envelope, {"platform": "x"}),
lambda: attach_approvals(envelope, [{"approver_id": "a"}]),
):
with pytest.raises(CoseStructureError, match="undecodable CBOR value"):
call()

# decode_cose_manifest never re-encodes the unprotected header, so this
# same envelope is expected to fail for an unrelated, earlier reason
# (there is no valid JSON payload here) rather than leak anything.
with pytest.raises(CoseError):
decode_cose_manifest(envelope)


def test_attach_unprotected_rejects_a_caller_supplied_unencodable_value():
"""``attach_unprotected`` also guards against a *caller's own* bad value.

``_decode_tagged`` only validates what came from *cose_bytes*; the value
a caller passes in to attach is never inspected before being merged into
the header and re-encoded. A plain, non-CBOR-encodable Python object
there hits the same ``cbor2.CBOREncodeError`` at re-encode time, so
``attach_unprotected``'s belt-and-suspenders ``except`` clause is the
only thing standing between a caller mistake and a leaked library
exception. This is that mistake, made deliberately.
"""

class NotCborEncodable:
pass

signed = sign_cose_sign1(base_manifest(), KP)
with pytest.raises(CoseStructureError, match="not re-encodable"):
attach_attestation(signed, NotCborEncodable())


def test_reject_cbor_sentinels_recurses_into_cbortag_and_set():
"""``_reject_cbor_sentinels`` must walk every CBOR container, not just
``Mapping``/``list``/``tuple``.

Code review on the PR that introduced this check found that it stopped
at the boundary of ``cbor2.CBORTag`` and ``set``/``frozenset``: cbor2
hands back a ``CBORTag`` for any semantic tag it has no built-in decoder
for, and auto-decodes tag 258 to a plain ``set``. A sentinel tucked
inside either of those - ``CBORTag(9999, [sentinel])`` or ``{sentinel}``
- decoded without error and without being caught, so it would reach
``attach_unprotected`` and die on re-encode, or (worse - see the next
test) sail straight through ``decode_cose_manifest`` with nothing raised
at all. This exercises the fixed recursion directly against every shape
it now covers.
"""
from agent_manifest._cose import _reject_cbor_sentinels

sentinel = object()
rejected_cases = {
"sentinel inside an unsupported CBORTag": cbor2.CBORTag(9999, [sentinel]),
"sentinel inside a CBORTag(258) wrapping a set-shaped payload": cbor2.CBORTag(
258, [sentinel]
),
"sentinel inside a bare set": {sentinel},
"sentinel inside a bare frozenset": frozenset([sentinel]),
"sentinel doubly nested: list -> CBORTag -> tuple": [
cbor2.CBORTag(5, (sentinel,))
],
"sentinel as a dict key inside a CBORTag": cbor2.CBORTag(7, {sentinel: 1}),
}
for value in rejected_cases.values():
with pytest.raises(CoseStructureError, match="undecodable CBOR value"):
_reject_cbor_sentinels({1: value}, what="unprotected header")

# A legitimate header using the same container types, with no sentinel
# anywhere in it, must not be rejected.
legitimate = {
1: cbor2.CBORTag(9999, ["a", "b", {2: 3}]),
2: {1, 2, 3},
3: frozenset({"x", "y"}),
4: [1, 2, {"nested": True}],
}
_reject_cbor_sentinels(legitimate, what="unprotected header") # must not raise


def test_unprotected_header_sentinel_nested_in_tag_or_set_is_checked_at_decode():
"""The sentinel check must catch a sentinel nested inside a tag or set
on a *real* envelope, not just when the recursive helper is called
directly - and, critically, before ``decode_cose_manifest`` /
``verify_cose_manifest`` hand the header back to a caller.

Hand-crafting the exact malformed CBOR bytes that make cbor2 place its
internal sentinel several containers deep (rather than as a direct
header value, which is what the fuzzer's reproducer above already
covers) is not practical without re-running the fuzzer. What can be
reproduced deterministically is the resulting Python object graph: this
patches ``cbor2.CBORDecoder.decode`` to return that exact graph - a
signed, otherwise fully valid envelope whose unprotected header carries
a sentinel inside ``CBORTag(258, [sentinel])`` - while still consuming
the real byte stream first, so every other check downstream of decoding
runs against a genuine envelope.

Before the fix, ``decode_cose_manifest``/``verify_cose_manifest`` raised
nothing at all here: they never re-encode the unprotected header, so the
old check's blind spot let the sentinel travel all the way into the
returned ``CoseVerification.unprotected``, silently, as a raw
unserializable ``cbor2.CBORTag``/``object()`` a caller had no reason to
expect.
"""
from unittest.mock import patch

signed = sign_cose_sign1(base_manifest(), KP)
signed = attach_attestation(signed, {"platform": "sim"})

decoded = cbor2.loads(signed)
tag, body = decoded.tag, list(decoded.value)
sentinel = object()
malicious_unprotected = dict(body[1])
malicious_unprotected[999] = cbor2.CBORTag(258, [sentinel])
body[1] = malicious_unprotected
fake_decoded = cbor2.CBORTag(tag, tuple(body))

real_decode = cbor2.CBORDecoder.decode

def smuggle_sentinel(self):
real_decode(self) # fully consume the real stream first
return fake_decoded

with patch.object(cbor2.CBORDecoder, "decode", smuggle_sentinel):
with pytest.raises(CoseStructureError, match="undecodable CBOR value"):
decode_cose_manifest(signed)
with pytest.raises(CoseStructureError, match="undecodable CBOR value"):
verify_cose_manifest(signed, trusted_keys=TRUSTED_KEYS)
for call in (
lambda: attach_unprotected(signed, 1, b"x"),
lambda: attach_receipt(signed, b"\xa0"),
lambda: attach_attestation(signed, {"platform": "x"}),
lambda: attach_approvals(signed, [{"approver_id": "a"}]),
):
with pytest.raises(CoseStructureError, match="undecodable CBOR value"):
call()
Loading