src/webhook.c

#include "webhook.h"

#include "auth.h"
#include "crypto.h"
#include "web.h"

#include <errno.h>
#include <fcntl.h>
#include <limits.h>
#include <netdb.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>

/* ---- payload helpers --------------------------------------------------- */

void json_put_esc(struct buf *b, const char *s, size_t n)
{
    for (size_t i = 0; i < n; i++) {
        unsigned char c = (unsigned char)s[i];
        switch (c) {
        case '"':  buf_puts(b, "\\\""); break;
        case '\\': buf_puts(b, "\\\\"); break;
        case '\n': buf_puts(b, "\\n");  break;
        case '\r': buf_puts(b, "\\r");  break;
        case '\t': buf_puts(b, "\\t");  break;
        default:
            if (c < 0x20)
                buf_printf(b, "\\u%04x", c);
            else
                buf_put(b, &c, 1);      /* UTF-8 passes through */
        }
    }
}

/* ---- enqueue ------------------------------------------------------------ */

/* comma-separated event list membership ("push,mr") — whole tokens */
static int ev_match(const char *list, const char *event)
{
    size_t el = strlen(event);
    const char *p = list;
    while (*p) {
        const char *cm = strchr(p, ',');
        size_t tl = cm ? (size_t)(cm - p) : strlen(p);
        if (tl == el && memcmp(p, event, el) == 0)
            return 1;
        if (!cm)
            break;
        p = cm + 1;
    }
    return 0;
}

int wh_enqueue(sqlite3 *db, long repo_id, const char *event,
               const char *payload)
{
    sqlite3_stmt *st;
    if (sqlite3_prepare_v2(db,
            "SELECT id, events FROM repo_webhooks"
            " WHERE repo_id=? AND active=1", -1, &st, NULL) != SQLITE_OK)
        return -1;
    sqlite3_bind_int64(st, 1, repo_id);

    /* collect ids first: inserting while stepping the same table on
     * one connection is legal but fragile */
    long ids[64];
    int ni = 0;
    while (sqlite3_step(st) == SQLITE_ROW && ni < 64) {
        const char *ev = (const char *)sqlite3_column_text(st, 1);
        if (ev && ev_match(ev, event))
            ids[ni++] = sqlite3_column_int64(st, 0);
    }
    sqlite3_finalize(st);

    sqlite3_stmt *ins;
    if (ni > 0 && sqlite3_prepare_v2(db,
            "INSERT INTO webhook_deliveries(hook_id,event,payload)"
            " VALUES(?,?,?)", -1, &ins, NULL) != SQLITE_OK)
        return -1;
    int n = 0;
    for (int i = 0; i < ni; i++) {
        sqlite3_reset(ins);
        sqlite3_bind_int64(ins, 1, ids[i]);
        sqlite3_bind_text(ins, 2, event, -1, SQLITE_TRANSIENT);
        sqlite3_bind_text(ins, 3, payload, -1, SQLITE_TRANSIENT);
        if (sqlite3_step(ins) == SQLITE_DONE)
            n++;
    }
    if (ni > 0)
        sqlite3_finalize(ins);
    return n;
}

void wh_notify(struct app_ctx *c, long repo_id, const char *event,
               const char *payload)
{
    if (wh_enqueue(c->db, repo_id, event, payload) > 0)
        wh_dispatch_async(c);
}

/* ---- detached dispatcher ------------------------------------------------ */

void wh_dispatch_async(struct app_ctx *c)
{
    (void)c;
    gt_spawn_task("webhook-run");
}

/* ---- transport: in-tree HTTP ------------------------------------------- */

/* connect with a hard deadline — a hanging endpoint must never stall
 * the dispatcher (or, through it, a push) */
static int wh_connect(const char *host, int port, char *err, size_t esz)
{
    char ps[8];
    snprintf(ps, sizeof ps, "%d", port);
    struct addrinfo hints;
    memset(&hints, 0, sizeof hints);
    hints.ai_family   = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    struct addrinfo *ai = NULL;
    int g = getaddrinfo(host, ps, &hints, &ai);
    if (g != 0 || !ai) {
        snprintf(err, esz, "dns: %s", gai_strerror(g));
        return -1;
    }
    int fd = -1;
    for (struct addrinfo *a = ai; a && fd < 0; a = a->ai_next) {
        fd = socket(a->ai_family, a->ai_socktype, a->ai_protocol);
        if (fd < 0)
            continue;
        fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
        if (connect(fd, a->ai_addr, a->ai_addrlen) == 0)
            break;
        if (errno != EINPROGRESS) {
            close(fd);
            fd = -1;
            continue;
        }
        fd_set ws;
        FD_ZERO(&ws);
        FD_SET(fd, &ws);
        struct timeval tv = { .tv_sec = 3, .tv_usec = 0 };
        if (select(fd + 1, NULL, &ws, NULL, &tv) <= 0) {
            close(fd);
            fd = -1;
            continue;
        }
        int so = 0;
        socklen_t sl = sizeof so;
        if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &so, &sl) != 0 ||
            so != 0) {
            close(fd);
            fd = -1;
            continue;
        }
        break;
    }
    freeaddrinfo(ai);
    if (fd < 0) {
        snprintf(err, esz, "connect failed");
        return -1;
    }
    fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) & ~O_NONBLOCK);
    struct timeval tv = { .tv_sec = 5, .tv_usec = 0 };
    setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof tv);
    setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv);
    return fd;
}

/* parse http://host[:port][/path] — https is handed to curl */
static int wh_post_http(const char *url, const char *event,
                        const char *sig, const char *body, size_t blen,
                        int *code, char *err, size_t esz)
{
    const char *p = url + 7;                /* skip http:// */
    const char *sep = p + strcspn(p, "/:");
    if (sep == p || sep - p > 250) {
        snprintf(err, esz, "bad url");
        return -1;
    }
    char host[256];
    memcpy(host, p, (size_t)(sep - p));
    host[sep - p] = '\0';
    int port = 80;
    const char *path = "/";
    if (*sep == ':') {
        port = atoi(sep + 1);
        if (port <= 0 || port > 65535) {
            snprintf(err, esz, "bad port");
            return -1;
        }
        const char *sl = strchr(sep + 1, '/');
        if (sl)
            path = sl;
    } else if (*sep == '/')
        path = sep;

    int fd = wh_connect(host, port, err, esz);
    if (fd < 0)
        return -1;

    struct buf req;
    buf_init(&req);
    buf_printf(&req,
        "POST %s HTTP/1.1\r\n"
        "Host: %s\r\n"
        "User-Agent: git-trace\r\n"
        "Content-Type: application/json\r\n"
        "Content-Length: %zu\r\n"
        "X-Git-Trace-Event: %s\r\n",
        path, host, blen, event);
    if (sig)
        buf_printf(&req, "X-Git-Trace-Signature: sha256=%s\r\n", sig);
    buf_puts(&req, "Connection: close\r\n\r\n");

    int rc = -1;
    for (size_t w = 0; w < req.len; ) {
        ssize_t n = send(fd, req.data + w, req.len - w, 0);
        if (n <= 0)
            goto out;
        w += (size_t)n;
    }
    for (size_t w = 0; w < blen; ) {
        ssize_t n = send(fd, body + w, blen - w, 0);
        if (n <= 0)
            goto out;
        w += (size_t)n;
    }
    {
        char resp[512];
        ssize_t r = recv(fd, resp, sizeof resp - 1, 0);
        if (r > 0) {
            resp[r] = '\0';
            if (strncmp(resp, "HTTP/", 5) == 0) {
                const char *sp = strchr(resp, ' ');
                *code = sp ? atoi(sp + 1) : 0;
                if (*code > 0)
                    rc = 0;
            }
        }
    }
    if (rc != 0)
        snprintf(err, esz, "no http response");
out:
    buf_free(&req);
    close(fd);
    return rc;
}

/* https:// or exotic schemes: delegate to curl as a process —
 * payload on stdin (@-), nothing sensitive on argv */
static int wh_post_curl(const char *url, const char *event,
                        const char *sig, const char *body, size_t blen,
                        int *code, char *err, size_t esz)
{
    int pin[2], pout[2];
    if (pipe(pin) != 0 || pipe(pout) != 0) {
        snprintf(err, esz, "pipe failed");
        return -1;
    }
    char evh[128], sih[128];
    snprintf(evh, sizeof evh, "X-Git-Trace-Event: %s", event);
    sih[0] = '\0';
    if (sig)
        snprintf(sih, sizeof sih, "X-Git-Trace-Signature: sha256=%s",
                 sig);

    pid_t pid = fork();
    if (pid == 0) {
        dup2(pin[0], 0);
        dup2(pout[1], 1);
        int devnull = open("/dev/null", O_WRONLY);
        if (devnull >= 0)
            dup2(devnull, 2);
        close(pin[0]); close(pin[1]);
        close(pout[0]); close(pout[1]);
        const char *argv[24];
        int i = 0;
        argv[i++] = "curl";
        argv[i++] = "-sS"; argv[i++] = "-m"; argv[i++] = "8";
        argv[i++] = "-o"; argv[i++] = "/dev/null";
        argv[i++] = "-w"; argv[i++] = "%{http_code}";
        argv[i++] = "-X"; argv[i++] = "POST";
        argv[i++] = "-H"; argv[i++] = "Content-Type: application/json";
        argv[i++] = "-H"; argv[i++] = evh;
        if (sih[0]) {
            argv[i++] = "-H";
            argv[i++] = sih;
        }
        argv[i++] = "--data-binary"; argv[i++] = "@-";
        argv[i++] = url;
        argv[i] = NULL;
        execvp("curl", (char *const *)argv);
        _exit(127);
    }
    close(pin[0]);
    close(pout[1]);
    if (pid < 0) {
        close(pin[1]); close(pout[0]);
        snprintf(err, esz, "fork failed");
        return -1;
    }
    /* payloads are capped small; a single write loop is enough —
     * curl drains stdin while it runs */
    for (size_t w = 0; w < blen; ) {
        ssize_t n = write(pin[1], body + w, blen - w);
        if (n <= 0)
            break;
        w += (size_t)n;
    }
    close(pin[1]);

    char codebuf[64];
    size_t cl = 0;
    while (cl < sizeof codebuf - 1) {
        ssize_t n = read(pout[0], codebuf + cl,
                         sizeof codebuf - 1 - cl);
        if (n <= 0)
            break;
        cl += (size_t)n;
    }
    codebuf[cl] = '\0';
    close(pout[0]);

    int st = 0, done = 0;
    for (int i = 0; i < 240 && !done; i++) {   /* ~12s budget */
        done = waitpid(pid, &st, WNOHANG) == pid;
        if (!done)
            usleep(50000);
    }
    if (!done) {
        kill(pid, SIGKILL);
        waitpid(pid, &st, 0);
        snprintf(err, esz, "curl timeout");
        return -1;
    }
    if (!WIFEXITED(st) || WEXITSTATUS(st) != 0) {
        snprintf(err, esz, "curl exit %d",
                 WIFEXITED(st) ? WEXITSTATUS(st) : -1);
        return -1;
    }
    *code = atoi(codebuf);
    if (*code <= 0) {
        snprintf(err, esz, "no http response");
        return -1;
    }
    return 0;
}

/* ---- one delivery ------------------------------------------------------- */

static int wh_deliver(const char *url, const char *secret,
                      const char *event, const char *payload,
                      int *code, char *err, size_t esz)
{
    char sighex[80];
    const char *sig = NULL;
    if (secret && *secret) {
        uint8_t mac[32];
        gt_hmac_sha256((const uint8_t *)secret, strlen(secret),
                       (const uint8_t *)payload, strlen(payload), mac);
        char hex[65];
        to_hex(mac, 32, hex);
        snprintf(sighex, sizeof sighex, "%s", hex);
        sig = sighex;
    }
    if (strncmp(url, "http://", 7) == 0)
        return wh_post_http(url, event, sig, payload,
                            strlen(payload), code, err, esz);
    if (strncmp(url, "https://", 8) == 0)
        return wh_post_curl(url, event, sig, payload,
                            strlen(payload), code, err, esz);
    snprintf(err, esz, "unsupported scheme");
    return -1;
}

int wh_run(struct app_ctx *c)
{
    sqlite3_stmt *st;
    if (sqlite3_prepare_v2(c->db,
            "SELECT id FROM webhook_deliveries"
            " WHERE status='pending' AND attempts<5"
            " ORDER BY id LIMIT 50", -1, &st, NULL) != SQLITE_OK)
        return -1;
    long ids[50];
    int n = 0;
    while (sqlite3_step(st) == SQLITE_ROW && n < 50)
        ids[n++] = sqlite3_column_int64(st, 0);
    sqlite3_finalize(st);

    int done = 0;
    for (int i = 0; i < n; i++) {
        if (sqlite3_prepare_v2(c->db,
                "SELECT h.url, h.secret, d.event, d.payload"
                " FROM webhook_deliveries d"
                " JOIN repo_webhooks h ON h.id=d.hook_id"
                " WHERE d.id=?", -1, &st, NULL) != SQLITE_OK)
            continue;
        sqlite3_bind_int64(st, 1, ids[i]);
        if (sqlite3_step(st) != SQLITE_ROW) {
            sqlite3_finalize(st);
            continue;
        }
        const char *url  = (const char *)sqlite3_column_text(st, 0);
        const char *sec  = (const char *)sqlite3_column_text(st, 1);
        const char *ev   = (const char *)sqlite3_column_text(st, 2);
        const char *pay  = (const char *)sqlite3_column_text(st, 3);
        int code = 0;
        char err[120] = "";
        int rc = wh_deliver(url ? url : "", sec ? sec : "",
                            ev ? ev : "push", pay ? pay : "{}",
                            &code, err, sizeof err);
        sqlite3_finalize(st);

        int ok = (rc == 0 && code >= 200 && code < 300);
        if (!ok && rc == 0)
            snprintf(err, sizeof err, "http %d", code);
        if (sqlite3_prepare_v2(c->db,
                "UPDATE webhook_deliveries SET"
                " attempts=attempts+1, status=?, code=?, error=?,"
                " done_at=strftime('%s','now') WHERE id=?",
                -1, &st, NULL) == SQLITE_OK) {
            sqlite3_bind_text(st, 1, ok ? "ok" : "fail", -1,
                              SQLITE_STATIC);
            sqlite3_bind_int(st, 2, code);
            sqlite3_bind_text(st, 3, err, -1, SQLITE_TRANSIENT);
            sqlite3_bind_int64(st, 4, ids[i]);
            sqlite3_step(st);
            sqlite3_finalize(st);
        }
        done++;
    }
    return done;
}