merge: store-backed download.add/list/get

2026-09-10 20:04:44 +04:00
parent 477a98a31b 3db0d01f9d
commit b02bdb0466
11 changed files with 286 additions and 53 deletions
@@ -12,8 +12,13 @@
 #include "rpc/dispatcher.hpp"
 #include "rpc/event_loop.hpp"
 #include "rpc/uds_server.hpp"
+#include "store/migrations.hpp"
+#include "store/settings.hpp"
+#include "store/sqlite.hpp"

 namespace rpc = velox::daemon::rpc;
+
+static std::string g_allowed_root;
 using velox::cli::CallError;
 using velox::cli::Client;

@@ -39,7 +44,22 @@ void run() {
    const std::string server_sock = velox_dir + "/velox.sock";

    rpc::EventLoop loop;
-    rpc::VeloxDispatcher dispatcher;
+    auto db = velox::daemon::store::Db::open(":memory:");
+    CHECK(db.has_value());
+    if (!db) return;
+    CHECK(velox::daemon::store::migrate_to_head(*db).has_value());
+
+    char root_tmpl[] = "/tmp/velox-cli-root-XXXXXX";
+    g_allowed_root = ::mkdtemp(root_tmpl);
+    CHECK(!g_allowed_root.empty());
+    {
+        velox::daemon::store::Settings settings(*db);
+        CHECK(settings.set_raw("saveTo.allowedRoots",
+                               "[\"" + g_allowed_root + "\"]").has_value());
+        CHECK(settings.set_raw("saveTo.defaultDir",
+                               "\"" + g_allowed_root + "\"").has_value());
+    }
+    rpc::VeloxDispatcher dispatcher(*db);
    rpc::UdsServer server(loop, dispatcher, server_sock);
    const auto ec = server.start();
    CHECK(!ec);
@@ -64,10 +84,42 @@ void run() {
            CHECK(ls->at("items").is_array());
        }

-        // A not-yet-implemented method surfaces the daemon's error, not a transport error.
-        auto add = c.call("download.add", {{"url", "https://example.com/x"}});
-        CHECK(!add.has_value());
-        if (!add) CHECK_EQ(add.error().code, -32603);
+        // download.add outside every allowed root -> -32011, original saveDir echoed.
+        auto bad = c.call("download.add",
+                          {{"url", "https://example.com/x"}, {"saveDir", "/etc"}});
+        CHECK(!bad.has_value());
+        if (!bad) {
+            CHECK_EQ(bad.error().code, -32011);
+            CHECK_EQ(bad.error().data.value("path", ""), std::string("/etc"));
+        }
+
+        // download.add into an allowed root -> a task id that then shows up in the list.
+        auto ok = c.call(
+            "download.add",
+            {{"url", "https://example.com/movie.mp4"}, {"saveDir", g_allowed_root}});
+        CHECK(ok.has_value());
+        std::string task_id;
+        if (ok) {
+            task_id = ok->value("taskId", "");
+            CHECK(!task_id.empty());
+            CHECK_EQ(ok->value("state", ""), std::string("queued"));
+        }
+
+        auto ls2 = c.call("download.list", nlohmann::json::object());
+        CHECK(ls2.has_value());
+        if (ls2) {
+            CHECK_EQ(ls2->value("total", -1), 1);
+            CHECK_EQ(ls2->at("items").at(0).value("filename", ""), std::string("movie.mp4"));
+        }
+
+        auto detail = c.call("download.get", {{"taskId", task_id}});
+        CHECK(detail.has_value());
+        if (detail) CHECK_EQ(detail->at("summary").value("taskId", ""), task_id);
+
+        auto missing =
+            c.call("download.get", {{"taskId", "00000000-0000-4000-8000-000000000000"}});
+        CHECK(!missing.has_value());
+        if (!missing) CHECK_EQ(missing.error().code, -32010);
    }

    loop.stop();
@@ -6,7 +6,7 @@ close. Kept here (not buried in commit messages) so the next pass can see them a
 | # | What | Where | Why deferred | Closes when |
 |---|---|---|---|---|
 | D1 | Pairing prompt is `EnvAutoApprover` (needs `VELOX_PAIR_AUTO=1`) | `rpc/pairing.hpp`, `main.cpp` | A GUI dialog / `org.freedesktop.Notifications` approver is integration work | Build step 7 (systemd + notifications) |
-| D2 | `download.add` → `-32603`, `download.probe` → `-32603` | `rpc/dispatcher.cpp` | The path boundary (`-32011`) is built and tested (`fs/safepath`, `daemon/docs/safepath-adversarial.md`); still need it wired into the `download.add` handler with the store behind it, plus the probe path (`-32013`) which needs the engine | `download.add` glue (dispatcher ↔ store ↔ `fs/safepath`); probe with the engine link |
-| D3 | Stub handlers for everything except `session.*`, `download.list`, `download.get` | `rpc/dispatcher.cpp` | No store behind them yet | Per method, as the store/scheduler wire in |
+| D2 | `download.probe` → `-32603` | `rpc/dispatcher.cpp` | `download.add` is wired (`fs/safepath` + store, real `-32011`); `download.probe` needs the engine's probe path for `-32013` | probe with the engine link (CORE stage 3 is landed; wire `Engine::probe`) |
+| D3 | Stub handlers for everything except `session.*`, `download.add/list/get` | `rpc/dispatcher.cpp` | No store behind them yet (categories/queues/rules/settings/limiter/schedule) | Per method, as the store query modules land behind them |
 | D4 | `sched/` is the pure `Governor` + schedule window only; no `Scheduler` wiring to store/engine/timer | `sched/` | `Engine` bodies land in CORE stage 8; `Scheduler` needs the UUID↔`vdm::TaskId` map, a store query layer, and a timer | After CORE stage 8 lands `Engine::start()` |
 | D5 | `event.*` fan-out not implemented; `session.subscribe` accepts and echoes but nothing is emitted | `rpc/uds_server.cpp`, `rpc/ws_server.cpp` | No task state to broadcast until the engine is wired | With the callback → `event.*` projection |
@@ -58,20 +58,38 @@ Roots for the examples: `allowedRoots = ["/home/u/Downloads", "/data/dl"]`, alre
   `-32011`. Then re-derive the final dir's path from its fd (`/proc/self/fd/N`) and
   re-assert containment. (A8 for the created tail, A14)
 5. **Best-effort leaf check:** `fstatat(dir_fd, leaf, AT_SYMLINK_NOFOLLOW)` — refuse if it
-   is already a symlink. The real close on the create-after-check race is CORE opening the
-   file `O_NOFOLLOW|O_EXCL` (or `O_NOFOLLOW` + explicit resume); that is CORE's contract,
-   stated in `daemon/docs/engine-api-review.md`.
+   is already a symlink. This narrows, but does not close, the create-after-check race on
+   the leaf: a symlink planted *after* this `fstatat` and *before* CORE opens the file is
+   still followed. Closing it needs CORE to open with `O_NOFOLLOW` (plus `O_EXCL` on a
+   fresh download). **Verified 2026-09-10: it does not yet** —
+   `core/src/io/sparse_file.cpp:77` is `O_WRONLY | O_CREAT | O_CLOEXEC`. The flag change
+   has been raised with CORE; until it lands this race is open, see the residual below.
 6. **Every failure is `-32011`, `data.path` = the *original* `saveDir`** — never the
   resolved path, which would leak where the roots actually live. The one exception is a
   `filename` that violates the schema's own `maxLength`, which is `-32602` at the param
   layer before this code runs.

-### Residual, accepted for M1
+### Residual — currently OPEN, tracked

-An **existing intermediate directory** swapped for an out-of-root symlink *between* our
-`realpath` and CORE's `open` is not caught by this code (step 3 trusts `realpath` for the
-pre-existing prefix; a full `O_NOFOLLOW` chase would reject legitimate symlinked
-directories mid-path, which A16 requires us to allow). It is closed in practice by CORE's
-`O_NOFOLLOW` open of the final file and by the download dir living under a `0700`
-`~/.local/share` / `~/Downloads` the attacker would already need write access to. A
-per-step "resolve, re-validate against roots" chase is the post-M1 hardening.
+Two TOCTOU gaps this code does not close on its own:
+
+1. **An existing intermediate directory** swapped for an out-of-root symlink between our
+   `realpath` (step 3) and the write. Step 3 trusts `realpath` for the pre-existing
+   prefix; a full `O_NOFOLLOW` chase would reject the legitimate symlinked directories
+   A16 requires us to allow.
+2. **The leaf** swapped for a symlink between our `fstatat` (step 5) and CORE's `open`.
+
+Both are closed by CORE opening the file `O_NOFOLLOW` (and, for a fresh download,
+`O_EXCL`). **As verified on 2026-09-10 that is not yet the case** —
+`core/src/io/sparse_file.cpp:77` opens `O_WRONLY | O_CREAT | O_CLOEXEC`. The flag change has
+been raised with CORE; when it lands, update step 5 and this paragraph and re-verify the
+flags at that line.
+
+What limits the exposure *today*: the download directory lives under `~/.local/share` /
+`~/Downloads`, both `0700` — an attacker planting a symlink there already has write access
+to the user's account. The unqualified "you are covered" version of this claim does not
+hold until the CORE change lands; this is exactly the boundary where a reader stops
+checking, so it is spelled out.
+
+The post-M1 hardening for gap 1 is a per-step "resolve one component, re-validate the
+running path against the roots" chase (systemd's `chase_symlinks` shape).
@@ -101,7 +101,7 @@ int main() {
        return 1;
    }

-    velox::daemon::rpc::VeloxDispatcher dispatcher;
+    velox::daemon::rpc::VeloxDispatcher dispatcher(*db);

    velox::daemon::rpc::UdsServer uds(loop, dispatcher, rt.socket_path());
    if (const auto ec = uds.start()) {
@@ -1,5 +1,15 @@
 #include "rpc/dispatcher.hpp"

+#include <random>
+#include <string>
+
+#include <nlohmann/json.hpp>
+
+#include "fs/safepath.hpp"
+#include "store/settings.hpp"
+#include "store/tasks.hpp"
+#include "util/time.hpp"
+
 namespace velox::daemon::rpc {

 namespace proto = velox::proto;
@@ -15,6 +25,56 @@ proto::HandlerResult<T> not_implemented(const char* method) {
        std::string("not implemented in this build: ") + method});
 }

+// A v4 UUID for a new task id.
+std::string new_task_id() {
+    std::random_device rd;
+    std::uniform_int_distribution<std::uint32_t> d;
+    std::uint32_t a = d(rd), b = d(rd), c = d(rd), e = d(rd);
+    b = (b & 0xFFFF0FFFu) | 0x00004000u;
+    c = (c & 0x3FFFFFFFu) | 0x80000000u;
+    char buf[37];
+    std::snprintf(buf, sizeof(buf), "%08x-%04x-%04x-%04x-%04x%08x", a, (b >> 16), (b & 0xFFFF),
+                  (c >> 16), (c & 0xFFFF), e);
+    return std::string(buf);
+}
+
+// Expand a leading "~" against $HOME. Configured dirs may be stored that way.
+std::string expand_tilde(std::string p) {
+    if (p == "~" || p.rfind("~/", 0) == 0) {
+        if (const char* home = std::getenv("HOME"); home != nullptr && home[0] != '\0')
+            p = std::string(home) + (p.size() > 1 ? p.substr(1) : std::string{});
+    }
+    return p;
+}
+
+// Last path segment of a URL, percent-decoded, as a filename fallback when the caller gave
+// none and there is no probe yet. Empty => the handler uses "download.bin".
+std::string filename_from_url(std::string_view url) {
+    auto q = url.find_first_of("?#");
+    if (q != std::string_view::npos) url = url.substr(0, q);
+    auto slash = url.find_last_of('/');
+    std::string_view leaf = slash == std::string_view::npos ? url : url.substr(slash + 1);
+    std::string out;
+    for (std::size_t i = 0; i < leaf.size(); ++i) {
+        if (leaf[i] == '%' && i + 2 < leaf.size()) {
+            auto hex = [](char ch) -> int {
+                if (ch >= '0' && ch <= '9') return ch - '0';
+                if (ch >= 'a' && ch <= 'f') return ch - 'a' + 10;
+                if (ch >= 'A' && ch <= 'F') return ch - 'A' + 10;
+                return -1;
+            };
+            int hi = hex(leaf[i + 1]), lo = hex(leaf[i + 2]);
+            if (hi >= 0 && lo >= 0) {
+                out.push_back(static_cast<char>((hi << 4) | lo));
+                i += 2;
+                continue;
+            }
+        }
+        out.push_back(leaf[i]);
+    }
+    return out;
+}
+
 }  // namespace

 // --- session.* : handled in the server layer, unreachable here in the running daemon ---
@@ -36,12 +96,81 @@ VeloxDispatcher::on_session_subscribe(const proto::SessionSubscribeParams&) {
    return not_implemented<proto::SessionSubscribeResult>("session.subscribe");
 }

-// --- download.list : an empty table, so a client can connect and render ---------------
+// --- download.list : the main table; the store does the filter / sort / page ----------

 proto::HandlerResult<proto::DownloadListResult>
-VeloxDispatcher::on_download_list(const proto::DownloadListParams&) {
+VeloxDispatcher::on_download_list(const proto::DownloadListParams& params) {
+    store::Tasks tasks(db_);
+    auto page = tasks.list(params.filter, params.sort, params.offset.value_or(0),
+                           params.limit.value_or(0));
+    if (!page)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "download.list: " + page.error().message});
+
    proto::DownloadListResult r;
-    r.total = 0;
+    r.total = page->total;
+    r.items.reserve(page->rows.size());
+    for (const auto& row : page->rows) r.items.push_back(store::to_summary(row));
+    return r;
+}
+
+// --- download.add : canonicalise + root-check the destination, then persist -----------
+
+proto::HandlerResult<proto::DownloadAddResult>
+VeloxDispatcher::on_download_add(const proto::DownloadSpec& spec) {
+    store::Settings settings(db_);
+
+    std::string save_dir = spec.saveDir && !spec.saveDir->empty()
+                               ? *spec.saveDir
+                               : settings.get_string("saveTo.defaultDir");
+    save_dir = expand_tilde(std::move(save_dir));
+
+    std::string leaf = spec.filename && !spec.filename->empty() ? *spec.filename
+                                                                : filename_from_url(spec.url);
+    if (leaf.empty()) leaf = "download.bin";
+
+    std::vector<std::string> roots;
+    for (const auto& r : settings.get_string_array("saveTo.allowedRoots")) {
+        if (auto c = fs::canonicalize_root(r)) roots.push_back(*c);
+    }
+
+    auto target = fs::resolve_target(save_dir, leaf, roots);
+    if (!target) {
+        // Everything path-destination-related is -32011 with the *original* saveDir in
+        // data.path (never the resolved path — daemon/docs/safepath-adversarial.md rule 6).
+        return std::unexpected(proto::HandlerError{
+            proto::ErrorCode::InvalidPath, target.error().message,
+            nlohmann::json{{"path", spec.saveDir.value_or(save_dir)}}});
+    }
+
+    store::TaskRow row;
+    row.task_id = new_task_id();
+    row.url = spec.url;
+    row.save_dir = target->dir;
+    row.filename = target->leaf;
+    row.created_at = velox::daemon::now_iso();
+    row.start_mode = spec.startMode ? std::string(proto::to_string(*spec.startMode)) : "auto";
+    // startMode 'manual' parks the task in `new`; anything else makes it eligible for the
+    // scheduler (`queued`). The scheduler itself is not wired yet (deferrals.md D4).
+    row.state = row.start_mode == "manual" ? "new" : "queued";
+    row.category_id = spec.categoryId;
+    row.queue_id = spec.queueId;
+    row.description = spec.description;
+    row.req_segments = spec.segments;
+    row.req_buffer_bytes = spec.bufferBytes;
+    if (spec.checksum) {
+        row.checksum_algo = std::string(proto::to_string(spec.checksum->algorithm));
+        row.checksum_value = spec.checksum->value;
+    }
+
+    store::Tasks tasks(db_);
+    if (auto ins = tasks.insert(row); !ins)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "download.add: " + ins.error().message});
+
+    proto::DownloadAddResult r;
+    r.taskId = row.task_id;
+    if (auto st = proto::parse_TaskState(row.state)) r.state = *st;
    return r;
 }

@@ -67,10 +196,6 @@ proto::HandlerResult<proto::CategoryUpsertResult>
 VeloxDispatcher::on_category_upsert(const proto::CategoryUpsertParams&) {
    return not_implemented<proto::CategoryUpsertResult>("category.upsert");
 }
-proto::HandlerResult<proto::DownloadAddResult>
-VeloxDispatcher::on_download_add(const proto::DownloadSpec&) {
-    return not_implemented<proto::DownloadAddResult>("download.add");
-}
 proto::HandlerResult<proto::DownloadAddBatchResult>
 VeloxDispatcher::on_download_addBatch(const proto::DownloadAddBatchParams&) {
    return not_implemented<proto::DownloadAddBatchResult>("download.addBatch");
@@ -81,11 +206,26 @@ VeloxDispatcher::on_download_cancel(const proto::DownloadCancelParams&) {
 }
 proto::HandlerResult<proto::TaskDetail>
 VeloxDispatcher::on_download_get(const proto::DownloadGetParams& params) {
-    // No store is wired yet, so no task exists and every id is genuinely not-found. This
-    // is the real -32010 answer (contracts/ error fixture download.get.not-found), not a
-    // placeholder; it becomes a store lookup when store/ is wired in.
+    store::Tasks tasks(db_);
+    auto got = tasks.get(params.taskId);
+    if (!got)
+        return std::unexpected(proto::HandlerError{proto::ErrorCode::InternalError,
+                                                   "download.get: " + got.error().message});
+    if (!got->has_value())
        return std::unexpected(proto::HandlerError{proto::ErrorCode::TaskNotFound, "no such task",
                                                   nlohmann::json{{"taskId", params.taskId}}});
+
+    const store::TaskRow& row = **got;
+    proto::TaskDetail d;
+    d.summary = store::to_summary(row);
+    // segmentDetail stays empty until the engine has segmented the task — the schema
+    // permits that ("empty before the task has been segmented").
+    d.mime = row.content_type;
+    d.bufferBytes = row.req_buffer_bytes;
+    d.effectiveBufferBytes = row.eff_buffer_bytes;
+    if (row.state != "complete" && row.state != "cancelled")
+        d.partPath = row.save_dir + "/" + row.filename + ".veloxpart";
+    return d;
 }
 proto::HandlerResult<proto::BulkTaskResult>
 VeloxDispatcher::on_download_pause(const proto::DownloadPauseParams&) {
@@ -5,22 +5,22 @@
 // parse; a method here only ever sees a validated, typed params struct and returns a
 // typed result.
 //
-// Scope of this drop (AGENT-DAEMON.md build order): the transport is real, the store is
-// not. session.hello / session.pair / session.subscribe are handled in the server layer
-// (they are connection- and transport-stateful) and never reach this class in the running
-// daemon. download.list answers with an empty table so a client can connect and render.
-// Every other method returns "not implemented in this build" — which the generated
-// dispatch() surfaces as -32603 — until the store and scheduler land.
-//
-// The -32603 collapse for genuine in-handler errors (-32010 / -32011 / -32013) is a known
-// codegen gap, filed as P1 in daemon/docs/proto-requests-m1.md. Not worked around here.
+// Scope of this drop (AGENT-DAEMON.md build order): the transports are real and the store
+// is behind download.add / download.list / download.get. session.hello / session.pair /
+// session.subscribe are handled in the server layer (connection- and transport-stateful)
+// and never reach this class in the running daemon. Everything else still returns
+// "not implemented in this build" (-> -32603) until its handler and the scheduler land;
+// see daemon/docs/deferrals.md.

+#include "store/sqlite.hpp"
 #include "velox_proto.hpp"

 namespace velox::daemon::rpc {

 class VeloxDispatcher final : public velox::proto::Dispatcher {
 public:
+    explicit VeloxDispatcher(velox::daemon::store::Db& db) : db_(db) {}
+
    velox::proto::HandlerResult<velox::proto::CaptureRules>
    on_capture_getRules(const velox::proto::CaptureGetRulesParams&) override;
    velox::proto::HandlerResult<velox::proto::CaptureOfferResult>
@@ -98,6 +98,9 @@ public:
    on_settings_get(const velox::proto::SettingsGetParams&) override;
    velox::proto::HandlerResult<velox::proto::SettingsSetResult>
    on_settings_set(const velox::proto::SettingsSetParams&) override;
+
+private:
+    velox::daemon::store::Db& db_;
 };

 }  // namespace velox::daemon::rpc
@@ -16,6 +16,7 @@

 #include "rpc/event_loop.hpp"
 #include "rpc/ws_handshake.hpp"
+#include "util/time.hpp"
 #include "store/pairings.hpp"
 #include "store/sqlite.hpp"
 #include "version.hpp"
@@ -32,15 +33,6 @@ std::error_code errc(int e) { return std::error_code(e, std::generic_category())
 constexpr std::size_t kMaxOutBytes = 16 * 1024 * 1024;
 constexpr std::size_t kMaxHandshakeBytes = 16 * 1024;

-std::string now_iso() {
-    std::time_t t = std::time(nullptr);
-    std::tm tm{};
-    ::gmtime_r(&t, &tm);
-    char buf[32];
-    std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%SZ", &tm);
-    return std::string(buf);
-}
-
 std::string uuid4() {
    std::random_device rd;
    std::uniform_int_distribution<std::uint32_t> d;
@@ -0,0 +1,21 @@
+#pragma once
+
+// RFC 3339 UTC timestamp, second precision ("2026-09-10T14:55:02Z"). The wire uses this
+// spelling verbatim for createdAt / lastTryAt / completedAt, so a stored value projects
+// with no reformatting.
+
+#include <ctime>
+#include <string>
+
+namespace velox::daemon {
+
+inline std::string now_iso() {
+    const std::time_t t = std::time(nullptr);
+    std::tm tm{};
+    ::gmtime_r(&t, &tm);
+    char buf[32];
+    std::strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%SZ", &tm);
+    return std::string(buf);
+}
+
+}  // namespace velox::daemon
@@ -11,11 +11,11 @@ function(veloxd_test name)
 endfunction()

 veloxd_test(ndjson           LIBS veloxd_rpc)
-veloxd_test(uds_roundtrip     LIBS veloxd_rpc)
+veloxd_test(uds_roundtrip     LIBS veloxd_rpc veloxd_store)
 veloxd_test(store_migrations  LIBS veloxd_store)
 veloxd_test(pairings          LIBS veloxd_store veloxd_rpc)
 veloxd_test(ws_frame          LIBS veloxd_rpc)
-veloxd_test(ws_server         LIBS veloxd_rpc)
+veloxd_test(ws_server         LIBS veloxd_rpc veloxd_store)
 veloxd_test(sched_window     LIBS veloxd_sched)
 veloxd_test(sched_governor   LIBS veloxd_sched)
 veloxd_test(safepath         LIBS veloxd_fs)
@@ -17,6 +17,8 @@
 #include "rpc/event_loop.hpp"
 #include "rpc/ndjson.hpp"
 #include "rpc/uds_server.hpp"
+#include "store/migrations.hpp"
+#include "store/sqlite.hpp"
 #include "velox_proto.hpp"

 using nlohmann::json;
@@ -63,8 +65,13 @@ json call(int fd, const json& request) {
 void run() {
    const std::string sock = make_temp_socket_path();

+    auto db = velox::daemon::store::Db::open(":memory:");
+    CHECK(db.has_value());
+    if (!db) return;
+    CHECK(velox::daemon::store::migrate_to_head(*db).has_value());
+
    rpc::EventLoop loop;
-    rpc::VeloxDispatcher dispatcher;
+    rpc::VeloxDispatcher dispatcher(*db);
    rpc::UdsServer server(loop, dispatcher, sock);
    const auto ec = server.start();
    CHECK(!ec);
@@ -123,7 +123,7 @@ void run() {
    rpc::RuntimeDir rt{dir ? dir : "/tmp"};

    rpc::EventLoop loop;
-    rpc::VeloxDispatcher dispatcher;
+    rpc::VeloxDispatcher dispatcher(*db);
    rpc::EnvAutoApprover approver;
    rpc::WsServer server(loop, dispatcher, *db, approver, rt);
    const auto ec = server.start();