diff options
Diffstat (limited to 'lib/ssh/src/ssh_sftp.erl')
| -rw-r--r-- | lib/ssh/src/ssh_sftp.erl | 375 |
1 files changed, 342 insertions, 33 deletions
diff --git a/lib/ssh/src/ssh_sftp.erl b/lib/ssh/src/ssh_sftp.erl index 0ea2366ac7..bab688f226 100644 --- a/lib/ssh/src/ssh_sftp.erl +++ b/lib/ssh/src/ssh_sftp.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2005-2013. All Rights Reserved. +%% Copyright Ericsson AB 2005-2014. All Rights Reserved. %% %% The contents of this file are subject to the Erlang Public License, %% Version 1.1, (the "License"); you may not use this file except in @@ -33,8 +33,8 @@ -export([start_channel/1, start_channel/2, start_channel/3, stop_channel/1]). --export([open/3, opendir/2, close/2, readdir/2, pread/4, read/3, - open/4, opendir/3, close/3, readdir/3, pread/5, read/4, +-export([open/3, open_tar/3, opendir/2, close/2, readdir/2, pread/4, read/3, + open/4, open_tar/4, opendir/3, close/3, readdir/3, pread/5, read/4, apread/4, aread/3, pwrite/4, write/3, apwrite/4, awrite/3, pwrite/5, write/4, position/3, real_path/2, read_file_info/2, get_file_info/2, @@ -57,7 +57,8 @@ rep_buf = <<>>, req_id, req_list = [], %% {ReqId, Fun} - inf %% list of fileinf + inf, %% list of fileinf, + opts }). -record(fileinf, @@ -68,6 +69,18 @@ mode }). +-record(bufinf, + { + mode, % read | write (=from or to buffer by user) + crypto_state, + crypto_fun, % For encode or decode depending on the mode field + size = 0, % # bytes "before" the current buffer for the postion call + + chunksize, % The size of the chunks to be sent or received + enc_text_buf = <<>>, % Encrypted text + plain_text_buf = <<>> % Decrypted text + }). + -define(FILEOP_TIMEOUT, infinity). -define(NEXT_REQID(S), @@ -85,10 +98,11 @@ start_channel(Host) when is_list(Host) -> start_channel(Host, []). start_channel(Cm, Opts) when is_pid(Cm) -> Timeout = proplists:get_value(timeout, Opts, infinity), + {_, SftpOpts} = handle_options(Opts, [], []), case ssh_xfer:attach(Cm, []) of {ok, ChannelId, Cm} -> case ssh_channel:start(Cm, ChannelId, - ?MODULE, [Cm, ChannelId, Timeout]) of + ?MODULE, [Cm, ChannelId, SftpOpts]) of {ok, Pid} -> case wait_for_version_negotiation(Pid, Timeout) of ok -> @@ -97,7 +111,7 @@ start_channel(Cm, Opts) when is_pid(Cm) -> TimeOut end; {error, Reason} -> - {error, Reason}; + {error, format_channel_start_error(Reason)}; ignore -> {error, ignore} end; @@ -108,11 +122,12 @@ start_channel(Cm, Opts) when is_pid(Cm) -> start_channel(Host, Opts) -> start_channel(Host, 22, Opts). start_channel(Host, Port, Opts) -> - Timeout = proplists:get_value(timeout, Opts, infinity), - case ssh_xfer:connect(Host, Port, proplists:delete(timeout, Opts)) of + {SshOpts, SftpOpts} = handle_options(Opts, [], []), + Timeout = proplists:get_value(timeout, SftpOpts, infinity), + case ssh_xfer:connect(Host, Port, SshOpts, Timeout) of {ok, ChannelId, Cm} -> case ssh_channel:start(Cm, ChannelId, ?MODULE, [Cm, - ChannelId, Timeout]) of + ChannelId, SftpOpts]) of {ok, Pid} -> case wait_for_version_negotiation(Pid, Timeout) of ok -> @@ -121,7 +136,7 @@ start_channel(Host, Port, Opts) -> TimeOut end; {error, Reason} -> - {error, Reason}; + {error, format_channel_start_error(Reason)}; ignore -> {error, ignore} end; @@ -159,6 +174,77 @@ open(Pid, File, Mode) -> open(Pid, File, Mode, FileOpTimeout) -> call(Pid, {open, false, File, Mode}, FileOpTimeout). +open_tar(Pid, File, Mode) -> + open_tar(Pid, File, Mode, ?FILEOP_TIMEOUT). +open_tar(Pid, File, Mode, FileOpTimeout) -> + case {lists:member(write,Mode), + lists:member(read,Mode), + Mode -- [read,write]} of + {true,false,[]} -> + {ok,Handle} = open(Pid, File, [write], FileOpTimeout), + erl_tar:init(Pid, write, + fun(write, {_,Data}) -> + write_to_remote_tar(Pid, Handle, to_bin(Data), FileOpTimeout); + (position, {_,Pos}) -> + position(Pid, Handle, Pos, FileOpTimeout); + (close, _) -> + close(Pid, Handle, FileOpTimeout) + end); + {true,false,[{crypto,{CryptoInitFun,CryptoEncryptFun,CryptoEndFun}}]} -> + {ok,SftpHandle} = open(Pid, File, [write], FileOpTimeout), + BI = #bufinf{mode = write, + crypto_fun = CryptoEncryptFun}, + {ok,BufHandle} = open_buf(Pid, CryptoInitFun, BI, FileOpTimeout), + erl_tar:init(Pid, write, + fun(write, {_,Data}) -> + write_buf(Pid, SftpHandle, BufHandle, to_bin(Data), FileOpTimeout); + (position, {_,Pos}) -> + position_buf(Pid, SftpHandle, BufHandle, Pos, FileOpTimeout); + (close, _) -> + {ok,#bufinf{ + plain_text_buf = PlainBuf0, + enc_text_buf = EncBuf0, + crypto_state = CState0 + }} = call(Pid, {get_bufinf,BufHandle}, FileOpTimeout), + {ok,EncTextTail} = CryptoEndFun(PlainBuf0, CState0), + EncTextBuf = <<EncBuf0/binary, EncTextTail/binary>>, + case write(Pid, SftpHandle, EncTextBuf, FileOpTimeout) of + ok -> + call(Pid, {erase_bufinf,BufHandle}, FileOpTimeout), + close(Pid, SftpHandle, FileOpTimeout); + Other -> + Other + end + end); + {false,true,[]} -> + {ok,Handle} = open(Pid, File, [read,binary], FileOpTimeout), + erl_tar:init(Pid, read, + fun(read2, {_,Len}) -> + read_repeat(Pid, Handle, Len, FileOpTimeout); + (position, {_,Pos}) -> + position(Pid, Handle, Pos, FileOpTimeout); + (close, _) -> + close(Pid, Handle, FileOpTimeout) + end); + {false,true,[{crypto,{CryptoInitFun,CryptoDecryptFun}}]} -> + {ok,SftpHandle} = open(Pid, File, [read,binary], FileOpTimeout), + BI = #bufinf{mode = read, + crypto_fun = CryptoDecryptFun}, + {ok,BufHandle} = open_buf(Pid, CryptoInitFun, BI, FileOpTimeout), + erl_tar:init(Pid, read, + fun(read2, {_,Len}) -> + read_buf(Pid, SftpHandle, BufHandle, Len, FileOpTimeout); + (position, {_,Pos}) -> + position_buf(Pid, SftpHandle, BufHandle, Pos, FileOpTimeout); + (close, _) -> + call(Pid, {erase_bufinf,BufHandle}, FileOpTimeout), + close(Pid, SftpHandle, FileOpTimeout) + end); + _ -> + {error,{illegal_mode,Mode}} + end. + + opendir(Pid, Path) -> opendir(Pid, Path, ?FILEOP_TIMEOUT). opendir(Pid, Path, FileOpTimeout) -> @@ -392,7 +478,8 @@ write_file_loop(Pid, Handle, Pos, Bin, Remain, PacketSz, FileOpTimeout) -> %% %% Description: %%-------------------------------------------------------------------- -init([Cm, ChannelId, Timeout]) -> +init([Cm, ChannelId, Options]) -> + Timeout = proplists:get_value(timeout, Options, infinity), erlang:monitor(process, Cm), case ssh_connection:subsystem(Cm, ChannelId, "sftp", Timeout) of success -> @@ -401,11 +488,12 @@ init([Cm, ChannelId, Timeout]) -> {ok, #state{xf = Xf, req_id = 0, rep_buf = <<>>, - inf = new_inf()}}; + inf = new_inf(), + opts = Options}}; failure -> - {stop, "server failed to start sftp subsystem"}; + {stop, {shutdown, "server failed to start sftp subsystem"}}; Error -> - {stop, Error} + {stop, {shutdown, Error}} end. %%-------------------------------------------------------------------- @@ -420,12 +508,12 @@ init([Cm, ChannelId, Timeout]) -> %%-------------------------------------------------------------------- handle_call({{timeout, infinity}, wait_for_version_negotiation}, From, #state{xf = #ssh_xfer{vsn = undefined} = Xf} = State) -> - {noreply, State#state{xf = Xf#ssh_xfer{vsn = From}}}; + {noreply, State#state{xf = Xf#ssh_xfer{vsn = {wait, From, undefined}}}}; handle_call({{timeout, Timeout}, wait_for_version_negotiation}, From, #state{xf = #ssh_xfer{vsn = undefined} = Xf} = State) -> - timer:send_after(Timeout, {timeout, undefined, From}), - {noreply, State#state{xf = Xf#ssh_xfer{vsn = From}}}; + TRef = erlang:send_after(Timeout, self(), {timeout, undefined, From}), + {noreply, State#state{xf = Xf#ssh_xfer{vsn = {wait, From, TRef}}}}; handle_call({_, wait_for_version_negotiation}, _, State) -> {reply, ok, State}; @@ -442,6 +530,15 @@ handle_cast(_,State) -> code_change(_OldVsn, State, _Extra) -> {ok, State}. +do_handle_call({get_bufinf,BufHandle}, _From, S=#state{inf=I0}) -> + {reply, dict:find(BufHandle,I0), S}; + +do_handle_call({put_bufinf,BufHandle,B}, _From, S=#state{inf=I0}) -> + {reply, ok, S#state{inf=dict:store(BufHandle,B,I0)}}; + +do_handle_call({erase_bufinf,BufHandle}, _From, S=#state{inf=I0}) -> + {reply, ok, S#state{inf=dict:erase(BufHandle,I0)}}; + do_handle_call({open, Async,FileName,Mode}, From, #state{xf = XF} = State) -> {Access,Flags,Attrs} = open_mode(XF#ssh_xfer.vsn, Mode), ReqID = State#state.req_id, @@ -546,12 +643,7 @@ do_handle_call({read,Async,Handle,Length}, From, State) -> do_handle_call({pwrite,Async,Handle,At,Data0}, From, State) -> case lseek_position(Handle, At, State) of {ok,Offset} -> - Data = if - is_binary(Data0) -> - Data0; - is_list(Data0) -> - list_to_binary(Data0) - end, + Data = to_bin(Data0), ReqID = State#state.req_id, Size = size(Data), ssh_xfer:write(?XF(State),ReqID,Handle,Offset,Data), @@ -564,12 +656,7 @@ do_handle_call({pwrite,Async,Handle,At,Data0}, From, State) -> do_handle_call({write,Async,Handle,Data0}, From, State) -> case lseek_position(Handle, cur, State) of {ok,Offset} -> - Data = if - is_binary(Data0) -> - Data0; - is_list(Data0) -> - list_to_binary(Data0) - end, + Data = to_bin(Data0), ReqID = State#state.req_id, Size = size(Data), ssh_xfer:write(?XF(State),ReqID,Handle,Offset,Data), @@ -707,8 +794,9 @@ handle_ssh_msg({ssh_cm, _, {exit_status, ChannelId, Status}}, State0) -> %% %% Description: Handles channel messages %%-------------------------------------------------------------------- -handle_msg({ssh_channel_up, _, _}, #state{xf = Xf} = State) -> - ssh_xfer:protocol_version_request(Xf), +handle_msg({ssh_channel_up, _, _}, #state{opts = Options, xf = Xf} = State) -> + Version = proplists:get_value(sftp_vsn, Options, ?SSH_SFTP_PROTOCOL_VERSION), + ssh_xfer:protocol_version_request(Xf, Version), {ok, State}; %% Version negotiation timed out @@ -754,6 +842,15 @@ terminate(_Reason, State) -> %%==================================================================== %% Internal functions %%==================================================================== +handle_options([], Sftp, Ssh) -> + {Ssh, Sftp}; +handle_options([{timeout, _} = Opt | Rest], Sftp, Ssh) -> + handle_options(Rest, [Opt | Sftp], Ssh); +handle_options([{sftp_vsn, _} = Opt| Rest], Sftp, Ssh) -> + handle_options(Rest, [Opt | Sftp], Ssh); +handle_options([Opt | Rest], Sftp, Ssh) -> + handle_options(Rest, Sftp, [Opt | Ssh]). + call(Pid, Msg, TimeOut) -> ssh_channel:call(Pid, {{timeout, TimeOut}, Msg}, infinity). @@ -768,7 +865,12 @@ do_handle_reply(#state{xf = Xf} = State, case Xf#ssh_xfer.vsn of undefined -> ok; - From -> + {wait, From, TRef} -> + if is_reference(TRef) -> + erlang:cancel_timer(TRef); + true -> + ok + end, ssh_channel:reply(From, ok) end, State#state{xf = Xf#ssh_xfer{vsn = Version, ext = Ext}, rep_buf = Rest}; @@ -1111,5 +1213,212 @@ lseek_pos({eof, Offset}, _CurOffset, CurSize) end; lseek_pos(_, _, _) -> {error, einval}. - +%%%================================================================ +%%% +to_bin(Data) when is_list(Data) -> list_to_binary(Data); +to_bin(Data) when is_binary(Data) -> Data. + + +read_repeat(Pid, Handle, Len, FileOpTimeout) -> + {ok,{_WindowSz,PacketSz}} = recv_window(Pid, FileOpTimeout), + read_rpt(Pid, Handle, Len, PacketSz, FileOpTimeout, <<>>). + +read_rpt(Pid, Handle, WantedLen, PacketSz, FileOpTimeout, Acc) when WantedLen > 0 -> + case read(Pid, Handle, min(WantedLen,PacketSz), FileOpTimeout) of + {ok, Data} -> + read_rpt(Pid, Handle, WantedLen-size(Data), PacketSz, FileOpTimeout, <<Acc/binary, Data/binary>>); + eof -> + {ok, Acc}; + Error -> + Error + end; +read_rpt(_Pid, _Handle, WantedLen, _PacketSz, _FileOpTimeout, Acc) when WantedLen >= 0 -> + {ok,Acc}. + + +write_to_remote_tar(_Pid, _SftpHandle, <<>>, _FileOpTimeout) -> + ok; +write_to_remote_tar(Pid, SftpHandle, Bin, FileOpTimeout) -> + {ok,{_Window,Packet}} = send_window(Pid, FileOpTimeout), + write_file_loop(Pid, SftpHandle, 0, Bin, size(Bin), Packet, FileOpTimeout). + +position_buf(Pid, SftpHandle, BufHandle, Pos, FileOpTimeout) -> + {ok,#bufinf{mode = Mode, + plain_text_buf = Buf0, + size = Size}} = call(Pid, {get_bufinf,BufHandle}, FileOpTimeout), + case Pos of + {cur,0} when Mode==write -> + {ok,Size+size(Buf0)}; + + {cur,0} when Mode==read -> + {ok,Size}; + + _ when Mode==read, is_integer(Pos) -> + Skip = Pos-Size, + if + Skip < 0 -> + {error, cannot_rewind}; + Skip == 0 -> + %% Optimization + {ok,Pos}; + Skip > 0 -> + case read_buf(Pid, SftpHandle, BufHandle, Skip, FileOpTimeout) of + %% A bit innefficient to fetch the bufinf again, but there are lots of + %% other more important optimizations waiting.... + {ok,_} -> + {ok,Pos}; + Other -> + Other + end + end; + + _ -> + {error,{not_yet_implemented,{pos,Pos}}} + end. + +read_buf(Pid, SftpHandle, BufHandle, WantedLen, FileOpTimeout) -> + {ok,{_Window,Packet}} = send_window(Pid, FileOpTimeout), + {ok,B0} = call(Pid, {get_bufinf,BufHandle}, FileOpTimeout), + case do_the_read_buf(Pid, SftpHandle, WantedLen, Packet, FileOpTimeout, B0) of + {ok,ResultBin,B} -> + call(Pid, {put_bufinf,BufHandle,B}, FileOpTimeout), + {ok,ResultBin}; + {error,Error} -> + {error,Error}; + {eof,B} -> + call(Pid, {put_bufinf,BufHandle,B}, FileOpTimeout), + eof + end. + +do_the_read_buf(_Pid, _SftpHandle, WantedLen, _Packet, _FileOpTimeout, + B=#bufinf{plain_text_buf=PlainBuf0, + size = Size}) + when size(PlainBuf0) >= WantedLen -> + %% We already have the wanted number of bytes decoded and ready! + <<ResultBin:WantedLen/binary, PlainBuf/binary>> = PlainBuf0, + {ok,ResultBin,B#bufinf{plain_text_buf=PlainBuf, + size = Size + WantedLen}}; + +do_the_read_buf(Pid, SftpHandle, WantedLen, Packet, FileOpTimeout, + B0=#bufinf{plain_text_buf = PlainBuf0, + enc_text_buf = EncBuf0, + chunksize = undefined + }) + when size(EncBuf0) > 0 -> + %% We have (at least) one decodable byte waiting for decodeing. + {ok,DecodedBin,B} = apply_crypto(EncBuf0, B0), + do_the_read_buf(Pid, SftpHandle, WantedLen, Packet, FileOpTimeout, + B#bufinf{plain_text_buf = <<PlainBuf0/binary, DecodedBin/binary>>, + enc_text_buf = <<>> + }); + +do_the_read_buf(Pid, SftpHandle, WantedLen, Packet, FileOpTimeout, + B0=#bufinf{plain_text_buf = PlainBuf0, + enc_text_buf = EncBuf0, + chunksize = ChunkSize0 + }) + when size(EncBuf0) >= ChunkSize0 -> + %% We have (at least) one chunk of decodable bytes waiting for decodeing. + <<ToDecode:ChunkSize0/binary, EncBuf/binary>> = EncBuf0, + {ok,DecodedBin,B} = apply_crypto(ToDecode, B0), + do_the_read_buf(Pid, SftpHandle, WantedLen, Packet, FileOpTimeout, + B#bufinf{plain_text_buf = <<PlainBuf0/binary, DecodedBin/binary>>, + enc_text_buf = EncBuf + }); + +do_the_read_buf(Pid, SftpHandle, WantedLen, Packet, FileOpTimeout, B=#bufinf{enc_text_buf = EncBuf0}) -> + %% We must read more bytes and append to the buffer of encoded bytes. + case read(Pid, SftpHandle, Packet, FileOpTimeout) of + {ok,EncryptedBin} -> + do_the_read_buf(Pid, SftpHandle, WantedLen, Packet, FileOpTimeout, + B#bufinf{enc_text_buf = <<EncBuf0/binary, EncryptedBin/binary>>}); + eof -> + {eof,B}; + Other -> + Other + end. + + +write_buf(Pid, SftpHandle, BufHandle, PlainBin, FileOpTimeout) -> + {ok,{_Window,Packet}} = send_window(Pid, FileOpTimeout), + {ok,B0=#bufinf{plain_text_buf=PTB}} = call(Pid, {get_bufinf,BufHandle}, FileOpTimeout), + case do_the_write_buf(Pid, SftpHandle, Packet, FileOpTimeout, + B0#bufinf{plain_text_buf = <<PTB/binary,PlainBin/binary>>}) of + {ok, B} -> + call(Pid, {put_bufinf,BufHandle,B}, FileOpTimeout), + ok; + {error,Error} -> + {error,Error} + end. + +do_the_write_buf(Pid, SftpHandle, Packet, FileOpTimeout, + B=#bufinf{enc_text_buf = EncBuf0, + size = Size}) + when size(EncBuf0) >= Packet -> + <<BinToWrite:Packet/binary, EncBuf/binary>> = EncBuf0, + case write(Pid, SftpHandle, BinToWrite, FileOpTimeout) of + ok -> + do_the_write_buf(Pid, SftpHandle, Packet, FileOpTimeout, + B#bufinf{enc_text_buf = EncBuf, + size = Size + Packet}); + Other -> + Other + end; + +do_the_write_buf(Pid, SftpHandle, Packet, FileOpTimeout, + B0=#bufinf{plain_text_buf = PlainBuf0, + enc_text_buf = EncBuf0, + chunksize = undefined}) + when size(PlainBuf0) > 0 -> + {ok,EncodedBin,B} = apply_crypto(PlainBuf0, B0), + do_the_write_buf(Pid, SftpHandle, Packet, FileOpTimeout, + B#bufinf{plain_text_buf = <<>>, + enc_text_buf = <<EncBuf0/binary, EncodedBin/binary>>}); + +do_the_write_buf(Pid, SftpHandle, Packet, FileOpTimeout, + B0=#bufinf{plain_text_buf = PlainBuf0, + enc_text_buf = EncBuf0, + chunksize = ChunkSize0 + }) + when size(PlainBuf0) >= ChunkSize0 -> + <<ToEncode:ChunkSize0/binary, PlainBuf/binary>> = PlainBuf0, + {ok,EncodedBin,B} = apply_crypto(ToEncode, B0), + do_the_write_buf(Pid, SftpHandle, Packet, FileOpTimeout, + B#bufinf{plain_text_buf = PlainBuf, + enc_text_buf = <<EncBuf0/binary, EncodedBin/binary>>}); + +do_the_write_buf(_Pid, _SftpHandle, _Packet, _FileOpTimeout, B) -> + {ok,B}. + +apply_crypto(In, B=#bufinf{crypto_state = CState0, + crypto_fun = F}) -> + case F(In,CState0) of + {ok,EncodedBin,CState} -> + {ok, EncodedBin, B#bufinf{crypto_state=CState}}; + {ok,EncodedBin,CState,ChunkSize} -> + {ok, EncodedBin, B#bufinf{crypto_state=CState, + chunksize=ChunkSize}} + end. + +open_buf(Pid, CryptoInitFun, BufInfo0, FileOpTimeout) -> + case CryptoInitFun() of + {ok,CryptoState} -> + open_buf1(Pid, BufInfo0, FileOpTimeout, CryptoState, undefined); + {ok,CryptoState,ChunkSize} -> + open_buf1(Pid, BufInfo0, FileOpTimeout, CryptoState, ChunkSize); + Other -> + Other + end. + +open_buf1(Pid, BufInfo0, FileOpTimeout, CryptoState, ChunkSize) -> + BufInfo = BufInfo0#bufinf{crypto_state = CryptoState, + chunksize = ChunkSize}, + BufHandle = make_ref(), + call(Pid, {put_bufinf,BufHandle,BufInfo}, FileOpTimeout), + {ok,BufHandle}. + +format_channel_start_error({shutdown, Reason}) -> + Reason; +format_channel_start_error(Reason) -> + Reason. |