/* Copyright (c) 2017 - 2019 LiteSpeed Technologies Inc. See LICENSE. */ /* * lsquic_qdec_hdl.c -- QPACK decoder streams handler */ #include #include #include #include #include #include #include "lsquic.h" #include "lsquic_types.h" #include "lsquic_int_types.h" #include "lsquic_sfcw.h" #include "lsquic_varint.h" #include "lsquic_hq.h" #include "lsquic_hash.h" #include "lsquic_stream.h" #include "lsquic_frab_list.h" #include "lsqpack.h" #include "lsquic_http1x_if.h" #include "lsquic_qdec_hdl.h" #include "lsquic_mm.h" #include "lsquic_engine_public.h" #include "lsquic_headers.h" #include "lsquic_http1x_if.h" #include "lsquic_conn.h" #define LSQUIC_LOGGER_MODULE LSQLM_QDEC_HDL #define LSQUIC_LOG_CONN_ID lsquic_conn_log_cid(qdh->qdh_conn) #include "lsquic_logger.h" static void qdh_hblock_unblocked (void *); static int qdh_write_decoder (struct qpack_dec_hdl *qdh, const unsigned char *buf, size_t sz) { ssize_t nw; if (!(qdh->qdh_dec_sm_out && lsquic_frab_list_empty(&qdh->qdh_fral))) { write_to_frab: if (0 == lsquic_frab_list_write(&qdh->qdh_fral, (unsigned char *) buf, sz)) { LSQ_DEBUG("wrote %zu bytes to frab list", sz); lsquic_stream_wantwrite(qdh->qdh_dec_sm_out, 1); return 0; } else { LSQ_INFO("error writing to frab list"); return -1; } } nw = lsquic_stream_write(qdh->qdh_dec_sm_out, buf, sz); if (nw < 0) { LSQ_INFO("error writing to outgoing QPACK decoder stream: %s", strerror(errno)); return -1; } LSQ_DEBUG("wrote %zd bytes to outgoing QPACK decoder stream", nw); if ((size_t) nw == sz) return 0; buf = buf + nw; sz -= (size_t) nw; goto write_to_frab; } static int qdh_write_type (struct qpack_dec_hdl *qdh) { int s; #ifndef NDEBUG const char *env = getenv("LSQUIC_RND_VARINT_LEN"); if (env && atoi(env)) { s = rand() & 3; LSQ_DEBUG("writing %d-byte stream type", 1 << s); } else #endif s = 0; switch (s) { case 0: return qdh_write_decoder(qdh, (unsigned char []) { HQUST_QPACK_DEC }, 1); case 1: return qdh_write_decoder(qdh, (unsigned char []) { 0x40, HQUST_QPACK_DEC }, 2); case 2: return qdh_write_decoder(qdh, (unsigned char []) { 0x80, 0x00, 0x00, HQUST_QPACK_DEC }, 4); default: return qdh_write_decoder(qdh, (unsigned char []) { 0xC0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, HQUST_QPACK_DEC }, 8); } } static void qdh_begin_out (struct qpack_dec_hdl *qdh) { if (0 != qdh_write_type(qdh)) { LSQ_WARN("%s: could not write to decoder", __func__); qdh->qdh_conn->cn_if->ci_internal_error(qdh->qdh_conn, "cannot write to decoder stream"); } } int lsquic_qdh_init (struct qpack_dec_hdl *qdh, struct lsquic_conn *conn, int is_server, const struct lsquic_engine_public *enpub, unsigned dyn_table_size, unsigned max_risked_streams) { qdh->qdh_conn = conn; lsquic_frab_list_init(&qdh->qdh_fral, 0x400, NULL, NULL, NULL); lsqpack_dec_init(&qdh->qdh_decoder, (void *) conn, dyn_table_size, max_risked_streams, qdh_hblock_unblocked); qdh->qdh_flags |= QDH_INITIALIZED; qdh->qdh_enpub = enpub; if (qdh->qdh_enpub->enp_hsi_if == lsquic_http1x_if) { qdh->qdh_h1x_ctor_ctx = (struct http1x_ctor_ctx) { .conn = conn, .max_headers_sz = 0x10000, /* XXX */ .is_server = is_server, }; qdh->qdh_hsi_ctx = &qdh->qdh_h1x_ctor_ctx; } else qdh->qdh_hsi_ctx = qdh->qdh_enpub->enp_hsi_ctx; if (qdh->qdh_dec_sm_out) qdh_begin_out(qdh); if (qdh->qdh_enc_sm_in) lsquic_stream_wantread(qdh->qdh_enc_sm_in, 1); LSQ_DEBUG("initialized"); return 0; } void lsquic_qdh_cleanup (struct qpack_dec_hdl *qdh) { if (qdh->qdh_flags & QDH_INITIALIZED) { LSQ_DEBUG("cleanup"); lsqpack_dec_cleanup(&qdh->qdh_decoder); lsquic_frab_list_cleanup(&qdh->qdh_fral); qdh->qdh_flags &= ~QDH_INITIALIZED; } } static lsquic_stream_ctx_t * qdh_out_on_new (void *stream_if_ctx, struct lsquic_stream *stream) { struct qpack_dec_hdl *const qdh = stream_if_ctx; qdh->qdh_dec_sm_out = stream; if (qdh->qdh_flags & QDH_INITIALIZED) qdh_begin_out(qdh); LSQ_DEBUG("initialized outgoing decoder stream"); return (void *) qdh; } static void qdh_out_on_write (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx) { struct qpack_dec_hdl *const qdh = (void *) ctx; struct lsquic_reader reader; ssize_t nw; unsigned char buf[LSQPACK_LONGEST_ICI]; if (lsqpack_dec_ici_pending(&qdh->qdh_decoder)) { nw = lsqpack_dec_write_ici(&qdh->qdh_decoder, buf, sizeof(buf)); if (nw > 0) { if (0 == qdh_write_decoder(qdh, buf, nw)) LSQ_DEBUG("wrote %zd-byte TSS instruction", nw); else goto err; } else if (nw < 0) { LSQ_WARN("could not generate TSS instruction"); goto err; } } if (lsquic_frab_list_empty(&qdh->qdh_fral)) { LSQ_DEBUG("%s: nothing to write", __func__); lsquic_stream_wantwrite(stream, 0); return; } reader = (struct lsquic_reader) { .lsqr_read = lsquic_frab_list_read, .lsqr_size = lsquic_frab_list_size, .lsqr_ctx = &qdh->qdh_fral, }; nw = lsquic_stream_writef(stream, &reader); if (nw >= 0) { LSQ_DEBUG("wrote %zd bytes to stream", nw); (void) lsquic_stream_flush(stream); if (lsquic_frab_list_empty(&qdh->qdh_fral)) lsquic_stream_wantwrite(stream, 0); } else { LSQ_WARN("cannot write to stream: %s", strerror(errno)); err: lsquic_stream_wantwrite(stream, 0); qdh->qdh_conn->cn_if->ci_internal_error(qdh->qdh_conn, "cannot write to stream"); } } static void qdh_out_on_close (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx) { struct qpack_dec_hdl *const qdh = (void *) ctx; qdh->qdh_dec_sm_out = NULL; LSQ_DEBUG("closed outgoing decoder stream"); } static void qdh_out_on_read (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx) { assert(0); } static const struct lsquic_stream_if qdh_dec_sm_out_if = { .on_new_stream = qdh_out_on_new, .on_read = qdh_out_on_read, .on_write = qdh_out_on_write, .on_close = qdh_out_on_close, }; const struct lsquic_stream_if *const lsquic_qdh_dec_sm_out_if = &qdh_dec_sm_out_if; static lsquic_stream_ctx_t * qdh_in_on_new (void *stream_if_ctx, struct lsquic_stream *stream) { struct qpack_dec_hdl *const qdh = stream_if_ctx; qdh->qdh_enc_sm_in = stream; if (qdh->qdh_flags & QDH_INITIALIZED) lsquic_stream_wantread(qdh->qdh_enc_sm_in, 1); LSQ_DEBUG("initialized incoming encoder stream"); return (void *) qdh; } static size_t qdh_read_encoder_stream (void *ctx, const unsigned char *buf, size_t sz, int fin) { struct qpack_dec_hdl *const qdh = (void *) ctx; const struct lsqpack_dec_err *qerr; int s; if (fin) { LSQ_INFO("encoder stream is closed"); qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1, HEC_CLOSED_CRITICAL_STREAM, "Peer closed QPACK encoder stream"); goto end; } s = lsqpack_dec_enc_in(&qdh->qdh_decoder, buf, sz); if (s != 0) { LSQ_INFO("error reading decoder stream"); qerr = lsqpack_dec_get_err_info(&qdh->qdh_decoder); qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1, HEC_QPACK_DECODER_STREAM_ERROR, "Error interpreting QPACK encoder " "stream; offset %"PRIu64", line %d", qerr->off, qerr->line); goto end; } if (qdh->qdh_dec_sm_out && lsqpack_dec_ici_pending(&qdh->qdh_decoder)) lsquic_stream_wantwrite(qdh->qdh_dec_sm_out, 1); LSQ_DEBUG("successfully fed %zu bytes to QPACK decoder", sz); end: return sz; } static void qdh_in_on_read (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx) { struct qpack_dec_hdl *const qdh = (void *) ctx; ssize_t nread; nread = lsquic_stream_readf(stream, qdh_read_encoder_stream, qdh); if (nread <= 0) { if (nread < 0) { LSQ_WARN("cannot read from encoder stream: %s", strerror(errno)); qdh->qdh_conn->cn_if->ci_internal_error(qdh->qdh_conn, "cannot read from encoder stream"); } else { LSQ_INFO("encoder stream closed by peer: abort connection"); qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1, HEC_CLOSED_CRITICAL_STREAM, "encoder stream closed"); } lsquic_stream_wantread(stream, 0); } } static void qdh_in_on_close (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx) { struct qpack_dec_hdl *const qdh = (void *) ctx; LSQ_DEBUG("closed incoming encoder stream"); qdh->qdh_enc_sm_in = NULL; } static void qdh_in_on_write (struct lsquic_stream *stream, lsquic_stream_ctx_t *ctx) { assert(0); } static const struct lsquic_stream_if qdh_enc_sm_in_if = { .on_new_stream = qdh_in_on_new, .on_read = qdh_in_on_read, .on_write = qdh_in_on_write, .on_close = qdh_in_on_close, }; const struct lsquic_stream_if *const lsquic_qdh_enc_sm_in_if = &qdh_enc_sm_in_if; static void qdh_hblock_unblocked (void *stream_p) { struct lsquic_stream *const stream = stream_p; struct qpack_dec_hdl *const qdh = lsquic_stream_get_qdh(stream); LSQ_DEBUG("header block for stream %"PRIu64" unblocked", stream->id); lsquic_stream_qdec_unblocked(stream); } static int qdh_supply_hset_to_stream (struct qpack_dec_hdl *qdh, struct lsquic_stream *stream, struct lsqpack_header_list *qlist) { const struct lsquic_hset_if *const hset_if = qdh->qdh_enpub->enp_hsi_if; struct uncompressed_headers *uh = NULL; const struct lsqpack_header *header; enum lsquic_header_status st; unsigned i; void *hset; hset = hset_if->hsi_create_header_set(qdh->qdh_hsi_ctx, 0); if (!hset) { LSQ_INFO("call to hsi_create_header_set failed"); return -1; } LSQ_DEBUG("got header set for stream %"PRIu64, stream->id); for (i = 0; i < qlist->qhl_count; ++i) { header = qlist->qhl_headers[i]; LSQ_DEBUG("%.*s: %.*s", header->qh_name_len, header->qh_name, header->qh_value_len, header->qh_value); st = hset_if->hsi_process_header(hset, header->qh_flags & QH_ID_SET ? 62 /* XXX: 62 */ + header->qh_static_id : 0, header->qh_name, header->qh_name_len, header->qh_value, header->qh_value_len); if (st != LSQUIC_HDR_OK) goto err; } lsqpack_dec_destroy_header_list(qlist); st = hset_if->hsi_process_header(hset, 0, 0, 0, 0, 0); if (st != LSQUIC_HDR_OK) goto err; uh = calloc(1, sizeof(*uh)); if (!uh) goto err; uh->uh_stream_id = stream->id; uh->uh_oth_stream_id = 0; uh->uh_weight = 0; uh->uh_exclusive = -1; if (hset_if == lsquic_http1x_if) uh->uh_flags |= UH_H1H; uh->uh_hset = hset; if (0 != lsquic_stream_uh_in(stream, uh)) goto err; LSQ_DEBUG("converted qlist to hset and gave it to stream %"PRIu64, stream->id); return 0; err: lsqpack_dec_destroy_header_list(qlist); hset_if->hsi_discard_header_set(hset); free(uh); return -1; } static enum lsqpack_read_header_status qdh_header_read_results (struct qpack_dec_hdl *qdh, struct lsquic_stream *stream, enum lsqpack_read_header_status rhs, struct lsqpack_header_list *qlist, const unsigned char *dec_buf, size_t dec_buf_sz) { const struct lsqpack_dec_err *qerr; if (rhs == LQRHS_DONE) { if (qlist) { if (0 != qdh_supply_hset_to_stream(qdh, stream, qlist)) return LQRHS_ERROR; if (qdh->qdh_dec_sm_out) { if (dec_buf_sz && 0 != qdh_write_decoder(qdh, dec_buf, dec_buf_sz)) { return LQRHS_ERROR; } if (dec_buf_sz || lsqpack_dec_ici_pending(&qdh->qdh_decoder)) lsquic_stream_wantwrite(qdh->qdh_dec_sm_out, 1); } } else { assert(0); /* XXX TODO What do we do here? */ return LQRHS_ERROR; } } else if (rhs == LQRHS_ERROR) { qerr = lsqpack_dec_get_err_info(&qdh->qdh_decoder); qdh->qdh_conn->cn_if->ci_abort_error(qdh->qdh_conn, 1, HEC_QPACK_DECOMPRESSION_FAILED, "QPACK decompression error; " "stream %"PRIu64", offset %"PRIu64", line %d", qerr->stream_id, qerr->off, qerr->line); } return rhs; } enum lsqpack_read_header_status lsquic_qdh_header_in_begin (struct qpack_dec_hdl *qdh, struct lsquic_stream *stream, uint64_t header_size, const unsigned char **buf, size_t bufsz) { enum lsqpack_read_header_status rhs; struct lsqpack_header_list *qlist; size_t dec_buf_sz; unsigned char dec_buf[LSQPACK_LONGEST_HEADER_ACK]; if (qdh->qdh_flags & QDH_INITIALIZED) { dec_buf_sz = sizeof(dec_buf); rhs = lsqpack_dec_header_in(&qdh->qdh_decoder, stream, stream->id, header_size, buf, bufsz, &qlist, dec_buf, &dec_buf_sz); return qdh_header_read_results(qdh, stream, rhs, qlist, dec_buf, dec_buf_sz); } else { LSQ_WARN("not initialized: cannot process header block"); return LQRHS_ERROR; } } enum lsqpack_read_header_status lsquic_qdh_header_in_continue (struct qpack_dec_hdl *qdh, struct lsquic_stream *stream, const unsigned char **buf, size_t bufsz) { enum lsqpack_read_header_status rhs; struct lsqpack_header_list *qlist; size_t dec_buf_sz; unsigned char dec_buf[LSQPACK_LONGEST_HEADER_ACK]; if (qdh->qdh_flags & QDH_INITIALIZED) { dec_buf_sz = sizeof(dec_buf); rhs = lsqpack_dec_header_read(&qdh->qdh_decoder, stream, buf, bufsz, &qlist, dec_buf, &dec_buf_sz); return qdh_header_read_results(qdh, stream, rhs, qlist, dec_buf, dec_buf_sz); } else { LSQ_WARN("not initialized: cannot process header block"); return LQRHS_ERROR; } } void lsquic_qdh_unref_stream (struct qpack_dec_hdl *qdh, struct lsquic_stream *stream) { if (0 == lsqpack_dec_unref_stream(&qdh->qdh_decoder, stream)) LSQ_DEBUG("unreffed stream %"PRIu64, stream->id); else LSQ_WARN("cannot unref stream %"PRIu64, stream->id); } void lsquic_qdh_cancel_stream (struct qpack_dec_hdl *qdh, struct lsquic_stream *stream) { ssize_t nw; unsigned char buf[LSQPACK_LONGEST_CANCEL]; nw = lsqpack_dec_cancel_stream(&qdh->qdh_decoder, stream, buf, sizeof(buf)); if (nw > 0) { if (0 == qdh_write_decoder(qdh, buf, nw)) LSQ_DEBUG("cancelled stream %"PRIu64" and wrote %zd-byte Cancel " "Stream instruction to the decoder stream", stream->id, nw); } else if (nw == 0) LSQ_WARN("cannot cancel stream %"PRIu64" -- not found", stream->id); else { LSQ_WARN("cannot cancel stream %"PRIu64" -- not enough buffer space " "to encode Cancel Stream instructin", stream->id); lsquic_qdh_unref_stream(qdh, stream); } }