//! `zit daemon`: one store served over a Unix socket to a client of the same
//! user, a desktop app's background committer above all, which keeps one
//! connection open and pipelines commits and reads over it.
//!
//! One thread answers, one connection at a time: a second client waits in
//! the listen backlog and is served once the first disconnects. That is the
//! shape the client needs (one persistent connection), and it keeps every
//! request strictly ordered without a lock of our own.
//!
//! The store's writer lock is taken per commit, or once per batch, and never
//! held while idle, so `zit commit`, `sync` and `pack` interleave with the
//! daemon as they do with each other; the daemon notices their appends and
//! rewrites before its next answer. A reply to a commit is written only after
//! the commit is durable.
//!
//! Waiting is done on several things at once: the listener or the open
//! connection, SIGINT/SIGTERM (through a self-pipe), stdin's end when asked to
//! follow it, and the idle deadline. Each waits as its own task in one
//! `std.Io.Select`, and the loop handles whichever finishes first. Stopping
//! cancels the waits, never a request: one being answered finishes, through
//! its fsync, before the socket is unlinked and the locks released.
const std = @import("std");
const repeat = @import("../repeat.zig").repeat;
const builtin = @import("builtin");
const net = std.Io.net;
const checkin = @import("../core/checkin.zig");
const feed = @import("../core/feed.zig");
const store_open = @import("../core/store_open.zig");
const storelock = @import("../storelock.zig");
const logstore = @import("../logstore.zig");
const protocol = @import("protocol.zig");
pub const Diag = checkin.Diag;
/// The daemon's single-instance lock, beside the store's writer lock.
pub const daemon_lock_name = ".zit.daemon";
/// The longest socket path `sockaddr_un` holds, its terminating NUL
/// excluded: 104 bytes of `sun_path` on macOS and the BSDs, 108 on Linux.
pub const socket_path_max: usize = (if (builtin.os.tag == .linux) 108 else 104) - 1;
/// Where a crash test stops the process: after a batch's appends are written
/// and before their fsync, or after the fsync and before the reply.
/// `sync_fails` makes the next batch's fsync report failure instead.
pub const Fault = enum { none, after_append, after_sync, sync_fails };
/// The exit status a fault seam ends the process with, so a test can tell
/// the planted stop from a real failure.
pub const fault_exit: u8 = 70;
/// What the store's settings say a commit checks, resolved by a caller that
/// can read the settings tiers. A message instead of a policy refuses to
/// serve the store at all.
pub const PolicyHook = struct {
ctx: ?*anyopaque = null,
resolve: *const fn (ctx: ?*anyopaque, arena: std.mem.Allocator, cfg: ?*const logstore.Config) PolicyAnswer,
};
pub const PolicyAnswer = union(enum) {
policy: checkin.Policy,
refused: []const u8,
};
pub const Options = struct {
socket_path: []const u8,
/// The store as the client named it, reported by `hello`.
store_abs: []const u8,
zit_version: []const u8,
exit_on_stdin_close: bool = false,
/// Exit after this long with no connection open. Ignored with
/// `exit_on_stdin_close`, whose client decides when the daemon goes.
idle_exit_ms: ?u64 = null,
limits: protocol.Limits = .{},
fault: Fault = .none,
policy_hook: ?PolicyHook = null,
elided_hook: ?feed.ElidedHook = null,
/// A connection whose request stops sending its trailers for this long
/// is closed. One between requests may stay idle indefinitely.
stall_ms: u64 = default_stall_ms,
/// One line per event an operator should know of: a stalled client
/// dropped, a fatal store fault.
log: ?*std.Io.Writer = null,
};
pub const default_stall_ms: u64 = 30_000;
pub const RunError = error{
OutOfMemory,
/// Another daemon serves this store. The diag names its process.
AlreadyRunning,
/// The socket path or its directory is unusable. The diag says why.
Socket,
/// The store could not be opened or locked. The diag says why.
Store,
/// The store's settings ask for what the daemon does not do.
Refused,
/// The Io implementation cannot run the concurrent waits the loop needs.
NoConcurrency,
};
// ---------------------------------------------------------------------------
// The socket's path
/// `TMPDIR/zit-UID/<first 16 hex of SHA3-256(store_abs)>.sock`: per user, so
/// the directory's permissions keep other users out, and per store, so two
/// daemons never meet. Null when the result would not fit `sockaddr_un`.
pub fn defaultSocketPath(buf: []u8, tmpdir: []const u8, uid: u32, store_abs: []const u8) ?[]const u8 {
var digest: [32]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(store_abs, &digest, .{});
const hex = std.fmt.bytesToHex(digest, .lower);
const trimmed = std.mem.trimEnd(u8, tmpdir, "/");
const path = std.fmt.bufPrint(buf, "{s}/zit-{d}/{s}.sock", .{ trimmed, uid, hex[0..16] }) catch return null;
if (path.len > socket_path_max) return null;
return path;
}
pub fn currentUid() u32 {
return sys.getuid();
}
/// Only a peer running as this user may talk to the daemon; a peer whose
/// credentials cannot be read is refused like a stranger.
pub fn peerAllowed(peer_uid: ?u32, own_uid: u32) bool {
const uid = peer_uid orelse return false;
return uid == own_uid;
}
/// The socket directory is safe to listen in only when this user owns it and
/// nobody else may enter it: a socket's own mode is not honoured everywhere,
/// so the directory is the guard.
pub fn socketDirSafe(owner_uid: u32, mode: u32, is_dir: bool, own_uid: u32) bool {
return is_dir and owner_uid == own_uid and mode & 0o077 == 0;
}
// ---------------------------------------------------------------------------
// Running
pub fn run(
gpa: std.mem.Allocator,
io: std.Io,
root_dir: std.Io.Dir,
opts: Options,
ready: *std.Io.Writer,
diag: *Diag,
) RunError!void {
diag.clear();
if (opts.socket_path.len == 0 or opts.socket_path.len > socket_path_max) {
diag.set("the socket path is {d} bytes; a Unix socket path holds at most {d}", .{ opts.socket_path.len, socket_path_max });
return error.Socket;
}
const store_dir_opt = store_open.storeDirOf(io, root_dir) catch {
diag.set("cannot follow this checkout to its store", .{});
return error.Store;
};
defer if (store_dir_opt) |d| d.close(io);
const store_dir = store_dir_opt orelse root_dir;
var lock = try takeDaemonLock(io, store_dir, diag);
defer lock.release(io);
// The directory is judged by its absolute path, which is what a relative
// one means while the daemon's cwd stays put, and what a message should
// name; the socket itself is bound as given, to keep within sun_path.
const cwd_abs = std.process.currentPathAlloc(io, gpa) catch return error.OutOfMemory;
defer gpa.free(cwd_abs);
const socket_abs = try std.fs.path.resolve(gpa, &.{ cwd_abs, opts.socket_path });
defer gpa.free(socket_abs);
try prepareSocketDir(io, socket_abs, diag);
try clearStaleSocket(io, opts.socket_path, socket_abs, diag);
const d = try gpa.create(Daemon);
defer gpa.destroy(d);
try d.open(gpa, io, root_dir, opts, diag);
defer d.close();
const address = net.UnixAddress.init(opts.socket_path) catch unreachable; // length checked above
var listener = address.listen(io, .{}) catch |e| {
diag.set("cannot listen on {s}: {s}", .{ opts.socket_path, @errorName(e) });
return error.Socket;
};
defer listener.deinit(io);
defer std.Io.Dir.cwd().deleteFile(io, opts.socket_path) catch {};
sys.chmod(opts.socket_path, 0o600) catch {
diag.set("cannot restrict {s} to its owner", .{opts.socket_path});
return error.Socket;
};
var signals = Signals.install() catch {
diag.set("cannot watch for SIGINT and SIGTERM", .{});
return error.Socket;
};
defer signals.restore(io);
ready.print("{s}\n", .{opts.socket_path}) catch {};
ready.flush() catch {};
try d.serve(&listener, &signals);
}
/// Remove a socket a dead daemon left, and nothing else: holding this store's
/// lock says nothing about a path another store's daemon, or the user, may
/// own. Not a socket, it stays; a socket something still answers on, too.
fn clearStaleSocket(io: std.Io, path: []const u8, path_abs: []const u8, diag: *Diag) RunError!void {
const st = sys.lstat(path) orelse return; // nothing there to clear
if (!st.is_sock) {
diag.set("{s} exists and is not a socket; refusing to replace it", .{path_abs});
return error.Socket;
}
const address = net.UnixAddress.init(path) catch unreachable; // length checked by the caller
if (address.connect(io)) |stream| {
stream.close(io);
diag.set("something already answers on {s}; another daemon may be using it", .{path_abs});
return error.Socket;
} else |_| {}
std.Io.Dir.cwd().deleteFile(io, path) catch |e| switch (e) {
error.FileNotFound => {},
else => {
diag.set("cannot remove the stale socket {s}: {s}", .{ path_abs, @errorName(e) });
return error.Socket;
},
};
}
fn takeDaemonLock(io: std.Io, store_dir: std.Io.Dir, diag: *Diag) RunError!storelock.Lock {
var discard: std.Io.Writer.Discarding = .init(&.{});
return storelock.acquire(io, store_dir, daemon_lock_name, 0, &discard.writer) catch |e| switch (e) {
error.Busy, error.Ambiguous => {
if (storelock.holderPid(io, store_dir, daemon_lock_name)) |pid| {
diag.set("another daemon (process {d}) is serving this store", .{pid});
} else {
diag.set("another daemon is serving this store; if none is running, delete {s} beside the store", .{daemon_lock_name});
}
return error.AlreadyRunning;
},
error.HeldByThisProcess, error.Io => {
diag.set("cannot take {s} beside the store", .{daemon_lock_name});
return error.Store;
},
};
}
/// Make the socket's directory 0700, or accept one that already is and is
/// ours; refuse anything else rather than listen where another user can.
fn prepareSocketDir(io: std.Io, socket_path: []const u8, diag: *Diag) RunError!void {
const dir_path = std.fs.path.dirname(socket_path) orelse ".";
std.Io.Dir.cwd().createDir(io, dir_path, .fromMode(0o700)) catch |e| switch (e) {
error.PathAlreadyExists => {},
else => {
diag.set("cannot create the socket directory {s}: {s}", .{ dir_path, @errorName(e) });
return error.Socket;
},
};
const st = sys.lstat(dir_path) orelse {
diag.set("cannot inspect the socket directory {s}", .{dir_path});
return error.Socket;
};
const uid = sys.getuid();
if (!socketDirSafe(st.uid, st.mode, st.is_dir, uid)) {
diag.set("refusing the socket directory {s}: it must be a directory owned by uid {d} with mode 0700 (it is uid {d}, mode {o})", .{ dir_path, uid, st.uid, st.mode & 0o7777 });
return error.Socket;
}
}
/// The store, as the daemon holds it between requests: the commit core and
/// the reader, each with its own view, neither holding a lock.
const Daemon = struct {
gpa: std.mem.Allocator,
io: std.Io,
opts: Options,
ctx: checkin.Context,
reader: feed.Reader,
/// Owns the policy's strings for the daemon's life.
policy_arena: std.heap.ArenaAllocator,
policy: checkin.Policy,
uid: u32,
/// Set when the store can no longer be trusted to hold what was answered
/// (an fsync failed): the daemon replies, says why, and exits.
fatal: ?[]const u8,
fn open(d: *Daemon, gpa: std.mem.Allocator, io: std.Io, root_dir: std.Io.Dir, opts: Options, diag: *Diag) RunError!void {
d.* = .{
.gpa = gpa,
.io = io,
.opts = opts,
.ctx = undefined,
.reader = undefined,
.policy_arena = .init(gpa),
.policy = .{},
.uid = sys.getuid(),
.fatal = null,
};
errdefer d.policy_arena.deinit();
d.ctx.open(gpa, io, root_dir, .{}, diag) catch |e| return openFailed(e, diag);
errdefer d.ctx.deinit();
d.reader.open(gpa, io, root_dir, opts.elided_hook, diag) catch |e| return openFailed(e, diag);
errdefer d.reader.deinit();
if (opts.policy_hook) |hook| {
var cfg = d.ctx.log.readConfig(gpa) catch null;
defer if (cfg) |*c| c.deinit(gpa);
switch (hook.resolve(hook.ctx, d.policy_arena.allocator(), if (cfg) |*c| &c.config else null)) {
.policy => |p| d.policy = p,
.refused => |why| {
diag.set("{s}", .{why});
return error.Refused;
},
}
}
}
fn openFailed(e: anyerror, diag: *Diag) RunError {
if (e == error.OutOfMemory) return error.OutOfMemory;
if (diag.len == 0) diag.set("cannot open the store ({s})", .{@errorName(e)});
return error.Store;
}
fn close(d: *Daemon) void {
d.reader.deinit();
d.ctx.deinit();
d.policy_arena.deinit();
}
fn backend(d: *Daemon) protocol.Backend {
return .{ .ctx = d, .vtable = &.{
.hello = hello,
.commit = commitAll,
.feed = feedFn,
.get = getFn,
.status = statusFn,
.lookup = lookupFn,
} };
}
fn hello(ptr: *anyopaque, arena: std.mem.Allocator) protocol.Answer(protocol.Hello) {
const d: *Daemon = @ptrCast(@alignCast(ptr));
var diag: Diag = .{};
const st = d.reader.status(arena, &diag) catch |e| return .{ .fault = protocol.feedFault(e, keep(arena, &diag)) };
return .{ .ok = .{ .zit = d.opts.zit_version, .store = d.opts.store_abs, .tip = st } };
}
/// Every commit under one hold of the writer lock and one fsync; the
/// replies wait for that fsync.
fn commitAll(ptr: *anyopaque, arena: std.mem.Allocator, works: []const protocol.CommitWork, results: []protocol.ItemResult) void {
const d: *Daemon = @ptrCast(@alignCast(ptr));
std.debug.assert(works.len == results.len and works.len != 0);
var diag: Diag = .{};
d.ctx.beginBatch(&diag) catch |e| {
const f = protocol.coreFault(e, keep(arena, &diag));
for (results) |*r| r.* = .{ .fault = f };
return;
};
for (works, results) |w, *r| r.* = d.commitOne(arena, w);
if (d.opts.fault == .after_append) std.process.exit(fault_exit);
if (d.opts.fault == .sync_fails) d.ctx.fail_next_sync = true;
d.ctx.endBatch(&diag) catch |e| {
// Nothing in the batch is durable, so nothing may read as done;
// and the page cache may still hold the frames, so no later
// answer from this process can be trusted either.
const f = protocol.coreFault(e, keep(arena, &diag));
for (results) |*r| if (r.* == .ok) {
r.* = .{ .fault = f };
};
d.fatal = "the store's fsync failed; exiting so that a restart reads the store from disk";
return;
};
if (d.opts.fault == .after_sync) std.process.exit(fault_exit);
}
fn commitOne(d: *Daemon, arena: std.mem.Allocator, w: protocol.CommitWork) protocol.ItemResult {
const c = w.body;
var report: checkin.Report = .{ .arena = arena };
var diag: Diag = .{};
const outcome = checkin.commit(&d.ctx, .{
.parent = c.parent,
.branch = c.branch,
.user = c.user,
.comment = c.comment,
.now = c.time,
.edits = w.edits,
.find_resend = true,
.allow_fork = c.allow_fork,
.no_checks = c.no_checks,
.policy = d.policy,
.report = &report,
}, &diag) catch |e| return .{ .fault = protocol.coreFault(e, keep(arena, &diag)) };
const word: protocol.OutcomeWord, const name = switch (outcome) {
.recorded => |n| .{ .recorded, n },
.unchanged => |n| .{ .unchanged, n },
.existing => |n| .{ .existing, n },
};
const hash = arena.dupe(u8, name.slice()) catch return oom();
const parent: ?[]const u8 = if (report.parent.len == 0) null else (arena.dupe(u8, report.parent.slice()) catch return oom());
return .{ .ok = .{ .outcome = word, .hash = hash, .parent = parent, .branch = report.branch, .forked = report.forked } };
}
fn oom() protocol.ItemResult {
return .{ .fault = protocol.coreFault(error.OutOfMemory, "out of memory") };
}
fn feedFn(ptr: *anyopaque, arena: std.mem.Allocator, cursor: []const u8) protocol.Answer(feed.Feed) {
const d: *Daemon = @ptrCast(@alignCast(ptr));
var diag: Diag = .{};
const f = d.reader.feed(arena, cursor, &diag) catch |e| return .{ .fault = protocol.feedFault(e, keep(arena, &diag)) };
return .{ .ok = f };
}
fn getFn(ptr: *anyopaque, arena: std.mem.Allocator, uuid: []const u8) protocol.Answer([]const u8) {
const d: *Daemon = @ptrCast(@alignCast(ptr));
var diag: Diag = .{};
const got = d.reader.get(arena, uuid, &diag) catch |e| return .{ .fault = protocol.feedFault(e, keep(arena, &diag)) };
return switch (got) {
.bytes => |b| .{ .ok = b },
.absent => |why| .{ .fault = absentFault(why) },
};
}
fn statusFn(ptr: *anyopaque, arena: std.mem.Allocator) protocol.Answer(feed.Status) {
const d: *Daemon = @ptrCast(@alignCast(ptr));
var diag: Diag = .{};
const st = d.reader.status(arena, &diag) catch |e| return .{ .fault = protocol.feedFault(e, keep(arena, &diag)) };
return .{ .ok = st };
}
/// From the reader, not the commit context: the reader catches up with
/// other writers' appends without the writer lock, so a lookup answers
/// while a CLI commit holds it.
fn lookupFn(ptr: *anyopaque, arena: std.mem.Allocator, queries: []const protocol.Lookup) protocol.Answer([]const ?protocol.Found) {
const d: *Daemon = @ptrCast(@alignCast(ptr));
var diag: Diag = .{};
const found = d.reader.lookup(arena, queries, &diag) catch |e| return .{ .fault = protocol.feedFault(e, keep(arena, &diag)) };
return .{ .ok = found };
}
// -----------------------------------------------------------------------
// The loop
const Stop = enum { signal, stdin_closed };
const Event = union(enum) {
accepted: net.Server.AcceptError!net.Stream,
request: protocol.ReadError!protocol.Incoming,
stop: Stop,
idle: void,
stall: void,
};
fn serve(d: *Daemon, listener: *net.Server, signals: *Signals) RunError!void {
const io = d.io;
// Declared before the select so they are released after it: the
// select's cancel must stop a pending read before its connection
// and arena go.
var request_arena: std.heap.ArenaAllocator = .init(d.gpa);
defer request_arena.deinit();
var conn: ?*Conn = null;
defer if (conn) |c| c.destroy(d.gpa, io);
// Every wait that can be pending at once, plus room: the select
// deadlocks on cancel if a finished task has nowhere to put its result.
var slots: [8]Event = undefined;
var sel = std.Io.Select(Event).init(io, &slots);
defer while (sel.cancel()) |left| switch (left) {
.accepted => |r| if (r) |s| s.close(io) else |_| {},
else => {},
};
sel.concurrent(.stop, waitSignal, .{ io, signals.read_end }) catch return error.NoConcurrency;
if (d.opts.exit_on_stdin_close) sel.concurrent(.stop, waitStdin, .{io}) catch return error.NoConcurrency;
const idle_ms: ?u64 = if (d.opts.exit_on_stdin_close) null else d.opts.idle_exit_ms;
if (idle_ms) |ms| sel.concurrent(.idle, sleepMs, .{ io, ms }) catch return error.NoConcurrency;
sel.concurrent(.accepted, accept, .{ listener, io }) catch return error.NoConcurrency;
var last_seen = std.Io.Timestamp.now(io, .awake);
// One stall watcher at a time, alive while a connection is open.
var stall_live = false;
var stall_seen: u64 = 0;
while (true) {
const event = sel.await() catch return;
switch (event) {
.stop => return,
.stall => {
stall_live = false;
const c = conn orelse continue;
const now_bytes = c.progress.bytes.load(.monotonic);
if (c.progress.in_trailers.load(.acquire) and now_bytes == stall_seen) {
// Wakes the blocked read with end of stream; the read
// then closes the connection the ordinary way.
c.stream.shutdown(io, .both) catch {};
d.logLine("zit daemon: closed a connection whose request sent nothing for {d} ms", .{d.opts.stall_ms});
}
stall_seen = now_bytes;
sel.concurrent(.stall, sleepMs, .{ io, d.opts.stall_ms }) catch return error.NoConcurrency;
stall_live = true;
},
.idle => {
const ms = idle_ms.?;
const quiet = elapsedMs(io, last_seen);
if (conn == null and quiet >= ms) return;
sel.concurrent(.idle, sleepMs, .{ io, if (conn == null) ms - quiet else ms }) catch return error.NoConcurrency;
},
.accepted => |result| {
const stream = result catch |e| switch (e) {
error.Canceled => return,
error.ConnectionAborted, error.WouldBlock, error.SystemResources, error.ProcessFdQuotaExceeded, error.SystemFdQuotaExceeded => {
sel.concurrent(.accepted, accept, .{ listener, io }) catch return error.NoConcurrency;
continue;
},
else => return,
};
if (!peerAllowed(sys.peerUid(stream.socket.handle), d.uid)) {
stream.close(io);
sel.concurrent(.accepted, accept, .{ listener, io }) catch return error.NoConcurrency;
continue;
}
conn = Conn.create(d.gpa, io, stream) catch {
stream.close(io);
sel.concurrent(.accepted, accept, .{ listener, io }) catch return error.NoConcurrency;
continue;
};
last_seen = std.Io.Timestamp.now(io, .awake);
_ = request_arena.reset(.{ .retain_with_limit = arena_retain_max });
sel.concurrent(.request, readNext, .{ conn.?, request_arena.allocator(), d.opts.limits }) catch return error.NoConcurrency;
if (!stall_live) {
stall_seen = conn.?.progress.bytes.load(.monotonic);
sel.concurrent(.stall, sleepMs, .{ io, d.opts.stall_ms }) catch return error.NoConcurrency;
stall_live = true;
}
},
.request => |result| {
const c = conn.?;
last_seen = std.Io.Timestamp.now(io, .awake);
const keep_open = d.answer(c, request_arena.allocator(), result);
if (keep_open) {
_ = request_arena.reset(.{ .retain_with_limit = arena_retain_max });
sel.concurrent(.request, readNext, .{ c, request_arena.allocator(), d.opts.limits }) catch return error.NoConcurrency;
} else {
c.destroy(d.gpa, io);
conn = null;
sel.concurrent(.accepted, accept, .{ listener, io }) catch return error.NoConcurrency;
}
},
}
}
}
/// Answer one read off the connection; false when the connection is
/// finished, by the client or because its framing can no longer be
/// trusted.
fn answer(d: *Daemon, c: *Conn, arena: std.mem.Allocator, result: protocol.ReadError!protocol.Incoming) bool {
const out = &c.writer.interface;
const incoming = result catch |e| {
const f: protocol.Fault = switch (e) {
error.Closed, error.Truncated, error.ReadFailed => return false,
error.HeaderTooLarge => .{ .code = "header_too_large", .exit = protocol.exit_usage, .message = "the header line is longer than 1 MiB" },
error.OutOfMemory => protocol.coreFault(error.OutOfMemory, "out of memory reading the request"),
};
protocol.writeFault(out, null, f) catch {};
out.flush() catch {};
return false;
};
protocol.dispatch(arena, d.backend(), &incoming, d.opts.limits, out) catch |e| {
if (e == error.OutOfMemory) protocol.writeFault(out, incoming.header.id, protocol.coreFault(error.OutOfMemory, "out of memory answering the request")) catch {};
};
out.flush() catch return false;
if (d.fatal) |why| {
d.logLine("zit daemon: {s}", .{why});
std.process.exit(1);
}
return incoming.header.trailers != null;
}
fn logLine(d: *Daemon, comptime fmt: []const u8, args: anytype) void {
const w = d.opts.log orelse return;
w.print(fmt ++ "\n", args) catch {};
w.flush() catch {};
}
};
fn keep(arena: std.mem.Allocator, diag: *const Diag) []const u8 {
return arena.dupe(u8, diag.text()) catch "";
}
fn absentFault(why: feed.Absence) protocol.Fault {
return switch (why) {
.purged => .{ .code = "content_absent", .exit = protocol.exit_absent, .message = "the content was purged from this store ('zit purge artifacts')" },
.shunned => .{ .code = "content_absent", .exit = protocol.exit_absent, .message = "the content is shunned in this repository" },
.elided => .{ .code = "content_absent", .exit = protocol.exit_absent, .message = "the content is elided by this store's cone or horizon" },
.missing => .{ .code = "not_found", .exit = protocol.exit_failure, .message = "no artifact by that name is stored here" },
};
}
/// What the per-request arena keeps between requests: enough for ordinary
/// headers and replies, not the gibibyte a single large put can borrow.
const arena_retain_max: usize = 8 << 20;
/// One client connection: the stream and its buffered reader and writer,
/// heap-pinned because the interfaces point into it.
const Conn = struct {
stream: net.Stream,
read_buf: []u8,
reader: net.Stream.Reader,
write_buf: [64 * 1024]u8,
writer: net.Stream.Writer,
progress: protocol.Progress,
fn create(gpa: std.mem.Allocator, io: std.Io, stream: net.Stream) error{OutOfMemory}!*Conn {
const c = try gpa.create(Conn);
errdefer gpa.destroy(c);
// A whole header line must fit the buffer to be read as one.
c.read_buf = try gpa.alloc(u8, protocol.header_max + 1);
c.stream = stream;
c.reader = stream.reader(io, c.read_buf);
c.writer = stream.writer(io, &c.write_buf);
c.progress = .{};
return c;
}
fn destroy(c: *Conn, gpa: std.mem.Allocator, io: std.Io) void {
c.stream.close(io);
gpa.free(c.read_buf);
gpa.destroy(c);
}
};
fn readNext(c: *Conn, arena: std.mem.Allocator, limits: protocol.Limits) protocol.ReadError!protocol.Incoming {
return protocol.readRequest(&c.reader.interface, arena, limits, &c.progress);
}
fn accept(listener: *net.Server, io: std.Io) net.Server.AcceptError!net.Stream {
return listener.accept(io);
}
fn waitSignal(io: std.Io, read_end: std.Io.File) Daemon.Stop {
var byte: [1]u8 = undefined;
var bufs = [_][]u8{&byte};
_ = read_end.readStreaming(io, &bufs) catch {};
return .signal;
}
/// Read stdin to its end, keeping nothing: the client closing it is the
/// whole message.
fn waitStdin(io: std.Io) Daemon.Stop {
var buf: [512]u8 = undefined;
var bufs = [_][]u8{&buf};
const stdin = std.Io.File.stdin();
while (true) {
const n = stdin.readStreaming(io, &bufs) catch return .stdin_closed;
if (n == 0) return .stdin_closed;
}
}
fn sleepMs(io: std.Io, ms: u64) void {
io.sleep(.fromMilliseconds(@intCast(@min(ms, std.math.maxInt(i64)))), .awake) catch {};
}
fn elapsedMs(io: std.Io, since: std.Io.Timestamp) u64 {
const ns = since.durationTo(std.Io.Timestamp.now(io, .awake)).toNanoseconds();
return if (ns <= 0) 0 else @intCast(@divTrunc(ns, std.time.ns_per_ms));
}
// ---------------------------------------------------------------------------
// Signals, through a self-pipe: the handler writes one byte, and a task
// blocked reading the pipe wakes the loop.
var signal_write_fd: std.posix.fd_t = -1;
fn onSignal(_: std.posix.SIG) callconv(.c) void {
const byte = [_]u8{1};
sys.writeByte(signal_write_fd, &byte);
}
const Signals = struct {
read_end: std.Io.File,
write_fd: std.posix.fd_t,
old_int: std.posix.Sigaction,
old_term: std.posix.Sigaction,
fn install() error{Pipe}!Signals {
const fds = sys.pipe() orelse return error.Pipe;
signal_write_fd = fds[1];
var s: Signals = .{ .read_end = .{ .handle = fds[0], .flags = .{ .nonblocking = false } }, .write_fd = fds[1], .old_int = undefined, .old_term = undefined };
const act: std.posix.Sigaction = .{
.handler = .{ .handler = onSignal },
.mask = std.posix.sigemptyset(),
.flags = 0,
};
std.posix.sigaction(.INT, &act, &s.old_int);
std.posix.sigaction(.TERM, &act, &s.old_term);
return s;
}
fn restore(s: *Signals, io: std.Io) void {
std.posix.sigaction(.INT, &s.old_int, null);
std.posix.sigaction(.TERM, &s.old_term, null);
signal_write_fd = -1;
s.read_end.close(io);
sys.closeFd(s.write_fd);
}
};
// ---------------------------------------------------------------------------
// What the standard library does not wrap: credentials, ownership, modes.
// Linux builds link no libc, so it is reached through raw system calls
// there; everywhere else zit links libc.
const sys = struct {
const linux = std.os.linux;
const is_linux = builtin.os.tag == .linux;
fn getuid() u32 {
return if (is_linux) linux.getuid() else std.c.getuid();
}
/// `SO_PEERCRED`'s layout.
const Ucred = extern struct { pid: i32, uid: u32, gid: u32 };
extern "c" fn getpeereid(fd: std.c.fd_t, uid: *std.c.uid_t, gid: *std.c.gid_t) c_int;
fn peerUid(fd: std.posix.fd_t) ?u32 {
if (is_linux) {
var cred: Ucred = undefined;
var len: linux.socklen_t = @sizeOf(Ucred);
const rc = linux.getsockopt(fd, linux.SOL.SOCKET, linux.SO.PEERCRED, @ptrCast(&cred), &len);
if (linux.errno(rc) != .SUCCESS or len != @sizeOf(Ucred)) return null;
return cred.uid;
}
var uid: std.c.uid_t = undefined;
var gid: std.c.gid_t = undefined;
if (getpeereid(fd, &uid, &gid) != 0) return null;
return uid;
}
const Owner = struct { uid: u32, mode: u32, is_dir: bool, is_sock: bool };
/// The path's own owner and mode, a symbolic link not followed.
fn lstat(path: []const u8) ?Owner {
var z: [std.fs.max_path_bytes + 1]u8 = undefined;
if (path.len >= z.len) return null;
@memcpy(z[0..path.len], path);
z[path.len] = 0;
const zp: [*:0]const u8 = z[0..path.len :0];
if (is_linux) {
var stx: linux.Statx = undefined;
const rc = linux.statx(linux.AT.FDCWD, zp, linux.AT.SYMLINK_NOFOLLOW, .{ .TYPE = true, .MODE = true, .UID = true }, &stx);
if (linux.errno(rc) != .SUCCESS) return null;
const kind = stx.mode & linux.S.IFMT;
return .{ .uid = stx.uid, .mode = stx.mode, .is_dir = kind == linux.S.IFDIR, .is_sock = kind == linux.S.IFSOCK };
}
var st: std.c.Stat = undefined;
if (std.c.fstatat(std.c.AT.FDCWD, zp, &st, std.c.AT.SYMLINK_NOFOLLOW) != 0) return null;
const kind = st.mode & std.c.S.IFMT;
return .{ .uid = st.uid, .mode = st.mode, .is_dir = kind == std.c.S.IFDIR, .is_sock = kind == std.c.S.IFSOCK };
}
fn chmod(path: []const u8, mode: u32) error{Failed}!void {
var z: [std.fs.max_path_bytes + 1]u8 = undefined;
if (path.len >= z.len) return error.Failed;
@memcpy(z[0..path.len], path);
z[path.len] = 0;
const zp: [*:0]const u8 = z[0..path.len :0];
if (is_linux) {
if (linux.errno(linux.fchmodat(linux.AT.FDCWD, zp, mode)) != .SUCCESS) return error.Failed;
return;
}
if (std.c.chmod(zp, @intCast(mode)) != 0) return error.Failed;
}
fn pipe() ?[2]std.posix.fd_t {
var fds: [2]std.posix.fd_t = undefined;
if (is_linux) {
if (linux.errno(linux.pipe2(&fds, .{ .CLOEXEC = true })) != .SUCCESS) return null;
return fds;
}
if (std.c.pipe(&fds) != 0) return null;
// No pipe2 here: mark both ends close-on-exec so a hook a later
// version runs cannot inherit them.
for (fds) |fd| _ = std.c.fcntl(fd, std.c.F.SETFD, @as(c_int, std.c.FD_CLOEXEC));
return fds;
}
/// Async-signal-safe: one `write`, its result ignored.
fn writeByte(fd: std.posix.fd_t, byte: *const [1]u8) void {
if (fd < 0) return;
if (is_linux) {
_ = linux.write(fd, byte, 1);
} else {
_ = std.c.write(fd, byte, 1);
}
}
fn closeFd(fd: std.posix.fd_t) void {
if (is_linux) {
_ = linux.close(fd);
} else {
_ = std.c.close(fd);
}
}
};
// ---------------------------------------------------------------------------
// Tests
const testing = std.testing;
test "peerAllowed: only this user, and never a peer whose credentials are unknown" {
try testing.expect(peerAllowed(501, 501));
try testing.expect(!peerAllowed(0, 501));
try testing.expect(!peerAllowed(502, 501));
try testing.expect(!peerAllowed(null, 501));
}
test "socketDirSafe: a directory this user owns that nobody else may enter" {
try testing.expect(socketDirSafe(501, 0o40700, true, 501));
try testing.expect(!socketDirSafe(501, 0o40750, true, 501));
try testing.expect(!socketDirSafe(501, 0o40701, true, 501));
try testing.expect(!socketDirSafe(0, 0o40700, true, 501));
try testing.expect(!socketDirSafe(501, 0o100600, false, 501));
}
test "defaultSocketPath: per user and per store, and refused past sockaddr_un" {
var buf: [512]u8 = undefined;
const a = defaultSocketPath(&buf, "/tmp/", 501, "/home/e/store").?;
try testing.expect(std.mem.startsWith(u8, a, "/tmp/zit-501/"));
try testing.expect(std.mem.endsWith(u8, a, ".sock"));
try testing.expectEqual(@as(usize, "/tmp/zit-501/".len + 16 + ".sock".len), a.len);
var buf2: [512]u8 = undefined;
const b = defaultSocketPath(&buf2, "/tmp", 501, "/home/e/other").?;
try testing.expect(!std.mem.eql(u8, a, b));
try testing.expect(defaultSocketPath(&buf, "/" ++ repeat("d", 100), 501, "/s") == null);
}