AI_Slop/calog
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
calog/libs/calogSsh.c

902 lines
33 KiB
C
Raw Blame History

// calogSsh.c -- calog SSH library (see calogSsh.h). SSH command execution and SFTP file
// transfer over the shared typed handle table, bound to libssh2. Every blocking call is an
// inline native, so it stalls only the calling script's context thread. Sessions run in
// libssh2 blocking mode. A connection handle has no owning context -- any context it is passed
// to may use it -- so the caller must serialize access (see the threading notes in calogSsh.h).
#define _GNU_SOURCE
#include "calogSsh.h"
#include "calogHandle.h"
#include "calogInternal.h"
#include "calogPlatform.h"
#include <pthread.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#ifndef _WIN32
#include <sys/select.h> // fd_set/select come from winsock2.h on Windows (via calogPlatform.h)
#endif
#include <libssh2.h>
#include <libssh2_sftp.h>
// Sole handle type tag for this library, distinct so a stray handle of another kind fails to
// resolve here.
#define SSH_TYPE_CONN 1u
#define SSH_PORT_MAX 65535
// Upper bound on a single read-to-EOF transfer (sshExec output, sftpGet file), matching
// calogNet/calogHttp: without it a huge/endless remote source would grow an unbounded buffer
// and OOM-kill the shared host process. Exceeding it fails with calogErrRangeE; a stream of
// exactly this many bytes succeeds (see sshDrainStep's boundary probe).
#define SSH_MAX_TRANSFER (64 * 1024 * 1024)
// libssh2_sftp_readdir fails the whole listing (LIBSSH2_ERROR_BUFFER_TOO_SMALL) on a remote
// entry name at or beyond this size; real filesystems cap names at 255 bytes (NAME_MAX), so
// this leaves comfortable margin.
#define SSH_NAME_MAX 512
// What a connection handle owns: the socket, the libssh2 session, and a lazily-opened+cached
// SFTP subsystem. The handle is a plain process-wide entry with no owning context: any context
// it is passed to may operate on it (see the threading notes in calogSsh.h).
typedef struct SshConnT {
CalogSocketT sock;
LIBSSH2_SESSION *session;
LIBSSH2_SFTP *sftp;
} SshConnT;
// Process-wide SSH library state shared by every runtime that registers the natives; refCount
// is one per registered runtime, so libssh2_exit runs (and the table is destroyed) only when
// the LAST runtime shuts down.
typedef struct SshLibT {
CalogHandleTableT *handles;
int32_t refCount;
} SshLibT;
// Read adapter so sshDrainStep works uniformly over a channel stream (libssh2_channel_read_ex)
// and an SFTP file (libssh2_sftp_read); the SFTP adapter ignores streamId.
typedef ssize_t (*SshReadFnT)(void *handle, int streamId, char *buffer, size_t length);
// Outcome of one sshDrainStep call.
typedef enum SshStepE {
sshStepDataE,
sshStepEofE,
sshStepAgainE
} SshStepE;
static pthread_mutex_t gSshLibMutex = PTHREAD_MUTEX_INITIALIZER;
static SshLibT *gSshLib = NULL;
static int32_t sftpGet(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpList(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpMkdir(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpPut(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpRemove(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sftpStat(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshAuthKey(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshAuthPassword(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshClose(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static void sshCloser(uint32_t type, void *resource);
static void sshConnDestroy(SshConnT *conn, const char *reason);
static int32_t sshConnect(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshDrainStep(SshReadFnT readFn, void *handle, int streamId, char **bytes, int64_t *cap, int64_t *length, SshStepE *step);
static int32_t sshEnsureSftp(SshConnT *conn, CalogValueT *result);
static int32_t sshExec(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData);
static int32_t sshExecDrain(CalogSocketT sock, LIBSSH2_SESSION *session, LIBSSH2_CHANNEL *channel, char **outBytes, int64_t *outLength, char **errBytes, int64_t *errLength);
static ssize_t sshReadChannel(void *handle, int streamId, char *buffer, size_t length);
static ssize_t sshReadSftpFile(void *handle, int streamId, char *buffer, size_t length);
static int32_t sshResolve(SshLibT *lib, int64_t handle, CalogValueT *result, SshConnT **out);
static void sshSessionDestroy(LIBSSH2_SESSION *session, CalogSocketT sock, const char *reason);
static void sshWaitSocket(CalogSocketT sock, LIBSSH2_SESSION *session);
int32_t calogSshRegister(CalogT *calog) {
pthread_mutex_lock(&gSshLibMutex);
if (gSshLib == NULL) {
SshLibT *lib;
// Windows needs WSAStartup before any socket use (no-op on POSIX).
if (calogPlatformNetInit() != 0) {
pthread_mutex_unlock(&gSshLibMutex);
return calogErrUnsupportedE;
}
lib = (SshLibT *)calloc(1, sizeof(*lib));
if (lib == NULL) {
calogPlatformNetShutdown();
pthread_mutex_unlock(&gSshLibMutex);
return calogErrOomE;
}
lib->handles = calogHandleTableCreate();
if (lib->handles == NULL) {
free(lib);
calogPlatformNetShutdown();
pthread_mutex_unlock(&gSshLibMutex);
return calogErrOomE;
}
// libssh2_init uses global state and is not thread safe; the lib mutex serialises it.
if (libssh2_init(0) != 0) {
calogHandleTableDestroy(lib->handles, NULL);
free(lib);
calogPlatformNetShutdown();
pthread_mutex_unlock(&gSshLibMutex);
return calogErrUnsupportedE;
}
gSshLib = lib;
}
gSshLib->refCount++;
pthread_mutex_unlock(&gSshLibMutex);
calogRegisterInline(calog, "sshConnect", sshConnect, gSshLib);
calogRegisterInline(calog, "sshAuthPassword", sshAuthPassword, gSshLib);
calogRegisterInline(calog, "sshAuthKey", sshAuthKey, gSshLib);
calogRegisterInline(calog, "sshExec", sshExec, gSshLib);
calogRegisterInline(calog, "sshClose", sshClose, gSshLib);
calogRegisterInline(calog, "sftpGet", sftpGet, gSshLib);
calogRegisterInline(calog, "sftpPut", sftpPut, gSshLib);
calogRegisterInline(calog, "sftpList", sftpList, gSshLib);
calogRegisterInline(calog, "sftpStat", sftpStat, gSshLib);
calogRegisterInline(calog, "sftpRemove", sftpRemove, gSshLib);
calogRegisterInline(calog, "sftpMkdir", sftpMkdir, gSshLib);
return calogOkE;
}
void calogSshShutdown(void) {
pthread_mutex_lock(&gSshLibMutex);
if (gSshLib == NULL) {
pthread_mutex_unlock(&gSshLibMutex);
return;
}
gSshLib->refCount--;
if (gSshLib->refCount <= 0) {
calogHandleTableDestroy(gSshLib->handles, sshCloser);
libssh2_exit();
calogPlatformNetShutdown();
free(gSshLib);
gSshLib = NULL;
}
pthread_mutex_unlock(&gSshLibMutex);
}
static int32_t sftpGet(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
LIBSSH2_SFTP_HANDLE *file;
char *buffer;
int64_t cap;
int64_t length;
SshStepE step;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
return calogFail(result, calogErrArgE, "sftpGet expects (handle, remotePath)");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
status = sshEnsureSftp(conn, result);
if (status != calogOkE) {
return status;
}
file = libssh2_sftp_open(conn->sftp, args[1].as.s.bytes, LIBSSH2_FXF_READ, 0);
if (file == NULL) {
return calogFail(result, calogErrNotFoundE, "sftpGet: could not open the remote file");
}
buffer = NULL;
cap = 0;
length = 0;
for (;;) {
status = sshDrainStep(sshReadSftpFile, file, 0, &buffer, &cap, &length, &step);
if (status != calogOkE) {
free(buffer);
libssh2_sftp_close(file);
if (status == calogErrRangeE) {
return calogFail(result, status, "sftpGet: file exceeds the maximum transfer size");
}
if (status == calogErrOomE) {
return calogFail(result, status, "sftpGet: out of memory");
}
return calogFail(result, status, "sftpGet: read failed");
}
if (step == sshStepEofE) {
break;
}
}
libssh2_sftp_close(file);
status = calogValueString(result, buffer, (int64_t)length);
free(buffer);
return status;
}
static int32_t sftpList(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
LIBSSH2_SFTP_HANDLE *dir;
CalogAggT *list;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
return calogFail(result, calogErrArgE, "sftpList expects (handle, path)");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
status = sshEnsureSftp(conn, result);
if (status != calogOkE) {
return status;
}
dir = libssh2_sftp_opendir(conn->sftp, args[1].as.s.bytes);
if (dir == NULL) {
return calogFail(result, calogErrArgE, "sftpList: could not open the directory");
}
status = calogAggCreate(&list, calogListE);
if (status != calogOkE) {
libssh2_sftp_closedir(dir);
return calogFail(result, status, "sftpList: out of memory");
}
for (;;) {
LIBSSH2_SFTP_ATTRIBUTES attrs;
CalogAggT *entry;
CalogValueT entryValue;
char name[SSH_NAME_MAX];
int rc;
rc = libssh2_sftp_readdir(dir, name, sizeof(name), &attrs);
if (rc == 0) {
break;
}
if (rc < 0) {
calogAggFree(list);
libssh2_sftp_closedir(dir);
return calogFail(result, calogErrArgE, "sftpList: read failed");
}
// Skip "." and ".." so callers see only real entries.
if ((rc == 1 && name[0] == '.') || (rc == 2 && name[0] == '.' && name[1] == '.')) {
continue;
}
status = calogAggCreate(&entry, calogMapE);
if (status != calogOkE) {
calogAggFree(list);
libssh2_sftp_closedir(dir);
return calogFail(result, status, "sftpList: out of memory");
}
status = calogMapSetStr(entry, "name", name, (int64_t)rc);
if (status == calogOkE) {
int64_t size;
size = (attrs.flags & LIBSSH2_SFTP_ATTR_SIZE) ? (int64_t)attrs.filesize : 0;
status = calogMapSetInt(entry, "size", size);
}
if (status == calogOkE) {
bool isDir;
isDir = (attrs.flags & LIBSSH2_SFTP_ATTR_PERMISSIONS) && LIBSSH2_SFTP_S_ISDIR(attrs.permissions);
status = calogMapSetBool(entry, "isDir", isDir);
}
if (status != calogOkE) {
calogAggFree(entry);
calogAggFree(list);
libssh2_sftp_closedir(dir);
return calogFail(result, status, "sftpList: failed to build an entry");
}
calogValueAgg(&entryValue, entry);
status = calogAggPush(list, &entryValue);
if (status != calogOkE) {
calogValueFree(&entryValue);
calogAggFree(list);
libssh2_sftp_closedir(dir);
return calogFail(result, status, "sftpList: out of memory");
}
}
libssh2_sftp_closedir(dir);
calogValueAgg(result, list);
return calogOkE;
}
static int32_t sftpMkdir(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
return calogFail(result, calogErrArgE, "sftpMkdir expects (handle, path)");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
status = sshEnsureSftp(conn, result);
if (status != calogOkE) {
return status;
}
if (libssh2_sftp_mkdir(conn->sftp, args[1].as.s.bytes, 0755) != 0) {
return calogFail(result, calogErrArgE, "sftpMkdir: could not create the directory");
}
return calogOkE;
}
static int32_t sftpPut(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
LIBSSH2_SFTP_HANDLE *file;
int64_t total;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 3 || args[0].type != calogIntE || args[1].type != calogStringE || args[2].type != calogStringE) {
return calogFail(result, calogErrArgE, "sftpPut expects (handle, remotePath, data)");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
status = sshEnsureSftp(conn, result);
if (status != calogOkE) {
return status;
}
file = libssh2_sftp_open(conn->sftp, args[1].as.s.bytes,
LIBSSH2_FXF_WRITE | LIBSSH2_FXF_CREAT | LIBSSH2_FXF_TRUNC, 0644);
if (file == NULL) {
return calogFail(result, calogErrArgE, "sftpPut: could not open the remote file");
}
total = 0;
while (total < args[2].as.s.length) {
ssize_t n;
n = libssh2_sftp_write(file, args[2].as.s.bytes + total, (size_t)(args[2].as.s.length - total));
if (n < 0) {
libssh2_sftp_close(file);
return calogFail(result, calogErrArgE, "sftpPut: write failed");
}
total += (int64_t)n;
}
libssh2_sftp_close(file);
return calogOkE;
}
static int32_t sftpRemove(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
return calogFail(result, calogErrArgE, "sftpRemove expects (handle, path)");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
status = sshEnsureSftp(conn, result);
if (status != calogOkE) {
return status;
}
if (libssh2_sftp_unlink(conn->sftp, args[1].as.s.bytes) != 0) {
return calogFail(result, calogErrArgE, "sftpRemove: could not remove the path");
}
return calogOkE;
}
static int32_t sftpStat(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
LIBSSH2_SFTP_ATTRIBUTES attrs;
CalogAggT *map;
int64_t size;
bool isDir;
int rc;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
return calogFail(result, calogErrArgE, "sftpStat expects (handle, path)");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
status = sshEnsureSftp(conn, result);
if (status != calogOkE) {
return status;
}
rc = libssh2_sftp_stat(conn->sftp, args[1].as.s.bytes, &attrs);
if (rc < 0) {
// A missing path (or any stat failure) reports as nil rather than an error.
return calogOkE;
}
status = calogAggCreate(&map, calogMapE);
if (status != calogOkE) {
return calogFail(result, status, "sftpStat: out of memory");
}
size = (attrs.flags & LIBSSH2_SFTP_ATTR_SIZE) ? (int64_t)attrs.filesize : 0;
isDir = (attrs.flags & LIBSSH2_SFTP_ATTR_PERMISSIONS) && LIBSSH2_SFTP_S_ISDIR(attrs.permissions);
status = calogMapSetInt(map, "size", size);
if (status == calogOkE) {
status = calogMapSetBool(map, "isDir", isDir);
}
if (status != calogOkE) {
calogAggFree(map);
return calogFail(result, status, "sftpStat: failed to build the result");
}
calogValueAgg(result, map);
return calogOkE;
}
static int32_t sshAuthKey(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
const char *publicKey;
const char *passphrase;
int rc;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount < 3 || argCount > 5 || args[0].type != calogIntE || args[1].type != calogStringE || args[2].type != calogStringE) {
return calogFail(result, calogErrArgE, "sshAuthKey expects (handle, user, privateKeyPath[, publicKeyPath, passphrase])");
}
publicKey = NULL;
passphrase = NULL;
if (argCount >= 4) {
if (args[3].type != calogStringE) {
return calogFail(result, calogErrArgE, "sshAuthKey: publicKeyPath must be a string");
}
if (args[3].as.s.length > 0) {
publicKey = args[3].as.s.bytes;
}
}
if (argCount == 5) {
if (args[4].type != calogStringE) {
return calogFail(result, calogErrArgE, "sshAuthKey: passphrase must be a string");
}
if (args[4].as.s.length > 0) {
passphrase = args[4].as.s.bytes;
}
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
rc = libssh2_userauth_publickey_fromfile_ex(conn->session, args[1].as.s.bytes, (unsigned int)args[1].as.s.length,
publicKey, args[2].as.s.bytes, passphrase);
calogValueBool(result, rc == 0);
return calogOkE;
}
static int32_t sshAuthPassword(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
int rc;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 3 || args[0].type != calogIntE || args[1].type != calogStringE || args[2].type != calogStringE) {
return calogFail(result, calogErrArgE, "sshAuthPassword expects (handle, user, password)");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
rc = libssh2_userauth_password_ex(conn->session, args[1].as.s.bytes, (unsigned int)args[1].as.s.length,
args[2].as.s.bytes, (unsigned int)args[2].as.s.length, NULL);
calogValueBool(result, rc == 0);
return calogOkE;
}
static int32_t sshClose(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 1 || args[0].type != calogIntE) {
return calogFail(result, calogErrArgE, "sshClose expects (handle)");
}
// Atomically claim the handle -- a concurrent sshClose of the same handle gets NULL and must
// not double-destroy. ssh handles are shareable across contexts (no owner check); the caller
// is responsible for ensuring nobody is mid-operation on the connection when it is closed
// (see the threading notes in calogSsh.h).
conn = (SshConnT *)calogHandleRemove(lib->handles, args[0].as.i, SSH_TYPE_CONN);
if (conn == NULL) {
return calogFail(result, calogErrNotFoundE, "sshClose: invalid ssh handle");
}
sshConnDestroy(conn, "calog: closing");
return calogOkE;
}
static void sshCloser(uint32_t type, void *resource) {
(void)type;
sshConnDestroy((SshConnT *)resource, "calog: shutdown");
}
// Tear down a connection handle: shut down any cached SFTP subsystem, then the session and
// socket, then free the handle itself. Shared by sshClose and the handle table's closer so
// the release order lives in exactly one place.
static void sshConnDestroy(SshConnT *conn, const char *reason) {
if (conn->sftp != NULL) {
libssh2_sftp_shutdown(conn->sftp);
}
sshSessionDestroy(conn->session, conn->sock, reason);
free(conn);
}
static int32_t sshConnect(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
LIBSSH2_SESSION *session;
struct addrinfo hints;
struct addrinfo *res;
struct addrinfo *rp;
char portBuffer[8];
int64_t port;
int64_t handle;
CalogSocketT fd;
int rc;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount < 1 || argCount > 2 || args[0].type != calogStringE) {
return calogFail(result, calogErrArgE, "sshConnect expects (host[, port])");
}
port = 22;
if (argCount == 2) {
if (args[1].type != calogIntE) {
return calogFail(result, calogErrArgE, "sshConnect expects (host[, port])");
}
port = args[1].as.i;
}
if (port < 1 || port > SSH_PORT_MAX) {
return calogFail(result, calogErrArgE, "sshConnect: port out of range");
}
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
snprintf(portBuffer, sizeof(portBuffer), "%u", (unsigned int)port);
rc = getaddrinfo(args[0].as.s.bytes, portBuffer, &hints, &res);
if (rc != 0) {
return calogFail(result, calogErrArgE, gai_strerror(rc));
}
fd = CALOG_INVALID_SOCKET;
for (rp = res; rp != NULL; rp = rp->ai_next) {
fd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
if (fd == CALOG_INVALID_SOCKET) {
continue;
}
if (connect(fd, rp->ai_addr, (socklen_t)rp->ai_addrlen) == 0) {
break;
}
calogSockClose(fd);
fd = CALOG_INVALID_SOCKET;
}
freeaddrinfo(res);
if (fd == CALOG_INVALID_SOCKET) {
return calogFail(result, calogErrArgE, "sshConnect: could not connect");
}
session = libssh2_session_init();
if (session == NULL) {
calogSockClose(fd);
return calogFail(result, calogErrOomE, "sshConnect: could not create a session");
}
libssh2_session_set_blocking(session, 1);
if (libssh2_session_handshake(session, fd) != 0) {
libssh2_session_free(session);
calogSockClose(fd);
return calogFail(result, calogErrArgE, "sshConnect: SSH handshake failed");
}
conn = (SshConnT *)calloc(1, sizeof(*conn));
if (conn == NULL) {
sshSessionDestroy(session, fd, "out of memory");
return calogFail(result, calogErrOomE, "sshConnect: out of memory");
}
conn->sock = fd;
conn->session = session;
conn->sftp = NULL;
handle = calogHandleAdd(lib->handles, SSH_TYPE_CONN, conn);
if (handle == 0) {
sshConnDestroy(conn, "out of memory");
return calogFail(result, calogErrOomE, "sshConnect: out of memory");
}
calogValueInt(result, handle);
return calogOkE;
}
// Attempt one read of *bytes/*cap/*length via readFn, growing the buffer first if it is
// full. At the SSH_MAX_TRANSFER cap, instead of failing outright, probes with a single-byte
// read so a stream of exactly SSH_MAX_TRANSFER bytes still succeeds -- only a stream with
// MORE than the cap fails (calogErrRangeE). Shared by sftpGet (a single SFTP file stream)
// and sshExecDrain (two interleaved channel streams). Never frees *bytes on failure -- that
// stays the caller's job, since callers may be juggling more than one buffer.
static int32_t sshDrainStep(SshReadFnT readFn, void *handle, int streamId, char **bytes, int64_t *cap, int64_t *length, SshStepE *step) {
void *grown;
char probe[1];
ssize_t n;
int32_t status;
if (*length == *cap) {
if (*cap >= SSH_MAX_TRANSFER) {
n = readFn(handle, streamId, probe, sizeof(probe));
if (n == 0) {
*step = sshStepEofE;
return calogOkE;
}
if (n == LIBSSH2_ERROR_EAGAIN) {
*step = sshStepAgainE;
return calogOkE;
}
if (n < 0) {
return calogErrArgE;
}
return calogErrRangeE;
}
grown = *bytes;
status = calogGrow(&grown, cap, *length + 1, 1);
if (status != calogOkE) {
return status;
}
*bytes = (char *)grown;
if (*cap > SSH_MAX_TRANSFER) {
*cap = SSH_MAX_TRANSFER;
}
}
n = readFn(handle, streamId, *bytes + *length, (size_t)(*cap - *length));
if (n > 0) {
*length += n;
*step = sshStepDataE;
} else if (n == 0) {
*step = sshStepEofE;
} else if (n == LIBSSH2_ERROR_EAGAIN) {
*step = sshStepAgainE;
} else {
return calogErrArgE;
}
return calogOkE;
}
// Open the SFTP subsystem on the connection if not already open, caching it for reuse.
static int32_t sshEnsureSftp(SshConnT *conn, CalogValueT *result) {
LIBSSH2_SFTP *sftp;
if (conn->sftp != NULL) {
return calogOkE;
}
sftp = libssh2_sftp_init(conn->session);
if (sftp == NULL) {
return calogFail(result, calogErrArgE, "could not start an SFTP session");
}
conn->sftp = sftp;
return calogOkE;
}
static int32_t sshExec(CalogValueT *args, int32_t argCount, CalogValueT *result, void *userData) {
SshLibT *lib;
SshConnT *conn;
LIBSSH2_CHANNEL *channel;
CalogAggT *map;
char *outBytes;
char *errBytes;
int64_t outLength;
int64_t errLength;
int exitCode;
int32_t status;
lib = (SshLibT *)userData;
calogValueNil(result);
if (argCount != 2 || args[0].type != calogIntE || args[1].type != calogStringE) {
return calogFail(result, calogErrArgE, "sshExec expects (handle, command)");
}
if (args[1].as.s.length > SSH_MAX_TRANSFER) {
return calogFail(result, calogErrRangeE, "sshExec: command too long");
}
status = sshResolve(lib, args[0].as.i, result, &conn);
if (status != calogOkE) {
return status;
}
channel = libssh2_channel_open_session(conn->session);
if (channel == NULL) {
return calogFail(result, calogErrArgE, "sshExec: could not open a channel");
}
// Not libssh2_channel_exec (it derives the length with strlen, truncating a command with an
// embedded NUL); pass the binary-safe length explicitly.
if (libssh2_channel_process_startup(channel, "exec", 4u, args[1].as.s.bytes, (unsigned int)args[1].as.s.length) != 0) {
libssh2_channel_free(channel);
return calogFail(result, calogErrArgE, "sshExec: could not start the command");
}
status = sshExecDrain(conn->sock, conn->session, channel, &outBytes, &outLength, &errBytes, &errLength);
if (status != calogOkE) {
libssh2_channel_free(channel);
if (status == calogErrRangeE) {
return calogFail(result, status, "sshExec: command output exceeds the maximum transfer size");
}
if (status == calogErrOomE) {
return calogFail(result, status, "sshExec: out of memory");
}
return calogFail(result, status, "sshExec: channel read failed");
}
libssh2_channel_close(channel);
exitCode = libssh2_channel_get_exit_status(channel);
libssh2_channel_free(channel);
status = calogAggCreate(&map, calogMapE);
if (status != calogOkE) {
free(outBytes);
free(errBytes);
return calogFail(result, status, "sshExec: out of memory");
}
status = calogMapSetStr(map, "stdout", outBytes, outLength);
if (status == calogOkE) {
status = calogMapSetStr(map, "stderr", errBytes, errLength);
}
if (status == calogOkE) {
status = calogMapSetInt(map, "exitCode", (int64_t)exitCode);
}
free(outBytes);
free(errBytes);
if (status != calogOkE) {
calogAggFree(map);
return calogFail(result, status, "sshExec: failed to build the result");
}
calogValueAgg(result, map);
return calogOkE;
}
// Run the remote command's stdout and stderr to completion, interleaving reads of both
// streams in one loop (never draining one to EOF before starting the other). The two streams
// share one flow-control window, so fully draining stdout first while the remote blocks
// writing a flood of stderr can hang forever waiting on stdout, which never arrives because
// the remote process never gets to exit. Drops the session into non-blocking mode for the
// duration (restored before every return) so a stream with nothing ready right now cannot
// block progress on the other; sshWaitSocket parks the calling thread only when neither
// stream can make progress right now. On failure both output buffers are freed here.
static int32_t sshExecDrain(CalogSocketT sock, LIBSSH2_SESSION *session, LIBSSH2_CHANNEL *channel, char **outBytes, int64_t *outLength, char **errBytes, int64_t *errLength) {
char *outBuf;
char *errBuf;
int64_t outCap;
int64_t errCap;
int64_t outLen;
int64_t errLen;
bool outDone;
bool errDone;
bool progressed;
SshStepE step;
int32_t status;
outBuf = NULL;
errBuf = NULL;
outCap = 0;
errCap = 0;
outLen = 0;
errLen = 0;
outDone = false;
errDone = false;
libssh2_session_set_blocking(session, 0);
while (!outDone || !errDone) {
progressed = false;
if (!outDone) {
status = sshDrainStep(sshReadChannel, channel, 0, &outBuf, &outCap, &outLen, &step);
if (status != calogOkE) {
libssh2_session_set_blocking(session, 1);
free(outBuf);
free(errBuf);
return status;
}
if (step == sshStepEofE) {
outDone = true;
progressed = true;
} else if (step == sshStepDataE) {
progressed = true;
}
}
if (!errDone) {
status = sshDrainStep(sshReadChannel, channel, SSH_EXTENDED_DATA_STDERR, &errBuf, &errCap, &errLen, &step);
if (status != calogOkE) {
libssh2_session_set_blocking(session, 1);
free(outBuf);
free(errBuf);
return status;
}
if (step == sshStepEofE) {
errDone = true;
progressed = true;
} else if (step == sshStepDataE) {
progressed = true;
}
}
if (!progressed && (!outDone || !errDone)) {
sshWaitSocket(sock, session);
}
}
libssh2_session_set_blocking(session, 1);
*outBytes = outBuf;
*outLength = outLen;
*errBytes = errBuf;
*errLength = errLen;
return calogOkE;
}
static ssize_t sshReadChannel(void *handle, int streamId, char *buffer, size_t length) {
return libssh2_channel_read_ex((LIBSSH2_CHANNEL *)handle, streamId, buffer, length);
}
static ssize_t sshReadSftpFile(void *handle, int streamId, char *buffer, size_t length) {
(void)streamId;
return libssh2_sftp_read((LIBSSH2_SFTP_HANDLE *)handle, buffer, length);
}
// Look up a connection handle. Not-found resolves to calogErrNotFoundE. There is NO owner
// check: an ssh handle may be used by any context it is passed to (see the threading notes in
// calogSsh.h) -- the caller must serialize access, since a libssh2 session is not thread-safe.
static int32_t sshResolve(SshLibT *lib, int64_t handle, CalogValueT *result, SshConnT **out) {
SshConnT *conn;
conn = (SshConnT *)calogHandleGet(lib->handles, handle, SSH_TYPE_CONN);
if (conn == NULL) {
return calogFail(result, calogErrNotFoundE, "invalid ssh handle");
}
*out = conn;
return calogOkE;
}
// Tear down a raw session and its socket (disconnect, free the session, close the fd). Used
// by sshConnDestroy and by sshConnect's post-handshake failure paths, before a SshConnT
// exists to own the session.
static void sshSessionDestroy(LIBSSH2_SESSION *session, CalogSocketT sock, const char *reason) {
libssh2_session_disconnect(session, reason);
libssh2_session_free(session);
calogSockClose(sock);
}
// Block the calling thread until the socket is ready in whatever direction libssh2 is
// currently waiting on. Used only by sshExecDrain's non-blocking interleave; every other
// native in this file runs its session in blocking mode and never calls this.
static void sshWaitSocket(CalogSocketT sock, LIBSSH2_SESSION *session) {
fd_set fdSet;
fd_set *readSet;
fd_set *writeSet;
int direction;
FD_ZERO(&fdSet);
FD_SET(sock, &fdSet);
readSet = NULL;
writeSet = NULL;
direction = libssh2_session_block_directions(session);
if (direction & LIBSSH2_SESSION_BLOCK_INBOUND) {
readSet = &fdSet;
}
if (direction & LIBSSH2_SESSION_BLOCK_OUTBOUND) {
writeSet = &fdSet;
}
select(sock + 1, readSet, writeSet, NULL, NULL);
}
Reference in a new issue View git blame Copy permalink