diff options
Diffstat (limited to 'src/9p_io.zig')
| -rw-r--r-- | src/9p_io.zig | 1454 |
1 files changed, 1454 insertions, 0 deletions
diff --git a/src/9p_io.zig b/src/9p_io.zig new file mode 100644 index 00000000..89611c2b --- /dev/null +++ b/src/9p_io.zig @@ -0,0 +1,1454 @@ +const std = @import("std"); +const libc = std.c; +const builtin = @import("builtin"); +const ninep = @import("9p.zig"); +const pardes = @import("pardes.zig"); +const limits = @import("memory.zig").limits; +pub const quic_enabled = @import("9p_options").quic; +const quic = if (quic_enabled) @import("9p_quic.zig") else struct {}; + +const log = std.log.scoped(.ninep); + +pub const darwin = switch (builtin.os.tag) { + .macos, .ios, .tvos, .watchos, .visionos => true, + else => false, +}; +pub const supported = builtin.os.tag == .linux or darwin; +pub const sun_path_len = @typeInfo(@FieldType(libc.sockaddr.un, "path")).array.len; + +pub fn setCloexec(fd: c_int) void { + _ = libc.fcntl(fd, libc.F.SETFD, @as(c_int, 1)); +} + +pub fn socketDir(buf: *[sun_path_len:0]u8) ?[:0]const u8 { + if (libc.getenv("XDG_RUNTIME_DIR")) |path| + return std.fmt.bufPrintSentinel(buf, "{s}", .{std.mem.span(path)}, 0) catch null; + const home = libc.getenv("HOME") orelse return null; + return std.fmt.bufPrintSentinel(buf, "{s}/.local/state/pardes", .{std.mem.span(home)}, 0) catch null; +} + +pub const FileFacts = struct { mode: u32, uid: libc.uid_t }; + +pub fn statNoFollow(path: [:0]const u8) ?FileFacts { + if (comptime darwin) { + var stat: libc.Stat = undefined; + if (libc.fstatat(libc.AT.FDCWD, path, &stat, libc.AT.SYMLINK_NOFOLLOW) != 0) return null; + return .{ .mode = stat.mode, .uid = stat.uid }; + } else { + const linux = std.os.linux; + var stat: linux.Statx = undefined; + const fields: linux.STATX = .{ .TYPE = true, .MODE = true, .UID = true }; + if (libc.statx(linux.AT.FDCWD, path, linux.AT.SYMLINK_NOFOLLOW, fields, &stat) != 0) return null; + return .{ .mode = stat.mode, .uid = stat.uid }; + } +} + +pub fn ensureSocketDir(dir: [:0]const u8) bool { + if (dir.len == 0) return false; + var partial: [sun_path_len:0]u8 = undefined; + @memcpy(partial[0 .. dir.len + 1], dir[0 .. dir.len + 1]); + for (1..dir.len) |i| { + if (dir[i] != '/') continue; + partial[i] = 0; + _ = libc.mkdir(partial[0..i :0], 0o700); + partial[i] = '/'; + } + _ = libc.mkdir(dir, 0o700); + const stat = statNoFollow(dir) orelse return false; + return stat.mode & 0o170000 == 0o040000 and stat.uid == libc.getuid() and stat.mode & 0o077 == 0; +} + +const prefix = "pardes-9p-"; + +pub const msize: u32 = 8192; + +pub const max_conns = 4; + +extern "c" fn inet_pton(family: c_int, src: [*:0]const u8, dst: *anyopaque) c_int; + +pub fn networkAddress(dial: []const u8, allow_zero_port: bool) error{BadDial}!std.Io.net.IpAddress { + if (comptime !supported) return error.BadDial; + const host_start: usize = if (std.mem.startsWith(u8, dial, "tcp!")) 4 else if (std.mem.startsWith(u8, dial, "quic!")) 5 else return error.BadDial; + const split = std.mem.lastIndexOfScalar(u8, dial, '!') orelse return error.BadDial; + if (split <= host_start or split + 1 == dial.len) return error.BadDial; + const port_text = dial[split + 1 ..]; + for (port_text) |c| if (c < '0' or c > '9') return error.BadDial; + const port = std.fmt.parseInt(u16, port_text, 10) catch return error.BadDial; + if (port == 0 and !allow_zero_port) return error.BadDial; + const host = dial[host_start..split]; + if (std.mem.indexOfScalar(u8, host, 0) != null) return error.BadDial; + var host_buf: [46]u8 = undefined; + const host_z = std.fmt.bufPrintSentinel(&host_buf, "{s}", .{host}, 0) catch return error.BadDial; + var ip4: std.Io.net.Ip4Address = .{ .port = port, .bytes = undefined }; + if (inet_pton(libc.AF.INET, host_z, &ip4.bytes) == 1) return .{ .ip4 = ip4 }; + var ip6: std.Io.net.Ip6Address = .{ .port = port, .bytes = undefined }; + if (inet_pton(libc.AF.INET6, host_z, &ip6.bytes) == 1) return canonicalIp(.{ .ip6 = ip6 }); + return error.BadDial; +} + +fn canonicalIp(address: std.Io.net.IpAddress) std.Io.net.IpAddress { + if (address == .ip6) { + if (std.Io.net.Ip4Address.fromIp6(address.ip6)) |ip4| return .{ .ip4 = ip4 }; + } + return address; +} + +fn ipSockaddr(address: std.Io.net.IpAddress, out: *libc.sockaddr.storage) libc.socklen_t { + return switch (address) { + .ip4 => |ip| blk: { + const addr: *libc.sockaddr.in = @ptrCast(out); + addr.* = .{ .port = std.mem.nativeToBig(u16, ip.port), .addr = @bitCast(ip.bytes) }; + break :blk @sizeOf(libc.sockaddr.in); + }, + .ip6 => |ip| blk: { + const addr: *libc.sockaddr.in6 = @ptrCast(out); + addr.* = .{ .port = std.mem.nativeToBig(u16, ip.port), .addr = ip.bytes, .flowinfo = 0, .scope_id = 0 }; + break :blk @sizeOf(libc.sockaddr.in6); + }, + }; +} + +fn sockaddrIp(address: *const libc.sockaddr) ?std.Io.net.IpAddress { + return switch (address.family) { + libc.AF.INET => blk: { + const addr: *const libc.sockaddr.in = @ptrCast(@alignCast(address)); + break :blk .{ .ip4 = .{ .port = std.mem.bigToNative(u16, addr.port), .bytes = @bitCast(addr.addr) } }; + }, + libc.AF.INET6 => blk: { + const addr: *const libc.sockaddr.in6 = @ptrCast(@alignCast(address)); + break :blk canonicalIp(.{ .ip6 = .{ .port = std.mem.bigToNative(u16, addr.port), .bytes = addr.addr } }); + }, + else => null, + }; +} + +const IfAddr = extern struct { + next: ?*IfAddr, + name: ?[*:0]u8, + flags: c_uint, + address: ?*libc.sockaddr, + netmask: ?*libc.sockaddr, + destination: ?*libc.sockaddr, + data: ?*anyopaque, +}; + +extern "c" fn getifaddrs(out: *?*IfAddr) c_int; +extern "c" fn freeifaddrs(first: *IfAddr) void; + +fn localIp(address: std.Io.net.IpAddress) bool { + switch (address) { + .ip4 => |ip| if (ip.bytes[0] == 127 or std.mem.allEqual(u8, &ip.bytes, 0)) return true, + .ip6 => |ip| if (ip.isLoopBack() or std.mem.allEqual(u8, &ip.bytes, 0)) return true, + } + var first: ?*IfAddr = null; + if (getifaddrs(&first) != 0) return false; + defer if (first) |head| freeifaddrs(head); + var next = first; + while (next) |entry| : (next = entry.next) { + var local = sockaddrIp(entry.address orelse continue) orelse continue; + local.setPort(address.getPort()); + if (address.eql(&local)) return true; + } + return false; +} + +const Srv = ninep.Server(pardes.filesystem, ninep.max_fids); + +const Conn = struct { + fd: c_int = -1, + quic: if (quic_enabled) ?quic.Connection else void = if (quic_enabled) null else {}, + draining: bool = false, + accepted_ms: i64 = 0, + srv: Srv = undefined, + in: [msize]u8 = undefined, + out: [2 * msize]u8 = undefined, + + fn step(c: *Conn, core: *pardes.Pardes, req: pardes.filesystem.Req) void { + core.update(.{ .fs_req = req }); + var answered = false; + while (core.nextEffect()) |effect| { + if (effect == .fs_reply) { + const reply = effect.fs_reply; + if (reply.tag == req.tag) answered = true; + c.srv.reply(&reply, core.fsPayload(reply)); + } else core.perform(effect); + } + if (!answered) { + const reply = pardes.filesystem.Reply.fail(req.tag, pardes.filesystem.E.IO); + c.srv.reply(&reply, ""); + } + } +}; + +const accept_pause_ms: i64 = 100; + +pub const Listener = struct { + fd: c_int = -1, + tcp_fd: c_int = -1, + tcp_address: ?std.Io.net.IpAddress = null, + quic: if (quic_enabled) ?quic.Listener else void = if (quic_enabled) null else {}, + quic_address: ?std.Io.net.IpAddress = null, + paused_ms: i64 = 0, + path_buf: [sun_path_len]u8 = undefined, + path_len: usize = 0, + conns: [max_conns]Conn = @splat(.{}), + control: [2]c_int = .{ -1, -1 }, + watcher: ?std.Thread = null, + stopping: std.atomic.Value(bool) = .init(false), + watch_lock: std.atomic.Mutex = .unlocked, + watch_fds: [max_conns + 3 + @as(usize, @intFromBool(quic_enabled))]libc.pollfd = undefined, + watch_len: usize = 0, + watch_timeout: c_int = -1, + wake_ctx: ?*anyopaque = null, + wake: ?*const fn (?*anyopaque) void = null, + + pub fn path(l: *const Listener) []const u8 { + return l.path_buf[0..l.path_len]; + } + + fn listenTcp(l: *Listener, address: std.Io.net.IpAddress) !void { + var addr: libc.sockaddr.storage = undefined; + const addr_len = ipSockaddr(address, &addr); + const fd = libc.socket(addr.family, libc.SOCK.STREAM, 0); + if (fd < 0) return error.SocketFailed; + errdefer _ = libc.close(fd); + setCloexec(fd); + setNonblock(fd); + const on: c_int = 1; + if (libc.setsockopt(fd, libc.SOL.SOCKET, libc.SO.REUSEADDR, &on, @sizeOf(c_int)) != 0) + return error.SocketOptionFailed; + if (address == .ip6) { + const v6only = if (darwin) 27 else std.os.linux.IPV6.V6ONLY; + if (libc.setsockopt(fd, libc.IPPROTO.IPV6, v6only, &on, @sizeOf(c_int)) != 0) + return error.SocketOptionFailed; + } + if (libc.bind(fd, @ptrCast(&addr), addr_len) != 0) return error.BindFailed; + if (libc.listen(fd, max_conns) != 0) return error.ListenFailed; + var actual_len: libc.socklen_t = @sizeOf(libc.sockaddr.storage); + if (libc.getsockname(fd, @ptrCast(&addr), &actual_len) != 0) return error.SocketAddressFailed; + l.tcp_address = sockaddrIp(@ptrCast(&addr)) orelse return error.SocketAddressFailed; + l.tcp_fd = fd; + log.info("serving 9P2000 over TCP on {f}", .{l.tcp_address.?}); + } + + pub fn accept(l: *Listener) void { + if (comptime !supported) return; + for ([_]c_int{ l.fd, l.tcp_fd }) |listener_fd| { + if (listener_fd < 0) continue; + for (0..max_conns + 1) |_| { + const fd = libc.accept(listener_fd, null, null); + if (fd < 0) switch (libc.errno(fd)) { + .AGAIN => break, + .INTR, .CONNABORTED => continue, + else => { + l.paused_ms = nowMs() +| accept_pause_ms; + log.warn("accept failed; pausing the listener for {d} ms", .{accept_pause_ms}); + return; + }, + }; + setCloexec(fd); + setNonblock(fd); + if (listener_fd == l.tcp_fd) { + const on: c_int = 1; + _ = libc.setsockopt(fd, libc.IPPROTO.TCP, libc.TCP.NODELAY, &on, @sizeOf(c_int)); + } + if (comptime darwin) { + const on: c_int = 1; + _ = libc.setsockopt(fd, libc.SOL.SOCKET, libc.SO.NOSIGPIPE, &on, @sizeOf(c_int)); + } + const c = for (&l.conns, 0..) |*cand, i| { + if (!l.live(@intCast(i)) and !cand.draining) break cand; + } else { + log.debug("refusing a connection, all {d} slots busy", .{max_conns}); + _ = libc.close(fd); + continue; + }; + c.fd = fd; + c.draining = false; + c.accepted_ms = nowMs(); + c.srv = .init(.{ + .in = &c.in, + .out = &c.out, + .root = pardes.filesystem.namespace_root, + }); + } + } + if (comptime quic_enabled) { + if (l.quic) |*listener| for (0..max_conns + 1) |_| { + var connection = (listener.accept() catch |err| { + log.warn("QUIC accept failed: {s}", .{@errorName(err)}); + return; + }) orelse break; + const c = for (&l.conns, 0..) |*cand, i| { + if (!l.live(@intCast(i)) and !cand.draining) break cand; + } else { + connection.deinit(); + continue; + }; + c.fd = -1; + c.quic = connection; + c.draining = false; + c.accepted_ms = nowMs(); + c.srv = .init(.{ .in = &c.in, .out = &c.out, .root = pardes.filesystem.namespace_root }); + }; + } + } + + pub const greet_deadline_ms: i64 = 5000; + + pub fn expire(l: *Listener) void { + if (comptime !supported) return; + const now = nowMs(); + if (now == 0) return; + for (&l.conns, 0..) |*c, i| { + if (!l.live(@intCast(i)) or c.srv.msize != 0) continue; + if (now - c.accepted_ms < greet_deadline_ms) continue; + log.debug("slot {d} never sent Tversion; taking it back", .{i}); + l.drop(@intCast(i)); + } + } + + pub fn accepting(l: *const Listener) bool { + if (comptime !supported) return false; + if (l.fd < 0 and l.tcp_fd < 0) return false; + if (l.paused_ms == 0) return true; + const now = nowMs(); + return now == 0 or now >= l.paused_ms; + } + + pub fn nextDue(l: *const Listener) ?i32 { + if (comptime !supported) return null; + const now = nowMs(); + if (now == 0) return null; + var due: ?i64 = null; + if (l.paused_ms > now) due = l.paused_ms; + if (comptime quic_enabled) { + if (l.quic) |*listener| if (listener.nextDue()) |ms| { + const at = now + ms; + due = if (due) |d| @min(d, at) else at; + }; + } + for (&l.conns, 0..) |*c, i| { + if (!l.live(@intCast(i))) continue; + if (comptime quic_enabled) { + if (c.quic) |*connection| if (connection.pending()) return 0; + } + if (c.srv.msize != 0) continue; + const at = c.accepted_ms + greet_deadline_ms; + due = if (due) |d| @min(d, at) else at; + } + const at = due orelse return null; + return @intCast(@max(0, at - now)); + } + + fn nowMs() i64 { + var ts: libc.timespec = undefined; + if (libc.clock_gettime(.MONOTONIC, &ts) != 0) return 0; + return @as(i64, ts.sec) * std.time.ms_per_s + @divTrunc(ts.nsec, std.time.ns_per_ms); + } + + pub fn fill(l: *Listener, i: u8) void { + if (comptime !supported) return; + const c = &l.conns[i]; + if (c.srv.dead) return l.drop(i); + const room = c.srv.in.len - c.srv.in_len; + if (room == 0) return; + var buf: [msize]u8 = undefined; + const got: isize = if (quic_enabled and c.quic != null) + @intCast((c.quic.?.read(buf[0..@min(room, buf.len)]) catch return l.drop(i)) orelse return) + else + libc.read(c.fd, &buf, @min(room, buf.len)); + if (got == 0) return l.drop(i); + if (got < 0) return switch (libc.errno(got)) { + .INTR, .AGAIN => {}, + else => l.drop(i), + }; + const n = c.srv.push(buf[0..@intCast(got)]); + if (c.srv.dead) return l.drop(i); + std.debug.assert(n == @as(usize, @intCast(got))); + } + + pub fn flush(l: *Listener, i: u8) void { + if (comptime !supported) return; + const c = &l.conns[i]; + if (!l.live(i)) return; + while (true) { + const bytes = c.srv.output(); + if (bytes.len == 0) return; + const n: isize = if (quic_enabled and c.quic != null) + @intCast(c.quic.?.write(bytes) catch return l.drop(i)) + else + libc.send(c.fd, bytes.ptr, bytes.len, nosignal); + if (n < 0) switch (libc.errno(n)) { + .INTR => continue, + .AGAIN => return, + else => return l.drop(i), + }; + if (n == 0) return; + c.srv.wrote(@intCast(n)); + } + } + + pub fn owes(l: *const Listener, i: u8) bool { + return l.conns[i].srv.output().len != 0; + } + + pub fn live(l: *const Listener, i: u8) bool { + return l.conns[i].fd >= 0 or (quic_enabled and l.conns[i].quic != null); + } + + pub fn drop(l: *Listener, i: u8) void { + const c = &l.conns[i]; + if (c.fd >= 0) { + _ = libc.close(c.fd); + c.fd = -1; + } + if (comptime quic_enabled) { + if (c.quic) |*connection| { + connection.deinit(); + c.quic = null; + } + } + if (c.draining or c.accepted_ms == 0) return; + c.srv.hangup(); + c.draining = true; + } + + pub const Drained = struct { count: usize = 0, pending: bool = false }; + + pub fn drain(l: *Listener, core: *pardes.Pardes) Drained { + core.fs.socket_path = l.path(); + core.fs.tcp_address = l.tcp_address; + core.fs.quic_address = l.quic_address; + l.expire(); + var result: Drained = .{}; + for (&l.conns, 0..) |*conn, i| { + if (!l.live(@intCast(i)) and !conn.draining) continue; + var count: usize = 0; + while (conn.srv.retry()) |req| { + conn.step(core, req); + l.collectOs(core); + count += 1; + } + while (count < 64) { + const req = conn.srv.next() orelse break; + conn.step(core, req); + l.collectOs(core); + count += 1; + } + result.count += count; + result.pending = result.pending or count >= 64; + if (conn.draining) { + if (count == 0) conn.draining = false else result.pending = true; + } else l.flush(@intCast(i)); + if (comptime quic_enabled) { + if (conn.quic) |*connection| result.pending = result.pending or connection.pending(); + } + } + return result; + } + + pub fn tick(l: *Listener, core: *pardes.Pardes) Drained { + if (comptime quic_enabled) { + if (l.quic) |*listener| listener.events() catch |err| { + log.warn("QUIC listener stopped: {s}", .{@errorName(err)}); + for (&l.conns, 0..) |*conn, i| if (conn.quic != null) l.drop(@intCast(i)); + listener.deinit(); + l.quic = null; + l.quic_address = null; + }; + } + if (l.accepting()) l.accept(); + for (0..max_conns) |i| if (l.live(@intCast(i))) l.fill(@intCast(i)); + const result = l.drain(core); + l.arm(); + return result; + } + + pub fn reset(l: *Listener, core: *pardes.Pardes) void { + for (0..max_conns) |i| l.drop(@intCast(i)); + while (l.drain(core).pending) {} + l.collectOs(core); + for (&l.conns) |*conn| conn.accepted_ms = 0; + l.arm(); + } + + fn collectOs(l: *Listener, core: *pardes.Pardes) void { + var i: usize = 0; + while (i < core.fs.os_paths.items.len) { + const entry = core.fs.os_paths.items[i]; + var held = false; + for (&l.conns, 0..) |*conn, j| { + if (!l.live(@intCast(j)) and !conn.draining) continue; + if (conn.srv.references(entry.node)) { + held = true; + break; + } + } + if (held) { + i += 1; + } else { + core.gpa.free(entry.path); + _ = core.fs.os_paths.swapRemove(i); + } + } + } + + pub fn wakeThread(l: *Listener, ctx: ?*anyopaque, wake: *const fn (?*anyopaque) void) !void { + if (l.watcher != null) return; + if (libc.pipe(&l.control) != 0) return error.PipeFailed; + errdefer { + _ = libc.close(l.control[0]); + _ = libc.close(l.control[1]); + l.control = .{ -1, -1 }; + } + for (l.control) |fd| { + setCloexec(fd); + setNonblock(fd); + } + l.wake_ctx = ctx; + l.wake = wake; + l.arm(); + l.watcher = try std.Thread.spawn(.{}, watch, .{l}); + } + + fn arm(l: *Listener) void { + if (l.control[1] < 0) return; + while (!l.watch_lock.tryLock()) std.atomic.spinLoopHint(); + l.watch_fds[0] = .{ .fd = l.control[0], .events = @intCast(libc.POLL.IN), .revents = 0 }; + l.watch_len = 1; + if (l.accepting()) { + for ([_]c_int{ l.fd, l.tcp_fd }) |fd| { + if (fd < 0) continue; + l.watch_fds[l.watch_len] = .{ .fd = fd, .events = @intCast(libc.POLL.IN), .revents = 0 }; + l.watch_len += 1; + } + } + if (comptime quic_enabled) { + if (l.quic) |*listener| { + l.watch_fds[l.watch_len] = listener.poll(); + l.watch_len += 1; + } + } + for (0..max_conns) |i| { + if (l.conns[i].fd < 0) continue; + l.watch_fds[l.watch_len] = .{ + .fd = l.conns[i].fd, + .events = @as(i16, @intCast(libc.POLL.IN)) | if (l.owes(@intCast(i))) @as(i16, @intCast(libc.POLL.OUT)) else 0, + .revents = 0, + }; + l.watch_len += 1; + } + l.watch_timeout = l.nextDue() orelse -1; + l.watch_lock.unlock(); + _ = libc.write(l.control[1], "w", 1); + } + + fn watch(l: *Listener) void { + var notified = false; + while (!l.stopping.load(.acquire)) { + var fds: [max_conns + 3 + @as(usize, @intFromBool(quic_enabled))]libc.pollfd = undefined; + while (!l.watch_lock.tryLock()) std.atomic.spinLoopHint(); + const len = if (notified) 1 else l.watch_len; + @memcpy(fds[0..len], l.watch_fds[0..len]); + const timeout = if (notified) -1 else l.watch_timeout; + l.watch_lock.unlock(); + const ready = libc.poll(&fds, @intCast(len), timeout); + if (ready < 0) continue; + if (fds[0].revents != 0) { + var buf: [64]u8 = undefined; + while (libc.read(l.control[0], &buf, buf.len) > 0) {} + notified = false; + continue; + } + if (!l.stopping.load(.acquire)) l.wake.?(l.wake_ctx); + notified = true; + } + } + + pub fn deinit(l: *Listener, gpa: std.mem.Allocator) void { + if (l.watcher) |thread| { + l.stopping.store(true, .release); + _ = libc.write(l.control[1], "q", 1); + thread.join(); + for (l.control) |fd| _ = libc.close(fd); + } + for (0..max_conns) |i| l.drop(@intCast(i)); + if (comptime quic_enabled) { + if (l.quic) |*listener| listener.deinit(); + } + if (l.tcp_fd >= 0) _ = libc.close(l.tcp_fd); + if (l.fd >= 0) { + _ = libc.close(l.fd); + l.fd = -1; + var z: [sun_path_len:0]u8 = undefined; + @memcpy(z[0..l.path_len], l.path_buf[0..l.path_len]); + z[l.path_len] = 0; + _ = libc.unlink(z[0..l.path_len :0]); + } + gpa.destroy(l); + } +}; + +pub fn socketPath(buf: *[sun_path_len]u8, dir: []const u8, name: []const u8) ?[:0]const u8 { + if (name.len == 0) return null; + if (std.mem.indexOfAny(u8, name, "/\x00") != null) return null; + return std.fmt.bufPrintSentinel(buf, "{s}/" ++ prefix ++ "{s}.sock", .{ dir, name }, 0) catch null; +} + +pub fn listen(gpa: std.mem.Allocator, named: []const u8, fallback: []const u8, tcp_dial: ?[]const u8, quic_dial: ?[]const u8) ?*Listener { + if (comptime !supported) return null; + var dir_buf: [sun_path_len:0]u8 = undefined; + const dir = socketDir(&dir_buf) orelse { + log.warn("no runtime directory for the socket", .{}); + return null; + }; + if (!ensureSocketDir(dir)) return null; + const l = gpa.create(Listener) catch return null; + l.* = .{}; + const p = socketPath(&l.path_buf, dir, if (named.len != 0) named else fallback) orelse { + gpa.destroy(l); + return null; + }; + var addr: libc.sockaddr.un = .{ .path = @splat(0) }; + @memcpy(addr.path[0 .. p.len + 1], p[0 .. p.len + 1]); + const fd = libc.socket(libc.AF.UNIX, libc.SOCK.STREAM, 0); + if (fd < 0) { + gpa.destroy(l); + return null; + } + setCloexec(fd); + if (libc.bind(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) != 0) { + const bind_error = libc.errno(-1); + const io = std.Io.Threaded.global_single_threaded.io(); + const existing = std.Io.Dir.cwd().statFile(io, p, .{ .follow_symlinks = false }) catch null; + if (bind_error != .ADDRINUSE or existing == null or existing.?.kind != .unix_domain_socket or alive(p)) { + log.warn("something is already listening on {s}", .{p}); + _ = libc.close(fd); + gpa.destroy(l); + return null; + } + if (libc.unlink(p) != 0 or libc.bind(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) != 0) { + _ = libc.close(fd); + gpa.destroy(l); + return null; + } + } + if (libc.chmod(p, 0o600) != 0 or libc.listen(fd, max_conns) != 0) { + _ = libc.close(fd); + _ = libc.unlink(p); + gpa.destroy(l); + return null; + } + setNonblock(fd); + l.fd = fd; + l.path_len = p.len; + if (tcp_dial) |dial| { + if (!std.mem.startsWith(u8, dial, "tcp!")) { + l.deinit(gpa); + return null; + } + const address = networkAddress(dial, true) catch { + log.warn("invalid TCP address {s}", .{dial}); + l.deinit(gpa); + return null; + }; + l.listenTcp(address) catch |err| { + log.warn("cannot listen on {s}: {s}", .{ dial, @errorName(err) }); + l.deinit(gpa); + return null; + }; + } + if (quic_dial) |dial| { + if (comptime quic_enabled) { + if (!std.mem.startsWith(u8, dial, "quic!")) { + l.deinit(gpa); + return null; + } + const address = networkAddress(dial, true) catch { + log.warn("invalid QUIC address {s}", .{dial}); + l.deinit(gpa); + return null; + }; + l.quic = quic.Listener.init(address) catch |err| { + log.warn("cannot listen on {s}: {s}", .{ dial, @errorName(err) }); + l.deinit(gpa); + return null; + }; + l.quic_address = l.quic.?.address; + log.info("serving 9P2000 over QUIC on {f}", .{l.quic_address.?}); + } else { + log.warn("QUIC is unavailable in this build", .{}); + l.deinit(gpa); + return null; + } + } + log.info("serving 9P2000 on {s}", .{p}); + return l; +} + +fn alive(path: [:0]const u8) bool { + var addr: libc.sockaddr.un = .{ .path = @splat(0) }; + if (path.len + 1 > sun_path_len) return true; // cannot ask; assume occupied + @memcpy(addr.path[0 .. path.len + 1], path[0 .. path.len + 1]); + const fd = libc.socket(libc.AF.UNIX, libc.SOCK.STREAM, 0); + if (fd < 0) return true; + defer _ = libc.close(fd); + setCloexec(fd); + setNonblock(fd); + if (libc.connect(fd, @ptrCast(&addr), @sizeOf(@TypeOf(addr))) == 0) return true; + return libc.errno(-1) != .CONNREFUSED; +} + +fn setNonblock(fd: c_int) void { + const flags = libc.fcntl(fd, libc.F.GETFL, @as(c_int, 0)); + if (flags < 0) return; + var o: libc.O = @bitCast(@as(u32, @bitCast(flags))); + o.NONBLOCK = true; + _ = libc.fcntl(fd, libc.F.SETFL, @as(c_int, @bitCast(@as(u32, @bitCast(o))))); +} + +const nosignal: u32 = if (darwin) 0 else libc.MSG.NOSIGNAL; + +const testing = std.testing; + +test "the socket name is a third prefix in the shared directory" { + var buf: [sun_path_len]u8 = undefined; + const p = socketPath(&buf, "/run/user/1000", "t9srv").?; + try testing.expectEqualStrings("/run/user/1000/pardes-9p-t9srv.sock", p); + try testing.expect(!std.mem.startsWith(u8, std.fs.path.basename(p), "pardes-detached-")); +} + +test "a name that is not one path component is no address at all" { + var buf: [sun_path_len]u8 = undefined; + try testing.expect(socketPath(&buf, "/run", "") == null); + try testing.expect(socketPath(&buf, "/run", "a/b") == null); + try testing.expect(socketPath(&buf, "/run", "a\x00b") == null); +} + +test "one connection's buffers are sized from the one msize constant" { + try testing.expect(msize >= ninep.min_msize); + const c: Conn = .{}; + try testing.expectEqual(@as(usize, msize), c.in.len); + try testing.expectEqual(@as(usize, 2 * msize), c.out.len); +} + +test "TCP addresses are numeric and normalize mapped IPv4" { + const loopback = try networkAddress("tcp!127.0.0.1!5640", false); + try testing.expectEqualDeep(loopback, try networkAddress("tcp!::ffff:127.0.0.1!5640", false)); + try testing.expectEqualDeep(loopback, try networkAddress("tcp!::ffff:7f00:1!5640", false)); + try testing.expectEqualDeep(try networkAddress("tcp!::1!5640", false), try networkAddress("tcp!0:0:0:0:0:0:0:1!5640", false)); + try testing.expectEqual(@as(u16, 0), (try networkAddress("tcp!127.0.0.1!0", true)).getPort()); + for ([_][]const u8{ "tcp!localhost!5640", "tcp!127.0.0.1!0", "tcp!127.0.0.1!-1", "tcp!127.0.0.1!65536", "tcp!127.0.0.1!", "tcp!!5640" }) |dial| + try testing.expectError(error.BadDial, networkAddress(dial, false)); + try Client.validateDial("tcp!127.0.0.1!5640"); + try Client.validateDial("/tmp/pardes-owned.sock"); + try Client.validateDial("unix!/tmp/pardes-owned.sock"); + try testing.expectError(error.BadDial, Client.validateDial("unix!work")); + try testing.expectError(error.BadDial, Client.validateDial("unix!")); + try testing.expectError(error.BadDial, Client.validateDial("unix!/tmp/a\x00b")); + try testing.expectError(error.BadDial, Client.validateDial("tcp!localhost!5640")); + try testing.expectError(error.BadDial, Client.validateDial("/tmp/a\x00b")); + if (quic_enabled) { + try Client.validateDial("quic!127.0.0.1!5640"); + } else try testing.expectError(error.QuicUnavailable, Client.validateDial("quic!127.0.0.1!5640")); +} + +test "same-session TCP mounts compare canonical endpoints and local wildcard destinations" { + if (comptime !supported) return error.SkipZigTest; + const Case = struct { bound: []const u8, dial: []const u8, same: bool }; + for ([_]Case{ + .{ .bound = "tcp!127.0.0.1!5640", .dial = "tcp!::ffff:127.0.0.1!5640", .same = true }, + .{ .bound = "tcp!::ffff:127.0.0.1!5640", .dial = "tcp!127.0.0.1!5640", .same = true }, + .{ .bound = "tcp!::1!5640", .dial = "tcp!0:0:0:0:0:0:0:1!5640", .same = true }, + .{ .bound = "tcp!127.0.0.1!5640", .dial = "tcp!0.0.0.0!5640", .same = true }, + .{ .bound = "tcp!::1!5640", .dial = "tcp!::!5640", .same = true }, + .{ .bound = "tcp!0.0.0.0!5640", .dial = "tcp!127.0.0.2!5640", .same = true }, + .{ .bound = "tcp!::!5640", .dial = "tcp!::1!5640", .same = true }, + .{ .bound = "tcp!127.0.0.1!5640", .dial = "tcp!127.0.0.1!5641", .same = false }, + .{ .bound = "tcp!0.0.0.0!5640", .dial = "tcp!192.0.2.1!5640", .same = false }, + .{ .bound = "tcp!::!5640", .dial = "tcp!2001:db8::1!5640", .same = false }, + .{ .bound = "tcp!::!5640", .dial = "tcp!127.0.0.1!5640", .same = false }, + }) |c| try testing.expectEqual(c.same, Client.sameSession(c.dial, "", try networkAddress(c.bound, true), null)); + try testing.expect(Client.sameSession("/tmp/pardes-owned.sock", "/tmp/pardes-owned.sock", null, null)); + try testing.expect(Client.sameSession("unix!/tmp/pardes-owned.sock", "/tmp/pardes-owned.sock", null, null)); + try testing.expect(!Client.sameSession("tcp!127.0.0.1!5640", "/tmp/pardes-owned.sock", null, null)); + if (quic_enabled) { + const endpoint = try networkAddress("quic!127.0.0.1!5640", false); + try testing.expect(Client.sameSession("quic!::ffff:127.0.0.1!5640", "", null, endpoint)); + try testing.expect(!Client.sameSession("tcp!127.0.0.1!5640", "", null, endpoint)); + try testing.expect(!Client.sameSession("quic!127.0.0.1!5640", "", endpoint, null)); + } +} + +extern "c" fn mkdtemp(template: [*:0]u8) ?[*:0]u8; +extern "c" fn rmdir(path: [*:0]const u8) c_int; + +test "Unix TCP and QUIC share one listener through reads writes reconnects and reset" { + if (comptime !supported) return error.SkipZigTest; + const gpa = testing.allocator; + var directory: [64:0]u8 = undefined; + _ = try std.fmt.bufPrintSentinel(&directory, "/tmp/pardes-tcp-XXXXXX", .{}, 0); + if (mkdtemp(&directory) == null) return error.TempDirectoryFailed; + defer _ = rmdir(&directory); + const old_runtime = if (libc.getenv("XDG_RUNTIME_DIR")) |v| try gpa.dupeZ(u8, std.mem.span(v)) else null; + defer { + if (old_runtime) |v| { + _ = setenv("XDG_RUNTIME_DIR", v, 1); + gpa.free(v); + } else _ = unsetenv("XDG_RUNTIME_DIR"); + } + try testing.expectEqual(@as(c_int, 0), setenv("XDG_RUNTIME_DIR", &directory, 1)); + const replacement = try gpa.alloc(u8, 3 * msize + 27); + defer gpa.free(replacement); + @memset(replacement, 'x'); + @memcpy(replacement[0.."changed café λ\n".len], "changed café λ\n"); + replacement[replacement.len - 1] = '\n'; + + const Worker = struct { + dial: []const u8, + body_path: []const u8, + expected: []const u8, + replacement: []const u8, + done: std.atomic.Value(bool) = .init(false), + failure: ?anyerror = null, + + fn run(w: *@This()) void { + defer w.done.store(true, .release); + w.check() catch |err| { + w.failure = err; + }; + } + + fn check(w: *@This()) !void { + const before = try Client.readLimit(testing.allocator, w.dial, w.body_path, w.body_path, w.expected.len); + defer testing.allocator.free(before); + try testing.expectEqualStrings(w.expected, before); + try testing.expectError(error.FileTooLarge, Client.readLimit(testing.allocator, w.dial, w.body_path, w.body_path, w.expected.len - 1)); + const listing = try Client.readLimit(testing.allocator, w.dial, "/self/pane", "/self/pane", 128); + defer testing.allocator.free(listing); + const exact_listing = try Client.readLimit(testing.allocator, w.dial, "/self/pane", "/self/pane", listing.len); + defer testing.allocator.free(exact_listing); + try testing.expectEqualStrings(listing, exact_listing); + try testing.expectError(error.FileTooLarge, Client.readLimit(testing.allocator, w.dial, "/self/pane", "/self/pane", listing.len - 1)); + try Client.write(testing.allocator, w.dial, w.body_path, w.replacement); + const after = try Client.read(testing.allocator, w.dial, w.body_path, w.body_path); + defer testing.allocator.free(after); + try testing.expectEqualStrings(w.replacement, after); + const screen = try Client.read(testing.allocator, w.dial, "/self/screen", "/self/screen"); + defer testing.allocator.free(screen); + const parsed = try std.json.parseFromSlice(struct { cols: u16, rows: u16 }, testing.allocator, screen, .{ .ignore_unknown_fields = true }); + defer parsed.deinit(); + try testing.expectEqual(@as(u16, 40), parsed.value.cols); + try testing.expectEqual(@as(u16, 12), parsed.value.rows); + } + }; + + const protocols: []const []const u8 = if (quic_enabled) &.{ "tcp", "quic" } else &.{"tcp"}; + for (protocols) |protocol| for ([_][]const u8{ "127.0.0.1", "::1" }) |host| { + const p = try pardes.Pardes.init(gpa, .{ .tty_only = true, .cols = 40, .rows = 12 }); + defer p.deinit(); + const pane = try p.setTestFile("initial\n"); + while (p.nextEffect()) |_| {} + var bind_buf: [64]u8 = undefined; + const bind = try std.fmt.bufPrint(&bind_buf, "{s}!{s}!0", .{ protocol, host }); + const tcp = std.mem.eql(u8, protocol, "tcp"); + const l = listen(gpa, "roundtrip", "", if (tcp) bind else null, if (tcp) null else bind) orelse return error.ListenFailed; + defer { + l.reset(p); + l.deinit(gpa); + } + try testing.expect(l.fd >= 0); + try testing.expectEqual(tcp, l.tcp_fd >= 0); + try testing.expectEqual(@as(usize, 4), l.conns.len); + try testing.expect(l.watcher == null); + const port = (if (tcp) l.tcp_address else l.quic_address).?.getPort(); + try testing.expect(port != 0); + var dial_buf: [64]u8 = undefined; + const network_dial = try std.fmt.bufPrint(&dial_buf, "{s}!{s}!{d}", .{ protocol, host, port }); + var body_buf: [64]u8 = undefined; + const body = try std.fmt.bufPrint(&body_buf, "/self/pane/{d}/body", .{pane.serial}); + for ([_][]const u8{ network_dial, l.path(), network_dial }, 0..) |dial, attempt| { + var worker: Worker = .{ .dial = dial, .body_path = body, .expected = if (attempt == 0) "initial\n" else replacement, .replacement = replacement }; + const thread = try std.Thread.spawn(.{}, Worker.run, .{&worker}); + defer thread.join(); + const deadline = Client.nowMs() + 3 * Client.budget_ms; + while (!worker.done.load(.acquire) and Client.nowMs() < deadline) { + _ = l.tick(p); + Client.nap(1); + } + try testing.expect(worker.done.load(.acquire)); + if (worker.failure) |err| return err; + const unix_fd = l.fd; + const tcp_fd = l.tcp_fd; + l.reset(p); + try testing.expectEqual(unix_fd, l.fd); + try testing.expectEqual(tcp_fd, l.tcp_fd); + for (&l.conns, 0..) |conn, i| try testing.expect(!l.live(@intCast(i)) and !conn.draining); + for (p.fs.snapshots) |snapshot| try testing.expect(snapshot.node == 0); + } + }; +} + +extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int; +extern "c" fn unsetenv(name: [*:0]const u8) c_int; + +pub fn start(gpa: std.mem.Allocator, core: *pardes.Pardes) ?*Listener { + var name: [16]u8 = undefined; + const fallback = std.fmt.bufPrint(&name, "{d}", .{@as(u32, @intCast(libc.getpid()))}) catch unreachable; + const listener = listen(gpa, core.opts.ninep_name, fallback, core.opts.ninep_tcp, core.opts.ninep_quic) orelse { + core.reportError(0, "9p listener", error.ListenFailed); + return null; + }; + core.fs.socket_path = listener.path(); + core.fs.tcp_address = listener.tcp_address; + core.fs.quic_address = listener.quic_address; + return listener; +} + +pub fn exportPaneEnv(listener: ?*const Listener, serial: u32, forward_look: bool) void { + _ = unsetenv("PARDES_FORWARD_LOOK"); + if (listener) |l| exporting: { + var sock: [sun_path_len]u8 = undefined; + const path = std.fmt.bufPrintSentinel(&sock, "{s}", .{l.path()}, 0) catch break :exporting; + var buf: [16]u8 = undefined; + const id = std.fmt.bufPrintSentinel(&buf, "{d}", .{serial}, 0) catch break :exporting; + if (setenv("PARDES_9P", path, 1) != 0) break :exporting; + if (setenv("PARDES_PANE", id, 1) != 0) break :exporting; + if (setenv("PARDES_FORWARD_LOOK", if (forward_look) "1" else "0", 1) == 0) return; + } + _ = unsetenv("PARDES_9P"); + _ = unsetenv("PARDES_PANE"); + _ = setenv("PARDES_FORWARD_LOOK", "0", 1); +} + +test "9P shell environment preserves identity when nested Look forwarding is disabled" { + const names = [_][*:0]const u8{ "XDG_RUNTIME_DIR", "HOME", "PARDES_9P", "PARDES_PANE", "PARDES_FORWARD_LOOK" }; + var saved: [names.len]?[:0]u8 = @splat(null); + for (names, &saved) |name, *value| { + if (libc.getenv(name)) |old| value.* = try testing.allocator.dupeZ(u8, std.mem.span(old)); + } + defer for (names, saved) |name, value| { + if (value) |old| { + _ = setenv(name, old, 1); + testing.allocator.free(old); + } else _ = unsetenv(name); + }; + + var dir: [sun_path_len:0]u8 = undefined; + _ = setenv("XDG_RUNTIME_DIR", "/run/user/1000", 1); + try testing.expectEqualStrings("/run/user/1000", socketDir(&dir).?); + _ = unsetenv("XDG_RUNTIME_DIR"); + _ = setenv("HOME", "/home/example", 1); + try testing.expectEqualStrings("/home/example/.local/state/pardes", socketDir(&dir).?); + _ = unsetenv("HOME"); + try testing.expect(socketDir(&dir) == null); + + const listener = try testing.allocator.create(Listener); + defer testing.allocator.destroy(listener); + listener.* = .{}; + const path = "/tmp/pardes-example.sock"; + @memcpy(listener.path_buf[0..path.len], path); + listener.path_len = path.len; + exportPaneEnv(listener, 7, false); + try testing.expectEqualStrings(path, std.mem.span(libc.getenv("PARDES_9P").?)); + try testing.expectEqualStrings("7", std.mem.span(libc.getenv("PARDES_PANE").?)); + try testing.expectEqualStrings("0", std.mem.span(libc.getenv("PARDES_FORWARD_LOOK").?)); + exportPaneEnv(listener, 8, true); + try testing.expectEqualStrings("8", std.mem.span(libc.getenv("PARDES_PANE").?)); + try testing.expectEqualStrings("1", std.mem.span(libc.getenv("PARDES_FORWARD_LOOK").?)); + exportPaneEnv(null, 0, false); + try testing.expect(libc.getenv("PARDES_9P") == null); + try testing.expect(libc.getenv("PARDES_PANE") == null); + try testing.expectEqualStrings("0", std.mem.span(libc.getenv("PARDES_FORWARD_LOOK").?)); +} + +pub const Client = struct { + pub const budget_ms: i64 = 2000; + + pub const max_depth: usize = 2 * ninep.max_welem; + + const uname = "pardes"; + + const Dial = union(enum) { unix: [:0]const u8, tcp: std.Io.net.IpAddress, quic: std.Io.net.IpAddress }; + + pub const Error = error{ + PathTooDeep, + BadDial, + Dial, + Hangup, + Timeout, + Botch, + Remote, + IsDirectory, + NotFound, + FileTooLarge, + QuicUnavailable, + }; + + pub fn read(gpa: std.mem.Allocator, dial: []const u8, path: []const u8, display_path: []const u8) ![]u8 { + return readLimit(gpa, dial, path, display_path, limits.max_file_bytes); + } + + pub fn readLimit(gpa: std.mem.Allocator, dial: []const u8, path: []const u8, display_path: []const u8, max_bytes: usize) ![]u8 { + if (comptime !supported) return error.Unsupported; + var names: [max_depth][]const u8 = undefined; + const n = try elements(path, &names); + var sock_buf: [sun_path_len]u8 = undefined; + const sock = try resolve(&sock_buf, dial); + var remote: RemoteError = .{}; + return fetchBytes(gpa, sock, names[0..n], &remote, null, display_path, @min(max_bytes, limits.max_file_bytes)); + } + + pub fn write(gpa: std.mem.Allocator, dial: []const u8, path: []const u8, bytes: []const u8) !void { + if (comptime !supported) return error.Unsupported; + var names: [max_depth][]const u8 = undefined; + const n = try elements(path, &names); + var sock_buf: [sun_path_len]u8 = undefined; + const sock = try resolve(&sock_buf, dial); + var remote: RemoteError = .{}; + const result = try fetchBytes(gpa, sock, names[0..n], &remote, bytes, path, limits.max_file_bytes); + gpa.free(result); + } + + const RemoteError = struct { + buf: [ninep.errmax]u8 = undefined, + len: usize = 0, + + fn set(r: *RemoteError, msg: []const u8) error{Remote} { + r.len = @min(msg.len, r.buf.len); + @memcpy(r.buf[0..r.len], msg[0..r.len]); + return error.Remote; + } + }; + + fn elements(path: []const u8, out: *[max_depth][]const u8) Error!usize { + var n: usize = 0; + var it = std.mem.tokenizeScalar(u8, path, '/'); + while (it.next()) |name| { + if (n == out.len) return Error.PathTooDeep; + out[n] = name; + n += 1; + } + return n; + } + + fn resolve(buf: *[sun_path_len]u8, dial: []const u8) error{ BadDial, QuicUnavailable }!Dial { + if (dial.len == 0) return error.BadDial; + if (std.mem.startsWith(u8, dial, "tcp!")) return .{ .tcp = try networkAddress(dial, false) }; + if (std.mem.startsWith(u8, dial, "quic!")) { + if (comptime !quic_enabled) return error.QuicUnavailable; + return .{ .quic = try networkAddress(dial, false) }; + } + const explicit_unix = std.mem.startsWith(u8, dial, "unix!"); + const path = if (explicit_unix) dial[5..] else dial; + if (explicit_unix and !std.mem.startsWith(u8, path, "/")) return error.BadDial; + if (std.mem.indexOfScalar(u8, path, '/') != null) { + if (std.mem.indexOfScalar(u8, path, 0) != null) return error.BadDial; + return .{ .unix = std.fmt.bufPrintSentinel(buf, "{s}", .{path}, 0) catch return error.BadDial }; + } + var dir_buf: [sun_path_len:0]u8 = undefined; + const dir = socketDir(&dir_buf) orelse return error.BadDial; + return .{ .unix = socketPath(buf, dir, dial) orelse return error.BadDial }; + } + + pub fn validateDial(dial: []const u8) error{ BadDial, QuicUnavailable }!void { + var buf: [sun_path_len]u8 = undefined; + _ = try resolve(&buf, dial); + } + + pub fn sameSession(dial: []const u8, socket_path: []const u8, tcp_address: ?std.Io.net.IpAddress, quic_address: ?std.Io.net.IpAddress) bool { + if (comptime !supported) return false; + var buf: [sun_path_len]u8 = undefined; + const address = resolve(&buf, dial) catch return false; + switch (address) { + .unix => |path| return socket_path.len != 0 and std.mem.eql(u8, path, socket_path), + .tcp, .quic => |destination| { + var ip = destination; + switch (ip) { + .ip4 => |v4| if (std.mem.allEqual(u8, &v4.bytes, 0)) { + ip = .{ .ip4 = .loopback(v4.port) }; + }, + .ip6 => |v6| if (std.mem.allEqual(u8, &v6.bytes, 0)) { + ip = .{ .ip6 = .loopback(v6.port) }; + }, + } + const bound = canonicalIp((if (address == .tcp) tcp_address else quic_address) orelse return false); + if (ip.getPort() != bound.getPort() or @as(std.Io.net.IpAddress.Family, ip) != @as(std.Io.net.IpAddress.Family, bound)) return false; + if (ip.eql(&bound)) return true; + const wildcard = switch (bound) { + .ip4 => |v4| std.mem.allEqual(u8, &v4.bytes, 0), + .ip6 => |v6| std.mem.allEqual(u8, &v6.bytes, 0), + }; + if (wildcard) return localIp(ip); + return false; + }, + } + } + + const Session = struct { + fd: c_int, + quic: if (quic_enabled) ?quic.Connection else void = if (quic_enabled) null else {}, + deadline: i64, + display_path: []const u8, + cl: ninep.Client = undefined, + in: [msize]u8 = undefined, + out: [msize]u8 = undefined, + stage: [msize]u8 = undefined, + + fn wait(s: *Session, events: i16) Error!void { + while (true) { + const left = s.deadline - nowMs(); + if (left <= 0) return Error.Timeout; + if (comptime quic_enabled) { + if (s.quic) |*connection| { + var fds = [1]libc.pollfd{connection.poll().?}; + const timeout = @min(left, connection.nextDue() orelse budget_ms); + const ready = libc.poll(&fds, 1, @intCast(timeout)); + if (ready < 0) { + if (libc.errno(ready) == .INTR) continue; + return Error.Hangup; + } + if (nowMs() >= s.deadline) return Error.Timeout; + if (fds[0].revents & @as(i16, @intCast(libc.POLL.NVAL)) != 0) return Error.Hangup; + connection.events() catch return Error.Hangup; + return; + } + } + var fds = [1]libc.pollfd{.{ .fd = s.fd, .events = events, .revents = 0 }}; + const ready = libc.poll(&fds, 1, @intCast(@min(left, budget_ms))); + if (ready < 0) { + if (libc.errno(ready) == .INTR) continue; + return Error.Hangup; + } + if (ready == 0 or nowMs() >= s.deadline) return Error.Timeout; + if (fds[0].revents & events != 0) return; + return Error.Hangup; + } + } + + fn flush(s: *Session) Error!void { + while (s.cl.output().len != 0) { + if (nowMs() >= s.deadline) return Error.Timeout; + const bytes = s.cl.output(); + if (comptime quic_enabled) { + if (s.quic) |*connection| { + const sent = connection.write(bytes) catch return Error.Hangup; + if (sent == 0) { + try s.wait(poll_out); + continue; + } + s.cl.wrote(sent); + continue; + } + } + try s.wait(poll_out); + const sent = libc.send(s.fd, bytes.ptr, bytes.len, nosignal); + if (sent < 0) switch (libc.errno(sent)) { + .INTR, .AGAIN => continue, + else => return Error.Hangup, + }; + if (sent == 0) return Error.Hangup; + s.cl.wrote(@intCast(sent)); + } + } + + fn settle(s: *Session) Error!ninep.Client.Done { + while (true) { + if (nowMs() >= s.deadline) return Error.Timeout; + try s.flush(); + if (s.cl.take()) |done| return done; + if (s.cl.dead) return Error.Botch; + const room = s.cl.in.len - s.cl.in_len; + if (room == 0) return Error.Botch; + if (comptime quic_enabled) { + if (s.quic) |*connection| { + const got = (connection.read(s.stage[0..@min(room, s.stage.len)]) catch return Error.Hangup) orelse { + try s.wait(poll_in); + continue; + }; + if (got == 0) return Error.Hangup; + const n = s.cl.push(s.stage[0..got]); + std.debug.assert(n == got); + continue; + } + } + try s.wait(poll_in); + const got = libc.read(s.fd, &s.stage, @min(room, s.stage.len)); + if (got == 0) return Error.Hangup; + if (got < 0) switch (libc.errno(got)) { + .INTR, .AGAIN => continue, + else => return Error.Hangup, + }; + const n = s.cl.push(s.stage[0..@intCast(got)]); + std.debug.assert(n == @as(usize, @intCast(got))); + } + } + + fn ask(s: *Session, req: ninep.Client.Request, remote: *RemoteError) Error!ninep.Client.Result { + _ = s.cl.submit(req) catch return Error.Botch; + const done = try s.settle(); + if (done.result == .fail) return remote.set(done.result.fail); + if (std.mem.eql(u8, @tagName(done.result), @tagName(std.meta.activeTag(req)))) return done.result; + return Error.Botch; + } + + fn drop(s: *Session, fid: u32) void { + _ = s.cl.submit(.{ .clunk = .{ .fid = fid } }) catch return; + _ = s.settle() catch {}; + } + + fn dropNoWait(s: *Session, fid: u32) void { + _ = s.cl.submit(.{ .clunk = .{ .fid = fid } }) catch return; + s.flush() catch {}; + } + }; + + fn transact( + s: *Session, + names: []const []const u8, + out: *std.Io.Writer.Allocating, + remote: *RemoteError, + write_bytes: ?[]const u8, + read_limit: usize, + ) !void { + _ = try s.ask(.{ .version = .{} }, remote); + if (s.cl.msize == 0) return Error.Botch; + + const root: u32 = 0; + var here = (try s.ask(.{ .attach = .{ .fid = root, .uname = uname } }, remote)).attach; + var cur: u32 = root; + var next: u32 = 1; + + var i: usize = 0; + while (i < names.len) { + const n = @min(ninep.max_welem, names.len - i); + const w = (try s.ask(.{ .walk = .{ + .fid = cur, + .newfid = next, + .names = names[i..][0..n], + } }, remote)).walk; + if (w.nwqid != n) return Error.NotFound; + here = w.wqid[n - 1]; + if (cur != root) s.drop(cur); + cur = next; + next = if (next == 1) 2 else 1; + i += n; + } + defer s.dropNoWait(cur); + + const directory = here.type & ninep.qtdir != 0; + + const mode: u8 = if (write_bytes != null) ninep.owrite else ninep.oread; + const truncate = write_bytes != null and names.len > 0 and + (std.mem.eql(u8, names[0], "os") or std.mem.eql(u8, names[names.len - 1], "body")); + _ = try s.ask(.{ .open = .{ .fid = cur, .mode = mode | if (truncate) ninep.otrunc else 0 } }, remote); + if (write_bytes) |bytes| { + if (directory) return Error.IsDirectory; + var written: usize = 0; + while (written < bytes.len) { + const chunk = bytes[written..][0..@min(bytes.len - written, s.cl.maxWrite())]; + const count = (try s.ask(.{ .write = .{ .fid = cur, .offset = written, .data = chunk } }, remote)).write; + if (count == 0 or count > chunk.len) return Error.Botch; + written += count; + } + return; + } + + const max_bytes: u64 = @min(read_limit, @as(usize, if (directory) limits.max_stream_bytes else limits.max_file_bytes)); + const wire_limit: u64 = if (directory) limits.max_stream_bytes else max_bytes; + var off: u64 = 0; + while (true) { + const want: u32 = @intCast(@min(@as(u64, s.cl.maxRead()), wire_limit + 1 - off)); + const data = (try s.ask(.{ .read = .{ .fid = cur, .offset = off, .count = want } }, remote)).read; + if (data.len == 0) return; + if (off + data.len > wire_limit) return Error.FileTooLarge; + if (directory) { + var pos: usize = 0; + while (pos < data.len) { + if (data.len - pos < 2) return Error.Botch; + const len: usize = 2 + @as(usize, std.mem.readInt(u16, data[pos..][0..2], .little)); + if (len > data.len - pos) return Error.Botch; + const entry = ninep.Stat.decode(data[pos..][0..len]) catch return Error.Botch; + const display_dir = std.mem.trimEnd(u8, s.display_path, "/"); + const row_len = display_dir.len + entry.name.len + 2 + @as(usize, @intFromBool(entry.qid.type & ninep.qtdir != 0)); + if (row_len > max_bytes - out.written().len) return Error.FileTooLarge; + try out.writer.print("{s}/{s}", .{ display_dir, entry.name }); + if (entry.qid.type & ninep.qtdir != 0) try out.writer.writeByte('/'); + try out.writer.writeByte('\n'); + pos += len; + } + } else try out.writer.writeAll(data); + off += data.len; + } + } + + fn fetchBytes( + gpa: std.mem.Allocator, + sock: Dial, + names: []const []const u8, + remote: *RemoteError, + write_bytes: ?[]const u8, + display_path: []const u8, + read_limit: usize, + ) ![]u8 { + if (comptime !supported) return Error.Dial; + const deadline = nowMs() +| budget_ms; + const s = try gpa.create(Session); + s.* = .{ .fd = -1, .deadline = deadline, .display_path = display_path }; + defer { + if (quic_enabled and s.quic != null) { + s.quic.?.deinit(); + } else if (s.fd >= 0) _ = libc.close(s.fd); + gpa.destroy(s); + } + if (sock == .quic) { + if (comptime quic_enabled) { + s.quic = quic.Connection.dial(sock.quic) catch return Error.Dial; + s.fd = s.quic.?.fd; + } else return Error.QuicUnavailable; + } else s.fd = try connect(sock, deadline); + s.cl = .init(.{ .in = &s.in, .out = &s.out }); + + var out: std.Io.Writer.Allocating = .init(gpa); + errdefer out.deinit(); + try transact(s, names, &out, remote, write_bytes, read_limit); + return out.toOwnedSlice(); + } + + fn connect(sock: Dial, deadline: i64) Error!c_int { + var addr: libc.sockaddr.storage = undefined; + const addr_len: libc.socklen_t = switch (sock) { + .unix => |path| blk: { + if (path.len + 1 > sun_path_len) return Error.BadDial; + const un: *libc.sockaddr.un = @ptrCast(&addr); + un.* = .{ .path = @splat(0) }; + @memcpy(un.path[0 .. path.len + 1], path[0 .. path.len + 1]); + break :blk @sizeOf(libc.sockaddr.un); + }, + .tcp => |ip| ipSockaddr(ip, &addr), + .quic => return Error.QuicUnavailable, + }; + const fd = libc.socket(addr.family, libc.SOCK.STREAM, 0); + if (fd < 0) return Error.Dial; + setCloexec(fd); + setNonblock(fd); + errdefer _ = libc.close(fd); + if (sock == .tcp) { + const on: c_int = 1; + _ = libc.setsockopt(fd, libc.IPPROTO.TCP, libc.TCP.NODELAY, &on, @sizeOf(c_int)); + } + while (true) { + if (libc.connect(fd, @ptrCast(&addr), addr_len) == 0) break; + switch (libc._errno().*) { + @intFromEnum(libc.E.AGAIN), @intFromEnum(libc.E.INTR) => { + if (nowMs() >= deadline) return Error.Dial; + nap(2); + }, + @intFromEnum(libc.E.INPROGRESS), @intFromEnum(libc.E.ALREADY) => { + const left = deadline - nowMs(); + if (left <= 0) return Error.Dial; + var pfd: [1]libc.pollfd = .{.{ .fd = fd, .events = poll_out, .revents = 0 }}; + if (libc.poll(&pfd, 1, @intCast(@min(left, 1000))) <= 0) continue; + var err: c_int = 0; + var len: libc.socklen_t = @sizeOf(c_int); + if (libc.getsockopt(fd, libc.SOL.SOCKET, libc.SO.ERROR, @ptrCast(&err), &len) != 0) + return Error.Dial; + if (err == 0) break; + return Error.Dial; + }, + @intFromEnum(libc.E.ISCONN) => break, + else => return Error.Dial, + } + } + if (comptime darwin) { + const on: c_int = 1; + _ = libc.setsockopt(fd, libc.SOL.SOCKET, libc.SO.NOSIGPIPE, &on, @sizeOf(c_int)); + } + return fd; + } + + fn nowMs() i64 { + var ts: libc.timespec = undefined; + if (libc.clock_gettime(.MONOTONIC, &ts) != 0) return std.math.maxInt(i64); + return @as(i64, ts.sec) * std.time.ms_per_s + @divTrunc(ts.nsec, std.time.ns_per_ms); + } + + const poll_in: i16 = @intCast(libc.POLL.IN); + const poll_out: i16 = @intCast(libc.POLL.OUT); + + fn nap(ms: c_int) void { + _ = libc.poll(&[0]libc.pollfd{}, 0, ms); + } + + test "a path becomes walk elements, normalised the way a shell would" { + var out: [max_depth][]const u8 = undefined; + try testing.expectEqual(@as(usize, 4), try elements("/self/pane/1/body", &out)); + try testing.expectEqualStrings("self", out[0]); + try testing.expectEqualStrings("pane", out[1]); + try testing.expectEqualStrings("1", out[2]); + try testing.expectEqualStrings("body", out[3]); + + try testing.expectEqual(@as(usize, 4), try elements("self/pane/1/body", &out)); + try testing.expectEqual(@as(usize, 4), try elements("//self//pane//1//body//", &out)); + try testing.expectEqual(@as(usize, 2), try elements("/self/index", &out)); + + try testing.expectEqual(@as(usize, 0), try elements("/", &out)); + + var deep: [8 * max_depth]u8 = @splat('/'); + for (0..max_depth + 1) |i| deep[i * 2 + 1] = 'a'; + try testing.expectError(Error.PathTooDeep, elements(deep[0 .. (max_depth + 1) * 2], &out)); + } + + test "a bare dial resolves to the socket --9p binds, and a path is taken as given" { + if (comptime !supported) return error.SkipZigTest; + var buf: [sun_path_len]u8 = undefined; + + const named = (try resolve(&buf, "work")).unix; + try testing.expect(std.mem.endsWith(u8, named, "/pardes-9p-work.sock")); + var expect: [sun_path_len]u8 = undefined; + var dir_buf: [sun_path_len:0]u8 = undefined; + const dir = socketDir(&dir_buf).?; + try testing.expectEqualStrings(socketPath(&expect, dir, "work").?, named); + + const path = (try resolve(&buf, "/tmp/somewhere.sock")).unix; + try testing.expectEqualStrings("/tmp/somewhere.sock", path); + try testing.expectEqualStrings("/tmp/somewhere.sock", (try resolve(&buf, "unix!/tmp/somewhere.sock")).unix); + + try testing.expectError(error.BadDial, resolve(&buf, "")); + try testing.expectError(error.BadDial, resolve(&buf, "/tmp/a\x00b")); + } + + test "a dial with nothing listening is one error and not a wait" { + if (comptime !supported) return error.SkipZigTest; + var names: [max_depth][]const u8 = undefined; + const n = try elements("/self/pane/1/body", &names); + var remote: RemoteError = .{}; + const before = nowMs(); + try testing.expectError( + Error.Dial, + fetchBytes(testing.allocator, .{ .unix = "/tmp/pardes-9p-no-such-socket.sock" }, names[0..n], &remote, null, "/self/pane/1/body", limits.max_file_bytes), + ); + try testing.expect(nowMs() - before < budget_ms); + } + + test "one fetch has three msize buffers and bounded transport metadata" { + const transport_bytes = if (quic_enabled) @sizeOf(?quic.Connection) else 0; + try testing.expectEqual(@as(usize, msize), @as(usize, (Session{ .fd = -1, .deadline = 0, .display_path = "" }).in.len)); + try testing.expect(transport_bytes <= 64); + try testing.expect(@sizeOf(Session) <= 3 * msize + 256 + transport_bytes); + try testing.expect(@sizeOf(ninep.Client) <= 256); + } + + test "expired sessions do not send or consume buffered protocol work" { + var session: Session = .{ .fd = -1, .deadline = 0, .display_path = "" }; + session.cl = .init(.{ .in = &session.in, .out = &session.out }); + _ = try session.cl.submit(.{ .version = .{} }); + const queued = session.cl.output().len; + try testing.expect(queued > 0); + try testing.expectError(Error.Timeout, session.flush()); + try testing.expectEqual(queued, session.cl.output().len); + try testing.expectError(Error.Timeout, session.settle()); + try testing.expectError(Error.Timeout, session.wait(poll_in)); + } +}; |