/* Licensed to the Apache Software Foundation (ASF) under one or more

 * contributor license agreements.  See the NOTICE file distributed with

 * this work for additional information regarding copyright ownership.

 * The ASF licenses this file to You under the Apache License, Version 2.0

 * (the "License"); you may not use this file except in compliance with

 * the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

#include <assert.h>

#include <apr_strings.h>

#include <httpd.h>

#include <http_core.h>

#include <http_protocol.h>

#include <http_log.h>

#include <http_connection.h>

#include <scoreboard.h>

#include "h2_private.h"

#include "h2.h"

#include "h2_config.h"

#include "h2_conn_io.h"

#include "h2_ctx.h"

#include "h2_mplx.h"

#include "h2_push.h"

#include "h2_task.h"

#include "h2_stream.h"

#include "h2_request.h"

#include "h2_headers.h"

#include "h2_stream.h"

#include "h2_session.h"

#include "h2_util.h"

#include "h2_version.h"

#include "h2_filter.h"

#define UNSET       -1

#define H2MIN(x,y) ((x) < (y) ? (x) : (y))

static apr_status_t recv_RAW_DATA(conn_rec *c, h2_filter_cin *cin, 

                                  apr_bucket *b, apr_read_type_e block)

{

    h2_session *session = cin->session;

    apr_status_t status = APR_SUCCESS;

    apr_size_t len;

    const char *data;

    ssize_t n;

    status = apr_bucket_read(b, &data, &len, block);

    while (status == APR_SUCCESS && len > 0) {

        n = nghttp2_session_mem_recv(session->ngh2, (const uint8_t *)data, len);

        ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,

                      H2_SSSN_MSG(session, "fed %ld bytes to nghttp2, %ld read"),

                      (long)len, (long)n);

        if (n < 0) {

            if (nghttp2_is_fatal((int)n)) {

                h2_session_event(session, H2_SESSION_EV_PROTO_ERROR, 

                                 (int)n, nghttp2_strerror((int)n));

                status = APR_EGENERAL;

            }

        }

        else {

            session->io.bytes_read += n;

            if (len <= n) {

                break;

            }

            len -= n;

            data += n;

        }

    }

    return status;

}

static apr_status_t recv_RAW_brigade(conn_rec *c, h2_filter_cin *cin, 

                                     apr_bucket_brigade *bb, 

                                     apr_read_type_e block)

{

    apr_status_t status = APR_SUCCESS;

    apr_bucket* b;

    int consumed = 0;

    h2_util_bb_log(c, c->id, APLOG_TRACE2, "RAW_in", bb);

    while (status == APR_SUCCESS && !APR_BRIGADE_EMPTY(bb)) {

        b = APR_BRIGADE_FIRST(bb);

        if (APR_BUCKET_IS_METADATA(b)) {

            /* nop */

        }

        else {

            status = recv_RAW_DATA(c, cin, b, block);

        }

        consumed = 1;

        apr_bucket_delete(b);

    }

    if (!consumed && status == APR_SUCCESS && block == APR_NONBLOCK_READ) {

        return APR_EAGAIN;

    }

    return status;

}

h2_filter_cin *h2_filter_cin_create(h2_session *session)

{

    h2_filter_cin *cin;

    cin = apr_pcalloc(session->pool, sizeof(*cin));

    if (!cin) {

        return NULL;

    }

    cin->session = session;

    return cin;

}

void h2_filter_cin_timeout_set(h2_filter_cin *cin, apr_interval_time_t timeout)

{

    cin->timeout = timeout;

}

apr_status_t h2_filter_core_input(ap_filter_t* f,

                                  apr_bucket_brigade* brigade,

                                  ap_input_mode_t mode,

                                  apr_read_type_e block,

                                  apr_off_t readbytes) 

{

    h2_filter_cin *cin = f->ctx;

    apr_status_t status = APR_SUCCESS;

    apr_interval_time_t saved_timeout = UNSET;

    const int trace1 = APLOGctrace1(f->c);

    if (trace1) {

        ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, f->c,

                      "h2_session(%ld): read, %s, mode=%d, readbytes=%ld", 

                      (long)f->c->id, (block == APR_BLOCK_READ)? 

                      "BLOCK_READ" : "NONBLOCK_READ", mode, (long)readbytes);

    }

    if (mode == AP_MODE_INIT || mode == AP_MODE_SPECULATIVE) {

        return ap_get_brigade(f->next, brigade, mode, block, readbytes);

    }

    if (mode != AP_MODE_READBYTES) {

        return (block == APR_BLOCK_READ)? APR_SUCCESS : APR_EAGAIN;

    }

    if (!cin->bb) {

        cin->bb = apr_brigade_create(cin->session->pool, f->c->bucket_alloc);

    }

    if (!cin->socket) {

        cin->socket = ap_get_conn_socket(f->c);

    }

    if (APR_BRIGADE_EMPTY(cin->bb)) {

        /* We only do a blocking read when we have no streams to process. So,

         * in httpd scoreboard lingo, we are in a KEEPALIVE connection state.

         */

        if (block == APR_BLOCK_READ) {

            if (cin->timeout > 0) {

                apr_socket_timeout_get(cin->socket, &saved_timeout);

                apr_socket_timeout_set(cin->socket, cin->timeout);

            }

        }

        status = ap_get_brigade(f->next, cin->bb, AP_MODE_READBYTES,

                                block, readbytes);

        if (saved_timeout != UNSET) {

            apr_socket_timeout_set(cin->socket, saved_timeout);

        }

    }

    switch (status) {

        case APR_SUCCESS:

            status = recv_RAW_brigade(f->c, cin, cin->bb, block);

            break;

        case APR_EOF:

        case APR_EAGAIN:

        case APR_TIMEUP:

            if (trace1) {

                ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, f->c,

                              "h2_session(%ld): read", f->c->id);

            }

            break;

        default:

            ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, f->c, APLOGNO(03046)

                          "h2_session(%ld): error reading", f->c->id);

            break;

    }

    return status;

}

/*******************************************************************************

 * http2 connection status handler + stream out source

 ******************************************************************************/

typedef struct {

    apr_bucket_refcount refcount;

    h2_bucket_event_cb *cb;

    void *ctx;

} h2_bucket_observer;

static apr_status_t bucket_read(apr_bucket *b, const char **str,

                                apr_size_t *len, apr_read_type_e block)

{

    (void)b;

    (void)block;

    *str = NULL;

    *len = 0;

    return APR_SUCCESS;

}

static void bucket_destroy(void *data)

{

    h2_bucket_observer *h = data;

    if (apr_bucket_shared_destroy(h)) {

        if (h->cb) {

            h->cb(h->ctx, H2_BUCKET_EV_BEFORE_DESTROY, NULL);

        }

        apr_bucket_free(h);

    }

}

apr_bucket * h2_bucket_observer_make(apr_bucket *b, h2_bucket_event_cb *cb,

                                 void *ctx)

{

    h2_bucket_observer *br;

    br = apr_bucket_alloc(sizeof(*br), b->list);

    br->cb = cb;

    br->ctx = ctx;

    b = apr_bucket_shared_make(b, br, 0, 0);

    b->type = &h2_bucket_type_observer;

    return b;

} 

apr_bucket * h2_bucket_observer_create(apr_bucket_alloc_t *list, 

                                       h2_bucket_event_cb *cb, void *ctx)

{

    apr_bucket *b = apr_bucket_alloc(sizeof(*b), list);

    APR_BUCKET_INIT(b);

    b->free = apr_bucket_free;

    b->list = list;

    b = h2_bucket_observer_make(b, cb, ctx);

    return b;

}

apr_status_t h2_bucket_observer_fire(apr_bucket *b, h2_bucket_event event)

{

    if (H2_BUCKET_IS_OBSERVER(b)) {

        h2_bucket_observer *l = (h2_bucket_observer *)b->data; 

        return l->cb(l->ctx, event, b);

    }

    return APR_EINVAL;

}

const apr_bucket_type_t h2_bucket_type_observer = {

    "H2OBS", 5, APR_BUCKET_METADATA,

    bucket_destroy,

    bucket_read,

    apr_bucket_setaside_noop,

    apr_bucket_split_notimpl,

    apr_bucket_shared_copy

};

apr_bucket *h2_bucket_observer_beam(struct h2_bucket_beam *beam,

                                    apr_bucket_brigade *dest,

                                    const apr_bucket *src)

{

    if (H2_BUCKET_IS_OBSERVER(src)) {

        h2_bucket_observer *l = (h2_bucket_observer *)src->data; 

        apr_bucket *b = h2_bucket_observer_create(dest->bucket_alloc, 

                                                  l->cb, l->ctx);

        APR_BRIGADE_INSERT_TAIL(dest, b);

        l->cb = NULL;

        l->ctx = NULL;

        h2_bucket_observer_fire(b, H2_BUCKET_EV_BEFORE_MASTER_SEND);

        return b;

    }

    return NULL;

}

static apr_status_t bbout(apr_bucket_brigade *bb, const char *fmt, ...)

                             __attribute__((format(printf,2,3)));

static apr_status_t bbout(apr_bucket_brigade *bb, const char *fmt, ...)

{

    va_list args;

    apr_status_t rv;

    va_start(args, fmt);

    rv = apr_brigade_vprintf(bb, NULL, NULL, fmt, args);

    va_end(args);

    return rv;

}

static void add_settings(apr_bucket_brigade *bb, h2_session *s, int last) 

{

    h2_mplx *m = s->mplx;

    bbout(bb, "  \"settings\": {\n");

    bbout(bb, "    \"SETTINGS_MAX_CONCURRENT_STREAMS\": %d,\n", m->max_streams); 

    bbout(bb, "    \"SETTINGS_MAX_FRAME_SIZE\": %d,\n", 16*1024); 

    bbout(bb, "    \"SETTINGS_INITIAL_WINDOW_SIZE\": %d,\n",

          h2_config_geti(s->config, H2_CONF_WIN_SIZE));

    bbout(bb, "    \"SETTINGS_ENABLE_PUSH\": %d\n", h2_session_push_enabled(s)); 

    bbout(bb, "  }%s\n", last? "" : ",");

}

static void add_peer_settings(apr_bucket_brigade *bb, h2_session *s, int last) 

{

    bbout(bb, "  \"peerSettings\": {\n");

    bbout(bb, "    \"SETTINGS_MAX_CONCURRENT_STREAMS\": %d,\n", 

        nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS)); 

    bbout(bb, "    \"SETTINGS_MAX_FRAME_SIZE\": %d,\n", 

        nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_MAX_FRAME_SIZE)); 

    bbout(bb, "    \"SETTINGS_INITIAL_WINDOW_SIZE\": %d,\n", 

        nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE)); 

    bbout(bb, "    \"SETTINGS_ENABLE_PUSH\": %d,\n", 

        nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_ENABLE_PUSH)); 

    bbout(bb, "    \"SETTINGS_HEADER_TABLE_SIZE\": %d,\n", 

        nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_HEADER_TABLE_SIZE)); 

    bbout(bb, "    \"SETTINGS_MAX_HEADER_LIST_SIZE\": %d\n", 

        nghttp2_session_get_remote_settings(s->ngh2, NGHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE)); 

    bbout(bb, "  }%s\n", last? "" : ",");

}

typedef struct {

    apr_bucket_brigade *bb;

    h2_session *s;

    int idx;

} stream_ctx_t;

static int add_stream(h2_stream *stream, void *ctx)

{

    stream_ctx_t *x = ctx;

    int32_t flowIn, flowOut;

    flowIn = nghttp2_session_get_stream_effective_local_window_size(x->s->ngh2, stream->id); 

    flowOut = nghttp2_session_get_stream_remote_window_size(x->s->ngh2, stream->id);

    bbout(x->bb, "%s\n    \"%d\": {\n", (x->idx? "," : ""), stream->id);

    bbout(x->bb, "    \"state\": \"%s\",\n", h2_stream_state_str(stream));

    bbout(x->bb, "    \"created\": %f,\n", ((double)stream->created)/APR_USEC_PER_SEC);

    bbout(x->bb, "    \"flowIn\": %d,\n", flowIn);

    bbout(x->bb, "    \"flowOut\": %d,\n", flowOut);

    bbout(x->bb, "    \"dataIn\": %"APR_OFF_T_FMT",\n", stream->in_data_octets);  

    bbout(x->bb, "    \"dataOut\": %"APR_OFF_T_FMT"\n", stream->out_data_octets);  

    bbout(x->bb, "    }");

    ++x->idx;

    return 1;

} 

static void add_streams(apr_bucket_brigade *bb, h2_session *s, int last) 

{

    stream_ctx_t x;

    x.bb = bb;

    x.s = s;

    x.idx = 0;

    bbout(bb, "  \"streams\": {");

    h2_mplx_stream_do(s->mplx, add_stream, &x);

    bbout(bb, "\n  }%s\n", last? "" : ",");

}

static void add_push(apr_bucket_brigade *bb, h2_session *s, 

                     h2_stream *stream, int last) 

{

    h2_push_diary *diary;

    apr_status_t status;

    bbout(bb, "    \"push\": {\n");

    diary = s->push_diary;

    if (diary) {

        const char *data;

        const char *base64_digest;

        apr_size_t len;

        status = h2_push_diary_digest_get(diary, bb->p, 256, 

                                          stream->request->authority, 

                                          &data, &len;;

        if (status == APR_SUCCESS) {

            base64_digest = h2_util_base64url_encode(data, len, bb->p);

            bbout(bb, "      \"cacheDigest\": \"%s\",\n", base64_digest);

        }

    }

    bbout(bb, "      \"promises\": %d,\n", s->pushes_promised);

    bbout(bb, "      \"submits\": %d,\n", s->pushes_submitted);

    bbout(bb, "      \"resets\": %d\n", s->pushes_reset);

    bbout(bb, "    }%s\n", last? "" : ",");

}

static void add_in(apr_bucket_brigade *bb, h2_session *s, int last) 

{

    bbout(bb, "    \"in\": {\n");

    bbout(bb, "      \"requests\": %d,\n", s->remote.emitted_count);

    bbout(bb, "      \"resets\": %d, \n", s->streams_reset);

    bbout(bb, "      \"frames\": %ld,\n", (long)s->frames_received);

    bbout(bb, "      \"octets\": %"APR_UINT64_T_FMT"\n", s->io.bytes_read);

    bbout(bb, "    }%s\n", last? "" : ",");

}

static void add_out(apr_bucket_brigade *bb, h2_session *s, int last) 

{

    bbout(bb, "    \"out\": {\n");

    bbout(bb, "      \"responses\": %d,\n", s->responses_submitted);

    bbout(bb, "      \"frames\": %ld,\n", (long)s->frames_sent);

    bbout(bb, "      \"octets\": %"APR_UINT64_T_FMT"\n", s->io.bytes_written);

    bbout(bb, "    }%s\n", last? "" : ",");

}

static void add_stats(apr_bucket_brigade *bb, h2_session *s, 

                     h2_stream *stream, int last) 

{

    bbout(bb, "  \"stats\": {\n");

    add_in(bb, s, 0);

    add_out(bb, s, 0);

    add_push(bb, s, stream, 1);

    bbout(bb, "  }%s\n", last? "" : ",");

}

static apr_status_t h2_status_insert(h2_task *task, apr_bucket *b)

{

    conn_rec *c = task->c->master;

    h2_ctx *h2ctx = h2_ctx_get(c, 0);

    h2_session *session;

    h2_stream *stream;

    apr_bucket_brigade *bb;

    apr_bucket *e;

    int32_t connFlowIn, connFlowOut;

    if (!h2ctx || (session = h2_ctx_session_get(h2ctx)) == NULL) {

        return APR_SUCCESS;

    }

    stream = h2_session_stream_get(session, task->stream_id);

    if (!stream) {

        /* stream already done */

        return APR_SUCCESS;

    }

    bb = apr_brigade_create(stream->pool, c->bucket_alloc);

    connFlowIn = nghttp2_session_get_effective_local_window_size(session->ngh2); 

    connFlowOut = nghttp2_session_get_remote_window_size(session->ngh2);

    bbout(bb, "{\n");

    bbout(bb, "  \"version\": \"draft-01\",\n");

    add_settings(bb, session, 0);

    add_peer_settings(bb, session, 0);

    bbout(bb, "  \"connFlowIn\": %d,\n", connFlowIn);

    bbout(bb, "  \"connFlowOut\": %d,\n", connFlowOut);

    bbout(bb, "  \"sentGoAway\": %d,\n", session->local.shutdown);

    add_streams(bb, session, 0);

    add_stats(bb, session, stream, 1);

    bbout(bb, "}\n");

    while ((e = APR_BRIGADE_FIRST(bb)) != APR_BRIGADE_SENTINEL(bb)) {

        APR_BUCKET_REMOVE(e);

        APR_BUCKET_INSERT_AFTER(b, e);

        b = e;

    }

    apr_brigade_destroy(bb);

    return APR_SUCCESS;

}

static apr_status_t status_event(void *ctx, h2_bucket_event event, 

                                 apr_bucket *b)

{

    h2_task *task = ctx;

    ap_log_cerror(APLOG_MARK, APLOG_TRACE3, 0, task->c->master, 

                  "status_event(%s): %d", task->id, event);

    switch (event) {

        case H2_BUCKET_EV_BEFORE_MASTER_SEND:

            h2_status_insert(task, b);

            break;

        default:

            break;

    }

    return APR_SUCCESS;

}

int h2_filter_h2_status_handler(request_rec *r)

{

    h2_ctx *ctx = h2_ctx_rget(r);

    conn_rec *c = r->connection;

    h2_task *task;

    apr_bucket_brigade *bb;

    apr_bucket *b;

    apr_status_t status;

    if (strcmp(r->handler, "http2-status")) {

        return DECLINED;

    }

    if (r->method_number != M_GET && r->method_number != M_POST) {

        return DECLINED;

    }

    task = ctx? h2_ctx_get_task(ctx) : NULL;

    if (task) {

        if ((status = ap_discard_request_body(r)) != OK) {

            return status;

        }

        /* We need to handle the actual output on the main thread, as

         * we need to access h2_session information. */

        r->status = 200;

        r->clength = -1;

        r->chunked = 1;

        apr_table_unset(r->headers_out, "Content-Length");

        /* Discourage content-encodings */

        apr_table_unset(r->headers_out, "Content-Encoding");

        apr_table_setn(r->subprocess_env, "no-brotli", "1");

        apr_table_setn(r->subprocess_env, "no-gzip", "1");

        ap_set_content_type(r, "application/json");

        apr_table_setn(r->notes, H2_FILTER_DEBUG_NOTE, "on");

        bb = apr_brigade_create(r->pool, c->bucket_alloc);

        b = h2_bucket_observer_create(c->bucket_alloc, status_event, task);

        APR_BRIGADE_INSERT_TAIL(bb, b);

        b = apr_bucket_eos_create(c->bucket_alloc);

        APR_BRIGADE_INSERT_TAIL(bb, b);

        ap_log_rerror(APLOG_MARK, APLOG_TRACE2, 0, r,

                      "status_handler(%s): checking for incoming trailers", 

                      task->id);

        if (r->trailers_in && !apr_is_empty_table(r->trailers_in)) {

            ap_log_rerror(APLOG_MARK, APLOG_TRACE2, 0, r,

                          "status_handler(%s): seeing incoming trailers", 

                          task->id);

            apr_table_setn(r->trailers_out, "h2-trailers-in", 

                           apr_itoa(r->pool, 1));

        }

        status = ap_pass_brigade(r->output_filters, bb);

        if (status == APR_SUCCESS

            || r->status != HTTP_OK

            || c->aborted) {

            return OK;

        }

        else {

            /* no way to know what type of error occurred */

            ap_log_rerror(APLOG_MARK, APLOG_TRACE1, status, r,

                          "status_handler(%s): ap_pass_brigade failed", 

                          task->id);

            return AP_FILTER_ERROR;

        }

    }

    return DECLINED;

}