Add sort of config and FastCGI parsing stuff

3 files changed, 327 insertions, 34 deletions

diff --git a/Request.zig b/Request.zig
new file mode 100644
index 0000000..02dbbd5
--- /dev/null
+++ b/Request.zig
@@ -0,0 +1,28 @@
+const std = @import("std");
+const main = @import("main.zig");
+const alloc = main.alloc;
+const Self = @This();
+
+front: ?*main.Front,
+
+/// FastCGI requestId at the front-end side.
+front_id: u16,
+
+/// Whether we should keep the front-end connection alive after this request has been processed.
+front_keep_conn: bool,
+
+// TODO: backend pointer
+back: ?*void = null,
+
+/// List of parameters in FastCGI format.
+params: std.ArrayList(u8) = std.ArrayList(u8).init(alloc),
+
+/// Free resources associated with this request.
+/// Does nothing when 'front' or 'back' are still set, as that means this request is still referenced from either side.
+pub fn deinit(self: *Self) void {
+    if (self.front != null or self.back != null) {
+        return;
+    }
+    self.params.deinit();
+    alloc.destroy(self);
+}
diff --git a/fastcgi.zig b/fastcgi.zig
new file mode 100644
index 0000000..d923e8b
--- /dev/null
+++ b/fastcgi.zig
@@ -0,0 +1,104 @@
+const std = @import("std");
+
+pub const Record = struct {
+    version: u8,
+    type: Type,
+    request_id: u16,
+    content_length: u16,
+    padding_length: u8,
+    reserved: u8,
+    // contentData[contentLength]
+    // paddingData[paddingLength]
+
+    pub const Self = @This();
+
+    pub const Type = enum(u8) {
+        begin_request = 1,
+        abort_request = 2,
+        end_request = 3,
+        params = 4,
+        stdin = 5,
+        stdout = 6,
+        stderr = 7,
+        data = 8,
+        get_values = 9,
+        get_values_result = 10,
+        unknown_type = 11,
+    };
+
+    pub const Role = enum(u16) {
+        responder = 1,
+        authorizer = 2,
+        filter = 3,
+    };
+
+    pub const Begin = struct {
+        role: Role,
+        keep_conn: bool,
+    };
+
+    // Read the record header, must be followed by exactly one of the methods defined below.
+    pub fn read(rd: anytype) !Self {
+        // (This could be done using a packed struct and readStruct(), but with the
+        // manual approach we more easily get endian conversion and an enum check)
+        var r: Self = undefined;
+        r.version = try rd.readByte();
+        r.type = try rd.readEnum(Type, undefined);
+        r.request_id = try rd.readIntBig(u16);
+        r.content_length = try rd.readIntBig(u16);
+        r.padding_length = try rd.readByte();
+        r.reserved = try rd.readByte();
+        if (r.version != 1) {
+            return error.InvalidFastCgiVersion;
+        }
+        return r;
+    }
+
+    // Discard any content and padding.
+    pub fn skip(self: *const Self, rd: anytype) !void {
+        try rd.skipBytes(self.content_length + self.padding_length, .{});
+    }
+
+    // Read the body of a begin_request.
+    pub fn readBegin(self: *const Self, rd: anytype) !Begin {
+        if (self.content_length < 8) return error.InvalidFastCgiBeginRequestBody;
+        var b: Begin = undefined;
+        b.role = try rd.readEnum(Role, .Big);
+        b.keep_conn = (try rd.readByte()) & 1 == 1;
+        try rd.skipBytes(self.content_length + self.padding_length - 3, .{});
+        return b;
+    }
+
+    // Append the record contents to an ArrayList.
+    pub fn readArrayList(self: *const Self, rd: anytype, array_list: *std.ArrayList(u8)) !void {
+        const start = array_list.items.len;
+        try array_list.resize(start + self.content_length);
+        try rd.readNoEof(array_list.items[start..]);
+        try rd.skipBytes(self.padding_length, .{});
+    }
+};
+
+test "Record.read" {
+    const rd = std.io.fixedBufferStream("\x01\x02\x03\x04\x05\x06\x07\x08\x09").reader();
+    const r = Record.read(rd) catch unreachable;
+    const eq = std.testing.expectEqual;
+    eq(r.version, 1);
+    eq(r.type, .abort_request);
+    eq(r.requestId, (3 << 8) + 4);
+    eq(r.content_length, (5 << 8) + 6);
+    eq(r.padding_length, 7);
+    eq(r.reserved, 8);
+    eq(rd.readByte() catch unreachable, 9);
+}
+
+test "invalid Record.read" {
+    inline for (.{
+        .{ .in = "", .err = error.EndOfStream },
+        .{ .in = "\x01\x02\x03\x04\x05\x06\x07", .err = error.EndOfStream },
+        .{ .in = "\x02\x02\x03\x04\x05\x06\x07\x08", .err = error.InvalidFastCgiVersion },
+        .{ .in = "\x01\x00\x03\x04\x05\x06\x07\x08", .err = error.InvalidValue },
+    }) |t| {
+        const rd = std.io.fixedBufferStream(t.in).reader();
+        std.testing.expectEqual(Record.read(rd), t.err);
+    }
+}
diff --git a/main.zig b/main.zig
index 9d6c8e8..21268b6 100644
--- a/main.zig
+++ b/main.zig
@@ -1,67 +1,228 @@
-// https://ziglang.org/documentation/master/
-// https://gist.github.com/andrewrk/34c21bdc1600b0884a3ab9fa9aa485b8
 const std = @import("std");
+const fastcgi = @import("fastcgi.zig");
+const Request = @import("Request.zig");
+const Address = std.net.Address;
 
 pub const io_mode = .evented;
 
 var general_purpose_allocator = std.heap.GeneralPurposeAllocator(.{}){};
-const alloc = &general_purpose_allocator.allocator;
+pub const alloc = &general_purpose_allocator.allocator;
 
 const mainloop = std.event.Loop.instance.?;
 
+const Config = struct {
+    version: bool = false,
+    help: bool = false,
+    listeners: std.ArrayList(Listener) = std.ArrayList(Listener).init(alloc),
+
+    const Listener = struct {
+        addr: ?Address,
+    };
+
+    // Should be marked as noreturn, but https://github.com/ziglang/zig/issues/5728
+    fn err(comptime fmt: []const u8, arg: anytype) void {
+        _ = std.io.getStdErr().writer().print(fmt, arg) catch {};
+        std.process.exit(1);
+    }
+
+    fn read() !Config {
+        const eq = std.mem.eql;
+        var args = std.process.args();
+        var conf = Config{};
+        _ = args.nextPosix(); // Program name
+        while (args.nextPosix()) |arg| {
+            if (eq(u8, arg, "-h")) {
+                conf.help = true;
+            } else if (eq(u8, arg, "-V")) {
+                conf.version = true;
+            } else if (eq(u8, arg, "listen")) {
+                try conf.listeners.append(try readListen(&args));
+            } else {
+                err("Unrecognized option: '{s}'.\n", .{arg});
+            }
+        }
+        return conf;
+    }
+
+    fn optArg(args: *std.process.ArgIterator, name: []const u8) [:0]const u8 {
+        if (args.nextPosix()) |arg| {
+            return arg;
+        }
+        err("Option '{s}' requires an argument.\n", .{name});
+        unreachable;
+    }
+
+    fn parseTypedAddr(arg: []const u8, ltype: *[]const u8) !?Address {
+        var str: []const u8 = arg;
+        if (std.mem.indexOfScalar(u8, str, ':')) |i| {
+            ltype.* = str[0..i];
+            str = str[i + 1 ..];
+        } else {
+            ltype.* = str;
+            return null;
+        }
+        if (str.len == 0) {
+            return null;
+        }
+
+        if (str[0] == '/' or str[0] == '.') {
+            return try Address.initUnix(str);
+        }
+
+        var port: u16 = 5000;
+        if (std.mem.lastIndexOfScalar(u8, str, ':')) |i| {
+            if (std.fmt.parseUnsigned(u16, str[i + 1 ..], 10)) |p| {
+                port = p;
+                str = str[0..i];
+            } else |_| {}
+        }
+        return try Address.resolveIp(if (str.len == 0) "0.0.0.0" else str, port);
+    }
+
+    fn readListen(args: *std.process.ArgIterator) !Listener {
+        const arg = optArg(args, "listen");
+        var ltype: []const u8 = undefined;
+        if (parseTypedAddr(arg, &ltype)) |a| {
+            if (!std.mem.eql(u8, ltype, "fastcgi")) {
+                err("Unknown listerer type '{s}'.\n", .{ltype});
+            }
+            return Listener{ .addr = a };
+        } else |e| {
+            err("Error parsing address '{s}': {}\n", .{ arg, e });
+            unreachable;
+        }
+    }
+};
+
 pub fn main() !void {
-    var server = std.net.StreamServer.init(.{ .reuse_address = true });
-    defer server.deinit();
+    const conf = try Config.read();
+
+    if (conf.version) {
+        std.io.getStdOut().writeAll("fcgy 0.1\n") catch unreachable;
+        std.process.exit(0);
+    }
+    if (conf.help) {
+        std.io.getStdOut().writeAll("fcgy <options> <listen-options> <spawn-options>\n") catch unreachable;
+        std.process.exit(0);
+    }
+    if (conf.listeners.items.len == 0) {
+        Config.err("No listeners defined\n", .{});
+    }
+    if (conf.listeners.items.len == 1 and conf.listeners.items[0].addr == null) {
+        Config.err("Listening on standard I/O is not supported yet.\n", .{});
+    }
 
-    const listenAddr = std.net.Address.parseIp("127.0.0.1", 5000) catch unreachable;
-    try server.listen(listenAddr);
-    std.debug.warn("Listening on {}\n", .{listenAddr});
+    for (conf.listeners.items) |l| {
+        var server = std.net.StreamServer.init(.{ .reuse_address = true });
+        if (l.addr) |a| {
+            try server.listen(l.addr.?);
+            std.debug.print("Listening on {}\n", .{l.addr});
+        } else {
+            Config.err("Cannot mix standard I/O and server-type listeners.\n", .{});
+        }
+        try mainloop.runDetached(alloc, listener, .{ &l, server });
+    }
+    mainloop.run();
+}
 
+fn listener(conf: *const Config.Listener, server_: std.net.StreamServer) void {
+    var server = server_;
+    defer server.deinit();
     while (true) {
-        const con = try server.accept();
-        const client = try alloc.create(clientNode);
-        client.* = .{
-            .data = Client{
+        const con = server.accept() catch unreachable;
+        const front = alloc.create(FrontNode) catch unreachable;
+        front.* = .{
+            .data = Front{
                 .addr = con.address,
                 .stream = con.stream,
             },
         };
-        clientList.append(client);
-        try mainloop.runDetached(alloc, Client.run, .{&client.data});
+        front_list.append(front);
+        mainloop.runDetached(alloc, Front.run, .{&front.data}) catch unreachable;
     }
 }
 
-const clientNode = std.TailQueue(Client).Node;
-var clientList: std.TailQueue(Client) = .{};
+// XXX: Do we even need to keep track of open Front connections?
+const FrontNode = std.TailQueue(Front).Node;
+var front_list: std.TailQueue(Front) = .{};
 
-const Client = struct {
-    addr: std.net.Address,
+pub const Front = struct {
+    addr: Address,
     stream: std.net.Stream,
+    requests: std.AutoHashMap(u16, *Request) = std.AutoHashMap(u16, *Request).init(alloc),
 
-    fn run(self: *Client) void {
-        std.debug.warn("Connection from {}\n", .{self.addr});
+    fn run(self: *Front) void {
+        std.debug.print("Connection from {}\n", .{self.addr});
         self.loop() catch |e| {
-            std.debug.warn("Connection from {} closed: {}\n", .{ self.addr, e });
+            std.debug.print("Connection from {} closed: {}\n", .{ self.addr, e });
         };
-        self.deinit();
+
+        const node = @fieldParentPtr(FrontNode, "data", self);
+        front_list.remove(node);
+        self.stream.close();
+        alloc.destroy(node);
     }
 
-    fn loop(self: *Client) !void {
-        var buf: [4096]u8 = undefined;
+    fn loop(self: *Front) !void {
+        var rd = std.io.bufferedReader(self.stream.reader()).reader();
         while (true) {
-            const n = try self.stream.read(&buf);
-            if (n == 0) {
-                return error.ConnectionClosed;
+            const r = try fastcgi.Record.read(rd);
+            std.debug.print("Received FastCGI record: {}\n", .{r});
+            switch (r.type) {
+                .begin_request => try self.handleBegin(r.request_id, try r.readBegin(rd)),
+                .params => try self.handleParams(&r, rd),
+                else => try r.skip(rd),
             }
-            try self.stream.writer().writeAll(buf[0..n]);
         }
     }
 
-    /// Closes the connection (if not closed already) and removes the connection from the clientList.
-    fn deinit(self: *Client) void {
-        const node = @fieldParentPtr(clientNode, "data", self);
-        clientList.remove(node);
-        self.stream.close();
-        alloc.destroy(node);
+    fn handleBegin(self: *Front, id: u16, b: fastcgi.Record.Begin) !void {
+        // TODO: Handle other role types (which servers support that, though?)
+        if (b.role != .responder) return error.unsupportedFastCgiRole;
+        std.debug.print("Begin request: {}\n", .{b});
+        var req = try alloc.create(Request);
+        req.* = .{
+            .front = self,
+            .front_id = id,
+            .front_keep_conn = b.keep_conn,
+        };
+        errdefer req.deinit();
+        try self.requests.putNoClobber(id, req);
+    }
+
+    fn handleParams(self: *Front, r: *const fastcgi.Record, rd: anytype) !void {
+        const req = self.requests.get(r.request_id) orelse return;
+        // We don't assign a back-end until we have all params, so this shouldn't happen.
+        if (req.back != null) return error.invalidState;
+        if (r.content_length > 0) return r.readArrayList(rd, &req.params);
+        // TODO: We received all params now, assign a backend and process the request
+        try r.skip(rd);
     }
 };
+
+test "parse typed address" {
+    const T = struct { in: []const u8, lt: []const u8, a: ?[]const u8 };
+    for ([_]T{
+        .{ .in = "fastcgi", .lt = "fastcgi", .a = null },
+        .{ .in = "fastcgi:", .lt = "fastcgi", .a = null },
+        .{ .in = "fastcgi:/", .lt = "fastcgi", .a = "/" },
+        .{ .in = "fastcgi:./some/path", .lt = "fastcgi", .a = "./some/path" },
+        .{ .in = "fastcgi::300", .lt = "fastcgi", .a = "0.0.0.0:300" },
+        .{ .in = "fastcgi:127.0.0.2", .lt = "fastcgi", .a = "127.0.0.2:5000" },
+        .{ .in = "fastcgi::::10", .lt = "fastcgi", .a = "[::]:10" },
+    }) |t| {
+        var ltype: []const u8 = undefined;
+        const addr = Config.parseTypedAddr(t.in, &ltype) catch unreachable;
+        std.testing.expectEqualStrings(t.lt, ltype);
+        if (t.a) |a| {
+            var buf: [128]u8 = undefined;
+            var naddr = std.fmt.bufPrint(&buf, "{}", .{addr.?}) catch unreachable;
+            if (std.mem.indexOfScalar(u8, naddr, 0)) |n| {
+                naddr = naddr[0..n]; // Formatting UNIX addresses may include zero bytes, remove those.
+            }
+            std.testing.expectEqualStrings(a, naddr);
+        } else {
+            std.testing.expect(addr == null);
+        }
+    }
+}