diff --git a/src/protocol.zig b/src/protocol.zig new file mode 100644 index 0000000..2acf48a --- /dev/null +++ b/src/protocol.zig @@ -0,0 +1,158 @@ +const std = @import("std"); +const encode = @import("encode.zig"); +const dbmod = @import("db.zig"); +const rowmod = @import("row.zig"); +const authmod = @import("auth.zig"); +const ttlmod = @import("ttl.zig"); +const collmod = @import("collections.zig"); +const aggregatemod = @import("aggregate.zig"); +const windowmod = @import("window.zig"); +const querymod = @import("query.zig"); + +const Allocator = std.mem.Allocator; + +/// Guard against a corrupt or hostile length prefix causing an unbounded +/// allocation before a single byte of the frame body has been validated. +pub const maxFrameBytes: u32 = 64 * 1024 * 1024; + +pub const Op = enum(u8) { + ping = 0, + get = 1, + set = 2, + del = 3, + scan = 4, + prefix = 5, + stats = 6, + flush = 7, + compact = 8, + batch = 9, + table_select = 10, + auth = 11, + user_admin = 12, + auth_token = 13, + metrics = 14, + ttl_op = 15, + coll_ttl_op = 16, + coll_op = 17, + txn_op = 18, + table_join = 19, + table_aggregate = 20, + table_explain = 21, + table_update = 22, + table_delete = 23, + table_export = 24, + table_import = 25, + table_window = 26, + subscribe = 27, + publish = 28, + unsubscribe = 29, + + pub fn fromByte(b: u8) !Op { + return std.meta.intToEnum(Op, b) catch error.UnknownOp; + } +}; + +pub const Status = enum(u8) { + ok = 0, + not_found = 1, + bad_request = 2, + io_error = 3, + conflict = 4, + internal = 5, + rate_limited = 6, + unauthorized = 7, + forbidden = 8, + + pub fn fromByte(b: u8) !Status { + return std.meta.intToEnum(Status, b) catch error.UnknownStatus; + } +}; + +pub const ProtoError = error{ + UnknownOp, + UnknownStatus, + FrameTooLarge, + Malformed, +}; + +/// A frame is the unit exchanged over the wire in both directions: a +/// single tag byte (an `Op` on the request side, a `Status` on the +/// response side) followed by a little-endian u32 payload length and the +/// payload itself. +pub const Frame = struct { + tag: u8, + payload: []u8, + allocator: Allocator, + + pub fn deinit(self: *Frame) void { + self.allocator.free(self.payload); + self.* = undefined; + } +}; + +pub fn writeFrame(writer: anytype, tag: u8, payload: []const u8) !void { + if (payload.len > maxFrameBytes) return error.FrameTooLarge; + var header: [5]u8 = undefined; + header[0] = tag; + std.mem.writeInt(u32, header[1..5], @intCast(payload.len), encode.endian); + try writer.writeAll(&header); + if (payload.len > 0) try writer.writeAll(payload); +} + +pub fn readFrame(allocator: Allocator, reader: anytype) !Frame { + var header: [5]u8 = undefined; + try reader.readNoEof(&header); + const tag = header[0]; + const len = std.mem.readInt(u32, header[1..5], encode.endian); + if (len > maxFrameBytes) return error.FrameTooLarge; + const payload = try allocator.alloc(u8, len); + errdefer allocator.free(payload); + if (len > 0) try reader.readNoEof(payload); + return .{ .tag = tag, .payload = payload, .allocator = allocator }; +} + +/// Bounds-checked cursor over a decoded payload. Every read returns +/// `error.Malformed` instead of panicking when the peer sent a short or +/// truncated buffer, since payload bytes come from an untrusted socket. +pub const Cursor = struct { + data: []const u8, + pos: usize = 0, + + pub fn takeByte(self: *Cursor) !u8 { + if (self.pos + 1 > self.data.len) return error.Malformed; + const v = self.data[self.pos]; + self.pos += 1; + return v; + } + + pub fn takeU32(self: *Cursor) !u32 { + if (self.pos + 4 > self.data.len) return error.Malformed; + const v = encode.readU32(self.data, self.pos); + self.pos += 4; + return v; + } + + pub fn takeU64(self: *Cursor) !u64 { + if (self.pos + 8 > self.data.len) return error.Malformed; + const v = encode.readU64(self.data, self.pos); + self.pos += 8; + return v; + } + + pub fn takeBytes(self: *Cursor, n: usize) ![]const u8 { + if (self.pos + n > self.data.len) return error.Malformed; + const v = self.data[self.pos .. self.pos + n]; + self.pos += n; + return v; + } + + pub fn takeLenPrefixed(self: *Cursor) ![]const u8 { + const n = try self.takeU32(); + return self.takeBytes(n); + } + + pub fn atEnd(self: *Cursor) bool { + return self.pos == self.data.len; + } +}; +