check validity of the expiration in basic.publish

Checking it always is not a good idea, since we don't always decode
the properties.

3 files changed, 29 insertions, 15 deletions

diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index d5f678af..66ffb156 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -710,19 +710,12 @@ message_properties(Message, Confirm, #q{ttl = TTL}) ->
                         needs_confirming = needs_confirming(Confirm)}.
 
 calculate_msg_expiry(#basic_message{content = Content}, TTL) ->
-    #content{properties = #'P_basic'{expiration = Expiration}} =
+    #content{properties = Props} =
         rabbit_binary_parser:ensure_content_decoded(Content),
-    ParseError =
-        fun () -> rabbit_log:warning("could not parse expiration '~p'~n.",
-                                     [Expiration])
-        end,
-    Milli = case Expiration of
-                undefined -> TTL;
-                B         -> case string:to_integer(binary_to_list(B)) of
-                                 {error, no_integer} -> ParseError(), TTL;
-                                 {N, ""}             -> N;
-                                 {_, _ }             -> ParseError(), TTL
-                             end
+    %% We assert that the expiration must be valid - we check in che channel.
+    Milli = case rabbit_basic:parse_expiration(Props) of
+                {ok, undefined} -> TTL;
+                {ok, N        } -> N
             end,
     case Milli of
         undefined -> undefined;
diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index 734456d3..eb304de5 100644
--- a/src/rabbit_basic.erl
+++ b/src/rabbit_basic.erl
@@ -20,7 +20,8 @@
 
 -export([publish/4, publish/6, publish/1,
          message/3, message/4, properties/1, append_table_header/3,
-         extract_headers/1, map_headers/2, delivery/4, header_routes/1]).
+         extract_headers/1, map_headers/2, delivery/4, header_routes/1,
+         parse_expiration/1]).
 -export([build_content/2, from_content/1]).
 
 %%----------------------------------------------------------------------------
@@ -72,6 +73,9 @@
                           binary() | [binary()]) -> rabbit_types:content()).
 -spec(from_content/1 :: (rabbit_types:content()) ->
                              {rabbit_framing:amqp_property_record(), binary()}).
+-spec(parse_expiration/1 ::
+        (rabbit_framing:amqp_property_record())
+        -> rabbit_types:ok_or_error2(non_neg_integer(), any())).
 
 -endif.
 
@@ -226,3 +230,13 @@ header_routes(HeadersTable) ->
            {Type, _Val}    -> throw({error, {unacceptable_type_in_header,
                                              binary_to_list(HeaderKey), Type}})
        end || HeaderKey <- ?ROUTING_HEADERS]).
+
+parse_expiration(#'P_basic'{expiration = Expiration}) ->
+    case Expiration of
+        undefined -> {ok, undefined};
+        B         -> case string:to_integer(binary_to_list(B)) of
+                         {error, no_integer} = E -> E;
+                         {N, ""}                 -> {ok, N};
+                         {_, _ }                 -> {error, leftover_string}
+                     end
+    end.
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 69fe0edc..78f5d3b8 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -608,8 +608,15 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
     check_internal_exchange(Exchange),
     %% We decode the content's properties here because we're almost
     %% certain to want to look at delivery-mode and priority.
-    DecodedContent = rabbit_binary_parser:ensure_content_decoded(Content),
-    check_user_id_header(DecodedContent#content.properties, State),
+    DecodedContent = #content {properties = Props} =
+        rabbit_binary_parser:ensure_content_decoded(Content),
+    check_user_id_header(Props, State),
+    case rabbit_basic:parse_expiration(Props) of
+        {ok, _}    -> ok;
+        {error, E} -> rabbit_misc:protocol_error(
+                        invalid_expiration, "cannot parse expiration '~p': ~p",
+                        [Props#'P_basic'.expiration, E])
+    end,
     {MsgSeqNo, State1} =
         case {TxStatus, ConfirmEnabled} of
             {none, false} -> {undefined, State};