feat: parliament daemon, login gate, peer context sync (0.9.13)

Login claims the chair, writes a session ticket, and starts a background
service for discovery plus revision-based context sync. Password Realm
opens require --login (or CST_REALM_PASSWORD); --service/--service-stop
manage the resident agent.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
WHC
2026-10-28 23:11:23 +08:00
co-authored by Cursor
parent de9ee28773
commit e853abac5e
13 changed files with 769 additions and 12 deletions
+68 -2
View File
@@ -1,7 +1,9 @@
#include "cst_agent.h"
#include "cst_ext.h"
#include "cst_models.h"
#include "cst_parliament.h"
#include "cst_parliament_cli.h"
#include "cst_realm.h"
#include "cst_tools.h"
#include "cst_version.h"
#include "cst_webui.h"
@@ -50,7 +52,10 @@ static void usage(const char *argv0) {
" --no-orchestrate override workspace orchestration setting\n"
" --realm-new create Realm file keys (legacy/dev)\n"
" --realm-password-init create password-derived Realm (no sk on disk)\n"
" --login unlock with password; this device becomes chair\n"
" --login unlock; claim chair; start parliament service\n"
" --service run parliament service (discover+context sync)\n"
" --service-fg same as --service, stay in foreground\n"
" --service-stop stop the parliament background service\n"
" --new-project ID replace all parliament context for a new project\n"
" --realm-enroll ID enroll this instance under the Realm\n"
" --realm-label S display label for enroll/login\n"
@@ -374,7 +379,8 @@ int main(int argc, char **argv) {
int new_session = 0, do_list = 0, do_migrate = 0;
const char *resume_id = NULL;
int do_realm_new = 0, do_realm_enroll = 0, do_peer_listen = 0, do_peer_dial = 0,
do_discover = 0, do_realm_pw_init = 0, do_login = 0, do_new_project = 0;
do_discover = 0, do_realm_pw_init = 0, do_login = 0, do_new_project = 0,
do_service = 0, do_service_fg = 0, do_service_stop = 0;
const char *realm_enroll_id = NULL, *realm_label = NULL;
const char *peer_dial_spec = NULL, *parliament_instance = NULL;
const char *new_project_id = NULL;
@@ -439,6 +445,9 @@ int main(int argc, char **argv) {
{"realm-password-init", no_argument, 0, 0x118},
{"login", no_argument, 0, 0x119},
{"new-project", required_argument, 0, 0x11a},
{"service", no_argument, 0, 0x11b},
{"service-fg", no_argument, 0, 0x11c},
{"service-stop", no_argument, 0, 0x11d},
{0, 0, 0, 0}};
int c;
while ((c = getopt_long(argc, argv, "w:s:b:k:m:P:r:n:T:Hp:dahVNcR:LyA:C:e:M:",
@@ -492,6 +501,15 @@ int main(int argc, char **argv) {
do_new_project = 1;
new_project_id = optarg;
break;
case 0x11b:
do_service = 1;
break;
case 0x11c:
do_service_fg = 1;
break;
case 0x11d:
do_service_stop = 1;
break;
case 'w':
snprintf(cfg.workspace_root, sizeof(cfg.workspace_root), "%s", optarg);
break;
@@ -675,6 +693,22 @@ int main(int argc, char **argv) {
curl_global_cleanup();
return rc;
}
if (do_service_stop) {
int rc = cst_parliament_cli_service_stop();
curl_global_cleanup();
return rc;
}
if (do_service || do_service_fg) {
const char *pw = getenv("CST_REALM_PASSWORD");
const char *iid = parliament_instance ? parliament_instance
: getenv("CST_INSTANCE_ID");
if (!iid || !iid[0])
iid = "chair-local";
int rc = cst_parliament_cli_service(pw, iid, discover_peer_port,
do_service_fg);
curl_global_cleanup();
return rc;
}
if (do_realm_enroll) {
int rc = cst_parliament_cli_realm_enroll(realm_enroll_id, realm_label);
curl_global_cleanup();
@@ -715,6 +749,38 @@ int main(int argc, char **argv) {
return rc;
}
/* Password Realm: interactive/agent open requires a chair login session. */
{
char rdir[OS_PATH_MAX];
if (cst_realm_dir(rdir, sizeof(rdir)) == 0 &&
cst_realm_is_password_mode(rdir) && !cst_parliament_session_ok(NULL, 0)) {
const char *pw = getenv("CST_REALM_PASSWORD");
if (pw && pw[0]) {
const char *iid = parliament_instance ? parliament_instance
: getenv("CST_INSTANCE_ID");
if (!iid || !iid[0])
iid = "chair-local";
char err[256];
char *hand =
cst_parliament_login(pw, iid, realm_label, err, sizeof(err));
if (!hand) {
fprintf(stderr, "%s\n", err[0] ? err : "login required");
curl_global_cleanup();
return 1;
}
free(hand);
(void)cst_parliament_daemon_start(pw, iid, discover_peer_port, err,
sizeof(err));
} else {
fprintf(stderr,
"password Realm active: run --login (or set "
"CST_REALM_PASSWORD) before opening the agent\n");
curl_global_cleanup();
return 1;
}
}
}
if (do_list) {
int rc = list_sessions(cfg.workspace_root);
curl_global_cleanup();
+114 -1
View File
@@ -165,7 +165,7 @@ int cst_parliament_claim_chair(const char *instance_id,
return cst_parliament_save(&m);
}
int cst_parliament_context_put(const char *text) {
static int context_write_file(const char *text) {
char path[OS_PATH_MAX], tmp[OS_PATH_MAX + 8];
if (context_path(path, sizeof(path)) != 0)
return -1;
@@ -185,6 +185,12 @@ int cst_parliament_context_put(const char *text) {
remove(tmp);
return -1;
}
return 0;
}
int cst_parliament_context_put(const char *text) {
if (context_write_file(text) != 0)
return -1;
cst_parliament_meta_t m;
if (cst_parliament_load(&m) == 0) {
m.revision++;
@@ -290,8 +296,115 @@ char *cst_parliament_login(const char *password, const char *instance_id,
cst_buf_puts(&hand, ctx ? ctx : "");
free(ctx);
free(dev);
if (cst_parliament_session_write(instance_id) != 0) {
if (err && err_sz)
snprintf(err, err_sz, "session ticket write failed");
crypto_wipe(keys.seed, sizeof(keys.seed));
crypto_wipe(keys.sk, sizeof(keys.sk));
crypto_wipe(sk, sizeof(sk));
cst_buf_free(&hand);
return NULL;
}
crypto_wipe(keys.seed, sizeof(keys.seed));
crypto_wipe(keys.sk, sizeof(keys.sk));
crypto_wipe(sk, sizeof(sk));
return cst_buf_steal(&hand);
}
static int session_path(char *out, size_t n) {
char dir[OS_PATH_MAX];
if (cst_realm_dir(dir, sizeof(dir)) != 0)
return -1;
return os_path_join(out, n, dir, "chair.session");
}
int cst_parliament_session_write(const char *instance_id) {
if (!instance_id || !instance_id[0])
return -1;
char path[OS_PATH_MAX], tmp[OS_PATH_MAX + 8];
if (session_path(path, sizeof(path)) != 0)
return -1;
if (snprintf(tmp, sizeof(tmp), "%s.tmp", path) >= (int)sizeof(tmp))
return -1;
FILE *f = fopen(tmp, "wb");
if (!f)
return -1;
int ok = fprintf(f, "instance=%s\nissued=%ld\n", instance_id,
(long)time(NULL)) > 0;
if (fclose(f) != 0)
ok = 0;
if (!ok) {
remove(tmp);
return -1;
}
if (os_rename_replace(tmp, path) != 0) {
remove(tmp);
return -1;
}
return 0;
}
int cst_parliament_session_ok(char *instance_id_out, size_t n) {
char path[OS_PATH_MAX];
if (session_path(path, sizeof(path)) != 0)
return 0;
FILE *f = fopen(path, "rb");
if (!f)
return 0;
char line[256], iid[64] = {0};
long issued = 0;
while (fgets(line, sizeof(line), f)) {
if (!strncmp(line, "instance=", 9)) {
const char *p = line + 9;
size_t i = 0;
while (p[i] && p[i] != '\n' && p[i] != '\r' && i + 1 < sizeof(iid)) {
iid[i] = p[i];
i++;
}
iid[i] = 0;
} else if (!strncmp(line, "issued=", 7))
issued = atol(line + 7);
}
fclose(f);
if (!iid[0] || issued <= 0)
return 0;
/* Ticket valid for 24h of wall clock. */
long now = (long)time(NULL);
if (now < issued || now - issued > 86400)
return 0;
cst_parliament_meta_t m;
if (cst_parliament_load(&m) != 0)
return 0;
if (!m.chair_instance[0] || strcmp(m.chair_instance, iid) != 0)
return 0;
if (instance_id_out && n)
snprintf(instance_id_out, n, "%s", iid);
return 1;
}
void cst_parliament_session_clear(void) {
char path[OS_PATH_MAX];
if (session_path(path, sizeof(path)) == 0)
remove(path);
}
int cst_parliament_apply_sync(unsigned long revision, const char *parliament_id,
const char *project_id, const char *chair,
const char *context) {
cst_parliament_meta_t m;
if (cst_parliament_load(&m) != 0)
return -1;
if (revision <= m.revision)
return 0;
if (parliament_id && parliament_id[0])
snprintf(m.parliament_id, sizeof(m.parliament_id), "%s", parliament_id);
if (project_id)
snprintf(m.project_id, sizeof(m.project_id), "%s", project_id);
if (chair)
snprintf(m.chair_instance, sizeof(m.chair_instance), "%s", chair);
m.revision = revision;
snprintf(m.status, sizeof(m.status), "%s", "open");
if (context_write_file(context ? context : "") != 0)
return -1;
return cst_parliament_save(&m);
}
+68 -3
View File
@@ -69,6 +69,13 @@ int cst_parliament_cli_realm_password_init(const char *password) {
int cst_parliament_cli_login(const char *password, const char *instance_id,
const char *device_label) {
return cst_parliament_cli_login_and_serve(password, instance_id, device_label,
9100);
}
int cst_parliament_cli_login_and_serve(const char *password,
const char *instance_id,
const char *device_label, int peer_port) {
char err[256];
char *hand = cst_parliament_login(password, instance_id, device_label, err,
sizeof(err));
@@ -78,6 +85,49 @@ int cst_parliament_cli_login(const char *password, const char *instance_id,
}
fputs(hand, stdout);
free(hand);
if (cst_parliament_daemon_start(password, instance_id,
peer_port > 0 ? peer_port : 9100, err,
sizeof(err)) != 0) {
fprintf(stderr, "service: %s\n", err[0] ? err : "start failed");
return 1;
}
if (cst_parliament_daemon_running())
printf("service=running\n");
return 0;
}
int cst_parliament_cli_service(const char *password, const char *instance_id,
int peer_port, int foreground) {
char err[256];
if (!instance_id || !instance_id[0]) {
fprintf(stderr, "instance id required\n");
return 1;
}
if (foreground) {
if (cst_parliament_daemon_run(password, instance_id,
peer_port > 0 ? peer_port : 9100, err,
sizeof(err)) != 0) {
fprintf(stderr, "%s\n", err[0] ? err : "service failed");
return 1;
}
return 0;
}
if (cst_parliament_daemon_start(password, instance_id,
peer_port > 0 ? peer_port : 9100, err,
sizeof(err)) != 0) {
fprintf(stderr, "%s\n", err[0] ? err : "service start failed");
return 1;
}
printf("service=running\n");
return 0;
}
int cst_parliament_cli_service_stop(void) {
if (cst_parliament_daemon_stop() != 0) {
fprintf(stderr, "service stop failed\n");
return 1;
}
printf("service=stopped\n");
return 0;
}
@@ -116,6 +166,19 @@ int cst_parliament_cli_realm_enroll(const char *instance_id, const char *label)
return 0;
}
static int load_realm_for_cli(cst_realm_keys_t *realm) {
char dir[OS_PATH_MAX];
if (cst_realm_dir(dir, sizeof(dir)) != 0)
return -1;
if (cst_realm_is_password_mode(dir)) {
const char *pw = getenv("CST_REALM_PASSWORD");
if (!pw || !pw[0])
return -1;
return cst_realm_unlock_password(dir, pw, realm);
}
return cst_realm_load(dir, realm);
}
int cst_parliament_cli_peer_listen(int port, const char *instance_id) {
if (!instance_id || !instance_id[0]) {
fprintf(stderr, "instance id required\n");
@@ -125,7 +188,7 @@ int cst_parliament_cli_peer_listen(int port, const char *instance_id) {
cst_realm_keys_t realm;
cst_realm_cert_t cert;
uint8_t sk[64];
if (cst_realm_dir(dir, sizeof(dir)) != 0 || cst_realm_load(dir, &realm) != 0 ||
if (cst_realm_dir(dir, sizeof(dir)) != 0 || load_realm_for_cli(&realm) != 0 ||
cst_realm_instance_load(instance_id, &cert, sk) != 0) {
fprintf(stderr, "cannot load realm/instance %s\n", instance_id);
return 1;
@@ -151,6 +214,7 @@ int cst_parliament_cli_peer_listen(int port, const char *instance_id) {
return 1;
}
printf("peer_ok remote=%s host=%s\n", sess.remote.instance_id, sess.host);
cst_peer_close(&sess);
os_net_close(ls);
return 0;
}
@@ -165,7 +229,7 @@ int cst_parliament_cli_peer_dial(const char *host, int port,
cst_realm_keys_t realm;
cst_realm_cert_t cert;
uint8_t sk[64];
if (cst_realm_dir(dir, sizeof(dir)) != 0 || cst_realm_load(dir, &realm) != 0 ||
if (cst_realm_dir(dir, sizeof(dir)) != 0 || load_realm_for_cli(&realm) != 0 ||
cst_realm_instance_load(instance_id, &cert, sk) != 0) {
fprintf(stderr, "cannot load realm/instance %s\n", instance_id);
return 1;
@@ -177,6 +241,7 @@ int cst_parliament_cli_peer_dial(const char *host, int port,
return 1;
}
printf("peer_ok remote=%s\n", sess.remote.instance_id);
cst_peer_close(&sess);
return 0;
}
@@ -188,7 +253,7 @@ int cst_parliament_cli_discover(const char *instance_id, int peer_port,
}
char dir[OS_PATH_MAX], fp[33];
cst_realm_keys_t realm;
if (cst_realm_dir(dir, sizeof(dir)) != 0 || cst_realm_load(dir, &realm) != 0 ||
if (cst_realm_dir(dir, sizeof(dir)) != 0 || load_realm_for_cli(&realm) != 0 ||
cst_realm_fingerprint(realm.pk, fp) != 0) {
fprintf(stderr, "cannot load realm\n");
return 1;
+321
View File
@@ -0,0 +1,321 @@
#ifndef _DEFAULT_SOURCE
#define _DEFAULT_SOURCE
#endif
#include "cst_parliament.h"
#include "cst_discover.h"
#include "cst_net.h"
#include "cst_peer.h"
#include "cst_realm.h"
#include "cst_util.h"
#include "monocypher.h"
#include "os_compat.h"
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#ifdef _WIN32
#include <windows.h>
#else
#include <fcntl.h>
#include <signal.h>
#include <sys/stat.h>
#include <unistd.h>
#endif
static volatile int g_stop;
#ifndef _WIN32
static void on_sig(int sig) {
(void)sig;
g_stop = 1;
}
#endif
int cst_parliament_daemon_pidfile(char *out, size_t n) {
char dir[OS_PATH_MAX];
if (cst_realm_dir(dir, sizeof(dir)) != 0)
return -1;
return os_path_join(out, n, dir, "parliament.pid");
}
static int read_pid(void) {
char path[OS_PATH_MAX];
if (cst_parliament_daemon_pidfile(path, sizeof(path)) != 0)
return -1;
FILE *f = fopen(path, "rb");
if (!f)
return -1;
int pid = -1;
if (fscanf(f, "%d", &pid) != 1)
pid = -1;
fclose(f);
return pid;
}
static int write_pid(int pid) {
char path[OS_PATH_MAX], tmp[OS_PATH_MAX + 8];
if (cst_parliament_daemon_pidfile(path, sizeof(path)) != 0)
return -1;
if (snprintf(tmp, sizeof(tmp), "%s.tmp", path) >= (int)sizeof(tmp))
return -1;
FILE *f = fopen(tmp, "wb");
if (!f)
return -1;
int ok = fprintf(f, "%d\n", pid) > 0;
if (fclose(f) != 0)
ok = 0;
if (!ok) {
remove(tmp);
return -1;
}
return os_rename_replace(tmp, path);
}
static void clear_pid(void) {
char path[OS_PATH_MAX];
if (cst_parliament_daemon_pidfile(path, sizeof(path)) == 0)
remove(path);
}
int cst_parliament_daemon_running(void) {
int pid = read_pid();
if (pid <= 0)
return 0;
#ifdef _WIN32
HANDLE h = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, FALSE, (DWORD)pid);
if (!h)
return 0;
CloseHandle(h);
return 1;
#else
return kill(pid, 0) == 0;
#endif
}
static void sleep_ms(int ms) {
#ifdef _WIN32
Sleep((DWORD)ms);
#else
struct timespec ts = {ms / 1000, (ms % 1000) * 1000000L};
nanosleep(&ts, NULL);
#endif
}
static int load_local_bundle(cst_peer_sync_bundle_t *b) {
memset(b, 0, sizeof(*b));
cst_parliament_meta_t m;
if (cst_parliament_load(&m) != 0)
return -1;
b->revision = m.revision;
snprintf(b->parliament_id, sizeof(b->parliament_id), "%s", m.parliament_id);
snprintf(b->project_id, sizeof(b->project_id), "%s", m.project_id);
snprintf(b->chair_instance, sizeof(b->chair_instance), "%s",
m.chair_instance);
b->context = cst_parliament_context_get();
return b->context ? 0 : -1;
}
static int apply_if_newer(cst_peer_sync_bundle_t *b, unsigned long before) {
if (b->revision <= before)
return 0;
return cst_parliament_apply_sync(b->revision, b->parliament_id, b->project_id,
b->chair_instance, b->context);
}
static int sync_on_session(cst_peer_session_t *sess) {
cst_peer_sync_bundle_t local;
if (load_local_bundle(&local) != 0)
return -1;
unsigned long before = local.revision;
char err[128];
int rc = cst_peer_sync(sess, &local, 15000, err, sizeof(err));
if (rc == 0)
apply_if_newer(&local, before);
free(local.context);
return rc;
}
static int unlock_realm(const char *password, cst_realm_keys_t *keys) {
char dir[OS_PATH_MAX];
if (cst_realm_dir(dir, sizeof(dir)) != 0)
return -1;
if (cst_realm_is_password_mode(dir))
return cst_realm_unlock_password(dir, password, keys);
return cst_realm_load(dir, keys);
}
int cst_parliament_daemon_run(const char *password, const char *instance_id,
int peer_port, char *err, size_t err_sz) {
if (err && err_sz)
err[0] = 0;
if (!instance_id || !instance_id[0]) {
if (err && err_sz)
snprintf(err, err_sz, "instance id required");
return -1;
}
cst_realm_keys_t realm;
if (unlock_realm(password, &realm) != 0) {
if (err && err_sz)
snprintf(err, err_sz, "cannot unlock realm");
return -1;
}
cst_realm_cert_t cert;
uint8_t sk[64];
if (cst_realm_instance_load(instance_id, &cert, sk) != 0) {
if (err && err_sz)
snprintf(err, err_sz, "instance not enrolled; login first");
crypto_wipe(realm.seed, sizeof(realm.seed));
crypto_wipe(realm.sk, sizeof(realm.sk));
return -1;
}
char fp[33];
cst_realm_fingerprint(realm.pk, fp);
os_net_init();
char nerr[128];
os_socket_t ls =
os_net_listen("0.0.0.0", peer_port > 0 ? peer_port : 0, nerr, sizeof(nerr));
if (ls == OS_SOCKET_BAD) {
if (err && err_sz)
snprintf(err, err_sz, "listen failed: %s", nerr);
crypto_wipe(sk, sizeof(sk));
crypto_wipe(realm.seed, sizeof(realm.seed));
crypto_wipe(realm.sk, sizeof(realm.sk));
return -1;
}
int bound = os_net_port(ls);
cst_discover_t *disc =
cst_discover_start(fp, instance_id, instance_id, bound);
write_pid(
#ifdef _WIN32
(int)GetCurrentProcessId()
#else
(int)getpid()
#endif
);
fprintf(stderr, "parliament_service pid=%d port=%d id=%s realm=%s\n",
#ifdef _WIN32
(int)GetCurrentProcessId(),
#else
(int)getpid(),
#endif
bound, instance_id, fp);
fflush(stderr);
#ifndef _WIN32
signal(SIGTERM, on_sig);
signal(SIGINT, on_sig);
signal(SIGHUP, on_sig);
#endif
g_stop = 0;
int tick = 0;
while (!g_stop) {
cst_peer_session_t sess;
char herr[128];
if (cst_peer_accept((intptr_t)ls, &cert, sk, realm.pk, 500, &sess, herr,
sizeof(herr)) == 0) {
sync_on_session(&sess);
cst_peer_close(&sess);
}
if (++tick % 10 == 0 && disc) {
cst_peer_info_t peers[16];
int n = cst_discover_peers(disc, peers, 16);
for (int i = 0; i < n; i++) {
if (peers[i].port <= 0)
continue;
if (cst_peer_dial(peers[i].host, peers[i].port, &cert, sk, realm.pk,
3000, &sess, herr, sizeof(herr)) == 0) {
sync_on_session(&sess);
cst_peer_close(&sess);
}
}
}
sleep_ms(200);
}
if (disc)
cst_discover_stop(disc);
os_net_close(ls);
clear_pid();
crypto_wipe(sk, sizeof(sk));
crypto_wipe(realm.seed, sizeof(realm.seed));
crypto_wipe(realm.sk, sizeof(realm.sk));
return 0;
}
int cst_parliament_daemon_stop(void) {
int pid = read_pid();
if (pid <= 0)
return 0;
#ifdef _WIN32
HANDLE h = OpenProcess(PROCESS_TERMINATE, FALSE, (DWORD)pid);
if (h) {
TerminateProcess(h, 0);
CloseHandle(h);
}
#else
if (kill(pid, SIGTERM) != 0 && errno != ESRCH)
return -1;
for (int i = 0; i < 25; i++) {
if (kill(pid, 0) != 0)
break;
sleep_ms(100);
}
if (kill(pid, 0) == 0)
kill(pid, SIGKILL);
#endif
clear_pid();
return 0;
}
int cst_parliament_daemon_start(const char *password, const char *instance_id,
int peer_port, char *err, size_t err_sz) {
if (err && err_sz)
err[0] = 0;
if (cst_parliament_daemon_running())
return 0;
#ifdef _WIN32
/* Windows: run in-process foreground loop under a detached console is
* deferred; callers should use --service in a service manager. */
if (err && err_sz)
snprintf(err, err_sz, "background daemon requires --service on Windows");
return -1;
#else
if (!password || !password[0]) {
if (err && err_sz)
snprintf(err, err_sz, "password required to start service");
return -1;
}
pid_t pid = fork();
if (pid < 0) {
if (err && err_sz)
snprintf(err, err_sz, "fork failed");
return -1;
}
if (pid > 0) {
/* Parent: wait briefly for pidfile. */
for (int i = 0; i < 50; i++) {
if (cst_parliament_daemon_running())
return 0;
sleep_ms(50);
}
if (err && err_sz)
snprintf(err, err_sz, "service did not start");
return -1;
}
/* Child: detach. */
setsid();
int fd = open("/dev/null", O_RDWR);
if (fd >= 0) {
dup2(fd, 0);
/* Keep stderr for service banner until closed by operator logs. */
if (fd > 2)
close(fd);
}
char derr[256];
(void)cst_parliament_daemon_run(password, instance_id, peer_port, derr,
sizeof(derr));
_exit(0);
#endif
}
+132 -2
View File
@@ -298,7 +298,7 @@ int cst_peer_dial(const char *host, int port, const cst_realm_cert_t *self,
out->remote = remote;
snprintf(out->host, sizeof(out->host), "%s", host);
out->port = port;
os_net_close(s);
out->sock = (intptr_t)s;
return 0;
}
@@ -329,6 +329,136 @@ int cst_peer_accept(intptr_t listen_sock, const cst_realm_cert_t *self,
out->remote = remote;
os_net_peer(s, out->host, sizeof(out->host));
out->port = 0;
os_net_close(s);
out->sock = (intptr_t)s;
return 0;
}
void cst_peer_close(cst_peer_session_t *s) {
if (!s)
return;
if (s->sock != 0 && s->sock != (intptr_t)OS_SOCKET_BAD)
os_net_close((os_socket_t)s->sock);
s->sock = (intptr_t)OS_SOCKET_BAD;
}
static int send_sync_hdr(os_socket_t s, const cst_peer_sync_bundle_t *b,
int timeout_ms) {
cJSON *o = cJSON_CreateObject();
if (!o)
return -1;
cJSON_AddNumberToObject(o, "v", 1);
cJSON_AddStringToObject(o, "op", "sync");
cJSON_AddNumberToObject(o, "revision", (double)b->revision);
cJSON_AddStringToObject(o, "parliament_id", b->parliament_id);
cJSON_AddStringToObject(o, "project_id", b->project_id);
cJSON_AddStringToObject(o, "chair", b->chair_instance);
size_t clen = b->context ? strlen(b->context) : 0;
cJSON_AddNumberToObject(o, "context_len", (double)clen);
char *line = cJSON_PrintUnformatted(o);
cJSON_Delete(o);
if (!line)
return -1;
int rc = send_line(s, line, timeout_ms);
free(line);
if (rc != 0)
return -1;
if (clen > 0 &&
os_net_send_all(s, b->context, clen, timeout_ms) != 0)
return -1;
return 0;
}
static int read_sync_hdr(os_socket_t s, cst_peer_sync_bundle_t *b,
int timeout_ms) {
memset(b, 0, sizeof(*b));
char *line = recv_line(s, timeout_ms);
if (!line)
return -1;
cJSON *o = cJSON_Parse(line);
free(line);
if (!o)
return -1;
const cJSON *op = cJSON_GetObjectItemCaseSensitive(o, "op");
if (!cJSON_IsString(op) || strcmp(op->valuestring, "sync") != 0) {
cJSON_Delete(o);
return -1;
}
const cJSON *v;
v = cJSON_GetObjectItemCaseSensitive(o, "revision");
if (cJSON_IsNumber(v))
b->revision = (unsigned long)v->valuedouble;
v = cJSON_GetObjectItemCaseSensitive(o, "parliament_id");
if (cJSON_IsString(v))
snprintf(b->parliament_id, sizeof(b->parliament_id), "%s", v->valuestring);
v = cJSON_GetObjectItemCaseSensitive(o, "project_id");
if (cJSON_IsString(v))
snprintf(b->project_id, sizeof(b->project_id), "%s", v->valuestring);
v = cJSON_GetObjectItemCaseSensitive(o, "chair");
if (cJSON_IsString(v))
snprintf(b->chair_instance, sizeof(b->chair_instance), "%s",
v->valuestring);
size_t clen = 0;
v = cJSON_GetObjectItemCaseSensitive(o, "context_len");
if (cJSON_IsNumber(v) && v->valuedouble >= 0)
clen = (size_t)v->valuedouble;
cJSON_Delete(o);
if (clen > 2 * 1024 * 1024)
return -1;
if (clen == 0) {
b->context = strdup("");
return b->context ? 0 : -1;
}
char *ctx = malloc(clen + 1);
if (!ctx)
return -1;
size_t got = 0;
while (got < clen) {
long r = os_net_recv(s, ctx + got, clen - got, timeout_ms);
if (r <= 0) {
free(ctx);
return -1;
}
got += (size_t)r;
}
ctx[clen] = 0;
b->context = ctx;
return 0;
}
int cst_peer_sync(cst_peer_session_t *s, cst_peer_sync_bundle_t *local,
int timeout_ms, char *err, size_t err_sz) {
if (err && err_sz)
err[0] = 0;
if (!s || !local || s->sock == (intptr_t)OS_SOCKET_BAD) {
if (err && err_sz)
snprintf(err, err_sz, "bad peer session");
return -1;
}
os_socket_t sock = (os_socket_t)s->sock;
if (send_sync_hdr(sock, local, timeout_ms) != 0) {
if (err && err_sz)
snprintf(err, err_sz, "sync send failed");
return -1;
}
cst_peer_sync_bundle_t remote;
if (read_sync_hdr(sock, &remote, timeout_ms) != 0) {
if (err && err_sz)
snprintf(err, err_sz, "sync recv failed");
return -1;
}
if (remote.revision > local->revision) {
free(local->context);
local->context = remote.context;
remote.context = NULL;
local->revision = remote.revision;
snprintf(local->parliament_id, sizeof(local->parliament_id), "%s",
remote.parliament_id);
snprintf(local->project_id, sizeof(local->project_id), "%s",
remote.project_id);
snprintf(local->chair_instance, sizeof(local->chair_instance), "%s",
remote.chair_instance);
} else {
free(remote.context);
}
return 0;
}