StreamDecoders.h

Go to the documentation of this file.

// StreamDecoders.h
// T1.1 step 1.2a: streaming transport decode primitives.
// 1.2a-i: HTTP chunked-transfer decoder (RFC 7230 sec 4.1) fed from byte
// buffers; no sockets. 1.2a-ii: SSE (text/event-stream) line parser.
// 1.2a-iii: incremental read-loop adapter (StreamByteSource + pump loops)
// feeding either decoder; SSLConnection::receiveAvailable() fits via the
// ConnectionByteSource template, tests inject mock byte sources.
 
#ifndef _STREAMDECODERS_H_
#define _STREAMDECODERS_H_
 
#include <string>
#include <vector>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <cstring>
#include "hash/crc32.h"  // vendored IEEE CRC32 (zlib polynomial) - eventstream CRCs
 
namespace cmsdk {
 
// Incremental HTTP chunked-transfer decoder.
// Feed raw bytes with feed(); decoded payload accumulates in body().
// isComplete() turns true once the terminating 0-chunk (and its final CRLF)
// has been consumed. hasError() reports malformed input (bad hex size line,
// missing CRLF after a chunk body).
class HTTPChunkedDecoder {
public:
    HTTPChunkedDecoder() : state(ST_SIZE), chunkRemaining(0), complete(false), error(false) {}
 
    // Feed count bytes; returns false if the decoder is in error state.
    bool feed(const char* data, size_t count) {
        for (size_t n = 0; n < count && !error; n++)
            feedByte(data[n]);
        return !error;
    }
    bool feed(const std::string& data) { return feed(data.data(), data.size()); }
 
    const std::string& body() const { return decoded; }
    bool isComplete() const { return complete; }
    bool hasError() const { return error; }
 
    void reset() {
        state = ST_SIZE; sizeLine.clear(); decoded.clear();
        chunkRemaining = 0; complete = false; error = false;
    }
 
private:
    enum State {
        ST_SIZE,       // reading hex size line (up to CRLF)
        ST_BODY,       // reading chunkRemaining body bytes
        ST_BODY_CRLF,  // expecting CR then LF after chunk body
        ST_BODY_LF,
        ST_TRAILER,    // after 0-chunk: consume trailer lines until blank line
        ST_DONE
    };
 
    void feedByte(char c) {
        switch (state) {
        case ST_SIZE:
            if (c == '\n') {
                // strip optional CR and any chunk extension after ';'
                size_t end = sizeLine.find(';');
                std::string hex = (end == std::string::npos) ? sizeLine : sizeLine.substr(0, end);
                while (!hex.empty() && (hex[hex.size()-1] == '\r' || hex[hex.size()-1] == ' '))
                    hex.erase(hex.size()-1);
                if (!parseHex(hex, chunkRemaining)) { error = true; return; }
                sizeLine.clear();
                if (chunkRemaining == 0) { state = ST_TRAILER; sizeLine.clear(); }
                else state = ST_BODY;
            }
            else sizeLine += c;
            break;
        case ST_BODY:
            decoded += c;
            if (--chunkRemaining == 0) state = ST_BODY_CRLF;
            break;
        case ST_BODY_CRLF:
            if (c == '\r') state = ST_BODY_LF;
            else if (c == '\n') state = ST_SIZE;  // tolerate bare LF
            else error = true;
            break;
        case ST_BODY_LF:
            if (c == '\n') state = ST_SIZE;
            else error = true;
            break;
        case ST_TRAILER:
            // consume trailer lines; a blank line ends the message
            if (c == '\n') {
                std::string line = sizeLine;
                if (!line.empty() && line[line.size()-1] == '\r') line.erase(line.size()-1);
                sizeLine.clear();
                if (line.empty()) { complete = true; state = ST_DONE; }
            }
            else sizeLine += c;
            break;
        case ST_DONE:
            break;  // ignore any bytes after completion
        }
    }
 
    static bool parseHex(const std::string& s, size_t& out) {
        if (s.empty()) return false;
        size_t v = 0;
        for (size_t n = 0; n < s.size(); n++) {
            char c = s[n]; int d;
            if (c >= '0' && c <= '9') d = c - '0';
            else if (c >= 'a' && c <= 'f') d = c - 'a' + 10;
            else if (c >= 'A' && c <= 'F') d = c - 'A' + 10;
            else return false;
            v = (v << 4) | (size_t)d;
        }
        out = v;
        return true;
    }
 
    State state;
    std::string sizeLine;   // also reused for trailer-line accumulation
    std::string decoded;
    size_t chunkRemaining;
    bool complete;
    bool error;
};
 
// Incremental SSE (Server-Sent Events, text/event-stream) line parser.
// Feed raw bytes with feed(); completed events (blank-line boundary) queue up
// and are drained with nextEvent(). Only `data:` fields are collected (an
// optional single leading space after the colon is stripped; multiple data
// lines in one event are joined with '\n'). Other fields (event:, id:,
// retry:, comments) are skipped safely. An event whose data is the literal
// sentinel "[DONE]" sets isDone() instead of being queued; bytes after that
// are ignored. hasError() is reserved for malformed input (currently only
// oversized single lines, guarding runaway buffers).
class SSEEventParser {
public:
    SSEEventParser() : haveData(false), done(false), error(false) {}
 
    // Feed count bytes; returns false if the parser is in error state.
    bool feed(const char* data, size_t count) {
        for (size_t n = 0; n < count && !error && !done; n++)
            feedByte(data[n]);
        return !error;
    }
    bool feed(const std::string& data) { return feed(data.data(), data.size()); }
 
    bool hasEvent() const { return !events.empty(); }
    // Pops the oldest completed event's data payload; returns false if none.
    bool nextEvent(std::string& out) {
        if (events.empty()) return false;
        out = events.front();
        events.erase(events.begin());
        return true;
    }
    bool isDone() const { return done; }
    bool hasError() const { return error; }
 
    void reset() {
        line.clear(); eventData.clear(); events.clear();
        haveData = false; done = false; error = false;
    }
 
private:
    static const size_t MAX_LINE = 1024 * 1024;  // 1 MB guard per line
 
    void feedByte(char c) {
        if (c == '\n') { processLine(); line.clear(); return; }
        line += c;
        if (line.size() > MAX_LINE) error = true;
    }
 
    void processLine() {
        // strip optional trailing CR (accept both CRLF and bare LF endings)
        if (!line.empty() && line[line.size()-1] == '\r') line.erase(line.size()-1);
        if (line.empty()) {  // blank line = event boundary
            if (haveData) {
                if (eventData == "[DONE]") done = true;
                else events.push_back(eventData);
            }
            eventData.clear(); haveData = false;
            return;
        }
        if (line.compare(0, 5, "data:") == 0) {
            size_t start = 5;
            if (start < line.size() && line[start] == ' ') start++;  // optional space
            if (haveData) eventData += '\n';  // spec: multiple data lines join with LF
            eventData += line.substr(start);
            haveData = true;
        }
        // else: event:/id:/retry:/comment lines are skipped safely
    }
 
    std::string line;       // current (possibly partial) line being assembled
    std::string eventData;  // accumulated data for the in-progress event
    bool haveData;
    std::vector<std::string> events;
    bool done;
    bool error;
};
 
// ---------------------------------------------------------------------------
// 1.2a-iii: incremental read-loop adapter.
// Abstracted byte source so tests can inject canned/staged buffers and the
// real transport is a thin wrapper over SSLConnection::receiveAvailable().
// ---------------------------------------------------------------------------
 
// Result of one read attempt from a byte source.
enum StreamReadResult {
    SRC_DATA,    // got >= 1 byte
    SRC_AGAIN,   // no data right now (would-block / timeout); retry later
    SRC_CLOSED,  // orderly end of stream
    SRC_ERROR    // transport failure
};
 
// Minimal pull-style byte source interface.
class StreamByteSource {
public:
    virtual ~StreamByteSource() {}
    // Read up to maxSize bytes into buf; set got. got is only meaningful
    // when SRC_DATA is returned (and must then be >= 1).
    virtual StreamReadResult read(char* buf, size_t maxSize, size_t& got) = 0;
};
 
// Thin adapter over a CMSDK network connection (e.g. SSLConnection).
// receiveAvailable(data, size, maxSize, timeout) returns true with size >= 0
// when bytes (possibly none yet) were pulled from the connection buffer, and
// false when nothing arrived within the timeout or the buffer path failed;
// isConnected() disambiguates would-block from end-of-stream.
template<class Connection>
class ConnectionByteSource : public StreamByteSource {
public:
    ConnectionByteSource(Connection& c, unsigned int timeoutMS)
        : con(c), timeout(timeoutMS) {}
    StreamReadResult read(char* buf, size_t maxSize, size_t& got) {
        unsigned int size = 0;
        bool ok = con.receiveAvailable(buf, size, (unsigned int)maxSize, timeout);
        if (ok && size > 0) { got = size; return SRC_DATA; }
        return con.isConnected() ? SRC_AGAIN : SRC_CLOSED;
    }
private:
    Connection& con;
    unsigned int timeout;
};
 
// Outcome of a pump loop.
enum StreamPumpResult {
    PUMP_COMPLETE,  // decoder reported natural completion ([DONE] / 0-chunk)
    PUMP_CLOSED,    // source closed before completion (may be normal for SSE)
    PUMP_ERROR,     // decoder rejected input or source failed
    PUMP_LIMIT      // safety cap on iterations hit (bug guard; avoids hangs)
};
 
// Pump bytes from src into a chunked decoder until the terminating 0-chunk,
// stream end or error. Newly decoded body bytes are surfaced incrementally
// through onData (may be empty). maxIterations bounds the loop.
inline StreamPumpResult pumpChunkedStream(StreamByteSource& src,
        HTTPChunkedDecoder& dec,
        const std::function<void(const char*, size_t)>& onData,
        size_t maxIterations = 100000, size_t readSize = 4096) {
    std::vector<char> buf(readSize);
    size_t delivered = dec.body().size();
    for (size_t iter = 0; iter < maxIterations; iter++) {
        size_t got = 0;
        StreamReadResult r = src.read(&buf[0], readSize, got);
        if (r == SRC_ERROR) return PUMP_ERROR;
        if (r == SRC_CLOSED) return dec.isComplete() ? PUMP_COMPLETE : PUMP_CLOSED;
        if (r == SRC_DATA) {
            if (!dec.feed(&buf[0], got)) return PUMP_ERROR;
            const std::string& body = dec.body();
            if (onData && body.size() > delivered)
                onData(body.data() + delivered, body.size() - delivered);
            delivered = body.size();
            if (dec.isComplete()) return PUMP_COMPLETE;
        }
        // SRC_AGAIN: just retry (source owns any waiting/timeout policy)
    }
    return PUMP_LIMIT;
}
 
// Pump bytes from src into an SSE parser until [DONE], stream end or error.
// Each completed event's data payload is surfaced through onEvent.
inline StreamPumpResult pumpSSEStream(StreamByteSource& src,
        SSEEventParser& parser,
        const std::function<void(const std::string&)>& onEvent,
        size_t maxIterations = 100000, size_t readSize = 4096) {
    std::vector<char> buf(readSize);
    for (size_t iter = 0; iter < maxIterations; iter++) {
        size_t got = 0;
        StreamReadResult r = src.read(&buf[0], readSize, got);
        if (r == SRC_ERROR) return PUMP_ERROR;
        if (r == SRC_DATA && !parser.feed(&buf[0], got)) {
            // drain any events completed before the malformed input
            std::string ev;
            while (parser.nextEvent(ev)) { if (onEvent) onEvent(ev); }
            return PUMP_ERROR;
        }
        std::string ev;
        while (parser.nextEvent(ev)) { if (onEvent) onEvent(ev); }
        if (parser.isDone()) return PUMP_COMPLETE;
        if (r == SRC_CLOSED) return PUMP_CLOSED;
    }
    return PUMP_LIMIT;
}
 
// ---------------------------------------------------------------------------
// 1.2b-i: AWS eventstream (Bedrock invoke-with-response-stream) binary frame
// parser. Incremental: feed raw bytes; completed frames queue up and are
// drained with nextFrame(). Wire format per frame:
//   [4 total-length BE][4 headers-length BE][4 prelude CRC]  (12-byte prelude)
//   headers: repeated { 1 key-len, key, 1 value-type, value }
//   payload bytes, then [4 message CRC]
// 1.2b-ii: both CRC fields are validated: prelude CRC covers the first 8
// bytes; message CRC covers everything before the trailing 4 bytes. Both are
// standard IEEE CRC32 (zlib polynomial), big-endian on the wire. Extracts the ':event-type' header (string, type 7) and the
// payload for each frame. Error flag + reset(), matching sibling decoders.
class EventStreamFrameParser {
public:
    struct Frame {
        std::string eventType;  // ':event-type' header value ('' if absent)
        std::string payload;    // raw payload bytes (JSON chunk for Bedrock)
    };
 
    EventStreamFrameParser() : error(false) {}
 
    // Feed count bytes; returns false if the parser is in error state.
    bool feed(const char* data, size_t count) {
        if (error) return false;
        if (buf.size() + count > MAX_BUFFER) { error = true; return false; }
        buf.append(data, count);
        size_t guard = 0;
        while (!error && tryParseFrame()) {
            if (++guard > MAX_FRAMES_PER_FEED) { error = true; break; }
        }
        return !error;
    }
    bool feed(const std::string& data) { return feed(data.data(), data.size()); }
 
    bool hasFrame() const { return !frames.empty(); }
    // Pops the oldest completed frame; returns false if none.
    bool nextFrame(Frame& out) {
        if (frames.empty()) return false;
        out = frames.front();
        frames.erase(frames.begin());
        return true;
    }
    bool hasError() const { return error; }
 
    void reset() { buf.clear(); frames.clear(); error = false; }
 
private:
    static const size_t MAX_BUFFER = 16 * 1024 * 1024;  // 16 MB runaway guard
    static const size_t MAX_FRAMES_PER_FEED = 100000;   // hang guard
    static const size_t PRELUDE_SIZE = 12;              // 2 lengths + prelude CRC
    static const size_t MIN_FRAME = 16;                 // prelude + message CRC
 
    static uint32_t be32(const unsigned char* p) {
        return ((uint32_t)p[0] << 24) | ((uint32_t)p[1] << 16) |
               ((uint32_t)p[2] << 8)  |  (uint32_t)p[3];
    }
    static uint16_t be16(const unsigned char* p) {
        return (uint16_t)(((uint16_t)p[0] << 8) | (uint16_t)p[1]);
    }
    // True if the IEEE CRC32 of [data,len) matches the 4 big-endian bytes at
    // wireCrc. hash::CRC32::getHash() already emits big-endian bytes.
    static bool crcMatches(const unsigned char* data, size_t len,
                           const unsigned char* wireCrc) {
        hash::CRC32 crc;
        crc.add(data, len);
        unsigned char computed[4];
        crc.getHash(computed);
        return memcmp(computed, wireCrc, 4) == 0;
    }
 
    // Attempts to parse one complete frame from the front of buf.
    // Returns true if a frame was consumed (parsed or skipped), false if more
    // bytes are needed. Sets error on malformed framing.
    bool tryParseFrame() {
        if (buf.size() < PRELUDE_SIZE) return false;
        const unsigned char* p = (const unsigned char*)buf.data();
        uint32_t totalLen = be32(p);
        uint32_t headersLen = be32(p + 4);
        // Prelude CRC (p+8) covers the two length fields. A mismatch means the
        // lengths themselves are untrustworthy: error out, do not consume.
        if (!crcMatches(p, 8, p + 8)) { error = true; return false; }
        if (totalLen < MIN_FRAME || totalLen > MAX_BUFFER ||
            (size_t)headersLen > (size_t)totalLen - MIN_FRAME)
            { error = true; return false; }
        if (buf.size() < totalLen) return false;  // incomplete frame; wait
 
        // Message CRC: trailing 4 bytes, over all preceding bytes of the frame.
        if (!crcMatches(p, (size_t)totalLen - 4, p + totalLen - 4))
            { error = true; return false; }
 
        Frame f;
        if (!parseHeaders(p + PRELUDE_SIZE, headersLen, f.eventType))
            { error = true; return false; }
        size_t payloadLen = (size_t)totalLen - MIN_FRAME - headersLen;
        f.payload.assign(buf.data() + PRELUDE_SIZE + headersLen, payloadLen);
        frames.push_back(f);
        buf.erase(0, totalLen);
        return true;
    }
 
    // Walks the headers block, extracting the ':event-type' string header.
    // Handles all AWS eventstream value types well enough not to desync.
    // Returns false on malformed/truncated header data.
    bool parseHeaders(const unsigned char* h, size_t len, std::string& eventType) {
        size_t pos = 0, guard = 0;
        while (pos < len) {
            if (++guard > 1000) return false;  // hang guard
            size_t keyLen = h[pos++];
            if (pos + keyLen + 1 > len) return false;
            std::string key((const char*)h + pos, keyLen);
            pos += keyLen;
            unsigned char type = h[pos++];
            size_t valLen;
            switch (type) {
                case 0: case 1: valLen = 0; break;         // bool true/false
                case 2: valLen = 1; break;                 // byte
                case 3: valLen = 2; break;                 // short
                case 4: valLen = 4; break;                 // int
                case 5: case 8: valLen = 8; break;         // long / timestamp
                case 6: case 7:                            // byte-buffer / string
                    if (pos + 2 > len) return false;
                    valLen = be16(h + pos); pos += 2;
                    break;
                case 9: valLen = 16; break;                // uuid
                default: return false;                     // unknown type: desync risk
            }
            if (pos + valLen > len) return false;
            if (type == 7 && key == ":event-type")
                eventType.assign((const char*)h + pos, valLen);
            pos += valLen;
        }
        return pos == len;
    }
 
    std::string buf;            // unconsumed raw bytes
    std::vector<Frame> frames;  // completed frames awaiting nextFrame()
    bool error;
};
 
// Pump bytes from src into an eventstream frame parser until stream end or
// error (eventstream has no in-band [DONE] sentinel at this layer; the
// caller decides completion from frame content, e.g. a message_stop event).
// Each completed frame is surfaced through onFrame. Bounded like siblings.
inline StreamPumpResult pumpEventStream(StreamByteSource& src,
        EventStreamFrameParser& parser,
        const std::function<void(const EventStreamFrameParser::Frame&)>& onFrame,
        size_t maxIterations = 100000, size_t readSize = 4096) {
    std::vector<char> buf(readSize);
    for (size_t iter = 0; iter < maxIterations; iter++) {
        size_t got = 0;
        StreamReadResult r = src.read(&buf[0], readSize, got);
        if (r == SRC_ERROR) return PUMP_ERROR;
        if (r == SRC_DATA && !parser.feed(&buf[0], got)) {
            // drain any frames completed before the malformed input
            EventStreamFrameParser::Frame f;
            while (parser.nextFrame(f)) { if (onFrame) onFrame(f); }
            return PUMP_ERROR;
        }
        EventStreamFrameParser::Frame f;
        while (parser.nextFrame(f)) { if (onFrame) onFrame(f); }
        if (r == SRC_CLOSED) return PUMP_COMPLETE;  // orderly close = end of stream
    }
    return PUMP_LIMIT;
}
 
} // namespace cmsdk
 
#endif // _STREAMDECODERS_H_