diff --git a/uppsrc/Core/SSH/Channels.cpp b/uppsrc/Core/SSH/Channels.cpp index a06d305cb..dd428bb66 100644 --- a/uppsrc/Core/SSH/Channels.cpp +++ b/uppsrc/Core/SSH/Channels.cpp @@ -35,7 +35,7 @@ void SshChannel::Exit() LLOG("Channel succesfully freed."); } return !rc; - }); + }, false); } bool SshChannel::Open() @@ -203,11 +203,11 @@ int SshChannel::Get(void *ptr, int size, int sid) { done = 0; Run([=]() mutable { - while(done < size && InProgress() && !IsEof() && !IsTimeout()) { + while(done < size && !IsEof() && !IsTimeout()) { int rc = Read(ptr, size, sid); - if(rc == 0) break; - if(rc > 0) RefreshUI(); + if(rc > 0) UpdateClient(); if(rc < 0) return false; + if(!rc) break; } return true; }); @@ -219,7 +219,7 @@ String SshChannel::Get(int size, int sid) StringBuffer sb(size); int len = Get(~sb, size, sid); sb.SetCount(len); - return sb; + return pick(String(sb)); } String SshChannel::GetLine(int maxlen, int sid) @@ -243,7 +243,7 @@ String SshChannel::GetLine(int maxlen, int sid) done++; } } - while(!eol && !IsEof() && !IsTimeout() && InProgress()); + while(!eol && !IsEof() && !IsTimeout()); return eol || IsEof(); }); return line; @@ -253,11 +253,11 @@ int SshChannel::Put(const void *ptr, int size, int sid) { done = 0; Run([=]() mutable { - while(done < size && InProgress() && !IsEof() && !IsTimeout()) { + while(done < size && !IsEof() && !IsTimeout()) { int rc = Write(ptr, size, sid); - if(rc == 0) break; - if(rc > 0) RefreshUI(); + if(rc > 0) UpdateClient(); if(rc < 0) return false; + if(!rc) break; } return true; }); @@ -271,9 +271,10 @@ int SshChannel::Read(void *ptr, int size, int sid) int rc = static_cast( libssh2_channel_read_ex(*channel, sid, (char*) ptr + done, size_t(sz)) ); - - if(rc < 0 && !WouldBlock(rc)) + if(rc < 0 && !WouldBlock(rc)) { SetError(rc); + } + else if(rc > 0) { done += rc; ssh->start_time = msecs(); @@ -296,9 +297,10 @@ int SshChannel::Write(const void *ptr, int size, int sid) int rc = static_cast( libssh2_channel_write_ex(*channel, sid, (const char*) ptr + done, size_t(sz)) ); - - if(!WouldBlock(rc) && rc < 0) + if(rc < 0 && !WouldBlock(rc)) { SetError(rc); + } + else if(rc > 0) { done += rc; ssh->start_time = msecs(); diff --git a/uppsrc/Core/SSH/Core.cpp b/uppsrc/Core/SSH/Core.cpp index 04af7fb79..6a98ac627 100644 --- a/uppsrc/Core/SSH/Core.cpp +++ b/uppsrc/Core/SSH/Core.cpp @@ -47,38 +47,32 @@ String GetName(int type, int64 id) static StaticMutex sLoopLock; -void Ssh::Check() +bool Ssh::Run(Gate<>&& fn, bool abortable) { - auto sock = ssh->socket; + auto Do = [=, &fn]() + { + Mutex::Lock __(sLoopLock); - if(IsTimeout()) - SetError(-1, "Operation timed out."); + if(IsTimeout()) + SetError(-1, "Operation timed out."); - if(ssh->status == ABORTED) - SetError(-1, "Operation aborted."); + if(abortable && ssh->status == ABORTED) + SetError(-1, "Operation aborted."); + + if(ssh->socket && ssh->socket->IsError()) + SetError(-1, "[Socket error]: " << ssh->socket->GetErrorDesc()); - if(sock && ssh->socket->IsError()) - SetError(-1, "[Socket error]: " << ssh->socket->GetErrorDesc()); -} - -bool Ssh::Do(Gate<>& fn) -{ - Mutex::Lock __(sLoopLock); - - Check(); - if(!ssh->init) - ssh->init = Init(); - return !ssh->init || !fn(); -} - -bool Ssh::Run(Gate<>&& fn) -{ + if(!ssh->init) + ssh->init = Init(); + + return !ssh->init || ! fn(); + }; + try { ssh->status = WORKING; ssh->start_time = msecs(); - - while(Do(fn)) - Wait(); + + while(Do()) Wait(); ssh->status = IDLE; } @@ -88,12 +82,13 @@ bool Ssh::Run(Gate<>&& fn) catch(...) { ReportError(-1, "Unhandled exception."); } + return !IsError(); } void Ssh::Wait() { - RefreshUI(); + UpdateClient(); if(!ssh->socket || !ssh->session) return; dword q = 0, r = libssh2_session_block_directions(ssh->session); @@ -131,8 +126,8 @@ void Ssh::ReportError(int rc, const String& reason) int64 Ssh::GetNewId() { - static int64 objectid; - return objectid == INT64_MAX ? objectid = 1 : ++objectid; + static int64 objectid = 0; + return ++objectid; } Ssh::Ssh() diff --git a/uppsrc/Core/SSH/Core.h b/uppsrc/Core/SSH/Core.h index 65a020e37..f0bddc0a7 100644 --- a/uppsrc/Core/SSH/Core.h +++ b/uppsrc/Core/SSH/Core.h @@ -57,15 +57,13 @@ protected: virtual bool Init() { return true; } virtual void Exit() {} void Wait(); - void Check(); - bool Do(Gate<>& fn); - bool Run(Gate<>&& fn); + bool Run(Gate<>&& fn, bool abortable = true); bool WouldBlock(int rc) { return rc == LIBSSH2_ERROR_EAGAIN; } bool WouldBlock() { return ssh->session && WouldBlock(libssh2_session_last_errno(ssh->session)); } bool IsTimeout() const { return !IsNull(ssh->timeout) && ssh->timeout > 0 && msecs(ssh->start_time) >= ssh->timeout; } void SetError(int rc, const String& reason = Null); void ReportError(int rc, const String& reason); - void RefreshUI() { WhenWait ? WhenWait() : ssh->whenwait(); } + void UpdateClient() { WhenWait ? WhenWait() : ssh->whenwait(); } private: static int64 GetNewId(); diff --git a/uppsrc/Core/SSH/SFtp.cpp b/uppsrc/Core/SSH/SFtp.cpp index b7b610b1c..8b34e1e2c 100644 --- a/uppsrc/Core/SSH/SFtp.cpp +++ b/uppsrc/Core/SSH/SFtp.cpp @@ -31,7 +31,7 @@ void SFtp::Exit() sftp_session.Clear(); LLOG("Session deinitalized."); return true; - }); + }, false); } int SFtp::FStat(SFtpHandle handle, SFtpAttrs& a, bool set) @@ -70,7 +70,7 @@ void SFtp::Close(SFtpHandle handle) int rc = libssh2_sftp_close_handle(handle); if(!rc) LLOG("File handle freed."); return !rc; - }); + }, false); } bool SFtp::Rename(const String& oldpath, const String& newpath) @@ -113,7 +113,7 @@ SFtp& SFtp::Seek(SFtpHandle handle, int64 position) { INTERLOCKED { - LLOG("Seeking to offset: " << position); + // LLOG("Seeking to offset: " << position); libssh2_sftp_seek64(handle, position); } return *this; @@ -126,103 +126,98 @@ int64 SFtp::GetPos(SFtpHandle handle) INTERLOCKED { pos = libssh2_sftp_tell64(handle); - LLOG("File position: " << pos); + // LLOG("File position: " << pos); }; return pos; } -int SFtp::Read(SFtpHandle handle, void* ptr, int size) -{ - int sz = min(size - done, ssh->chunk_size); - - int rc = static_cast( - libssh2_sftp_read(handle, (char*) ptr + done, size_t(sz)) - ); - - if(!WouldBlock(rc) && rc < 0) - SetError(rc); - if(rc > 0) { - done += rc; - ssh->start_time = msecs(); - RefreshUI(); - } - if(!rc) - LLOG("EOF received."); - return rc; -} - -int SFtp::Write(SFtpHandle handle, const void* ptr, int size) -{ - int sz = min(size - done, ssh->chunk_size); - - int rc = static_cast( - libssh2_sftp_write(handle, (const char*) ptr + done, size_t(sz)) - ); - - if(!WouldBlock(rc) && rc < 0) - SetError(rc); - if(rc > 0) { - done += rc; - ssh->start_time = msecs(); - RefreshUI(); - } - if(!rc) - LLOG("EOF received."); - return rc; -} - int SFtp::Get(SFtpHandle handle, void *ptr, int size) { done = 0; + Run([=]() mutable { - while(done < size && !IsTimeout() && InProgress()) { - int rc = Read(handle, ptr, size); - if(rc < 0) return false; - if(!rc) break; + while(done < size && !IsTimeout()) { + int rc = static_cast( + libssh2_sftp_read(handle, (char*) ptr + done, min(size - done, ssh->chunk_size)) + ); + if(rc < 0) { + if(!WouldBlock(rc)) + SetError(rc); + return false; + } + else + if(rc == 0) { + LLOG("EOF received."); + break; + } + done += rc; + ssh->start_time = msecs(); + UpdateClient(); } return true; }); + return GetDone(); } int SFtp::Put(SFtpHandle handle, const void *ptr, int size) { done = 0; + Run([=]() mutable { - while(done < size && !IsTimeout() && InProgress()) { - int rc = Write(handle, ptr, size); - if(rc < 0) return false; - if(!rc) break; + while(done < size && !IsTimeout()) { + int rc = static_cast( + libssh2_sftp_write(handle, (const char*) ptr + done, min(size - done, ssh->chunk_size)) + ); + if(rc < 0) { + if(!WouldBlock(rc)) + SetError(rc); + return false; + } + else + if(rc == 0) { + LLOG("EOF received."); + break; + } + done += rc; + ssh->start_time = msecs(); + UpdateClient(); } return true; }); + return GetDone(); } bool SFtp::CopyData(Stream& dest, Stream& src, int64 maxsize) { - int64 size = src.GetSize(); - String err; - if(IsError()) return false; - - if(size < 0 || size >= maxsize) { - err = Format("Buffer overflow. size = %d (allowed size >= 0 && < %d", size, maxsize); - goto Bailout; - } - LLOG("Transfer chunk size: " << ssh->chunk_size); - if(CopyStream(dest, src, src.GetSize(), WhenProgress, ssh->chunk_size) < 0) { - err = "File transfer is aborted."; - goto Bailout; - } - return !IsError(); -Bailout: - src.Close(); - dest.Close(); - ReportError(-1, err); - return false; + int64 size = src.GetSize(), count = 0; + Buffer chunk(ssh->chunk_size, 0); + + WhenProgress(0, size); + + while(!src.IsEof()) { + int n = src.Get(chunk, (int) min(size - count, ssh->chunk_size)); + if(n > 0) { + dest.Put(chunk, n); + if(dest.IsError()) { + LLOG("Stream write error. " + src.GetErrorText()); + return false; + } + count += n; + if(WhenProgress(count, size)) { + return false; + } + } + if(src.IsError()) { + LLOG("Stream read error. " + src.GetErrorText()); + break; + } + } + return !src.IsError(); } bool SFtp::SaveFile(const char *path, const String& data) diff --git a/uppsrc/Core/SSH/Scp.cpp b/uppsrc/Core/SSH/Scp.cpp index e991f3f5a..aca9e91ae 100644 --- a/uppsrc/Core/SSH/Scp.cpp +++ b/uppsrc/Core/SSH/Scp.cpp @@ -44,21 +44,22 @@ bool Scp::Load(Stream& s, ScpAttrs a, int64 maxsize) int64 done_ = 0; int64 size = a.st_size; String msg; - + if(size < 0 || size >= maxsize) { msg = "Invald stream size."; } - else - while(done_ < size && !IsEof() && !IsError()) { - int csz = (int) min(size - done_, ssh->chunk_size); - Buffer chunk(csz, 0); - int n = Get(chunk, csz); - if(n > 0) { - done_ += n; - s.Put(chunk, n); - if((nowait = WhenProgress(done_, size))) { - msg = "File transfer is aborted."; - break; + else { + WhenProgress(0, size); + Buffer chunk(ssh->chunk_size); + while(done_ < size && !IsEof() && !IsError()) { + int n = Get(chunk, (int) min(size - done_, ssh->chunk_size)); + if(n > 0) { + done_ += n; + s.Put(chunk, n); + if((nowait = WhenProgress(done_, size))) { + msg = "File transfer is aborted."; + break; + } } } } @@ -72,8 +73,8 @@ bool Scp::Save(Stream& s) int64 size = s.GetSize(); String msg; - Buffer chunk(ssh->chunk_size, 0); - + WhenProgress(0, size); + Buffer chunk(ssh->chunk_size); while(done_ < size && !IsEof() && !IsError()) { int l = s.Get(chunk, (int) min(size - done_, ssh->chunk_size)); int n = Put(chunk, l); diff --git a/uppsrc/Core/SSH/Session.cpp b/uppsrc/Core/SSH/Session.cpp index 30abcd655..bbeb0e833 100644 --- a/uppsrc/Core/SSH/Session.cpp +++ b/uppsrc/Core/SSH/Session.cpp @@ -12,7 +12,7 @@ static void ssh_keyboard_callback(const char *name, int name_len, const char *in int instruction_len, int num_prompts, const LIBSSH2_USERAUTH_KBDINT_PROMPT *prompts, LIBSSH2_USERAUTH_KBDINT_RESPONSE *responses, void **abstract) { - SshSession *session = static_cast(*abstract); + SshSession *session = reinterpret_cast(*abstract); for(auto i = 0; i < num_prompts; i++) { auto response = session->WhenKeyboard( String(name, name_len), @@ -36,7 +36,7 @@ static void ssh_keyboard_callback(const char *name, int name_len, const char *in static void ssh_password_change(LIBSSH2_SESSION *session, char **pwd, int *len, void **abstract) { - String newpwd = static_cast(*abstract)->WhenPasswordChange(); + String newpwd = reinterpret_cast(*abstract)->WhenPasswordChange(); #ifdef UPP_HEAP *pwd = (char*) ssh_malloc(newpwd.GetLength(), abstract); memcpy(*pwd, ~newpwd, newpwd.GetLength()); @@ -49,17 +49,17 @@ static void ssh_password_change(LIBSSH2_SESSION *session, char **pwd, int *len, static void ssh_x11_request(LIBSSH2_SESSION *session, LIBSSH2_CHANNEL *channel, char *shost, int sport, void **abstract) { - static_cast(*abstract)->WhenX11((SshX11Handle) channel); + reinterpret_cast(*abstract)->WhenX11((SshX11Handle) channel); } // ssh_session_libtrace: Allows full-level logging (redirection) of libsssh2 diagnostic messages. #ifdef flagLIBSSH2TRACE -static void ssh_session_libtrace(LIBSSH2_SESSION *session, void* context, const char*data, size_t length) +static void ssh_session_libtrace(LIBSSH2_SESSION *session, void *context, const char *data, size_t length) { if(!session || !SSH::sTraceVerbose) return; - auto* ssh_obj = static_cast(context); + auto* ssh_obj = reinterpret_cast(context); RLOG(SSH::GetName(ssh_obj->GetType(), ssh_obj->GetId()) << String(data, int64(length))); } #endif @@ -91,22 +91,18 @@ void SshSession::Exit() session->connected = false; LLOG("Session handles freed."); return true; - }); + }, false); } bool SshSession::Connect(const String& url) { UrlInfo u(url); - auto b = u.scheme == "ssh" || - u.scheme == "scp" || - u.scheme == "sftp" || - u.scheme == "exec" || - (u.scheme.IsEmpty() && !u.host.IsEmpty()); + auto b = findarg(u.scheme, "ssh", "sftp", "scp", "exec") >= 0 || (u.scheme.IsEmpty() && !u.host.IsEmpty()); int port = (u.port.IsEmpty() || !b) ? 22 : StrInt(u.port); - - return b ? Connect(u.host, port, u.username, u.password) - : Run([=]{ SetError(-1, "Malformed secure shell URL."); return false; }); + if(b) return Connect(u.host, port, u.username, u.password); + ReportError(-1, "Malformed secure shell URL."); + return false; } bool SshSession::Connect(const String& host, int port, const String& user, const String& password)