Repository: localstack/localstack
Pull request: fix CBOR timestamp parsing
## Motivation
Kinesis supports _CBOR_ (Concise Binary Object Representation) encoding as `Content-Type` in addition to JSON.
CBOR support was added for LocalStack with #6494 by extending our ASF parsers and serializers, and patching CBOR support into `botocore` (when being used as a proxy to `kinesis-mock`).
Unfortunately, AWS does not properly implement the CBOR spec when it comes to timestamp handling:
- [RFC8949](https://datatracker.ietf.org/doc/html/rfc8949) defines CBOR timestamps as floating-point epoch seconds:
  > The tag content MUST be an unsigned or negative integer (major types 0 and 1) or a floating-point number (major type 7 with additional information 25, 26, or 27).
  > ...
  > To indicate fractional seconds, floating-point values can be used within tag number 1 instead of integer values.
- AWS however uses milliseconds as nicely described in this issue: https://github.com/aws/aws-sdk-java-v2/issues/4661
  - This means that when using a standard-compliant parser, the timestamps multiplied by 1000 (resulting in an error - "ValueError: year 56474 is out of range"):
    ```
    2024-07-03T13:30:33.095 ERROR --- [et.reactor-1] l.aws.handlers.logging     : exception during call chain
    ValueError: year 56474 is out of range
    
    The above exception was the direct cause of the following exception:
    
    Traceback (most recent call last):
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/protocol/parser.py", line 871, in _parse_body_as_json
        return cbor2.loads(body_contents)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^
    _cbor2.CBORDecodeValueError: error decoding datetime from epoch
    
    The above exception was the direct cause of the following exception:
    
    Traceback (most recent call last):
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/rolo/gateway/chain.py", line 166, in handle
        handler(self, self.context, response)
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/handlers/service.py", line 63, in __call__
        return self.parse_and_enrich(context)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/handlers/service.py", line 67, in parse_and_enrich
        operation, instance = parser.parse(context.request)
                              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/protocol/parser.py", line 172, in wrapper
        return func(*args, **kwargs)
               ^^^^^^^^^^^^^^^^^^^^^
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/protocol/parser.py", line 921, in parse
        final_parsed = self._do_parse(request, shape, uri_params)
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/protocol/parser.py", line 933, in _do_parse
        parsed = self._handle_json_body(request, shape, uri_params)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/protocol/parser.py", line 945, in _handle_json_body
        parsed_json = self._parse_body_as_json(request)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      File "/opt/code/localstack/.venv/lib/python3.11/site-packages/localstack/aws/protocol/parser.py", line 873, in _parse_body_as_json
        raise ProtocolParserError("HTTP body could not be parsed as CBOR.") from e
    localstack.aws.protocol.parser.ProtocolParserError: HTTP body could not be parsed as CBOR.
    2024-07-03T13:30:33.097  INFO --- [et.reactor-1] localstack.request.http    : POST / => 500
    2024-07-03T13:30:33.151 ERROR --- [t.reactor-11] l.aws.handlers.logging     : exception during call chain
    ```

This already caused a few issues in the past, which were addressed with #6791 and #11071.
But these fixes are only working if the datetime fields are serialized as numbers instead of native timestamps.
The AWS SDK does perform a native timestamp serialization, which again leads to parsing errors (mentioned above, see reproducer linked in "Testing").

This PR addresses the CBOR timestamp parsing/serialization issues directly when using `cbor2` to en-/decode the binary data.

Unfortunately, cbor2 does not easily allow patching standard en-/decoder functions, which is why the fix only works if the Python-native cbor loading code is explicitly imported from the respective private modules (happy for any input on how to make this more stable / better): https://github.com/agronholm/cbor2/issues/192

## Changes
- Patches `cbor2` datetime encoding and decoding.
- Adjusts all imports of `cbor2` to explicitly use the Python implementation instead of the C / native library.
- Fixes the parser (removes an invalid assumption that binary data is not also base64 encoded).
- Fixes the serializer (removed timestamp handling in the serializer - done natively by `cbor2` now).

## Testing
- I created a reproducer using the AWS Java SDK to showcase this issue: https://github.com/alexrashed/kinesis-cbor-reproducer and verified the changes here with this reproducer / the AWS Java SDK.
- I added an AWS-verified plain-request CBOR integration test, testing CBOR serialization and deserialization.

Fixes #9699

Problem statement:
HTTP body could not be parsed as CBOR
When using apache flink and running the code locally, below error happened  in the localstack from docker console - 

localstack.aws.protocol.parser.ProtocolParserError: HTTP body could not be parsed as CBOR.
localstack-main  | 2023-11-21T05:35:06.808  INFO --- [   asgi_gw_2] localstack.request.http    : POST / => 500
localstack-main  | 2023-11-21T05:35:06.908 ERROR --- [   asgi_gw_6] l.aws.handlers.logging     : exception during call chain
localstack-main  | Traceback (most recent call last):
localstack-main  |   File "/opt/code/localstack/localstack/aws/protocol/parser.py", line 862, in _parse_body_as_json
localstack-main  |     return cbor2.loads(body_contents)
localstack-main  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^
localstack-main  | ValueError: year 55858 is out of range
localstack-main  |
localstack-main  | The above exception was the direct cause of the following exception:
localstack-main  |
localstack-main  | Traceback (most recent call last):
localstack-main  |   File "/opt/code/localstack/localstack/aws/chain.py", line 90, in handle
localstack-main  |     handler(self, self.context, response)
localstack-main  |   File "/opt/code/localstack/localstack/aws/handlers/service.py", line 64, in __call__
localstack-main  |     return self.parse_and_enrich(context)
localstack-main  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
localstack-main  |   File "/opt/code/localstack/localstack/aws/handlers/service.py", line 77, in parse_and_enrich
localstack-main  |     operation, instance = parser.parse(context.request)
localstack-main  |                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
localstack-main  |   File "/opt/code/localstack/localstack/aws/protocol/parser.py", line 172, in wrapper
localstack-main  |     return func(*args, **kwargs)
localstack-main  |            ^^^^^^^^^^^^^^^^^^^^^
localstack-main  |   File "/opt/code/localstack/localstack/aws/protocol/parser.py", line 903, in parse
localstack-main  |     final_parsed = self._do_parse(request, shape, uri_params)
localstack-main  |                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
localstack-main  |   File "/opt/code/localstack/localstack/aws/protocol/parser.py", line 915, in _do_parse
localstack-main  |     parsed = self._handle_json_body(request, shape, uri_params)
localstack-main  |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
localstack-main  |   File "/opt/code/localstack/localstack/aws/protocol/parser.py", line 927, in _handle_json_body
localstack-main  |     parsed_json = self._parse_body_as_json(request)
localstack-main  |                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Program arguments - 
-aws.region us-east-1 -aws.endpoint localhost.localstack.cloud:4566 -kinesis.source.stream ExampleInputStream -kinesis.sink.stream ExampleOutputStream


Error at the client side - 


Caused by: java.lang.RuntimeException: Retries exceeded for getShardIterator operation - all 3 retry attempts failed.
	at org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy.getShardIterator(KinesisProxy.java:409)
	at org.apache.flink.streaming.connectors.kinesis.proxy.KinesisProxy.getShardIterator(KinesisProxy.java:374)
	at org.apache.flink.streaming.connectors.kinesis.internals.publisher.polling.PollingRecordPublisher.getShardIterator(PollingRecordPublisher.java:195)
	at org.apache.flink.streaming.connectors.kinesis.internals.publisher.polling.PollingRecordPublisher.<init>(PollingRecordPublisher.java:96)
	at org.apache.flink.streaming.connectors.kinesis.internals.publisher.polling.PollingRecordPublisherFactory.create(PollingRecordPublisherFactory.java:86)
	at org.apache.flink.streaming.connectors.kinesis.internals.publisher.polling.PollingRecordPublisherFactory.create(PollingRecordPublisherFactory.java:34)
	at org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher.createRecordPublisher(KinesisDataFetcher.java:497)
	at org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher.createShardConsumer(KinesisDataFetcher.java:466)
	at org.apache.flink.streaming.connectors.kinesis.internals.KinesisDataFetcher.runFetcher(KinesisDataFetcher.java:591)
	at org.apache.flink.streaming.connectors.kinesis.FlinkKinesisConsumer.run(FlinkKinesisConsumer.java:392)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:67)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:332)

Hints from issue discussion:
Welcome to LocalStack! Thanks for reporting your first issue and our team will be working towards fixing the issue for you or reach out for more background information. We recommend joining our [Slack Community](https://localstack.cloud/contact/) for real-time help and drop a message to LocalStack Pro Support if you are a Pro user! If you are willing to contribute towards fixing this issue, please have a look at our [contributing guidelines](https://github.com/localstack/.github/blob/main/CONTRIBUTING.md) and our [contributing guide](https://docs.localstack.cloud/contributing/).
Hi @lengyuewuhua!
Would it be possible to provide a minimum reproducible sample for this issue?
We have some tests in place which try to cover the CBOR protocol for Kinesis, but it always depends on the client (i.e. your flink code) if it sends the request in a specific manner. These cases are hard to debug without a specific reproducible client environment...
Thanks!
Hi @alexrashed,

Thanks for your quick response. Please see my code snipt below - 


```java
@Slf4j
public class JobHandler {

  private static final String DEFAULT_SOURCE_STREAM = "ExampleInputStream";
  // "POLLING" for standard consumer, "EFO" for Enhanced Fan-Out
  private static final String DEFAULT_PUBLISHER_TYPE = RecordPublisherType.POLLING.name();
  private static final String DEFAULT_SINK_STREAM = "ExampleOutputStream";
  private static final String DEFAULT_AWS_REGION = "us-east-1";
  private static final String DEFAULT_AWS_ENDPOINT = "http://localhost.localstack.cloud:4566";


  /**
   * Get configuration properties from Amazon Managed Service for Apache Flink runtime properties
   * GroupID "FlinkApplicationProperties", or from command line parameters when running locally
   */
  private static ParameterTool loadApplicationParameters(String[] args,
      StreamExecutionEnvironment env) throws IOException {
    if (env instanceof LocalStreamEnvironment) {
      return ParameterTool.fromArgs(args);
    } else {
      Map<String, Properties> applicationProperties = KinesisAnalyticsRuntime.getApplicationProperties();
      Properties flinkProperties = applicationProperties.get("FlinkApplicationProperties");
      if (flinkProperties == null) {
        throw new RuntimeException(
            "Unable to load FlinkApplicationProperties properties from runtime properties");
      }
      Map<String, String> map = new HashMap<>(flinkProperties.size());
      flinkProperties.forEach((k, v) -> map.put((String) k, (String) v));
      return ParameterTool.fromMap(map);
    }
  }

  private static FlinkKinesisConsumer<String> createKinesisSource(
      ParameterTool applicationProperties) {

    // Properties for Amazon Kinesis Data Streams Source, we need to specify from where we want to consume the data.
    // STREAM_INITIAL_POSITION: LATEST: consume messages that have arrived from the moment application has been deployed
    // STREAM_INITIAL_POSITION: TRIM_HORIZON: consume messages starting from first available in the Kinesis Stream
    Properties kinesisConsumerConfig = new Properties();
    kinesisConsumerConfig.put(AWSConfigConstants.AWS_REGION,
        applicationProperties.get("kinesis.region", DEFAULT_AWS_REGION));
    kinesisConsumerConfig.put(AWSConfigConstants.AWS_ENDPOINT,
        applicationProperties.get("aws.endpoint", DEFAULT_AWS_ENDPOINT));
    kinesisConsumerConfig.put(STREAM_INITIAL_POSITION, "LATEST");

    kinesisConsumerConfig.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, "111");
    kinesisConsumerConfig.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, "111");

    // Set up publisher type: POLLING (standard consumer) or EFO (Enhanced Fan-Out)
    kinesisConsumerConfig.put(RECORD_PUBLISHER_TYPE,
        applicationProperties.get("kinesis.source.type", DEFAULT_PUBLISHER_TYPE));
    return new FlinkKinesisConsumer<>(
        applicationProperties.get("kinesis.source.stream", DEFAULT_SOURCE_STREAM),
        new SimpleStringSchema(), kinesisConsumerConfig);
  }


  private static KinesisStreamsSink<String> createKinesisSink(ParameterTool applicationProperties) {

    Properties sinkProperties = new Properties();
    // Required
    sinkProperties.put(AWSConfigConstants.AWS_REGION,
        applicationProperties.get("kinesis.region", DEFAULT_AWS_REGION));
    sinkProperties.put(AWSConfigConstants.AWS_ENDPOINT,
        applicationProperties.get("aws.endpoint", DEFAULT_AWS_ENDPOINT));

    return KinesisStreamsSink.<String>builder().setKinesisClientProperties(sinkProperties)
        .setSerializationSchema(new SimpleStringSchema())
        .setPartitionKeyGenerator(element -> String.valueOf(element.hashCode()))
        .setStreamName(applicationProperties.get("kinesis.sink.stream", DEFAULT_SINK_STREAM))
        .build();
  }


  public static void main(String[] args) throws Exception {
    // set up the streaming execution environment
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    final ParameterTool applicationProperties = loadApplicationParameters(args, env);
    log.warn("Application properties: {}", applicationProperties.toMap());

    FlinkKinesisConsumer<String> source = createKinesisSource(applicationProperties);
    DataStream<String> input = env.addSource(source, "Kinesis source");

    KinesisStreamsSink<String> sink = createKinesisSink(applicationProperties);
    input.sinkTo(sink);

    env.execute("Flink Kinesis Source and Sink examples");
  }
}
```

Commit message: fix CBOR timestamp parsing

Patch to review:
diff --git a/localstack-core/localstack/aws/client.py b/localstack-core/localstack/aws/client.py
index dfe475147188c..54f15968419f3 100644
--- a/localstack-core/localstack/aws/client.py
+++ b/localstack-core/localstack/aws/client.py
@@ -2,7 +2,7 @@
 
 import io
 import logging
-from datetime import datetime
+from datetime import datetime, timezone
 from typing import Dict, Iterable, Optional
 from urllib.parse import urlsplit
 
@@ -173,16 +173,89 @@ def _parse_body_as_json(fn, self, body_contents):
         try:
             return fn(self, body_contents)
         except UnicodeDecodeError as json_exception:
-            import cbor2
+            # cbor2: explicitly load from private _decoder module to avoid using the (non-patched) C-version
+            from cbor2._decoder import loads
 
             try:
                 LOG.debug("botocore failed decoding JSON. Trying to decode as CBOR.")
-                return cbor2.loads(body_contents)
+                return loads(body_contents)
             except Exception as cbor_exception:
                 LOG.debug("CBOR fallback decoding failed.")
                 raise cbor_exception from json_exception
 
 
+@hooks.on_infra_start()
+def _patch_cbor2():
+    """
+    Patch fixing the AWS CBOR en-/decoding of datetime fields.
+
+    Unfortunately, Kinesis (the only known service using CBOR) does not use the number of seconds (with floating-point
+    milliseconds - according to RFC8949), but uses milliseconds.
+    Python cbor2 is highly optimized by using a C-implementation by default, which cannot be patched.
+    Instead of `from cbor2 import loads`, directly import the python-native loads implementation to avoid loading the
+    unpatched C implementation:
+    ```
+    from cbor2._decoder import loads
+    from cbor2._decoder import dumps
+    ```
+
+    See https://github.com/aws/aws-sdk-java-v2/issues/4661
+    """
+    from cbor2._decoder import CBORDecodeValueError, semantic_decoders
+    from cbor2._encoder import CBOREncodeValueError, default_encoders
+    from cbor2._types import CBORTag
+
+    def _patched_decode_epoch_datetime(self) -> datetime:
+        """
+        Replaces `cbor2._decoder.CBORDecoder.decode_epoch_datetime` as default datetime semantic_decoder.
+        """
+        # Semantic tag 1
+        value = self._decode()
+
+        try:
+            # The next line is the only change in this patch compared to the original function.
+            # AWS breaks the CBOR spec by using the millis instead of seconds (with floating point support for millis)
+            # https://github.com/aws/aws-sdk-java-v2/issues/4661
+            value = value / 1000
+            tmp = datetime.fromtimestamp(value, timezone.utc)
+        except (OverflowError, OSError, ValueError) as exc:
+            raise CBORDecodeValueError("error decoding datetime from epoch") from exc
+
+        return self.set_shareable(tmp)
+
+    def _patched_encode_datetime(self, value: datetime) -> None:
+        """
+        Replaces `cbor2._encoder.CBOREncoder.encode_datetime` as default datetime default_encoder.
+        """
+        if not value.tzinfo:
+            if self._timezone:
+                value = value.replace(tzinfo=self._timezone)
+            else:
+                raise CBOREncodeValueError(
+                    f"naive datetime {value!r} encountered and no default timezone " "has been set"
+                )
+
+        if self.datetime_as_timestamp:
+            from calendar import timegm
+
+            if not value.microsecond:
+                timestamp: float = timegm(value.utctimetuple())
+            else:
+                timestamp = timegm(value.utctimetuple()) + value.microsecond / 1000000
+            # The next line is the only change in this patch compared to the original function.
+            # AWS breaks the CBOR spec by using the millis instead of seconds (with floating point support for millis)
+            # https://github.com/aws/aws-sdk-java-v2/issues/4661
+            timestamp = timestamp * 1000
+            self.encode_semantic(CBORTag(1, timestamp))
+        else:
+            datestring = value.isoformat().replace("+00:00", "Z")
+            self.encode_semantic(CBORTag(0, datestring))
+
+    # overwrite the default epoch datetime en-/decoder with patched versions
+    default_encoders[datetime] = _patched_encode_datetime
+    semantic_decoders[1] = _patched_decode_epoch_datetime
+
+
 def _create_and_enrich_aws_request(
     fn, self: Endpoint, params: dict, operation_model: OperationModel = None
 ):
diff --git a/localstack-core/localstack/aws/protocol/parser.py b/localstack-core/localstack/aws/protocol/parser.py
index ae926e0d48259..54713ecfb3dda 100644
--- a/localstack-core/localstack/aws/protocol/parser.py
+++ b/localstack-core/localstack/aws/protocol/parser.py
@@ -72,7 +72,6 @@
 from typing import IO, Any, Dict, List, Mapping, Optional, Tuple, Union
 from xml.etree import ElementTree as ETree
 
-import cbor2
 import dateutil.parser
 from botocore.model import (
     ListShape,
@@ -83,6 +82,9 @@
     Shape,
     StructureShape,
 )
+
+# cbor2: explicitly load from private _decoder module to avoid using the (non-patched) C-version
+from cbor2._decoder import loads as cbor2_loads
 from werkzeug.exceptions import BadRequest, NotFound
 
 from localstack.aws.protocol.op_router import RestServiceOperationRouter
@@ -868,7 +870,7 @@ def _parse_body_as_json(self, request: Request) -> dict:
             return {}
         if request.mimetype.startswith("application/x-amz-cbor"):
             try:
-                return cbor2.loads(body_contents)
+                return cbor2_loads(body_contents)
             except ValueError as e:
                 raise ProtocolParserError("HTTP body could not be parsed as CBOR.") from e
         else:
@@ -888,18 +890,14 @@ def _parse_timestamp(
         if not shape.serialization.get("timestampFormat") and request.mimetype.startswith(
             "application/x-amz-cbor"
         ):
+            # cbor2 has native support for timestamp decoding, so this node could already have the right type
+            if isinstance(node, datetime.datetime):
+                return node
+            # otherwise parse the timestamp using the AWS CBOR timestamp format
+            # (non-CBOR-standard conform, uses millis instead of floating-point-millis)
             return self._convert_str_to_timestamp(node, self.CBOR_TIMESTAMP_FORMAT)
         return super()._parse_timestamp(request, shape, node, uri_params)
 
-    def _parse_blob(
-        self, request: Request, shape: Shape, node: bool, uri_params: Mapping[str, Any] = None
-    ) -> bytes:
-        if isinstance(node, bytes) and request.mimetype.startswith("application/x-amz-cbor"):
-            # CBOR does not base64 encode binary data
-            return bytes(node)
-        else:
-            return super()._parse_blob(request, shape, node, uri_params)
-
 
 class JSONRequestParser(BaseJSONRequestParser):
     """
diff --git a/localstack-core/localstack/aws/protocol/serializer.py b/localstack-core/localstack/aws/protocol/serializer.py
index 41154390396b2..86cabdd3487b6 100644
--- a/localstack-core/localstack/aws/protocol/serializer.py
+++ b/localstack-core/localstack/aws/protocol/serializer.py
@@ -85,11 +85,13 @@
 from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
 from xml.etree import ElementTree as ETree
 
-import cbor2
 import xmltodict
 from botocore.model import ListShape, MapShape, OperationModel, ServiceModel, Shape, StructureShape
 from botocore.serialize import ISO8601, ISO8601_MICRO
 from botocore.utils import calculate_md5, is_json_value_header, parse_to_aware_datetime
+
+# cbor2: explicitly load from private _encoder module to avoid using the (non-patched) C-version
+from cbor2._encoder import dumps as cbor2_dumps
 from werkzeug import Request as WerkzeugRequest
 from werkzeug import Response as WerkzeugResponse
 from werkzeug.datastructures import Headers, MIMEAccept
@@ -514,10 +516,6 @@ def _timestamp_iso8601(value: datetime) -> str:
     def _timestamp_unixtimestamp(value: datetime) -> float:
         return value.timestamp()
 
-    @staticmethod
-    def _timestamp_unixtimestampmillis(value: datetime) -> int:
-        return int(value.timestamp() * 1000)
-
     def _timestamp_rfc822(self, value: datetime) -> str:
         if isinstance(value, datetime):
             value = self._timestamp_unixtimestamp(value)
@@ -1244,7 +1242,7 @@ def _serialize_error(
                 body["message"] = message
 
         if mime_type in self.CBOR_TYPES:
-            response.set_response(cbor2.dumps(body))
+            response.set_response(cbor2_dumps(body, datetime_as_timestamp=True))
             response.content_type = mime_type
         else:
             response.set_json(body)
@@ -1282,7 +1280,7 @@ def _serialize_body_params(
             self._serialize(body, params, shape, None, mime_type)
 
         if mime_type in self.CBOR_TYPES:
-            return cbor2.dumps(body)
+            return cbor2_dumps(body, datetime_as_timestamp=True)
         else:
             return json.dumps(body)
 
@@ -1364,13 +1362,12 @@ def _default_serialize(self, body: dict, value: Any, _, key: str, __):
     def _serialize_type_timestamp(
         self, body: dict, value: Any, shape: Shape, key: str, mime_type: str
     ):
-        timestamp_format = (
-            shape.serialization.get("timestampFormat")
-            # CBOR always uses unix timestamp milliseconds
-            if mime_type not in self.CBOR_TYPES
-            else "unixtimestampmillis"
-        )
-        body[key] = self._convert_timestamp_to_str(value, timestamp_format)
+        if mime_type in self.CBOR_TYPES:
+            # CBOR has native support for timestamps
+            body[key] = value
+        else:
+            timestamp_format = shape.serialization.get("timestampFormat")
+            body[key] = self._convert_timestamp_to_str(value, timestamp_format)
 
     def _serialize_type_blob(
         self, body: dict, value: Union[str, bytes], _, key: str, mime_type: str


Code review comments:
diff_hunk: @@ -567,6 +578,77 @@ def test_get_records_shard_iterator_with_surrounding_quotes(
 
         assert aws_client.kinesis.get_records(ShardIterator=f'"{shard_iterator}"')["Records"]
 
+    @markers.aws.validated
+    def test_subscribe_to_shard_with_at_timestamp_cbor(
+        self,
+        kinesis_create_stream,
+        wait_for_stream_ready,
+        kinesis_register_consumer,
+        wait_for_kinesis_consumer_ready,
+        aws_client,
+        aws_http_client_factory,
+        account_id,
+        region_name,
+    ):
+        # create stream
+        pre_create_timestamp = (datetime.now() - timedelta(hours=0, minutes=1)).astimezone()
+        stream_name = kinesis_create_stream(ShardCount=1)
+        stream_arn = aws_client.kinesis.describe_stream(StreamName=stream_name)[
+            "StreamDescription"
+        ]["StreamARN"]
+        wait_for_stream_ready(stream_name)
+        post_create_timestamp = (datetime.now() + timedelta(hours=0, minutes=1)).astimezone()
+
+        # perform a raw DescribeStream request to test the datetime serialization by LocalStack
+        kinesis_http_client = aws_http_client_factory("kinesis", signer_factory=SigV4Auth)
+        endpoint = (
+            f"https://{account_id}.control-kinesis.{region_name}.amazonaws.com"
+            if is_aws_cloud()
+            else config.internal_service_url()
+        )
+        describe_response = kinesis_http_client.post(
+            f"{endpoint}",
path: tests/aws/services/kinesis/test_kinesis.py
text: @user1:
nit: could use the endpoint variable directly here (same in the next call under on line#631)