Bugfix/copy object bugs

riak_test/src/rtcs.erl

@@ -1020,16 +1020,22 @@ pbc({multibag, _} = Flavor, ObjectKind, RiakNodes, Opts) ->
     rtcs_bag:pbc(Flavor, ObjectKind, RiakNodes, Opts).
 
 make_authorization(Method, Resource, ContentType, Config, Date) ->
-    make_authorization(s3, Method, Resource, ContentType, Config, Date).
+    make_authorization(Method, Resource, ContentType, Config, Date, []).
 
-make_authorization(Type, Method, Resource, ContentType, Config, Date) ->
-    StringToSign = [Method, $\n, [], $\n, ContentType, $\n, Date, $\n, Resource],
-    Signature =
-        base64:encode_to_string(sha_mac(Config#aws_config.secret_access_key, StringToSign)),
+make_authorization(Method, Resource, ContentType, Config, Date, AmzHeaders) ->
+    make_authorization(s3, Method, Resource, ContentType, Config, Date, AmzHeaders).
+
+make_authorization(Type, Method, Resource, ContentType, Config, Date, AmzHeaders) ->
     Prefix = case Type of
                  s3 -> "AWS";
                  velvet -> "MOSS"
              end,
+    StsAmzHeaderPart = [[K, $:, V, $\n] || {K, V} <- AmzHeaders],
+    StringToSign = [Method, $\n, [], $\n, ContentType, $\n, Date, $\n,
+                    StsAmzHeaderPart, Resource],
+    lager:debug("StringToSign~n~s~n", [StringToSign]),
+    Signature =
+        base64:encode_to_string(sha_mac(Config#aws_config.secret_access_key, StringToSign)),
     lists:flatten([Prefix, " ", Config#aws_config.access_key_id, $:, Signature]).
 
 sha_mac(Key,STS) -> crypto:hmac(sha, Key,STS).

riak_test/tests/put_copy_test.erl

@@ -36,6 +36,10 @@
 -define(KEY2, "sidepocket").
 -define(KEY3, "superpocket").
 -define(BUCKET3, "the-other-put-target-bucket").
+-define(BUCKET4, "no-cl-bucket").
+-define(SRC_KEY, "source").
+-define(TGT_KEY, "target").
+-define(MP_TGT_KEY, "mp-target").
 
 confirm() ->
     {UserConfig, {RiakNodes, _CSNodes, _}} = rtcs:setup(1),
@@ -59,6 +63,8 @@ confirm() ->
     ok = verify_others_copy(UserConfig, UserConfig2),
     ok = verify_multipart_copy(UserConfig),
     ok = verify_security(UserConfig, UserConfig2, UserConfig3),
+    ok = verify_source_not_found(UserConfig),
+    ok = verify_without_cl_header(UserConfig),
 
     ?assertEqual([{delete_marker, false}, {version_id, "null"}],
                  erlcloud_s3:delete_object(?BUCKET, ?KEY, UserConfig)),
@@ -237,3 +243,78 @@ verify_security(Alice, Bob, Charlie) ->
     ?assertEqual("403", Status),
 
     ok.
+
+verify_source_not_found(UserConfig) ->
+    NonExistingKey = "non-existent-source",
+    {'EXIT', {{aws_error, {http_error, 404, _, ErrorXml}}, _Stack}} =
+        (catch erlcloud_s3:copy_object(?BUCKET2, ?KEY2,
+                                       ?BUCKET, NonExistingKey, UserConfig)),
+    lager:debug("ErrorXml: ~s", [ErrorXml]),
+    ?assert(string:str(ErrorXml,
+                       "<Resource>/" ++ ?BUCKET ++
+                           "/" ++ NonExistingKey ++ "</Resource>") > 0).
+
+%% Verify reuqests without Content-Length header, they should succeed.
+%% To avoid automatic Content-Length header addition by HTTP client library,
+%% this test uses `curl' command line utitlity, intended.
+verify_without_cl_header(UserConfig) ->
+    ?assertEqual(ok, erlcloud_s3:create_bucket(?BUCKET4, UserConfig)),
+    Data = ?DATA0,
+    ?assertEqual([{version_id, "null"}],
+                 erlcloud_s3:put_object(?BUCKET4, ?SRC_KEY, Data, UserConfig)),
+    verify_without_cl_header(UserConfig, normal, Data),
+    verify_without_cl_header(UserConfig, mp, Data),
+    ok.
+
+verify_without_cl_header(UserConfig, normal, Data) ->
+    lager:info("Verify basic (non-MP) PUT copy without Content-Length header"),
+    Target = fmt("/~s/~s", [?BUCKET4, ?TGT_KEY]),
+    Source = fmt("/~s/~s", [?BUCKET4, ?SRC_KEY]),
+    _Res = exec_curl(UserConfig, "PUT", Target, [{"x-amz-copy-source", Source}]),
+
+    Props = erlcloud_s3:get_object(?BUCKET4, ?TGT_KEY, UserConfig),
+    ?assertEqual(Data, proplists:get_value(content, Props)),
+    ok;
+verify_without_cl_header(UserConfig, mp, Data) ->
+    lager:info("Verify Multipart upload copy without Content-Length header"),
+    InitUploadRes = erlcloud_s3_multipart:initiate_upload(
+                      ?BUCKET4, ?MP_TGT_KEY, "application/octet-stream",
+                      [], UserConfig),
+    UploadId = erlcloud_s3_multipart:upload_id(InitUploadRes),
+    lager:info("~p ~p", [InitUploadRes, UploadId]),
+    Source = fmt("/~s/~s", [?BUCKET4, ?SRC_KEY]),
+    MpTarget = fmt("/~s/~s?partNumber=1&uploadId=~s", [?BUCKET4, ?MP_TGT_KEY, UploadId]),
+    _Res = exec_curl(UserConfig, "PUT", MpTarget,
+                     [{"x-amz-copy-source", Source},
+                      {"x-amz-copy-source-range", "bytes=1-2"}]),
+
+    ListPartsXml = erlcloud_s3_multipart:list_parts(?BUCKET4, ?MP_TGT_KEY, UploadId, [], UserConfig),
+    lager:debug("ListParts: ~p", [ListPartsXml]),
+    ListPartsRes = erlcloud_s3_multipart:parts_to_term(ListPartsXml),
+    Parts = proplists:get_value(parts, ListPartsRes),
+    EtagList = [{PartNum, Etag} || {PartNum, [{etag, Etag}, {size, _Size}]} <- Parts],
+    lager:debug("EtagList: ~p", [EtagList]),
+    ?assertEqual(ok, erlcloud_s3_multipart:complete_upload(
+                       ?BUCKET4, ?MP_TGT_KEY, UploadId, EtagList, UserConfig)),
+    Props = erlcloud_s3:get_object(?BUCKET4, ?MP_TGT_KEY, UserConfig),
+    ExpectedBody = binary:part(Data, 1, 2),
+    ?assertEqual(ExpectedBody, proplists:get_value(content, Props)),
+    ok.
+
+exec_curl(#aws_config{s3_port=Port} = UserConfig, Method, Resource, AmzHeaders) ->
+    ContentType = "application/octet-stream",
+    Date = httpd_util:rfc1123_date(),
+    Auth = rtcs:make_authorization(Method, Resource, ContentType, UserConfig, Date,
+                                   AmzHeaders),
+    HeaderArgs = [fmt("-H '~s: ~s' ", [K, V]) ||
+                     {K, V} <- [{"Date", Date}, {"Authorization", Auth},
+                                {"Content-Type", ContentType} | AmzHeaders]],
+    Cmd="curl -X " ++ Method ++ " -v -s " ++ HeaderArgs ++
+        "'http://127.0.0.1:" ++ integer_to_list(Port) ++ Resource ++ "'",
+    lager:debug("Curl command line: ~s", [Cmd]),
+    Res = os:cmd(Cmd),
+    lager:debug("Curl result: ~s", [Res]),
+    Res.
+
+fmt(Fmt, Args) ->
+    lists:flatten(io_lib:format(Fmt, Args)).

riak_test/tests/regression_tests.erl

@@ -79,7 +79,7 @@ verify_cs347(_SetupInfo = {UserConfig, {_RiakNodes, _CSNodes, _Stanchion}}, Buck
             Result ->
                 Result
         end,
-    ?assert(rtcs:check_no_such_bucket(ListObjectRes1, "/"++?TEST_BUCKET_CS347)),
+    ?assert(rtcs:check_no_such_bucket(ListObjectRes1, "/" ++ ?TEST_BUCKET_CS347 ++ "/")),
 
     lager:info("creating bucket ~p", [BucketName]),
     ?assertEqual(ok, erlcloud_s3:create_bucket(BucketName, UserConfig)),
@@ -97,7 +97,7 @@ verify_cs347(_SetupInfo = {UserConfig, {_RiakNodes, _CSNodes, _Stanchion}}, Buck
             Result2 ->
                 Result2
         end,
-    ?assert(rtcs:check_no_such_bucket(ListObjectRes2, "/"++?TEST_BUCKET_CS347)),
+    ?assert(rtcs:check_no_such_bucket(ListObjectRes2, "/" ++ ?TEST_BUCKET_CS347 ++ "/")),
     ok.
 
 

riak_test/tests/stats_test.erl

@@ -143,7 +143,7 @@ query_stats(Type, UserConfig, Port) ->
                                stanchion -> {"/stats", velvet}
                            end,
     Cmd="curl -s -H 'Date: " ++ Date ++ "' -H 'Authorization: " ++
-        rtcs:make_authorization(SignType, "GET", Resource, [], UserConfig, Date) ++
+        rtcs:make_authorization(SignType, "GET", Resource, [], UserConfig, Date, []) ++
         "' http://localhost:" ++
         integer_to_list(Port) ++ Resource,
     lager:info("Stats query cmd: ~p", [Cmd]),

src/riak_cs_copy_object.erl

@@ -50,7 +50,7 @@ test_condition_and_permission(RcPid, SrcManifest, RD, Ctx) ->
     %% TODO: write tests around these conditions, any kind of test is okay
     case condition_check(RD, ETag, LastUpdate) of
         ok ->
-            _ = authorize_on_src(RcPid, SrcManifest, RD, Ctx);
+            authorize_on_src(RcPid, SrcManifest, RD, Ctx);
         Other ->
             {Other, RD, Ctx}
     end.

src/riak_cs_s3_response.erl

@@ -26,7 +26,6 @@
          error_message/1,
          error_code/1,
          error_response/1,
-         error_response/5,
          copy_object_response/3,
          copy_part_response/3,
          no_such_upload_response/3,
@@ -53,6 +52,8 @@ error_message(invalid_email_address) ->
     "The email address you provided is not a valid.";
 error_message(access_denied) ->
     "Access Denied";
+error_message(copy_source_access_denied) ->
+    "Access Denied";
 error_message(reqtime_tooskewed) ->
     "The difference between the request time and the current time is too large.";
 error_message(bucket_not_empty) ->
@@ -88,6 +89,7 @@ error_message(malformed_policy_principal) -> "Invalid principal in policy";
 error_message(malformed_policy_action) -> "Policy has invalid action";
 error_message(malformed_policy_condition) -> "Policy has invalid condition";
 error_message(no_such_key) -> "The specified key does not exist.";
+error_message(no_copy_source_key) -> "The specified key does not exist.";
 error_message(no_such_bucket_policy) -> "The specified bucket does not have a bucket policy.";
 error_message(no_such_upload) ->
     "The specified upload does not exist. The upload ID may be invalid, "
@@ -111,6 +113,7 @@ error_message(ErrorName) ->
 -spec error_code(error_reason()) -> string().
 error_code(invalid_access_key_id) -> "InvalidAccessKeyId";
 error_code(access_denied) -> "AccessDenied";
+error_code(copy_source_access_denied) -> "AccessDenied";
 error_code(reqtime_tooskewed) -> "RequestTimeTooSkewed";
 error_code(bucket_not_empty) -> "BucketNotEmpty";
 error_code(bucket_already_exists) -> "BucketAlreadyExists";
@@ -123,6 +126,7 @@ error_code(bad_etag_order) -> "InvalidPartOrder";
 error_code(invalid_user_update) -> "InvalidUserUpdate";
 error_code(no_such_bucket) -> "NoSuchBucket";
 error_code(no_such_key) -> "NoSuchKey";
+error_code(no_copy_source_key) -> "NoSuchKey";
 error_code({riak_connect_failed, _}) -> "RiakConnectFailed";
 error_code(admin_key_undefined) -> "ServiceUnavailable";
 error_code(admin_secret_undefined) -> "ServiceUnavailable";
@@ -161,6 +165,7 @@ error_code(ErrorName) ->
 status_code(invalid_access_key_id) -> 403;
 status_code(invalid_email_address) -> 400;
 status_code(access_denied) ->  403;
+status_code(copy_source_access_denied) ->  403;
 status_code(reqtime_tooskewed) -> 403;
 status_code(bucket_not_empty) ->  409;
 status_code(bucket_already_exists) -> 409;
@@ -174,6 +179,7 @@ status_code(bad_etag_order) -> 400;
 status_code(invalid_user_update) -> 400;
 status_code(no_such_bucket) -> 404;
 status_code(no_such_key) -> 404;
+status_code(no_copy_source_key) -> 404;
 status_code({riak_connect_failed, _}) -> 503;
 status_code(admin_key_undefined) -> 503;
 status_code(admin_secret_undefined) -> 503;
@@ -224,7 +230,8 @@ respond(StatusCode, Body, ReqData, Ctx) ->
     {{halt, StatusCode}, UpdReqData, Ctx}.
 
 api_error(Error, RD, Ctx) when is_atom(Error) ->
-    error_response(status_code(Error),
+    error_response(Error,
+                   status_code(Error),
                    error_code(Error),
                    error_message(Error),
                    RD,
@@ -233,7 +240,8 @@ api_error({Tag, _}=Error, RD, Ctx)
   when Tag =:= riak_connect_failed orelse
        Tag =:= malformed_policy_version orelse
        Tag =:= auth_not_supported ->
-    error_response(status_code(Error),
+    error_response(Tag,
+                   status_code(Error),
                    error_code(Error),
                    error_message(Error),
                    RD,
@@ -248,14 +256,23 @@ error_response(ErrorDoc) when length(ErrorDoc) =:= 0 ->
 error_response(ErrorDoc) ->
     {error, error_code_to_atom(xml_error_code(ErrorDoc))}.
 
-error_response(StatusCode, Code, Message, RD, Ctx) ->
-    {OrigResource, _} = riak_cs_s3_rewrite:original_resource(RD),
+error_response(ErrorTag, StatusCode, Code, Message, RD, Ctx) ->
     XmlDoc = [{'Error', [{'Code', [Code]},
                          {'Message', [Message]},
-                         {'Resource', [string:strip(OrigResource, right, $/)]},
+                         {'Resource', [error_resource(ErrorTag, RD)]},
                          {'RequestId', [""]}]}],
     respond(StatusCode, riak_cs_xml:to_xml(XmlDoc), RD, Ctx).
 
+-spec error_resource(atom(), #wm_reqdata{}) -> iodata().
+error_resource(Tag, RD)
+  when Tag =:= no_copy_source_key;
+       Tag =:= copy_source_access_denied->
+    {B, K} = riak_cs_copy_object:get_copy_source(RD),
+    <<$/, B/binary, $/, K/binary>>;
+error_resource(_Tag, RD) ->
+    {OrigResource, _} = riak_cs_s3_rewrite:original_resource(RD),
+    OrigResource.
+
 toomanybuckets_response(Current,BucketLimit,RD,Ctx) ->
     XmlDoc = {'Error',
               [

src/riak_cs_wm_object.erl

@@ -89,25 +89,14 @@ allowed_methods() ->
     ['HEAD', 'GET', 'DELETE', 'PUT'].
 
 -spec valid_entity_length(#wm_reqdata{}, #context{}) -> {boolean(), #wm_reqdata{}, #context{}}.
-valid_entity_length(RD, Ctx=#context{response_module=ResponseMod}) ->
-    case wrq:method(RD) of
-        'PUT' ->
-            case catch(
-                   list_to_integer(
-                     wrq:get_req_header("Content-Length", RD))) of
-                Length when is_integer(Length) ->
-                    case Length =< riak_cs_lfs_utils:max_content_len() of
-                        false ->
-                            ResponseMod:api_error(
-                              entity_too_large, RD, Ctx);
-                        true ->
-                            check_0length_metadata_update(Length, RD, Ctx)
-                    end;
-                _ ->
-                    {false, RD, Ctx}
-            end;
-        _ ->
-            {true, RD, Ctx}
+valid_entity_length(RD, Ctx) ->
+    MaxLen = riak_cs_lfs_utils:max_content_len(),
+    case riak_cs_wm_utils:valid_entity_length(MaxLen, RD, Ctx) of
+        {true, NewRD, NewCtx} ->
+            check_0length_metadata_update(riak_cs_wm_utils:content_length(RD),
+                                          NewRD, NewCtx);
+        Other ->
+            Other
     end.
 
 -spec content_types_provided(#wm_reqdata{}, #context{}) -> {[{string(), atom()}], #wm_reqdata{}, #context{}}.
@@ -399,18 +388,20 @@ handle_copy_put(RD, Ctx, SrcBucket, SrcKey) ->
                     {true, _RD, _OtherCtx} ->
                         %% access to source object not authorized
                         %% TODO: check the return value / http status
-                        _ = lager:debug("access to source object denied (~s, ~s)", [SrcBucket, SrcKey]),
-                        {{halt, 403}, RD, Ctx};
-
+                        ResponseMod:api_error(copy_source_access_denied, RD, Ctx);
+                    {{halt, 403}, _RD, _OtherCtx} = Error ->
+                        %% access to source object not authorized either, but
+                        %% in different return value
+                        ResponseMod:api_error(copy_source_access_denied, RD, Ctx);
                     {Result, _, _} = Error ->
                         _ = lager:debug("~p on ~s ~s", [Result, SrcBucket, SrcKey]),
                         Error
 
                 end;
             {error, notfound} ->
-                ResponseMod:api_error(no_such_key, RD, Ctx);
+                ResponseMod:api_error(no_copy_source_key, RD, Ctx);
             {error, no_active_manifest} ->
-                ResponseMod:api_error(no_such_key, RD, Ctx);
+                ResponseMod:api_error(no_copy_source_key, RD, Ctx);
             {error, Err} ->
                 ResponseMod:api_error(Err, RD, Ctx)
         end

src/riak_cs_wm_object_upload_part.erl

@@ -141,27 +141,9 @@ process_post(RD, Ctx=#context{local_context=LocalCtx, riak_client=RcPid}) ->
     end.
 
 -spec valid_entity_length(#wm_reqdata{}, #context{}) -> {boolean(), #wm_reqdata{}, #context{}}.
-valid_entity_length(RD, Ctx=#context{local_context=LocalCtx}) ->
-    case wrq:method(RD) of
-        'PUT' ->
-            case catch(
-                   list_to_integer(
-                     wrq:get_req_header("Content-Length", RD))) of
-                Length when is_integer(Length) ->
-                    case Length =< riak_cs_lfs_utils:max_content_len() of
-                        false ->
-                            riak_cs_s3_response:api_error(
-                              entity_too_large, RD, Ctx);
-                        true ->
-                            UpdLocalCtx = LocalCtx#key_context{size=Length},
-                            {true, RD, Ctx#context{local_context=UpdLocalCtx}}
-                    end;
-                _ ->
-                    {false, RD, Ctx}
-            end;
-        _ ->
-            {true, RD, Ctx}
-    end.
+valid_entity_length(RD, Ctx) ->
+    MaxLen = riak_cs_lfs_utils:max_content_len(),
+    riak_cs_wm_utils:valid_entity_length(MaxLen, RD, Ctx).
 
 -spec delete_resource(#wm_reqdata{}, #context{}) ->
                              {boolean() | {'halt', term()}, #wm_reqdata{}, #context{}}.
@@ -284,7 +266,7 @@ accept_body(RD, Ctx0=#context{local_context=LocalCtx0,
                                     riak_cs_riak_client:checkin(ReadRcPid)
                                 end
                         end;
-                    {error, notfond} ->
+                    {error, notfound} ->
                         riak_cs_s3_response:no_such_upload_response(UploadId, RD, Ctx0);
                     {error, Reason} ->
                         riak_cs_s3_response:api_error(Reason, RD, Ctx0)