icecast-server/src/format_ebml.c

/* Icecast
 *
 * This program is distributed under the GNU General Public License,
 * version 2. A copy of this license is included with this source.
 * At your option, this specific source file can also be distributed
 * under the GNU GPL version 3.
 *
 * Copyright 2012,      David Richards, Mozilla Foundation,
 *                      and others (see AUTHORS for details).
 * Copyright 2014,      Philipp "ph3-der-loewe" Schafft <lion@lion.leolix.org>.
 */

/* format_ebml.c
 *
 * format plugin for WebM/EBML
 *
 */

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "refbuf.h"
#include "source.h"
#include "client.h"

#include "stats.h"
#include "format.h"
#include "format_ebml.h"

#define CATMODULE "format-ebml"

#include "logging.h"

#define EBML_HEADER_MAX_SIZE 131072
#define EBML_SLICE_SIZE 4096

/* A value that no EBML var-int is allowed to take. */
#define EBML_UNKNOWN ((unsigned long long) -1)

typedef enum ebml_read_mode {
    EBML_STATE_READING_HEADER = 0,
    EBML_STATE_READING_CLUSTERS
} ebml_read_mode;

typedef enum ebml_parsing_state {
    EBML_STATE_PARSING_HEADER = 0,
    EBML_STATE_COPYING_TO_HEADER,
    EBML_STATE_START_CLUSTER,
    EBML_STATE_PARSING_CLUSTERS,
    EBML_STATE_COPYING_TO_DATA
} ebml_parsing_state;

typedef enum ebml_chunk_type {
    EBML_CHUNK_HEADER = 0,
    EBML_CHUNK_CLUSTER_START,
    EBML_CHUNK_CLUSTER_CONTINUE
} ebml_chunk_type;

typedef struct ebml_client_data_st ebml_client_data_t;

struct ebml_client_data_st {

    refbuf_t *header;
    int header_pos;

};

struct ebml_st {

    ebml_read_mode output_state;
    ebml_parsing_state parse_state;
    unsigned long long copy_len;
    
    int cluster_start;

    int position;
    unsigned char *buffer;

    int input_position;
    unsigned char *input_buffer;
    
    int header_size;
    int header_position;
    int header_read_position;
    unsigned char *header;

};

static void ebml_free_plugin(format_plugin_t *plugin);
static refbuf_t *ebml_get_buffer(source_t *source);
static int ebml_write_buf_to_client(client_t *client);
static void ebml_write_buf_to_file(source_t *source, refbuf_t *refbuf);
static int ebml_create_client_data(source_t *source, client_t *client);
static void ebml_free_client_data(client_t *client);

static ebml_t *ebml_create();
static void ebml_destroy(ebml_t *ebml);
static int ebml_read_space(ebml_t *ebml);
static int ebml_read(ebml_t *ebml, char *buffer, int len, ebml_chunk_type *chunk_type);
static unsigned char *ebml_get_write_buffer(ebml_t *ebml, int *bytes);
static int ebml_wrote(ebml_t *ebml, int len);
static int ebml_parse_tag(unsigned char      *buffer,
                          unsigned char      *buffer_end,
                          unsigned long long *payload_length);
static int ebml_parse_var_int(unsigned char      *buffer,
                              unsigned char      *buffer_end,
                              unsigned long long *out_value);

int format_ebml_get_plugin(source_t *source)
{

    ebml_source_state_t *ebml_source_state = calloc(1, sizeof(ebml_source_state_t));
    format_plugin_t *plugin = calloc(1, sizeof(format_plugin_t));

    plugin->get_buffer = ebml_get_buffer;
    plugin->write_buf_to_client = ebml_write_buf_to_client;
    plugin->create_client_data = ebml_create_client_data;
    plugin->free_plugin = ebml_free_plugin;
    plugin->write_buf_to_file = ebml_write_buf_to_file;
    plugin->set_tag = NULL;
    plugin->apply_settings = NULL;

    plugin->contenttype = httpp_getvar(source->parser, "content-type");

    plugin->_state = ebml_source_state;
    vorbis_comment_init(&plugin->vc);
    source->format = plugin;

    ebml_source_state->ebml = ebml_create();
    
    return 0;
}

static void ebml_free_plugin(format_plugin_t *plugin)
{

    ebml_source_state_t *ebml_source_state = plugin->_state;

    refbuf_release(ebml_source_state->header);
    ebml_destroy(ebml_source_state->ebml);
    free(ebml_source_state);
    vorbis_comment_clear(&plugin->vc);
    free(plugin);
}

/* Write to a client from the header buffer.
 */
static int send_ebml_header(client_t *client)
{

    ebml_client_data_t *ebml_client_data = client->format_data;
    int len = EBML_SLICE_SIZE;
    int ret;

    if (ebml_client_data->header->len - ebml_client_data->header_pos < len)
    {
        len = ebml_client_data->header->len - ebml_client_data->header_pos;
    }
    ret = client_send_bytes (client,
                             ebml_client_data->header->data + ebml_client_data->header_pos,
                             len);

    if (ret > 0)
    {
        ebml_client_data->header_pos += ret;
    }

    return ret;

}

/* Initial write-to-client function.
 */
static int ebml_write_buf_to_client (client_t *client)
{

    ebml_client_data_t *ebml_client_data = client->format_data;

    if (ebml_client_data->header_pos != ebml_client_data->header->len)
    {
        return send_ebml_header (client);
    }
    else
    {
        /* Now that the header's sent, short-circuit to the generic
         * write-refbufs function. */
        client->write_to_client = format_generic_write_to_client;
        return client->write_to_client(client);
    }

}

/* Return a refbuf to add to the queue.
 */
static refbuf_t *ebml_get_buffer(source_t *source)
{

    ebml_source_state_t *ebml_source_state = source->format->_state;
    format_plugin_t *format = source->format;
    unsigned char *write_buffer = NULL;
    int read_bytes = 0;
    int write_bytes = 0;
    ebml_chunk_type chunk_type;
    refbuf_t *refbuf;
    int ret;

    while (1)
    {
        read_bytes = ebml_read_space(ebml_source_state->ebml);
        if (read_bytes > 0) {
            /* A chunk is available for reading */
            refbuf = refbuf_new(read_bytes);
            ebml_read(ebml_source_state->ebml, refbuf->data, read_bytes, &chunk_type);

            if (ebml_source_state->header == NULL)
            {
                /* Capture header before adding clusters to the queue */
                ebml_source_state->header = refbuf;
                continue;
            }

/*            ICECAST_LOG_DEBUG("EBML: generated refbuf, size %i : %hhi %hhi %hhi",
 *                            read_bytes, refbuf->data[0], refbuf->data[1], refbuf->data[2]);
 */
                              
            if (chunk_type == EBML_CHUNK_CLUSTER_START)
            {
                refbuf->sync_point = 1;
/*                ICECAST_LOG_DEBUG("EBML: ^ was sync point"); */
            }
            return refbuf;

        } else if(read_bytes == 0) {
            /* Feed more bytes into the parser */
            write_buffer = ebml_get_write_buffer(ebml_source_state->ebml, &write_bytes);
            read_bytes = client_read_bytes (source->client, write_buffer, write_bytes);
            if (read_bytes <= 0) {
                ebml_wrote (ebml_source_state->ebml, 0);
                return NULL;
            }
            format->read_bytes += read_bytes;
            ret = ebml_wrote (ebml_source_state->ebml, read_bytes);
            if (ret != read_bytes) {
                ICECAST_LOG_ERROR("Problem processing stream");
                source->running = 0;
                return NULL;
            }
        } else {
            ICECAST_LOG_ERROR("Problem processing stream");
            source->running = 0;
            return NULL;
        }
    }
}

/* Initialize client state.
 */
static int ebml_create_client_data(source_t *source, client_t *client)
{
    ebml_client_data_t *ebml_client_data;
    ebml_source_state_t *ebml_source_state = source->format->_state;

    if (!ebml_source_state->header)
        return -1;

    ebml_client_data = calloc(1, sizeof(ebml_client_data_t));
    if (!ebml_client_data)
        return -1;

    ebml_client_data->header = ebml_source_state->header;
    refbuf_addref(ebml_client_data->header);
    client->format_data = ebml_client_data;
    client->free_client_data = ebml_free_client_data;
    return 0;
}


static void ebml_free_client_data (client_t *client)
{

    ebml_client_data_t *ebml_client_data = client->format_data;

    refbuf_release (ebml_client_data->header);
    free (client->format_data);
    client->format_data = NULL;
}


static void ebml_write_buf_to_file_fail (source_t *source)
{
    ICECAST_LOG_WARN("Write to dump file failed, disabling");
    fclose (source->dumpfile);
    source->dumpfile = NULL;
}


static void ebml_write_buf_to_file (source_t *source, refbuf_t *refbuf)
{

    ebml_source_state_t *ebml_source_state = source->format->_state;

    if (ebml_source_state->file_headers_written == 0)
    {
        if (fwrite (ebml_source_state->header->data, 1,
                    ebml_source_state->header->len,
                    source->dumpfile) != ebml_source_state->header->len)
            ebml_write_buf_to_file_fail(source);
        else
            ebml_source_state->file_headers_written = 1;
    }

    if (fwrite (refbuf->data, 1, refbuf->len, source->dumpfile) != refbuf->len)
    {
        ebml_write_buf_to_file_fail(source);
    }

}


/* internal ebml parsing */

static void ebml_destroy(ebml_t *ebml)
{

    free(ebml->header);
    free(ebml->input_buffer);
    free(ebml->buffer);
    free(ebml);

}

static ebml_t *ebml_create()
{

    ebml_t *ebml = calloc(1, sizeof(ebml_t));

    ebml->output_state = EBML_STATE_READING_HEADER;

    ebml->header = calloc(1, EBML_HEADER_MAX_SIZE);
    ebml->buffer = calloc(1, EBML_SLICE_SIZE);
    ebml->input_buffer = calloc(1, EBML_SLICE_SIZE);

    ebml->cluster_start = -1;

    return ebml;

}

/* Return the size of a buffer needed to store the next
 * chunk that ebml_read can yield.
 */
static int ebml_read_space(ebml_t *ebml)
{

    int read_space;

    switch (ebml->output_state) {
        case EBML_STATE_READING_HEADER:
        
            if (ebml->header_size != 0) {
                /* The header can be read */
                return ebml->header_size;
            } else {
                /* The header's not ready yet */
                return 0;
            }
            break;
            
        case EBML_STATE_READING_CLUSTERS:
            
            if (ebml->cluster_start > 0) {
                /* return up until just before a new cluster starts */
                read_space = ebml->cluster_start;
            } else {
                /* return what we have */
                read_space = ebml->position;
            }

            return read_space;
    }
    
    ICECAST_LOG_ERROR("EBML: Invalid parser read state");
    return 0;
}

/* Return a chunk of the EBML/MKV/WebM stream.
 * The header will be buffered until it can be returned as one chunk.
 * A cluster element's opening tag will always start a new chunk.
 * 
 * chunk_type will be set to indicate if the chunk is the header,
 * the start of a cluster, or continuing the current cluster.
 */
static int ebml_read(ebml_t *ebml, char *buffer, int len, ebml_chunk_type *chunk_type)
{

    int read_space;
    int to_read;
    
    *chunk_type = EBML_CHUNK_HEADER;

    if (len < 1) {
        return 0;
    }

    switch (ebml->output_state) {
        case EBML_STATE_READING_HEADER:
        
            if (ebml->header_size != 0)
            {
                /* Can read a chunk of the header */
                read_space = ebml->header_size - ebml->header_read_position;

                if (read_space >= len) {
                    to_read = len;
                } else {
                    to_read = read_space;
                }

                memcpy(buffer, ebml->header, to_read);
                ebml->header_read_position += to_read;
                
                *chunk_type = EBML_CHUNK_HEADER;
                
                if (ebml->header_read_position == ebml->header_size) {
                    ebml->output_state = EBML_STATE_READING_CLUSTERS;
                }
            } else {
                /* The header's not ready yet */
                return 0;
            }
        
            break;
            
        case EBML_STATE_READING_CLUSTERS:
        
            *chunk_type = EBML_CHUNK_CLUSTER_CONTINUE;
            read_space = ebml->position;
            
            if (ebml->cluster_start == 0) {
                /* new cluster is starting now */
                *chunk_type = EBML_CHUNK_CLUSTER_START;
                
                /* mark end of cluster */
                ebml->cluster_start = -1;
            } else if (ebml->cluster_start > 0) {
                /* return up until just before a new cluster starts */
                read_space = ebml->cluster_start;
            }

            if (read_space < 1) {
                return 0;
            }

            if (read_space >= len ) {
                to_read = len;
            } else {
                to_read = read_space;
            }

            memcpy(buffer, ebml->buffer, to_read);
            
            /* Shift unread data down to the start of the buffer */
            memmove(ebml->buffer, ebml->buffer + to_read, ebml->position - to_read);
            ebml->position -= to_read;

            if (ebml->cluster_start > 0) {
                ebml->cluster_start -= to_read;
            }
        
            break;
    }

    return to_read;

}

/* Get pointer & length of the buffer able to accept input.
 * 
 * Returns the start of the writable space;
 * Sets bytes to the amount of space available.
 */
static unsigned char *ebml_get_write_buffer(ebml_t *ebml, int *bytes)
{
    *bytes = EBML_SLICE_SIZE - ebml->input_position;
    return ebml->input_buffer + ebml->input_position;
}

/* Process data that has been written to the EBML parser's input buffer.
 */
static int ebml_wrote(ebml_t *ebml, int len)
{
    int processing = 1;
    int cursor = 0;
    int to_copy;
    unsigned char *end_of_buffer;

    int tag_length;
    unsigned long long payload_length;
    int copy_state;

    char *segment_id = "\x18\x53\x80\x67";
    char *cluster_id = "\x1F\x43\xB6\x75";
    
    ebml->input_position += len;
    end_of_buffer = ebml->input_buffer + ebml->input_position;
    
    while (processing) {
        
        /*ICECAST_LOG_DEBUG("Parse State: %i", ebml->parse_state);*/
        
        switch (ebml->parse_state) {
            
            case EBML_STATE_PARSING_HEADER:
            case EBML_STATE_PARSING_CLUSTERS:
                
                if (ebml->parse_state == EBML_STATE_PARSING_HEADER) {
                    copy_state = EBML_STATE_COPYING_TO_HEADER;
                } else {
                    copy_state = EBML_STATE_COPYING_TO_DATA;
                }
                
                tag_length = ebml_parse_tag(ebml->input_buffer + cursor,
                                            end_of_buffer, &payload_length);
                
                if (tag_length > 0) {
                    
                    if (payload_length == EBML_UNKNOWN) {
                        /* Parse all children for tags we can't skip */
                        payload_length = 0;
                    }
                    
                    if (!memcmp(ebml->input_buffer + cursor, cluster_id, 4)) {
                        /* Found a cluster */
                        ebml->parse_state = EBML_STATE_START_CLUSTER;
                        
                    } else if (!memcmp(ebml->input_buffer + cursor, segment_id, 4)) {
                        /* Segment tag, copy tag to buffer but parse children */
                        ebml->copy_len = tag_length;
                        ebml->parse_state = copy_state;
                        
                    } else {
                        /* Non-cluster tag, copy it & children into buffer */
                        ebml->copy_len = tag_length + payload_length;
                        ebml->parse_state = copy_state;
                        
                    }

                } else if (tag_length == 0) {
                    /* Wait for more data */
                    /* ICECAST_LOG_DEBUG("Wait"); */
                    processing = 0;
                } else if (tag_length < 0) {
                    /* Parse error */
                    /* ICECAST_LOG_DEBUG("Stop"); */
                    return -1;
                }
                break;
            
            case EBML_STATE_START_CLUSTER:
                /* found a cluster; wait to process it until
                 * any previous cluster tag has been flushed
                 * from the read buffer, so as to not lose the
                 * sync point.
                 */
                if (ebml->cluster_start >= 0) {
                    processing = 0;
                } else {
                    
                    tag_length = ebml_parse_tag(ebml->input_buffer + cursor,
                                                end_of_buffer, &payload_length);
                    
                    /* The header has been fully read by now, publish its size. */
                    ebml->header_size = ebml->header_position;
                    
                    /* Mark this sync point */
                    ebml->cluster_start = ebml->position;
                    
                    /* Copy cluster tag to read buffer */
                    ebml->copy_len = tag_length;
                    ebml->parse_state = EBML_STATE_COPYING_TO_DATA;
                }
                break;
            
            case EBML_STATE_COPYING_TO_HEADER:
            case EBML_STATE_COPYING_TO_DATA:
                to_copy = ebml->input_position - cursor;
                if (to_copy > ebml->copy_len) {
                    to_copy = ebml->copy_len;
                }
                
                if (ebml->parse_state == EBML_STATE_COPYING_TO_HEADER) {
                    if ((ebml->header_position + to_copy) > EBML_HEADER_MAX_SIZE) {
                        ICECAST_LOG_ERROR("EBML Header too large, failing");
                        return -1;
                    }
                    
                    memcpy(ebml->header + ebml->header_position, ebml->input_buffer + cursor, to_copy);
                    ebml->header_position += to_copy;
                    
                } else if (ebml->parse_state == EBML_STATE_COPYING_TO_DATA) {
                    if ((ebml->position + to_copy) > EBML_SLICE_SIZE) {
                        to_copy = EBML_SLICE_SIZE - ebml->position;
                    }
                    
                    memcpy(ebml->buffer + ebml->position, ebml->input_buffer + cursor, to_copy);
                    ebml->position += to_copy;
                }
                /* ICECAST_LOG_DEBUG("Copied %i of %hhu", to_copy, ebml->copy_len); */
                
                cursor += to_copy;
                ebml->copy_len -= to_copy;
                
                if (ebml->copy_len == 0) {
                    /* resume parsing */
                    if (ebml->parse_state == EBML_STATE_COPYING_TO_HEADER) {
                        ebml->parse_state = EBML_STATE_PARSING_HEADER;
                    } else {
                        ebml->parse_state = EBML_STATE_PARSING_CLUSTERS;
                    }
                } else {
                    /* wait for more data */
                    processing = 0;
                }
                
                break;
            
            default:
                processing = 0;
            
        }
    
    }
    
    /* Shift unprocessed data down to the start of the buffer */
    memmove(ebml->input_buffer, ebml->input_buffer + cursor, ebml->input_position - cursor);
    ebml->input_position -= cursor;
    
    return len;
    
}

/* Try to parse an EBML tag at the given location, returning the
 * length of the tag & the length of the associated payload.
 * 
 * Returns the length of the tag on success, and writes the payload
 * size to *payload_length.
 * 
 * Return 0 if it would be necessary to read past the
 * given end-of-buffer address to read a complete tag.
 * 
 * Returns -1 if the tag is corrupt.
 */

static int ebml_parse_tag(unsigned char *buffer,
                          unsigned char *buffer_end,
                          unsigned long long *payload_length)
{
    int type_length;
    int size_length;
    unsigned long long value;
    
    *payload_length = 0;
    
    /* read past the type tag */
    type_length = ebml_parse_var_int(buffer, buffer_end, &value);
    
    if (type_length <= 0) {
        return type_length;
    }
    
    /* read the length tag */
    size_length = ebml_parse_var_int(buffer + type_length, buffer_end, payload_length);
    
    if (size_length <= 0) {
        return size_length;
    }
    
    return type_length + size_length;
}

/* Try to parse an EBML variable-length integer.
 * Returns 0 if there's not enough space to read the number;
 * Returns -1 if the number is malformed.
 * Else, returns the length of the number in bytes and writes the
 * value to *out_value.
 */
static int ebml_parse_var_int(unsigned char *buffer,
                              unsigned char *buffer_end,
                              unsigned long long *out_value)
{
    int size = 1;
    int i;
    unsigned char mask = 0x80;
    unsigned long long value;
    unsigned long long unknown_marker;
    
    if (buffer >= buffer_end) {
        return 0;
    }
    
    /* find the length marker bit in the first byte */
    value = buffer[0];
    
    while (mask) {
        if (value & mask) {
            value = value & ~mask;
            unknown_marker = mask - 1;
            break;
        }
        size++;
        mask = mask >> 1;
    }
    
    /* catch malformed number (no prefix) */
    if (mask == 0) {
        ICECAST_LOG_DEBUG("Corrupt var-int");
        return -1;
    }
    
    /* catch number bigger than parsing buffer */
    if (buffer + size - 1 >= buffer_end) {
        return 0;
    }
    
    /* read remaining bytes of (big-endian) number */
    for (i = 1; i < size; i++) {
        value = (value << 8) + buffer[i];
        unknown_marker = (unknown_marker << 8) + 0xFF;
    }
    
    /* catch special "unknown" length */
    
    if (value == unknown_marker) {
        *out_value = EBML_UNKNOWN;
    } else {
        *out_value = value;
    }

/*
    ICECAST_LOG_DEBUG("Varint: value %lli, unknown %llu, mask %hhu, size %i", value, unknown_marker, mask, size);
*/
    
    return size;
}