Hermit/src/parliament_daemon.c

#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 <pthread.h>
#include <signal.h>
#include <sys/stat.h>
#include <sys/time.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");
}

typedef struct {
  int pid;
  char instance[64];
  int port;
} daemon_meta_t;

static int read_meta(daemon_meta_t *m) {
  memset(m, 0, sizeof(*m));
  m->pid = -1;
  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;
  char line[256];
  while (fgets(line, sizeof(line), f)) {
    if (!strncmp(line, "pid=", 4))
      m->pid = atoi(line + 4);
    else 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(m->instance)) {
        m->instance[i] = p[i];
        i++;
      }
      m->instance[i] = 0;
    } else if (!strncmp(line, "port=", 5))
      m->port = atoi(line + 5);
    else if (m->pid <= 0) {
      int pid = atoi(line);
      if (pid > 0)
        m->pid = pid;
    }
  }
  fclose(f);
  return m->pid > 0 ? 0 : -1;
}

static int read_pid(void) {
  daemon_meta_t m;
  return read_meta(&m) == 0 ? m.pid : -1;
}

static int write_pid(int pid, const char *instance_id, int peer_port) {
  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, "pid=%d\ninstance=%s\nport=%d\n", pid,
                   instance_id ? instance_id : "",
                   peer_port > 0 ? peer_port : 0) > 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);
}

static int pid_alive(int 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
}

int cst_parliament_daemon_running(void) {
  daemon_meta_t m;
  if (read_meta(&m) != 0)
    return 0;
  if (!pid_alive(m.pid)) {
    clear_pid();
    return 0;
  }
  return 1;
}

int cst_parliament_daemon_info(int *pid_out, char *instance_out, size_t in_sz,
                               int *port_out) {
  daemon_meta_t m;
  if (read_meta(&m) != 0 || !pid_alive(m.pid)) {
    if (read_meta(&m) == 0 && !pid_alive(m.pid))
      clear_pid();
    return -1;
  }
  if (pid_out)
    *pid_out = m.pid;
  if (instance_out && in_sz)
    snprintf(instance_out, in_sz, "%s", m.instance);
  if (port_out)
    *port_out = m.port;
  return 0;
}

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 int64_t now_ms(void) {
#ifdef _WIN32
  return (int64_t)GetTickCount64();
#else
  struct timeval tv;
  gettimeofday(&tv, NULL);
  return (int64_t)tv.tv_sec * 1000 + (int64_t)tv.tv_usec / 1000;
#endif
}

#ifndef _WIN32
static pthread_mutex_t g_sync_mu = PTHREAD_MUTEX_INITIALIZER;
#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);
}

/* Short sync budget: accept thread must not stall the listen backlog. */
static int sync_on_session(cst_peer_session_t *sess) {
  cst_peer_sync_bundle_t local;
#ifndef _WIN32
  pthread_mutex_lock(&g_sync_mu);
#endif
  if (load_local_bundle(&local) != 0) {
#ifndef _WIN32
    pthread_mutex_unlock(&g_sync_mu);
#endif
    return -1;
  }
  unsigned long before = local.revision;
  char err[128];
  int rc = cst_peer_sync(sess, &local, 3000, err, sizeof(err));
  if (rc == 0)
    apply_if_newer(&local, before);
  free(local.context);
#ifndef _WIN32
  pthread_mutex_unlock(&g_sync_mu);
#endif
  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);
}

typedef struct {
  os_socket_t ls;
  cst_realm_cert_t cert;
  uint8_t sk[64];
  uint8_t realm_pk[32];
} accept_ctx_t;

#ifndef _WIN32
static void *accept_main(void *arg) {
  accept_ctx_t *ac = arg;
  while (!g_stop) {
    cst_peer_session_t sess;
    char herr[128];
    if (cst_peer_accept((intptr_t)ac->ls, &ac->cert, ac->sk, ac->realm_pk, 400,
                        &sess, herr, sizeof(herr)) != 0)
      continue;
    sync_on_session(&sess);
    cst_peer_close(&sess);
  }
  return NULL;
}
#endif

/* Per-peer dial cooldown so stuck remotes do not dominate the dial loop. */
typedef struct {
  char iid[64];
  int fails;
  int64_t cool_until_ms;
} dial_cool_t;

static dial_cool_t *cool_find(dial_cool_t *tab, int *n, int cap,
                              const char *iid) {
  for (int i = 0; i < *n; i++) {
    if (!strcmp(tab[i].iid, iid))
      return &tab[i];
  }
  if (*n >= cap)
    return NULL;
  dial_cool_t *c = &tab[(*n)++];
  memset(c, 0, sizeof(*c));
  snprintf(c->iid, sizeof(c->iid), "%s", iid);
  return c;
}

static int cool_blocked(dial_cool_t *c, int64_t now) {
  return c && c->cool_until_ms > now;
}

static void cool_fail(dial_cool_t *c, int64_t now) {
  if (!c)
    return;
  c->fails = c->fails < 8 ? c->fails + 1 : 8;
  /* 4s, 8s, 16s, … capped at 120s */
  int64_t wait = 4000LL << (c->fails - 1);
  if (wait > 120000)
    wait = 120000;
  c->cool_until_ms = now + wait;
}

static void cool_ok(dial_cool_t *c) {
  if (!c)
    return;
  c->fails = 0;
  c->cool_until_ms = 0;
}

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
      ,
      instance_id, bound);
  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;

  accept_ctx_t ac;
  memset(&ac, 0, sizeof(ac));
  ac.ls = ls;
  ac.cert = cert;
  memcpy(ac.sk, sk, 64);
  memcpy(ac.realm_pk, realm.pk, 32);

#ifndef _WIN32
  pthread_t ath;
  int have_ath = pthread_create(&ath, NULL, accept_main, &ac) == 0;
#else
  int have_ath = 0;
#endif

  int tick = 0, dial_rr = 0;
  dial_cool_t cool[24];
  int ncool = 0;

  while (!g_stop) {
    cst_peer_session_t sess;
    char herr[128];

#ifdef _WIN32
    /* No accept thread: keep prefer-accept single-threaded path. */
    for (int a = 0; a < 4; a++) {
      if (cst_peer_accept((intptr_t)ls, &cert, sk, realm.pk, 200, &sess, herr,
                          sizeof(herr)) != 0)
        break;
      sync_on_session(&sess);
      cst_peer_close(&sess);
    }
#endif

    int push = cst_parliament_sync_pending();
    tick++;
    int period = push ? 1 : 8;
    if ((push || tick % period == 0) && disc) {
      cst_peer_info_t peers[16];
      int n = cst_discover_peers(disc, peers, 16);
      int64_t now = now_ms();
      int tried = 0;
      for (int k = 0; k < n; k++) {
        int i = (dial_rr + k) % n;
        if (peers[i].port <= 0)
          continue;
        if (!strcmp(peers[i].instance_id, instance_id))
          continue;
        if (!push && strcmp(instance_id, peers[i].instance_id) > 0)
          continue;
        dial_cool_t *c = cool_find(cool, &ncool, 24, peers[i].instance_id);
        if (!push && cool_blocked(c, now))
          continue;
        dial_rr = (i + 1) % n;
        /* Dial budget stays short; accept thread covers inbound meanwhile. */
        int ok = cst_peer_dial(peers[i].host, peers[i].port, &cert, sk,
                               realm.pk, 2000, &sess, herr, sizeof(herr)) == 0;
        if (ok) {
          sync_on_session(&sess);
          cst_peer_close(&sess);
          cool_ok(c);
        } else {
          cool_fail(c, now);
        }
        tried = 1;
        break;
      }
      if (push && (tried || n == 0))
        cst_parliament_sync_ack();
    }
    /* Pace dial loop; accept runs concurrently on Unix. */
    sleep_ms(have_ath ? 150 : 50);
  }

#ifndef _WIN32
  if (have_ath) {
    /* Wake accept's poll by shutting down listen sock after stop. */
    os_net_shutdown(ls);
    pthread_join(ath, NULL);
  }
#endif
  if (disc)
    cst_discover_stop(disc);
  os_net_close(ls);
  clear_pid();
  crypto_wipe(sk, sizeof(sk));
  crypto_wipe(ac.sk, sizeof(ac.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 (!instance_id || !instance_id[0]) {
    if (err && err_sz)
      snprintf(err, err_sz, "instance id required");
    return -1;
  }
  int want_port = peer_port > 0 ? peer_port : 9100;
  {
    int cur_pid = 0, cur_port = 0;
    char cur_id[64] = {0};
    if (cst_parliament_daemon_info(&cur_pid, cur_id, sizeof(cur_id),
                                   &cur_port) == 0) {
      int same = (!cur_id[0] || strcmp(cur_id, instance_id) == 0) &&
                 (cur_port <= 0 || cur_port == want_port);
      if (same)
        return 0;
      (void)cst_parliament_daemon_stop();
    }
  }
#ifdef _WIN32
  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) {
    for (int i = 0; i < 50; i++) {
      int cur_pid = 0, cur_port = 0;
      char cur_id[64] = {0};
      if (cst_parliament_daemon_info(&cur_pid, cur_id, sizeof(cur_id),
                                     &cur_port) == 0 &&
          (!cur_id[0] || strcmp(cur_id, instance_id) == 0))
        return 0;
      sleep_ms(50);
    }
    if (err && err_sz)
      snprintf(err, err_sz, "service did not start");
    return -1;
  }
  setsid();
  int fd = open("/dev/null", O_RDWR);
  if (fd >= 0) {
    dup2(fd, 0);
    dup2(fd, 1);
    dup2(fd, 2);
    if (fd > 2)
      close(fd);
  }
  char derr[256];
  (void)cst_parliament_daemon_run(password, instance_id, want_port, derr,
                                  sizeof(derr));
  _exit(0);
#endif
}