view release on metacpan or  search on metacpan

xs/proto_http.c  view on Meta::CPAN

    const char *p = buf;
    const char *end = buf + len;
    int status;

    /* skip "HTTP/1.x " */
    while (p < end && *p != ' ') p++;
    if (p >= end) return 0;
    p++;

    status = atoi(p);
    if (status < 100 || status > 599) return 500; /* treat malformed as server error */
    return status;
}

/* Find header value (case-insensitive). Returns pointer into buf or NULL. */
static const char* find_header(const char *headers, size_t headers_len,
                                const char *name, size_t *value_len) {
    size_t name_len = strlen(name);
    const char *p = headers;
    const char *end = headers + headers_len;

    while (p < end) {
        const char *line_end = p;
        while (line_end < end && *line_end != '\r') line_end++;

        if ((size_t)(line_end - p) > name_len + 1 && p[name_len] == ':') {
            int match = 1;
            size_t i;
            for (i = 0; i < name_len; i++) {
                if (tolower((unsigned char)p[i]) != tolower((unsigned char)name[i])) {
                    match = 0;
                    break;
                }
            }
            if (match) {
                const char *val = p + name_len + 1;
                while (val < line_end && *val == ' ') val++;
                *value_len = line_end - val;
                return val;
            }
        }

        /* advance past \r\n */
        if (line_end + 2 <= end) p = line_end + 2;
        else break;
    }
    return NULL;
}

/* Parse a complete HTTP response from recv_buf. */
static void process_http_response(ev_clickhouse_t *self) {
    size_t hdr_end;
    int status;
    const char *val;
    size_t val_len;
    size_t content_length = 0;
    int chunked = 0;
    int is_gzip = 0;
    const char *body;
    size_t body_len;
    char *decoded = NULL;
    size_t decoded_len = 0;
    size_t decoded_cap = 0;

    if (self->recv_len == 0 || self->send_count == 0) return;

    /* find headers end */
    hdr_end = find_header_end(self->recv_buf, self->recv_len);
    if (hdr_end == 0) return; /* need more data */

    /* parse status */
    status = parse_http_status(self->recv_buf, hdr_end);

    /* parse Content-Length */
    val = find_header(self->recv_buf, hdr_end, "Content-Length", &val_len);
    if (val) {
        content_length = (size_t)strtoul(val, NULL, 10);
    }

    /* check Transfer-Encoding: chunked */
    val = find_header(self->recv_buf, hdr_end, "Transfer-Encoding", &val_len);
    if (val && val_len >= 7 && strncasecmp(val, "chunked", 7) == 0) {
        chunked = 1;
    }

    /* check Content-Encoding: gzip */
    val = find_header(self->recv_buf, hdr_end, "Content-Encoding", &val_len);
    if (val && val_len >= 4 && strncasecmp(val, "gzip", 4) == 0) {
        is_gzip = 1;
    }

    /* parse X-ClickHouse-Summary: {"read_rows":"N","read_bytes":"N",...}
     * to populate profile_rows/profile_bytes for HTTP, mirroring the
     * native protocol's SERVER_PROFILE_INFO packet. */
    val = find_header(self->recv_buf, hdr_end, "X-ClickHouse-Summary", &val_len);
    if (val && val_len > 12) {
        size_t i;
        for (i = 0; i + 12 < val_len; i++) {
            if (val[i] != '"') continue;
            /* val isn't NUL-terminated (it points into recv_buf), so use
             * parse_uint_within which takes an explicit length cap rather
             * than scanning until a non-digit/NUL like strtoull would. */
            if (i + 13 < val_len
                && memcmp(val + i + 1, "read_rows\":\"", 12) == 0)
                self->profile_rows = parse_uint_within(val + i + 13,
                                                       val_len - (i + 13));
            else if (i + 14 < val_len
                     && memcmp(val + i + 1, "read_bytes\":\"", 13) == 0)
                self->profile_bytes = parse_uint_within(val + i + 14,
                                                        val_len - (i + 14));
        }
    }

    size_t consumed;
    if (chunked) {
        /* decode chunked transfer encoding */
        const char *cp = self->recv_buf + hdr_end;
        const char *cp_end = self->recv_buf + self->recv_len;

        {
            int chunked_complete = 0;
            while (cp < cp_end) {
                /* read chunk size */
                const char *nl = cp;
                unsigned long chunk_size;
                while (nl < cp_end && *nl != '\r') nl++;
                if (nl + 2 > cp_end) goto need_more; /* need more data */

                chunk_size = strtoul(cp, NULL, 16);
                cp = nl + 2; /* skip \r\n */

                if (chunk_size == 0) {
                    /* terminal chunk; skip trailing \r\n */
                    if (cp + 2 > cp_end) goto need_more;
                    cp += 2;
                    chunked_complete = 1;
                    break;
                }

                if ((size_t)(cp_end - cp) < 2
                    || chunk_size > (size_t)(cp_end - cp) - 2) goto need_more;

                /* guard against overflow and unbounded growth —
                 * close connection since remaining chunks would
                 * corrupt the stream for subsequent requests.
                 * Apply both the hard ceiling and the user's
                 * opt-in max_recv_buffer (when set) to the
                 * post-decode body size. */
                size_t cap = CH_MAX_DECOMPRESS_SIZE;
                if (self->max_recv_buffer > 0
                    && self->max_recv_buffer < cap)
                    cap = self->max_recv_buffer;
                if (decoded_len + chunk_size < decoded_len
                    || decoded_len + chunk_size > cap) {
                    if (decoded) Safefree(decoded);
                    self->send_count--;
                    teardown_after_deliver(self,
                        "chunked response too large", "connection closed");
                    return;
                }
                if (decoded == NULL) {
                    decoded_cap = chunk_size + 256;
                    Newx(decoded, decoded_cap, char);
                } else if (decoded_len + chunk_size > decoded_cap) {
                    decoded_cap = (decoded_len + chunk_size) * 2;
                    Renew(decoded, decoded_cap, char);
                }
                Copy(cp, decoded + decoded_len, chunk_size, char);
                decoded_len += chunk_size;
                cp += chunk_size + 2; /* skip chunk data + \r\n */
            }

            if (!chunked_complete) goto need_more;
        }

        body = decoded;
        body_len = decoded_len;
        consumed = cp - self->recv_buf;
    } else {
        /* Content-Length based — bound content_length before any pointer
         * arithmetic so a malicious huge value cannot overflow
         * hdr_end + content_length and slip past the need-more check. */
        size_t cl_cap = CH_MAX_DECOMPRESS_SIZE;
        if (self->max_recv_buffer > 0 && self->max_recv_buffer < cl_cap)
            cl_cap = self->max_recv_buffer;
        if (content_length > cl_cap) {
            self->send_count--;
            teardown_after_deliver(self,
                "Content-Length exceeds limit", "connection closed");
            return;
        }
        if (self->recv_len < hdr_end + content_length) return; /* need more data */
        body = self->recv_buf + hdr_end;
        body_len = content_length;
        consumed = hdr_end + content_length;
    }

    /* deliver response (body, body_len, consumed are set; decoded may be NULL) */
    self->send_count--;
    if (status == 200) {
        char *final_body = (char *)body;
        size_t final_len = body_len;

        if (is_gzip && body_len > 0) {
            size_t dec_len;
            char *dec = gzip_decompress(body, body_len, &dec_len);
            if (!dec) {
                if (decoded) Safefree(decoded);
                recv_consume(self, consumed);
                int destroyed = deliver_error(self, "gzip decompression failed");
                if (destroyed) return;
                goto done;
            }
            /* Apply user's opt-in max_recv_buffer to the gzip-decoded
             * body too — same trip-wire semantics as the chunked path
             * and the on_readable raw-recv check. Tear the connection
             * down on overflow so subsequent queries can't slip past
             * the cap on the same socket (matches the chunked path's
             * behaviour and the POD contract). */
            if (self->max_recv_buffer > 0
                && dec_len > self->max_recv_buffer) {
                Safefree(dec);
                if (decoded) Safefree(decoded);
                teardown_after_deliver(self,
                    "gzip body exceeds max_recv_buffer", "connection closed");
                return;
            }
            final_body = dec;
            final_len = dec_len;
        }

        {
            int is_raw = peek_cb_raw(self);
            int destroyed;
            if (is_raw) {
                /* raw mode — deliver body as scalar, skip TSV parsing */
                destroyed = deliver_raw_body(self, final_body, final_len);
            } else {
                AV *rows = NULL;
                if (final_len > 0)
                    rows = parse_tab_separated(final_body, final_len);
                destroyed = deliver_rows(self, rows);
            }
            if (final_body != body) Safefree(final_body);
            if (decoded) Safefree(decoded);
            recv_consume(self, consumed);
            if (destroyed) return;
        }
    } else {
        /* error */
        char *errmsg = format_http_error(self, status, body, body_len, is_gzip);
        if (decoded) Safefree(decoded);
        recv_consume(self, consumed);
        int destroyed = deliver_error(self, errmsg);
        Safefree(errmsg);
        if (destroyed) return;
    }

    if (self->magic != EV_CH_MAGIC) return;

done:
    /* Stop query timeout timer on response */
    stop_timing(self);
    pipeline_advance(self);
    return;

need_more:
    /* incomplete response — keep reading */
    if (decoded) Safefree(decoded);
    return;
}