#include "client.h"
#include <cerrno>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <poll.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <unistd.h>
extern "C" {
#include <compact.h>
#include <rpc.h>
#include "swarmd_hrpc.h"
#include "swarmd_schema.h"
}
#include "../src/uds_server.h" // UDS_MAX_FRAME — one definition of the cap
namespace swarmd {
// ---- framing -------------------------------------------------------------
static uint32_t le32(const uint8_t *p) {
return (uint32_t)p[0] | ((uint32_t)p[1] << 8) | ((uint32_t)p[2] << 16) | ((uint32_t)p[3] << 24);
}
static bool write_all(int fd, const uint8_t *p, size_t n) {
size_t off = 0;
while (off < n) {
const ssize_t w = ::write(fd, p + off, n - off);
if (w > 0) { off += (size_t)w; continue; }
if (w < 0 && (errno == EINTR)) continue;
if (w < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
// The socket is blocking, so this is not expected — but a short wait
// is still the right answer to a full kernel buffer, and dropping
// half a frame is not.
struct pollfd pfd = { fd, POLLOUT, 0 };
if (::poll(&pfd, 1, 1000) > 0) continue;
}
return false;
}
return true;
}
Client::~Client() { close(); }
void Client::close() {
if (_fd >= 0) ::close(_fd);
_fd = -1;
_in.clear();
_reply.clear();
}
void Client::_fail(const std::string &why) {
if (_fd < 0) return; // already reported
_err = why;
close();
if (on.on_disconnect) on.on_disconnect(why);
}
bool Client::_send_frame(uint8_t *buf, size_t len) {
if (_fd < 0 || buf == nullptr) { free(buf); return false; }
uint8_t hdr[4] = { (uint8_t)(len & 0xff), (uint8_t)((len >> 8) & 0xff),
(uint8_t)((len >> 16) & 0xff), (uint8_t)((len >> 24) & 0xff) };
const bool ok = write_all(_fd, hdr, 4) && write_all(_fd, buf, len);
free(buf); // ⛔ the caller frees what *_encode_* allocated
if (!ok) _fail("the daemon closed while we were writing to it");
return ok;
}
// ---- reading -------------------------------------------------------------
bool Client::_read_some(int timeout_ms) {
if (_fd < 0) return false;
struct pollfd pfd = { _fd, POLLIN, 0 };
const int r = ::poll(&pfd, 1, timeout_ms);
if (r < 0) {
if (errno == EINTR) return true;
_fail("poll failed on the daemon socket");
return false;
}
if (r == 0) return true; // nothing yet, not an error
uint8_t tmp[65536];
const ssize_t n = ::read(_fd, tmp, sizeof(tmp));
if (n == 0) { _fail("the daemon closed the connection"); return false; }
if (n < 0) {
if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) return true;
_fail("read failed on the daemon socket");
return false;
}
_bytes_in += (uint64_t)n;
_in.insert(_in.end(), tmp, tmp + n);
return true;
}
void Client::_dispatch_event(const void *vmsg) {
const rpc_message_t *msg = (const rpc_message_t *)vmsg;
swarmd_hrpc_handlers_t h;
memset(&h, 0, sizeof(h));
h.ctx = this;
h.on_on_message = [](void *ctx, const swarmd_message_t *m) {
Client *c = (Client *)ctx;
if (c->on.on_message)
c->on.on_message(m->channel, m->slot, m->origin_kind, m->origin_id,
m->bytes, m->bytes_len);
};
h.on_on_peer_up = [](void *ctx, const swarmd_peer_up_t *m) {
Client *c = (Client *)ctx;
if (c->on.on_peer_up) c->on.on_peer_up(m->peer, m->pubkey);
};
h.on_on_peer_down = [](void *ctx, const swarmd_peer_down_t *m) {
Client *c = (Client *)ctx;
if (c->on.on_peer_down) c->on.on_peer_down(m->peer);
};
h.on_on_drain = [](void *ctx, const swarmd_channel_ref_t *m) {
Client *c = (Client *)ctx;
if (c->on.on_drain) c->on.on_drain(m->channel);
};
h.on_on_content_progress = [](void *ctx, const swarmd_content_progress_t *m) {
Client *c = (Client *)ctx;
if (c->on.on_content_progress) c->on.on_content_progress(m->hash, m->received, m->total);
};
h.on_on_content_done = [](void *ctx, const swarmd_content_done_t *m) {
Client *c = (Client *)ctx;
if (c->on.on_content_done)
c->on.on_content_done(m->hash, std::string((const char *)m->path.data, m->path.len));
};
h.on_on_content_failed = [](void *ctx, const swarmd_content_failed_t *m) {
Client *c = (Client *)ctx;
if (c->on.on_content_failed)
c->on.on_content_failed(m->hash, std::string((const char *)m->reason.data, m->reason.len));
};
// An event carries no reply, so anything the dispatcher hands back is
// discarded rather than sent — but it must still be freed.
uint8_t *rep = nullptr;
size_t rl = 0;
swarmd_hrpc_dispatch(&h, msg, &rep, &rl);
if (rep) free(rep);
}
bool Client::_drain_frames(uint64_t await_id, const uint8_t **body, size_t *body_len) {
for (;;) {
if (_in.size() < 4) return false;
const uint32_t n = le32(_in.data());
if (n > UDS_MAX_FRAME) {
// A length this large is a framing bug, not a frame — and the stream
// is desynchronised, so there is nothing to resume from.
_fail("the daemon sent a frame length past the cap; the stream is out of sync");
return false;
}
if (_in.size() < 4 + (size_t)n) return false;
// Copy out before consuming: a decoded view points INTO this buffer, and
// the buffer moves as soon as the next frame is taken off it.
std::vector<uint8_t> frame(_in.begin() + 4, _in.begin() + 4 + n);
_in.erase(_in.begin(), _in.begin() + 4 + n);
compact_state_t st = { 0, frame.size(), frame.data() };
rpc_message_t msg;
if (rpc_decode_message(&st, &msg) < 0) {
_fail("the daemon sent a message this client could not decode");
return false;
}
if (msg.type == rpc_response) {
if (msg.id != await_id) continue; // a reply nobody is waiting for
if (msg.error) {
_err = "the daemon answered with an error";
// Not a disconnect: the call failed, the session is fine.
_reply.clear();
if (body) *body = nullptr;
if (body_len) *body_len = 0;
return true;
}
_reply.assign(msg.data, msg.data + msg.len);
if (body) *body = _reply.empty() ? nullptr : _reply.data();
if (body_len) *body_len = _reply.size();
return true;
}
// ⛔ EVENTS ARE DELIVERED EVEN WHILE A CALL IS WAITING. A peer arriving,
// or a fetch finishing, does not queue politely behind the answer to
// channel_open.
if (msg.type == rpc_request) _dispatch_event(&msg);
}
}
bool Client::_await(uint64_t id, int timeout_ms, const uint8_t **body, size_t *body_len) {
if (body) *body = nullptr;
if (body_len) *body_len = 0;
// Anything already buffered may hold the answer.
if (_drain_frames(id, body, body_len)) return true;
if (_fd < 0) return false;
int left = timeout_ms;
while (left > 0) {
const int slice = left > 200 ? 200 : left;
if (!_read_some(slice)) return false;
if (_drain_frames(id, body, body_len)) return true;
if (_fd < 0) return false;
left -= slice;
}
_err = "the daemon did not answer in time";
return false;
}
bool Client::poll(int timeout_ms) {
if (_fd < 0) return false;
if (!_read_some(timeout_ms)) return false;
// id 0 is never a request id, so nothing matches and every frame is an event.
_drain_frames(0, nullptr, nullptr);
return _fd >= 0;
}
// ---- calls ---------------------------------------------------------------
bool Client::connect(const std::string &socket_path, const std::string &name,
const uint8_t app_identity[32], int timeout_ms) {
close();
_err.clear();
_fd = ::socket(AF_UNIX, SOCK_STREAM, 0);
if (_fd < 0) { _err = "could not create a unix socket"; return false; }
struct sockaddr_un a;
memset(&a, 0, sizeof(a));
a.sun_family = AF_UNIX;
strncpy(a.sun_path, socket_path.c_str(), sizeof(a.sun_path) - 1);
if (::connect(_fd, (struct sockaddr *)&a, sizeof(a)) != 0) {
::close(_fd);
_fd = -1;
_err = "no daemon at " + socket_path;
return false;
}
swarmd_hello_request_t req;
memset(&req, 0, sizeof(req));
req.name.data = (utf8_t *)name.c_str();
req.name.len = name.size();
if (app_identity) memcpy(req.app_identity, app_identity, 32);
req.version = 1;
const uint64_t id = _next_id++;
uint8_t *buf = nullptr;
size_t len = 0;
if (swarmd_encode_hello(id, &req, &buf, &len) < 0) { _fail("could not encode HELLO"); return false; }
if (!_send_frame(buf, len)) return false;
const uint8_t *body = nullptr;
size_t body_len = 0;
if (!_await(id, timeout_ms, &body, &body_len)) { close(); return false; }
swarmd_hello_response_t res;
memset(&res, 0, sizeof(res));
if (body) {
compact_state_t in = { 0, body_len, (uint8_t *)body };
if (swarmd_hello_response_decode(&in, &res) < 0) {
_fail("the daemon's HELLO reply did not decode");
return false;
}
}
_client_id = res.client_id;
memcpy(_daemon_pubkey, res.daemon_pubkey, 32);
return true;
}
// One shape for every call that answers with `ok`.
#define SWARMD_OK_CALL(ENCODE, REQPTR) \
do { \
if (_fd < 0) { _err = "not connected"; return false; } \
const uint64_t id = _next_id++; \
uint8_t *buf = nullptr; \
size_t len = 0; \
if (ENCODE(id, REQPTR, &buf, &len) < 0) { _fail("encode failed"); return false; } \
if (!_send_frame(buf, len)) return false; \
const uint8_t *body = nullptr; \
size_t body_len = 0; \
if (!_await(id, timeout_ms, &body, &body_len)) return false; \
if (!body) return false; \
swarmd_ok_t ok; \
memset(&ok, 0, sizeof(ok)); \
compact_state_t in = { 0, body_len, (uint8_t *)body }; \
if (swarmd_ok_decode(&in, &ok) < 0) { _fail("ok reply did not decode"); return false; } \
if (!ok.ok) _err = "the daemon refused the request"; \
return ok.ok; \
} while (0)
bool Client::join(const uint8_t topic[32], bool server, bool client, int timeout_ms) {
swarmd_swarm_join_t req;
memset(&req, 0, sizeof(req));
memcpy(req.topic, topic, 32);
req.server = server;
req.client = client;
SWARMD_OK_CALL(swarmd_encode_swarm_join, &req);
}
bool Client::leave(const uint8_t topic[32], int timeout_ms) {
swarmd_topic_ref_t req;
memset(&req, 0, sizeof(req));
memcpy(req.topic, topic, 32);
SWARMD_OK_CALL(swarmd_encode_swarm_leave, &req);
}
bool Client::close_channel(uint64_t channel, int timeout_ms) {
swarmd_channel_ref_t req;
memset(&req, 0, sizeof(req));
req.channel = channel;
SWARMD_OK_CALL(swarmd_encode_channel_close, &req);
}
bool Client::content_serve(const uint8_t hash[32], const std::string &path, int timeout_ms) {
swarmd_content_serve_t req;
memset(&req, 0, sizeof(req));
memcpy(req.hash, hash, 32);
req.path.data = (utf8_t *)path.c_str();
req.path.len = path.size();
SWARMD_OK_CALL(swarmd_encode_content_serve, &req);
}
bool Client::content_fetch(const uint8_t hash[32], const std::string &dest_path, int timeout_ms) {
swarmd_content_fetch_t req;
memset(&req, 0, sizeof(req));
memcpy(req.hash, hash, 32);
req.dest_path.data = (utf8_t *)dest_path.c_str();
req.dest_path.len = dest_path.size();
SWARMD_OK_CALL(swarmd_encode_content_fetch, &req);
}
bool Client::content_cancel(const uint8_t hash[32], bool discard, int timeout_ms) {
swarmd_content_cancel_t req;
memset(&req, 0, sizeof(req));
memcpy(req.hash, hash, 32);
req.discard = discard;
SWARMD_OK_CALL(swarmd_encode_content_cancel, &req);
}
uint64_t Client::open_channel(const std::string &protocol, const std::vector<uint8_t> &id_bytes,
uint32_t slots, int timeout_ms) {
if (_fd < 0) { _err = "not connected"; return 0; }
swarmd_channel_open_t req;
memset(&req, 0, sizeof(req));
req.protocol.data = (utf8_t *)protocol.c_str();
req.protocol.len = protocol.size();
req.id = id_bytes.empty() ? nullptr : (uint8_t *)id_bytes.data();
req.id_len = id_bytes.size();
req.slots = slots ? slots : 1;
const uint64_t id = _next_id++;
uint8_t *buf = nullptr;
size_t len = 0;
if (swarmd_encode_channel_open(id, &req, &buf, &len) < 0) { _fail("encode failed"); return 0; }
if (!_send_frame(buf, len)) return 0;
const uint8_t *body = nullptr;
size_t body_len = 0;
if (!_await(id, timeout_ms, &body, &body_len) || !body) return 0;
swarmd_channel_ref_t ref;
memset(&ref, 0, sizeof(ref));
compact_state_t in = { 0, body_len, (uint8_t *)body };
if (swarmd_channel_ref_decode(&in, &ref) < 0) { _fail("channel reply did not decode"); return 0; }
if (ref.channel == 0) _err = "the daemon did not open the channel";
return ref.channel;
}
bool Client::send(uint64_t channel, uint32_t slot, uint64_t target,
const uint8_t *bytes, size_t len_bytes, bool *writable, int timeout_ms) {
if (writable) *writable = false;
if (_fd < 0) { _err = "not connected"; return false; }
swarmd_send_request_t req;
memset(&req, 0, sizeof(req));
req.channel = channel;
req.slot = slot;
req.target = target;
req.bytes = (uint8_t *)bytes;
req.bytes_len = len_bytes;
const uint64_t id = _next_id++;
uint8_t *buf = nullptr;
size_t len = 0;
if (swarmd_encode_send(id, &req, &buf, &len) < 0) { _fail("encode failed"); return false; }
if (!_send_frame(buf, len)) return false;
const uint8_t *body = nullptr;
size_t body_len = 0;
if (!_await(id, timeout_ms, &body, &body_len) || !body) return false;
swarmd_send_result_t res;
memset(&res, 0, sizeof(res));
compact_state_t in = { 0, body_len, (uint8_t *)body };
if (swarmd_send_result_decode(&in, &res) < 0) { _fail("send reply did not decode"); return false; }
if (writable) *writable = res.writable;
if (!res.accepted) _err = "no peer has this channel open yet";
return res.accepted;
}
bool Client::status(Status &out, int timeout_ms) {
if (_fd < 0) { _err = "not connected"; return false; }
swarmd_none_t req;
memset(&req, 0, sizeof(req));
const uint64_t id = _next_id++;
uint8_t *buf = nullptr;
size_t len = 0;
if (swarmd_encode_status_get(id, &req, &buf, &len) < 0) { _fail("encode failed"); return false; }
if (!_send_frame(buf, len)) return false;
const uint8_t *body = nullptr;
size_t body_len = 0;
if (!_await(id, timeout_ms, &body, &body_len) || !body) return false;
swarmd_status_t st;
memset(&st, 0, sizeof(st));
compact_state_t in = { 0, body_len, (uint8_t *)body };
if (swarmd_status_decode(&in, &st) < 0) { _fail("status reply did not decode"); return false; }
out.dht_state = st.dht_state;
out.bootstrapped = st.bootstrapped;
out.peers_known = st.peers_known;
out.peers_connected = st.peers_connected;
out.connecting = st.connecting;
out.queued = st.queued;
out.local_clients = st.local_clients;
return true;
}
bool Client::content_hash(const std::string &path, uint8_t out_hash[32], int timeout_ms) {
memset(out_hash, 0, 32);
if (_fd < 0) { _err = "not connected"; return false; }
swarmd_path_t req;
memset(&req, 0, sizeof(req));
req.path.data = (utf8_t *)path.c_str();
req.path.len = path.size();
const uint64_t id = _next_id++;
uint8_t *buf = nullptr;
size_t len = 0;
if (swarmd_encode_content_hash(id, &req, &buf, &len) < 0) { _fail("encode failed"); return false; }
if (!_send_frame(buf, len)) return false;
const uint8_t *body = nullptr;
size_t body_len = 0;
if (!_await(id, timeout_ms, &body, &body_len) || !body) return false;
swarmd_hash_t res;
memset(&res, 0, sizeof(res));
compact_state_t in = { 0, body_len, (uint8_t *)body };
if (swarmd_hash_decode(&in, &res) < 0) { _fail("hash reply did not decode"); return false; }
memcpy(out_hash, res.hash, 32);
static const uint8_t zero[32] = { 0 };
if (memcmp(out_hash, zero, 32) == 0) {
// The daemon answers all-zero when it cannot read the file. Saying so is
// better than handing back a hash nothing will ever match.
_err = "the daemon could not hash " + path;
return false;
}
return true;
}
} // namespace swarmd