feat(core): allow streaming authenticity proofs

diff --git a/core/src/apps/management/authenticate_device.py b/core/src/apps/management/authenticate_device.py
index 5a8712b7..9d6c8fab 100644
--- a/core/src/apps/management/authenticate_device.py
+++ b/core/src/apps/management/authenticate_device.py
@@ -1,15 +1,17 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from trezor.messages import AuthenticateDevice, AuthenticityProof
+    from buffer_types import AnyBytes
 
+    from trezor.messages import AuthenticateDevice, AuthenticityProof, Success
 
-async def authenticate_device(msg: AuthenticateDevice) -> AuthenticityProof:
+
+async def authenticate_device(msg: AuthenticateDevice) -> AuthenticityProof | Success:
     from trezor import TR, utils, wire
     from trezor.crypto import optiga
     from trezor.crypto.hashlib import sha256
     from trezor.loop import sleep
-    from trezor.messages import AuthenticityProof
+    from trezor.messages import AuthenticityProof, Success
     from trezor.ui.layouts import confirm_action
     from trezor.ui.layouts.progress import progress
     from trezor.utils import BufferReader, bootloader_locked
@@ -72,6 +74,74 @@ async def authenticate_device(msg: AuthenticateDevice) -> AuthenticityProof:
 
         spinner.report(1000)
 
+    if msg.stream:
+        from trezor.enums import AuthenticityProofType
+        from trezor.messages import (
+            AuthenticityProofChunk,
+            AuthenticityProofSizes,
+            GetAuthenticityProofChunk,
+        )
+        from trezor.wire import DataError
+        from trezor.wire.context import call
+
+        def _cert_sizes(certificates: list[AnyBytes] | None) -> list[int] | None:
+            if certificates is None:
+                return None
+            return [len(cert) for cert in certificates]
+
+        def _sig_size(signature: AnyBytes | None) -> int | None:
+            return len(signature) if signature is not None else None
+
+        req = await call(
+            AuthenticityProofSizes(
+                optiga_certificates=_cert_sizes(optiga_certificates),
+                tropic_certificates=_cert_sizes(tropic_certificates),
+                mcu_certificates=_cert_sizes(mcu_certificates),
+                optiga_signature=len(optiga_signature),
+                tropic_signature=_sig_size(tropic_signature),
+                mcu_signature=_sig_size(mcu_signature),
+            ),
+            GetAuthenticityProofChunk,
+        )
+
+        all_certificates: dict[AuthenticityProofType, list[AnyBytes] | None] = {
+            AuthenticityProofType.OPTIGA: optiga_certificates,
+            AuthenticityProofType.TROPIC: tropic_certificates,
+            AuthenticityProofType.MCU: mcu_certificates,
+        }
+        all_signatures: dict[AuthenticityProofType, AnyBytes | None] = {
+            AuthenticityProofType.OPTIGA: optiga_signature,
+            AuthenticityProofType.TROPIC: tropic_signature,
+            AuthenticityProofType.MCU: mcu_signature,
+        }
+
+        while req.proof_type is not None:
+
+            if req.index is None:
+                blob = all_signatures[req.proof_type]
+                if blob is None:
+                    raise DataError("No signature")
+            else:
+                certificates = all_certificates[req.proof_type] or []
+                if req.index >= len(certificates):
+                    raise DataError("No certificate")
+                blob = certificates[req.index]
+
+            blob = memoryview(blob)
+            blob_len = len(blob)
+            # `req.offset` and `req.size` cannot be negative (defined as `uint32`)
+            offset = req.offset
+            end = offset + req.size
+            if offset > blob_len or end > blob_len:
+                raise DataError("Invalid chunk range")
+            chunk = blob[offset:end]
+
+            resp = AuthenticityProofChunk(chunk=chunk)
+            req = await call(resp, GetAuthenticityProofChunk)
+
+        return Success()
+
+    # support non-chunked response for backwards compatibility
     return AuthenticityProof(
         optiga_certificates=optiga_certificates,
         optiga_signature=optiga_signature,
diff --git a/python/src/trezorlib/device.py b/python/src/trezorlib/device.py
index e172ab53..92afddef 100644
--- a/python/src/trezorlib/device.py
+++ b/python/src/trezorlib/device.py
@@ -18,16 +18,25 @@ from __future__ import annotations
 
 import hashlib
 import hmac
+import io
 import random
 import secrets
 import time
 import warnings
-from typing import TYPE_CHECKING, Callable, Iterable, Optional, Tuple
+from typing import (
+    TYPE_CHECKING,
+    Callable,
+    Iterable,
+    Optional,
+    Sequence,
+    Tuple,
+    overload,
+)
 
 from slip10 import SLIP10
 
 from . import messages
-from .exceptions import Cancelled, TrezorException
+from .exceptions import Cancelled, TrezorException, UnexpectedMessageError
 from .tools import Address, parse_path, workflow
 
 if TYPE_CHECKING:
@@ -630,12 +639,150 @@ def set_busy(session: "Session", expiry_ms: Optional[int]) -> None:
     session.call(messages.SetBusy(expiry_ms=expiry_ms), expect=messages.Success)
 
 
-@workflow()
-def authenticate(session: "Session", challenge: bytes) -> messages.AuthenticityProof:
-    return session.call(
-        messages.AuthenticateDevice(challenge=challenge),
-        expect=messages.AuthenticityProof,
+# AuthenticityProof-related data cannot be too large.
+PROOF_BLOB_SIZE_LIMIT = 100 * 1024
+PROOF_CERTS_LEN_LIMIT = 10
+
+
+def _fetch_proof_chunks(
+    session: "Session",
+    chunk_size: int,
+    proof_type: messages.AuthenticityProofType,
+    index: int | None,
+    size: int,
+) -> bytes:
+    result = io.BytesIO()
+    if size > PROOF_BLOB_SIZE_LIMIT:
+        raise ValueError("Unexpected blob size")
+
+    while size > 0:
+        req_size = min(size, chunk_size)
+        resp = session.call(
+            messages.GetAuthenticityProofChunk(
+                proof_type=proof_type,
+                index=index,
+                offset=result.tell(),
+                size=req_size,
+            ),
+            expect=messages.AuthenticityProofChunk,
+        )
+        if req_size != len(resp.chunk):
+            raise ValueError("Unexpected response size")
+
+        size -= len(resp.chunk)
+        result.write(resp.chunk)
+
+    assert size == 0
+    return result.getvalue()
+
+
+if TYPE_CHECKING:
+    # the fetched `signature` is None iff `signature_size` parameter is None.
+    # (needed since `optiga_signature` field is required)
+
+    @overload
+    def _fetch_proof_part(
+        session: "Session",
+        chunk_size: int,
+        part: messages.AuthenticityProofType,
+        signature_size: int,
+        certificate_sizes: Sequence[int] | None,
+    ) -> tuple[bytes, list[bytes]]: ...
+
+    @overload
+    def _fetch_proof_part(
+        session: "Session",
+        chunk_size: int,
+        part: messages.AuthenticityProofType,
+        signature_size: None,
+        certificate_sizes: Sequence[int] | None,
+    ) -> tuple[None, list[bytes]]: ...
+
+
+def _fetch_proof_part(
+    session: "Session",
+    chunk_size: int,
+    part: messages.AuthenticityProofType,
+    signature_size: int | None,
+    certificate_sizes: Sequence[int] | None,
+) -> tuple[bytes | None, list[bytes]]:
+    signature = None
+    if signature_size is not None:
+        signature = _fetch_proof_chunks(
+            session, chunk_size, part, index=None, size=signature_size
+        )
+
+    certificate_sizes = certificate_sizes or []
+    if len(certificate_sizes) > PROOF_CERTS_LEN_LIMIT:
+        raise ValueError("Too many certificates")
+
+    certificates = [
+        _fetch_proof_chunks(session, chunk_size, part, index, size)
+        for index, size in enumerate(certificate_sizes)
+    ]
+    return (signature, certificates)
+
+
+def _fetch_proof(
+    session: "Session",
+    chunk_size: int,
+    sizes: messages.AuthenticityProofSizes,
+) -> messages.AuthenticityProof:
+    optiga_signature, optiga_certificates = _fetch_proof_part(
+        session,
+        chunk_size,
+        part=messages.AuthenticityProofType.OPTIGA,
+        signature_size=sizes.optiga_signature,
+        certificate_sizes=sizes.optiga_certificates,
+    )
+    tropic_signature, tropic_certificates = _fetch_proof_part(
+        session,
+        chunk_size,
+        part=messages.AuthenticityProofType.TROPIC,
+        signature_size=sizes.tropic_signature,
+        certificate_sizes=sizes.tropic_certificates,
     )
+    mcu_signature, mcu_certificates = _fetch_proof_part(
+        session,
+        chunk_size,
+        part=messages.AuthenticityProofType.MCU,
+        signature_size=sizes.mcu_signature,
+        certificate_sizes=sizes.mcu_certificates,
+    )
+    return messages.AuthenticityProof(
+        optiga_certificates=optiga_certificates,
+        optiga_signature=optiga_signature,
+        tropic_certificates=tropic_certificates,
+        tropic_signature=tropic_signature,
+        mcu_certificates=mcu_certificates,
+        mcu_signature=mcu_signature,
+    )
+
+
+@workflow()
+def authenticate(
+    session: "Session", challenge: bytes, chunk_size: int = 1024
+) -> messages.AuthenticityProof:
+    stream = chunk_size > 0
+    try:
+        sizes = session.call(
+            messages.AuthenticateDevice(challenge=challenge, stream=stream),
+            expect=messages.AuthenticityProofSizes,
+        )
+    except UnexpectedMessageError as exc:
+        # support older FW versions (without streaming support)
+        if isinstance(exc.actual, messages.AuthenticityProof):
+            return exc.actual
+        raise
+
+    try:
+        return _fetch_proof(session, chunk_size, sizes)
+    finally:
+        # stop the workflow using `proof_type=None`.
+        session.call(
+            messages.GetAuthenticityProofChunk(proof_type=None, offset=0, size=0),
+            expect=messages.Success,
+        )
 
 
 @workflow()