diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fc0e19c --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.acton.lock +.build +out +zig-cache +zig-out diff --git a/Build.act b/Build.act new file mode 100644 index 0000000..c202459 --- /dev/null +++ b/Build.act @@ -0,0 +1,9 @@ +name = "acton_ssh" +fingerprint = 0xb650ac2857701cb1 +zig_dependencies = { + "libssh": ( + url="https://github.com/precla/libssh/archive/refs/heads/zig-build.tar.gz", + hash="libssh-0.11.0-kUQJaoJiKgDjZpTAyxxo6GamiMzj9LPxlEr2FacwOxIR", + artifacts=[ "ssh" ] + ) +} diff --git a/build.act.json b/build.act.json deleted file mode 100644 index ec3c9b7..0000000 --- a/build.act.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "dependencies": {}, - "zig_dependencies": { - "libssh": { - "url": "https://github.com/actonlang/libssh/archive/refs/heads/zig-build.tar.gz", - "hash": "122076b4676ec7d777e5efc24ae4b7f1ab4fa3df407ab9649fdd55f9fc594ca053ec", - "artifacts": [ - "ssh" - ] - } - } -} diff --git a/src/netconf.h b/src/netconf.h new file mode 100644 index 0000000..e373b5a --- /dev/null +++ b/src/netconf.h @@ -0,0 +1,35 @@ +// NETCONF message +#define NETCONF_HELLO_MSG \ + "\n" \ + "\n" \ + " \n" \ + " urn:ietf:params:netconf:base:1.0\n" \ + " \n" \ + "]]>]]>" + +#define NETCONF_GET_CONFIG_MSG \ + "\n" \ + "\n" \ + " \n" \ + " \n" \ + " \n" \ + " \n" \ + " \n" \ + "]]>]]>" + +#define NETCONF_GET_STATE \ + "\n" \ + "\n" \ + " " \ + " " \ + " " \ + " " \ + " " \ + "]]>]]>" + +// NETCONF message +#define NETCONF_CLOSE_SESSION_MSG \ + "\n" \ + "\n" \ + " \n" \ + "]]>]]>" diff --git a/src/ssh.act b/src/ssh.act old mode 100644 new mode 100755 index a8b0268..c9daed4 --- a/src/ssh.act +++ b/src/ssh.act @@ -1,69 +1,125 @@ import net -import testing + def version() -> str: - """Get the libssh version""" + """Get the acton-ssh version""" NotImplemented -def _test_version(): - testing.assertEqual("0.11.0", version()) actor Client(cap: net.TCPConnectCap, host: str, username: str, on_connect: action(Client) -> None, - on_close: action(Client, str) -> None, - key: ?str=None, + on_receive: action(Client, bytes) -> None, + on_error: action(Client, str) -> None, + on_remote_close: ?action(Client) -> None, password: ?str=None, - port: u16=22, - ): - """SSH Client""" + key: ?str=None, + port: int=22, + subsystem: ?str=None): + """SSH Client + + Opens an SSH connection. If subsystem is None, opens a shell channel. + If subsystem is set (e.g. "netconf"), opens a subsystem channel instead. + """ + + var _client: u64 = 0 - # haha, this is really a pointer :P - var _ssh_session: u64 = 0 + proc def _pin_affinity() -> None: + NotImplemented + _pin_affinity() + + action def write(data: bytes) -> None: + """Write data to the SSH channel""" + NotImplemented + + action def close(on_close: action(Client) -> None) -> None: + """Close the connection""" + NotImplemented + + def reconnect(): + close(_connect) + + def _connect(c): + NotImplemented proc def _init() -> None: """Initialize the SSH client""" NotImplemented _init() - print("SSH Client connected") -# action def close(on_close: action(TLSConnection) -> None) -> None: -# """Close the connection""" -# NotImplemented -# -# def reconnect(): -# close(_connect) -# -# def _connect(c): -# NotImplemented -# -# TODO: implement support for channels -# AFAIK, all things over ssh are done via channels, so need some channel -# primitive, maybe an actor per channel that then multiplexes into the Client -# session? Prolly need some higher level wrappers for common things like -# starting a shell or running a single command. SFTP / SCP would be nice too, -# but for sometime in the future. Custom subsystems need to be supported too. +actor main(env): + var _buf: bytes = b"" + NETCONF_HELLO = b""" + + + urn:ietf:params:netconf:base:1.0 + +]]>]]>""" + NETCONF_GET_CONFIG = b""" + + + + + + +]]>]]>""" + NETCONF_CLOSE = b""" + + +]]>]]>""" + + var _state: int = 0 # 0=wait hello, 1=wait get-config reply, 2=wait close reply -actor main(env): def on_connect(client: Client): - print("Connected") + print("Connected! Sending NETCONF hello...") + client.write(NETCONF_HELLO) + + def on_receive(client: Client, data: bytes): + _buf += data + s = _buf.decode() + if "]]>]]>" in s: + print("=== NETCONF message ===") + print(s) + print("=======================") + _buf = b"" + + st = _state + if st == 0: + print("Got server hello, sending get-config...") + _state = 1 + client.write(NETCONF_GET_CONFIG) + elif st == 1: + print("Got config, sending close-session...") + _state = 2 + client.write(NETCONF_CLOSE) + elif st == 2: + print("Got close reply, done!") + def _on_close(c: Client): + env.exit(0) + client.close(_on_close) + + def on_error(client: Client, error: str): + print("Error:", error) + env.exit(1) - def on_close(client: Client, error: str): - print("Error", error) + def on_remote_close(client: Client): + print("Remote closed connection") + env.exit(0) - print(version()) c = Client( net.TCPConnectCap(net.TCPCap(net.NetCap(env.cap))), "localhost", "foo", on_connect, - on_close, + on_receive, + on_error, + on_remote_close, password="bar", - port=2223, + port=830, + subsystem="netconf", ) - env.exit(0) diff --git a/src/ssh.ext.c b/src/ssh.ext.c index 2112598..54394d9 100644 --- a/src/ssh.ext.c +++ b/src/ssh.ext.c @@ -1,120 +1,417 @@ +#include #include -#include -// TODO: figure out how to include rts/log so we get access to log_error etc +#include +#include +#include +// acton includes +#include -void noop_free(void *ptr) { +/* Declared in rts/io.h but not always in the include path */ +uv_loop_t *get_uv_loop(); + +enum client_state { + S_AUTH, + S_AUTH_KBDINT, + S_CHANNEL_OPEN, + S_REQUEST_TYPE, + S_READY, + S_ERROR +}; + +typedef struct { + ssh_session session; + ssh_channel channel; + uv_poll_t *poll; + int fd; + enum client_state state; + + /* pending write buffer */ + char *write_buf; + size_t write_len; + size_t write_sent; + + /* actor reference for callbacks */ + sshQ_Client actor; +} client_t; + +/* Forward declarations */ +static void poll_cb(uv_poll_t *handle, int status, int events); + +/* Helper: set up SSH session, connect, and start polling on the RTS loop */ +static int client_setup(client_t *c, sshQ_Client self) { + int err = 0; + + c->actor = self; + c->channel = NULL; + c->write_buf = NULL; + c->write_len = 0; + c->write_sent = 0; + + c->session = ssh_new(); + if (!c->session) + return -1; + + const char *host = (const char *)fromB_str(self->host); + int port = (int)fromB_int(self->port); + const char *user = (const char *)fromB_str(self->username); + + ssh_options_set(c->session, SSH_OPTIONS_HOST, host); + ssh_options_set(c->session, SSH_OPTIONS_PORT, &port); + ssh_options_set(c->session, SSH_OPTIONS_USER, user); + ssh_session_set_disconnect_message(c->session, "Disconnecting SSH, powered by Acton"); + + /* Blocking TCP connect (fast) */ + err = ssh_connect(c->session); + if (err != SSH_OK) { + ssh_free(c->session); + c->session = NULL; + return -1; + } + + c->state = S_AUTH; + ssh_set_blocking(c->session, 0); + + c->fd = ssh_get_fd(c->session); + if (c->fd < 0) { + ssh_disconnect(c->session); + ssh_free(c->session); + c->session = NULL; + return -1; + } + + /* Use the RTS event loop instead of a private loop */ + c->poll = (uv_poll_t *)acton_malloc(sizeof(uv_poll_t)); + err = uv_poll_init(get_uv_loop(), c->poll, c->fd); + if (err < 0) { + ssh_disconnect(c->session); + ssh_free(c->session); + c->session = NULL; + return -1; + } + + c->poll->data = c; + err = uv_poll_start(c->poll, UV_READABLE | UV_WRITABLE, poll_cb); + if (err < 0) { + uv_close((uv_handle_t *)c->poll, NULL); + ssh_disconnect(c->session); + ssh_free(c->session); + c->session = NULL; + return -1; + } + + return 0; +} + +/* Called whenever the polled fd has activity — drives connection state machine + * and delivers data to Acton via callbacks */ +static void poll_cb(uv_poll_t *handle, int status, int events) { + if (!handle || !handle->data) + return; + + client_t *c = (client_t *)handle->data; + sshQ_Client self = c->actor; + int rc; + + if (status < 0) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str((char *)uv_strerror(status))); + uv_poll_stop(c->poll); + return; + } + + switch (c->state) { + case S_AUTH: { + if ((void *)self->password == (void *)B_None || self->password == NULL) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str("No authentication method available")); + uv_poll_stop(c->poll); + return; + } + rc = ssh_userauth_password(c->session, NULL, (const char *)fromB_str(self->password)); + if (rc == SSH_AUTH_SUCCESS) { + c->state = S_CHANNEL_OPEN; + } else if (rc == SSH_AUTH_AGAIN) { + return; + } else { + /* Password auth failed/denied — try keyboard-interactive */ + c->state = S_AUTH_KBDINT; + } + break; + } + case S_AUTH_KBDINT: { + rc = ssh_userauth_kbdint(c->session, NULL, NULL); + if (rc == SSH_AUTH_INFO) { + int nprompts = ssh_userauth_kbdint_getnprompts(c->session); + for (int i = 0; i < nprompts; i++) { + ssh_userauth_kbdint_setanswer(c->session, i, + (const char *)fromB_str(self->password)); + } + /* Need another round — stay in S_AUTH_KBDINT */ + return; + } else if (rc == SSH_AUTH_SUCCESS) { + c->state = S_CHANNEL_OPEN; + } else if (rc == SSH_AUTH_AGAIN) { + return; + } else { + char errmsg[512]; + snprintf(errmsg, sizeof(errmsg), "Authentication failed: %s", ssh_get_error(c->session)); + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str(errmsg)); + uv_poll_stop(c->poll); + return; + } + break; + } + case S_CHANNEL_OPEN: + if (!c->channel) { + c->channel = ssh_channel_new(c->session); + if (!c->channel) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str("Failed to create SSH channel")); + uv_poll_stop(c->poll); + return; + } + } + rc = ssh_channel_open_session(c->channel); + if (rc == SSH_OK) { + c->state = S_REQUEST_TYPE; + } else if (rc == SSH_AGAIN) { + return; + } else { + char errmsg[512]; + snprintf(errmsg, sizeof(errmsg), "Failed to open channel: %s", ssh_get_error(c->session)); + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str(errmsg)); + uv_poll_stop(c->poll); + return; + } + break; + case S_REQUEST_TYPE: + if ((void *)self->subsystem == (void *)B_None || self->subsystem == NULL) { + /* No subsystem — request a shell */ + rc = ssh_channel_request_shell(c->channel); + } else { + rc = ssh_channel_request_subsystem(c->channel, (const char *)fromB_str(self->subsystem)); + } + if (rc == SSH_OK) { + c->state = S_READY; + /* Connection fully established — notify Acton */ + $action f = ($action)self->on_connect; + f->$class->__asyn__(f, self); + } else if (rc == SSH_AGAIN) { + return; + } else { + char errmsg[512]; + snprintf(errmsg, sizeof(errmsg), "Failed to request shell/subsystem: %s", ssh_get_error(c->session)); + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str(errmsg)); + uv_poll_stop(c->poll); + return; + } + break; + + case S_READY: + /* Flush any pending write data */ + if (c->write_buf && c->write_sent < c->write_len && (events & UV_WRITABLE)) { + int wrote = ssh_channel_write(c->channel, + c->write_buf + c->write_sent, + (uint32_t)(c->write_len - c->write_sent)); + if (wrote > 0) { + c->write_sent += (size_t)wrote; + if (c->write_sent >= c->write_len) { + free(c->write_buf); + c->write_buf = NULL; + c->write_len = 0; + c->write_sent = 0; + } + } else if (wrote == SSH_ERROR) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str("SSH channel write failed")); + uv_poll_stop(c->poll); + return; + } + /* SSH_AGAIN / 0: try again on next poll */ + } + + /* Read all available data from the channel */ + { + char buf[8192]; + int n; + while ((n = ssh_channel_read_nonblocking(c->channel, buf, sizeof(buf), 0)) > 0) { + $action2 f = ($action2)self->on_receive; + f->$class->__asyn__(f, self, to$bytesD_len(buf, n)); + } + if (ssh_channel_is_eof(c->channel)) { + uv_poll_stop(c->poll); + if (self->on_remote_close) { + $action f = ($action)self->on_remote_close; + f->$class->__asyn__(f, self); + } + } else if (n < 0) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str((char *)ssh_get_error(c->session))); + uv_poll_stop(c->poll); + } + } + break; + + case S_ERROR: + uv_poll_stop(c->poll); + break; + } } +/* ---- Module-level functions ---- */ + void sshQ___ext_init__() { - // TODO: can we avoid custom malloc in libssh? like let libssh use stock - // malloc and instead we would explicitly call free() from a finalizer() - // All things related to buffers for receiving data and similarly would have - // to be allocated on the GC-heap though since that data is passed outside - // of the SSH actor - libssh_replace_allocator( - acton_gc_malloc, - acton_gc_realloc, - acton_gc_calloc, - noop_free, - acton_gc_strdup, - acton_gc_strndup); int r = ssh_init(); - printf("SSH extension initialized %d\n", r); + if (r != SSH_OK) + printf("SSH init failed (%d)\n", r); +} + +B_str sshQ_version() { + return to$str("0.2.0"); } -B_str sshQ_version () { - if (LIBSSH_VERSION_MINOR != 11) - return to$str("invalid"); - return to$str("0.11.0"); +/* ---- Client actor methods ---- */ + +$R sshQ_ClientD__pin_affinityG_local(sshQ_Client self, $Cont c$cont) { + pin_actor_affinity(); + return $R_CONT(c$cont, B_None); } -// TODO: crap function for test, to be replaced with something -int show_remote_processes(ssh_session session) -{ - ssh_channel channel; - int rc; - char buffer[256]; - int nbytes; - - channel = ssh_channel_new(session); - if (channel == NULL) - return SSH_ERROR; - - rc = ssh_channel_open_session(channel); - if (rc != SSH_OK) - { - ssh_channel_free(channel); - return rc; - } - - rc = ssh_channel_request_exec(channel, "ps aux"); - if (rc != SSH_OK) - { - ssh_channel_close(channel); - ssh_channel_free(channel); - return rc; - } - - nbytes = ssh_channel_read(channel, buffer, sizeof(buffer), 0); - while (nbytes > 0) - { - if (write(1, buffer, nbytes) != (unsigned int) nbytes) - { - ssh_channel_close(channel); - ssh_channel_free(channel); - return SSH_ERROR; - } - nbytes = ssh_channel_read(channel, buffer, sizeof(buffer), 0); - } - - if (nbytes < 0) - { - ssh_channel_close(channel); - ssh_channel_free(channel); - return SSH_ERROR; - } - - ssh_channel_send_eof(channel); - ssh_channel_close(channel); - ssh_channel_free(channel); - - return SSH_OK; +$R sshQ_ClientD__initG_local(sshQ_Client self, $Cont c$cont) { + pin_actor_affinity(); + + client_t *client = (client_t *)acton_malloc(sizeof(client_t)); + memset(client, 0, sizeof(client_t)); + + if (client_setup(client, self) != 0) { + char errmsg[512]; + if (client->session) { + snprintf(errmsg, sizeof(errmsg), "SSH connect failed: %s", ssh_get_error(client->session)); + } else { + snprintf(errmsg, sizeof(errmsg), "SSH initialization failed"); + } + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str(errmsg)); + return $R_CONT(c$cont, B_None); + } + + self->_client = toB_u64((unsigned long)client); + return $R_CONT(c$cont, B_None); } -$R sshQ_ClientD__initG_local (sshQ_Client self, $Cont c$cont) { - ssh_session session = ssh_new(); - if (session == NULL) { - //log_error("Failed to create SSH session"); +$R sshQ_ClientD_writeG_local(sshQ_Client self, $Cont c$cont, B_bytes data) { + client_t *client = (client_t *)fromB_u64(self->_client); + if (!client || !client->channel || client->state != S_READY) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str("Cannot write: not connected")); + return $R_CONT(c$cont, B_None); + } + + /* If there is already pending data, append to the write buffer */ + if (client->write_buf) { + size_t remaining = client->write_len - client->write_sent; + size_t new_len = remaining + data->nbytes; + char *new_buf = malloc(new_len); + if (!new_buf) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str("Write buffer allocation failed")); + return $R_CONT(c$cont, B_None); + } + memcpy(new_buf, client->write_buf + client->write_sent, remaining); + memcpy(new_buf + remaining, data->str, data->nbytes); + free(client->write_buf); + client->write_buf = new_buf; + client->write_len = new_len; + client->write_sent = 0; return $R_CONT(c$cont, B_None); } - printf("session: %p\n", session); - self->_ssh_session = toB_u64((unsigned long)session); - printf("init self->session: %p\n", self->_ssh_session); - - ssh_options_set(session, SSH_OPTIONS_HOST, fromB_str(self->host)); - ssh_options_set(session, SSH_OPTIONS_PORT, &self->port->val); - ssh_options_set(session, SSH_OPTIONS_USER, fromB_str(self->username)); - - ssh_set_blocking(session, 1); - printf("Connecting to \n"); - int rc = ssh_connect(session); - if (rc != SSH_OK) { - //log_error("Error connecting to SSH server: %s", ssh_get_error(session)); - $action2 f = ($action2) self->on_close; - f->$class->__asyn__(f, self, to$str(ssh_get_error(session))); + + /* Try to write directly */ + int wrote = ssh_channel_write(client->channel, (const char *)data->str, (uint32_t)data->nbytes); + if (wrote == (int)data->nbytes) { + return $R_CONT(c$cont, B_None); + } + + /* Partial or failed write — buffer the remainder for poll_cb to flush */ + size_t sent = (wrote > 0) ? (size_t)wrote : 0; + size_t remaining = data->nbytes - sent; + client->write_buf = malloc(remaining); + if (!client->write_buf) { + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str("Write buffer allocation failed")); + return $R_CONT(c$cont, B_None); + } + memcpy(client->write_buf, (const char *)data->str + sent, remaining); + client->write_len = remaining; + client->write_sent = 0; + + return $R_CONT(c$cont, B_None); +} + +$R sshQ_ClientD_closeG_local(sshQ_Client self, $Cont c$cont, $action on_close) { + client_t *client = (client_t *)fromB_u64(self->_client); + if (!client) { + on_close->$class->__asyn__(on_close, self); return $R_CONT(c$cont, B_None); } - rc = ssh_userauth_password(session, NULL, fromB_str(self->password)); - if (rc == SSH_OK) { - printf("Connected\n"); - show_remote_processes(session); - } else { - printf("Error: %s\n", ssh_get_error(session)); + /* Stop polling */ + if (client->poll) { + uv_poll_stop(client->poll); + uv_close((uv_handle_t *)client->poll, NULL); + client->poll = NULL; + } + + /* Close SSH channel */ + if (client->channel) { + ssh_channel_send_eof(client->channel); + ssh_channel_close(client->channel); + ssh_channel_free(client->channel); + client->channel = NULL; + } + + /* Disconnect SSH session */ + if (client->session) { + ssh_disconnect(client->session); + ssh_free(client->session); + client->session = NULL; + } + + /* Free write buffer */ + if (client->write_buf) { + free(client->write_buf); + client->write_buf = NULL; + } + + self->_client = toB_u64(0); + + /* Invoke on_close callback */ + on_close->$class->__asyn__(on_close, self); + + return $R_CONT(c$cont, B_None); +} + +$R sshQ_ClientD__connectG_local(sshQ_Client self, $Cont c$cont, sshQ_Client c) { + pin_actor_affinity(); + + client_t *client = (client_t *)acton_malloc(sizeof(client_t)); + memset(client, 0, sizeof(client_t)); + + if (client_setup(client, self) != 0) { + char errmsg[512]; + if (client->session) { + snprintf(errmsg, sizeof(errmsg), "SSH reconnect failed: %s", ssh_get_error(client->session)); + } else { + snprintf(errmsg, sizeof(errmsg), "SSH reconnect initialization failed"); + } + $action2 f = ($action2)self->on_error; + f->$class->__asyn__(f, self, to$str(errmsg)); + return $R_CONT(c$cont, B_None); } -// self->_connected = true; -// $action f = ($action) self->on_connect; -// f->$class->__asyn__(f, self); + self->_client = toB_u64((unsigned long)client); return $R_CONT(c$cont, B_None); }